Merge pull request #614 from ess-dmsc/meta_data

Meta data

changes.md

@@ -12,6 +12,7 @@
     ([#607](https://github.com/ess-dmsc/kafka-to-nexus/pull/607))
 - Fixed a bug where the file-writer will not abandon an alternative command topic if it fails to start a file-writing job.
 - The f142 value statistics written to file is now done so according to the NeXus format.
+- Added the HDF5/NeXus file structure to the "writing finished"-message and a system test to check this.
 
 
 ## Version 4.0.0: Many, many changes

src/CommandSystem/Handler.cpp

@@ -68,9 +68,10 @@ void Handler::revertCommandTopic() {
 }
 
 void Handler::sendHasStoppedMessage(std::string FileName,
-                                    std::string Metadata) {
+                                    nlohmann::json Metadata) {
+  Metadata["hdf_structure"] = NexusStructure;
   CommandResponse->publishStoppedMsg(ActionResult::Success, JobId, "", FileName,
-                                     Metadata);
+                                     Metadata.dump());
   PollForJob = true;
   revertCommandTopic();
 }
@@ -229,6 +230,7 @@ void Handler::handleStartCommand(FileWriter::Msg CommandMsg,
              JobId = StartJob.JobID;
              PollForJob = false;
              JobPool->disconnectFromPool();
+             NexusStructure = StartJob.NexusStructure;
            } catch (std::exception const &E) {
              PollForJob = true;
              JobId = "not_currently_writing";

src/CommandSystem/Handler.h

@@ -38,7 +38,7 @@ class HandlerBase {
   registerIsWritingFunction(IsWritingFuncType IsWritingFunction) = 0;
 
   virtual void sendHasStoppedMessage(std::string FileName,
-                                     std::string Metadata) = 0;
+                                     nlohmann::json Metadata) = 0;
   virtual void sendErrorEncounteredMessage(std::string FileName,
                                            std::string Metadata,
                                            std::string ErrorMessage) = 0;
@@ -61,7 +61,7 @@ class Handler : public HandlerBase {
   void registerIsWritingFunction(IsWritingFuncType IsWritingFunction) override;
 
   void sendHasStoppedMessage(std::string FileName,
-                             std::string Metadata) override;
+                             nlohmann::json Metadata) override;
   void sendErrorEncounteredMessage(std::string FileName, std::string Metadata,
                                    std::string ErrorMessage) override;
 
@@ -73,6 +73,7 @@ class Handler : public HandlerBase {
   void handleStopCommand(FileWriter::Msg CommandMsg);
   std::string const ServiceId;
   std::string JobId;
+  std::string NexusStructure;
   StartFuncType DoStart{
       [](auto) { throw std::runtime_error("Not implemented."); }};
   StopTimeFuncType DoSetStopTime{

src/Master.cpp

@@ -112,7 +112,7 @@ void Master::setToIdle() {
     }
     CurrentJSONStatus.update(StaticMetaData);
     CommandAndControl->sendHasStoppedMessage(CurrentFileName,
-                                             CurrentJSONStatus.dump());
+                                             CurrentJSONStatus);
   }
   CurrentStreamController.reset(nullptr);
   CurrentState = WriterState::Idle;

src/tests/MasterTests.cpp

@@ -24,7 +24,8 @@ class CommandHandlerStandIn : public Command::HandlerBase {
   MAKE_MOCK1(registerIsWritingFunction, void(Command::IsWritingFuncType),
              override);
 
-  MAKE_MOCK2(sendHasStoppedMessage, void(std::string, std::string), override);
+  MAKE_MOCK2(sendHasStoppedMessage, void(std::string, nlohmann::json),
+             override);
   MAKE_MOCK3(sendErrorEncounteredMessage,
              void(std::string, std::string, std::string), override);
 

system-tests/conftest.py

@@ -229,7 +229,7 @@ def start_file_writer(request):
 
 
 @pytest.fixture(scope="function", autouse=True)
-def writer_channel(request):
+def writer_channel(request) -> WorkerCommandChannel:
     """
     :type request: _pytest.python.FixtureRequest
     """

system-tests/requirements.txt

@@ -6,4 +6,4 @@ h5py>=3.1.0
 flatbuffers>=1.12
 black==19.3b0
 ess-streaming_data_types>=0.13.1
-file-writer-control>=1.1.7
+file-writer-control>=1.2.0

system-tests/test_filewriter_kafka_meta_data.py

@@ -0,0 +1,41 @@
+from helpers.kafkahelpers import (
+    create_producer,
+)
+from helpers.nexushelpers import OpenNexusFile
+from datetime import datetime, timedelta
+import numpy as np
+from file_writer_control.WriteJob import WriteJob
+from helpers.writer import (
+    wait_start_job,
+    wait_writers_available,
+    wait_no_working_writers,
+)
+
+
+def test_static_data_reaches_file(writer_channel, worker_pool, kafka_address):
+    wait_writers_available(writer_channel, nr_of=1, timeout=10)
+    now = datetime.now()
+    start_time = now - timedelta(seconds=10)
+    stop_time = now
+    file_name = "output_file_kafka_meta_data.nxs"
+    with open("commands/nexus_structure_static.json", "r") as f:
+        structure = f.read()
+    write_job = WriteJob(
+        nexus_structure=structure,
+        file_name=file_name,
+        broker=kafka_address,
+        start_time=start_time,
+        stop_time=stop_time,
+    )
+    wait_start_job(worker_pool, write_job, timeout=20)
+
+    wait_no_working_writers(writer_channel, timeout=30)
+    current_jobs = writer_channel.list_known_jobs()
+    for c_job in current_jobs:
+        if c_job.job_id == write_job.job_id:
+            assert "extra" in c_job.metadata
+            assert "hdf_structure" in c_job.metadata
+            assert c_job.metadata["hdf_structure"] == structure
+            return
+
+    assert False, "Unable to find current job."