307 lines
12 KiB
C++
Executable File
307 lines
12 KiB
C++
Executable File
/**
|
|
* 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
|