WIP: refactory cpp metric

core/CMakeLists.txt

@@ -81,7 +81,7 @@ include(${CMAKE_CURRENT_SOURCE_DIR}/dependencies.cmake)
 # Subdirectories (modules).
 set(SUB_DIRECTORIES_LIST
         aggregator app_config checkpoint common config config_manager config_server_pb
-        controller event event_handler event_listener helper log_pb logger monitor
+        controller event event_handler event_listener helper log_pb logger monitor ilogtail_metric
         parser plugin polling processor profiler reader profile_sender sender shennong sdk
         fuse sls_control
         )
@@ -119,6 +119,7 @@ target_link_libraries(${LOGTAIL_TARGET}
         controller
         plugin
         monitor
+        iLogtail_metric
         sender
         profiler
         profile_sender

core/config/Config.cpp

@@ -563,6 +563,9 @@ LogFileReader* Config::CreateLogFileReader(const std::string& dir,
     }
 
     if (reader != NULL) {
+
+        reader->SetFileMetric(ILogtailMetric::GetInstance()->createFileMetric(mConfigName, "FileInput", dir + file));
+
         if ("customized" == mTopicFormat) {
             reader->SetTopicName(STRING_DEEP_COPY(mCustomizedTopic));
         }

core/config/Config.h

@@ -29,6 +29,8 @@
 #include "processor/BaseFilterNode.h"
 #include "LogType.h"
 #include "IntegrityConfig.h"
+#include "ILogtailMetric.h"
+#include "MetricConstants.h"
 
 namespace logtail {
 

core/config_manager/ConfigManagerBase.cpp

@@ -793,6 +793,7 @@ void ConfigManagerBase::LoadSingleUserConfig(const std::string& logName, const J
 
                 UserLogConfigParser::ParseAdvancedConfig(value, *config);
             }
+
             if (logType == DELIMITER_LOG) {
                 config->mTimeFormat = GetStringValue(value, "timeformat", "");
                 string separatorStr = GetStringValue(value, "delimiter_separator");

core/config_manager/ConfigManagerBase.h

@@ -32,6 +32,8 @@
 #include "common/Lock.h"
 #include "common/Thread.h"
 #include "event/Event.h"
+#include "ILogtailMetric.h"
+#include "MetricConstants.h"
 
 DECLARE_FLAG_BOOL(https_verify_peer);
 DECLARE_FLAG_STRING(https_ca_cert);

core/controller/EventDispatcherBase.cpp

@@ -62,6 +62,7 @@
 #include "shennong/MetricSender.h"
 #include "polling/PollingDirFile.h"
 #include "polling/PollingModify.h"
+#include "ilogtail_metric/MetricExportor.h"
 #ifdef APSARA_UNIT_TEST_MAIN
 #include "polling/PollingEventQueue.h"
 #endif
@@ -868,6 +869,7 @@ bool EventDispatcherBase::Dispatch() {
         DumpCheckPointPeriod(curTime);
         if (curTime - lastCheckDir >= INT32_FLAG(main_loop_check_interval)) {
             LogFileProfiler::GetInstance()->SendProfileData();
+            MetricExportor::GetInstance()->pushMetrics();
 #if defined(__linux__)
             CheckShennong();
 #endif

core/ilogtail_metric/CMakeLists.txt

@@ -0,0 +1,30 @@
+# Copyright 2022 iLogtail Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+cmake_minimum_required(VERSION 2.9)
+project(iLogtail_metric)
+
+if (MSVC)
+    add_definitions(-DTARGETLIBS=Psapi.lib)
+    add_definitions(-DPSAPI_VERSION=1)
+endif ()
+
+file(GLOB LIB_SOURCE_FILES *.cpp *.h)
+append_source_files(LIB_SOURCE_FILES)
+add_library(${PROJECT_NAME} STATIC ${LIB_SOURCE_FILES})
+target_link_libraries(${PROJECT_NAME} log_pb)
+
+if (MSVC)
+    target_link_libraries(${PROJECT_NAME} "Psapi.lib")
+endif ()

core/ilogtail_metric/ILogtaiMetric.cpp

@@ -0,0 +1,119 @@
+#include "ILogtailMetric.h"
+
+namespace logtail {
+
+
+ILogtailMetric::ILogtailMetric() {
+}
+
+ILogtailMetric::~ILogtailMetric() {
+}
+
+BaseMetric::BaseMetric() {
+    mMetricObj = new MetricObj;
+    mMetricObj->val = {0};
+    mMetricObj->timestamp = time(NULL);
+}
+
+BaseMetric::BaseMetric(MetricObj*  metricObj) {
+    mMetricObj = metricObj;
+}
+
+BaseMetric::~BaseMetric() {
+    if (mMetricObj) {
+        delete mMetricObj;
+    }
+}
+
+
+void BaseMetric::baseMetricAdd(uint64_t val) {
+    mMetricObj->val += val;
+    mMetricObj->timestamp = {time(NULL)};
+}
+
+BaseMetric::MetricObj*  BaseMetric::snapShotMetricObj() {
+    MetricObj*  newMetricObj = new MetricObj;
+
+    LOG_INFO(sLogger, ("mMetricObj->val exchange before", mMetricObj->val));
+    long value = mMetricObj->val.exchange(0);
+    newMetricObj->val = {value};
+    LOG_INFO(sLogger, ("mMetricObj->val exchange after", mMetricObj->val));
+
+    LOG_INFO(sLogger, ("mMetricObj->timestamp exchange after", mMetricObj->timestamp));
+    newMetricObj->timestamp = {time(NULL)};
+    LOG_INFO(sLogger, ("mMetricObj->timestamp exchange after", mMetricObj->timestamp));
+
+    return newMetricObj;
+}
+
+BaseMetric::MetricObj*  BaseMetric::getMetricObj() {
+    return mMetricObj;
+}
+
+
+BaseMetricPtr PipelineMetric::getBaseMetric(std::string metricField) {
+    LOG_INFO(sLogger, ("getBaseMetric", metricField));
+    std::unordered_map<std::string, BaseMetricPtr>::iterator iter = mBaseMetrics.find(metricField);
+    if (iter != mBaseMetrics.end()) {
+        return iter->second;
+    } else {
+        BaseMetricPtr base(new BaseMetric());
+        mBaseMetrics.insert(std::pair<std::string, BaseMetricPtr>(metricField, base));
+        return base;
+    }
+}
+
+
+PipelineMetricPtr ILogtailMetric::createPipelineMetric(std::list<std::string> fields , std::unordered_map<std::string, std::string> labels) {
+    PipelineMetricPtr fileMetric(new PipelineMetric());
+
+    LOG_INFO(sLogger, ("label size before", fileMetric->mLabels.size()));
+
+    fileMetric->mLabels = labels;
+
+    LOG_INFO(sLogger, ("label size before", fileMetric->mLabels.size()));
+
+    for (std::list<std::string>::iterator it = fields.begin(); it != fields.end(); ++it) {
+        BaseMetricPtr base(new BaseMetric());
+        fileMetric->mBaseMetrics.insert(std::pair<std::string, BaseMetricPtr>(*it, base));
+    }
+    std::lock_guard<std::mutex> lock(mMetricsLock);
+    mPipelineMetrics.push_back(fileMetric);
+    return fileMetric;
+}
+
+void ILogtailMetric::deletePipelineMetric(PipelineMetricPtr pipelineMetric) {
+    std::lock_guard<std::mutex> lock(mMetricsLock);
+    mPipelineMetrics.remove(pipelineMetric);
+}
+
+
+PipelineMetricPtr ILogtailMetric::createFileMetric(std::string configUid, std::string pluginUid, std::string filePath) {
+    std::unordered_map<std::string, std::string> labels;
+    LOG_INFO(sLogger, ("label size before", labels.size()));
+
+    labels.insert(std::pair<std::string, std::string>("configUid", configUid));
+    labels.insert(std::pair<std::string, std::string>("pluginUid", pluginUid));
+    labels.insert(std::pair<std::string, std::string>("filePath", filePath));
+
+    LOG_INFO(sLogger, ("label size before", labels.size()));
+    std::list<std::string> fields;
+    fields.push_back(METRIC_FILE_READ_COUNT);
+    fields.push_back(METRIC_FILE_READ_BYTES);
+    return createPipelineMetric(fields, labels);
+}
+
+
+BaseMetricPtr ILogtailMetric::getBaseMetric(PipelineMetricPtr pipelineMetric, std::string metricField) {
+    LOG_INFO(sLogger, ("getBaseMetric", metricField));
+    std::unordered_map<std::string, BaseMetricPtr>::iterator iter = pipelineMetric->mBaseMetrics.find(metricField);
+    if (iter != pipelineMetric->mBaseMetrics.end()) {
+        return iter->second;
+    } else {
+        BaseMetricPtr base(new BaseMetric());
+        pipelineMetric->mBaseMetrics.insert(std::pair<std::string, BaseMetricPtr>(metricField, base));
+        return base;
+    }
+}
+
+}

core/ilogtail_metric/ILogtailMetric.h

@@ -0,0 +1,71 @@
+#pragma once
+#include <string>
+#include <unordered_map>
+#include <list>
+#include <atomic>
+#include "logger/Logger.h"
+#include <MetricConstants.h>
+
+
+namespace logtail {
+
+
+class BaseMetric{
+    struct MetricObj {
+        /* counters and gauges */
+        std::atomic_long val;
+
+        std::atomic_long timestamp;
+    };
+    public:
+        BaseMetric();
+        BaseMetric(MetricObj*  metricObj);
+        ~BaseMetric();
+        MetricObj* mMetricObj;
+        void baseMetricAdd(uint64_t val);
+        MetricObj*  getMetricObj();
+        MetricObj*  snapShotMetricObj();
+};
+
+typedef std::shared_ptr<BaseMetric> BaseMetricPtr;
+
+
+class PipelineMetric {
+    public:
+        std::unordered_map<std::string, BaseMetricPtr> mBaseMetrics;
+        std::unordered_map<std::string, std::string> mLabels;
+
+        BaseMetricPtr getBaseMetric(std::string metricField);
+
+};
+
+typedef std::shared_ptr<PipelineMetric> PipelineMetricPtr;
+
+class ILogtailMetric {
+
+    private:
+        ILogtailMetric();
+        ~ILogtailMetric();
+
+
+    public:
+        static ILogtailMetric* GetInstance() {
+            static ILogtailMetric* ptr = new ILogtailMetric();
+            return ptr;
+        }
+
+        std::mutex mMetricsLock;
+
+        std::list<PipelineMetricPtr> mPipelineMetrics;
+        PipelineMetricPtr mInstanceMetric;
+
+        PipelineMetricPtr createPipelineMetric(std::list<std::string> fields , std::unordered_map<std::string, std::string> labels);
+
+        PipelineMetricPtr createFileMetric(std::string configUid, std::string pluginUid, std::string filePath);
+
+        BaseMetricPtr getBaseMetric(PipelineMetricPtr pipelineMetric, std::string metricField);
+
+        void deletePipelineMetric(PipelineMetricPtr pipelineMetric);
+
+    };
+}

core/ilogtail_metric/MetricConstants.h

@@ -0,0 +1,25 @@
+/*
+ * Copyright 2022 iLogtail Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#pragma once
+#include <string>
+
+namespace logtail {
+
+const std::string METRIC_FILE_READ_COUNT = "file_read_count";
+const std::string METRIC_FILE_READ_BYTES = "file_read_bytes";
+
+} // namespace logtail