Add shared statistics and Prometheus metrics

src/Application.cpp

@@ -14,7 +14,8 @@
 using namespace Qt::StringLiterals;
 
 Application::Application(QObject *parent) :
-    QObject{parent}
+    QObject{parent},
+    m_metrics{qEnvironmentVariable("STATS_FILE")}
 {
     configureHttpRoutes();
 
@@ -52,6 +53,14 @@ void Application::configureHttpRoutes()
         return QHttpServerResponse{
             QJsonObject{{u"status"_s, u"ok"_s}, {u"version"_s, QString::fromLatin1(IMMICH_SYNC_VERSION)}}};
     });
+    m_httpServer.route(u"/api/stats"_s, [this] {
+        QJsonObject statistics = m_metrics.statistics();
+        statistics.insert(u"version"_s, QString::fromLatin1(IMMICH_SYNC_VERSION));
+        return QHttpServerResponse{statistics};
+    });
+    m_httpServer.route(u"/metrics"_s, [this] {
+        return QHttpServerResponse{"text/plain; version=0.0.4; charset=utf-8", m_metrics.prometheus()};
+    });
     m_httpServer.setMissingHandler(this, [](const QHttpServerRequest &, QHttpServerResponder &responder) {
         responder.write(QByteArrayLiteral("Not found\n"), QByteArrayLiteral("text/plain; charset=utf-8"),
                         QHttpServerResponder::StatusCode::NotFound);
@@ -104,7 +113,7 @@ void Application::acceptWebSocket()
         QWebSocket *socket = pending.release();
         socket->setParent(this);
         socket->setMaxAllowedIncomingMessageSize(64 * 1024);
-        auto *session = new SyncSession{socket, socket};
+        auto *session = new SyncSession{socket, m_metrics, socket};
         connect(socket, &QWebSocket::textMessageReceived, session, &SyncSession::handleTextMessage);
         connect(socket, &QWebSocket::disconnected, session, &SyncSession::abort);
         connect(socket, &QWebSocket::disconnected, socket, &QObject::deleteLater);

src/Application.h

@@ -1,5 +1,7 @@
 #pragma once
 
+#include "Metrics.h"
+
 #include <QHostAddress>
 #include <QHttpServer>
 #include <QObject>
@@ -26,6 +28,7 @@ class Application final : public QObject {
     [[nodiscard]] static QHttpServerResponse resourceResponse(const QString &path, const QByteArray &mimeType);
     static void applySecurityHeaders(QHttpServerResponse &response);
 
+    Metrics m_metrics;
     QHttpServer m_httpServer;
     QTcpServer m_tcpServer;
 };

src/ImmichClient.cpp

@@ -221,6 +221,7 @@ AssetInfo ImmichClient::parseAsset(const QJsonValue &value)
         .id = asset.value(u"id"_s).toString(),
         .checksum = asset.value(u"checksum"_s).toString(),
         .fileName = asset.value(u"originalFileName"_s).toString(),
+        .type = asset.value(u"type"_s).toString(),
         .fileCreatedAt = asset.value(u"fileCreatedAt"_s).toString(),
         .fileModifiedAt = asset.value(u"fileModifiedAt"_s).toString(),
         .favorite = asset.value(u"isFavorite"_s).toBool(),

src/Metrics.cpp

@@ -0,0 +1,249 @@
+#include "Metrics.h"
+
+#include <QDebug>
+#include <QFile>
+#include <QJsonValue>
+#include <QLockFile>
+#include <QSaveFile>
+
+#include <iterator>
+#include <utility>
+
+using namespace Qt::StringLiterals;
+
+namespace {
+qint64 nonNegativeInteger(const QJsonObject &object, const QString &name)
+{
+    return qMax<qint64>(0, object.value(name).toInteger());
+}
+
+QJsonObject statsJson(const Metrics::LifetimeStats &stats)
+{
+    return {
+        {u"syncs"_s, stats.syncs},   {u"assets"_s, stats.assets}, {u"images"_s, stats.images},
+        {u"videos"_s, stats.videos}, {u"other"_s, stats.other},   {u"bytes"_s, stats.bytes},
+    };
+}
+
+void appendMetric(QByteArray &output, const QByteArray &name, const QByteArray &labels, qint64 value)
+{
+    output += name;
+    if (!labels.isEmpty())
+    {
+        output += '{';
+        output += labels;
+        output += '}';
+    }
+    output += ' ';
+    output += QByteArray::number(value);
+    output += '\n';
+}
+}
+
+Metrics::Metrics(QString statsFile) :
+    m_statsFile{std::move(statsFile)}
+{
+    m_memoryStats = loadLifetimeStats();
+}
+
+void Metrics::synchronizationStarted()
+{
+    ++m_syncsStarted;
+    ++m_activeSyncs;
+}
+
+void Metrics::synchronizationFinished(bool success)
+{
+    m_activeSyncs = qMax<qint64>(0, m_activeSyncs - 1);
+    if (success)
+    {
+        ++m_syncsSucceeded;
+    }
+    else
+    {
+        ++m_syncsPartial;
+    }
+    updateLifetimeStats([](LifetimeStats &stats) { ++stats.syncs; });
+}
+
+void Metrics::synchronizationAborted()
+{
+    m_activeSyncs = qMax<qint64>(0, m_activeSyncs - 1);
+    ++m_syncsAborted;
+}
+
+void Metrics::assetFinished(const QString &assetType, bool success, qint64 bytes)
+{
+    const MediaType type = mediaType(assetType);
+    const auto index = static_cast<size_t>(type);
+    if (!success)
+    {
+        ++m_assetsFailed[index];
+        return;
+    }
+
+    ++m_assetsSucceeded[index];
+    m_assetBytes[index] += qMax<qint64>(0, bytes);
+    updateLifetimeStats([type, bytes](LifetimeStats &stats) {
+        ++stats.assets;
+        stats.bytes += qMax<qint64>(0, bytes);
+        switch (type)
+        {
+        case MediaType::Image:
+            ++stats.images;
+            break;
+        case MediaType::Video:
+            ++stats.videos;
+            break;
+        case MediaType::Other:
+            ++stats.other;
+            break;
+        }
+    });
+}
+
+QJsonObject Metrics::statistics() const
+{
+    return statsJson(loadLifetimeStats());
+}
+
+QByteArray Metrics::prometheus() const
+{
+    const LifetimeStats lifetime = loadLifetimeStats();
+    QByteArray output;
+    output.reserve(2'048);
+    output += "# HELP immich_sync_build_info Build and version information.\n";
+    output += "# TYPE immich_sync_build_info gauge\n";
+    output += "immich_sync_build_info{version=\"" IMMICH_SYNC_VERSION "\"} 1\n";
+    output += "# HELP immich_sync_active_synchronizations Synchronizations currently running on this replica.\n";
+    output += "# TYPE immich_sync_active_synchronizations gauge\n";
+    appendMetric(output, "immich_sync_active_synchronizations", {}, m_activeSyncs);
+    output += "# HELP immich_sync_synchronizations_started_total Synchronizations started on this process.\n";
+    output += "# TYPE immich_sync_synchronizations_started_total counter\n";
+    appendMetric(output, "immich_sync_synchronizations_started_total", {}, m_syncsStarted);
+    output += "# HELP immich_sync_synchronizations_finished_total Synchronization outcomes on this process.\n";
+    output += "# TYPE immich_sync_synchronizations_finished_total counter\n";
+    appendMetric(output, "immich_sync_synchronizations_finished_total", "result=\"success\"", m_syncsSucceeded);
+    appendMetric(output, "immich_sync_synchronizations_finished_total", "result=\"partial\"", m_syncsPartial);
+    appendMetric(output, "immich_sync_synchronizations_finished_total", "result=\"aborted\"", m_syncsAborted);
+    output += "# HELP immich_sync_assets_total Asset transfer outcomes on this process.\n";
+    output += "# TYPE immich_sync_assets_total counter\n";
+    static constexpr const char *mediaTypes[]{"image", "video", "other"};
+    for (size_t index = 0; index < std::size(mediaTypes); ++index)
+    {
+        const QByteArray mediaTypeLabel = QByteArrayLiteral("media_type=\"") + mediaTypes[index] + '"';
+        appendMetric(output, "immich_sync_assets_total", mediaTypeLabel + ",result=\"success\"",
+                     m_assetsSucceeded[index]);
+        appendMetric(output, "immich_sync_assets_total", mediaTypeLabel + ",result=\"failed\"", m_assetsFailed[index]);
+    }
+    output += "# HELP immich_sync_asset_bytes_total Successfully transferred original asset bytes on this process.\n";
+    output += "# TYPE immich_sync_asset_bytes_total counter\n";
+    for (size_t index = 0; index < std::size(mediaTypes); ++index)
+    {
+        const QByteArray labels = QByteArrayLiteral("media_type=\"") + mediaTypes[index] + '"';
+        appendMetric(output, "immich_sync_asset_bytes_total", labels, m_assetBytes[index]);
+    }
+    output += "# HELP immich_sync_lifetime_synchronizations_total Completed synchronizations in persistent shared "
+              "statistics.\n";
+    output += "# TYPE immich_sync_lifetime_synchronizations_total counter\n";
+    appendMetric(output, "immich_sync_lifetime_synchronizations_total", {}, lifetime.syncs);
+    output +=
+        "# HELP immich_sync_lifetime_assets_total Successfully synchronized assets in persistent shared statistics.\n";
+    output += "# TYPE immich_sync_lifetime_assets_total counter\n";
+    appendMetric(output, "immich_sync_lifetime_assets_total", "media_type=\"image\"", lifetime.images);
+    appendMetric(output, "immich_sync_lifetime_assets_total", "media_type=\"video\"", lifetime.videos);
+    appendMetric(output, "immich_sync_lifetime_assets_total", "media_type=\"other\"", lifetime.other);
+    output += "# HELP immich_sync_lifetime_asset_bytes_total Successfully synchronized original bytes in persistent "
+              "shared statistics.\n";
+    output += "# TYPE immich_sync_lifetime_asset_bytes_total counter\n";
+    appendMetric(output, "immich_sync_lifetime_asset_bytes_total", {}, lifetime.bytes);
+    return output;
+}
+
+Metrics::MediaType Metrics::mediaType(const QString &assetType)
+{
+    if (assetType.compare(u"IMAGE"_s, Qt::CaseInsensitive) == 0)
+    {
+        return MediaType::Image;
+    }
+    if (assetType.compare(u"VIDEO"_s, Qt::CaseInsensitive) == 0)
+    {
+        return MediaType::Video;
+    }
+    return MediaType::Other;
+}
+
+Metrics::LifetimeStats Metrics::loadLifetimeStats() const
+{
+    if (m_statsFile.isEmpty())
+    {
+        return m_memoryStats;
+    }
+
+    QFile file{m_statsFile};
+    if (!file.exists())
+    {
+        return {};
+    }
+    if (!file.open(QIODevice::ReadOnly))
+    {
+        qWarning() << "Unable to read statistics file" << m_statsFile << file.errorString();
+        return m_memoryStats;
+    }
+    QJsonParseError error;
+    const QJsonValue value = QJsonValue::fromJson(file.readAll(), &error);
+    if (error.error != QJsonParseError::NoError || !value.isObject())
+    {
+        qWarning() << "Unable to parse statistics file" << m_statsFile << error.errorString();
+        return m_memoryStats;
+    }
+    const QJsonObject object = value.toObject();
+    return {
+        .syncs = nonNegativeInteger(object, u"syncs"_s),
+        .assets = nonNegativeInteger(object, u"assets"_s),
+        .images = nonNegativeInteger(object, u"images"_s),
+        .videos = nonNegativeInteger(object, u"videos"_s),
+        .other = nonNegativeInteger(object, u"other"_s),
+        .bytes = nonNegativeInteger(object, u"bytes"_s),
+    };
+}
+
+bool Metrics::storeLifetimeStats(const LifetimeStats &stats) const
+{
+    QSaveFile file{m_statsFile};
+    if (!file.open(QIODevice::WriteOnly))
+    {
+        qWarning() << "Unable to write statistics file" << m_statsFile << file.errorString();
+        return false;
+    }
+    const QByteArray json = QJsonValue{statsJson(stats)}.toJson(QJsonValue::JsonFormat::Compact) + '\n';
+    if (file.write(json) != json.size() || !file.commit())
+    {
+        qWarning() << "Unable to commit statistics file" << m_statsFile << file.errorString();
+        return false;
+    }
+    return true;
+}
+
+void Metrics::updateLifetimeStats(const std::function<void(LifetimeStats &)> &update)
+{
+    if (m_statsFile.isEmpty())
+    {
+        update(m_memoryStats);
+        return;
+    }
+
+    QLockFile lock{m_statsFile + u".lock"_s};
+    lock.setStaleLockTime(30'000);
+    if (!lock.tryLock(5'000))
+    {
+        qWarning() << "Unable to lock statistics file" << m_statsFile << lock.error();
+        return;
+    }
+    LifetimeStats stats = loadLifetimeStats();
+    update(stats);
+    if (storeLifetimeStats(stats))
+    {
+        m_memoryStats = stats;
+    }
+}

src/Metrics.h

@@ -0,0 +1,52 @@
+#pragma once
+
+#include <QByteArray>
+#include <QJsonObject>
+#include <QString>
+
+#include <functional>
+
+class Metrics final {
+  public:
+    struct LifetimeStats {
+        qint64 syncs = 0;
+        qint64 assets = 0;
+        qint64 images = 0;
+        qint64 videos = 0;
+        qint64 other = 0;
+        qint64 bytes = 0;
+    };
+
+    explicit Metrics(QString statsFile = {});
+
+    void synchronizationStarted();
+    void synchronizationFinished(bool success);
+    void synchronizationAborted();
+    void assetFinished(const QString &assetType, bool success, qint64 bytes);
+
+    [[nodiscard]] QJsonObject statistics() const;
+    [[nodiscard]] QByteArray prometheus() const;
+
+  private:
+    enum class MediaType {
+        Image,
+        Video,
+        Other,
+    };
+
+    [[nodiscard]] static MediaType mediaType(const QString &assetType);
+    [[nodiscard]] LifetimeStats loadLifetimeStats() const;
+    bool storeLifetimeStats(const LifetimeStats &stats) const;
+    void updateLifetimeStats(const std::function<void(LifetimeStats &)> &update);
+
+    QString m_statsFile;
+    LifetimeStats m_memoryStats;
+    qint64 m_syncsStarted = 0;
+    qint64 m_syncsSucceeded = 0;
+    qint64 m_syncsPartial = 0;
+    qint64 m_syncsAborted = 0;
+    qint64 m_assetsSucceeded[3]{};
+    qint64 m_assetsFailed[3]{};
+    qint64 m_assetBytes[3]{};
+    qint64 m_activeSyncs = 0;
+};

src/SyncSession.cpp

@@ -10,9 +10,10 @@
 
 using namespace Qt::StringLiterals;
 
-SyncSession::SyncSession(QWebSocket *socket, QObject *parent) :
+SyncSession::SyncSession(QWebSocket *socket, Metrics &metrics, QObject *parent) :
     QObject{parent},
-    m_socket{socket}
+    m_socket{socket},
+    m_metrics{metrics}
 {
     m_requestTimer.setSingleShot(true);
     m_requestTimer.setInterval(10'000);
@@ -315,6 +316,7 @@ void SyncSession::startSync(const QString &direction)
     m_completed = 0;
     m_failed = 0;
     m_active = true;
+    m_metrics.synchronizationStarted();
     qInfo().noquote() << u"Synchronization started: %1, %2 asset(s)"_s.arg(direction).arg(m_total);
     send({{u"type"_s, u"sync-status"_s},
           {u"stage"_s, u"started"_s},
@@ -333,6 +335,7 @@ void SyncSession::startNextTransfer()
     if (m_queue.isEmpty())
     {
         m_active = false;
+        m_metrics.synchronizationFinished(m_failed == 0);
         send({{u"type"_s, u"sync-status"_s},
               {u"stage"_s, u"complete"_s},
               {u"completed"_s, static_cast<qint64>(m_completed)},
@@ -346,8 +349,7 @@ void SyncSession::startNextTransfer()
 
     const TransferSpec &transfer = m_queue.front();
     const quint64 serial = ++m_transferSerial;
-    qInfo().noquote() << u"Transfer %1/%2 started: %3 (%4)"_s
-                             .arg(m_completed + m_failed + 1)
+    qInfo().noquote() << u"Transfer %1/%2 started: %3 (%4)"_s.arg(m_completed + m_failed + 1)
                              .arg(m_total)
                              .arg(transfer.asset.fileName, transfer.direction);
     send({{u"type"_s, u"asset-status"_s},
@@ -490,6 +492,7 @@ void SyncSession::finishTransfer(quint64 serial, bool success, const QString &me
     }
     ++m_transferSerial;
     const TransferSpec transfer = m_queue.takeFirst();
+    const qint64 transferredBytes = m_temporaryFile ? m_temporaryFile->size() : 0;
     qInfo().noquote() << u"Transfer finished: %1 (%2): %3"_s.arg(transfer.asset.fileName,
                                                                  success ? u"success"_s : u"failed"_s, message);
     if (success)
@@ -508,6 +511,7 @@ void SyncSession::finishTransfer(quint64 serial, bool success, const QString &me
             m_download->abort();
         }
     }
+    m_metrics.assetFinished(transfer.asset.type, success, transferredBytes);
     send({{u"type"_s, u"asset-status"_s},
           {u"stage"_s, success ? u"complete"_s : u"failed"_s},
           {u"direction"_s, transfer.direction},
@@ -587,6 +591,7 @@ void SyncSession::abort()
     }
     if (wasActive)
     {
+        m_metrics.synchronizationAborted();
         qInfo() << "Synchronization aborted because its WebSocket disconnected";
         send({{u"type"_s, u"sync-status"_s},
               {u"stage"_s, u"aborted"_s},

src/SyncSession.h

@@ -1,6 +1,7 @@
 #pragma once
 
 #include "ImmichClient.h"
+#include "Metrics.h"
 
 #include <QElapsedTimer>
 #include <QJsonObject>
@@ -17,7 +18,7 @@ class SyncSession final : public QObject {
     Q_OBJECT
 
   public:
-    explicit SyncSession(QWebSocket *socket, QObject *parent = nullptr);
+    explicit SyncSession(QWebSocket *socket, Metrics &metrics, QObject *parent = nullptr);
     ~SyncSession() override;
 
   public slots:
@@ -57,6 +58,7 @@ class SyncSession final : public QObject {
     [[nodiscard]] bool directionAllowed(const QString &direction, QString *reason = nullptr) const;
 
     QPointer<QWebSocket> m_socket;
+    Metrics &m_metrics;
     QTimer m_requestTimer;
     JobKind m_jobKind = JobKind::None;
     QString m_direction;

src/Types.h

@@ -9,6 +9,7 @@ struct AssetInfo {
     QString id;
     QString checksum;
     QString fileName;
+    QString type;
     QString fileCreatedAt;
     QString fileModifiedAt;
     bool favorite = false;