Parse protobuf with custom parser

client_interface.go

@@ -280,8 +280,9 @@ type ClientInterface interface {
 	// @note if you want to pull logs continuous, set endCursor = ""
 	PullLogs(project, logstore string, shardID int, cursor, endCursor string,
 		logGroupMaxCount int) (gl *LogGroupList, nextCursor string, err error)
-	// Deprecated: Use PullLogsWithQuery instead.
+	// Deprecated: Use PullLogsV3 instead.
 	PullLogsV2(plr *PullLogRequest) (gl *LogGroupList, nextCursor string, err error)
+	PullLogsV3(plr *PullLogRequest) (*PullLogsResponse, error)
 	PullLogsWithQuery(plr *PullLogRequest) (gl *LogGroupList, plm *PullLogMeta, err error)
 	// GetHistograms query logs with [from, to) time range
 	GetHistograms(project, logstore string, topic string, from int64, to int64, queryExp string) (*GetHistogramsResponse, error)

client_store.go

@@ -228,6 +228,11 @@ func (c *Client) PullLogsV2(plr *PullLogRequest) (gl *LogGroupList, nextCursor s
 	return ls.PullLogsV2(plr)
 }
 
+func (c *Client) PullLogsV3(plr *PullLogRequest) (r *PullLogsResponse, err error) {
+	ls := convertLogstore(c, plr.Project, plr.Logstore)
+	return ls.PullLogsV3(plr)
+}
+
 func (c *Client) PullLogsWithQuery(plr *PullLogRequest) (gl *LogGroupList, plm *PullLogMeta, err error) {
 	ls := convertLogstore(c, plr.Project, plr.Logstore)
 	return ls.PullLogsWithQuery(plr)

consumer/consumer_client.go

@@ -139,7 +139,7 @@ func (consumer *ConsumerClient) getCursor(shardId int, from string) (string, err
 	return cursor, err
 }
 
-func (consumer *ConsumerClient) pullLogs(shardId int, cursor string) (gl *sls.LogGroupList, plm *sls.PullLogMeta, err error) {
+func (consumer *ConsumerClient) pullLogs(shardId int, cursor string) (r *sls.PullLogsResponse, err error) {
 	plr := &sls.PullLogRequest{
 		Project:          consumer.option.Project,
 		Logstore:         consumer.option.Logstore,
@@ -149,28 +149,29 @@ func (consumer *ConsumerClient) pullLogs(shardId int, cursor string) (gl *sls.Lo
 		LogGroupMaxCount: consumer.option.MaxFetchLogGroupCount,
 	}
 	for retry := 0; retry < 3; retry++ {
-		gl, plm, err = consumer.client.PullLogsWithQuery(plr)
-		if err != nil {
-			slsError, ok := err.(*sls.Error)
-			if ok {
-				level.Warn(consumer.logger).Log("msg", "shard pull logs failed, occur sls error",
-					"shard", shardId,
-					"error", slsError,
-					"tryTimes", retry+1,
-					"cursor", cursor,
-				)
-				if slsError.HTTPCode == 403 {
-					time.Sleep(5 * time.Second)
-				}
-			} else {
-				level.Warn(consumer.logger).Log("msg", "unknown error when pull log",
-					"shardId", shardId,
-					"cursor", cursor,
-					"error", err,
-					"tryTimes", retry+1)
+		r, err = consumer.client.PullLogsV3(plr)
+		if err == nil || retry >= 2 {
+			break
+		}
+		slsError, ok := err.(*sls.Error)
+		if ok {
+			level.Warn(consumer.logger).Log("msg", "shard pull logs failed, occur sls error",
+				"shard", shardId,
+				"error", slsError,
+				"tryTimes", retry+1,
+				"cursor", cursor,
+			)
+			if slsError.HTTPCode == 403 {
+				time.Sleep(5 * time.Second)
 			}
-			time.Sleep(200 * time.Millisecond)
+		} else {
+			level.Warn(consumer.logger).Log("msg", "unknown error when pull log",
+				"shardId", shardId,
+				"cursor", cursor,
+				"error", err,
+				"tryTimes", retry+1)
 		}
+		time.Sleep(200 * time.Millisecond)
 	}
 	// If you can't retry the log three times, it will return to empty list and start pulling the log cursor,
 	// so that next time you will come in and pull the function again, which is equivalent to a dead cycle.

consumer/tasks.go

@@ -3,6 +3,7 @@ package consumerLibrary
 import (
 	"errors"
 	"fmt"
+	sls "github.com/aliyun/aliyun-log-go-sdk"
 	"runtime"
 	"time"
 
@@ -49,19 +50,21 @@ func (consumer *ShardConsumerWorker) nextFetchTask() error {
 	// update last fetch time, for control fetch frequency
 	consumer.lastFetchTime = time.Now()
 
-	logGroup, pullLogMeta, err := consumer.client.pullLogs(consumer.shardId, consumer.nextFetchCursor)
+	r, err := consumer.client.pullLogs(consumer.shardId, consumer.nextFetchCursor)
 	if err != nil {
 		return err
 	}
 	// set cursors user to decide whether to save according to the execution of `process`
 	consumer.consumerCheckPointTracker.setCurrentCursor(consumer.nextFetchCursor)
-	consumer.lastFetchLogGroupList = logGroup
-	consumer.nextFetchCursor = pullLogMeta.NextCursor
-	consumer.lastFetchRawSize = pullLogMeta.RawSize
-	consumer.lastFetchGroupCount = GetLogGroupCount(consumer.lastFetchLogGroupList)
+	consumer.lastFetchLogGroupList = &sls.LogGroupList{
+		FastLogGroups: r.LogGroups,
+	}
+	consumer.nextFetchCursor = r.NextCursor
+	consumer.lastFetchRawSize = r.RawSize
+	consumer.lastFetchGroupCount = len(r.LogGroups)
 	if consumer.client.option.Query != "" {
-		consumer.lastFetchRawSizeBeforeQuery = pullLogMeta.RawSizeBeforeQuery
-		consumer.lastFetchGroupCountBeforeQuery = pullLogMeta.RawDataCountBeforeQuery
+		consumer.lastFetchRawSizeBeforeQuery = r.RawSizeBeforeQuery
+		consumer.lastFetchGroupCountBeforeQuery = r.RawDataCountBeforeQuery
 		if consumer.lastFetchRawSizeBeforeQuery == -1 {
 			consumer.lastFetchRawSizeBeforeQuery = 0
 		}

consumer/worker.go

@@ -1,8 +1,8 @@
 package consumerLibrary
 
 import (
-	"os"
 	"io"
+	"os"
 	"sync"
 	"time"
 
@@ -143,14 +143,14 @@ func (consumerWorker *ConsumerWorker) getShardConsumer(shardId int) *ShardConsum
 
 }
 
-func (consumerWorker *ConsumerWorker) cleanShardConsumer(owned_shards []int) {
+func (consumerWorker *ConsumerWorker) cleanShardConsumer(ownedShards []int) {
 
 	consumerWorker.shardConsumer.Range(
 		func(key, value interface{}) bool {
 			shard := key.(int)
 			consumer := value.(*ShardConsumerWorker)
 
-			if !Contain(shard, owned_shards) {
+			if !Contain(shard, ownedShards) {
 				level.Info(consumerWorker.Logger).Log("msg", "try to call shut down for unassigned consumer shard", "shardId", shard)
 				consumer.consumerShutDown()
 				level.Info(consumerWorker.Logger).Log("msg", "Complete call shut down for unassigned consumer shard", "shardId", shard)

example/consumer/copy_data/copy_data.go

@@ -29,7 +29,7 @@ func main() {
 		Logstore:          "",
 		ConsumerGroupName: "",
 		ConsumerName:      "",
-		// This options is used for initialization, will be ignored once consumer group is created and each shard has been started to be consumed.
+		// These options are used for initialization, will be ignored once consumer group is created and each shard has been started to be consumed.
 		// Could be "begin", "end", "specific time format in time stamp", it's log receiving time.
 		CursorPosition: consumerLibrary.BEGIN_CURSOR,
 	}

example/consumer/demo/simple_demo.go

@@ -23,9 +23,9 @@ func main() {
 		Logstore:          "",
 		ConsumerGroupName: "",
 		ConsumerName:      "",
-		// This options is used for initialization, will be ignored once consumer group is created and each shard has been started to be consumed.
+		// These options are used for initialization, will be ignored once consumer group is created and each shard has been started to be consumed.
 		// Could be "begin", "end", "specific time format in time stamp", it's log receiving time.
-		CursorPosition: consumerLibrary.END_CURSOR,
+		CursorPosition: consumerLibrary.BEGIN_CURSOR,
 	}
 
 	consumerWorker := consumerLibrary.InitConsumerWorkerWithCheckpointTracker(option, process)
@@ -41,7 +41,22 @@ func main() {
 // Fill in your consumption logic here, and be careful not to change the parameters of the function and the return value,
 // otherwise you will report errors.
 func process(shardId int, logGroupList *sls.LogGroupList, checkpointTracker consumerLibrary.CheckPointTracker) (string, error) {
-	fmt.Println(shardId, logGroupList)
+	//fmt.Println(shardId, logGroupList)
+	fmt.Println(shardId)
+	for _, lg := range logGroupList.FastLogGroups {
+		for _, tag := range lg.LogTags {
+			fmt.Printf("[tag] %s : %s\n", tag.Key, tag.Value)
+		}
+		fmt.Printf("[source] %s\n", lg.Source)
+		fmt.Printf("[topic] %s\n", lg.Topic)
+		for _, log := range lg.Logs {
+			fmt.Printf("[log] time = %d, nsTimePs = %d\n", log.Time, log.TimeNs)
+			for _, content := range log.Contents {
+				fmt.Printf("%s : %s\n", content.Key, content.Value)
+			}
+		}
+	}
+
 	checkpointTracker.SaveCheckPoint(false)
 	return "", nil
 }