support configServer V2

config_server/protocol/v2/agent.proto

@@ -70,7 +70,7 @@ enum RequestFlags {
 // API: /Agent/Heartbeat
 
 // Agent sends requests to the ConfigServer to get config updates and receive commands.
-message HearbeatRequest {
+message HeartbeatRequest {
     bytes request_id = 1;
     uint64 sequence_num = 2;                    // Increment every request, for server to check sync status
     uint64 capabilities = 3;                    // Bitmask of flags defined by AgentCapabilities enum
@@ -85,7 +85,7 @@ message HearbeatRequest {
     repeated CommandInfo custom_commands = 12;  // Information about command history
     uint64 flags = 13;                          // Predefined command flag
     bytes opaque = 14;                          // Opaque data for extension
-    // before 100 (inclusive) are reserved for future official fields}
+    // before 100 (inclusive) are reserved for future official fields
 }
 
 // Define Config's detail

core/CMakeLists.txt

@@ -99,7 +99,7 @@ include(${CMAKE_CURRENT_SOURCE_DIR}/flusher/links.cmake)
 
 # Subdirectories (modules). except for common, input, processor, flusher, observer, helper and spl.
 set(SUB_DIRECTORIES_LIST
-        batch application app_config checkpoint compression config config/provider config/watcher config_manager config_server_pb 
+        batch application app_config checkpoint compression config config/feedbacker config/provider config/watcher config_manager config_server_pb/v1 config_server_pb/v2 
         container_manager controller event event_handler event_listener file_server go_pipeline log_pb logger 
         models monitor parser pipeline plugin plugin/creator plugin/instance plugin/interface polling
         profile_sender queue reader sdk sender serializer sls_control fuse

core/application/Application.cpp

@@ -42,7 +42,8 @@
 #include "monitor/LogFileProfiler.h"
 #include "monitor/MetricExportor.h"
 #include "monitor/Monitor.h"
-#include "pipeline/PipelineManager.h"
+#include "pipeline/PipelineConfigManager.h"
+#include "pipeline/ProcessConfigManager.h"
 #include "plugin/PluginRegistry.h"
 #include "processor/daemon/LogProcess.h"
 #include "sender/Sender.h"
@@ -55,6 +56,7 @@
 #endif
 #else
 #include "config/provider/CommonConfigProvider.h"
+#include "config/provider/LegacyCommonConfigProvider.h"
 #endif
 #include "queue/ExactlyOnceQueueManager.h"
 
@@ -185,31 +187,47 @@ void Application::Init() {
 }
 
 void Application::Start() {
-    LogtailMonitor::GetInstance()->UpdateConstMetric("start_time", GetTimeStamp(time(NULL), "%Y-%m-%d %H:%M:%S"));
+    LogtailMonitor::GetInstance()->UpdateConstMetric("start_time", GetTimeStamp(time(NULL), "%Y-%m-%d %H:%M:%S")); // LCOV_EXCL_START
 
 #if defined(__ENTERPRISE__) && defined(_MSC_VER)
     InitWindowsSignalObject();
 #endif
     // flusher_sls should always be loaded, since profiling will rely on this.
     Sender::Instance()->Init();
 
-    // add local config dir
-    filesystem::path localConfigPath
-        = filesystem::path(AppConfig::GetInstance()->GetLogtailSysConfDir()) / "config" / "local";
-    error_code ec;
-    filesystem::create_directories(localConfigPath, ec);
-    if (ec) {
-        LOG_WARNING(sLogger,
-                    ("failed to create dir for local config",
-                     "manual creation may be required")("error code", ec.value())("error msg", ec.message()));
+    {
+        // add local config dir
+        filesystem::path localConfigPath
+            = filesystem::path(AppConfig::GetInstance()->GetLogtailSysConfDir()) / "config" / "local";
+        error_code ec;
+        filesystem::create_directories(localConfigPath, ec);
+        if (ec) {
+            LOG_WARNING(sLogger,
+                        ("failed to create dir for local pipelineconfig",
+                         "manual creation may be required")("error code", ec.value())("error msg", ec.message()));
+        }
+        ConfigWatcher::GetInstance()->AddPipelineSource(localConfigPath.string());
+    }
+    {
+        // add local config dir
+        filesystem::path localConfigPath
+            = filesystem::path(AppConfig::GetInstance()->GetLogtailSysConfDir()) / "processconfig" / "local";
+        error_code ec;
+        filesystem::create_directories(localConfigPath, ec);
+        if (ec) {
+            LOG_WARNING(sLogger,
+                        ("failed to create dir for local processconfig",
+                         "manual creation may be required")("error code", ec.value())("error msg", ec.message()));
+        }
+        ConfigWatcher::GetInstance()->AddProcessSource(localConfigPath.string());
     }
-    ConfigWatcher::GetInstance()->AddSource(localConfigPath.string());
 
 #ifdef __ENTERPRISE__
     EnterpriseConfigProvider::GetInstance()->Init("enterprise");
     LegacyConfigProvider::GetInstance()->Init("legacy");
 #else
-    CommonConfigProvider::GetInstance()->Init("common");
+    CommonConfigProvider::GetInstance()->Init("common_v2");
+    LegacyCommonConfigProvider::GetInstance()->Init("common");
 #endif
 
     LogtailAlarm::GetInstance()->Init();
@@ -251,9 +269,13 @@ void Application::Start() {
             lastCheckTagsTime = curTime;
         }
         if (curTime - lastConfigCheckTime >= INT32_FLAG(config_scan_interval)) {
-            ConfigDiff diff = ConfigWatcher::GetInstance()->CheckConfigDiff();
-            if (!diff.IsEmpty()) {
-                PipelineManager::GetInstance()->UpdatePipelines(diff);
+            PipelineConfigDiff pipelineConfigDiff = ConfigWatcher::GetInstance()->CheckPipelineConfigDiff();
+            if (!pipelineConfigDiff.IsEmpty()) {
+                PipelineManager::GetInstance()->UpdatePipelines(pipelineConfigDiff);
+            }
+            ProcessConfigDiff processConfigDiff = ConfigWatcher::GetInstance()->CheckProcessConfigDiff();
+            if (!processConfigDiff.IsEmpty()) {
+                ProcessConfigManager::GetInstance()->UpdateProcessConfigs(processConfigDiff);
             }
             lastConfigCheckTime = curTime;
         }
@@ -295,7 +317,7 @@ void Application::Start() {
         ConfigManager::GetInstance()->DeleteHandlers();
 
         this_thread::sleep_for(chrono::seconds(1));
-    }
+    } // LCOV_EXCL_STOP
 }
 
 void Application::GenerateInstanceId() {
@@ -330,6 +352,7 @@ void Application::Exit() {
     LegacyConfigProvider::GetInstance()->Stop();
 #else
     CommonConfigProvider::GetInstance()->Stop();
+    LegacyCommonConfigProvider::GetInstance()->Stop();
 #endif
 
     LogtailMonitor::GetInstance()->Stop();

core/config/ConfigDiff.h

@@ -19,16 +19,26 @@
 #include <string>
 #include <vector>
 
-#include "config/Config.h"
+#include "config/PipelineConfig.h"
+#include "config/ProcessConfig.h"
 
 namespace logtail {
 
-struct ConfigDiff {
-    std::vector<Config> mAdded;
-    std::vector<Config> mModified;
+class PipelineConfigDiff {
+public:
+    std::vector<PipelineConfig> mAdded;
+    std::vector<PipelineConfig> mModified;
     std::vector<std::string> mRemoved;
     std::vector<std::string> mUnchanged; // 过渡使用，仅供插件系统用
+    bool IsEmpty() { return mRemoved.empty() && mAdded.empty() && mModified.empty(); }
+};
 
+class ProcessConfigDiff {
+public:
+    std::vector<ProcessConfig> mAdded;
+    std::vector<ProcessConfig> mModified;
+    std::vector<std::string> mRemoved;
+    std::vector<std::string> mUnchanged; // 过渡使用，仅供插件系统用
     bool IsEmpty() { return mRemoved.empty() && mAdded.empty() && mModified.empty(); }
 };
 

core/config/Config.cpp → core/config/PipelineConfig.cpp

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#include "config/Config.h"
+#include "config/PipelineConfig.h"
 
 #include <string>
 
@@ -94,7 +94,7 @@ static void ReplaceEnvVarRef(Json::Value& value, bool& res) {
     }
 }
 
-bool Config::Parse() {
+bool PipelineConfig::Parse() {
     if (BOOL_FLAG(enable_env_ref_in_config)) {
         if (ReplaceEnvVar()) {
             LOG_INFO(sLogger, ("env vars in config are replaced, config", mDetail->toStyledString())("config", mName));
@@ -698,7 +698,7 @@ bool Config::Parse() {
     return true;
 }
 
-bool Config::ReplaceEnvVar() {
+bool PipelineConfig::ReplaceEnvVar() {
     bool res = false;
     ReplaceEnvVarRef(*mDetail, res);
     return res;

core/config/Config.h → core/config/PipelineConfig.h

@@ -27,7 +27,7 @@
 
 namespace logtail {
 
-struct Config {
+struct PipelineConfig {
     std::string mName;
     std::unique_ptr<Json::Value> mDetail;
     uint32_t mCreateTime = 0;
@@ -49,7 +49,7 @@ struct Config {
     std::string mLogstore;
     std::string mRegion;
 
-    Config(const std::string& name, std::unique_ptr<Json::Value>&& detail) : mName(name), mDetail(std::move(detail)) {}
+    PipelineConfig(const std::string& name, std::unique_ptr<Json::Value>&& detail) : mName(name), mDetail(std::move(detail)) {}
 
     bool Parse();
 
@@ -75,11 +75,11 @@ struct Config {
     bool ReplaceEnvVar();
 };
 
-inline bool operator==(const Config& lhs, const Config& rhs) {
+inline bool operator==(const PipelineConfig& lhs, const PipelineConfig& rhs) {
     return (lhs.mName == rhs.mName) && (*lhs.mDetail == *rhs.mDetail);
 }
 
-inline bool operator!=(const Config& lhs, const Config& rhs) {
+inline bool operator!=(const PipelineConfig& lhs, const PipelineConfig& rhs) {
     return !(lhs == rhs);
 }
 

core/config/ProcessConfig.cpp

@@ -0,0 +1,30 @@
+// Copyright 2023 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.
+
+#include "config/ProcessConfig.h"
+
+#include <string>
+
+#include "app_config/AppConfig.h"
+#include "common/FileSystemUtil.h"
+#include "common/Flags.h"
+#include "common/JsonUtil.h"
+#include "common/ParamExtractor.h"
+#include "common/YamlUtil.h"
+#include "plugin/PluginRegistry.h"
+
+
+using namespace std;
+
+namespace logtail {} // namespace logtail

core/config/ProcessConfig.h

@@ -0,0 +1,88 @@
+/*
+ * Copyright 2023 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 <json/json.h>
+#include <re2/re2.h>
+
+#include <cstdint>
+#include <filesystem>
+#include <memory>
+#include <string>
+#include <vector>
+
+namespace logtail {
+
+struct ProcessConfig {
+    std::string mName;
+    std::unique_ptr<Json::Value> mDetail;
+
+    // for alarm only
+    std::string mProject;
+    std::string mLogstore;
+    std::string mRegion;
+
+    ProcessConfig(const std::string& name, std::unique_ptr<Json::Value>&& detail)
+        : mName(name), mDetail(std::move(detail)) {
+        mProject = "";
+        mLogstore = "";
+        mRegion = "";
+    }
+    ProcessConfig(const logtail::ProcessConfig& config) {
+        mName = config.mName;
+        mDetail = std::make_unique<Json::Value>(*config.mDetail);
+        mProject = "";
+        mLogstore = "";
+        mRegion = "";
+    }
+
+    ProcessConfig& operator=(ProcessConfig&& other) {
+        if (this != &other) {
+            mName = std::move(other.mName);
+            mDetail = std::move(other.mDetail);
+            mProject = "";
+            mLogstore = "";
+            mRegion = "";
+        }
+        return *this;
+    }
+
+    ProcessConfig& operator=(const ProcessConfig& other) {
+        if (this != &other) {
+            mName = other.mName;
+            mDetail = std::make_unique<Json::Value>(*other.mDetail);
+            mProject = "";
+            mLogstore = "";
+            mRegion = "";
+        }
+        return *this;
+    }
+
+    bool Parse() { return true; }
+
+    const Json::Value& GetConfig() const { return *mDetail; }
+};
+
+inline bool operator==(const ProcessConfig& lhs, const ProcessConfig& rhs) {
+    return (lhs.mName == rhs.mName) && (*lhs.mDetail == *rhs.mDetail);
+}
+
+inline bool operator!=(const ProcessConfig& lhs, const ProcessConfig& rhs) {
+    return !(lhs == rhs);
+}
+
+} // namespace logtail