Test Coverage and mruby logging

CMakeLists.txt

@@ -2,7 +2,7 @@
 set(AGENT_VERSION_MAJOR 2)
 set(AGENT_VERSION_MINOR 7)
 set(AGENT_VERSION_PATCH 0)
-set(AGENT_VERSION_BUILD 12)
+set(AGENT_VERSION_BUILD 13)
 set(AGENT_VERSION_RC "")
 
 # This minimum version is to support Visual Studio 2019 and C++ feature checking and FetchContent

src/mtconnect/agent.cpp

@@ -124,6 +124,12 @@ namespace mtconnect {
     m_printers["xml"] = make_unique<printer::XmlPrinter>(m_pretty, m_validation);
     m_printers["json"] = make_unique<printer::JsonPrinter>(jsonVersion, m_pretty, m_validation);
 
+    if (!m_schemaVersion)
+    {
+      m_xmlParser->parseFile(m_deviceXmlPath, dynamic_cast<printer::XmlPrinter *>(m_printers["xml"].get()));
+      m_schemaVersion = m_xmlParser->getSchemaVersion();
+    }
+
     if (m_schemaVersion)
     {
       m_intSchemaVersion = IntSchemaVersion(*m_schemaVersion);

src/mtconnect/configuration/agent_config.cpp

@@ -809,6 +809,10 @@ namespace mtconnect::configuration {
       throw;
     }
 
+    // Expand $VAR and ${VAR} references from sibling config values and the
+    // environment before any options are read.
+    expandConfigVariables(config);
+
     if (m_logChannels.empty())
     {
       configureLogger(config);
@@ -970,7 +974,19 @@ namespace mtconnect::configuration {
     // Make the PipelineContext
     m_pipelineContext = std::make_shared<pipeline::PipelineContext>();
     m_pipelineContext->m_contract = m_agent->makePipelineContract();
-
+    if (!HasOption(options, configuration::SchemaVersion))
+    {
+      if (m_agent->getSchemaVersion())
+      {
+        options[configuration::SchemaVersion] = *m_agent->getSchemaVersion();
+      }
+      else
+      {
+        options[configuration::SchemaVersion] = std::format("{}.{}", std::to_string(AGENT_VERSION_MAJOR),
+                                                            std::to_string(AGENT_VERSION_MINOR));
+
+      }
+    }
     loadSinks(config, options);
 
     m_agent->initialize(m_pipelineContext);

src/mtconnect/configuration/agent_config.hpp

@@ -272,6 +272,14 @@ namespace mtconnect {
       ///@param path the path to set for the config file directory
       void setConfigPath(const std::filesystem::path &path) { m_configPath = path; }
 
+      /// @brief Expand `$VAR` and `${VAR}` references in the config tree.
+      ///
+      /// Each value is resolved against sibling config values that precede it and,
+      /// failing that, against environment variables. Unresolved references are left
+      /// in place. Called by loadConfig() before any options are parsed.
+      /// @param config the parsed configuration tree to expand in place
+      void expandConfigVariables(boost::property_tree::ptree &config);
+
     protected:
       DevicePtr getDefaultDevice();
       void loadAdapters(const ptree &tree, const ConfigOptions &options);
@@ -349,8 +357,6 @@ namespace mtconnect {
         }
       }
 
-      void expandConfigVariables(boost::property_tree::ptree &);
-
     protected:
       using text_sink = boost::log::sinks::synchronous_sink<boost::log::sinks::text_file_backend>;
       using console_sink =

src/mtconnect/pipeline/message_mapper.hpp

@@ -90,8 +90,21 @@ namespace mtconnect::pipeline {
             msg = *topic;
           else
             msg = "unknown topic";
-          LOG(error) << "Cannot find data item for topic: " << msg
-                     << " and data: " << data->getValue<std::string>();
+
+          // Render the value as a string in a type-safe way. The value is known to not
+          // be a string here, so convert a copy to avoid throwing on bad_variant_access.
+          std::string valueStr;
+          try
+          {
+            entity::Value v = data->getValue();
+            entity::ConvertValueToType(v, entity::ValueType::STRING);
+            valueStr = std::get<std::string>(v);
+          }
+          catch (...)
+          {
+            valueStr = "<non-string value>";
+          }
+          LOG(error) << "Cannot find data item for topic: " << msg << " and data: " << valueStr;
         }
         return nullptr;
       }

src/mtconnect/ruby/ruby_observation.hpp

@@ -207,4 +207,38 @@ namespace mtconnect::ruby {
           MRB_ARGS_REQ(1));
     }
   };
+
+  /// @struct RubyObservation
+  /// @remark Ruby Observation Wrapper
+  /// @code
+  /// class Observation -> mtconnect::observation::Observation
+  ///   def self.make(data_item, properties, timestamp = now) -> Observation::make(...)
+  ///   def initialize(data_item, properties, timestamp = now) -> Observation::make(...)
+  ///   def dup -> mtconnect::observation::Observation::copy()
+  ///   def data_item -> mtconnect::observation::Observation::getDataItem()
+  ///   def timestamp -> mtconnect::observation::Observation::getTimestamp()
+  /// end
+  ///
+  /// class Condition -> mtconnect::observation::Condition
+  ///   def level -> mtconnect::observation::Condition::getLevel()
+  ///   def level=(level) -> mtconnect::observation::Condition::setLevel(level)
+  /// end
+  /// @endcode
+  ///
+  /// @note Both `make` and `initialize` go through Observation::make, which
+  ///       decides whether the observation is unavailable before the object is
+  ///       built. Two things follow that are easy to trip over from ruby:
+  ///
+  ///       1. An `UNAVAILABLE` value marks the observation unavailable and is
+  ///          then erased, so the observation does not carry the literal text.
+  ///          The match is case insensitive, and it is the `level` property for
+  ///          a condition and `VALUE` for everything else.
+  ///       2. **Omitting the property has the same effect.** A condition with no
+  ///          `level`, or any other observation with no `VALUE`, is silently
+  ///          unavailable rather than an error. A transform that means to report
+  ///          a value and passes an empty properties hash by mistake produces an
+  ///          unavailable observation, not a failure, so the mistake shows up as
+  ///          missing data rather than as a raised exception.
+  ///
+  ///       See Observation::make in observation.cpp.
 }  // namespace mtconnect::ruby

src/mtconnect/ruby/ruby_transform.hpp

@@ -325,6 +325,11 @@ namespace mtconnect::ruby {
         {
           // auto str = mrb_any_to_s(mrb, rv);
           LOG(error) << "Error in transform: " << mrb_str_to_cstr(mrb, mrb_inspect(mrb, rv));
+          if (mrb->exc)
+          {
+            auto exc = mrb_funcall(mrb, mrb_obj_value(mrb->exc), "to_s", 0);
+            LOG(error) << "Error calling transform " << m_name << ": " << RSTRING_CSTR(mrb, exc);
+          }
           rv = mrb_nil_value();
         }
 

src/mtconnect/ruby/ruby_vm.hpp

@@ -66,9 +66,7 @@ namespace mtconnect::ruby {
     {
       mrb_value msg;
       mrb_get_args(mrb, "S", &msg);
-      BOOST_LOG_STREAM_WITH_PARAMS(::boost::log::trivial::logger::get(),
-                                   (::boost::log::keywords::severity = level))
-          << mrb_str_to_cstr(mrb, msg);
+      BOOST_LOG_SEV(agent_logger::get(), level) <<  mrb_str_to_cstr(mrb, msg);
     }
 
     void defineLogger()

src/mtconnect/sink/rest_sink/rest_service.cpp

@@ -795,11 +795,11 @@ namespace mtconnect {
           "deviceType={string}&format={string}");
       m_server->addRouting({boost::beast::http::verb::get, "/current?" + qp, handler})
           .document("MTConnect current request",
-                    "Gets a stapshot of the state of all the observations for all devices "
+                    "Gets a snapshot of the state of all the observations for all devices "
                     "optionally filtered by the `path`");
       m_server->addRouting({boost::beast::http::verb::get, "/{device}/current?" + qp, handler})
           .document("MTConnect current request",
-                    "Gets a stapshot of the state of all the observations for device `device` "
+                    "Gets a snapshot of the state of all the observations for device `device` "
                     "optionally filtered by the `path`")
           .command("current");
     }

src/mtconnect/source/adapter/agent_adapter/agent_adapter.cpp

@@ -155,6 +155,8 @@ namespace mtconnect::source::adapter::agent_adapter {
       LOG(error) << "Agent adapter cannot start: fatal error";
       return false;
     }
+
+    LOG(debug) << "Starting agent adapter for " << m_identity;
     m_pipeline.start();
 
     if (m_url.m_protocol == "https")
@@ -201,6 +203,7 @@ namespace mtconnect::source::adapter::agent_adapter {
 
   void AgentAdapter::clear()
   {
+    LOG(debug) << "Clearing adapter state for " << m_url;
     m_streamRequest.reset();
     m_assetRequest.reset();
 
@@ -237,9 +240,15 @@ namespace mtconnect::source::adapter::agent_adapter {
             // request's `from` is baked in when the stream opens and never advances,
             // so replaying it re-delivers every observation seen during the session.
             if (self->canRecover() && self->m_streamRequest)
+            {
+              LOG(info) << "Attempting to recover stream for " << self->m_url;
               self->recover();
+            }
             else
+            {
+              LOG(info) << "Cannot recover stream for " << self->m_url << ", restarting session";
               self->run();
+            }
           }
         }));
   }
@@ -253,6 +262,7 @@ namespace mtconnect::source::adapter::agent_adapter {
           m_strand, [weak = std::weak_ptr(getptr())](boost::system::error_code ec) {
             if (auto self = weak.lock(); self && !ec && self->m_assetRequest)
             {
+              LOG(debug) << "Recovering asset request for " << self->m_url;
               self->m_assetSession->makeRequest(*self->m_assetRequest);
             }
           }));
@@ -269,18 +279,21 @@ namespace mtconnect::source::adapter::agent_adapter {
       switch (static_cast<ErrorCode>(ec.value()))
       {
         case ErrorCode::ADAPTER_FAILED:
+          LOG(debug) << "Assets Failed, adapter failed for " << m_url;
           stop();
           if (m_handler->m_disconnected)
             m_handler->m_disconnected(m_identity);
           m_pipeline.getContext()->m_contract->sourceFailed(m_identity);
           break;
 
         case ErrorCode::RETRY_REQUEST:
+          LOG(debug) << "Assets Failed, Retrying asset request for " << m_url;
           recoverAssetRequest();
           break;
 
         default:
-
+          LOG(debug) << "Assets Failed, unknown failure message " << ec.message() << " for "
+                     << m_identity;
           break;
       }
     }
@@ -291,13 +304,15 @@ namespace mtconnect::source::adapter::agent_adapter {
     if (m_stopped)
       return;
 
+    LOG(debug) << "Streams Failed: " << ec.message() << " for " << m_url;
     if (ec.category() == source::TheErrorCategory())
     {
       switch (static_cast<ErrorCode>(ec.value()))
       {
         case ErrorCode::INSTANCE_ID_CHANGED:
         case ErrorCode::RESTART_STREAM:
         {
+          LOG(debug) << "Streams must be restarted for " << m_url;
           if (m_handler->m_disconnected)
             m_handler->m_disconnected(m_identity);
           clear();
@@ -306,29 +321,33 @@ namespace mtconnect::source::adapter::agent_adapter {
         }
 
         case ErrorCode::RETRY_REQUEST:
+          LOG(debug) << "Streams must be recovered for " << m_url;
           recoverStreams();
           break;
 
         case ErrorCode::STREAM_CLOSED:
+          LOG(debug) << "Streams must be recovered for " << m_url;
           if (m_handler->m_disconnected)
             m_handler->m_disconnected(m_identity);
           recoverStreams();
           break;
 
         case ErrorCode::ADAPTER_FAILED:
+          LOG(debug) << "Stopping adapter for " << m_url;
           stop();
           if (m_handler->m_disconnected)
             m_handler->m_disconnected(m_identity);
           m_pipeline.getContext()->m_contract->sourceFailed(m_identity);
           break;
 
         case ErrorCode::MULTIPART_STREAM_FAILED:
+          LOG(debug) << "Multipart stream failed, switching to polling for " << m_url;
           m_usePolling = true;
           recoverStreams();
           break;
 
         default:
-          LOG(error) << "Unknown error: " << ec.message();
+          LOG(error) << "Unknown error: " << ec.message() << " for " << m_url;
           break;
       }
     }
@@ -340,6 +359,7 @@ namespace mtconnect::source::adapter::agent_adapter {
       switch (static_cast<beast::http::error>(ec.value()))
       {
         case beast::http::error::end_of_stream:
+          LOG(debug) << "Streams Failed, end of stream for " << m_url << ", attempting to recover";
           recoverStreams();
           break;
 
@@ -354,6 +374,7 @@ namespace mtconnect::source::adapter::agent_adapter {
     if (m_stopped)
       return;
 
+    LOG(debug) << "Starting session for " << m_url;
     clear();
 
     if (m_probeAgent)
@@ -372,6 +393,7 @@ namespace mtconnect::source::adapter::agent_adapter {
     if (m_stopped)
       return;
 
+    LOG(debug) << "Recovering session for " << m_url;
     sample();
   }
 
@@ -380,6 +402,7 @@ namespace mtconnect::source::adapter::agent_adapter {
     if (m_stopped)
       return false;
 
+    LOG(debug) << "Requesting probe for " << m_url;
     m_streamRequest.emplace(m_sourceDevice, "probe", UrlQuery(), false, [this]() {
       m_agentVersion = m_feedback.m_agentVersion;
       assets();
@@ -393,6 +416,7 @@ namespace mtconnect::source::adapter::agent_adapter {
     if (m_stopped)
       return false;
 
+    LOG(debug) << "Requesting current for " << m_url;
     m_streamRequest.emplace(m_sourceDevice, "current", UrlQuery(), false,
                             [this]() { return sample(); });
     return m_session->makeRequest(*m_streamRequest);
@@ -405,6 +429,7 @@ namespace mtconnect::source::adapter::agent_adapter {
 
     if (m_usePolling)
     {
+      LOG(debug) << "Starting polling sample stream for " << m_url << " from " << m_feedback.m_next;
       using namespace boost;
       UrlQuery query({{"from", lexical_cast<string>(m_feedback.m_next)},
                       {"count", lexical_cast<string>(m_count)}});
@@ -423,6 +448,8 @@ namespace mtconnect::source::adapter::agent_adapter {
     }
     else
     {
+      LOG(debug) << "Starting long poll sample stream for " << m_url << " from "
+                 << m_feedback.m_next;
       using namespace boost;
       UrlQuery query({{"from", lexical_cast<string>(m_feedback.m_next)},
                       {"count", lexical_cast<string>(m_count)},
@@ -437,6 +464,8 @@ namespace mtconnect::source::adapter::agent_adapter {
 
   void AgentAdapter::stop()
   {
+    LOG(debug) << "Stopping agent adapter for " << m_url;
+
     m_stopped = true;
     clear();
     if (m_session)
@@ -469,7 +498,7 @@ namespace mtconnect::source::adapter::agent_adapter {
                    back_inserter(idList),
                    [](const EntityPtr entity) { return entity->getValue<string>(); });
     string ids = boost::join(idList, ";");
-
+    LOG(debug) << "Updating assets with ids: " << ids << " for " << m_url;
     m_assetRequest.emplace(nullopt, "assets/" + ids, UrlQuery(), false, [this]() {
       m_assetRequest.reset();
       return true;

src/mtconnect/utilities.hpp

@@ -997,7 +997,7 @@ namespace mtconnect {
       /// @brief Format the URL as text
       /// @param device optional device to add to the URL
       /// @return formatted URL
-      std::string getUrlText(const std::optional<std::string> &device)
+      std::string getUrlText(const std::optional<std::string> &device) const
       {
         std::stringstream url;
         url << m_protocol << "://" << getHost() << ':' << getPort() << getTarget();
@@ -1010,5 +1010,15 @@ namespace mtconnect {
       /// @return parsed URL
       static Url parse(const std::string_view &url);
     };
+
+    /// @brief output operator for URL
+    /// @param os the output stream
+    /// @param url the URL to output
+    inline std::ostream &operator<<(std::ostream &os, const Url &url)
+    {
+      os << url.getUrlText(std::nullopt);
+      return os;
+    }
+
   }  // namespace url
 }  // namespace mtconnect