add getlogsV3ToComplete (aliyun#227)

Author: crimson-gao

client_interface.go

@@ -287,6 +287,8 @@ type ClientInterface interface {
 	GetLogsToCompleted(project, logstore string, topic string, from int64, to int64, queryExp string, maxLineNum int64, offset int64, reverse bool) (*GetLogsResponse, error)
 	// GetLogsToCompletedV2 query logs with [from, to) time range to completed
 	GetLogsToCompletedV2(project, logstore string, req *GetLogRequest) (*GetLogsResponse, error)
+	// GetLogsToCompletedV3 query logs with [from, to) time range to completed
+	GetLogsToCompletedV3(project, logstore string, req *GetLogRequest) (*GetLogsV3Response, error)
 
 	// #################### Index Operations #####################
 	// CreateIndex ...

client_store.go

@@ -280,6 +280,12 @@ func (c *Client) GetLogsToCompletedV2(project, logstore string, req *GetLogReque
 	return ls.GetLogsToCompletedV2(req)
 }
 
+// GetLogsToCompletedV3 ...
+func (c *Client) GetLogsToCompletedV3(project, logstore string, req *GetLogRequest) (*GetLogsV3Response, error) {
+	ls := convertLogstore(c, project, logstore)
+	return ls.GetLogsToCompletedV3(req)
+}
+
 // GetLogLinesV2 ...
 func (c *Client) GetLogLinesV2(project, logstore string, req *GetLogRequest) (*GetLogLinesResponse, error) {
 	ls := convertLogstore(c, project, logstore)

client_store_test.go

@@ -112,3 +112,35 @@ func (s *LostoreTestSuite) TestSplitShardDefault() {
 	assert.True(s.T(), id3 != -1)
 	assert.Equal(s.T(), 4, num3)
 }
+
+func (s *LostoreTestSuite) TestGetLogsV3ToCompleted() {
+	key := "key"
+	value := "val"
+	n := uint32(time.Now().Unix())
+	lg := &LogGroup{
+		Logs: []*Log{
+			{
+				Time: &n,
+				Contents: []*LogContent{
+					{
+						Key:   &key,
+						Value: &value,
+					},
+				},
+			},
+		},
+	}
+	project := s.projectName
+	logstore := s.logstoreName
+	err := s.client.PostLogStoreLogs(project, logstore, lg, nil)
+	s.Require().NoError(err)
+	time.Sleep(time.Second * 10)
+	resp, err := s.client.GetLogsToCompletedV3(project, logstore, &GetLogRequest{
+		From:  time.Now().Unix() - 30000,
+		To:    time.Now().Unix(),
+		Lines: 100,
+		Topic: "",
+	})
+	s.Require().NoError(err)
+	s.GreaterOrEqual(resp.Meta.Count, int64(1))
+}

log_store.go

@@ -752,6 +752,21 @@ func (s *LogStore) GetLogsToCompletedV2(req *GetLogRequest) (*GetLogsResponse, e
 	return res, err
 }
 
+// GetLogsToCompletedV3 query logs with [from, to) time range to completed
+func (s *LogStore) GetLogsToCompletedV3(req *GetLogRequest) (*GetLogsV3Response, error) {
+	var res *GetLogsV3Response
+	var err error
+	f := func() (bool, error) {
+		res, err = s.GetLogsV3(req)
+		if err == nil {
+			return res.IsComplete(), nil
+		}
+		return false, err
+	}
+	s.getToCompleted(f)
+	return res, err
+}
+
 // GetHistogramsToCompleted query logs with [from, to) time range to completed
 func (s *LogStore) GetHistogramsToCompleted(topic string, from int64, to int64, queryExp string) (*GetHistogramsResponse, error) {
 	var res *GetHistogramsResponse

model.go

@@ -207,6 +207,10 @@ type GetLogsV3Response struct {
 	Logs []map[string]string   `json:"data"`
 }
 
+func (resp *GetLogsV3Response) IsComplete() bool {
+	return strings.ToLower(resp.Meta.Progress) == "complete"
+}
+
 // GetLogLinesResponse defines response from GetLogLines call
 // note: GetLogLinesResponse.Logs is nil when use GetLogLinesResponse
 type GetLogLinesResponse struct {

token_auto_update_client.go

@@ -874,6 +874,16 @@ func (c *TokenAutoUpdateClient) GetLogsToCompletedV2(project, logstore string, r
 	return
 }
 
+func (c *TokenAutoUpdateClient) GetLogsToCompletedV3(project, logstore string, req *GetLogRequest) (r *GetLogsV3Response, err error) {
+	for i := 0; i < c.maxTryTimes; i++ {
+		r, err = c.logClient.GetLogsToCompletedV3(project, logstore, req)
+		if !c.processError(err) {
+			return
+		}
+	}
+	return
+}
+
 func (c *TokenAutoUpdateClient) GetLogLinesV2(project, logstore string, req *GetLogRequest) (r *GetLogLinesResponse, err error) {
 	for i := 0; i < c.maxTryTimes; i++ {
 		r, err = c.logClient.GetLogLinesV2(project, logstore, req)