Merge pull request hpcc-systems#19607 from ghalliday/eventlogging

HPCC-33609 Create a POC for an Event Recording framework

Reviewed-By: Tim Klemm <tim.klemm@lexisnexisrisk.com>
Reviewed-by: Jake Smith <jake.smith@lexisnexisrisk.com>
Merged-by: Gavin Halliday <ghalliday@hpccsystems.com>

Author: ghalliday

roxie/ccd/ccdmain.cpp

@@ -1162,6 +1162,13 @@ int CCD_API roxie_main(int argc, const char *argv[], const char * defaultYaml)
         else
             multicastTTL = ttlTmp;
 
+        if (topology->getPropBool("@recordAllEvents", false))
+        {
+            const char * recordEventOptions = topology->queryProp("@recordEventOptions");
+            const char * optRecordEventFilename = topology->queryProp("@recordEventFilename");
+            startRoxieEventRecording(recordEventOptions, optRecordEventFilename);
+        }
+
         workunitGraphCacheEnabled = topology->getPropBool("expert/@workunitGraphCacheEnabled", workunitGraphCacheEnabled);
 
         indexReadChunkSize = topology->getPropInt("@indexReadChunkSize", 60000);
@@ -1766,6 +1773,7 @@ int CCD_API roxie_main(int argc, const char *argv[], const char * defaultYaml)
 #ifndef _CONTAINERIZED
     stopPerformanceMonitor();
 #endif
+    stopRoxieEventRecording();
     cleanupPlugins();
     unloadHpccProtocolPlugin();
     closeMulticastSockets();

roxie/ccd/ccdstate.cpp

@@ -22,6 +22,7 @@
 #include "jhash.hpp"
 #include "jsort.hpp"
 #include "jregexp.hpp"
+#include "jevent.hpp"
 
 #include "udptopo.hpp"
 #include "ccd.hpp"
@@ -241,6 +242,45 @@ void stopDelayedReleaser()
 }
 
 
+//-------------------------------------------------------------------------
+
+void startRoxieEventRecording(const char * options, const char * filename)
+{
+    if (isEmptyString(options))
+        options = "threadid";
+
+    StringBuffer outputFilename;
+    const char * path = filename;
+    if (!isAbsolutePath(filename))
+    {
+        getTempFilePath(outputFilename, "eventrecorder", nullptr);
+        outputFilename.append(PATHSEPCHAR);
+        if (!isEmptyString(filename))
+        {
+            outputFilename.append(filename);
+        }
+        else
+        {
+            //MORE: Revisit this at a later date
+            unsigned seq = (unsigned)(get_cycles_now() % 100000);
+            outputFilename.append("roxieevents.").append((unsigned)GetCurrentProcessId()).append(".").append(seq).append(".evt");
+        }
+
+        path = outputFilename.str();
+        //MORE: The caller will need to know the full pathname
+    }
+
+    recursiveCreateDirectoryForFile(path);
+    queryRecorder().startRecording(options, path);
+
+    //MORE: Add all the meta information for the files - either now or in stopRoxieEventRecording()
+}
+
+void stopRoxieEventRecording()
+{
+    queryRecorder().stopRecording();
+}
+
 //-------------------------------------------------------------------------
 
 class CSimpleSuperFileArray : implements ISimpleSuperFileEnquiry, public CInterface

roxie/ccd/ccdstate.hpp

@@ -171,7 +171,7 @@ extern const char *queryNodeIndexName(const IPropertyTree &graphNode, ThorActivi
 extern void createDelayedReleaser();
 extern void stopDelayedReleaser();
 
-extern void createDelayedReleaser();
-extern void stopDelayedReleaser();
+extern void startRoxieEventRecording(const char * optionsText, const char * filename);
+extern void stopRoxieEventRecording();
 
 #endif

roxie/roxiemem/roxiemem.cpp

@@ -7054,7 +7054,7 @@ class SimpleRowBuffer : implements IBufferedRowCallback
 };
 
 // A row buffer which does not allocate memory for the row array from roxiemem
-class TestingRowBuffer : implements IBufferedRowCallback
+class TestingRowBuffer final : implements IBufferedRowCallback
 {
 public:
     TestingRowBuffer(unsigned _cost, unsigned _id) : cost(_cost), id(_id)
@@ -7136,7 +7136,7 @@ class CallbackBlockAllocator : implements IBufferedRowCallback
 
 
 //Free the block as soon as requested
-class SimpleCallbackBlockAllocator : public CallbackBlockAllocator
+class SimpleCallbackBlockAllocator final : public CallbackBlockAllocator
 {
 public:
     SimpleCallbackBlockAllocator(IRowManager * _rowManager, memsize_t _size, unsigned _cost, unsigned _id)
@@ -7154,7 +7154,7 @@ class SimpleCallbackBlockAllocator : public CallbackBlockAllocator
 };
 
 //Allocate another row before disposing of the first
-class NastyCallbackBlockAllocator : public CallbackBlockAllocator
+class NastyCallbackBlockAllocator final : public CallbackBlockAllocator
 {
 public:
     NastyCallbackBlockAllocator(IRowManager * _rowManager, unsigned _size, unsigned _cost, unsigned _id)

system/jhtree/jhtree.cpp

@@ -64,6 +64,7 @@
 #include "rtldynfield.hpp"
 #include "eclhelper_base.hpp"
 #include "jmetrics.hpp"
+#include "jevent.hpp"
 
 constexpr __uint64 defaultFetchThresholdNs = 20000; // Assume anything < 20us comes from the page cache, everything above probably went to disk
 
@@ -671,23 +672,51 @@ class DelayedCacheEntryReleaser : public IRemovedMappingCallback
 public:
     ~DelayedCacheEntryReleaser()
     {
+        if (unlikely(recordingEvents()))
+        {
+            for (unsigned i1 = 0; i1 < numFixed; i1++)
+                noteEviction(fixedFileId[i1], fixedPending[i1]);
+
+            ForEachItemIn(i2, pending)
+                noteEviction(pendingFileIds.item(i2), &pending.item(i2));
+        }
+
         for (unsigned i=0; i < numFixed; i++)
             fixedPending[i]->Release();
     }
     virtual void noteRemoval(void * _mapping) override
     {
         CNodeMapping *mapping = reinterpret_cast<CNodeMapping *>(_mapping);
-        //Save the node onto a list, so it will be released when this object is released.
+        unsigned keyId = mapping->queryFindValue().keyId;
+        // Save the node onto a list, so it will be released when this object is released.
         CJHTreeNode * node = const_cast<CJHTreeNode *>(mapping->query().getNode());
+        // The key id needs to be stored separately because the node does not currently contain it.
         if (numFixed < maxFixed)
-            fixedPending[numFixed++] = node;
+        {
+            fixedPending[numFixed] = node;
+            fixedFileId[numFixed] = keyId;
+            numFixed++;
+        }
         else
+        {
             pending.append(*node);
+            pendingFileIds.append(keyId);
+        }
     }
+
+protected:
+    void noteEviction(unsigned keyId, CJHTreeNode * node)
+    {
+        offset_t pos = node->getFpos();
+        queryRecorder().recordIndexEviction(keyId, pos, node->getNodeType(), node->getMemSize());
+    }
+
 protected:
-    CIArray pending;
+    CIArrayOf<CJHTreeNode> pending;
+    UnsignedArray pendingFileIds;
     //Use a fixed array for a small number of allocations to avoid a heap allocation inside the critsec
     CJHTreeNode * fixedPending[maxFixed]; // deliberately uninitialized
+    unsigned fixedFileId[maxFixed]; // deliberately uninitialized
     unsigned numFixed = 0;
 
 };
@@ -2969,6 +2998,8 @@ const CJHTreeNode *CNodeCache::getCachedNode(const INodeLoader *keyIndex, unsign
                 block.leave();
 
                 //Update ctx stats outside of the critical section.
+                if (unlikely(recordingEvents()))
+                    queryRecorder().recordIndexLookup(iD, pos, type, true);
                 if (ctx) ctx->noteStatistic(hitStatId[cacheType], 1);
                 return fastPathMatch;
             }
@@ -2986,6 +3017,9 @@ const CJHTreeNode *CNodeCache::getCachedNode(const INodeLoader *keyIndex, unsign
         ownedCacheEntry.setown(cacheEntry);
     }
 
+    if (unlikely(recordingEvents()))
+        queryRecorder().recordIndexLookup(iD, pos, type, false);
+
     //If an exception is thrown before the node is cleanly loaded we need to remove the partially constructed
     //node from the cache otherwise it may never get loaded, and can prevent items being removed from the cache
     //note: noteStatistic() can throw an exception if a worker has been aborted...
@@ -3041,6 +3075,10 @@ const CJHTreeNode *CNodeCache::getCachedNode(const INodeLoader *keyIndex, unsign
         }
         cycle_t endLoadCycles = get_cycles_now();
         cycle_t actualLoadCycles = endLoadCycles - startLoadCycles;
+
+        if (unlikely(recordingEvents()))
+            queryRecorder().recordIndexLoad(iD, pos, type, ownedCacheEntry->queryNode()->getMemSize(), cycle_to_nanosec(actualLoadCycles), cycle_to_nanosec(fetchCycles));
+
         if (actualLoadCycles > traceNodeLoadThreshold)
         {
             if ((endLoadCycles - lastLoadReportCycles) >= traceNodeLoadFrequency)

system/jlib/CMakeLists.txt

@@ -63,6 +63,7 @@ set (    SRCS
          jcrc.cpp
          jdebug.cpp
          jencrypt.cpp
+         jevent.cpp
          jexcept.cpp
          jfile.cpp
          jflz.cpp
@@ -127,6 +128,7 @@ set (    INCLUDES
         jelogtype.hpp
         jencrypt.hpp
         jerror.hpp
+        jevent.hpp
         jexcept.hpp
         jfcmp.hpp
         jfile.hpp