MINIFICPP-2613 - Move base metrics to libminifi

Signed-off-by: Gabor Gyimesi <gamezbird@gmail.com>

This closes #2020

Author: adamdebreceni

core-framework/include/core/ProcessorImpl.h

@@ -33,8 +33,7 @@
 #include "minifi-cpp/core/Annotation.h"
 #include "minifi-cpp/core/DynamicProperty.h"
 #include "minifi-cpp/core/Scheduling.h"
-#include "minifi-cpp/core/state/nodes/MetricsBase.h"
-#include "minifi-cpp/core/ProcessorMetrics.h"
+#include "minifi-cpp/core/ProcessorMetricsExtension.h"
 #include "minifi-cpp/utils/gsl.h"
 #include "utils/Id.h"
 #include "minifi-cpp/core/OutputAttributeDefinition.h"
@@ -118,8 +117,8 @@ class ProcessorImpl : public virtual ProcessorApi {
 
   annotation::Input getInputRequirement() const override = 0;
 
-  gsl::not_null<std::shared_ptr<ProcessorMetrics>> getMetrics() const override {
-    return metrics_;
+  std::shared_ptr<ProcessorMetricsExtension> getMetricsExtension() const override {
+    return metrics_extension_;
   }
 
   static constexpr auto DynamicProperties = std::array<DynamicProperty, 0>{};
@@ -142,7 +141,7 @@ class ProcessorImpl : public virtual ProcessorApi {
 
   std::atomic<bool> trigger_when_empty_;
 
-  gsl::not_null<std::shared_ptr<ProcessorMetrics>> metrics_;
+  std::shared_ptr<ProcessorMetricsExtension> metrics_extension_;
 
   std::shared_ptr<logging::Logger> logger_;
 

core-framework/src/core/ProcessorImpl.cpp

@@ -36,7 +36,6 @@
 #include "range/v3/algorithm/any_of.hpp"
 #include "fmt/format.h"
 #include "minifi-cpp/Exception.h"
-#include "core/ProcessorMetrics.h"
 #include "minifi-cpp/core/ProcessorDescriptor.h"
 
 using namespace std::literals::chrono_literals;
@@ -46,7 +45,6 @@ namespace org::apache::nifi::minifi::core {
 ProcessorImpl::ProcessorImpl(ProcessorMetadata metadata)
     : metadata_(std::move(metadata)),
       trigger_when_empty_(false),
-      metrics_(std::make_shared<ProcessorMetricsImpl>(*this)),
       logger_(metadata_.logger) {
   logger_->log_debug("Processor {} created with uuid {}", getName(), getUUIDStr());
 }

extensions/llamacpp/processors/RunLlamaCppInference.cpp

@@ -64,13 +64,13 @@ void RunLlamaCppInference::onSchedule(core::ProcessContext& context, core::Proce
 }
 
 void RunLlamaCppInference::increaseTokensIn(uint64_t token_count) {
-  auto* const llamacpp_metrics = dynamic_cast<RunLlamaCppInferenceMetrics*>(metrics_.get());
+  auto* const llamacpp_metrics = dynamic_cast<RunLlamaCppInferenceMetrics*>(metrics_extension_.get());
   gsl_Assert(llamacpp_metrics);
   llamacpp_metrics->tokens_in += token_count;
 }
 
 void RunLlamaCppInference::increaseTokensOut(uint64_t token_count) {
-  auto* const llamacpp_metrics = dynamic_cast<RunLlamaCppInferenceMetrics*>(metrics_.get());
+  auto* const llamacpp_metrics = dynamic_cast<RunLlamaCppInferenceMetrics*>(metrics_extension_.get());
   gsl_Assert(llamacpp_metrics);
   llamacpp_metrics->tokens_out += token_count;
 }

extensions/llamacpp/processors/RunLlamaCppInference.h

@@ -24,36 +24,34 @@
 #include "core/logging/LoggerFactory.h"
 #include "core/PropertyDefinitionBuilder.h"
 #include "LlamaContext.h"
-#include "core/ProcessorMetrics.h"
+#include "minifi-cpp/core/ProcessorMetricsExtension.h"
+#include "core/state/Value.h"
 
 namespace org::apache::nifi::minifi::extensions::llamacpp::processors {
 
 using LlamaContextProvider =
   std::function<std::unique_ptr<LlamaContext>(const std::filesystem::path& model_path, const LlamaSamplerParams& llama_sampler_params, const LlamaContextParams& llama_ctx_params)>;
 
-class RunLlamaCppInferenceMetrics : public core::ProcessorMetricsImpl {
+class RunLlamaCppInferenceMetrics : public core::ProcessorMetricsExtension {
  public:
-  explicit RunLlamaCppInferenceMetrics(const core::ProcessorImpl& source_processor)
-  : core::ProcessorMetricsImpl(source_processor) {
-  }
+  RunLlamaCppInferenceMetrics() = default;
 
   std::vector<state::response::SerializedResponseNode> serialize() override {
-    auto resp = core::ProcessorMetricsImpl::serialize();
-    auto& root_node = resp[0];
+    std::vector<state::response::SerializedResponseNode> resp;
 
     state::response::SerializedResponseNode tokens_in_node{"TokensIn", tokens_in.load()};
-    root_node.children.push_back(tokens_in_node);
+    resp.push_back(tokens_in_node);
 
     state::response::SerializedResponseNode tokens_out_node{"TokensOut", tokens_out.load()};
-    root_node.children.push_back(tokens_out_node);
+    resp.push_back(tokens_out_node);
 
     return resp;
   }
 
   std::vector<state::PublishedMetric> calculateMetrics() override {
-    auto metrics = core::ProcessorMetricsImpl::calculateMetrics();
-    metrics.push_back({"tokens_in", static_cast<double>(tokens_in.load()), getCommonLabels()});
-    metrics.push_back({"tokens_out", static_cast<double>(tokens_out.load()), getCommonLabels()});
+    std::vector<state::PublishedMetric> metrics;
+    metrics.push_back({"tokens_in", static_cast<double>(tokens_in.load()), {}});
+    metrics.push_back({"tokens_out", static_cast<double>(tokens_out.load()), {}});
     return metrics;
   }
 
@@ -66,7 +64,7 @@ class RunLlamaCppInference : public core::ProcessorImpl {
   explicit RunLlamaCppInference(core::ProcessorMetadata metadata, LlamaContextProvider llama_context_provider = {})
       : core::ProcessorImpl(metadata),
         llama_context_provider_(std::move(llama_context_provider)) {
-    metrics_ = gsl::make_not_null(std::make_shared<RunLlamaCppInferenceMetrics>(*this));
+    metrics_extension_ = gsl::make_not_null(std::make_shared<RunLlamaCppInferenceMetrics>());
   }
   ~RunLlamaCppInference() override = default;
 

extensions/mqtt/processors/PublishMQTT.cpp

@@ -326,25 +326,22 @@ uint16_t PublishMQTT::InFlightMessageCounter::getCounter() const {
   return counter_;
 }
 
-PublishMQTT::PublishMQTTMetrics::PublishMQTTMetrics(const core::ProcessorImpl& source_processor, const InFlightMessageCounter& in_flight_message_counter)
-  : core::ProcessorMetricsImpl(source_processor),
-    in_flight_message_counter_(&in_flight_message_counter) {
+PublishMQTT::PublishMQTTMetrics::PublishMQTTMetrics(const InFlightMessageCounter& in_flight_message_counter)
+  : in_flight_message_counter_(&in_flight_message_counter) {
 }
 
 std::vector<state::response::SerializedResponseNode> PublishMQTT::PublishMQTTMetrics::serialize() {
-  auto metrics_vector = core::ProcessorMetricsImpl::serialize();
-  gsl_Expects(!metrics_vector.empty());
-  auto& metrics = metrics_vector[0];
+  std::vector<state::response::SerializedResponseNode> metrics;
 
   state::response::SerializedResponseNode in_flight_message_count_node{"InFlightMessageCount", static_cast<uint32_t>(in_flight_message_counter_->getCounter())};
-  metrics.children.push_back(in_flight_message_count_node);
+  metrics.push_back(in_flight_message_count_node);
 
-  return metrics_vector;
+  return metrics;
 }
 
 std::vector<state::PublishedMetric> PublishMQTT::PublishMQTTMetrics::calculateMetrics() {
-  auto metrics = core::ProcessorMetricsImpl::calculateMetrics();
-  metrics.push_back({"in_flight_message_count", static_cast<double>(in_flight_message_counter_->getCounter()), getCommonLabels()});
+  std::vector<state::PublishedMetric> metrics;
+  metrics.push_back({"in_flight_message_count", static_cast<double>(in_flight_message_counter_->getCounter()), {}});
   return metrics;
 }
 

extensions/mqtt/processors/PublishMQTT.h

@@ -41,7 +41,7 @@ class PublishMQTT : public processors::AbstractMQTTProcessor {
  public:
   explicit PublishMQTT(core::ProcessorMetadata metadata)
       : processors::AbstractMQTTProcessor(metadata) {
-    metrics_ = gsl::make_not_null(std::make_shared<PublishMQTTMetrics>(*this, in_flight_message_counter_));
+    metrics_extension_ = gsl::make_not_null(std::make_shared<PublishMQTTMetrics>(in_flight_message_counter_));
   }
 
   EXTENSIONAPI static constexpr const char* Description = "PublishMQTT serializes FlowFile content as an MQTT payload, sending the message to the configured topic and broker.";
@@ -111,9 +111,9 @@ class PublishMQTT : public processors::AbstractMQTTProcessor {
     uint16_t limit_{MQTT_MAX_RECEIVE_MAXIMUM};
   };
 
-  class PublishMQTTMetrics : public core::ProcessorMetricsImpl {
+  class PublishMQTTMetrics : public core::ProcessorMetricsExtension {
    public:
-    PublishMQTTMetrics(const core::ProcessorImpl& source_processor, const InFlightMessageCounter& in_flight_message_counter);
+    explicit PublishMQTTMetrics(const InFlightMessageCounter& in_flight_message_counter);
     std::vector<state::response::SerializedResponseNode> serialize() override;
     std::vector<state::PublishedMetric> calculateMetrics() override;
 

extensions/standard-processors/processors/GetFile.cpp

@@ -178,7 +178,7 @@ bool GetFile::fileMatchesRequestCriteria(const std::filesystem::path& full_name,
     return false;
   }
 
-  auto* const getfile_metrics = dynamic_cast<GetFileMetrics*>(metrics_.get());
+  auto* const getfile_metrics = dynamic_cast<GetFileMetrics*>(metrics_extension_.get());
   gsl_Assert(getfile_metrics);
   getfile_metrics->input_bytes += file_size;
   ++getfile_metrics->accepted_files;

extensions/standard-processors/processors/GetFile.h

@@ -34,7 +34,7 @@
 #include "core/Core.h"
 #include "core/logging/LoggerFactory.h"
 #include "minifi-cpp/utils/Export.h"
-#include "core/ProcessorMetrics.h"
+#include "minifi-cpp/core/ProcessorMetricsExtension.h"
 
 namespace org::apache::nifi::minifi::processors {
 
@@ -52,29 +52,24 @@ struct GetFileRequest {
   std::filesystem::path inputDirectory;
 };
 
-class GetFileMetrics : public core::ProcessorMetricsImpl {
+class GetFileMetrics : public core::ProcessorMetricsExtension {
  public:
-  explicit GetFileMetrics(const core::ProcessorImpl& source_processor)
-    : core::ProcessorMetricsImpl(source_processor) {
-  }
-
   std::vector<state::response::SerializedResponseNode> serialize() override {
-    auto resp = core::ProcessorMetricsImpl::serialize();
-    auto& root_node = resp[0];
+    std::vector<state::response::SerializedResponseNode> resp;
 
     state::response::SerializedResponseNode accepted_files_node{"AcceptedFiles", accepted_files.load()};
-    root_node.children.push_back(accepted_files_node);
+    resp.push_back(accepted_files_node);
 
     state::response::SerializedResponseNode input_bytes_node{"InputBytes", input_bytes.load()};
-    root_node.children.push_back(input_bytes_node);
+    resp.push_back(input_bytes_node);
 
     return resp;
   }
 
   std::vector<state::PublishedMetric> calculateMetrics() override {
-    auto metrics = core::ProcessorMetricsImpl::calculateMetrics();
-    metrics.push_back({"accepted_files", static_cast<double>(accepted_files.load()), getCommonLabels()});
-    metrics.push_back({"input_bytes", static_cast<double>(input_bytes.load()), getCommonLabels()});
+    std::vector<state::PublishedMetric> metrics;
+    metrics.push_back({"accepted_files", static_cast<double>(accepted_files.load()), {}});
+    metrics.push_back({"input_bytes", static_cast<double>(input_bytes.load()), {}});
     return metrics;
   }
 
@@ -86,7 +81,7 @@ class GetFile : public core::ProcessorImpl {
  public:
   explicit GetFile(core::ProcessorMetadata metadata)
       : ProcessorImpl(metadata) {
-    metrics_ = gsl::make_not_null(std::make_shared<GetFileMetrics>(*this));
+    metrics_extension_ = std::make_shared<GetFileMetrics>();
   }
   ~GetFile() override = default;
 

libminifi/include/core/ProcessSession.h

@@ -38,7 +38,7 @@
 #include "provenance/Provenance.h"
 #include "core/Relationship.h"
 #include "minifi-cpp/utils/gsl.h"
-#include "minifi-cpp/core/ProcessorMetrics.h"
+#include "core/ProcessorMetrics.h"
 #include "minifi-cpp/core/ProcessSession.h"
 
 namespace org::apache::nifi::minifi::core::detail {
@@ -123,7 +123,7 @@ class ProcessSessionImpl : public ReferenceContainerImpl, public virtual Process
 
   bool existsFlowFileInRelationship(const Relationship &relationship) override;
 
-  void setMetrics(const std::shared_ptr<ProcessorMetrics>& metrics) override {
+  void setMetrics(const std::shared_ptr<ProcessorMetrics>& metrics) {
     metrics_ = metrics;
   }
 

libminifi/include/core/Processor.h

@@ -38,7 +38,7 @@
 #include "minifi-cpp/core/DynamicProperty.h"
 #include "minifi-cpp/core/Scheduling.h"
 #include "minifi-cpp/core/state/nodes/MetricsBase.h"
-#include "minifi-cpp/core/ProcessorMetrics.h"
+#include "core/ProcessorMetrics.h"
 #include "minifi-cpp/utils/gsl.h"
 #include "utils/Id.h"
 #include "minifi-cpp/core/OutputAttributeDefinition.h"
@@ -113,6 +113,7 @@ class Processor : public ConnectableImpl, public ConfigurableComponentImpl, publ
   [[nodiscard]] bool supportsDynamicRelationships() const override;
   state::response::SharedResponseNode getResponseNode() override;
   gsl::not_null<std::shared_ptr<ProcessorMetrics>> getMetrics() const;
+  std::shared_ptr<ProcessorMetricsExtension> getMetricsExtension() const;
   std::string getProcessGroupName() const;
   void setProcessGroupName(const std::string &name);
   std::string getProcessGroupPath() const;
@@ -170,6 +171,8 @@ class Processor : public ConnectableImpl, public ConfigurableComponentImpl, publ
   std::string process_group_name_;
   std::string process_group_path_;
 
+  gsl::not_null<std::shared_ptr<ProcessorMetrics>> metrics_;
+
  protected:
   std::unique_ptr<ProcessorApi> impl_;
 };