engine: Stream json api data to a file

Author: tobifalk

engine/src/server.hpp

@@ -21,13 +21,15 @@
 
 #pragma once
 
+#include <memory>   // for unique_ptr
 #include <utility>  // for make_pair
 
-#include <cloe/handler.hpp>    // for Request, Response
-#include <cloe/registrar.hpp>  // for Registrar
+#include <cloe/handler.hpp>                         // for Request, Response
+#include <cloe/registrar.hpp>                       // for Registrar
+#include <cloe/utility/output_serializer_json.hpp>  // for JsonFileSerializer
 
-#include "stack.hpp"      // for ServerConf
 #include "oak/server.hpp"  // for Server, StaticRegistrar, ...
+#include "stack.hpp"       // for ServerConf
 
 namespace engine {
 
@@ -67,6 +69,11 @@ class Server {
  public:
   bool is_listening() { return server_.is_listening(); }
 
+  bool is_streaming() { return is_streaming_; }
+
+  /**
+   * Start the web server.
+   */
   void start() {
     assert(!is_listening());
 
@@ -77,13 +84,31 @@ class Server {
     server_.listen();
   }
 
+  /**
+   * Open a file for api data streaming. This does not require a running web
+   * server.
+   */
+  void init_stream(const std::string& filename) {
+    serializer_ = make_json_file_serializer(cloe::utility::JsonFileType::JSON_GZIP, logger());
+    serializer_->open_file(filename);
+  }
+
+  /**
+   * Stop all server-related procedures.
+   */
   void stop() {
     if (is_listening()) {
       logger()->info("Stopping server...");
       server_.stop();
     }
+    if (serializer_ != nullptr) {
+      serializer_->close_file();
+    }
   }
 
+  /**
+   * Register a list of all endpoints.
+   */
   void enroll(cloe::Registrar& r) {
     r.register_api_handler(
         "/endpoints", cloe::HandlerType::STATIC,
@@ -107,12 +132,32 @@ class Server {
   }
 
   /**
-   * Refresh the server buffer.
+   * Refresh and/or start streaming api data to a file.
+   */
+  void refresh_buffer_start_stream() {
+    is_streaming_ = serializer_ != nullptr;
+    if (is_listening() || is_streaming()) {
+      buffer_api_registrar_.refresh_buffer();
+    }
+    if (is_streaming()) {
+      // Write static endpoints at the beginning of the file.
+      write_data_stream(static_api_registrar_.endpoints());
+      write_data_stream(locked_api_registrar_.endpoints());
+      write_data_stream(buffer_api_registrar_.endpoints());
+    }
+  }
+
+  /**
+   * Refresh and/or write api data to a file.
    */
   void refresh_buffer() {
-    if (config.listen) {
+    if (is_listening() || is_streaming()) {
       buffer_api_registrar_.refresh_buffer();
     }
+    if (is_streaming()) {
+      write_data_stream(locked_api_registrar_.endpoints());
+      write_data_stream(buffer_api_registrar_.endpoints());
+    }
   }
 
   /**
@@ -123,12 +168,22 @@ class Server {
  protected:
   cloe::Logger logger() const { return cloe::logger::get("cloe"); }
 
+ private:
+  void write_data_stream(const std::vector<std::string>& endpoints) const {
+    auto j = server_.endpoints_to_json(endpoints);
+    if (!j.empty()) {
+      serializer_->serialize(j);
+    }
+  };
+
  private:  // State
   oak::Server server_;
   oak::StaticRegistrar static_registrar_{&server_, config.static_prefix, nullptr};
   oak::StaticRegistrar static_api_registrar_{&server_, config.api_prefix, nullptr};
   oak::LockedRegistrar locked_api_registrar_{&server_, config.api_prefix, nullptr};
   oak::BufferRegistrar buffer_api_registrar_{&server_, config.api_prefix, nullptr};
+  bool is_streaming_{false};
+  std::unique_ptr<cloe::utility::JsonFileSerializer> serializer_;
 };
 
 }  // namespace engine

engine/src/simulation.cpp

@@ -707,7 +707,7 @@ StateId SimulationMachine::Connect::impl(SimulationContext& ctx) {
   }
 
   ctx.progress.init_end();
-  ctx.server->refresh_buffer();
+  ctx.server->refresh_buffer_start_stream();
   logger()->info("Simulation initialization complete.");
   return START;
 }
@@ -1217,6 +1217,13 @@ SimulationResult Simulation::run() {
     if (config_.server.listen) {
       ctx.server->start();
     }
+    // Stream data to the requested file
+    if (r.config.engine.output_file_data_stream) {
+      auto filepath = r.get_output_filepath(*r.config.engine.output_file_data_stream);
+      if (is_writable(filepath)) {
+        ctx.server->init_stream(filepath.native());
+      }
+    }
 
     // Run pre-connect hooks
     ctx.commander->set_enabled(config_.engine.security_enable_hooks);

engine/src/stack.hpp

@@ -325,6 +325,7 @@ struct EngineConf : public Confable {
   boost::optional<boost::filesystem::path> output_file_config{"config.json"};
   boost::optional<boost::filesystem::path> output_file_result{"result.json"};
   boost::optional<boost::filesystem::path> output_file_triggers{"triggers.json"};
+  boost::optional<boost::filesystem::path> output_file_data_stream;
   bool output_clobber_files{true};
 
   /**
@@ -426,6 +427,7 @@ struct EngineConf : public Confable {
               {"config", make_schema(&output_file_config, file_proto(), "file to store config in")},
               {"result", make_schema(&output_file_result, file_proto(), "file to store simulation result in")},
               {"triggers", make_schema(&output_file_triggers, file_proto(), "file to store triggers in")},
+              {"api_recording", make_schema(&output_file_data_stream, file_proto(), "file to store api data stream")},
            }},
         }},
         {"triggers", Struct{

tests/setup_bats.bash

@@ -73,6 +73,20 @@ cloe_engine() {
     )
 }
 
+export cloe_tmp_registry="${HOME}/.cache/cloe/tmp-registry"
+
+cloe_engine_with_tmp_registry() {
+    test $1 == "run"
+    shift
+
+    (
+        export CLOE_WRITE_OUTPUT=true
+        export CLOE_TMP_REGISTRY=${cloe_tmp_registry}
+        export CLOE_OVERRIDE_ENV=(CLOE_TMP_REGISTRY)
+        cloe_engine run <(echo '{"version":"4","engine":{"registry_path":"${CLOE_TMP_REGISTRY}"}}') "$@"
+    )
+}
+
 cloe_shell() {
     local cloe_log_level="${CLOE_LOG_LEVEL-debug}"
     local cloe_profile_file="${CLOE_PROFILE_FILE-${CLOE_ROOT}/conantest.py}"

tests/test_engine_json_schema.json

@@ -683,6 +683,18 @@
             "files": {
               "additionalProperties": false,
               "properties": {
+                "api_recording": {
+                  "description": "file to store api data stream",
+                  "oneOf": [
+                    {
+                      "type": "null"
+                    },
+                    {
+                      "comment": "path should either not exist or be a file",
+                      "type": "string"
+                    }
+                  ]
+                },
                 "config": {
                   "description": "file to store config in",
                   "oneOf": [

tests/test_engine_json_schema_with_vtd.json

@@ -683,6 +683,18 @@
             "files": {
               "additionalProperties": false,
               "properties": {
+                "api_recording": {
+                  "description": "file to store api data stream",
+                  "oneOf": [
+                    {
+                      "type": "null"
+                    },
+                    {
+                      "comment": "path should either not exist or be a file",
+                      "type": "string"
+                    }
+                  ]
+                },
                 "config": {
                   "description": "file to store config in",
                   "oneOf": [

tests/test_engine_replica_smoketest.bats

@@ -6,25 +6,10 @@
 
 load setup_bats
 
-export cloe_tmp_registry="/tmp/cloe-ci-registry"
-
-cloe_engine_with_tmp_registry() {
-    test $1 == "run"
-    shift
-
-    (
-        export CLOE_WRITE_OUTPUT=true
-        export CLOE_TMP_REGISTRY=${cloe_tmp_registry}
-        export CLOE_OVERRIDE_ENV=(CLOE_TMP_REGISTRY)
-        cloe_engine run <(echo '{"version":"4","engine":{"registry_path":"${CLOE_TMP_REGISTRY}"}}') "$@"
-    )
-}
-
-
 @test "Expect exact replication: test_engine_replica_smoketest.json" {
     # Clean up in case temporary registry already exists.
     if [[ -d "$cloe_tmp_registry" ]]; then
-        rm -rf "$cloe_tmp_registry"
+        rm -r "$cloe_tmp_registry" || true
     fi
 
     # Run our initial smoketest, writing to the temporary registry.
@@ -64,5 +49,5 @@ cloe_engine_with_tmp_registry() {
     md5sum -c <(md5sum $cloe_tmp_registry/original/{config,triggers}.json | sed 's/original/replica-2/')
 
     # Remove the test registry.
-    rm -rf "$cloe_tmp_registry"
+    rm -r "$cloe_tmp_registry" || true
 }

tests/test_vtd.bats

@@ -18,6 +18,8 @@ teardown() {
         echo "Teardown VTD (from BATS)"
         teardown_vtd
     fi
+    # Remove the temporary registry.
+    rm -r "$cloe_tmp_registry" || true
 }
 
 @test "Expect check/run success: test_vtd_smoketest.json" {
@@ -28,6 +30,16 @@ teardown() {
     cloe_engine run test_vtd_smoketest.json
 }
 
+@test "Expect check/run success: test_vtd_api_recording.json" {
+    if ! test_vtd_plugin_exists; then
+        skip "required simulator vtd not present"
+    fi
+    cloe_engine check test_vtd_api_recording.json
+    cloe_engine_with_tmp_registry run test_vtd_api_recording.json
+    # Remove the test registry.
+    rm -r "$cloe_tmp_registry" || true
+}
+
 @test "Expect check/run success: test_vtd_watchdog.json" {
     if ! test_vtd_plugin_exists; then
         skip "required simulator vtd not present"

tests/test_vtd_api_recording.json

@@ -0,0 +1,23 @@
+{
+    "version": "4",
+    "include": [
+        "config_vtd_smoketest.json",
+        "bootstrap_vtd.json"
+    ],
+    "engine": {
+        "output": {
+            "files": {
+                "api_recording": "data.json.gz"
+            }
+        }
+    },
+    "server": {
+        "listen": false
+    },
+    "triggers": [
+        {
+            "event": "time=2",
+            "action": "succeed"
+        }
+    ]
+}