Add ProcessorParseContainerLogNative

core/file_server/MultilineOptions.cpp

@@ -135,7 +135,24 @@ bool MultilineOptions::Init(const Json::Value& config, const PipelineContext& ct
                                      ctx.GetProjectName(),
                                      ctx.GetLogstoreName(),
                                      ctx.GetRegion());
-        } else if (mStartPatternRegPtr || mEndPatternRegPtr) {
+        } else if (mStartPatternRegPtr && mContinuePatternRegPtr && mEndPatternRegPtr) {
+            mContinuePatternRegPtr.reset();
+            LOG_WARNING(
+                ctx.GetLogger(),
+                ("problem encountered in config parsing",
+                 "none of param Multiline.StartPattern, Multiline.ContinuePattern and Multiline.EndPattern are empty")(
+                    "action", "ignore param Multiline.ContinuePattern")("module", pluginName)("config",
+                                                                                              ctx.GetConfigName()));
+            ctx.GetAlarm().SendAlarm(
+                CATEGORY_CONFIG_ALARM,
+                "none of param Multiline.StartPattern, Multiline.ContinuePattern and Multiline.EndPattern are empty: "
+                "ignore param Multiline.ContinuePattern, module: "
+                    + pluginName + ", config: " + ctx.GetConfigName(),
+                ctx.GetProjectName(),
+                ctx.GetLogstoreName(),
+                ctx.GetRegion());
+        }
+        if (mStartPatternRegPtr || mEndPatternRegPtr) {
             mIsMultiline = true;
         }
     }
@@ -181,4 +198,17 @@ bool MultilineOptions::ParseRegex(const string& pattern, shared_ptr<boost::regex
     return true;
 }
 
+const std::string& MultilineOptions::UnmatchedContentTreatmentToString() {
+    switch (mUnmatchedContentTreatment) {
+        case UnmatchedContentTreatment::DISCARD:
+            static std::string discardStr = "discard";
+            return discardStr;
+        case UnmatchedContentTreatment::SINGLE_LINE:
+            static std::string singleLine = "single line";
+            return singleLine;
+        default:
+            static std::string unkonwn = "";
+            return unkonwn;
+    }
+}
 } // namespace logtail

core/file_server/MultilineOptions.h

@@ -28,6 +28,7 @@ namespace logtail {
 
 class MultilineOptions {
 public:
+    const std::string& UnmatchedContentTreatmentToString();
     enum class Mode { CUSTOM, JSON };
     enum class UnmatchedContentTreatment { DISCARD, SINGLE_LINE };
 

core/models/LogEvent.cpp

@@ -87,7 +87,7 @@ bool LogEvent::FromJson(const Json::Value& root) {
     if (root.isMember("contents")) {
         Json::Value contents = root["contents"];
         for (const auto& key : contents.getMemberNames()) {
-            SetContent(key, contents[key].asString());
+            SetContent(GetSourceBuffer()->CopyString(key), contents[key].asString());
         }
     }
     return true;

core/models/PipelineEventGroup.cpp

@@ -157,6 +157,7 @@ const std::string EVENT_GROUP_META_LOG_FILE_PATH_RESOLVED = "log.file.path_resol
 const std::string EVENT_GROUP_META_LOG_FILE_INODE = "log.file.inode";
 const std::string EVENT_GROUP_META_LOG_FILE_OFFSET = "log.file.offset";
 const std::string EVENT_GROUP_META_LOG_FILE_LENGTH = "log.file.length";
+const std::string EVENT_GROUP_META_CONTAINER_TYPE = "container.type";
 
 const std::string EVENT_GROUP_META_K8S_CLUSTER_ID = "k8s.cluster.id";
 const std::string EVENT_GROUP_META_K8S_NODE_NAME = "k8s.node.name";
@@ -189,6 +190,8 @@ const std::string& EventGroupMetaKeyToString(EventGroupMetaKey key) {
             return EVENT_GROUP_META_LOG_FILE_OFFSET;
         case EventGroupMetaKey::LOG_READ_LENGTH:
             return EVENT_GROUP_META_LOG_FILE_LENGTH;
+        case EventGroupMetaKey::LOG_FORMAT:
+            return EVENT_GROUP_META_CONTAINER_TYPE;
         default:
             static std::string sEmpty = "unknown";
             return sEmpty;

core/models/PipelineEventGroup.h

@@ -42,6 +42,7 @@ enum class EventGroupMetaKey {
     LOG_FILE_INODE,
     LOG_READ_OFFSET,
     LOG_READ_LENGTH,
+    LOG_FORMAT,
 
     K8S_CLUSTER_ID,
     K8S_NODE_NAME,

core/monitor/MetricConstants.cpp

@@ -40,6 +40,7 @@ const std::string METRIC_PROC_PARSE_IN_SIZE_BYTES = "proc_parse_in_size_bytes";
 const std::string METRIC_PROC_PARSE_OUT_SIZE_BYTES = "proc_parse_out_size_bytes";
 
 const std::string METRIC_PROC_PARSE_ERROR_TOTAL = "proc_parse_error_total";
+const std::string METRIC_PROC_PARSE_SUCCESS_TOTAL = "proc_parse_success_total";
 const std::string METRIC_PROC_KEY_COUNT_NOT_MATCH_ERROR_TOTAL = "proc_key_count_not_match_error_total";
 const std::string METRIC_PROC_HISTORY_FAILURE_TOTAL = "proc_history_failure_total";
 

core/monitor/MetricConstants.h

@@ -40,6 +40,7 @@ extern const std::string METRIC_PROC_TIME_MS;
 extern const std::string METRIC_PROC_PARSE_IN_SIZE_BYTES;
 extern const std::string METRIC_PROC_PARSE_OUT_SIZE_BYTES;
 extern const std::string METRIC_PROC_PARSE_ERROR_TOTAL;
+extern const std::string METRIC_PROC_PARSE_SUCCESS_TOTAL;
 extern const std::string METRIC_PROC_KEY_COUNT_NOT_MATCH_ERROR_TOTAL;
 extern const std::string METRIC_PROC_HISTORY_FAILURE_TOTAL;
 

core/plugin/PluginRegistry.cpp

@@ -40,7 +40,9 @@
 #include "plugin/creator/StaticProcessorCreator.h"
 #include "processor/ProcessorDesensitizeNative.h"
 #include "processor/ProcessorFilterNative.h"
+#include "processor/ProcessorMergeMultilineLogNative.h"
 #include "processor/ProcessorParseApsaraNative.h"
+#include "processor/ProcessorParseContainerLogNative.h"
 #include "processor/ProcessorParseDelimiterNative.h"
 #include "processor/ProcessorParseJsonNative.h"
 #include "processor/ProcessorParseRegexNative.h"
@@ -259,6 +261,8 @@ void PluginRegistry::LoadStaticPlugins() {
     RegisterProcessorCreator(new StaticProcessorCreator<ProcessorParseTimestampNative>());
     RegisterProcessorCreator(new StaticProcessorCreator<ProcessorTagNative>());
     RegisterProcessorCreator(new StaticProcessorCreator<ProcessorFilterNative>());
+    RegisterProcessorCreator(new StaticProcessorCreator<ProcessorMergeMultilineLogNative>());
+    RegisterProcessorCreator(new StaticProcessorCreator<ProcessorParseContainerLogNative>());
 #if defined(__linux__) && !defined(__ANDROID__)
     if (BOOL_FLAG(enable_processor_spl)) {
         RegisterProcessorCreator(new StaticProcessorCreator<ProcessorSPL>());