datalogger.cpp 44 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384
  1. #include <string.h>
  2. #include <cmath>
  3. #include <malloc.h>
  4. #include <limits.h>
  5. #include <signal.h>
  6. #include <sys/stat.h>
  7. #include "fileutil.h"
  8. #include "strutil.h"
  9. #include "datalogger.h"
  10. #include "processclock.h"
  11. #include "debug.h"
  12. /////////////////////////////////////////////////////////////////////////////
  13. #define _USE_MODIFIED_INDEX 1
  14. #define _TRACK_TIME 1
  15. #define _FILE_SIZE_DELETE_MARGIN_PERCENT 10
  16. #define _SIZE_GUARD_TIMESTAMP_FILE_NAME "sguard.ts"
  17. #define _INVALID_TIMESTAMP_VALUE ((time_t)-1)
  18. /////////////////////////////////////////////////////////////////////////////
  19. inline static bool _IsLoggable(double d)
  20. {
  21. int c = std::fpclassify(d);
  22. return (c == FP_NORMAL) || (c == FP_ZERO);
  23. }
  24. /////////////////////////////////////////////////////////////////////////////
  25. CDataLogger::CDataLogger(LPCDLPARAMS pdlp, CLogfile &rlf) : m_lf(rlf),
  26. m_tidSGThread(0),
  27. m_bSGHasSizeLimitPrerequisites(false),
  28. m_bSGInProgress(false),
  29. m_bSGConfigured(false),
  30. m_nSGCurPassUTC(0),
  31. m_nSGLastPassUTC(0),
  32. m_condmtx1(PTHREAD_MUTEX_INITIALIZER),
  33. m_cond1(PTHREAD_COND_INITIALIZER),
  34. m_nLastLogTimestamp(0),
  35. m_bBadDateLogsDetected(false)
  36. {
  37. memset(&m_dlp, 0, sizeof(m_dlp));
  38. memset(&m_gv, 0, sizeof(m_gv));
  39. memset(&m_mtx, 0, sizeof(m_mtx));
  40. memset(&m_mutexAttr, 0, sizeof(m_mutexAttr));
  41. if(pdlp)
  42. memcpy(&m_dlp, pdlp, sizeof(m_dlp));
  43. if(!(m_bSGConfigured = !!m_dlp.nMaxSize || !!m_dlp.nMaxAge))
  44. m_lf.Error("Size guard not configured.\n");
  45. if(m_dlp.pszBaseDir)
  46. strncpy(m_szAppDir, m_dlp.pszBaseDir, sizeof(m_szAppDir) - 1);
  47. else
  48. ::GetAppDirectory(m_szAppDir, sizeof(m_szAppDir));
  49. ::pthread_mutexattr_init(&m_mutexAttr);
  50. ::pthread_mutexattr_settype(&m_mutexAttr, PTHREAD_MUTEX_RECURSIVE);
  51. ::pthread_mutex_init(&m_mtx, &m_mutexAttr);
  52. }
  53. /////////////////////////////////////////////////////////////////////////////
  54. CDataLogger::~CDataLogger(void)
  55. {
  56. Release();
  57. ::pthread_mutex_destroy(&m_mtx);
  58. ::pthread_mutexattr_destroy(&m_mutexAttr);
  59. ::pthread_cond_destroy(&m_cond1);
  60. }
  61. /////////////////////////////////////////////////////////////////////////////
  62. void CDataLogger::Release(void)
  63. {
  64. if(m_tidSGThread)
  65. {
  66. ::pthread_mutex_lock(&m_condmtx1);
  67. bool bSgInProg = m_bSGInProgress;
  68. ::pthread_mutex_unlock(&m_condmtx1);
  69. if(bSgInProg)
  70. CLogfile::StdErr("\nDatalogger is terminating. This may take some time ...\nPlease wait, before powering off the device!\n");
  71. Lock();
  72. m_lf.Lock();
  73. ::pthread_cancel(m_tidSGThread);
  74. m_lf.Unlock();
  75. Unlock();
  76. ::pthread_join(m_tidSGThread, NULL);
  77. m_tidSGThread = 0;
  78. }
  79. }
  80. /////////////////////////////////////////////////////////////////////////////
  81. unsigned long CDataLogger::GetTagID(const char *pszVarPath, int nDataType, int nLogType)
  82. {
  83. CMySqlDB db;
  84. const char *pszID = NULL;
  85. char *pszEndptr;
  86. unsigned long nID = ULONG_MAX;
  87. std::string sSql;
  88. sSql = formatString("select `tagid` from `%s` where `path` = '%s' and `dataType` = %d and `logType` = %d", m_dlp.szTagsTable, pszVarPath, nDataType, nLogType);
  89. if(!db.Connect("localhost", m_dlp.szDBUser, m_dlp.szDBPass, m_dlp.szDBName))
  90. {
  91. m_lf.Error("CDataLogger::GetTagID: DB Error: %s\n", db.LastError().c_str());
  92. return nID;
  93. }
  94. CMySqlResult res = db.Query(sSql.c_str());
  95. bool bError = res.error();
  96. if(bError)
  97. m_lf.Error("CDataLogger::GetTagID: DB Error: %s\n", db.LastError().c_str());
  98. else
  99. {
  100. my_ulonglong nCount = res.RowCount();
  101. if(nCount > 0)
  102. {
  103. MYSQL_ROW pRow = res.FetchRow();
  104. pszID = pRow[0];
  105. nID = strtoul(pszID, &pszEndptr, 10);
  106. }
  107. else
  108. {
  109. sSql = formatString("insert into `%s` (`dataType`, `logType`, `path`) values(%d, %d, '%s')", m_dlp.szTagsTable, nDataType, nLogType, pszVarPath);
  110. res = db.Query(sSql.c_str());
  111. bError = res.error();
  112. if(bError)
  113. m_lf.Error("DB Error: %s\n", db.LastError().c_str());
  114. else
  115. {
  116. sSql = formatString("select `tagid` from `%s` where `path`='%s'", m_dlp.szTagsTable, pszVarPath);
  117. CMySqlResult res = db.Query(sSql.c_str());
  118. bool bError = res.error();
  119. if(bError)
  120. m_lf.Error("DB Error: %s\n", db.LastError().c_str());
  121. else if(res.RowCount() > 0)
  122. {
  123. MYSQL_ROW pRow = res.FetchRow();
  124. pszID = pRow[0];
  125. nID = strtoul(pszID, &pszEndptr, 10);
  126. }
  127. }
  128. }
  129. }
  130. return nID;
  131. }
  132. /////////////////////////////////////////////////////////////////////////////
  133. bool CDataLogger::TableFileExists(const char *pszTableName)
  134. {
  135. char szDataFile[PATH_MAX];
  136. sprintf(szDataFile, "%s%s/%s.ibd", m_gv.szDataDir, m_dlp.szDBName, pszTableName);
  137. return !!FileExist(szDataFile);
  138. }
  139. /////////////////////////////////////////////////////////////////////////////
  140. int64_t CDataLogger::TableFileSize(const char *pszTableName)
  141. {
  142. struct stat statbuf;
  143. char szDataFile[PATH_MAX];
  144. sprintf(szDataFile, "%s%s/%s.ibd", m_gv.szDataDir, m_dlp.szDBName, pszTableName);
  145. if(stat(szDataFile, &statbuf) == -1)
  146. return 0;
  147. return (int64_t)statbuf.st_size;
  148. }
  149. /////////////////////////////////////////////////////////////////////////////
  150. bool CDataLogger::QueryServerVariable(CMySqlDB &rdb, const char *pszVarname, CMySqlVar &val)
  151. {
  152. if(!pszVarname || !*pszVarname)
  153. {
  154. m_lf.Error("CDataLogger::QueryServerVariable: No Variable name!\n");
  155. return false;
  156. }
  157. bool bError;
  158. std::string sSql;
  159. sSql = formatString("select @@%s;", pszVarname);
  160. CMySqlResult res = rdb.Query(sSql.c_str());
  161. bError = res.error();
  162. if(!bError)
  163. {
  164. bError = true;
  165. do
  166. {
  167. my_ulonglong nRowCount = res.RowCount();
  168. unsigned int nFldCount = res.FieldCount();
  169. const MYSQL_FIELD *pFields = res.FetchFields();
  170. if(nRowCount != 1 || nFldCount != 1 || !pFields)
  171. break;
  172. MYSQL_ROW pRow = res.FetchRow();
  173. if(!pRow || !*pRow || !**pRow)
  174. break;
  175. bError = !val.FromField(pFields[0], pRow[0]);
  176. }
  177. while(false);
  178. }
  179. return !bError;
  180. }
  181. /////////////////////////////////////////////////////////////////////////////
  182. #if 0
  183. bool CDataLogger::QueryStatusVariable(CMySqlDB &rdb, const char *pszVarname, CMySqlVar &val)
  184. {
  185. if(!pszVarname || !*pszVarname)
  186. {
  187. m_lf.Error("CDataLogger::QueryStatusVariable: No Variable name!\n");
  188. return false;
  189. }
  190. bool bError;
  191. std::string sSql;
  192. sSql = formatString("show session status like '%s'", pszVarname);
  193. CMySqlResult res = rdb.Query(sSql.c_str());
  194. bError = res.error();
  195. if(!bError)
  196. {
  197. bError = true;
  198. do
  199. {
  200. my_ulonglong nRowCount = res.RowCount();
  201. unsigned int nFldCount = res.FieldCount();
  202. const MYSQL_FIELD *pFields = res.FetchFields();
  203. if(nRowCount != 1 || nFldCount != 2 || !pFields)
  204. break;
  205. MYSQL_ROW pRow = res.FetchRow();
  206. if(!pRow || !pRow[1] || !*pRow[1])
  207. break;
  208. bError = !val.FromField(pFields[1], pRow[1]);
  209. }
  210. while(false);
  211. }
  212. return !bError;
  213. }
  214. #endif
  215. /////////////////////////////////////////////////////////////////////////////
  216. bool CDataLogger::ReadGlobalOptions(CMySqlDB &rdb)
  217. {
  218. CMySqlVar val;
  219. memset(&m_gv, 0, sizeof(m_gv));
  220. if(QueryServerVariable(rdb, "datadir", val))
  221. {
  222. memset(m_gv.szDataDir, 0, sizeof(m_gv.szDataDir));
  223. if(!val.IsNull())
  224. val.CopyStrVal(m_gv.szDataDir, sizeof(m_gv.szDataDir) - 1, 0);
  225. }
  226. if(QueryServerVariable(rdb, "innodb_file_per_table", val))
  227. {
  228. if(!val.IsNull())
  229. m_gv.bInnoDbFilePerTable = val;
  230. }
  231. if(QueryServerVariable(rdb, "innodb_strict_mode", val))
  232. {
  233. if(!val.IsNull())
  234. m_gv.bInnoDbIsStrictMode = val;
  235. }
  236. if(QueryServerVariable(rdb, "innodb_file_format", val))
  237. {
  238. memset(m_gv.szInnoDbFileFormat, 0, sizeof(m_gv.szInnoDbFileFormat));
  239. if(!val.IsNull())
  240. {
  241. val.CopyStrVal(m_gv.szInnoDbFileFormat, sizeof(m_gv.szInnoDbFileFormat) - 1, 0);
  242. m_gv.bInnoDbIsBarracuda = !strcasecmp(m_gv.szInnoDbFileFormat, "barracuda");
  243. }
  244. }
  245. return true;
  246. }
  247. /////////////////////////////////////////////////////////////////////////////
  248. void CDataLogger::SizeGuardTrigger(time_t ts)
  249. {
  250. if(m_bSGConfigured && m_tidSGThread)
  251. {
  252. ::pthread_mutex_lock(&m_condmtx1);
  253. if( !m_bSGInProgress &&
  254. !SizeGuardDayWorkDone(ts))
  255. {
  256. unsigned long long nMnUTC = (unsigned long long)_MIDNIGHT_TIMESTAMP_UTC(ts);
  257. if(SizeGuardLastPassRead() != nMnUTC)
  258. {
  259. m_nSGCurPassUTC = ts;
  260. ::pthread_cond_signal(&m_cond1);
  261. }
  262. else
  263. m_nSGLastPassUTC = nMnUTC;
  264. }
  265. ::pthread_mutex_unlock(&m_condmtx1);
  266. }
  267. }
  268. /////////////////////////////////////////////////////////////////////////////
  269. #if 0
  270. bool CDataLogger::QueryTableSizes(CMySqlDB &rdb, const char *pszTableName)
  271. {
  272. bool bError;
  273. char *pszEndptr;
  274. MYSQL_ROW pRow;
  275. uint64_t nDataLength = 0, nDataFree = 0, nIndexLength = 0, nFreeExtents, nTotalExtents;
  276. std::string sSql;
  277. sSql = formatString("select `DATA_LENGTH`, `INDEX_LENGTH`, `DATA_FREE` from `information_schema`.`TABLES` where (`TABLE_SCHEMA` = '%s') AND (`TABLE_NAME` = '%s');", m_dlp.szDBName, pszTableName);
  278. CMySqlResult res = rdb.Query(sSql.c_str());
  279. bError = res.error();
  280. if(bError)
  281. {
  282. m_lf.Error("CDataLogger::QueryTableSizes: DB Error: %s\n", rdb.LastError().c_str());
  283. return false;
  284. }
  285. pRow = res.FetchRow();
  286. nDataLength = strtoull(pRow[0], &pszEndptr, 10);
  287. nIndexLength = strtoull(pRow[1], &pszEndptr, 10);
  288. nDataFree = strtoull(pRow[2], &pszEndptr, 10);
  289. //////////////////////////////////////////////////////////////////////////////////////////////
  290. sSql = formatString("select `FREE_EXTENTS`, `TOTAL_EXTENTS`, `DATA_FREE` from `information_schema`.`FILES` where `FILE_NAME` like '%%demo/%s.ibd';", pszTableName);
  291. CMySqlResult res2 = rdb.Query(sSql.c_str());
  292. bError = res2.error();
  293. if(bError)
  294. {
  295. m_lf.Error("CDataLogger::QueryTableSizes: DB Error: %s\n", rdb.LastError().c_str());
  296. return false;
  297. }
  298. pRow = res2.FetchRow();
  299. nFreeExtents = strtoull(pRow[0], &pszEndptr, 10);
  300. nTotalExtents = strtoull(pRow[1], &pszEndptr, 10);
  301. int64_t nFileSize = TableFileSize(pszTableName);
  302. m_lf.Info("Data length: %ju, Data free: %ju, Total extents: %ju, Free extents: %ju, File size: %ju.\n", nDataLength + nIndexLength, nDataFree, nTotalExtents, nFreeExtents, nFileSize);
  303. TRACE("Data length: %ju, Data free: %ju, Total extents: %ju, Free extents: %ju, File size: %ju.\n", nDataLength + nIndexLength, nDataFree, nTotalExtents, nFreeExtents, nFileSize);
  304. return true;
  305. }
  306. #endif
  307. /////////////////////////////////////////////////////////////////////////////
  308. bool CDataLogger::CheckTable(CMySqlDB &rdb, const char *pszTableName, bool &bExists, bool &bUseResortTable, bool &bIsInnoDB)
  309. {
  310. bool bError;
  311. std::string sSql;
  312. const char *pszEngine = NULL;
  313. uint64_t nDataLength = 0, nIndexLength = 0;
  314. int64_t nFileSize = -1;
  315. bExists = bUseResortTable = false;
  316. bIsInnoDB = false;
  317. sSql = formatString("select `ENGINE`, `DATA_LENGTH`, `INDEX_LENGTH` from `information_schema`.`TABLES` where (`TABLE_SCHEMA` = '%s') AND (`TABLE_NAME` = '%s')", m_dlp.szDBName, pszTableName);
  318. CMySqlResult res1 = rdb.Query(sSql.c_str());
  319. bError = res1.error();
  320. if(bError)
  321. {
  322. m_lf.Error("CDataLogger::CheckTables: DB Error: %s\n", rdb.LastError().c_str());
  323. return false;
  324. }
  325. else
  326. {
  327. my_ulonglong nCount = res1.RowCount();
  328. if(nCount > 0)
  329. {
  330. MYSQL_ROW pRow = res1.FetchRow();
  331. unsigned int nFc = res1.FieldCount();
  332. const MYSQL_FIELD *pFields = res1.FetchFields();
  333. CMySqlVar vals[3];
  334. bExists = true;
  335. for(unsigned int i = 0; i < nFc; i++)
  336. {
  337. vals[i].FromField(pFields[i], pRow[i]);
  338. if(!vals[i].FieldnameCmp("ENGINE"))
  339. {
  340. pszEngine = vals[i].StrVal();
  341. bIsInnoDB = !strcasecmp(pszEngine, "InnoDB");
  342. }
  343. else if(!vals[i].FieldnameCmp("DATA_LENGTH"))
  344. nDataLength = (uint64_t)vals[i];
  345. else if(!vals[i].FieldnameCmp("INDEX_LENGTH"))
  346. nIndexLength = (uint64_t)vals[i];
  347. }
  348. if(bIsInnoDB && m_gv.bInnoDbFilePerTable)
  349. {
  350. if(!TableFileExists(pszTableName))
  351. m_lf.Warning("'innodb_file_per_table' configured, but table '%s' does not have a data file!\n", pszTableName);
  352. else
  353. nFileSize = TableFileSize(pszTableName);
  354. }
  355. if(nFileSize >= 0)
  356. m_lf.Info("Found info on table '%s': Engine: '%s', Data size: %s, Index size: %s, File size: %s\n", pszTableName, pszEngine ? pszEngine : "unknown", strFormatByteSize(nDataLength, 1).c_str(), strFormatByteSize(nIndexLength, 1).c_str(), strFormatByteSize(nFileSize, 1).c_str());
  357. else
  358. m_lf.Info("Found info on table '%s': Engine: '%s', Data size: %s, Index size: %s\n", pszTableName, pszEngine ? pszEngine : "unknown", strFormatByteSize(nDataLength, 1).c_str(), strFormatByteSize(nIndexLength, 1).c_str());
  359. if(!pszEngine)
  360. m_lf.Warning("Failed to determine the Database engine of Table '%s'!\n", pszTableName);
  361. else if(strcmp(pszEngine, _DL_DATABSE_ENGINE_NAME))
  362. m_lf.Warning("Current Database engine of Table '%s' is '%s'! Please consider to migrate to '%s'!\n", pszTableName, pszEngine, _DL_DATABSE_ENGINE_NAME);
  363. sSql = formatString("select * from `%s` limit 0, 10", pszTableName);
  364. CMySqlResult res2 = rdb.Query(sSql.c_str()); // try to query table
  365. if(res2.error()) // if the query results in an error, the table may be corrupt!
  366. {
  367. char szTblName[256];
  368. m_lf.Error("Failed to query table '%s': %s\n", pszTableName, rdb.LastError().c_str());
  369. // if the table is corrupt, try to rename it and create a new instane of the original table.
  370. sprintf(szTblName, "%s_corrupt_%08zX", pszTableName, (size_t)time(NULL));
  371. sSql = formatString("rename table `%s` to `%s`", pszTableName, szTblName);
  372. CMySqlResult res3 = rdb.Query(sSql.c_str()); // try to rename the corrupt table
  373. if(res3.error())
  374. {
  375. // if the renaming fails, try to create a new working table with a different name
  376. m_lf.Error("Failed to rename table '%s': %s\n", pszTableName, rdb.LastError().c_str());
  377. bExists = false;
  378. bUseResortTable = true;
  379. }
  380. else
  381. {
  382. m_lf.Warning("Renamed table '%s' to '%s'\n", pszTableName, szTblName);
  383. bExists = false;
  384. }
  385. }
  386. }
  387. }
  388. return !bError;
  389. }
  390. /////////////////////////////////////////////////////////////////////////////
  391. bool CDataLogger::InitDatabase(bool bEnforceCreate)
  392. {
  393. CMySqlDB db;
  394. time_t nMaxLogTimestamp = m_nLastLogTimestamp = 0;
  395. bool bExists, bUseResortTable, bDummy;
  396. m_lf.Info("Connecting to Database server @ 'localhost'.\n");
  397. if(!db.Connect("localhost", m_dlp.szDBUser, m_dlp.szDBPass, NULL))
  398. {
  399. m_lf.Error("CDataLogger::InitDatabase: DB Error: %s\n", db.LastError().c_str());
  400. return false;
  401. }
  402. m_lf.Info("Success!\n");
  403. if(!ReadGlobalOptions(db))
  404. return false;
  405. if(!m_gv.bInnoDbIsBarracuda || !m_gv.bInnoDbFilePerTable || !m_gv.bInnoDbIsStrictMode)
  406. m_lf.Warning("InnoDB file format: '%s', InnoDB table space: '%s', InnoDB strict mode: %s.\n", m_gv.szInnoDbFileFormat, m_gv.bInnoDbFilePerTable ? "File per table" : "System", m_gv.bInnoDbIsStrictMode ? "yes" : "no");
  407. else
  408. m_lf.Info("InnoDB file format: '%s', InnoDB table space: '%s', InnoDB strict mode: %s.\n", m_gv.szInnoDbFileFormat, m_gv.bInnoDbFilePerTable ? "File per table" : "System", m_gv.bInnoDbIsStrictMode ? "yes" : "no");
  409. if(!CreateDatabase(db, bEnforceCreate))
  410. return false;
  411. m_lf.Info("Opening Database '%s'.\n", m_dlp.szDBName);
  412. if(db.SelectDB(m_dlp.szDBName))
  413. {
  414. m_lf.Error("CDataLogger::InitDatabase: DB Error: %s\n", db.LastError().c_str());
  415. return false;
  416. }
  417. m_lf.Info("Success!\n");
  418. if(!CheckTable(db, m_dlp.szTagsTable, bExists, bUseResortTable, bDummy))
  419. return false;
  420. if(!bExists)
  421. {
  422. if(bUseResortTable)
  423. {
  424. char szOrigTable[_DL_MAX_TABLE_NAME_LENGTH];
  425. memcpy(szOrigTable, m_dlp.szTagsTable, sizeof(szOrigTable));
  426. sprintf(m_dlp.szTagsTable, "%s_resort_%08zX", szOrigTable, (size_t)time(NULL));
  427. m_lf.Warning("Resorting to work-around table '%s'.\n", m_dlp.szTagsTable);
  428. }
  429. if(!CreateTagsTable(db))
  430. return false;
  431. }
  432. if(!AlterTagsTable(db)) // 23.10.2020, extend logtype enum if not up to date
  433. return false;
  434. if(!CheckTable(db, m_dlp.szLogsTable, bExists, bUseResortTable, m_gv.bLogsTblIsInnoDB))
  435. return false;
  436. if(!bExists)
  437. {
  438. if(bUseResortTable)
  439. {
  440. char szOrigTable[_DL_MAX_TABLE_NAME_LENGTH];
  441. memcpy(szOrigTable, m_dlp.szLogsTable, sizeof(szOrigTable));
  442. sprintf(m_dlp.szLogsTable, "%s_resort_%08zX", szOrigTable, (size_t)time(NULL));
  443. m_lf.Warning("Resorting to work-around table '%s'.\n", m_dlp.szLogsTable);
  444. }
  445. if(!CreateLogsTable(db))
  446. return false;
  447. #if _DL_DATABSE_ENGINE == _DL_DATABASE_ENGINE_INNODB
  448. m_gv.bLogsTblIsInnoDB = true;
  449. #endif // _DL_DATABSE_ENGINE == _DL_DATABASE_ENGINE_INNODB
  450. }
  451. if(m_bSGConfigured)
  452. {
  453. if(!(m_bSGHasSizeLimitPrerequisites = m_gv.bLogsTblIsInnoDB && m_gv.bInnoDbFilePerTable && TableFileExists(m_dlp.szLogsTable)))
  454. {
  455. m_lf.Warning("SG: Database configuration: InnoDB: %s, File-per-table tablespace: %s\n", m_gv.bLogsTblIsInnoDB ? "yes" : "no", m_gv.bInnoDbFilePerTable ? "yes" : "no");
  456. m_lf.Warning("SG: The current database configuration does not support monitoring of size limits. Size guard will operate on age limits only!\n");
  457. }
  458. ::pthread_mutex_lock(&m_condmtx1);
  459. if(::pthread_create(&m_tidSGThread, NULL, &CDataLogger::SizeGuardWorker, reinterpret_cast<void*>(this)))
  460. {
  461. m_tidSGThread = 0;
  462. m_lf.Error("SG: Failed to start size monitoring thread!\n");
  463. ::pthread_mutex_unlock(&m_condmtx1);
  464. return false;
  465. }
  466. ::pthread_cond_wait(&m_cond1, &m_condmtx1);
  467. ::pthread_mutex_unlock(&m_condmtx1);
  468. }
  469. if(!CreateLogsBDTable(db))
  470. return false;
  471. if(((nMaxLogTimestamp = GetLastLogTimestamp(db)) != _INVALID_TIMESTAMP_VALUE))
  472. m_nLastLogTimestamp = nMaxLogTimestamp;
  473. m_lf.Info("Database initialization complete.\n");
  474. return true;
  475. }
  476. /////////////////////////////////////////////////////////////////////////////
  477. bool CDataLogger::CreateDatabase(CMySqlDB &rdb, bool bEnforceCreate)
  478. {
  479. bool bError = false;
  480. std::string sSql;
  481. if(bEnforceCreate)
  482. {
  483. sSql = formatString("drop database if exists `%s`", m_dlp.szDBName);
  484. CMySqlResult res = rdb.Query(sSql.c_str());
  485. bError = res.error();
  486. if(bError)
  487. m_lf.Error("CDataLogger::InitDatabase: DB Error: %s\n", rdb.LastError().c_str());
  488. }
  489. if(!bError)
  490. {
  491. sSql = formatString("create database if not exists `%s`", m_dlp.szDBName);
  492. CMySqlResult res = rdb.Query(sSql.c_str());
  493. bError = res.error();
  494. if(bError)
  495. m_lf.Error("CDataLogger::InitDatabase: DB Error: %s\n", rdb.LastError().c_str());
  496. }
  497. return !bError;
  498. }
  499. /////////////////////////////////////////////////////////////////////////////
  500. bool CDataLogger::CreateTagsTable(CMySqlDB &rdb)
  501. {
  502. std::string sSql;
  503. const char *pszFormat =
  504. "CREATE TABLE IF NOT EXISTS `%s` (" \
  505. " `tagid` smallint(6) unsigned NOT NULL AUTO_INCREMENT," \
  506. " `dataType` enum('bool', 'I1', 'UI1', 'I2', 'UI2', 'I4', 'UI4', 'I8', 'UI8', 'float', 'double', 'string') NOT NULL," \
  507. " `logType` enum('IC', 'IU', 'VC', 'VU', 'ICR', 'IUR', 'VCR', 'VUR') NOT NULL," \
  508. " `path` varchar(%u) NOT NULL," \
  509. " PRIMARY KEY (`tagid`, `dataType`, `logType`)," \
  510. " KEY `path` (`path`)" \
  511. ") ENGINE=%s DEFAULT CHARSET=ascii";
  512. m_lf.Info("Creating Table '%s' (if not exists) using engine '%s'.\n", m_dlp.szTagsTable, _DL_DATABSE_ENGINE_NAME);
  513. sSql = formatString(pszFormat, m_dlp.szTagsTable, _DL_MAX_VARPATH_LENGTH, _DL_DATABSE_ENGINE_NAME);
  514. CMySqlResult res = rdb.Query(sSql.c_str());
  515. bool bError = res.error();
  516. if(bError)
  517. m_lf.Error("CDataLogger::CreateTagsTable: DB Error: %s\n", rdb.LastError().c_str());
  518. else
  519. m_lf.Info("Success!\n");
  520. return !bError;
  521. }
  522. bool CDataLogger::AlterTagsTable(CMySqlDB &rdb)
  523. {
  524. std::string sSql = formatString("ALTER TABLE `%s` MODIFY COLUMN `logType` enum('IC','IU','VC','VU','ICR','IUR','VCR','VUR') NOT NULL", m_dlp.szTagsTable);
  525. CMySqlResult res = rdb.Query(sSql.c_str());
  526. bool bError = res.error();
  527. if(bError)
  528. m_lf.Error("CDataLogger::AlterTagsTable: DB Error: %s\n", rdb.LastError().c_str());
  529. return !bError;
  530. }
  531. /////////////////////////////////////////////////////////////////////////////
  532. bool CDataLogger::CreateLogsTable(CMySqlDB &rdb)
  533. {
  534. std::string sSql;
  535. const char *pszFormat =
  536. #if _USE_MODIFIED_INDEX
  537. "CREATE TABLE IF NOT EXISTS `%s` (" \
  538. " `tagid` smallint(6) unsigned NOT NULL," \
  539. " `tslog` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP," \
  540. " `value` double," \
  541. " `valueMin` double DEFAULT NULL," \
  542. " `valueMax` double DEFAULT NULL," \
  543. " PRIMARY KEY (`tslog`, `tagid`)," \
  544. " KEY `taglog` (`tagid`, `tslog`)" \
  545. ") ENGINE=%s DEFAULT CHARSET=ascii ROW_FORMAT=COMPRESSED";
  546. #else // _USE_MODIFIED_INDEX
  547. "CREATE TABLE IF NOT EXISTS `%s` (" \
  548. " `tagid` smallint(6) unsigned NOT NULL," \
  549. " `tslog` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP," \
  550. " `value` double," \
  551. " `valueMin` double DEFAULT NULL," \
  552. " `valueMax` double DEFAULT NULL," \
  553. " PRIMARY KEY (`tagid`, `tslog`)," \
  554. " KEY `tslog` (`tslog`)" \
  555. ") ENGINE=%s DEFAULT CHARSET=ascii ROW_FORMAT=COMPRESSED";
  556. #endif // _USE_MODIFIED_INDEX
  557. m_lf.Info("Creating Table '%s' if not exists using engine '%s'.\n", m_dlp.szLogsTable, _DL_DATABSE_ENGINE_NAME);
  558. sSql = formatString(pszFormat, m_dlp.szLogsTable, _DL_DATABSE_ENGINE_NAME);
  559. CMySqlResult res = rdb.Query(sSql.c_str());
  560. bool bError = res.error();
  561. if(bError)
  562. m_lf.Error("CDataLogger::CreateLogsTable: DB Error: %s\n", rdb.LastError().c_str());
  563. else
  564. m_lf.Info("Success!\n");
  565. return !bError;
  566. }
  567. /////////////////////////////////////////////////////////////////////////////
  568. bool CDataLogger::CreateLogsBDTable(CMySqlDB &rdb)
  569. {
  570. std::string sSql;
  571. const char *pszFormat =
  572. "CREATE TABLE IF NOT EXISTS `%s` (" \
  573. " `id` int unsigned unsigned NOT NULL AUTO_INCREMENT," \
  574. " `tagid` smallint(6) unsigned NOT NULL," \
  575. " `tslog` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP," \
  576. " `value` double," \
  577. " `valueMin` double DEFAULT NULL," \
  578. " `valueMax` double DEFAULT NULL," \
  579. " PRIMARY KEY (`id`)," \
  580. " KEY `tslog` (`tslog`)," \
  581. " KEY `tagid` (`tagid`)" \
  582. ") ENGINE=%s DEFAULT CHARSET=ascii ROW_FORMAT=COMPRESSED";
  583. m_lf.Info("Creating Table '%s' if not exists using engine '%s'.\n", m_dlp.szLogsTableBD, _DL_DATABSE_ENGINE_NAME);
  584. sSql = formatString(pszFormat, m_dlp.szLogsTableBD, _DL_DATABSE_ENGINE_NAME);
  585. CMySqlResult res = rdb.Query(sSql.c_str());
  586. bool bError = res.error();
  587. if(bError)
  588. m_lf.Error("CDataLogger::CreateLogsBDTable: DB Error: %s\n", rdb.LastError().c_str());
  589. else
  590. m_lf.Info("Success!\n");
  591. return !bError;
  592. }
  593. /////////////////////////////////////////////////////////////////////////////
  594. bool CDataLogger::Log(unsigned long nTagID, double fValue, double fMin, double fMax, time_t nTimestamp, int nIndex, LogTypes lt, bool bNull, bool bNoBadDateCheck)
  595. {
  596. bNull = bNull || !_IsLoggable(fValue) || !_IsLoggable(fMin) || !_IsLoggable(fMax);
  597. DL_LOG_ENTRY log;
  598. log.nTagID = nTagID;
  599. log.nTimestamp = nTimestamp;
  600. log.fValue = fValue;
  601. log.fMin = fMin;
  602. log.fMax = fMax;
  603. log.nIndex = nIndex;
  604. log.lt = lt;
  605. log.bNull = bNull;
  606. if(bNoBadDateCheck || (nTimestamp > m_nLastLogTimestamp))
  607. m_logs.push_back(log);
  608. else
  609. m_logsBD.push_back(log);
  610. return true;
  611. }
  612. /////////////////////////////////////////////////////////////////////////////
  613. time_t CDataLogger::GetLastLogTimestamp(CMySqlDB &rdb)
  614. {
  615. time_t nTs = _INVALID_TIMESTAMP_VALUE;
  616. std::string sSql;
  617. do
  618. {
  619. sSql = formatString("select unix_timestamp(max(`tslog`)) from `%s`", m_dlp.szLogsTable);
  620. CMySqlResult res = rdb.Query(sSql.c_str());
  621. if(res.error())
  622. {
  623. m_lf.Error("CDataLogger::GetLastLogTimestamp: DB Error: %s\n", rdb.LastError().c_str());
  624. break;
  625. }
  626. else
  627. {
  628. CMySqlVar val;
  629. my_ulonglong nRowCount = res.RowCount();
  630. unsigned int nFldCount = res.FieldCount();
  631. const MYSQL_FIELD *pFields = res.FetchFields();
  632. if(nRowCount != 1 || nFldCount != 1 || !pFields)
  633. {
  634. // m_lf.Error("CDataLogger::GetLastLogTimestamp: Unexpected error!\n");
  635. break;
  636. }
  637. MYSQL_ROW pRow = res.FetchRow();
  638. if( !pRow ||
  639. !val.FromField(pFields[0], pRow[0]))
  640. {
  641. // m_lf.Error("CDataLogger::GetLastLogTimestamp: Unexpected error!\n");
  642. break;
  643. }
  644. nTs = (time_t)(uint64_t)val;
  645. }
  646. }
  647. while(false);
  648. return nTs;
  649. }
  650. /////////////////////////////////////////////////////////////////////////////
  651. bool CDataLogger::Flush(time_t nTimestamp)
  652. {
  653. ::pthread_mutex_lock(&m_condmtx1);
  654. bool bGoodBadTransition = false, bSgInProg = m_bSGInProgress;
  655. ::pthread_mutex_unlock(&m_condmtx1);
  656. time_t nMaxLogTimestamp = 0, nFirstGoodTimestamp = 0;
  657. size_t nCountValidDate = m_logs.size();
  658. size_t nCountBadDate = m_logsBD.size();
  659. if(!nCountValidDate && !nCountBadDate)
  660. return true; // nothing to do
  661. TRACE("Trying to flush %zu logs ...\n", nCountValidDate + nCountBadDate);
  662. if(bSgInProg && nCountValidDate)
  663. {
  664. TRACE("Flush: SG in progress - flushing of %zu logs deferred!\n", nCountValidDate);
  665. m_lf.Info("Flush: SG in progress - flushing of %zu logs deferred!\n", nCountValidDate);
  666. nCountValidDate = 0; // size guard operates on the valid-date-logs-table only, so we can safely flush invalid-date-logs
  667. if(!nCountBadDate)
  668. return true; // no error
  669. }
  670. if(!m_bBadDateLogsDetected && nCountBadDate)
  671. {
  672. m_bBadDateLogsDetected = true;
  673. bGoodBadTransition = true;
  674. }
  675. else if(m_bBadDateLogsDetected && !nCountBadDate)
  676. {
  677. m_bBadDateLogsDetected = false;
  678. bGoodBadTransition = true;
  679. }
  680. CMySqlDB db;
  681. std::string strUnlock, strSaveFgnKey, strDisableFgnKey, strRestoreFgnKey, strSaveTZ, strSetTZ, strRestoreTZ;
  682. bool bError = false;
  683. std::string sSql;
  684. if(!db.Connect("localhost", m_dlp.szDBUser, m_dlp.szDBPass, m_dlp.szDBName))
  685. {
  686. m_lf.Error("CDataLogger::Flush: DB Error: %s\n", db.LastError().c_str());
  687. return false;
  688. }
  689. strUnlock = formatString("unlock tables");
  690. strSaveFgnKey = formatString("set @old_foreign_key_checks = @@foreign_key_checks");
  691. strDisableFgnKey = formatString("set foreign_key_checks = 0");
  692. strRestoreFgnKey = formatString("set foreign_key_checks = @old_foreign_key_checks");
  693. strSaveTZ = formatString("set @old_time_zone = @@time_zone");
  694. strSetTZ = formatString("set time_zone = '+00:00'");
  695. strRestoreTZ = formatString("set time_zone = @old_time_zone");
  696. db.Query(strSaveTZ.c_str());
  697. db.Query(strSetTZ.c_str());
  698. db.Query(strSaveFgnKey.c_str());
  699. db.Query(strDisableFgnKey.c_str());
  700. if(nCountValidDate > 0)
  701. {
  702. std::string strSql, strLock;
  703. strLock = formatString("lock tables `%s` write", m_dlp.szLogsTable);
  704. strSql.reserve(130 + 111 * nCountValidDate);
  705. auto itFirst = m_logs.begin();
  706. const DL_LOG_ENTRY &rle0 = *itFirst;
  707. if(nMaxLogTimestamp < rle0.nTimestamp)
  708. nMaxLogTimestamp = rle0.nTimestamp;
  709. nFirstGoodTimestamp = rle0.nTimestamp;
  710. if(m_dlp.bMinMax && _IS_INTERVAL_LOGTYPE(rle0.lt))
  711. {
  712. if(!rle0.bNull)
  713. sSql = formatString("insert into `%s` (`tagid`, `tslog`, `value`, `valueMin`, `valueMax`) values (%lu, timestamp(from_unixtime(%lu)), %.20g, %.20g, %.20g)", m_dlp.szLogsTable, rle0.nTagID, rle0.nTimestamp, rle0.fValue, rle0.fMin, rle0.fMax);
  714. else
  715. sSql = formatString("insert into `%s` (`tagid`, `tslog`, `value`, `valueMin`, `valueMax`) values (%lu, timestamp(from_unixtime(%lu)), NULL, NULL, NULL)", m_dlp.szLogsTable, rle0.nTagID, rle0.nTimestamp);
  716. }
  717. else
  718. {
  719. if(!rle0.bNull)
  720. sSql = formatString("insert into `%s` (`tagid`, `tslog`, `value`, `valueMin`, `valueMax`) values (%lu, timestamp(from_unixtime(%lu)), %.20g, NULL, NULL)", m_dlp.szLogsTable, rle0.nTagID, rle0.nTimestamp, rle0.fValue);
  721. else
  722. sSql = formatString("insert into `%s` (`tagid`, `tslog`, `value`, `valueMin`, `valueMax`) values (%lu, timestamp(from_unixtime(%lu)), NULL, NULL, NULL)", m_dlp.szLogsTable, rle0.nTagID, rle0.nTimestamp);
  723. }
  724. strSql = sSql;
  725. for(++itFirst; itFirst < m_logs.end(); itFirst++)
  726. {
  727. const DL_LOG_ENTRY &rle = *itFirst;
  728. if(nMaxLogTimestamp < rle.nTimestamp)
  729. nMaxLogTimestamp = rle.nTimestamp;
  730. if(m_dlp.bMinMax && _IS_INTERVAL_LOGTYPE(rle.lt))
  731. {
  732. if(!rle0.bNull)
  733. sSql = formatString(",(%lu, timestamp(from_unixtime(%lu)), %.20g, %.20g, %.20g)", rle.nTagID, rle.nTimestamp, rle.fValue, rle.fMin, rle.fMax);
  734. else
  735. sSql = formatString(",(%lu, timestamp(from_unixtime(%lu)), NULL, NULL, NULL)", rle.nTagID, rle.nTimestamp);
  736. }
  737. else
  738. {
  739. if(!rle.bNull)
  740. sSql = formatString(",(%lu, timestamp(from_unixtime(%lu)), %.20g, NULL, NULL)", rle.nTagID, rle.nTimestamp, rle.fValue);
  741. else
  742. sSql = formatString(",(%lu, timestamp(from_unixtime(%lu)), NULL, NULL, NULL)", rle.nTagID, rle.nTimestamp);
  743. }
  744. strSql += sSql;
  745. }
  746. if(TryLock())
  747. {
  748. db.Query(strLock.c_str());
  749. CMySqlResult res = db.Query(strSql.c_str());
  750. bError = res.error();
  751. db.Query(strUnlock.c_str());
  752. Unlock();
  753. if(bError)
  754. m_lf.Error("CDataLogger::Flush: DB Error: %s\n", db.LastError().c_str());
  755. else
  756. m_nLastLogTimestamp = nMaxLogTimestamp;
  757. m_logs.clear();
  758. }
  759. }
  760. if(bGoodBadTransition)
  761. {
  762. if(m_bBadDateLogsDetected)
  763. {
  764. bool bError;
  765. char szTs[64];
  766. unsigned long long nNextAIVal = 0;
  767. sSql = formatString("select auto_increment from `information_schema`.`TABLES` where `TABLE_SCHEMA` = '%s' and `TABLE_NAME` = '%s';", m_dlp.szDBName, m_dlp.szLogsTableBD);
  768. CMySqlResult res = db.Query(sSql.c_str());
  769. if(!(bError = res.error()))
  770. {
  771. CMySqlVar val;
  772. my_ulonglong nRowCount = res.RowCount();
  773. unsigned int nFldCount = res.FieldCount();
  774. const MYSQL_FIELD *pFields = res.FetchFields();
  775. do
  776. {
  777. if(nRowCount != 1 || nFldCount != 1 || !pFields)
  778. break;
  779. MYSQL_ROW pRow = res.FetchRow();
  780. if(!pRow || !*pRow || !**pRow)
  781. break;
  782. if(!val.FromField(pFields[0], pRow[0]))
  783. break;
  784. nNextAIVal = val;
  785. }
  786. while(false);
  787. }
  788. Timestamp2String(m_nLastLogTimestamp, szTs, sizeof(szTs));
  789. if(nNextAIVal)
  790. {
  791. TRACE("Flush: Transition valid -> invalid date detected! The last Timestamp in Table `%s` was '%s+00:00'. Subsequent logs will be written to table `%s` starting with `id` %llu.\n", m_dlp.szLogsTable, szTs, m_dlp.szLogsTableBD, nNextAIVal);
  792. m_lf.Warning("Flush: Transition valid -> invalid date detected! The last Timestamp in Table `%s` was '%s+00:00'. Subsequent logs will be written to table `%s` starting with `id` %llu.\n", m_dlp.szLogsTable, szTs, m_dlp.szLogsTableBD, nNextAIVal);
  793. }
  794. else
  795. {
  796. TRACE("Flush: Transition valid -> invalid date detected! The last Timestamp in Table `%s` was '%s+00:00'. Subsequent logs will be written to table `%s`.\n", m_dlp.szLogsTable, szTs, m_dlp.szLogsTableBD);
  797. m_lf.Warning("Flush: Transition valid -> invalid date detected! The last Timestamp in Table `%s` was '%s+00:00'. Subsequent logs will be written to table `%s`.\n", m_dlp.szLogsTable, szTs, m_dlp.szLogsTableBD);
  798. }
  799. }
  800. else
  801. {
  802. bool bError;
  803. char szTs[64];
  804. unsigned long long nLastID = 0;
  805. sSql = formatString("select max(`id`) from `%s`.`%s`;", m_dlp.szDBName, m_dlp.szLogsTableBD);
  806. CMySqlResult res = db.Query(sSql.c_str());
  807. if(!(bError = res.error()))
  808. {
  809. CMySqlVar val;
  810. my_ulonglong nRowCount = res.RowCount();
  811. unsigned int nFldCount = res.FieldCount();
  812. const MYSQL_FIELD *pFields = res.FetchFields();
  813. do
  814. {
  815. if(nRowCount != 1 || nFldCount != 1 || !pFields)
  816. break;
  817. MYSQL_ROW pRow = res.FetchRow();
  818. if(!pRow || !*pRow || !**pRow)
  819. break;
  820. if(!val.FromField(pFields[0], pRow[0]))
  821. break;
  822. nLastID = val;
  823. }
  824. while(false);
  825. }
  826. Timestamp2String(nFirstGoodTimestamp, szTs, sizeof(szTs));
  827. if(nLastID)
  828. {
  829. TRACE("Flush: Transition invalid -> valid date detected! The last `id` in table `%s` was %llu. Subsequent logs will be written to table `%s` beginning with timestamp '%s+00:00'.\n", m_dlp.szLogsTableBD, nLastID, m_dlp.szLogsTable, szTs);
  830. m_lf.Warning("Flush: Transition invalid -> valid date detected! The last `id` in table `%s` was %llu. Subsequent logs will be written to table `%s` beginning with timestamp '%s+00:00'.\n", m_dlp.szLogsTableBD, nLastID, m_dlp.szLogsTable, szTs);
  831. }
  832. else
  833. {
  834. TRACE("Flush: Transition invalid -> valid date detected! Subsequent logs will be written to table `%s` beginning with timestamp '%s+00:00'.\n", m_dlp.szLogsTable, szTs);
  835. m_lf.Warning("Flush: Transition invalid -> valid date detected! Subsequent logs will be written to table `%s` beginning with timestamp '%s+00:00'.\n", m_dlp.szLogsTable, szTs);
  836. }
  837. }
  838. }
  839. if(nCountBadDate > 0)
  840. {
  841. std::string strSql, strLock;
  842. strLock = formatString("lock tables `%s` write", m_dlp.szLogsTableBD);
  843. strSql.reserve(130 + 111 * nCountBadDate);
  844. auto itFirst = m_logsBD.begin();
  845. const DL_LOG_ENTRY &rle0 = *itFirst;
  846. if(m_dlp.bMinMax && _IS_INTERVAL_LOGTYPE(rle0.lt))
  847. {
  848. if(!rle0.bNull)
  849. sSql = formatString("insert into `%s` (`tagid`, `tslog`, `value`, `valueMin`, `valueMax`) values (%lu, timestamp(from_unixtime(%lu)), %.20g, %.20g, %.20g)", m_dlp.szLogsTableBD, rle0.nTagID, rle0.nTimestamp, rle0.fValue, rle0.fMin, rle0.fMax);
  850. else
  851. sSql = formatString("insert into `%s` (`tagid`, `tslog`, `value`, `valueMin`, `valueMax`) values (%lu, timestamp(from_unixtime(%lu)), NULL, NULL, NULL)", m_dlp.szLogsTableBD, rle0.nTagID, rle0.nTimestamp);
  852. }
  853. else
  854. {
  855. if(!rle0.bNull)
  856. sSql = formatString("insert into `%s` (`tagid`, `tslog`, `value`, `valueMin`, `valueMax`) values (%lu, timestamp(from_unixtime(%lu)), %.20g, NULL, NULL)", m_dlp.szLogsTableBD, rle0.nTagID, rle0.nTimestamp, rle0.fValue);
  857. else
  858. sSql = formatString("insert into `%s` (`tagid`, `tslog`, `value`, `valueMin`, `valueMax`) values (%lu, timestamp(from_unixtime(%lu)), NULL, NULL, NULL)", m_dlp.szLogsTableBD, rle0.nTagID, rle0.nTimestamp);
  859. }
  860. strSql = sSql;
  861. for(++itFirst; itFirst < m_logsBD.end(); itFirst++)
  862. {
  863. const DL_LOG_ENTRY &rle = *itFirst;
  864. if(m_dlp.bMinMax && _IS_INTERVAL_LOGTYPE(rle.lt))
  865. {
  866. if(!rle.bNull)
  867. sSql = formatString(",(%lu, timestamp(from_unixtime(%lu)), %.20g, %.20g, %.20g)", rle.nTagID, rle.nTimestamp, rle.fValue, rle.fMin, rle.fMax);
  868. else
  869. sSql = formatString(",(%lu, timestamp(from_unixtime(%lu)), NULL, NULL, NULL)", rle.nTagID, rle.nTimestamp);
  870. }
  871. else
  872. {
  873. if(!rle.bNull)
  874. sSql = formatString(",(%lu, timestamp(from_unixtime(%lu)), %.20g, NULL, NULL)", rle.nTagID, rle.nTimestamp, rle.fValue);
  875. else
  876. sSql = formatString(",(%lu, timestamp(from_unixtime(%lu)), NULL, NULL, NULL)", rle.nTagID, rle.nTimestamp);
  877. }
  878. strSql += sSql;
  879. }
  880. db.Query(strLock.c_str());
  881. CMySqlResult res = db.Query(strSql.c_str());
  882. bError = res.error();
  883. db.Query(strUnlock.c_str());
  884. if(bError)
  885. m_lf.Error("CDataLogger::Flush: DB Error: %s\n", db.LastError().c_str());
  886. m_logsBD.clear();
  887. }
  888. db.Query(strRestoreFgnKey.c_str());
  889. db.Query(strRestoreTZ.c_str());
  890. return !bError;
  891. }
  892. /////////////////////////////////////////////////////////////////////////////
  893. unsigned long long CDataLogger::SizeGuardLastPassRead(void)
  894. {
  895. char szPath[PATH_MAX];
  896. sprintf(szPath, "%s/%s", m_szAppDir, _SIZE_GUARD_TIMESTAMP_FILE_NAME);
  897. FILE *pf = fopen(szPath, "rb");
  898. unsigned long long ts = 0;
  899. if(pf)
  900. {
  901. if(fread(&ts, sizeof(ts), 1, pf) != 1)
  902. ts = 0;
  903. fclose(pf);
  904. }
  905. return ts;
  906. }
  907. /////////////////////////////////////////////////////////////////////////////
  908. void CDataLogger::SizeGuardLastPassWrite(unsigned long long ts)
  909. {
  910. char szPath[PATH_MAX];
  911. sprintf(szPath, "%s/%s", m_szAppDir, _SIZE_GUARD_TIMESTAMP_FILE_NAME);
  912. FILE *pf = fopen(szPath, "wb");
  913. if(pf)
  914. {
  915. fwrite(&ts, sizeof(ts), 1, pf);
  916. fclose(pf);
  917. }
  918. }
  919. /////////////////////////////////////////////////////////////////////////////
  920. size_t CDataLogger::Timestamp2String(time_t t, char *pszBuffer, size_t nCbBuffer)
  921. {
  922. const struct tm * ptm = gmtime(&t);
  923. return strftime(pszBuffer, nCbBuffer, "%F %T", ptm);
  924. }
  925. /////////////////////////////////////////////////////////////////////////////
  926. const char* CDataLogger::Ns2String(unsigned long long nNs, char *pszBuffer, size_t nCbBuffer)
  927. {
  928. *pszBuffer = '\0';
  929. if(nNs < 1000)
  930. snprintf(pszBuffer, nCbBuffer, "%llu ns", nNs);
  931. else if(nNs < 1000000)
  932. snprintf(pszBuffer, nCbBuffer, "%.1f us", (double)nNs / 1000.0);
  933. else
  934. return Ms2String((double)nNs / 1000000.0, pszBuffer, nCbBuffer);
  935. return pszBuffer;
  936. }
  937. /////////////////////////////////////////////////////////////////////////////
  938. const char* CDataLogger::Ms2String(double fMs, char *pszBuffer, size_t nCbBuffer)
  939. {
  940. *pszBuffer = '\0';
  941. if(fMs < 1000.0)
  942. snprintf(pszBuffer, nCbBuffer, "%.2f ms", fMs);
  943. else if(fMs < 60000.0)
  944. snprintf(pszBuffer, nCbBuffer, "%.2f sec", fMs / 1000.0);
  945. else if(fMs < 3600000.0)
  946. snprintf(pszBuffer, nCbBuffer, "%.2f min", fMs / 60000.0);
  947. else
  948. snprintf(pszBuffer, nCbBuffer, "%.2f h", fMs / 3600000.0);
  949. return pszBuffer;
  950. }
  951. /////////////////////////////////////////////////////////////////////////////
  952. bool CDataLogger::DoSizeGuard(void)
  953. {
  954. if(!m_bSGConfigured)
  955. return true;
  956. CMySqlDB db;
  957. bool bError;
  958. MYSQL_ROW pRow;
  959. unsigned long long nElapsed;
  960. char szT1[32], szT2[32], szMs[32];
  961. CProcessClock pc;
  962. CMySqlVar vMinTs, vCountDelete;
  963. int64_t nMinTs, nTsCurMidnightUTC, nTsDeleteUpTo;
  964. std::string sSql;
  965. /////////////////////////////////////////////////////////////////////////
  966. nTsCurMidnightUTC = _MIDNIGHT_TIMESTAMP_UTC(m_nSGCurPassUTC);
  967. if(SizeGuardDayWorkDone(nTsCurMidnightUTC))
  968. return true;
  969. m_nSGLastPassUTC = nTsCurMidnightUTC;
  970. if(SizeGuardLastPassRead() == (unsigned long long)(nTsCurMidnightUTC))
  971. return true;
  972. SizeGuardLastPassWrite(nTsCurMidnightUTC);
  973. /////////////////////////////////////////////////////////////////////////
  974. if(!db.Connect("localhost", m_dlp.szDBUser, m_dlp.szDBPass, NULL))
  975. {
  976. m_lf.Error("CDataLogger::DoSizeGuard(%d): - DB Error: %s\n", __LINE__, db.LastError().c_str());
  977. return false;
  978. }
  979. sSql = formatString("select unix_timestamp(min(`tslog`)) from `%s`.`%s`;", m_dlp.szDBName, m_dlp.szLogsTable);
  980. m_lf.Info("SG: \"%s\".\n", sSql.c_str());
  981. pc.ClockTrigger();
  982. CMySqlResult res = db.Query(sSql.c_str());
  983. nElapsed = pc.ClockGetElapsed();
  984. bError = res.error();
  985. if(bError)
  986. {
  987. m_lf.Error("CDataLogger::DoSizeGuard(%d): \"%s\" - DB Error: %s\n", __LINE__, sSql.c_str(), db.LastError().c_str());
  988. return false;
  989. }
  990. if(!(pRow = res.FetchRow()))
  991. {
  992. m_lf.Error("CDataLogger::DoSizeGuard(%d): Unexpected Error!\n", __LINE__);
  993. return false;
  994. }
  995. my_ulonglong nRowCount = res.RowCount();
  996. unsigned int nFldCount = res.FieldCount();
  997. const MYSQL_FIELD *pFields = res.FetchFields();
  998. if(nRowCount != 1 || nFldCount != 1)
  999. {
  1000. m_lf.Error("CDataLogger::DoSizeGuard(%d): Unexpected Error!\n", __LINE__);
  1001. return false;
  1002. }
  1003. if(!vMinTs.FromField(pFields[0], pRow[0]))
  1004. {
  1005. m_lf.Error("CDataLogger::DoSizeGuard(%d): Unexpected Error!\n", __LINE__);
  1006. return false;
  1007. }
  1008. nMinTs = vMinTs;
  1009. m_lf.Info("SG: operation completed in %s. Min. timestamp: %ju\n", Ns2String(nElapsed, szMs, sizeof(szMs)), nMinTs);
  1010. /////////////////////////////////////////////////////////////////////////
  1011. // process max. size
  1012. if(m_bSGHasSizeLimitPrerequisites && m_dlp.nMaxSize)
  1013. {
  1014. unsigned long long nFileSize = TableFileSize(m_dlp.szLogsTable);
  1015. unsigned long long nGuardThreshold = m_dlp.nMaxSize * (100 - _FILE_SIZE_DELETE_MARGIN_PERCENT) / 100;
  1016. m_lf.Info("SG: File size: %s.\n", strFormatByteSize(nFileSize, 1).c_str());
  1017. if(nFileSize > nGuardThreshold)
  1018. {
  1019. Timestamp2String(nTsCurMidnightUTC - _SECONDS_PER_DAY, szT1, sizeof(szT1));
  1020. Timestamp2String(nTsCurMidnightUTC - 1, szT2, sizeof(szT2));
  1021. sSql = formatString("select count(*) from `%s`.`%s` where `tslog` between '%s' and '%s';", m_dlp.szDBName, m_dlp.szLogsTable, szT1, szT2);
  1022. m_lf.Info("SG: \"%s\"\n", sSql.c_str());
  1023. pc.ClockTrigger();
  1024. CMySqlResult res = db.Query(sSql.c_str());
  1025. bError = res.error();
  1026. if(bError)
  1027. {
  1028. m_lf.Error("CDataLogger::DoSizeGuard(%d): \"%s\" - DB Error: %s\n", __LINE__, sSql.c_str(), db.LastError().c_str());
  1029. return false;
  1030. }
  1031. else
  1032. {
  1033. nElapsed = pc.ClockGetElapsed();
  1034. nRowCount = res.RowCount();
  1035. nFldCount = res.FieldCount();
  1036. pFields = res.FetchFields();
  1037. pRow = res.FetchRow();
  1038. if(pRow && pFields && nRowCount == 1 && nFldCount == 1 && vCountDelete.FromField(pFields[0], pRow[0]))
  1039. {
  1040. unsigned long long nCountDelete = (uint64_t)vCountDelete;
  1041. m_lf.Info("SG: operation completed in %s. Estimated number of rows to delete: %llu.\n", Ns2String(nElapsed, szMs, sizeof(szMs)), nCountDelete);
  1042. if(nCountDelete > 0)
  1043. {
  1044. #if _USE_MODIFIED_INDEX
  1045. sSql = formatString("delete from `%s`.`%s` limit %llu;", m_dlp.szDBName, m_dlp.szLogsTable, nCountDelete);
  1046. #else // _USE_MODIFIED_INDEX
  1047. sSql = formatString("delete from `%s`.`%s` where `tslog` <= (select `tslog` from (select `tslog` from `demo`.`logs` order by `tslog` asc limit %llu, 1) x);", m_dlp.szDBName, m_dlp.szLogsTable, nCountDelete - 1);
  1048. #endif // _USE_MODIFIED_INDEX
  1049. m_lf.Info("SG: triggered on size limit: \"%s\".\n", sSql.c_str());
  1050. pc.ClockTrigger();
  1051. if(db.Query(sSql.c_str()).error())
  1052. {
  1053. m_lf.Error("CDataLogger::DoSizeGuard(%d): \"%s\" - DB Error: %s\n", __LINE__, sSql.c_str(), db.LastError().c_str());
  1054. return false;
  1055. }
  1056. else
  1057. {
  1058. nElapsed = pc.ClockGetElapsed();
  1059. m_lf.Info("SG: operation completed in %s.\n", Ns2String(nElapsed, szMs, sizeof(szMs)));
  1060. }
  1061. }
  1062. }
  1063. else
  1064. {
  1065. m_lf.Error("CDataLogger::DoSizeGuard(%d): Unexpected Error!\n", __LINE__);
  1066. return false;
  1067. }
  1068. }
  1069. }
  1070. }
  1071. /////////////////////////////////////////////////////////////////////////
  1072. // process max. age entries
  1073. if(m_dlp.nMaxAge)
  1074. {
  1075. nTsDeleteUpTo = nTsCurMidnightUTC - m_dlp.nMaxAge * _SECONDS_PER_DAY;
  1076. if(nMinTs <= nTsDeleteUpTo)
  1077. {
  1078. Timestamp2String(nTsDeleteUpTo, szT1, sizeof(szT1));
  1079. sSql = formatString("delete from `%s`.`%s` where `tslog` < '%s';", m_dlp.szDBName, m_dlp.szLogsTable, szT1);
  1080. m_lf.Info("SG: triggered on age limit: \"%s\".\n", sSql.c_str());
  1081. pc.ClockTrigger();
  1082. CMySqlResult res = db.Query(sSql.c_str());
  1083. nElapsed = pc.ClockGetElapsed();
  1084. bError = res.error();
  1085. if(bError)
  1086. {
  1087. m_lf.Error("CDataLogger::DoSizeGuard(%d): DB Error: %s\n", __LINE__, db.LastError().c_str());
  1088. return false;
  1089. }
  1090. else
  1091. {
  1092. m_lf.Info("SG: operation completed in %s.\n", Ns2String(nElapsed, szMs, sizeof(szMs)));
  1093. }
  1094. }
  1095. }
  1096. return true;
  1097. }
  1098. /////////////////////////////////////////////////////////////////////////////
  1099. /////////////////////////////////////////////////////////////////////////////
  1100. /////////////////////////////////////////////////////////////////////////////
  1101. // size guard worker thread
  1102. void* CDataLogger::SizeGuardWorker(void* pParam)
  1103. {
  1104. unsigned long long nElapsed;
  1105. char szMs[32];
  1106. CProcessClock pc;
  1107. CDataLogger *pThis = reinterpret_cast<CDataLogger*>(pParam);
  1108. ::pthread_mutex_lock(&pThis->m_condmtx1);
  1109. ::pthread_cond_signal(&pThis->m_cond1);
  1110. while(true)
  1111. {
  1112. pThis->m_bSGInProgress = false;
  1113. ::pthread_cond_wait(&pThis->m_cond1, &pThis->m_condmtx1);
  1114. pThis->m_bSGInProgress = true;
  1115. ::pthread_mutex_unlock(&pThis->m_condmtx1);
  1116. TRACE("Size guard start.\n");
  1117. pThis->Lock();
  1118. pc.ClockTrigger();
  1119. pThis->DoSizeGuard();
  1120. nElapsed = pc.ClockGetElapsed();
  1121. pThis->Unlock();
  1122. pThis->m_lf.Info("SG: finished in %s.\n", Ns2String(nElapsed, szMs, sizeof(szMs)));
  1123. TRACE("Size guard end - [%s].\n", szMs);
  1124. ::pthread_yield();
  1125. ::pthread_mutex_lock(&pThis->m_condmtx1);
  1126. }
  1127. return NULL;
  1128. }