feat: consumer add log id

Author: crimson-gao

consumer/worker_test.go

@@ -82,3 +82,40 @@ func TestConsumerQueryNoData(t *testing.T) {
 	worker.StopAndWait()
 
 }
+
+func TestConsumerWithLogId(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,
+	}
+
+	worker := InitConsumerWorkerWithCheckpointTracker(option, process_with_log_id)
+
+	worker.Start()
+	time.Sleep(time.Second * 2000)
+	worker.StopAndWait()
+
+}
+
+func process_with_log_id(shardId int, logGroupList *sls.LogGroupList, checkpointTracker CheckPointTracker) (string, error) {
+	fmt.Printf("time: %s, shardId %d processing works success, logGroupSize: %d, currentCursor: %s\n",
+		time.Now().Format("2006-01-02 15:04:05 000"),
+		shardId, len(logGroupList.LogGroups),
+		checkpointTracker.GetCurrentCursor())
+	for _, logGroup := range logGroupList.LogGroups {
+		lgoGroupId := logGroup.GetLogGroupId()
+		fmt.Println("log group id: ", lgoGroupId)
+		for i, log := range logGroup.Logs {
+			log_id := logGroup.GetLogId(i)
+			fmt.Printf("log %d has %d keys, and log id: %s\n", i, len(log.Contents), log_id)
+		}
+	}
+	return "", nil
+}

example/consumer/with_log_id/main.go

@@ -0,0 +1,53 @@
+package main
+
+import (
+	"fmt"
+	"os"
+	"time"
+
+	sls "github.com/aliyun/aliyun-log-go-sdk"
+	consumerLibrary "github.com/aliyun/aliyun-log-go-sdk/consumer"
+)
+
+func main() {
+	option := consumerLibrary.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:    consumerLibrary.END_CURSOR,
+	}
+
+	worker := consumerLibrary.InitConsumerWorkerWithCheckpointTracker(option, process_with_log_id)
+
+	worker.Start()
+	defer worker.StopAndWait()
+	for {
+		time.Sleep(time.Second)
+	}
+	// worker.StopAndWait()
+}
+
+func process_with_log_id(shardId int, logGroupList *sls.LogGroupList, checkpointTracker consumerLibrary.CheckPointTracker) (string, error) {
+	fmt.Printf("time: %s, shardId %d processing works success, logGroupSize: %d,\n",
+		time.Now().Format("2006-01-02 15:04:05 000"),
+		shardId, len(logGroupList.LogGroups))
+
+	// start consume logs
+	for _, logGroup := range logGroupList.LogGroups {
+		// logGroupId is empty string if failed
+		logGroupId := logGroup.GetLogGroupId() // eg: "4|MTczMjE1NDI4NzIxMzIzMjAzNg=="
+		fmt.Println("log group id: ", logGroupId)
+
+		for i, log := range logGroup.Logs {
+			// log id is empty string if failed
+			log_id := logGroup.GetLogId(i) // eg: "4|MTczMjE1NDI4NzIxMzIzMjAzNg==|5|4"
+			fmt.Printf("log %d has %d keys, and log id: %s\n", i, len(log.Contents), log_id)
+		}
+	}
+	return "", nil
+}

log.pb.go

@@ -547,11 +547,13 @@ func (s *LogStore) GetLogsBytesWithQuery(plr *PullLogRequest) ([]byte, *PullLogM
 	if err != nil {
 		return nil, nil, err
 	}
+	readLastCursor, _ := parseHeaderString(r.Header, "X-Log-Read-Last-Cursor")
 	pullMeta := &PullLogMeta{
-		RawSize:    rawSize,
-		NextCursor: nextCursor,
-		Netflow:    netflow,
-		Count:      count,
+		RawSize:        rawSize,
+		NextCursor:     nextCursor,
+		Netflow:        netflow,
+		Count:          count,
+		readLastCursor: readLastCursor,
 	}
 	// If query is not nil, extract more headers
 	if plr.Query != "" {
@@ -640,7 +642,9 @@ func (s *LogStore) PullLogsWithQuery(plr *PullLogRequest) (gl *LogGroupList, plm
 	if err != nil {
 		return nil, nil, err
 	}
-
+	if plm.Count > 0 && plm.readLastCursor != "" {
+		gl.addIdIfPossible(plm.readLastCursor, plr.ShardID)
+	}
 	return
 }
 

log_store.go

@@ -2,10 +2,13 @@ package sls
 
 import (
 	"encoding/json"
+	"fmt"
 	"net/http"
 	"net/url"
 	"strconv"
 	"strings"
+
+	"github.com/go-kit/kit/log/level"
 )
 
 // GetLogRequest for GetLogsV2
@@ -75,10 +78,11 @@ func (plr *PullLogRequest) ToURLParams() url.Values {
 }
 
 type PullLogMeta struct {
-	NextCursor string
-	Netflow    int
-	RawSize    int
-	Count      int
+	NextCursor     string
+	Netflow        int
+	RawSize        int
+	Count          int
+	readLastCursor string // int64 string, eg: "1732154287213232020"
 	// these fields are only present when query is set
 	RawSizeBeforeQuery   int // processed raw size before query
 	Lines                int // result lines after query
@@ -375,3 +379,50 @@ type ListStoreViewsResponse struct {
 	Count      int      `json:"count"`
 	StoreViews []string `json:"storeviews"`
 }
+
+type logGroupIdentity struct {
+	cursor string
+	shard  int
+}
+
+// GetLogGroupId returns the log group id (shard|cursor)
+// If id is unknown, returns empty string
+func (l *LogGroup) GetLogGroupId() string {
+	if l.identity == nil {
+		return ""
+	}
+	return fmt.Sprintf("%d|%s", l.identity.shard, l.identity.cursor)
+}
+
+// index must be less than len(LogGroup.Logs)
+// if no log id found, the returned log id is empty string
+func (l *LogGroup) GetLogId(index int) string {
+	if index > len(l.Logs) {
+		return ""
+	}
+	logGroupId := l.GetLogGroupId()
+	if logGroupId == "" {
+		return ""
+	}
+	return fmt.Sprintf("%s|%d|%d", logGroupId, len(l.Logs), index)
+}
+
+func (l *LogGroupList) addIdIfPossible(readLastCursor string, shard int) error {
+	lastCursorInt, err := strconv.ParseInt(readLastCursor, 10, 64)
+	if err != nil {
+		if IsDebugLevelMatched(1) {
+			level.Error(Logger).Log("msg", "decode readLastCursor failed",
+				"cursor", readLastCursor, "err", err)
+		}
+		return err
+	}
+	cursor := lastCursorInt - int64(len(l.LogGroups)) + 1
+	for i := 0; i < len(l.LogGroups); i++ {
+		l.LogGroups[i].identity = &logGroupIdentity{
+			cursor: encodeCursor(cursor),
+			shard:  shard,
+		}
+		cursor++
+	}
+	return nil
+}

model.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"testing"
 
+	"github.com/golang/protobuf/proto"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -70,3 +71,39 @@ func TestIndex_MarshalJSON(t *testing.T) {
 	}
 
 }
+
+func TestLogGroupIdentity(t *testing.T) {
+
+	sampleLog := &Log{
+		Time: proto.Uint32(1732774880),
+		Contents: []*LogContent{
+			{
+				Key:   proto.String("test"),
+				Value: proto.String("test"),
+			},
+		},
+	}
+	logGroupList := &LogGroupList{
+		LogGroups: []*LogGroup{
+			{
+				Logs: []*Log{sampleLog},
+			},
+			{
+				Logs: []*Log{sampleLog, sampleLog},
+			},
+			{
+				Logs: []*Log{sampleLog, sampleLog, sampleLog},
+			},
+		},
+	}
+	err := logGroupList.addIdIfPossible("MTcyOTA3MDYxODQyODA0NDY1NQ==", 0)
+	assert.NoError(t, err)
+
+	assert.Equal(t, "0|MTcyOTA3MDYxODQyODA0NDY1NQ==", logGroupList.LogGroups[2].GetLogGroupId())
+	assert.Equal(t, "0|MTcyOTA3MDYxODQyODA0NDY1NA==", logGroupList.LogGroups[1].GetLogGroupId())
+	assert.Equal(t, "0|MTcyOTA3MDYxODQyODA0NDY1Mw==", logGroupList.LogGroups[0].GetLogGroupId())
+	empty := &LogGroup{}
+	assert.Equal(t, empty.GetLogGroupId(), "")
+	assert.Equal(t, "0|MTcyOTA3MDYxODQyODA0NDY1NA==|2|1", logGroupList.LogGroups[1].GetLogId(1))
+	assert.Equal(t, "0|MTcyOTA3MDYxODQyODA0NDY1NQ==|3|0", logGroupList.LogGroups[2].GetLogId(0))
+}

model_test.go

@@ -0,0 +1,19 @@
+package sls
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestCursorEncode(t *testing.T) {
+	assert.Equal(t, encodeCursor(1729070618428044655), "MTcyOTA3MDYxODQyODA0NDY1NQ==")
+	cursor, err := decodeCursor("MTcyOTA3MDYxODQyODA0NDY1NQ==")
+	assert.Nil(t, err)
+	assert.Equal(t, cursor, int64(1729070618428044655))
+
+	assert.Equal(t, encodeCursor(0), "MA==")
+	cursor, err = decodeCursor("MA==")
+	assert.Nil(t, err)
+	assert.Equal(t, cursor, int64(0))
+}

util_test.go

@@ -1,6 +1,7 @@
 package sls
 
 import (
+	"encoding/base64"
 	"fmt"
 	"net/http"
 	"strconv"
@@ -50,3 +51,19 @@ func parseHeaderString(header http.Header, headerName string) (string, error) {
 	}
 	return v[0], nil
 }
+
+func decodeCursor(cursor string) (int64, error) {
+	c, err := base64.StdEncoding.DecodeString(cursor)
+	if err != nil {
+		return 0, err
+	}
+	cursorInt, err := strconv.ParseInt(string(c), 10, 64)
+	if err != nil {
+		return 0, err
+	}
+	return cursorInt, nil
+}
+
+func encodeCursor(cursor int64) string {
+	return base64.StdEncoding.EncodeToString([]byte(strconv.FormatInt(cursor, 10)))
+}