126 grapher registration with chronovisor

ChronoGrapher/ChronoGrapher.cpp

@@ -104,6 +104,7 @@ int main(int argc, char**argv)
     LOG_INFO("[ChronoGrapher] DataStoreAdminService started successfully.");
 
     // Instantiate GrapherRecordingService
+    chronolog::RecordingGroupId recording_group_id = 7;
     std::string RECORDING_SERVICE_PROTOCOL = confManager.GRAPHER_CONF.RECORDING_SERVICE_CONF.PROTO_CONF;
     std::string RECORDING_SERVICE_IP = confManager.GRAPHER_CONF.RECORDING_SERVICE_CONF.IP;
     uint16_t RECORDING_SERVICE_PORT = confManager.GRAPHER_CONF.RECORDING_SERVICE_CONF.BASE_PORT;
@@ -125,8 +126,8 @@ int main(int argc, char**argv)
     LOG_INFO("[ChronoGrapher] RecordingService started successfully.");
 
     // create GrapherIdCard to identify this Grapher process in ChronoVisor's Registry
-    chronolog::GrapherIdCard processIdCard(recording_endpoint.first, recording_endpoint.second
-                                         , recording_service_provider_id);
+    chronolog::GrapherIdCard processIdCard(recording_group_id, recording_endpoint.first, recording_endpoint.second,
+                                           recording_service_provider_id);
 
     std::stringstream process_id_string;
     process_id_string << processIdCard;

ChronoKeeper/CSVFileChunkExtractor.cpp

@@ -23,14 +23,18 @@ chronolog::CSVFileStoryChunkExtractor::~CSVFileStoryChunkExtractor()
 void chronolog::CSVFileStoryChunkExtractor::processStoryChunk(chronolog::StoryChunk*story_chunk)
 {
     std::ofstream chunk_fstream;
+
+    // chunk_filename: /rootDirectory/storyId.chunkStartTime.keeperIP.port.csv
+
     std::string chunk_filename(rootDirectory);
+    chunk_filename += "/" + std::to_string(story_chunk->getStoryId()) + "." + std::to_string(story_chunk->getStartTime() / 1000000000) + ".";
     keeperIdCard.getIPasDottedString(chunk_filename);
-    chunk_filename += "." + std::to_string(story_chunk->getStoryId()) + "." +
-                      std::to_string(story_chunk->getStartTime() / 1000000000) + ".csv";
+    chunk_filename += "." + std::to_string(keeperIdCard.getPort()) + ".csv";
 
     tl::xstream es = tl::xstream::self();
     LOG_INFO("[CSVFileStoryChunkExtractor] Processing StoryChunk: ES={}, ULT={}, StoryID={}, StartTime={}", es.get_rank()
          , tl::thread::self_id(), story_chunk->getStoryId(), story_chunk->getStartTime());
+
     // current thread if the only one that has this storyChunk and the only one that's writing to this chunk csv file 
     // thus no additional locking is needed ... 
     chunk_fstream.open(chunk_filename, std::ofstream::out|std::ofstream::app);

ChronoVisor/CMakeLists.txt

@@ -15,6 +15,7 @@ target_sources(chronovisor_server PRIVATE
     ./src/ClientRegistryRecord.cpp
     ./src/ChronicleMetaDirectory.cpp
     ./src/KeeperRegistry.cpp
+    ../chrono_common/ConfigurationManager.cpp
     ../ChronoAPI/ChronoLog/src/city.cpp
     ../ChronoAPI/ChronoLog/src/log.cpp)
 

ChronoVisor/include/ChronicleMetaDirectory.h

@@ -32,17 +32,15 @@ class ChronicleMetaDirectory
 
     int destroy_chronicle(const std::string &name);
 
-    //int create_story(std::string const& chronicle_name, const std::string& story_name,
-    //                 const std::unordered_map<std::string, std::string>& attrs);
     int destroy_story(std::string const &chronicle_name, const std::string &story_name);
 
     int
     acquire_story(chronolog::ClientId const &client_id, const std::string &chronicle_name, const std::string &story_name
-                  , const std::unordered_map <std::string, std::string> &attrs, int &flags, StoryId &, bool &);
+                  , const std::unordered_map <std::string, std::string> &attrs, int &flags, StoryId &);
 
     int
     release_story(chronolog::ClientId const &client_id, const std::string &chronicle_name, const std::string &story_name
-                  , StoryId &, bool &);
+                  , StoryId &);
 
     int get_chronicle_attr(std::string const &name, const std::string &key, std::string &value);
 

ChronoVisor/include/KeeperRegistry.h

@@ -1,16 +1,20 @@
 #ifndef KEEPER_REGISTRY_H
 #define KEEPER_REGISTRY_H
 
+#include <chrono>
+#include <map>
 #include <mutex>
+#include <random>
 #include <vector>
-#include <map>
-#include <chrono>
+
 #include <thallium.hpp>
 
 #include "chronolog_types.h"
 #include "KeeperIdCard.h"
 #include "KeeperStatsMsg.h"
 #include "KeeperRegistrationMsg.h"
+#include "GrapherIdCard.h"
+#include "GrapherRegistrationMsg.h"
 #include "ConfigurationManager.h"
 
 namespace chronolog
@@ -29,94 +33,148 @@ class DataStoreAdminClient;
 
 class KeeperRegistryService;
 
-class KeeperRegistry
+class KeeperProcessEntry
 {
+public:
+    KeeperProcessEntry(KeeperIdCard const& keeper_id_card, ServiceId const& admin_service_id)
+        : idCard(keeper_id_card)
+        , adminServiceId(admin_service_id)
+        , keeperAdminClient(nullptr)
+        , active(false)
+        , lastStatsTime(0)
+        , activeStoryCount(0)
+    {
+        idCardString += idCard;
+    }
+
+    KeeperProcessEntry(KeeperProcessEntry const& other) = default;
+    ~KeeperProcessEntry() = default;// Registry is reponsible for creating & deleting keeperAdminClient
+
+    KeeperIdCard idCard;
+    ServiceId adminServiceId;
+    std::string idCardString;
+    DataStoreAdminClient* keeperAdminClient;
+    bool active;
+    uint64_t lastStatsTime;
+    uint32_t activeStoryCount;
+    std::list<std::pair<std::time_t, DataStoreAdminClient*>> delayedExitClients;
+};
+
+class GrapherProcessEntry
+{
+public:
+    GrapherProcessEntry(GrapherIdCard const& id_card, ServiceId const& admin_service_id)
+        : idCard(id_card)
+        , adminServiceId(admin_service_id)
+        , adminClient(nullptr)
+        , active(false)
+        , lastStatsTime(0)
+        , activeStoryCount(0)
+    {
+        idCardString += idCard;
+    }
+
+    GrapherProcessEntry(GrapherProcessEntry const& other) = default;
+    ~GrapherProcessEntry() = default;// Registry is reponsible for creating & deleting keeperAdminClient
+
+    GrapherIdCard idCard;
+    ServiceId adminServiceId;
+    std::string idCardString;
+    DataStoreAdminClient* adminClient;
+    bool active;
+    uint64_t lastStatsTime;
+    uint32_t activeStoryCount;
+    std::list<std::pair<std::time_t, DataStoreAdminClient*>> delayedExitGrapherClients;
+    };
 
-    struct KeeperProcessEntry {
+
+    class RecordingGroup
+    {
     public:
-        KeeperProcessEntry(KeeperIdCard const& keeper_id_card, ServiceId const& admin_service_id)
-            : idCard(keeper_id_card)
-            , adminServiceId(admin_service_id)
-            , keeperAdminClient(nullptr)
-            , active(false)
-            , lastStatsTime(0)
-            , activeStoryCount(0)
+        RecordingGroup(RecordingGroupId group_id, GrapherProcessEntry* grapher_ptr = nullptr)
+            : groupId(group_id)
+            , activeKeeperCount(0)
+            , grapherProcess(grapher_ptr)
         {}
 
-        KeeperProcessEntry(KeeperProcessEntry const &other) = default;
+        RecordingGroup(RecordingGroup const& other) = default;
+        ~RecordingGroup() = default;
 
-        void reset()
-        {
-            keeperAdminClient = nullptr;
-            active = false;
-            lastStatsTime = 0;
-            activeStoryCount = 0;
-        }
-
-        ~KeeperProcessEntry() = default;   // Registry is reponsible for creating & deleting keeperAdminClient
-
-        KeeperIdCard idCard;
-        ServiceId adminServiceId;
-        DataStoreAdminClient* keeperAdminClient;
-        bool active;
-        uint64_t lastStatsTime;
-        uint32_t activeStoryCount;
-        std::list<std::pair<std::time_t, DataStoreAdminClient*>> delayedExitClients;
-    };
+        bool isActive() const;
+        void startDelayedGrapherExit(GrapherProcessEntry&, std::time_t);
+        void clearDelayedExitGrapher(GrapherProcessEntry&, std::time_t);
+        void startDelayedKeeperExit(KeeperProcessEntry&, std::time_t);
+        void clearDelayedExitKeeper(KeeperProcessEntry&, std::time_t);
 
-    enum RegistryState
-    {
-        UNKNOWN = 0, INITIALIZED = 1, // RegistryService is initialized, no active keepers
-        RUNNING = 2,    // RegistryService and active Keepers
-        SHUTTING_DOWN = 3    // Shutting down services
-    };
+        std::vector<KeeperIdCard>& getActiveKeepers(std::vector<KeeperIdCard>& keeper_id_cards);
 
-public:
-    KeeperRegistry(): registryState(UNKNOWN), registryEngine(nullptr), keeperRegistryService(nullptr)
-    {}
+        RecordingGroupId groupId;
+        size_t activeKeeperCount;
+        GrapherProcessEntry* grapherProcess;
+        std::map<std::pair<uint32_t, uint16_t>, KeeperProcessEntry> keeperProcesses;
+    };
 
-    ~KeeperRegistry();
+    class KeeperRegistry
+    {
+        enum RegistryState
+        {
+            UNKNOWN = 0,
+            INITIALIZED = 1, // RegistryService is initialized, no active keepers
+            RUNNING = 2,     // RegistryService and active Keepers
+            SHUTTING_DOWN = 3// Shutting down services
+        };
 
-    bool is_initialized() const
-    { return (INITIALIZED == registryState); }
+    public:
+        KeeperRegistry();
 
-    bool is_running() const
-    { return (RUNNING == registryState); }
+        ~KeeperRegistry();
 
-    bool is_shutting_down() const
-    { return (SHUTTING_DOWN == registryState); }
+        bool is_initialized() const { return (INITIALIZED == registryState); }
 
-    int InitializeRegistryService(ChronoLog::ConfigurationManager const &);
+        bool is_running() const { return (RUNNING == registryState); }
 
-    int ShutdownRegistryService();
+        bool is_shutting_down() const { return (SHUTTING_DOWN == registryState); }
 
-    int registerKeeperProcess(KeeperRegistrationMsg const &keeper_reg_msg);
+        int InitializeRegistryService(ChronoLog::ConfigurationManager const&);
 
-    int unregisterKeeperProcess(KeeperIdCard const &keeper_id_card);
+        int ShutdownRegistryService();
 
-    void updateKeeperProcessStats(KeeperStatsMsg const &keeperStatsMsg);
+        int registerKeeperProcess(KeeperRegistrationMsg const& keeper_reg_msg);
 
-    std::vector <KeeperIdCard> &getActiveKeepers(std::vector <KeeperIdCard> &keeper_id_cards);
+        int unregisterKeeperProcess(KeeperIdCard const& keeper_id_card);
 
-    int notifyKeepersOfStoryRecordingStart(std::vector<KeeperIdCard>&, ChronicleName const&, StoryName const&,
-                                           StoryId const&);
+        void updateKeeperProcessStats(KeeperStatsMsg const& keeperStatsMsg);
 
-    int notifyKeepersOfStoryRecordingStop(std::vector <KeeperIdCard> const &, StoryId const &);
+        int notifyRecordingGroupOfStoryRecordingStart(ChronicleName const&, StoryName const&, StoryId const&,
+                                                      std::vector<KeeperIdCard>&);
+        int notifyRecordingGroupOfStoryRecordingStop(StoryId const&);
 
+        int registerGrapherProcess(GrapherRegistrationMsg const& reg_msg);
+        int unregisterGrapherProcess(GrapherIdCard const& id_card);
 
-private:
+    private:
+        KeeperRegistry(KeeperRegistry const&) = delete;//disable copying
+        KeeperRegistry& operator=(KeeperRegistry const&) = delete;
 
-    KeeperRegistry(KeeperRegistry const &) = delete; //disable copying
-    KeeperRegistry &operator=(KeeperRegistry const &) = delete;
+        int notifyGrapherOfStoryRecordingStart(RecordingGroup&, ChronicleName const&, StoryName const&, StoryId const&,
+                                               uint64_t);
+        int notifyGrapherOfStoryRecordingStop(RecordingGroup&, StoryId const&);
+        int notifyKeepersOfStoryRecordingStart(RecordingGroup&, std::vector<KeeperIdCard>&, ChronicleName const&,
+                                               StoryName const&, StoryId const&, uint64_t);
+        int notifyKeepersOfStoryRecordingStop(RecordingGroup&, std::vector<KeeperIdCard> const&, StoryId const&);
 
-    RegistryState registryState;
-    std::mutex registryLock;
-    std::map <std::pair <uint32_t, uint16_t>, KeeperProcessEntry> keeperProcessRegistry;
-    thallium::engine*registryEngine;
-    KeeperRegistryService*keeperRegistryService;
-    size_t delayedDataAdminExitSeconds;
-};
+        RegistryState registryState;
+        std::mutex registryLock;
+        thallium::engine* registryEngine;
+        KeeperRegistryService* keeperRegistryService;
+        size_t delayedDataAdminExitSeconds;
 
+        std::map<RecordingGroupId, RecordingGroup> recordingGroups;
+        std::vector<RecordingGroup*> activeGroups;
+        std::mt19937 mt_random;//mersene twister random int generator
+        std::uniform_int_distribution<size_t> group_id_distribution;
+        std::map<StoryId, RecordingGroup*> activeStories;
+    };
 }
 
 #endif

ChronoVisor/include/KeeperRegistryService.h

@@ -2,13 +2,14 @@
 #define KEEPER_REGISTRY_SERVICE_H
 
 #include <iostream>
-//#include <margo.h>
 #include <thallium.hpp>
 #include <thallium/serialization/stl/string.hpp>
 
 #include "KeeperIdCard.h"
 #include "KeeperRegistrationMsg.h"
 #include "KeeperStatsMsg.h"
+#include "GrapherIdCard.h"
+#include "GrapherRegistrationMsg.h"
 #include "log.h"
 #include "KeeperRegistry.h"
 
@@ -53,13 +54,32 @@ class KeeperRegistryService: public tl::provider <KeeperRegistryService>
         theKeeperProcessRegistry.updateKeeperProcessStats(keeper_stats_msg);
     }
 
+    void register_grapher(tl::request const &request, chronolog::GrapherRegistrationMsg const & registrationMsg)
+    {
+        int return_code = 0;
+        std::stringstream ss;
+        ss << registrationMsg;
+        LOG_INFO("[KeeperRegistryService] register_grapher: {}", ss.str());
+        return_code = theKeeperProcessRegistry.registerGrapherProcess(registrationMsg);
+        request.respond(return_code);
+    }
+
+    void unregister_grapher(tl::request const &request, chronolog::GrapherIdCard const &id_card)
+    {
+        int return_code = 0;
+        return_code = theKeeperProcessRegistry.unregisterGrapherProcess(id_card);
+        request.respond(return_code);
+    }
+
     KeeperRegistryService(tl::engine &tl_engine, uint16_t service_provider_id, KeeperRegistry &keeperRegistry)
             : tl::provider <KeeperRegistryService>(tl_engine, service_provider_id), theKeeperProcessRegistry(
             keeperRegistry)
     {
         define("register_keeper", &KeeperRegistryService::register_keeper);
         define("unregister_keeper", &KeeperRegistryService::unregister_keeper);
         define("handle_stats_msg", &KeeperRegistryService::handle_stats_msg, tl::ignore_return_value());
+        define("register_grapher", &KeeperRegistryService::register_grapher);
+        define("unregister_grapher", &KeeperRegistryService::unregister_grapher);
         //setup finalization callback in case this ser vice provider is still alive when the engine is finalized
         get_engine().push_finalize_callback(this, [p = this]()
         { delete p; });