Merge pull request #31 from acoustid/json-log

Json log

.github/workflows/build.yml

@@ -10,15 +10,15 @@ env:
 
 jobs:
   build:
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-22.04
     permissions:
       contents: read
       packages: write
     steps:
       - uses: actions/checkout@v2
       - run: |
           sudo apt-get update
-          sudo apt-get install -y qtbase5-dev libgtest-dev
+          sudo apt-get install -y qt6-base-dev libgtest-dev
       - run: cmake -DCMAKE_BUILD_TYPE=Release .
       - run: make
       - run: make check

CMakeLists.txt

@@ -16,7 +16,7 @@ set(CMAKE_AUTORCC ON)
 
 find_package(Threads REQUIRED)
 find_package(GTest)
-find_package(Qt5 COMPONENTS Core Network Concurrent REQUIRED)
+find_package(Qt6 COMPONENTS Core Network Concurrent REQUIRED)
 
 set(CPACK_GENERATOR "DEB")
 set(CPACK_PACKAGE_NAME acoustid-index)
@@ -68,9 +68,10 @@ set(fpindexlib_SOURCES
 	src/store/ram_output_stream.cpp
 	src/util/crc.c
 	src/util/options.cpp
+	src/util/tracing.cpp
 )
 add_library(fpindexlib ${fpindexlib_SOURCES})
-target_link_libraries(fpindexlib Qt5::Core Qt5::Network Qt5::Concurrent)
+target_link_libraries(fpindexlib Qt6::Core Qt6::Network Qt6::Concurrent)
 
 set(qhttp_SOURCES
     ./src/3rdparty/qhttp/src/qhttpserverconnection.cpp
@@ -85,7 +86,7 @@ set(qhttp_SOURCES
 )
 add_library(qhttp ${qhttp_SOURCES})
 target_include_directories(qhttp PRIVATE ./src/3rdparty/)
-target_link_libraries(qhttp Qt5::Core Qt5::Network Qt5::Concurrent)
+target_link_libraries(qhttp Qt6::Core Qt6::Network Qt6::Concurrent)
 
 set(fpserver_SOURCES
 	src/server/listener.cpp

Dockerfile

@@ -1,9 +1,9 @@
-FROM ubuntu:20.04
+FROM ubuntu:22.04
 
 RUN useradd -m -s /bin/bash -u 1000 acoustid
 
 RUN apt-get update && \
-    apt-get install -y libqt5network5 libqt5core5a libstdc++6 libgcc1 libgcc-s1
+    apt-get install -y libqt6network6 libqt6core6 libstdc++6 libgcc1 libgcc-s1
 
 ADD acoustid-index.deb /tmp/
 RUN dpkg -i /tmp/acoustid-index.deb && rm /tmp/acoustid-index.deb

src/index/index.cpp

@@ -15,7 +15,7 @@
 using namespace Acoustid;
 
 Index::Index(DirectorySharedPtr dir, bool create)
-	: m_mutex(QMutex::Recursive), m_dir(dir), m_open(false),
+	: m_mutex(), m_dir(dir), m_open(false),
 	  m_hasWriter(false),
 	  m_deleter(new IndexFileDeleter(dir))
 {
@@ -28,9 +28,11 @@ Index::~Index()
 
 void Index::open(bool create)
 {
+	QMutexLocker locker(&m_mutex);
 	if (!m_info.load(m_dir.data(), true)) {
 		if (create) {
 			IndexWriter(m_dir, m_info).commit();
+			locker.unlock();
 			return open(false);
 	 	}
 		throw IOException("there is no index in the directory");
@@ -45,6 +47,7 @@ QSharedPointer<IndexReader> Index::openReader()
     if (!m_open) {
        throw IndexIsNotOpen("index is not open");
     }
+    locker.unlock();
     return QSharedPointer<IndexReader>::create(sharedFromThis());
 }
 
@@ -55,6 +58,7 @@ QSharedPointer<IndexWriter> Index::openWriter(bool wait, int64_t timeoutInMSecs)
         throw IndexIsNotOpen("index is not open");
     }
     acquireWriterLockInt(wait, timeoutInMSecs);
+    locker.unlock();
     return QSharedPointer<IndexWriter>::create(sharedFromThis(), true);
 }
 

src/index/index_writer.cpp

@@ -114,7 +114,7 @@ void IndexWriter::merge(const QList<int>& merge)
 		throw CorruptIndexException("checksum mismatch after merge");
 	}
 
-	QSet<int> merged = merge.toSet();
+	QSet<int> merged = QSet<int>(merge.begin(), merge.end());
 	info.clearSegments();
 	for (size_t i = 0; i < segments.size(); i++) {
 		const SegmentInfo& s = segments.at(i);

src/index/segment_merge_policy.cpp

@@ -48,7 +48,7 @@ QList<int> SegmentMergePolicy::findMerges(const SegmentInfoList& infos)
 	for (size_t i = 0; i < infos.size(); i++) {
 		segments.append(i);
 	}
-	qStableSort(segments.begin(), segments.end(), SegmentSizeLessThan(&infos));
+	std::stable_sort(segments.begin(), segments.end(), SegmentSizeLessThan(&infos));
 	//qDebug() << "Order after sorting is " << segments;
 
 	size_t minSegmentSize = infos.at(segments.last()).blockCount();

src/server/connection.cpp

@@ -10,6 +10,7 @@
 #include "errors.h"
 #include "protocol.h"
 #include "metrics.h"
+#include "util/tracing.h"
 
 using namespace Acoustid;
 using namespace Acoustid::Server;
@@ -133,6 +134,7 @@ void Connection::readIncomingData()
     }
 
     auto futureResult = QtConcurrent::run([=]() {
+	setTraceId(m_session->getTraceId());
         QString response;
         try {
             response = renderResponse(handler());

src/server/http/request.cpp

@@ -24,6 +24,15 @@ QString HttpRequest::param(const QString &name, const QString &defaultValue) con
     return defaultValue;
 }
 
+QString HttpRequest::header(const QString &name, const QString &defaultValue) const {
+    auto encodedName = name.toUtf8();
+    auto it = m_headers.find(encodedName);
+    if (it != m_headers.end()) {
+	return *it;
+    }
+    return defaultValue;
+}
+
 QUrl HttpRequest::url() const { return m_url; }
 
 QJsonDocument HttpRequest::json() const {

src/server/http/request.h

@@ -37,6 +37,8 @@ class HttpRequest {
 
     QByteArray body() const { return m_body; }
 
+    QString header(const QString &name, const QString &defaultValue = QString()) const;
+
     void setMethod(HttpMethod method) { m_method = method; }
 
     void setUrl(const QUrl &url) {

src/server/http/response.cpp

@@ -28,7 +28,7 @@ void HttpResponse::send(qhttp::server::QHttpRequest *req, qhttp::server::QHttpRe
         res->end();
         return;
     }
-    res->addHeaderValue("Content-Length", m_body.size());
+    res->addHeaderValue("Content-Length", static_cast<size_t>(m_body.size()));
     res->end(m_body);
 }
 

src/server/http/router.cpp

@@ -1,7 +1,8 @@
 #include "router.h"
+#include "util/tracing.h"
 
 #include <QRegularExpression>
-#include <QtConcurrent>
+#include <QThreadPool>
 
 namespace Acoustid {
 namespace Server {
@@ -10,7 +11,7 @@ void HttpRouter::route(HttpMethod method, const QString &path, HttpHandlerFunc h
     auto pathParts = path.split('/');
     for (auto &pathPart : pathParts) {
         if (pathPart.startsWith(':')) {
-            auto paramName = pathPart.midRef(1);
+            auto paramName = pathPart.mid(1);
             pathPart = "(?<" + paramName + ">[^/]+)";
         }
     }
@@ -51,7 +52,11 @@ void HttpRouter::handle(qhttp::server::QHttpRequest *req, qhttp::server::QHttpRe
         HttpRequest request(req->method(), req->url());
         request.setHeaders(req->headers());
         request.setBody(req->collectedData());
-        QtConcurrent::run([=]() {
+	QThreadPool::globalInstance()->start([=]() {
+            auto traceId = request.header("X-Trace-Id");
+	    if (!traceId.isEmpty()) {
+                setTraceId(traceId);
+	    }
             HttpResponse response;
             try {
                 response = handle(request);

src/server/main.cpp

@@ -3,6 +3,8 @@
 
 #include <QCoreApplication>
 #include <QThreadPool>
+#include <QJsonObject>
+#include <QJsonDocument>
 #include "qhttpserver.hpp"
 #include "qhttpserverrequest.hpp"
 #include "qhttpserverresponse.hpp"
@@ -16,8 +18,47 @@ using namespace Acoustid::Server;
 
 using namespace qhttp::server;
 
+static QTextStream stderrStream(stderr);
+
+void handleLogMessage(QtMsgType type, const QMessageLogContext &context, const QString &msg)
+{
+    QString time = QDateTime::currentDateTimeUtc().toString(Qt::ISODateWithMs);
+
+    QString level;
+    switch (type) {
+	case QtDebugMsg:
+	    level = "debug";
+	    break;
+	case QtInfoMsg:
+	    level = "info";
+	    break;
+	case QtWarningMsg:
+	    level = "warning";
+	    break;
+	case QtCriticalMsg:
+	    level = "error";
+	    break;
+	case QtFatalMsg:
+	    level = "error";
+	    break;
+    }
+
+    QJsonObject obj;
+    obj.insert("time", time);
+    obj.insert("level", level);
+    obj.insert("message", msg);
+
+    stderrStream << QJsonDocument(obj).toJson(QJsonDocument::Compact) << Qt::endl;
+
+    if (type == QtFatalMsg) {
+	abort();
+    }
+}
+
 int main(int argc, char **argv)
 {
+	qInstallMessageHandler(handleLogMessage);
+
 	OptionParser parser("%prog [options]");
 	parser.addOption("directory", 'd')
 		.setArgument()
@@ -72,16 +113,16 @@ int main(int argc, char **argv)
 	Listener listener(path, opts->contains("mmap"));
 	listener.setMetrics(metrics);
 	listener.listen(QHostAddress(address), port);
-	qDebug() << "Simple server listening on" << address << "port" << port;
+	qInfo() << "Simple server listening on" << address << "port" << port;
 
     HttpRequestHandler handler(listener.index(), metrics);
 
 	QHttpServer httpListener(&app);
     httpListener.listen(QHostAddress(httpAddress), httpPort, [&](QHttpRequest *req, QHttpResponse *res) {
         handler.router().handle(req, res);
     });
-    qDebug() << "HTTP server listening on" << httpAddress << "port" << httpPort;
-    qDebug() << "Prometheus metrics available at" << QString("http://%1:%2/_metrics").arg(httpAddress).arg(httpPort);
+    qInfo() << "HTTP server listening on" << httpAddress << "port" << httpPort;
+    qInfo() << "Prometheus metrics available at" << QString("http://%1:%2/_metrics").arg(httpAddress).arg(httpPort);
 
 	return app.exec();
 }

src/server/protocol.cpp

@@ -65,13 +65,13 @@ ScopedHandlerFunc buildHandler(const QString &command, const QStringList &args)
     } else if (command == "begin") {
         return [=](QSharedPointer<Session> session) { session->begin(); return QString(); };
     } else if (command == "commit") {
-        return [=](QSharedPointer<Session> session) { session->commit(); return QString(); };
+        return [=](QSharedPointer<Session> session) { session->commit(); session->clearTraceId(); return QString(); };
     } else if (command == "rollback") {
-        return [=](QSharedPointer<Session> session) { session->rollback(); return QString(); };
+        return [=](QSharedPointer<Session> session) { session->rollback(); session->clearTraceId(); return QString(); };
     } else if (command == "optimize") {
-        return [=](QSharedPointer<Session> session) { session->optimize(); return QString(); };
+        return [=](QSharedPointer<Session> session) { session->optimize(); session->clearTraceId(); return QString(); };
     } else if (command == "cleanup") {
-        return [=](QSharedPointer<Session> session) { session->cleanup(); return QString(); };
+        return [=](QSharedPointer<Session> session) { session->cleanup(); session->clearTraceId(); return QString(); };
     } else if (command == "insert") {
         if (args.size() != 2) {
             throw BadRequest("expected two arguments");
@@ -80,6 +80,7 @@ ScopedHandlerFunc buildHandler(const QString &command, const QStringList &args)
             auto id = args.at(0).toInt();
             auto hashes = parseFingerprint(args.at(1));
             session->insert(id, hashes);
+	    session->clearTraceId();
             return QString();
         };
     } else if (command == "search") {
@@ -94,7 +95,9 @@ ScopedHandlerFunc buildHandler(const QString &command, const QStringList &args)
             for (int i = 0; i < results.size(); i++) {
                 output.append(QString("%1:%2").arg(results[i].id()).arg(results[i].score()));
             }
-            return output.join(" ");
+	    QString outputString = output.join(" ");
+	    session->clearTraceId();
+            return outputString;
         };
     } else {
         throw BadRequest(QString("unknown command %1").arg(command));

src/server/session.cpp

@@ -89,6 +89,10 @@ void Session::setAttribute(const QString &name, const QString &value) {
         m_idle_timeout = value.toInt();
         return;
     }
+    if (name == "trace_id") {
+	m_traceId = value;
+	return;
+    }
     if (m_indexWriter.isNull()) {
         throw NotInTransactionException();
     }
@@ -114,3 +118,13 @@ QList<Result> Session::search(const QVector<uint32_t> &hashes) {
     }
     return collector.topResults();
 }
+
+QString Session::getTraceId() {
+    QMutexLocker locker(&m_mutex);
+    return m_traceId;
+}
+
+void Session::clearTraceId() {
+    QMutexLocker locker(&m_mutex);
+    m_traceId.clear();
+}

src/server/session.h

@@ -39,6 +39,9 @@ class Session
 
     QSharedPointer<Metrics> metrics() const { return m_metrics; }
 
+    QString getTraceId();
+    void clearTraceId();
+
 private:
 	QMutex m_mutex;
     QSharedPointer<Index> m_index;
@@ -48,6 +51,7 @@ class Session
 	int m_maxResults { 500 };
     int64_t m_timeout { 0 };
     int64_t m_idle_timeout { 60 * 1000 };
+    QString m_traceId;
 };
 
 }

src/util/tracing.cpp

@@ -0,0 +1,14 @@
+#include <QThreadStorage>
+#include "util/tracing.h"
+
+static QThreadStorage<QString> traceId;
+
+void setTraceId(const QString &value)
+{
+    traceId.setLocalData(value);
+}
+
+QString getTraceId()
+{
+    return traceId.localData();
+}

src/util/tracing.h

@@ -0,0 +1,6 @@
+#pragma once
+
+#include <QString>
+
+QString getTraceId();
+void setTraceId(const QString &id);