Merge branch 'main' into feature/metadata

core/CMakeLists.txt

@@ -33,6 +33,8 @@ cmake_dependent_option(ENABLE_STATIC_LINK_CRT "Build Logtail by linking CRT stat
 option(WITHOUTGDB "Build Logtail without gdb")
 option(WITHSPL "Build Logtail and UT with SPL" ON)
 option(BUILD_LOGTAIL_UT "Build unit test for Logtail")
+set(PROVIDER_PATH "provider" CACHE PATH "Path to the provider module") # external provider path can be set with -DPROVIDER_PATH
+set(UNITTEST_PATH "unittest" CACHE PATH "Path to the unittest module") # external unittest path can be set with -DUNITTEST_PATH
 
 if (BUILD_LOGTAIL_SHARED_LIBRARY AND WITHSPL)
     message(FATEL_ERROR, "Generating logtail shared library is not supported to be linked with SPL. WITHSPL should be set OFF.")
@@ -109,13 +111,13 @@ include(${CMAKE_CURRENT_SOURCE_DIR}/plugin/processor/links.cmake)
 include(${CMAKE_CURRENT_SOURCE_DIR}/plugin/flusher/flusher.cmake)
 include(${CMAKE_CURRENT_SOURCE_DIR}/plugin/flusher/links.cmake)
 
-# Subdirectories (modules). except for common, input, processor, flusher, observer, helper and spl.
+# Subdirectories (modules). except for common, input, processor, flusher, observer, helper, spl, and provider.
 set(SUB_DIRECTORIES_LIST
         application app_config checkpoint container_manager logger go_pipeline monitor profile_sender models
-        metadata config config/feedbacker config/provider config/watcher
+        metadata config config/watcher
         pipeline pipeline/batch pipeline/limiter pipeline/plugin pipeline/plugin/creator pipeline/plugin/instance pipeline/plugin/interface pipeline/queue pipeline/route pipeline/serializer
         runner runner/sink/http
-        protobuf/config_server/v1 protobuf/config_server/v2 protobuf/sls
+        protobuf/sls
         file_server file_server/event file_server/event_handler file_server/event_listener file_server/reader file_server/polling
         prometheus prometheus/labels prometheus/schedulers prometheus/async
         ebpf ebpf/observer ebpf/security ebpf/handler
@@ -161,6 +163,10 @@ endif()
 # remove several files in go_pipeline
 list(REMOVE_ITEM FRAMEWORK_SOURCE_FILES ${CMAKE_CURRENT_SOURCE_DIR}/go_pipeline/LogtailPluginAdapter.cpp ${CMAKE_CURRENT_SOURCE_DIR}/go_pipeline/LogtailPluginAdapter.h)
 
+
+# add provider
+add_subdirectory("${PROVIDER_PATH}" "${CMAKE_BINARY_DIR}/provider")
+
 if(MSVC)
     # remove linux event listener
     file(GLOB REMOVE_EVENT_LISTENER_SOURCES file_server/event_listener/*_Linux.cpp file_server/event_listener/*_Linux.h)
@@ -231,12 +237,12 @@ if(BUILD_LOGTAIL OR BUILD_LOGTAIL_SHARED_LIBRARY)
     flusher_link(${LOGTAIL_TARGET})
     all_link(${LOGTAIL_TARGET})
     common_link(${LOGTAIL_TARGET})
+    target_link_libraries(${LOGTAIL_TARGET} provider)
 endif()
 
 # Logtail UT.
 if (BUILD_LOGTAIL_UT)
-    message(STATUS "Build unittest.")
-    add_definitions(-DVIRTUAL=virtual)
+    message(STATUS "Build unittest.")    
     function(delete_gcda_files target_directory)
         if(EXISTS "${target_directory}")
             message(STATUS "Deleting .gcda files in ${target_directory}")
@@ -249,7 +255,5 @@ if (BUILD_LOGTAIL_UT)
     delete_gcda_files(".")
     include(CTest)
     enable_testing()
-    add_subdirectory(unittest)
-else()
-    add_definitions(-DVIRTUAL= )
+    add_subdirectory("${UNITTEST_PATH}" "${CMAKE_BINARY_DIR}/unittest")
 endif ()

core/application/Application.cpp

@@ -60,8 +60,7 @@
 #include "streamlog/StreamLogManager.h"
 #endif
 #else
-#include "config/provider/CommonConfigProvider.h"
-#include "config/provider/LegacyCommonConfigProvider.h"
+#include "provider/Provider.h"
 #endif
 
 DEFINE_FLAG_BOOL(ilogtail_disable_core, "disable core in worker process", true);
@@ -242,8 +241,7 @@ void Application::Start() { // GCOVR_EXCL_START
     EnterpriseConfigProvider::GetInstance()->Init("enterprise");
     LegacyConfigProvider::GetInstance()->Init("legacy");
 #else
-    CommonConfigProvider::GetInstance()->Init("common_v2");
-    LegacyCommonConfigProvider::GetInstance()->Init("common");
+    InitRemoteConfigProviders();
 #endif
 
     LogtailAlarm::GetInstance()->Init();
@@ -370,8 +368,10 @@ void Application::Exit() {
     EnterpriseConfigProvider::GetInstance()->Stop();
     LegacyConfigProvider::GetInstance()->Stop();
 #else
-    CommonConfigProvider::GetInstance()->Stop();
-    LegacyCommonConfigProvider::GetInstance()->Stop();
+    auto remoteConfigProviders = GetRemoteConfigProviders();
+    for (auto& provider : remoteConfigProviders) {
+        provider->Stop();
+    }
 #endif
 
     LogtailMonitor::GetInstance()->Stop();

core/config/provider/CommonConfigProvider.cpp → core/config/common_provider/CommonConfigProvider.cpp

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#include "config/provider/CommonConfigProvider.h"
+#include "CommonConfigProvider.h"
 
 #include <json/json.h>
 

core/config/provider/CommonConfigProvider.h → core/config/common_provider/CommonConfigProvider.h

@@ -125,7 +125,7 @@ class CommonConfigProvider : public ConfigProvider, ConfigFeedbackable {
 
     ConfigServerAddress GetOneConfigServerAddress(bool changeConfigServer);
 
-    VIRTUAL bool SendHttpRequest(const std::string& operation,
+    virtual bool SendHttpRequest(const std::string& operation,
                                  const std::string& reqBody,
                                  const std::string& configType,
                                  const std::string& requestId,

core/config/provider/LegacyCommonConfigProvider.cpp → core/config/common_provider/LegacyCommonConfigProvider.cpp

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#include "config/provider/LegacyCommonConfigProvider.h"
+#include "LegacyCommonConfigProvider.h"
 
 #include <json/json.h>
 

core/go_pipeline/LogtailPlugin.cpp

@@ -30,7 +30,7 @@
 #include "monitor/LogtailAlarm.h"
 #include "pipeline/PipelineManager.h"
 #include "pipeline/queue/SenderQueueManager.h"
-#include "profile_sender/ProfileSender.h"
+#include "provider/Provider.h"
 
 DEFINE_FLAG_BOOL(enable_sls_metrics_format, "if enable format metrics in SLS metricstore log pattern", false);
 DEFINE_FLAG_BOOL(enable_containerd_upper_dir_detect,
@@ -171,22 +171,22 @@ int LogtailPlugin::SendPbV2(const char* configName,
     FlusherSLS* pConfig = NULL;
     if (configNameStr == alarmConfig->mLogstore) {
         pConfig = alarmConfig;
-        pConfig->mProject = ProfileSender::GetInstance()->GetDefaultProfileProjectName();
-        pConfig->mRegion = ProfileSender::GetInstance()->GetDefaultProfileRegion();
+        pConfig->mProject = GetProfileSender()->GetDefaultProfileProjectName();
+        pConfig->mRegion = GetProfileSender()->GetDefaultProfileRegion();
         if (pConfig->mProject.empty()) {
             return 0;
         }
     } else if (configNameStr == profileConfig->mLogstore) {
         pConfig = profileConfig;
-        pConfig->mProject = ProfileSender::GetInstance()->GetDefaultProfileProjectName();
-        pConfig->mRegion = ProfileSender::GetInstance()->GetDefaultProfileRegion();
+        pConfig->mProject = GetProfileSender()->GetDefaultProfileProjectName();
+        pConfig->mRegion = GetProfileSender()->GetDefaultProfileRegion();
         if (pConfig->mProject.empty()) {
             return 0;
         }
     } else if (configNameStr == containerConfig->mLogstore) {
         pConfig = containerConfig;
-        pConfig->mProject = ProfileSender::GetInstance()->GetDefaultProfileProjectName();
-        pConfig->mRegion = ProfileSender::GetInstance()->GetDefaultProfileRegion();
+        pConfig->mProject = GetProfileSender()->GetDefaultProfileProjectName();
+        pConfig->mRegion = GetProfileSender()->GetDefaultProfileRegion();
         if (pConfig->mProject.empty()) {
             return 0;
         }

core/monitor/LogFileProfiler.cpp

@@ -27,7 +27,7 @@
 #include "common/version.h"
 #include "file_server/ConfigManager.h"
 #include "logger/Logger.h"
-#include "profile_sender/ProfileSender.h"
+#include "provider/Provider.h"
 #include "pipeline/queue/QueueKeyManager.h"
 
 DEFINE_FLAG_INT32(profile_data_send_interval, "interval of send LogFile/DomainSocket profile data, seconds", 600);
@@ -261,7 +261,7 @@ void LogFileProfiler::SendProfileData(bool forceSend) {
             }
         }
         UpdateDumpData(logGroup, detail, logstore);
-        ProfileSender::GetInstance()->SendToProfileProject(region, logGroup);
+        GetProfileSender()->SendToProfileProject(region, logGroup);
     } while (true);
     DumpToLocal(curTime, forceSend, detail, logstore);
     mLastSendTime = curTime;

core/monitor/LogtailAlarm.cpp

@@ -23,7 +23,7 @@
 #include "common/TimeUtil.h"
 #include "common/version.h"
 #include "protobuf/sls/sls_logs.pb.h"
-#include "profile_sender/ProfileSender.h"
+#include "provider/Provider.h"
 #include "pipeline/queue/QueueKeyManager.h"
 #include "pipeline/queue/SenderQueueManager.h"
 
@@ -195,7 +195,7 @@ void LogtailAlarm::SendAllRegionAlarm() {
             // check sender queue status, if invalid jump this region
 
             QueueKey alarmPrjLogstoreKey = QueueKeyManager::GetInstance()->GetKey(
-                "-flusher_sls-" + ProfileSender::GetInstance()->GetProfileProjectName(region) + "#logtail_alarm");
+                "-flusher_sls-" + GetProfileSender()->GetProfileProjectName(region) + "#logtail_alarm");
             if (!SenderQueueManager::GetInstance()->IsValidToPush(alarmPrjLogstoreKey)) {
                 // jump this region
                 ++sendRegionIndex;
@@ -264,7 +264,7 @@ void LogtailAlarm::SendAllRegionAlarm() {
             continue;
         }
         // this is an anonymous send and non lock send
-        ProfileSender::GetInstance()->SendToProfileProject(region, logGroup);
+        GetProfileSender()->SendToProfileProject(region, logGroup);
     } while (true);
 }
 
@@ -298,7 +298,7 @@ void LogtailAlarm::SendAlarm(const LogtailAlarmType alarmType,
     }
 
     // ignore alarm for profile data
-    if (ProfileSender::GetInstance()->IsProfileData(region, projectName, category)) {
+    if (GetProfileSender()->IsProfileData(region, projectName, category)) {
         return;
     }
     // LOG_DEBUG(sLogger, ("Add Alarm", region)("projectName", projectName)("alarm index",

core/monitor/MetricExportor.cpp

@@ -90,10 +90,10 @@ void MetricExportor::SendToSLS(std::map<std::string, sls_logs::LogGroup*>& logGr
         logGroup->set_source(LogFileProfiler::mIpAddr);
         logGroup->set_topic(METRIC_TOPIC_TYPE);
         if (METRIC_REGION_DEFAULT == iter->first) {
-            ProfileSender::GetInstance()->SendToProfileProject(ProfileSender::GetInstance()->GetDefaultProfileRegion(),
+            GetProfileSender()->SendToProfileProject(GetProfileSender()->GetDefaultProfileRegion(),
                                                                *logGroup);
         } else {
-            ProfileSender::GetInstance()->SendToProfileProject(iter->first, *logGroup);
+            GetProfileSender()->SendToProfileProject(iter->first, *logGroup);
         }
         delete logGroup;
     }

core/monitor/MetricExportor.h

@@ -17,7 +17,7 @@
 #pragma once
 
 #include "Monitor.h"
-#include "profile_sender/ProfileSender.h"
+#include "provider/Provider.h"
 
 
 namespace logtail {

core/monitor/Monitor.cpp

@@ -50,7 +50,7 @@
 #include "config/provider/EnterpriseConfigProvider.h"
 #endif
 #include "pipeline/PipelineManager.h"
-#include "profile_sender/ProfileSender.h"
+#include "provider/Provider.h"
 
 using namespace std;
 using namespace sls_logs;
@@ -299,7 +299,7 @@ bool LogtailMonitor::SendStatusProfile(bool suicide) {
     AddLogContent(logPtr, "plugin_stats", PipelineManager::GetInstance()->GetPluginStatistics());
     // Metrics.
     vector<string> allProfileRegion;
-    ProfileSender::GetInstance()->GetAllProfileRegion(allProfileRegion);
+    GetProfileSender()->GetAllProfileRegion(allProfileRegion);
     UpdateMetric("region", allProfileRegion);
 #ifdef __ENTERPRISE__
     UpdateMetric("config_update_count", EnterpriseConfigProvider::GetInstance()->GetConfigUpdateTotalCount());
@@ -347,10 +347,10 @@ bool LogtailMonitor::SendStatusProfile(bool suicide) {
         }
 
         if (i == allProfileRegion.size() - 1) {
-            ProfileSender::GetInstance()->SendToProfileProject(allProfileRegion[i], logGroup);
+            GetProfileSender()->SendToProfileProject(allProfileRegion[i], logGroup);
         } else {
             LogGroup copyLogGroup = logGroup;
-            ProfileSender::GetInstance()->SendToProfileProject(allProfileRegion[i], copyLogGroup);
+            GetProfileSender()->SendToProfileProject(allProfileRegion[i], copyLogGroup);
         }
     }
     return true;

core/monitor/Monitor.h

@@ -24,6 +24,7 @@
 #include "LogtailMetric.h"
 #include "MetricConstants.h"
 #include "MetricStore.h"
+
 #if defined(_MSC_VER)
 #include <Windows.h>
 #endif

core/plugin/flusher/sls/DiskBufferWriter.cpp

@@ -27,7 +27,7 @@
 #include "protobuf/sls/sls_logs.pb.h"
 #include "logger/Logger.h"
 #include "monitor/LogtailAlarm.h"
-#include "profile_sender/ProfileSender.h"
+#include "provider/Provider.h"
 #include "pipeline/queue/QueueKeyManager.h"
 #include "pipeline/queue/SLSSenderQueueItem.h"
 #include "sdk/Exception.h"
@@ -765,7 +765,7 @@ SendResult DiskBufferWriter::SendToNetSync(sdk::Client* sendClient,
                                                            bufferMeta.logstore(),
                                                            "");
                 // no region
-                if (!ProfileSender::GetInstance()->IsProfileData("", bufferMeta.project(), bufferMeta.logstore()))
+                if (!GetProfileSender()->IsProfileData("", bufferMeta.project(), bufferMeta.logstore()))
                     LOG_ERROR(sLogger,
                               ("send data to SLS fail, error_code", errorCode)("error_message", ex.GetMessage())(
                                   "endpoint", sendClient->GetRawSlsHost())("projectName", bufferMeta.project())(

core/plugin/flusher/sls/FlusherSLS.cpp

@@ -33,7 +33,7 @@
 #include "plugin/flusher/sls/SLSClientManager.h"
 #include "plugin/flusher/sls/SLSResponse.h"
 #include "plugin/flusher/sls/SendResult.h"
-#include "profile_sender/ProfileSender.h"
+#include "provider/Provider.h"
 #include "runner/FlusherRunner.h"
 #include "sdk/Common.h"
 #include "sls_control/SLSControl.h"
@@ -610,7 +610,7 @@ void FlusherSLS::OnSendDone(const HttpResponse& response, SenderQueueItem* item)
 
     auto data = static_cast<SLSSenderQueueItem*>(item);
     string configName = HasContext() ? GetContext().GetConfigName() : "";
-    bool isProfileData = ProfileSender::GetInstance()->IsProfileData(mRegion, mProject, data->mLogstore);
+    bool isProfileData = GetProfileSender()->IsProfileData(mRegion, mProject, data->mLogstore);
     int32_t curTime = time(NULL);
     auto curSystemTime = chrono::system_clock::now();
     if (slsResponse.mStatusCode == 200) {

core/prometheus/Constants.h

@@ -50,6 +50,7 @@ const char* const SCHEME = "scheme";
 const char* const METRICS_PATH = "metrics_path";
 const char* const SCRAPE_INTERVAL = "scrape_interval";
 const char* const SCRAPE_TIMEOUT = "scrape_timeout";
+const char* const SCRAPE_PROTOCOLS = "scrape_protocols";
 const char* const HEADERS = "headers";
 const char* const PARAMS = "params";
 const char* const QUERY_STRING = "query_string";
@@ -71,6 +72,14 @@ const char* const PASSWORD = "password";
 const char* const PASSWORD_FILE = "password_file";
 const char* const BASIC_PREFIX = "Basic ";
 
+// scrape protocols, from https://prometheus.io/docs/prometheus/latest/configuration/configuration/#scrape_config
+// text/plain, application/openmetrics-text will be used
+// version of openmetrics is 1.0.0 or 0.0.1, from
+// https://github.com/OpenObservability/OpenMetrics/blob/main/specification/OpenMetrics.md#extensions-and-improvements
+const char* const PrometheusProto = "PrometheusProto";
+const char* const PrometheusText0_0_4 = "PrometheusText0.0.4";
+const char* const OpenMetricsText0_0_1 = "OpenMetricsText0.0.1";
+const char* const OpenMetricsText1_0_0 = "OpenMetricsText1.0.0";
 
 // metric labels
 const char* const JOB = "job";
@@ -94,4 +103,10 @@ const char* const UP = "up";
 
 const char* const SCRAPE_TIMESTAMP_MILLISEC = "scrape_timestamp_millisec";
 
+// scrape config compression
+const char* const ENABLE_COMPRESSION = "enable_compression";
+const char* const ACCEPT_ENCODING = "Accept-Encoding";
+const char* const GZIP = "gzip";
+const char* const IDENTITY = "identity";
+
 } // namespace logtail::prometheus