Merge branch 'aliyun:master' into master

Author: neverafraid1

.github/workflows/go-ut.yml

@@ -19,7 +19,7 @@ jobs:
     - name: Set up Go
       uses: actions/setup-go@v3
       with:
-        go-version: 1.17
+        go-version: 1.19
 
     - name: Build
       run: go build -v ./...

.github/workflows/go.yml

@@ -16,4 +16,16 @@ jobs:
     - name: Set up Go
       uses: actions/setup-go@v2
       with:
-        go-version: 1.13
+        go-version: 1.19
+
+    - name: Build
+      run: go build -v ./
+
+    - name: Build consumer
+      run: go build -v ./consumer/...
+
+    - name: Build producer
+      run: go build -v ./producer/...
+
+    - name: Build util
+      run: go build -v ./util/...

README.md

@@ -38,6 +38,14 @@ go get -u github.com/aliyun/aliyun-log-go-sdk
    Client = sls.CreateNormalInterfaceV2(Endpoint, credentialsProvider)
    ```
 
+   为了防止出现配置错误，您可以在创建 Client 之后，测试 Client 是否能成功调用 SLS API
+   ```go
+   _, err := Client.ListProject()
+   if err != nil {
+      panic(err)
+   }
+   ```
+
 2. **创建project**
 
    参考 [log_project.go](https://github.com/aliyun/aliyun-log-go-sdk/blob/master/example/project/log_project.go)文件
@@ -97,9 +105,7 @@ go get -u github.com/aliyun/aliyun-log-go-sdk
 
 5. **写数据**
 
-   参考[put_log.go](https://github.com/aliyun/aliyun-log-go-sdk/blob/master/example/loghub/put_log.go)
-
-   这里展示了用sdk中原生的API接口去发送数据简单示例，但是我们不推荐用API直接向logstore写入数据，推荐使用SDK 中提供的[producer](https://github.com/aliyun/aliyun-log-go-sdk/tree/master/producer) 包向logstore 写入数据，自动压缩数据并且提供安全退出机制，不会使数据丢失。
+   这里展示了用sdk中原生的API接口去发送数据简单示例，但是我们不推荐用API直接向logstore写入数据，推荐使用SDK 中提供的[producer](https://github.com/aliyun/aliyun-log-go-sdk/tree/master/producer) 包向logstore 写入数据，自动压缩数据并且提供安全退出机制，不会使数据丢失, 对于MetricStore类型, 使用`PutLogsWithMetricStoreURL`API 发送数据可以提升大基数时间线下查询性能 。
 
    ```go
    logs := []*sls.Log{}
@@ -143,8 +149,6 @@ go get -u github.com/aliyun/aliyun-log-go-sdk
 
 6.**读数据**
 
-参考[pull_log.go](https://github.com/aliyun/aliyun-log-go-sdk/blob/master/example/loghub/pull_log.go)
-
 这里展示了使用SDK中原生API接口调用去拉取数据的方式，我们不推荐使用这种方式去读取消费logstore中的数据，推荐使用SDK中 [consumer](https://github.com/aliyun/aliyun-log-go-sdk/tree/master/consumer) 消费组去拉取数据，消费组提供自动负载均衡以及失败重试等机制，并且会自动保存拉取断点，再次拉取不会拉取重复数据。
 
 ```go

cgo/README.md

@@ -0,0 +1,77 @@
+## 依赖
+使用此扩展，需要设置 go env 中的 CGO_ENABLED=1，且环境中已安装了合适的编译器，linux 上通常是 gcc。  
+可以通过以下命令查看当前 CGO_ENABLED 是否打开。
+
+```bash
+go env | grep CGO_ENABLED
+```
+查看默认的编译器。
+```bash
+go env | grep CC
+```
+
+如果 CGO_ENABLED 值是 1，则可跳过下面开启 CGO_ENABLED 的步骤。  
+
+### 全局永久开启
+```bash 
+go env -w CGO_ENABLED=1
+```
+
+### 临时开启
+```bash
+CGO_ENABLED=1 go build
+```
+
+## 使用方法
+开启 cgo-zstd 扩展
+
+```golang
+import (
+    cgo "github.com/aliyun/aliyun-log-go-sdk/cgo"
+    sls "github.com/aliyun/aliyun-log-go-sdk"
+)
+cgo.SetZstdCgoCompressor(1)
+```
+
+
+使用 zstd 压缩写入日志的示例
+```golang
+import (
+	"time"
+
+	cgo "github.com/aliyun/aliyun-log-go-sdk/cgo"
+	sls "github.com/aliyun/aliyun-log-go-sdk"
+	"github.com/golang/protobuf/proto"
+)
+
+func main() {
+	cgo.SetZstdCgoCompressor(1)
+	client := sls.CreateNormalInterface("endpoint",
+		"accessKeyId", "accessKeySecret", "")
+	lg := &sls.LogGroup{
+		Logs: []*sls.Log{
+			{
+				Time: proto.Uint32(uint32(time.Now().Unix())),
+				Contents: []*sls.LogContent{
+					{
+						Key:   proto.String("HELLO"),
+						Value: proto.String("world"),
+					},
+				},
+			},
+		},
+	}
+	err := client.PostLogStoreLogsV2(
+		"your-project",
+		"your-logstore",
+		&sls.PostLogStoreLogsRequest{
+			LogGroup:     lg,
+			CompressType: sls.Compress_ZSTD, // 指定压缩方式为 ZSTD
+		},
+	)
+	if err != nil {
+		panic(err)
+	}
+
+}
+```

cgo/compressor.go

@@ -0,0 +1,42 @@
+package cgo
+
+import (
+	"sync"
+
+	"github.com/DataDog/zstd"
+	sls "github.com/aliyun/aliyun-log-go-sdk"
+)
+
+func SetZstdCgoCompressor(compressLevel int) error {
+	sls.SetZstdCompressor(newZstdCompressor(compressLevel))
+	return nil
+}
+
+type zstdCompressor struct {
+	ctxPool sync.Pool
+	level   int
+}
+
+func newZstdCompressor(level int) *zstdCompressor {
+	res := &zstdCompressor{
+		level: level,
+	}
+	res.ctxPool = sync.Pool{
+		New: func() interface{} {
+			return zstd.NewCtx()
+		},
+	}
+	return res
+}
+
+func (c *zstdCompressor) Compress(src, dst []byte) ([]byte, error) {
+	zstdCtx := c.ctxPool.Get().(zstd.Ctx)
+	defer c.ctxPool.Put(zstdCtx)
+	return zstdCtx.CompressLevel(dst, src, c.level)
+}
+
+func (c *zstdCompressor) Decompress(src, dst []byte) ([]byte, error) {
+	zstdCtx := c.ctxPool.Get().(zstd.Ctx)
+	defer c.ctxPool.Put(zstdCtx)
+	return zstdCtx.Decompress(dst, src)
+}

client.go

@@ -8,8 +8,11 @@ import (
 	"net/http"
 	"net/url"
 	"strconv"
+	"strings"
 	"sync"
 	"time"
+
+	"github.com/aliyun/aliyun-log-go-sdk/util"
 )
 
 // GlobalForceUsingHTTP if GlobalForceUsingHTTP is true, then all request will use HTTP(ignore LogProject's UsingHTTP flag)
@@ -28,6 +31,7 @@ var MaxCompletedRetryLatency = 5 * time.Minute
 const (
 	Compress_LZ4  = iota // 0
 	Compress_None        // 1
+	Compress_ZSTD        // 2
 	Compress_Max         // max compress type(just for filter invalid compress type)
 )
 
@@ -165,6 +169,11 @@ func (c *Client) SetHTTPClient(client *http.Client) {
 	c.HTTPClient = client
 }
 
+// SetRetryTimeout set retry timeout
+func (c *Client) SetRetryTimeout(timeout time.Duration) {
+	c.RetryTimeOut = timeout
+}
+
 // SetAuthVersion set signature version that the client used
 func (c *Client) SetAuthVersion(version AuthVersionType) {
 	c.accessKeyLock.Lock()
@@ -395,3 +404,11 @@ func (c *Client) DeleteProject(name string) error {
 func (c *Client) Close() error {
 	return nil
 }
+
+func (c *Client) setSignV4IfInAcdr(endpoint string) {
+	region, err := util.ParseRegion(endpoint)
+	if err == nil && strings.Contains(region, "-acdr-ut-") {
+		c.AuthVersion = AuthV4
+		c.Region = region
+	}
+}

client_interface.go

@@ -3,6 +3,8 @@ package sls
 import (
 	"net/http"
 	"time"
+
+	"github.com/aliyun/aliyun-log-go-sdk/util"
 )
 
 // CreateNormalInterface create a normal client.
@@ -11,10 +13,10 @@ import (
 // If you keep using long-lived AccessKeyID and AccessKeySecret,
 // use the example code below.
 //
-//	  provider := NewStaticCredProvider(accessKeyID, accessKeySecret, securityToken)
+//	  provider := NewStaticCredentialsProvider(accessKeyID, accessKeySecret, securityToken)
 //		client := CreateNormalInterfaceV2(endpoint, provider)
 func CreateNormalInterface(endpoint, accessKeyID, accessKeySecret, securityToken string) ClientInterface {
-	return &Client{
+	client := &Client{
 		Endpoint:        endpoint,
 		AccessKeyID:     accessKeyID,
 		AccessKeySecret: accessKeySecret,
@@ -26,6 +28,8 @@ func CreateNormalInterface(endpoint, accessKeyID, accessKeySecret, securityToken
 			securityToken,
 		),
 	}
+	client.setSignV4IfInAcdr(endpoint)
+	return client
 }
 
 // CreateNormalInterfaceV2 create a normal client, with a CredentialsProvider.
@@ -35,13 +39,15 @@ func CreateNormalInterface(endpoint, accessKeyID, accessKeySecret, securityToken
 //
 // See [credentials_provider.go] for more details.
 func CreateNormalInterfaceV2(endpoint string, credentialsProvider CredentialsProvider) ClientInterface {
-	return &Client{
+	client := &Client{
 		Endpoint:            endpoint,
 		credentialsProvider: credentialsProvider,
 	}
+	client.setSignV4IfInAcdr(endpoint)
+	return client
 }
 
-type UpdateTokenFunction = func() (accessKeyID, accessKeySecret, securityToken string, expireTime time.Time, err error)
+type UpdateTokenFunction = util.UpdateTokenFunction
 
 // CreateTokenAutoUpdateClient create a TokenAutoUpdateClient,
 // this client will auto fetch security token and retry when operation is `Unauthorized`
@@ -79,6 +85,8 @@ type ClientInterface interface {
 	SetUserAgent(userAgent string)
 	// SetHTTPClient set a custom http client, all request will send to sls by this client
 	SetHTTPClient(client *http.Client)
+	// SetRetryTimeout set retry timeout, client will retry util retry timeout
+	SetRetryTimeout(timeout time.Duration)
 	// #################### Client Operations #####################
 	// ResetAccessKeyToken reset client's access key token
 	ResetAccessKeyToken(accessKeyID, accessKeySecret, securityToken string)
@@ -162,6 +170,20 @@ type ClientInterface interface {
 	// ListEventStore returns all eventStore names of project p.
 	ListEventStore(project string, offset, size int) ([]string, error)
 
+	// #################### StoreView Operations #####################
+	// CreateStoreView creates a new storeView.
+	CreateStoreView(project string, storeView *StoreView) error
+	// UpdateStoreView updates a storeView.
+	UpdateStoreView(project string, storeView *StoreView) error
+	// DeleteStoreView deletes a storeView.
+	DeleteStoreView(project string, storeViewName string) error
+	// GetStoreView returns storeView.
+	GetStoreView(project string, storeViewName string) (*StoreView, error)
+	// ListStoreViews returns all storeView names of a project.
+	ListStoreViews(project string, req *ListStoreViewsRequest) (*ListStoreViewsResponse, error)
+	// GetStoreViewIndex returns all index config of logstores in the storeView, only support storeType logstore.
+	GetStoreViewIndex(project string, storeViewName string) (*GetStoreViewIndexResponse, error)
+
 	// #################### Logtail Operations #####################
 	// ListMachineGroup returns machine group name list and the total number of machine groups.
 	// The offset starts from 0 and the size is the max number of machine groups could be returned.
@@ -220,7 +242,6 @@ type ClientInterface interface {
 	CreateEtlMeta(project string, etlMeta *EtlMeta) (err error)
 	UpdateEtlMeta(project string, etlMeta *EtlMeta) (err error)
 	DeleteEtlMeta(project string, etlMetaName, etlMetaKey string) (err error)
-	listEtlMeta(project string, etlMetaName, etlMetaKey, etlMetaTag string, offset, size int) (total int, count int, etlMeta []*EtlMeta, err error)
 	GetEtlMeta(project string, etlMetaName, etlMetaKey string) (etlMeta *EtlMeta, err error)
 	ListEtlMeta(project string, etlMetaName string, offset, size int) (total int, count int, etlMetaList []*EtlMeta, err error)
 	ListEtlMetaWithTag(project string, etlMetaName, etlMetaTag string, offset, size int) (total int, count int, etlMetaList []*EtlMeta, err error)
@@ -237,12 +258,14 @@ type ClientInterface interface {
 	MergeShards(project, logstore string, shardID int) (shards []*Shard, err error)
 
 	// #################### Log Operations #####################
+	PutLogsWithMetricStoreURL(project, logstore string, lg *LogGroup) (err error)
 	// PutLogs put logs into logstore.
 	// The callers should transform user logs into LogGroup.
 	PutLogs(project, logstore string, lg *LogGroup) (err error)
 	// PostLogStoreLogs put logs into Shard logstore by hashKey.
 	// The callers should transform user logs into LogGroup.
 	PostLogStoreLogs(project, logstore string, lg *LogGroup, hashKey *string) (err error)
+	PostLogStoreLogsV2(project, logstore string, req *PostLogStoreLogsRequest) (err error)
 	// PostRawLogWithCompressType put logs into logstore with specific compress type and hashKey.
 	PostRawLogWithCompressType(project, logstore string, rawLogData []byte, compressType int, hashKey *string) (err error)
 	// PutLogsWithCompressType put logs into logstore with specific compress type.
@@ -262,14 +285,18 @@ type ClientInterface interface {
 	// The nextCursor is the next curosr can be used to read logs at next time.
 	GetLogsBytes(project, logstore string, shardID int, cursor, endCursor string,
 		logGroupMaxCount int) (out []byte, nextCursor string, err error)
+	// Deprecated: Use GetLogsBytesWithQuery instead.
 	GetLogsBytesV2(plr *PullLogRequest) (out []byte, nextCursor string, err error)
+	GetLogsBytesWithQuery(plr *PullLogRequest) (out []byte, plm *PullLogMeta, err error)
 	// PullLogs gets logs from shard specified by shardId according cursor and endCursor.
 	// The logGroupMaxCount is the max number of logGroup could be returned.
 	// The nextCursor is the next cursor can be used to read logs at next time.
 	// @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.
 	PullLogsV2(plr *PullLogRequest) (gl *LogGroupList, nextCursor string, err 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)
 	// GetLogs query logs with [from, to) time range

client_job.go

@@ -16,6 +16,7 @@ const (
 	DataSourceKafka      DataSourceType = "Kafka"
 	DataSourceCMS        DataSourceType = "AliyunCloudMonitor"
 	DataSourceGeneral    DataSourceType = "General"
+	DataSourceS3         DataSourceType = "S3"
 
 	OSSDataFormatTypeLine          OSSDataFormatType = "Line"
 	OSSDataFormatTypeMultiline     OSSDataFormatType = "Multiline"
@@ -190,7 +191,23 @@ type (
 		TimePattern      string            `json:"timePattern"`
 		TimeFormat       string            `json:"timeFormat"`
 		TimeZone         string            `json:"timeZone"`
+		Communication    string            `json:"communication"`
+		NameResolutions  string            `json:"nameResolutions"`
 		AdditionalProps  map[string]string `json:"additionalProps"`
+		VpcId            string            `json:"vpcId"`
+	}
+
+	S3Source struct {
+		DataSource
+		AWSAccessKey       string      `json:"awsAccessKey"`
+		AWSAccessKeySecret string      `json:"awsAccessKeySecret"`
+		AWSRegion          string      `json:"awsRegion"`
+		Bucket             string      `json:"bucket"`
+		Prefix             string      `json:"prefix,omitempty"`
+		Pattern            string      `json:"pattern,omitempty"`
+		CompressionCodec   string      `json:"compressionCodec,omitempty"`
+		Encoding           string      `json:"encoding,omitempty"`
+		Format             interface{} `json:"format,omitempty"`
 	}
 
 	// ingestion JDBC source