Improve timeseries effiency when handling large result sets

Author: jcoffland

src/cbang/api/Timeseries.cpp

@@ -89,34 +89,12 @@ void Timeseries::query(uint64_t since, unsigned maxResults, const cb_t &cb) {
 }
 
 
-void Timeseries::add(uint64_t time, const JSON::ValuePtr &value) {
-  // Load last data point if necessary
-  if (last.isNull()) {
-    auto cb = [=] (
-      const SmartPointer<Exception> &err, const JSON::ValuePtr &data) {
-      if (err.isNull() && data.isSet() && data->isList() && data->size())
-        last = data->get(0)->get("value");
-      add(time, value);
-    };
-
-    last = JSON::Null::instancePtr();
-    return query(0, 1, cb);
-  }
-
-  // Don't record if result is unchanged
-  if (*last != *value) {
-    last = value;
+void Timeseries::broadcast(uint64_t time, const JSON::ValuePtr &value) {
+  if (subscribers.empty()) return;
 
-    auto key = getKey(time);
-    LOG_DEBUG(5, "key: " << key << " value: " << *value);
-    db.set(key, value->toString());
-
-    if (!subscribers.empty()) {
-      auto entry = makeEntry(time, value);
-      for (auto p: subscribers)
-        if (p.second.isSet()) p.second->next(entry);
-    }
-  }
+  auto entry = makeEntry(time, value);
+  for (auto p: subscribers)
+    if (p.second.isSet()) p.second->next(entry);
 }
 
 
@@ -152,8 +130,3 @@ void Timeseries::unsubscribe(uint64_t id) {
     THROWX("Timeseries does not have subscriber with id " << id,
       HTTP::Status::HTTP_NOT_FOUND);
 }
-
-
-string Timeseries::getKey(uint64_t ts) const {
-  return Time(handler.getTimePeriod(ts)).toString(TIME_FMT);
-}

src/cbang/api/Timeseries.h

@@ -60,14 +60,13 @@ namespace cb {
       Timeseries(TimeseriesHandler &handler, const std::string &key);
 
       void query(uint64_t since, unsigned maxResults, const cb_t &cb);
-      void add(uint64_t time, const JSON::ValuePtr &value);
+      void broadcast(uint64_t time, const JSON::ValuePtr &value);
 
       SmartPointer<Subscriber> subscribe(
         uint64_t id, uint64_t since, unsigned maxResults, const cb_t &cb);
       void unsubscribe(uint64_t id);
 
     protected:
-      std::string getKey(uint64_t ts) const;
       void query(uint64_t ts);
       void query();
     };

src/cbang/api/handler/TimeseriesHandler.cpp

@@ -29,6 +29,37 @@
                           joseph@cauldrondevelopment.com
 
 \******************************************************************************/
+/*
+  This class handles the storing and recalling of timeseries data.  Queries
+  are made to the SQL DB and those results are stored as a timeseries in
+  LevelDB.
+
+  Timeseries data may either be a single value or a list of values keyed by
+  one or more variables.  The key or keys are specified via the ``key``
+  configuration option.  The key value(s) are removed from the timeseries
+  data to save space.  If the resulting data is a dictionary with a single
+  value then only the value is stored.
+
+  Clients may subscribe to the timeseries via Websocket.  Subscribers will
+  first receive a list of all the recent timeseries data limited by time
+  and total number of results.  Subsequently, new results will be broadcast
+  to all subscribers for as long as they stay connected.
+
+  The last result is tracked in LevelDB.  If a new result is the same as the
+  last result it is not added to the timeseries.  In this way, gaps in the time
+  series are assumed to be repeats of the previous value.
+
+  The "last" result is stored at key "\0".  If the timeseries is a list of
+  values then "last" will be dictionary of key, value pairs, otherwise just the
+  one value.  The "last" value is reloaded when the server is restarted before
+  making further queries.
+
+  The implementation is complicated because care is taken to not dominate the
+  main thread when processing large lists of data.  A low priority event is
+  used, repetitive operations are spread across multiple event callbacks and
+  LevelDB writes are batched.
+*/
+
 
 #include "TimeseriesHandler.h"
 
@@ -44,6 +75,10 @@ using namespace std;
 using namespace cb;
 using namespace cb::API;
 
+#undef CBANG_LOG_PREFIX
+#define CBANG_LOG_PREFIX "TS:" << name << ":"
+#define TIME_FMT "%Y%m%d%H%M%S"
+
 
 TimeseriesHandler::TimeseriesHandler(
   API &api, const string &name, const JSON::ValuePtr &config) :
@@ -76,9 +111,9 @@ TimeseriesHandler::TimeseriesHandler(
     key = config->createList(); // Empty key
   }
 
-  event->setPriority(4);
+  event->setPriority(7);
 
-  load();
+  schedule();
 }
 
 
@@ -98,7 +133,7 @@ double TimeseriesHandler::getNext() const {
   auto   now  = Timer::now();
   double next = getTimePeriod(ceil(now)) + period - now;
 
-  LOG_DEBUG(5, "Querying timeseries in " << next << " seconds "
+  LOG_DEBUG(4, "Querying in " << next << " seconds "
     << " current=" << Time(getTimePeriod(ceil(now))).toString()
     << " period="  << period
     << " now="     << Time(now).toString());
@@ -114,95 +149,76 @@ SmartPointer<Timeseries> TimeseriesHandler::get(
   if (it != series.end()) return it->second;
   if (!create) return 0;
 
-  LOG_DEBUG(4, "Adding `" << name << "` timeseries with key `" << key << "`");
+  LOG_DEBUG(5, "Adding key `" << key << "`");
   return series[key] = new Timeseries(*this, key);
 }
 
 
-
-void TimeseriesHandler::action(const CtxPtr &ctx) {
-  // Resolve variables
-  auto resolver = ctx->getResolver();
-  auto action   = resolver->selectString("args.action", "query");
-  auto since    = resolver->selectTime("args.since", 0);
-  auto maxCount = resolver->selectU64("args.max_count", 0);
-  auto key      = resolveKey(*resolver->select("args"));
-
-  // Get Timeseries
-  auto ts = get(key);
-  if (ts.isNull() && action == "unsubscribe") return;
-
-  // Execute action
-  auto cb = [this, ctx] (
-    const SmartPointer<Exception> &err, const JSON::ValuePtr &data) {
-      if (err.isNull()) ctx->reply(data);
-      else ctx->reply(*err);
-    };
-
-  if (action == "query") return ts->query(since, maxCount, cb);
-
-  auto ws = ctx->getWebsocket();
-  if (ws.isSet())  {
-    if (action == "subscribe")   return ws->subscribe(*ts, since, maxCount, cb);
-    if (action == "unsubscribe") return ws->unsubscribe(*ts);
-  }
-
-  THROW("Unrecognized timeseries action '" << action << "'");
+void TimeseriesHandler::schedule() {
+  if (last.isNull() || results.isSet()) event->add(0);
+  else event->add(getNext());
 }
 
 
-void TimeseriesHandler::load() {schedule();}
-
-
-void TimeseriesHandler::query(uint64_t time) {
-  auto cb = [=] (HTTP::Status status, const JSON::ValuePtr &results) {
-    if (status == HTTP::Status::HTTP_OK) try {
-      if (ret != "list") get("")->add(time, results);
-      else {
-        LOG_DEBUG(3, "Storing timeseries results " << name);
-        this->results = results;
-        resultsTime   = time;
-        it            = results->begin();
-      }
-    } CATCH_ERROR;
+void TimeseriesHandler::process() {
+  if (last.isNull()) {
+    // Try to load the last value
+    LOG_DEBUG(3, "Querying last value");
 
-    schedule();
-  };
+    auto cb = [this] (bool success, const string &value) {
+      LOG_DEBUG(3, "Last value success=" << (success ? "true" : "false"));
 
-  LOG_DEBUG(3, "Querying timeseries " << name);
-  LOG_DEBUG(5, "Timeseries query: " << sql);
-
-  query(sql, cb);
-}
+      last = new JSON::Dict;
+      schedule();
 
+      try {
+        if (success) last = JSON::Reader::parse(value);
+      } catch (const Exception &e) {
+        LOG_ERROR("Failed to parse last timeseries result: " << e);
+      }
+    };
 
-void TimeseriesHandler::schedule() {
-  if (results.isSet()) event->add(0);
-  else if (!event->isPending()) event->add(getNext());
-}
+    db.get("\0"s, cb);
 
+  } else if (results.isSet()) {
+    // Process a batch of results
+    LOG_DEBUG(4, "Processing result batch");
 
-void TimeseriesHandler::process() {
-  if (results.isSet()) {
     try {
       for (unsigned i = 0; i < 1000; i++) {
         if (it == results->end()) {
           LOG_DEBUG(3, "Done " << name);
+          if (batch.isSet()) {
+            auto cb = [=] () {
+              batch->set("\0"s, last->toString());
+              batch->commit();
+            };
+
+            batch.release();
+            db.getPool()->submit(0, cb);
+          }
           results.release();
           break;
         }
 
         auto result = *it++;
         string key  = resolveKey(*result);
 
-        // Don't bother storing key data
+        // Don't store key data
         for (auto it: *this->key)
           result->erase(it->asString());
 
         // If there's only one key store just its value
         if (result->size() == 1) result = *result->begin();
 
-        get(key)->add(resultsTime, result);
+        // Don't record if result is unchanged
+        auto it = last->find(key);
+        if (it != last->end() && **it == *result) continue;
+        last->insert(key, result);
+
+        if (batch.isNull()) batch = new LevelDB::Batch(db.batch());
+        batch->set(key + "\0"s + resultsTimeKey, result->toString());
+        get(key)->broadcast(resultsTime, result);
       }
     } catch (const Exception &e) {
       LOG_ERROR(e);
@@ -212,13 +228,75 @@ void TimeseriesHandler::process() {
 
     schedule();
 
-  } else query(getTimePeriod(Time::now()));
+  } else query(getTimePeriod(Time::now())); // Load next timeseries result
+}
+
+
+void TimeseriesHandler::query(uint64_t time) {
+  auto cb = [=] (HTTP::Status status, const JSON::ValuePtr &results) {
+    if (status == HTTP::Status::HTTP_OK) try {
+      resultsTimeKey = Time(getTimePeriod(time)).toString(TIME_FMT);
+
+      if (ret != "list") {
+        if (*last != *results) {
+          last = results;
+          db.set("\0"s, last->toString());
+          db.set("\0"s + resultsTimeKey, results->toString());
+          get("")->broadcast(time, results);
+        }
+
+      } else {
+        LOG_DEBUG(3, "Storing results");
+        this->results  = results;
+        resultsTime    = time;
+        it             = results->begin();
+      }
+    } CATCH_ERROR;
+
+    schedule();
+  };
+
+  LOG_DEBUG(3, "Querying");
+  LOG_DEBUG(5, "query: " << sql);
+
+  query(sql, cb);
+}
+
+
+void TimeseriesHandler::action(const CtxPtr &ctx) {
+  // Resolve variables
+  auto resolver = ctx->getResolver();
+  auto action   = resolver->selectString("args.action", "query");
+  auto since    = resolver->selectTime("args.since", 0);
+  auto maxCount = resolver->selectU64("args.max_count", 0);
+
+  // Get Timeseries
+  auto key = ret == "list" ? resolveKey(*resolver->select("args")) : "";
+  auto ts  = get(key, action != "unsubscribe");
+  if (ts.isNull()) return;
+
+  // Execute action
+  auto cb = [this, ctx] (
+    const SmartPointer<Exception> &err, const JSON::ValuePtr &data) {
+      if (err.isNull()) ctx->reply(data);
+      else ctx->reply(*err);
+    };
+
+  if (action == "query") return ts->query(since, maxCount, cb);
+
+  auto ws = ctx->getWebsocket();
+  if (ws.isSet())  {
+    if (action == "subscribe")   return ws->subscribe(*ts, since, maxCount, cb);
+    if (action == "unsubscribe") return ws->unsubscribe(*ts);
+  }
+
+  THROW("Unrecognized timeseries action '" << action << "'");
 }
 
 
 SmartPointer<MariaDB::EventDB> TimeseriesHandler::getDBConnection() const {
   auto db = QueryDef::getDBConnection();
-  db->setPriority(5); // Lower priority to avoid blocking regular API requests
+  db->setPriority(8); // Lower priority to avoid blocking regular API requests
   return db;
 }
 

src/cbang/api/handler/TimeseriesHandler.h

@@ -52,27 +52,30 @@ namespace cb {
 
       std::map<std::string, SmartPointer<Timeseries>> series;
 
-      JSON::ValuePtr results;
+      JSON::ValuePtr        last;
+      JSON::ValuePtr        results;
       JSON::Value::iterator it;
-      uint64_t resultsTime = 0;
+      uint64_t              resultsTime = 0;
+      std::string           resultsTimeKey;
+      SmartPointer<LevelDB::Batch> batch;
 
       TimeseriesHandler(
         API &api, const std::string &name, const JSON::ValuePtr &config);
 
       const std::string &getName() const {return name;}
+
+    protected:
       std::string resolveKey(const JSON::Value &dict) const;
       uint64_t getTimePeriod(uint64_t ts) const {return (ts / period) * period;}
       double getNext() const;
-
-    protected:
       SmartPointer<Timeseries> get(const std::string &key, bool create = true);
-      void action(const CtxPtr &ctx);
-      void load();
+
       void schedule();
-      using QueryDef::query;
-      void query(uint64_t time);
       void process();
 
+      using QueryDef::query;
+      void query(uint64_t time);
+      void action(const CtxPtr &ctx);
 
       // From QueryDef
       SmartPointer<MariaDB::EventDB> getDBConnection() const override;