Add fs top endpoint and refactor performance collector

README.md

@@ -8,6 +8,9 @@ Add the following config section to your telegraf.yml:
 [[inputs.exec]]
   commands = ["/etc/telegraf/bcachefs.sh"]
   data_format = "influx"
+[[inputs.exec]]
+  commands = ["/etc/telegraf/bcachefs_fs_top.sh"]
+  data_format = "influx"
 [[inputs.exec]]
   commands = ["/etc/telegraf/bcachefs_options.sh"]
   data_format = "influx"

bcachefs_fs_top.sh

@@ -0,0 +1,41 @@
+#!/bin/bash
+
+set -euo pipefail
+
+function h2b()
+{
+    local val="$1"
+    local num unit
+    if [[ $val =~ ^([0-9.]+)([BKMGTP]?)$ ]]; then
+        num="${BASH_REMATCH[1]}"
+        unit="${BASH_REMATCH[2]}"
+        case "$unit" in
+            B|"") echo "${num%.*}" ;;
+            K) echo "$(awk "BEGIN{printf \"%d\", $num*1024}")" ;;
+            M) echo "$(awk "BEGIN{printf \"%d\", $num*1024*1024}")" ;;
+            G) echo "$(awk "BEGIN{printf \"%d\", $num*1024*1024*1024}")" ;;
+            T) echo "$(awk "BEGIN{printf \"%d\", $num*1024*1024*1024*1024}")" ;;
+            P) echo "$(awk "BEGIN{printf \"%d\", $num*1024*1024*1024*1024*1024}")" ;;
+            *) echo "$num" ;;
+        esac
+    else
+        echo 0
+    fi
+}
+
+curl localhost:8000/fs_top 2>/dev/null | while IFS= read -r line; do
+    # data_read               0B/sec     55660544B    512118784B
+    if [[ "$line" =~ ^([a-zA-Z0-9_]+)[[:space:]]+([0-9.]+[BKMGTP]?)/sec[[:space:]]+([0-9.]+[BKMGTP]?)[[:space:]]+([0-9.]+[BKMGTP]?)$ ]]; then
+        stat="${BASH_REMATCH[1]}"
+        rate_raw="${BASH_REMATCH[2]}"
+        total_raw="${BASH_REMATCH[3]}"
+        mount_raw="${BASH_REMATCH[4]}"
+
+        rate=$(h2b "$rate_raw")
+        total=$(h2b "$total_raw")
+        mount=$(h2b "$mount_raw")
+
+        printf "bcachefs_fs_top,stat=%s rate=%di,total=%di,mount=%di\n" \
+               "$stat" "$rate" "$total" "$mount"
+    fi
+done

perf-collector/fstop.cpp

@@ -0,0 +1,7 @@
+#include "fstop.h"
+
+FsTop::FsTop(QHttpServer &server, QObject *parent) :
+    ProcessFrameCapturer{server, "/fs_top",
+                         "bcachefs", {"fs", "top", "/mnt"},
+                         parent}
+{}

perf-collector/fstop.h

@@ -0,0 +1,11 @@
+#pragma once
+
+#include "processframecapturer.h"
+
+class FsTop : public ProcessFrameCapturer
+{
+    Q_OBJECT
+
+public:
+    explicit FsTop(QHttpServer &server, QObject *parent = nullptr);
+};

perf-collector/main.cpp

@@ -1,15 +1,11 @@
 #include <QCoreApplication>
-#include <QProcess>
 #include <QDebug>
 #include <QTcpServer>
 #include <QHttpServer>
-#include <QJsonObject>
 #include <memory>
-#include <sys/prctl.h>
-#include <signal.h>
-#include <unistd.h>
 
-QHash<QByteArray, QHash<QByteArray, QHash<QByteArray, int>>> counters;
+#include "perfjson.h"
+#include "fstop.h"
 
 int main(int argc, char *argv[])
 {
@@ -20,111 +16,21 @@ int main(int argc, char *argv[])
     QHttpServer server;
 
     if (!tcpServer->listen(QHostAddress::LocalHost, 8000))
+    {
+        qFatal("could not listen: %s", qPrintable(tcpServer->errorString()));
         return -1;
+    }
 
     if (!server.bind(tcpServer.get()))
+    {
+        qFatal("could not bind webserver to socket");
         return -2;
+    }
 
-    tcpServer.release();
+    tcpServer.release(); // QHttpServer now owns QTcpServer
 
-    server.route("/perf_json", [](){
-        auto filesystems = QJsonObject{};
-
-        for (auto iter = std::cbegin(counters); iter != std::cend(counters); iter++)
-        {
-            auto functions = QJsonObject{};
-
-            for (auto iter2 = std::cbegin(iter.value()); iter2 != std::cend(iter.value()); iter2++)
-            {
-                auto strs = QJsonObject{};
-
-                for (auto iter3 = std::cbegin(iter2.value()); iter3 != std::cend(iter2.value()); iter3++)
-                    strs[iter3.key()] = iter3.value();
-
-                functions[iter2.key()] = std::move(strs);
-            }
-
-            filesystems[iter.key()] = std::move(functions);
-        }
-
-        return filesystems;
-    });
-
-    QProcess process;
-
-    process.setProcessChannelMode(QProcess::MergedChannels);
-
-    process.setChildProcessModifier([] {
-        prctl(PR_SET_PDEATHSIG, SIGKILL);
-    });
-
-    QObject::connect(&process, &QProcess::started, [](){
-        qDebug() << "started";
-    });
-
-    QObject::connect(&process, &QProcess::finished, [&](int exitCode, QProcess::ExitStatus exitStatus){
-        qDebug() << "finished exitCode =" << exitCode << "exitStatus =" << exitStatus;
-        app.exit(exitCode);
-    });
-
-    QObject::connect(&process, &QProcess::readyRead, [&](){
-        while (process.canReadLine())
-        {
-            auto lineStr = process.readLine();
-
-            QByteArrayView line = lineStr;
-            line = line.trimmed();
-
-            //qDebug() << "received" << line;
-
-            qsizetype tsEnd = line.indexOf(' ');
-            QByteArrayView timestamp = line.left(tsEnd);
-
-            qsizetype lparen = line.indexOf('(', tsEnd + 1);
-
-            qsizetype funcStart = line.lastIndexOf(' ', lparen);
-            QByteArrayView function = line.mid(funcStart + 1, lparen - funcStart - 1);
-
-            QByteArrayView thread = line.mid(tsEnd + 1, funcStart - tsEnd - 1);
-
-            qsizetype rparen = line.lastIndexOf(')');
-            QByteArrayView args = line.mid(lparen + 1, rparen - lparen - 1);
-
-            //qDebug() << "timestamp =" << timestamp;
-            //qDebug() << "thread =" << thread;
-            //qDebug() << "function =" << function;
-            //qDebug() << "args =" << args;
-
-            QByteArrayView fs, str;
-            while (!args.isEmpty())
-            {
-                qsizetype colon = args.indexOf(':');
-                qsizetype quote1 = args.indexOf('"', colon);
-                qsizetype quote2 = args.indexOf('"', quote1 + 1);
-
-                QByteArrayView key   = args.left(colon).trimmed();
-                QByteArrayView value = args.mid(quote1 + 1, quote2 - quote1 - 1);
-
-                //qDebug() << "key =" << key << "value =" << value;
-
-                if (key == "fs")
-                    fs = value;
-                else if (key == "str")
-                    str = value;
-
-                qsizetype next = args.indexOf(',', quote2);
-                if (next == -1) break;
-                args = args.mid(next + 1).trimmed();
-            }
-
-            if (fs.isEmpty())
-                continue;
-
-            counters[fs][function][str]++;
-        }
-    });
-
-    process.start("sudo", {"perf", "trace", "-e", "bcachefs:error_throw", "-e", "bcachefs:write_super"});
+    PerfJson perfJson{server};
+    FsTop fsTop{server};
 
     return app.exec();
 }

perf-collector/perf-collector.pro

@@ -5,4 +5,12 @@ CONFIG += c++17 cmdline
 DEFINES += QT_DISABLE_DEPRECATED_BEFORE=0x060000    # disables all the APIs deprecated before Qt 6.0.0
 
 SOURCES += \
-        main.cpp
+        fstop.cpp \
+        main.cpp \
+        perfjson.cpp \
+        processframecapturer.cpp
+
+HEADERS += \
+    fstop.h \
+    perfjson.h \
+    processframecapturer.h

perf-collector/perfjson.cpp

@@ -0,0 +1,121 @@
+#include "perfjson.h"
+
+#include <QHttpServer>
+#include <QJsonObject>
+#include <QDebug>
+#include <QCoreApplication>
+
+#include <unistd.h>
+#include <sys/prctl.h>
+#include <signal.h>
+
+PerfJson::PerfJson(QHttpServer &server, QObject *parent) :
+    QObject{parent}
+{
+    m_process.setProcessChannelMode(QProcess::MergedChannels);
+
+    m_process.setChildProcessModifier([] {
+        prctl(PR_SET_PDEATHSIG, SIGKILL);
+    });
+
+    connect(&m_process, &QProcess::started, this, &PerfJson::processStarted);
+    connect(&m_process, &QProcess::finished, this, &PerfJson::processFinished);
+    connect(&m_process, &QProcess::readyRead, this, &PerfJson::readyRead);
+
+    m_process.start("sudo", {"perf", "trace", "-e", "bcachefs:error_throw", "-e", "bcachefs:write_super"});
+
+    server.route("/perf_json", this, &PerfJson::getPerfJson);
+}
+
+void PerfJson::processStarted()
+{
+    qDebug() << "started";
+}
+
+void PerfJson::processFinished(int exitCode, QProcess::ExitStatus exitStatus)
+{
+    qDebug() << "finished exitCode =" << exitCode << "exitStatus =" << exitStatus;
+    QCoreApplication::exit(exitCode);
+}
+
+void PerfJson::readyRead()
+{
+    while (m_process.canReadLine())
+    {
+        auto lineStr = m_process.readLine();
+
+        QByteArrayView line = lineStr;
+        line = line.trimmed();
+
+        //qDebug() << "received" << line;
+
+        qsizetype tsEnd = line.indexOf(' ');
+        QByteArrayView timestamp = line.left(tsEnd);
+
+        qsizetype lparen = line.indexOf('(', tsEnd + 1);
+
+        qsizetype funcStart = line.lastIndexOf(' ', lparen);
+        QByteArrayView function = line.mid(funcStart + 1, lparen - funcStart - 1);
+
+        QByteArrayView thread = line.mid(tsEnd + 1, funcStart - tsEnd - 1);
+
+        qsizetype rparen = line.lastIndexOf(')');
+        QByteArrayView args = line.mid(lparen + 1, rparen - lparen - 1);
+
+        //qDebug() << "timestamp =" << timestamp;
+        //qDebug() << "thread =" << thread;
+        //qDebug() << "function =" << function;
+        //qDebug() << "args =" << args;
+
+        QByteArrayView fs, str;
+        while (!args.isEmpty())
+        {
+            qsizetype colon = args.indexOf(':');
+            qsizetype quote1 = args.indexOf('"', colon);
+            qsizetype quote2 = args.indexOf('"', quote1 + 1);
+
+            QByteArrayView key   = args.left(colon).trimmed();
+            QByteArrayView value = args.mid(quote1 + 1, quote2 - quote1 - 1);
+
+            //qDebug() << "key =" << key << "value =" << value;
+
+            if (key == "fs")
+                fs = value;
+            else if (key == "str")
+                str = value;
+
+            qsizetype next = args.indexOf(',', quote2);
+            if (next == -1) break;
+            args = args.mid(next + 1).trimmed();
+        }
+
+        if (fs.isEmpty())
+            continue;
+
+        counters[fs][function][str]++;
+    }
+}
+
+QJsonObject PerfJson::getPerfJson() const
+{
+    auto filesystems = QJsonObject{};
+
+    for (auto iter = std::cbegin(counters); iter != std::cend(counters); iter++)
+    {
+        auto functions = QJsonObject{};
+
+        for (auto iter2 = std::cbegin(iter.value()); iter2 != std::cend(iter.value()); iter2++)
+        {
+            auto strs = QJsonObject{};
+
+            for (auto iter3 = std::cbegin(iter2.value()); iter3 != std::cend(iter2.value()); iter3++)
+                strs[iter3.key()] = iter3.value();
+
+            functions[iter2.key()] = std::move(strs);
+        }
+
+        filesystems[iter.key()] = std::move(functions);
+    }
+
+    return filesystems;
+}

perf-collector/perfjson.h

@@ -0,0 +1,29 @@
+#pragma once
+
+#include <QObject>
+#include <QHash>
+#include <QByteArray>
+#include <QProcess>
+
+class QHttpServer;
+class QJsonObject;
+
+class PerfJson : public QObject
+{
+    Q_OBJECT
+
+public:
+    PerfJson(QHttpServer &server, QObject *parent = nullptr);
+
+private slots:
+    void processStarted();
+    void processFinished(int exitCode, QProcess::ExitStatus exitStatus);
+    void readyRead();
+
+private:
+    QJsonObject getPerfJson() const;
+
+    QHash<QByteArray, QHash<QByteArray, QHash<QByteArray, int>>> counters;
+
+    QProcess m_process;
+};

perf-collector/processframecapturer.cpp

@@ -0,0 +1,54 @@
+#include "processframecapturer.h"
+
+#include <QHttpServer>
+#include <QDebug>
+#include <QCoreApplication>
+
+#include <unistd.h>
+#include <sys/prctl.h>
+#include <signal.h>
+
+ProcessFrameCapturer::ProcessFrameCapturer(QHttpServer &server, const QString &pathPattern,
+                                           const QString &program, const QStringList &arguments,
+                                           QObject *parent) :
+    QObject{parent}
+{
+    m_process.setProcessChannelMode(QProcess::MergedChannels);
+
+    m_process.setChildProcessModifier([] {
+        prctl(PR_SET_PDEATHSIG, SIGKILL);
+    });
+
+    connect(&m_process, &QProcess::started, this, &ProcessFrameCapturer::processStarted);
+    connect(&m_process, &QProcess::finished, this, &ProcessFrameCapturer::processFinished);
+    connect(&m_process, &QProcess::readyRead, this, &ProcessFrameCapturer::readyRead);
+
+    m_process.start(program, arguments);
+
+    server.route(pathPattern, this, &ProcessFrameCapturer::getProcessFrame);
+}
+
+void ProcessFrameCapturer::processStarted()
+{
+    qDebug() << "started";
+}
+
+void ProcessFrameCapturer::processFinished(int exitCode, QProcess::ExitStatus exitStatus)
+{
+    qDebug() << "finished exitCode =" << exitCode << "exitStatus =" << exitStatus;
+    QCoreApplication::exit(exitCode);
+}
+
+void ProcessFrameCapturer::readyRead()
+{
+    auto buf = m_process.readAll();
+    qDebug() << buf;
+    constexpr QByteArrayView search = "\x1B[2J\x1B[H";
+    const auto index = buf.lastIndexOf(search);
+    if (index != -1)
+    {
+        m_buffer.clear();
+        m_buffer.remove(0, index + search.size());
+    }
+    m_buffer.append(buf);
+}

perf-collector/processframecapturer.h

@@ -0,0 +1,28 @@
+#pragma once
+
+#include <QObject>
+#include <QByteArray>
+#include <QProcess>
+
+class QHttpServer;
+
+class ProcessFrameCapturer : public QObject
+{
+    Q_OBJECT
+public:
+    explicit ProcessFrameCapturer(QHttpServer &server, const QString &pathPattern,
+                                  const QString &program, const QStringList &arguments = {},
+                                  QObject *parent = nullptr);
+
+private slots:
+    void processStarted();
+    void processFinished(int exitCode, QProcess::ExitStatus exitStatus);
+    void readyRead();
+
+private:
+    QByteArray getProcessFrame() const { return m_buffer; }
+
+    QByteArray m_buffer;
+
+    QProcess m_process;
+};