support user define PushNativeProcessMaxCachedSize

alibaba · Sep 23, 2024 · b675236 · b675236
1 parent 59db229
commit b675236
Show file tree

Hide file tree

Showing 5 changed files with 11 additions and 9 deletions.
diff --git a/pkg/config/global_config.go b/pkg/config/global_config.go
@@ -42,6 +42,8 @@ type GlobalConfig struct {
 
 	// Go Input to Native Processor
 	GoInputToNativeProcessor bool
+	PushNativeMaxCachedSize  int
+	PushNativeTimeoutMs      int
 }
 
 // LogtailGlobalConfig is the singleton instance of GlobalConfig.
@@ -60,6 +62,8 @@ func newGlobalConfig() (cfg GlobalConfig) {
 		DefaultLogGroupQueueSize: 4,
 		LogtailSysConfDir:        ".",
 		DelayStopSec:             300,
+		PushNativeMaxCachedSize:  100,
+		PushNativeTimeoutMs:      1000,
 	}
 	return
 }
diff --git a/pluginmanager/plugin_runner_v1.go b/pluginmanager/plugin_runner_v1.go
@@ -147,9 +147,8 @@ func (p *pluginv1Runner) addMetricInput(pluginMeta *pipeline.PluginMeta, input p
 	wrapper.LogsChan = p.LogsChan
 	wrapper.LatencyMetric = p.LogstoreConfig.Statistics.CollecLatencytMetric
 	if p.LogstoreConfig.GlobalConfig.GoInputToNativeProcessor {
-		// TODO : give config for MaxCachedSize and PushNativeTimeout
-		wrapper.MaxCachedSize = 1000
-		wrapper.PushNativeTimeout = time.Duration(1000) * time.Millisecond
+		wrapper.MaxCachedSize = p.LogstoreConfig.GlobalConfig.PushNativeMaxCachedSize
+		wrapper.PushNativeTimeout = time.Duration(p.LogstoreConfig.GlobalConfig.PushNativeTimeoutMs) * time.Millisecond
 	}
 	p.MetricPlugins = append(p.MetricPlugins, &wrapper)
 	return wrapper.Init(pluginMeta, inputInterval)
@@ -161,9 +160,8 @@ func (p *pluginv1Runner) addServiceInput(pluginMeta *pipeline.PluginMeta, input
 	wrapper.Input = input
 	wrapper.LogsChan = p.LogsChan
 	if p.LogstoreConfig.GlobalConfig.GoInputToNativeProcessor {
-		// TODO : give config for MaxCachedSize and PushNativeTimeout
-		wrapper.MaxCachedSize = 1000
-		wrapper.PushNativeTimeout = time.Duration(1000) * time.Millisecond
+		wrapper.MaxCachedSize = p.LogstoreConfig.GlobalConfig.PushNativeMaxCachedSize
+		wrapper.PushNativeTimeout = time.Duration(p.LogstoreConfig.GlobalConfig.PushNativeTimeoutMs) * time.Millisecond
 	}
 	p.ServicePlugins = append(p.ServicePlugins, &wrapper)
 	return wrapper.Init(pluginMeta)

diff --git a/plugins/input/goprofile/golang_profile.go b/plugins/input/goprofile/golang_profile.go
@@ -73,5 +73,5 @@ func init() {
 }
 
 func (g *GoProfile) GetMode() pipeline.InputModeType {
-	return pipeline.PUSH
+	return pipeline.PULL
 }
diff --git a/plugins/input/mysql/mysql.go b/plugins/input/mysql/mysql.go
@@ -465,5 +465,5 @@ func init() {
 }
 
 func (m *Mysql) GetMode() pipeline.InputModeType {
-	return pipeline.PUSH
+	return pipeline.PULL
 }
diff --git a/plugins/input/redis/input_redis.go b/plugins/input/redis/input_redis.go
@@ -275,5 +275,5 @@ func init() {
 }
 
 func (r *InputRedis) GetMode() pipeline.InputModeType {
-	return pipeline.PUSH
+	return pipeline.PULL
 }