feat: forward service self metric (#2489)

* feat: forward service self metric

Change-Id: I994caa69c976c9d112b78d8ec53eb1ccf7b667e5
Co-developed-by: Cursor <noreply@cursor.com>

* fix

Change-Id: If80c18eff78e36812f6a0e37c5fdb8666c219b55
Co-developed-by: Cursor <noreply@cursor.com>

Author: Abingcbc

core/forward/BaseService.h

@@ -22,11 +22,15 @@ namespace logtail {
 
 class BaseService {
 public:
+    BaseService(const std::string& address) : mAddress(address) {}
     virtual ~BaseService() = default;
 
     virtual bool Update(std::string configName, const Json::Value& config) = 0;
     virtual bool Remove(std::string configName, const Json::Value& config) = 0;
     [[nodiscard]] virtual const std::string& Name() const = 0;
+
+protected:
+    std::string mAddress;
 };
 
 }; // namespace logtail

core/forward/GrpcInputManager.cpp

@@ -76,7 +76,7 @@ bool GrpcInputManager::AddListenInput(const std::string& configName,
     std::lock_guard<std::mutex> lock(mListenAddressToInputMapMutex);
     auto it = mListenAddressToInputMap.find(address);
     // generate a new service instance to get name
-    std::unique_ptr<T> service = std::make_unique<T>();
+    std::unique_ptr<T> service = std::make_unique<T>(address);
     if (it != mListenAddressToInputMap.end()) {
         if (it->second.mServer == nullptr || it->second.mService == nullptr) {
             // should never happen

core/forward/loongsuite/LoongSuiteForwardService.cpp

@@ -22,6 +22,7 @@
 
 #include "common/Flags.h"
 #include "common/ParamExtractor.h"
+#include "common/TimeKeeper.h"
 #include "grpcpp/support/status.h"
 #include "logger/Logger.h"
 #include "models/PipelineEventGroup.h"
@@ -69,6 +70,20 @@ bool LoongSuiteForwardServiceImpl::Update(std::string configName, const Json::Va
     return true;
 }
 
+LoongSuiteForwardServiceImpl::LoongSuiteForwardServiceImpl(const std::string& address) : BaseService(address) {
+    WriteMetrics::GetInstance()->CreateMetricsRecordRef(mMetricsRecordRef,
+                                                        MetricCategory::METRIC_CATEGORY_COMPONENT,
+                                                        {
+                                                            {METRIC_LABEL_KEY_COMPONENT_NAME, "loongsuite_forward"},
+                                                            {METRIC_LABEL_KEY_SERVICE_ADDRESS, address},
+                                                        });
+    mInEventsTotal = mMetricsRecordRef.CreateCounter(METRIC_COMPONENT_IN_EVENTS_TOTAL);
+    mInSizeBytes = mMetricsRecordRef.CreateCounter(METRIC_COMPONENT_IN_SIZE_BYTES);
+    mTotalDelayMs = mMetricsRecordRef.CreateTimeCounter(METRIC_COMPONENT_TOTAL_DELAY_MS);
+    mDiscardedEventsTotal = mMetricsRecordRef.CreateCounter(METRIC_COMPONENT_DISCARDED_ITEMS_TOTAL);
+    WriteMetrics::GetInstance()->CommitMetricsRecordRef(mMetricsRecordRef);
+}
+
 bool LoongSuiteForwardServiceImpl::Remove(std::string configName, const Json::Value& config) {
     std::string errorMsg;
 
@@ -89,6 +104,9 @@ grpc::ServerUnaryReactor* LoongSuiteForwardServiceImpl::Forward(grpc::CallbackSe
                                                                 LoongSuiteForwardResponse* response) {
     auto* reactor = context->DefaultReactor();
     grpc::Status status(grpc::StatusCode::NOT_FOUND, "No matching config found for forward request");
+    ADD_COUNTER(mInEventsTotal, 1);
+    ADD_COUNTER(mInSizeBytes, request->data_size());
+    auto before = TimeKeeper::GetInstance()->NowMs();
 
     std::shared_ptr<ForwardConfig> config;
     if (FindMatchingConfig(context, config)) {
@@ -99,10 +117,13 @@ grpc::ServerUnaryReactor* LoongSuiteForwardServiceImpl::Forward(grpc::CallbackSe
         if (status.ok()) {
             mRetryTimeController.UpRetryTimes(config->configName);
         } else {
+            ADD_COUNTER(mDiscardedEventsTotal, 1);
             mRetryTimeController.DownRetryTimes(config->configName);
         }
+    } else {
+        ADD_COUNTER(mDiscardedEventsTotal, 1);
     }
-
+    ADD_COUNTER(mTotalDelayMs, std::chrono::milliseconds(TimeKeeper::GetInstance()->NowMs() - before));
     reactor->Finish(status);
     return reactor;
 }

core/forward/loongsuite/LoongSuiteForwardService.h

@@ -24,6 +24,7 @@
 
 #include "collection_pipeline/queue/QueueKey.h"
 #include "forward/BaseService.h"
+#include "monitor/MetricManager.h"
 #include "protobuf/forward/loongsuite.grpc.pb.h"
 
 namespace logtail {
@@ -78,7 +79,7 @@ class RetryTimeController {
 
 class LoongSuiteForwardServiceImpl : public BaseService, public LoongSuiteForwardService::CallbackService {
 public:
-    LoongSuiteForwardServiceImpl() = default;
+    LoongSuiteForwardServiceImpl(const std::string& address);
     ~LoongSuiteForwardServiceImpl() override = default;
 
     bool Update(std::string configName, const Json::Value& config) override;
@@ -99,6 +100,12 @@ class LoongSuiteForwardServiceImpl : public BaseService, public LoongSuiteForwar
 
     RetryTimeController mRetryTimeController;
 
+    MetricsRecordRef mMetricsRecordRef;
+    CounterPtr mInEventsTotal;
+    CounterPtr mInSizeBytes;
+    TimeCounterPtr mTotalDelayMs;
+    CounterPtr mDiscardedEventsTotal;
+
     bool AddToIndex(std::string& configName, ForwardConfig&& config, std::string& errorMsg);
     bool FindMatchingConfig(grpc::CallbackServerContext* context, std::shared_ptr<ForwardConfig>& config) const;
     void ProcessForwardRequest(const LoongSuiteForwardRequest* request,

core/monitor/metric_constants/ComponentMetrics.cpp

@@ -22,6 +22,7 @@ namespace logtail {
 // label keys
 const string METRIC_LABEL_KEY_COMPONENT_NAME = "component_name";
 const string METRIC_LABEL_KEY_FLUSHER_PLUGIN_ID = "flusher_plugin_id";
+const string METRIC_LABEL_KEY_SERVICE_ADDRESS = "service_address";
 
 /**********************************************************
  *   queue

core/monitor/metric_constants/MetricConstants.h

@@ -234,6 +234,7 @@ extern const std::string METRIC_LABEL_KEY_EXACTLY_ONCE_ENABLED;
 extern const std::string METRIC_LABEL_KEY_QUEUE_TYPE;
 extern const std::string METRIC_LABEL_KEY_TARGET;
 extern const std::string METRIC_LABEL_KEY_GROUP_BATCH_ENABLED;
+extern const std::string METRIC_LABEL_KEY_SERVICE_ADDRESS;
 
 // label values
 extern const std::string METRIC_LABEL_VALUE_COMPONENT_NAME_BATCHER;

core/unittest/forward/GrpcInputManagerUnittest.cpp

@@ -90,7 +90,8 @@ void GrpcInputManagerUnittest::TestUpdateListenInputExistingAddressSameTypeUpdat
 }
 
 void GrpcInputManagerUnittest::TestUpdateListenInputExistingAddressDifferentTypeError() const {
-    MockServiceImpl mockService;
+    const std::string address = "0.0.0.0:50053";
+    MockServiceImpl mockService(address);
     auto* runner = GrpcInputManager::GetInstance();
     runner->Init();
     Json::Value config;
@@ -99,7 +100,6 @@ void GrpcInputManagerUnittest::TestUpdateListenInputExistingAddressDifferentType
     config["QueueKey"] = 1;
     config["InputIndex"] = 1;
     config["Protocol"] = "LoongSuite";
-    const std::string address = "0.0.0.0:50053";
     bool ret = runner->AddListenInput<LoongSuiteForwardServiceImpl>("configA", address, config);
     APSARA_TEST_TRUE_FATAL(ret);
     ret = runner->AddListenInput<MockServiceImpl>("configB", address, config);

core/unittest/forward/LoongSuiteForwardServiceUnittest.cpp

@@ -57,7 +57,7 @@ class LoongSuiteForwardServiceUnittest : public testing::Test {
 
 protected:
     void SetUp() override {
-        service = std::unique_ptr<LoongSuiteForwardServiceImpl>(new LoongSuiteForwardServiceImpl());
+        service = std::unique_ptr<LoongSuiteForwardServiceImpl>(new LoongSuiteForwardServiceImpl("0.0.0.0:50054"));
         // Initialize ProcessorRunner for testing
         ProcessorRunner::GetInstance()->Init();
     }

core/unittest/forward/MockServiceImpl.h

@@ -21,7 +21,7 @@ namespace logtail {
 
 class MockServiceImpl : public BaseService, public LoongSuiteForwardService::CallbackService {
 public:
-    MockServiceImpl() = default;
+    explicit MockServiceImpl(const std::string& address) : BaseService(address) {}
     ~MockServiceImpl() override = default;
 
     bool Update(std::string configName, const Json::Value& config) override { return true; }