Initial commit

Lts/Services/RecordingManagement/db_writer_snapshot.cpp

@@ -0,0 +1,306 @@
+/**
+  * DbWriterSnapshot for RecordingManagement
+  * $Id$
+  * $HeadURL$
+  * $Author$
+  */
+#include "db_writer_snapshot.h"
+
+#include "logger.h"
+
+#include <QSqlError>
+#include <QTimer>
+#include <QVariant>
+
+#include <nglsettings/nglsettings_settings.h>
+#include <ngldb/dbtransaction.h>
+#include <ngldb/nsqlquery.h>
+
+#include <PluginInterfaces/record_replay_plugin.h>
+#include <RecordingManagement/rec_db_impl.h>
+
+namespace RecordingManagement
+{
+
+void DbWriterSnapshotThread::run()
+{
+  if (!m_dbWriterSnapshot)
+    return;
+  m_dbWriterSnapshot->init();
+  // make sure m_dbWriterSnapshot lives in our thread
+  Q_ASSERT(m_dbWriterSnapshot->thread() == this);
+  exec();
+  m_dbWriterSnapshot->cleanup();
+}
+
+DbWriterSnapshot::DbWriterSnapshot(const QVector<RecordReplay::RecordReplayPlugin *> &plugins,
+                                   int exerciseId,
+                                   NGL::Settings *settings)
+  : QObject()
+  , m_plugins(plugins)
+  , m_snapshotInterval(1 * 60 * 1000) // TODO: make config variable   // Interval: 1 min in milliseconds
+  , m_exerciseId(exerciseId)
+  , m_settings(settings)
+  , m_logger(getLogger("DbWriterSnapshot"))
+{
+}
+
+DbWriterSnapshot::~DbWriterSnapshot()
+{
+  cleanup();  // should do nothing because cleanup() should be called from within the thread
+}
+
+void DbWriterSnapshot::init()
+{
+  m_dbInterface = DatabaseInterface::create(m_exerciseId);
+
+  QSqlDatabase dbConn = QSqlDatabase::addDatabase(m_settings->databaseDriver(),
+                                                  dbConnectionName());
+  dbConn.setHostName(m_settings->databaseHostname());
+  dbConn.setDatabaseName(m_settings->databaseName());
+  dbConn.setUserName(m_dbInterface->dbUser());
+  dbConn.setPassword(m_settings->databasePassword());
+
+  if (!dbConn.open())
+  {
+    FATAL_LOG(m_logger) <<  "Error opening database " << m_settings->databaseHostname()
+                         <<  ":" << m_settings->databaseName()
+                          << " as user " << m_dbInterface->dbUser();
+    FATAL_LOG(m_logger) << dbConn.lastError().driverText() << " "
+                        << dbConn.lastError().databaseText() << " "
+                        << dbConn.lastError().nativeErrorCode();
+    m_dbInterface = nullptr;
+    return;
+  }
+
+
+  // Set search path
+  const QString stmt = QLatin1String("SET SEARCH_PATH TO ") + m_dbInterface->schemaName() + QLatin1String(", PUBLIC");
+  INFO_LOG(m_logger) << "SearchPath Stmt: " << stmt;
+  NSqlQuery q(dbConn);
+
+  q.exec(stmt);
+
+  if (!m_dbInterface->init(dbConn))
+  {
+    ERROR_LOG(m_logger) << "Error initializing database interface - can not initialize DbWriterSnapshot";
+    m_dbInterface = nullptr;
+    return;
+  }
+
+  QTimer::singleShot(0, Qt::VeryCoarseTimer,
+                     this, &DbWriterSnapshot::initSnapshot);
+}
+
+void DbWriterSnapshot::cleanup()
+{
+  m_dbInterface = nullptr;
+  QSqlDatabase::database(dbConnectionName()).close();
+}
+
+void DbWriterSnapshot::setMinRecordedSampleTimes(const QHash<QString, qint64> &minRecordedSampleTimes)
+{
+  QMutexLocker l(&m_minRecordedSampleTimesMutex);
+  for(auto it = minRecordedSampleTimes.begin(); it != minRecordedSampleTimes.end(); ++it)
+  {
+    const qint64 minRecordedSampleTime = it.value();
+    const QString &topicName = it.key();
+    if (minRecordedSampleTime == 0)
+    {
+      WARNING_LOG(m_logger) << "Minimum recorded Sample Time for topic " << topicName <<
+                               " is zero - looks like a service did not properly set the sample time!";
+      continue;
+    }
+    auto mrstIt = m_minRecordedSampleTimes.constFind(topicName);
+    if (mrstIt == m_minRecordedSampleTimes.constEnd())
+    {
+      // no data until now - just take minRecordedSampleTime
+      m_minRecordedSampleTimes[topicName] = minRecordedSampleTime;
+    }
+    else
+    {
+      // already a minimum time - make sure to get the minimum from both
+      m_minRecordedSampleTimes[topicName] = std::min(minRecordedSampleTime, mrstIt.value());
+    }
+    INFO_LOG(m_logger) << "Current minimum sample time for " << topicName
+                       << " is " << m_minRecordedSampleTimes.value(topicName)
+                       << " = " << QDateTime::fromMSecsSinceEpoch(m_minRecordedSampleTimes.value(topicName)).toString();
+  }
+}
+
+
+QHash<QString, qint64> DbWriterSnapshot::getMinRecordedSampleTimes(qint64 getUntil)
+{
+  QMutexLocker l(&m_minRecordedSampleTimesMutex);
+  QHash<QString, qint64> ret;
+  // we can't remove them from m_minRecordedSampleTimes because
+  // there can be samples between getUntil and now which would not get indexed
+  // if there's no m_minRecordedSampleTimes entry for this topic
+  for (auto it = m_minRecordedSampleTimes.begin() ; it != m_minRecordedSampleTimes.end(); ++it)
+  {
+    if (it.value() < getUntil)
+    {
+      // return for recording
+      ret[it.key()] = it.value();
+      // and update minimum rec timestamp
+      *it = getUntil;
+    }
+  }
+  return ret;
+}
+
+void DbWriterSnapshot::initSnapshot()
+{
+  NSqlQuery minRecSampleTimeQuery(QSqlDatabase::database(dbConnectionName()));
+  NSqlQuery topicRecIdQuery(QSqlDatabase::database(dbConnectionName()));
+
+  topicRecIdQuery.prepare(QStringLiteral("SELECT MAX(time_stamp) FROM rec_snapshot WHERE svc_name = :svc_name AND topic_name = :topic_name"));
+
+  const qint64 tNow(time(nullptr) * 1000 - 60 * 1 * 1000); // jetzt vor einer Minute
+  INFO_LOG(m_logger) << "tNow (-1min): " << tNow;
+
+  QHash<QString, qint64> minRecordedSampleTimes;
+
+  // Recording already initialised => read last arrays for each plugin/topic to
+  // get a hint where to start with indexing each topic
+  for (RecordReplay::RecordReplayPlugin *recPlugin : qAsConst(m_plugins))
+  {
+    // get all record topics and check if there's a index for them
+    // if not -> check if there's a sample for this topic where we've to start indexing
+    const QVector<QString> recordTopicNames = recPlugin->getRecordTopicNames();
+
+    topicRecIdQuery.bindValue(QStringLiteral(":svc_name"), recPlugin->svcShortName());
+    for (const QString &topicName : recordTopicNames)
+    {
+      qint64 timeToStartSnapshotCreationFrom = 0;
+      bool bHasSnapshot = false;
+      topicRecIdQuery.bindValue(QStringLiteral(":topic_name"), topicName);
+      if (topicRecIdQuery.execAndFetch() && !topicRecIdQuery.value(0).isNull())
+      {
+        // we found a snapshot -> assume that all data until this point is properly indexed
+        // so next snapshot must be created at t + m_snapshotInterval
+        timeToStartSnapshotCreationFrom = topicRecIdQuery.value(0).toLongLong() + m_snapshotInterval;
+        bHasSnapshot = true;
+      }
+      else
+      {
+        // no snapshot for this topic. This may be because
+        // - no data until now for this topic
+        // - the index should be rebuild (== rec_snapshot entries for this topic were removed)
+        // -> check if data is available
+
+        // tNow -> default value when no data is available until now
+        timeToStartSnapshotCreationFrom = recPlugin->getMinimumSampleTime(minRecSampleTimeQuery, topicName, tNow);
+      }
+
+      const char *fromWhere = "last sample";
+      if (bHasSnapshot)
+        fromWhere = "snapshot + last sample";
+      else if (timeToStartSnapshotCreationFrom == tNow)
+        fromWhere = "current time";
+      INFO_LOG(m_logger) << "last snapshot time for topic '" << topicName
+                         << "' is: " << timeToStartSnapshotCreationFrom
+                         << " from " << fromWhere;
+      minRecordedSampleTimes[topicName] = timeToStartSnapshotCreationFrom;
+    }
+  }
+  setMinRecordedSampleTimes(minRecordedSampleTimes);
+  writeSnapshotData();
+}
+
+void DbWriterSnapshot::writeSnapshotData()
+{
+  const qint64 endTime(QDateTime::currentMSecsSinceEpoch() - 60 * 1 * 1000); // jetzt vor einer Minute
+
+  // calculate last snapshot time to create during this run
+  qint64 lastSnapshotTimeToGenerateNow = endTime - (endTime % m_snapshotInterval);
+
+  const QHash<QString, qint64> minRecSampleTimes(getMinRecordedSampleTimes(lastSnapshotTimeToGenerateNow));
+  if (minRecSampleTimes.isEmpty())
+  {
+    INFO_LOG(m_logger) << "minRecSampleTimes has no entries with time < " << lastSnapshotTimeToGenerateNow
+                       << " nothing to do...";
+    QTimer::singleShot(m_snapshotInterval, Qt::VeryCoarseTimer,
+                       this, &DbWriterSnapshot::writeSnapshotData);
+    return;
+  }
+
+  // shortcut to avoid DBTransaction and loop over all plugins
+  qint64 startTime = std::numeric_limits<qint64>::max();
+  for(const qint64 st : qAsConst(minRecSampleTimes))
+    startTime = std::min(startTime, st);
+  INFO_LOG(m_logger) << "writeSnapshotData - lastSnapshotTimeToGenerateNow: " << lastSnapshotTimeToGenerateNow
+                     << ", minRecTime is " << startTime;
+  if (startTime <= 0 || startTime > endTime)
+  {
+    WARNING_LOG(m_logger) << "Spurious start time (" << startTime
+                          << ") - using end time for start (" << endTime << ")";
+    startTime = endTime - 1;  // -1 -> create at least one interval
+  }
+
+  if (startTime > lastSnapshotTimeToGenerateNow)
+  {
+    INFO_LOG(m_logger) << "Nothing changed in the last interval until " << lastSnapshotTimeToGenerateNow << " youngest topic is at " << startTime;
+    QTimer::singleShot(m_snapshotInterval, Qt::VeryCoarseTimer,
+                       this, &DbWriterSnapshot::writeSnapshotData);
+    return;
+  }
+
+  // we know something changed between startTime and lastSnapshotTimeToGenerateNow - let's do some work
+  INFO_LOG(m_logger) << "startTime: " << startTime
+                     << " lastSnapshotTimeToGenerateNow: " << lastSnapshotTimeToGenerateNow << " - BEGIN";
+  DBTransaction trans(QSqlDatabase::database(dbConnectionName()));
+  NSqlQuery recSnapshotQuery(QSqlDatabase::database(dbConnectionName()));
+  for (RecordReplay::RecordReplayPlugin *recPlugin : qAsConst(m_plugins))
+    writeSnapshotDataPlugin(lastSnapshotTimeToGenerateNow, recSnapshotQuery, recPlugin, minRecSampleTimes);
+
+  trans.commit();
+  INFO_LOG(m_logger) << "startTime: " << startTime
+                     << " lastSnapshotTimeToGenerateNow: " << lastSnapshotTimeToGenerateNow << " - FINISHED";
+
+
+  QTimer::singleShot(m_snapshotInterval, Qt::VeryCoarseTimer,
+                     this, &DbWriterSnapshot::writeSnapshotData);
+}
+
+void DbWriterSnapshot::writeSnapshotDataPlugin(qint64 endInterval,
+                                               NSqlQuery &recSnapshotQuery,
+                                               RecordReplay::RecordReplayPlugin *recPlugin,
+                                               const QHash<QString, qint64> &minRecordedSampleTimes)
+{
+  RecordingManagement::Database::RecSnapshot recSnapshot;
+  recSnapshot.svcName = recPlugin->svcShortName();
+
+  auto snapshots =
+    recPlugin->getSnapshotTimes(recSnapshotQuery,
+                                endInterval,
+                                m_snapshotInterval,
+                                minRecordedSampleTimes);
+
+  for (auto &it : snapshots)
+  {
+    const QString &topicName = it.first;
+    auto &vec = it.second;
+
+    recSnapshot.topicName = topicName;
+    recSnapshot.timeStamp = vec.at(0).timeStamp;  // vec is sorted by timeStamp
+    m_dbInterface->removeNewerEqual(recSnapshot);
+
+    for (auto &snapshotData : vec)
+    {
+      INFO_LOG(m_logger) << "svc: " << recSnapshot.svcName
+                         << " topic: " << recSnapshot.topicName
+                         << " timestamp: " << snapshotData.timeStamp
+                         << " #elements: " << snapshotData.recIdArray.size();
+
+      recSnapshot.id = 0; // to be inserted
+      recSnapshot.timeStamp = snapshotData.timeStamp;
+      recSnapshot.recIdArray = LtsUtils::BigintArray2D(std::move(snapshotData.recIdArray));
+
+      if (!m_dbInterface->insert(recSnapshot))
+        ERROR_LOG(m_logger) << "Error while recording snaphshot for topic " << recSnapshot.topicName;
+    }
+  }
+}
+
+}  // RecordingManagement