add UsingOldContentTag. When UsingOldContentTag is set to false, the tag is now put into the context during cgo.

core/config/Config.h

@@ -92,6 +92,7 @@ class Config {
         uint16_t mSearchCheckpointDirDepth = 0; // Max directory depth when search checkpoint.
 
         bool mEnableTimestampNanosecond = false;
+        bool mUsingOldContentTag = false;
         // Deprecated
         bool mEnablePreciseTimestamp = false;
         std::string mPreciseTimestampKey;

core/config/UserLogConfigParser.cpp

@@ -150,6 +150,12 @@ void UserLogConfigParser::ParseAdvancedConfig(const Json::Value& originalVal, Co
             cfg.mAdvancedConfig.mSpecifiedYear = static_cast<int32_t>(val.asUInt());
         }
     }
+    // using_old_content_tag
+    {
+        if (advancedVal.isMember("using_old_content_tag") && advancedVal["using_old_content_tag"].isBool()) {
+            cfg.mAdvancedConfig.mUsingOldContentTag = GetBoolValue(advancedVal, "using_old_content_tag");
+        }
+    }
     // precise_timestamp
     {
         if (advancedVal.isMember("enable_timestamp_nanosecond") && advancedVal["enable_timestamp_nanosecond"].isBool()) {

core/config_manager/ConfigManagerBase.cpp

@@ -867,11 +867,17 @@ void ConfigManagerBase::LoadSingleUserConfig(const std::string& logName, const J
                 SetNotFoundJsonMember(pluginConfigJson["global"],
                                       "EnableTimestampNanosecond",
                                       config->mAdvancedConfig.mEnableTimestampNanosecond);
+                SetNotFoundJsonMember(pluginConfigJson["global"],
+                                      "UsingOldContentTag",
+                                      config->mAdvancedConfig.mUsingOldContentTag);
             } else {
                 Json::Value pluginGlobalConfigJson;
                 SetNotFoundJsonMember(pluginGlobalConfigJson,
                                       "EnableTimestampNanosecond",
                                       config->mAdvancedConfig.mEnableTimestampNanosecond);
+                SetNotFoundJsonMember(pluginGlobalConfigJson,
+                                      "UsingOldContentTag",
+                                      config->mAdvancedConfig.mUsingOldContentTag);
                 pluginConfigJson["global"] = pluginGlobalConfigJson;
             }
             config->mPluginConfig = pluginConfigJson.toStyledString();

core/config_manager/ConfigYamlToJson.cpp

@@ -96,6 +96,7 @@ ConfigYamlToJson::ConfigYamlToJson() {
     mFileAdvancedConfigMap["DirBlackList"] = "dir_blacklist";
     mFileAdvancedConfigMap["FilepathBlackList"] = "filepath_blacklist";
     mFileAdvancedConfigMap["EnableTimestampNanosecond"] = "enable_timestamp_nanosecond";
+    mFileAdvancedConfigMap["UsingOldContentTag"] = "using_old_content_tag";
     mFileAdvancedConfigMap["EnablePreciseTimestamp"] = "enable_precise_timestamp";
     mFileAdvancedConfigMap["PreciseTimestampKey"] = "precise_timestamp_key";
     mFileAdvancedConfigMap["PreciseTimestampUnit"] = "precise_timestamp_unit";

pkg/config/global_config.go

@@ -36,6 +36,7 @@ type GlobalConfig struct {
 	DelayStopSec int
 
 	EnableTimestampNanosecond      bool
+	UsingOldContentTag             bool
 	EnableContainerdUpperDirDetect bool
 	EnableSlsMetricsFormat         bool
 }

pluginmanager/logstore_config.go

@@ -298,13 +298,17 @@ func (lc *LogstoreConfig) ProcessLogGroup(logByte []byte, packID string) int {
 		if len(topic) > 0 {
 			log.Contents = append(log.Contents, &protocol.Log_Content{Key: "__log_topic__", Value: topic})
 		}
-		for _, tag := range logGroup.LogTags {
-			log.Contents = append(log.Contents, &protocol.Log_Content{
-				Key:   tagPrefix + tag.GetKey(),
-				Value: tag.GetValue(),
-			})
+		if lc.GlobalConfig.UsingOldContentTag {
+			for _, tag := range logGroup.LogTags {
+				log.Contents = append(log.Contents, &protocol.Log_Content{
+					Key:   tagPrefix + tag.GetKey(),
+					Value: tag.GetValue(),
+				})
+			}
+			lc.PluginRunner.ReceiveRawLog(&pipeline.LogWithContext{Log: log, Context: map[string]interface{}{"source": packID, "topic": topic}})
+		} else {
+			lc.PluginRunner.ReceiveRawLog(&pipeline.LogWithContext{Log: log, Context: map[string]interface{}{"source": packID, "topic": topic, "tags": logGroup.LogTags}})
 		}
-		lc.PluginRunner.ReceiveRawLog(&pipeline.LogWithContext{Log: log, Context: map[string]interface{}{"source": packID, "topic": topic}})
 	}
 	return 0
 }

pluginmanager/plugin_runner_v1.go

@@ -65,9 +65,16 @@ func (p *pluginv1Runner) Init(inputQueueSize int, flushQueueSize int) error {
 
 func (p *pluginv1Runner) Initialized() error {
 	if len(p.AggregatorPlugins) == 0 {
-		logger.Debug(p.LogstoreConfig.Context.GetRuntimeContext(), "add default aggregator")
-		if err := loadAggregator("aggregator_default", p.LogstoreConfig, nil); err != nil {
-			return err
+		if p.LogstoreConfig.GlobalConfig.UsingOldContentTag {
+			logger.Debug(p.LogstoreConfig.Context.GetRuntimeContext(), "add default aggregator")
+			if err := loadAggregator("aggregator_default", p.LogstoreConfig, nil); err != nil {
+				return err
+			}
+		} else {
+			logger.Debug(p.LogstoreConfig.Context.GetRuntimeContext(), "add aggregator_context")
+			if err := loadAggregator("aggregator_context", p.LogstoreConfig, nil); err != nil {
+				return err
+			}
 		}
 	}
 	if len(p.FlusherPlugins) == 0 {

plugins/aggregator/context/aggregator_context.go

@@ -95,16 +95,36 @@ func (p *AggregatorContext) Add(log *protocol.Log, ctx map[string]interface{}) e
 		logGroupList = make([]*protocol.LogGroup, 0, p.MaxLogGroupCount)
 	}
 	if len(logGroupList) == 0 {
-		logGroupList = append(logGroupList, p.newLogGroup(source, topic))
+		if _, ok := ctx["tags"]; ok {
+			newLogGroup := p.newLogGroup(source, topic)
+			fillTags(ctx["tags"].([]*protocol.LogTag), newLogGroup)
+			logGroupList = append(logGroupList, newLogGroup)
+		} else {
+			logGroupList = append(logGroupList, p.newLogGroup(source, topic))
+		}
 	}
 	nowLogGroup := logGroupList[len(logGroupList)-1]
+	tagChanged := false
+	// Determine whether the tag of the current log is consistent with that of the previous logs. If it is inconsistent, need to create a new logGroup.
+	if _, ok := ctx["tags"]; ok && len(nowLogGroup.Logs) > 0 {
+		tags := map[string]string{}
+		for _, tag := range nowLogGroup.LogTags {
+			tags[tag.GetKey()] = tag.GetValue()
+		}
+		for _, tag := range ctx["tags"].([]*protocol.LogTag) {
+			if v, ok := tags[tag.GetKey()]; !ok || v != tag.GetValue() {
+				tagChanged = true
+				break
+			}
+		}
+	}
 
 	logSize := p.evaluateLogSize(log)
 	// When current log group is full (log count or no more capacity for current log),
 	// allocate a new log group.
-	if len(nowLogGroup.Logs) >= p.MaxLogCount || p.nowLogGroupSizeMap[source]+logSize > MaxLogGroupSize {
+	if len(nowLogGroup.Logs) >= p.MaxLogCount || p.nowLogGroupSizeMap[source]+logSize > MaxLogGroupSize || tagChanged {
 		// The number of log group exceeds limit, make a quick flush.
-		if len(logGroupList) == p.MaxLogGroupCount {
+		if len(logGroupList) == p.MaxLogGroupCount || (tagChanged && len(logGroupList) > 0) {
 			// Quick flush to avoid becoming bottleneck when large logs come.
 			if err := p.queue.Add(logGroupList[0]); err == nil {
 				// add success, remove head log group
@@ -115,7 +135,13 @@ func (p *AggregatorContext) Add(log *protocol.Log, ctx map[string]interface{}) e
 		}
 		// New log group, reset size.
 		p.nowLogGroupSizeMap[source] = 0
-		logGroupList = append(logGroupList, p.newLogGroup(source, topic))
+		if _, ok := ctx["tags"]; ok {
+			newLogGroup := p.newLogGroup(source, topic)
+			fillTags(ctx["tags"].([]*protocol.LogTag), newLogGroup)
+			logGroupList = append(logGroupList, newLogGroup)
+		} else {
+			logGroupList = append(logGroupList, p.newLogGroup(source, topic))
+		}
 		nowLogGroup = logGroupList[len(logGroupList)-1]
 	}
 
@@ -126,6 +152,10 @@ func (p *AggregatorContext) Add(log *protocol.Log, ctx map[string]interface{}) e
 	return nil
 }
 
+func fillTags(logTags []*protocol.LogTag, logGroup *protocol.LogGroup) {
+	logGroup.LogTags = append(logGroup.LogTags, logTags...)
+}
+
 // Flush ...
 func (p *AggregatorContext) Flush() []*protocol.LogGroup {
 	p.lock.Lock()