compress logs in database

CMakeLists.txt

@@ -84,7 +84,7 @@ generate_compressed_bins(${CMAKE_BINARY_DIR} js/angular.min.js js/angular-route.
 add_executable(laminard src/database.cpp src/main.cpp src/server.cpp src/laminar.cpp
     src/conf.cpp src/resources.cpp src/run.cpp laminar.capnp.c++ ${COMPRESSED_BINS})
 # TODO: some alternative to boost::filesystem?
-target_link_libraries(laminard capnp-rpc capnp kj-async kj pthread boost_filesystem boost_system sqlite3)
+target_link_libraries(laminard capnp-rpc capnp kj-async kj pthread boost_filesystem boost_system sqlite3 z)
 
 ## Client
 add_executable(laminarc src/client.cpp laminar.capnp.c++)

src/database.cpp

@@ -19,6 +19,7 @@
 #include "database.h"
 
 #include <sqlite3.h>
+#include <string.h>
 
 Database::Database(const char *path) {
     sqlite3_open(path, &hdl);
@@ -47,15 +48,17 @@ void Database::Statement::bindValue(int i, int e) {
 
 void Database::Statement::bindValue(int i, const char* e) {
     sqlite3_bind_text(stmt, i, e, -1, NULL);
-
 }
 
 void Database::Statement::bindValue(int i, std::string e) {
-    sqlite3_bind_text(stmt, i, e.c_str(), e.length(), NULL);
+    sqlite3_bind_blob(stmt, i, e.data(), e.size(), NULL);
 }
 
 template<> std::string Database::Statement::fetchColumn(int col) {
-    return (char*)sqlite3_column_text(stmt, col);
+    int sz = sqlite3_column_bytes(stmt, col);
+    std::string res(sz, '\0');
+    memcpy(&res[0], sqlite3_column_blob(stmt, col), sz);
+    return res;
 }
 
 template<> const char* Database::Statement::fetchColumn(int col) {

src/laminar.cpp

@@ -23,6 +23,7 @@
 
 #include <sys/wait.h>
 #include <fstream>
+#include <zlib.h>
 
 #include <boost/filesystem.hpp>
 namespace fs = boost::filesystem;
@@ -71,7 +72,7 @@ Laminar::Laminar() {
     db->exec("CREATE TABLE IF NOT EXISTS builds("
              "name TEXT, number INT UNSIGNED, node TEXT, queuedAt INT, "
              "startedAt INT, completedAt INT, result INT, output TEXT, "
-             "parentJob TEXT, parentBuild INT, reason TEXT,"
+             "outputLen INT, parentJob TEXT, parentBuild INT, reason TEXT, "
              "PRIMARY KEY (name, number))");
     db->exec("CREATE INDEX IF NOT EXISTS idx_completion_time ON builds("
              "completedAt DESC)");
@@ -128,10 +129,16 @@ void Laminar::sendStatus(LaminarClient* client) {
         if(const Run* run = activeRun(client->scope.job, client->scope.num)) {
             client->sendMessage(run->log.c_str());
         } else { // it must be finished, fetch it from the database
-            db->stmt("SELECT output FROM builds WHERE name = ? AND number = ?")
+            db->stmt("SELECT output, outputLen FROM builds WHERE name = ? AND number = ?")
               .bind(client->scope.job, client->scope.num)
-              .fetch<const char*>([=](const char* log) {
+              .fetch<str,int>([=](str zipped, unsigned long sz) {
+                std::string log(sz+1,'\0');
+                int res = ::uncompress((unsigned char*)&log[0], &sz,
+                        (unsigned char*)zipped.data(), zipped.size());
+                if(res == Z_OK)
                     client->sendMessage(log);
+                else
+                    LLOG(ERROR, "Failed to uncompress log", res);
             });
         }
         return;
@@ -625,7 +632,7 @@ void Laminar::assignNewJobs() {
                 }
 
                 // setup run completion handler
-                run->notifyCompletion = [this](const Run* r) { runFinished(r); };
+                run->notifyCompletion = [this](Run* r) { runFinished(r); };
 
                 // trigger the first step of the run
                 if(stepRun(run)) {
@@ -647,15 +654,24 @@ void Laminar::assignNewJobs() {
     }
 }
 
-void Laminar::runFinished(const Run * r) {
+void Laminar::runFinished(Run * r) {
     Node* node = r->node;
 
     node->busyExecutors--;
     LLOG(INFO, "Run completed", r->name, to_string(r->result));
     time_t completedAt = time(0);
-    db->stmt("INSERT INTO builds VALUES(?,?,?,?,?,?,?,?,?,?,?)")
+
+    // compress log
+    std::string zipped(r->log.size(), '\0');
+    size_t logsize = r->log.length();
+    size_t zippedSize = zipped.size();
+    ::compress((unsigned char*)&zipped[0], &zippedSize,
+            (unsigned char*)&r->log[0], logsize);
+    zipped.resize(zippedSize);
+
+    db->stmt("INSERT INTO builds VALUES(?,?,?,?,?,?,?,?,?,?,?,?)")
      .bind(r->name, r->build, node->name, r->queuedAt, r->startedAt, completedAt, int(r->result),
-           r->log, r->parentName, r->parentBuild, r->reason())
+           zipped, logsize, r->parentName, r->parentBuild, r->reason())
      .exec();
 
     // notify clients

src/laminar.h

@@ -61,7 +61,7 @@ private:
     void reapAdvance();
     void assignNewJobs();
     bool stepRun(std::shared_ptr<Run> run);
-    void runFinished(const Run*);
+    void runFinished(Run*);
     bool nodeCanQueue(const Node&, const Run&) const;
     // expects that Json has started an array
     void populateArtifacts(Json& out, std::string job, int num) const;

src/run.h

@@ -68,7 +68,7 @@ public:
 
     std::string reason() const;
 
-    std::function<void(const Run*)> notifyCompletion;
+    std::function<void(Run*)> notifyCompletion;
     Node* node;
     RunState result;
     RunState lastResult;