add spl consume processor (#348)

Co-authored-by: weilong.pwl <[email protected]>

Author: panawala

consumer/README.md

@@ -66,6 +66,7 @@ LogHubConfig是提供给用户的配置类，用于配置消费策略，您可
 |AutoCommitDisabled|是否禁用sdk自动提交checkpoint|非必填，默认不会禁用|
 |AutoCommitIntervalInMS|自动提交checkpoint的时间间隔|非必填，单位为MS，默认时间为60s|
 |Query|过滤规则  基于规则消费时必须设置对应规则 如 *| where a = 'xxx'|非必填|
+|Processor|消费处理器  基于规则消费时设置对应的消费处理标识 如 consume-processor-1, 推荐使用Processor代替Query|非必填|
 
 
 **自定义 logger**

consumer/config.go

@@ -16,6 +16,7 @@ type LogHubConfig struct {
 	//:param Project:
 	//:param Logstore:
 	//:param Query: Filter rules Corresponding rules must be set when consuming based on rules, such as *| where a = 'xxx'
+	//:param Processor: name of consume processor, default is empty
 	//:param ConsumerGroupName:
 	//:param ConsumerName:
 	//:param CursorPosition: This options is used for initialization, will be ignored once consumer group is created and each shard has beeen started to be consumed.
@@ -60,6 +61,7 @@ type LogHubConfig struct {
 	Project                   string
 	Logstore                  string
 	Query                     string
+	Processor                 string
 	ConsumerGroupName         string
 	ConsumerName              string
 	CursorPosition            string

consumer/consumer_client.go

@@ -148,6 +148,7 @@ func (consumer *ConsumerClient) pullLogs(shardId int, cursor string) (gl *sls.Lo
 		Query:            consumer.option.Query,
 		LogGroupMaxCount: consumer.option.MaxFetchLogGroupCount,
 		CompressType:     consumer.option.CompressType,
+		Processor:        consumer.option.Processor,
 	}
 	for retry := 0; retry < 3; retry++ {
 		gl, plm, err = consumer.client.PullLogsWithQuery(plr)

consumer/shard_worker.go

@@ -197,7 +197,7 @@ func (c *ShardConsumerWorker) sleepUtilNextFetch(lastFetchSuccessTime time.Time,
 
 	lastFetchRawSize := plm.RawSize
 	lastFetchGroupCount := plm.Count
-	if c.client.option.Query != "" {
+	if c.client.option.Query != "" || c.client.option.Processor != "" {
 		lastFetchRawSize = plm.RawSizeBeforeQuery
 		lastFetchGroupCount = plm.DataCountBeforeQuery
 	}

consumer/worker_test.go

@@ -83,6 +83,28 @@ func TestConsumerQueryNoData(t *testing.T) {
 
 }
 
+func TestConsumerProcessor(t *testing.T) {
+	option := LogHubConfig{
+		Endpoint: os.Getenv("LOG_TEST_ENDPOINT"),
+		CredentialsProvider: sls.NewStaticCredentialsProvider(
+			os.Getenv("LOG_TEST_ACCESS_KEY_ID"),
+			os.Getenv("LOG_TEST_ACCESS_KEY_SECRET"), ""),
+		Project:           os.Getenv("LOG_TEST_PROJECT"),
+		Logstore:          os.Getenv("LOG_TEST_LOGSTORE"),
+		ConsumerGroupName: "test-consumer",
+		ConsumerName:      "test-consumer-1",
+		CursorPosition:    END_CURSOR,
+		Processor:         "consume-processor-1753854227-0001",
+	}
+
+	worker := InitConsumerWorkerWithCheckpointTracker(option, process)
+
+	worker.Start()
+	time.Sleep(time.Second * 2000)
+	worker.StopAndWait()
+
+}
+
 func TestConsumerWithLogId(t *testing.T) {
 	option := LogHubConfig{
 		Endpoint: os.Getenv("LOG_TEST_ENDPOINT"),

example/consumer/query_demo/simple_demo_with_query.go

@@ -28,7 +28,11 @@ func main() {
 		CursorPosition:  consumerLibrary.SPECIAL_TIMER_CURSOR,
 		CursorStartTime: 1706077849,
 		// Query is for log pre-handling before return to client, more info refer to https://www.alibabacloud.com/help/zh/sls/user-guide/rule-based-consumption
-		Query: "* | where cast(body_bytes_sent as bigint) > 14000",
+		// Query: "* | where cast(body_bytes_sent as bigint) > 14000",
+
+		// Processor is for rule based consumption, more info refer to https://www.alibabacloud.com/help/zh/sls/user-guide/rule-based-consumption
+		// prefer to use processor instead of query, query will be deprecated in the future
+		Processor: "consume-processor-1753854227-0001",
 	}
 
 	consumerWorker := consumerLibrary.InitConsumerWorkerWithCheckpointTracker(option, process)

model.go

@@ -73,7 +73,7 @@ type PullLogRequest struct {
 	Query            string
 	// Deprecated: PullMode is not used
 	PullMode     string
-	QueryId      string
+	Processor    string
 	CompressType int
 }
 
@@ -88,10 +88,11 @@ func (plr *PullLogRequest) ToURLParams() url.Values {
 	if plr.Query != "" {
 		urlVal.Add("query", plr.Query)
 		urlVal.Add("pullMode", "scan_on_stream")
-		if plr.QueryId != "" {
-			urlVal.Add("queryId", plr.QueryId)
-		}
 	}
+	if plr.Processor != "" {
+		urlVal.Add("processor", plr.Processor)
+	}
+
 	return urlVal
 }