tencentcloud_cls_sdk_go

package module
v1.0.14-consumer Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 20, 2026 License: Apache-2.0 Imports: 34 Imported by: 48

README

Tencent CLS Log SDK

一、SDK 功能说明

tencentcloud-cls-sdk-go 是腾讯云日志服务（CLS）官方 Go SDK，提供日志上传与日志消费两大能力：

  • 日志上传：异步 / 同步生产者，支持攒批聚合、失败重试、回调通知、多种压缩（lz4 / zstd / deflate），开箱即用。
  • 日志消费：基于消费组（Consumer Group）的多分区并发消费、自动 offset 持久化与断点续传，业务侧只需实现一个 Processor 接口即可。

通用特性：

  • 支持 AccessKeyID + AccessKeySecret 永久密钥与 AccessToken 临时密钥，并提供运行时刷新。
  • 支持按地域自动拼接 endpoint。
  • 失败重试：429 / 5xx 等可恢复错误自动指数退避；4xx 类客户端错误直接失败。
  • 高性能：充分利用 Go 协程并发能力，单实例可支撑高吞吐上报。
go get github.com/tencentcloud/tencentcloud-cls-sdk-go

二、上传 SDK 说明

USAGE

go get github.com/tencentcloud/tencentcloud-cls-sdk-go

为什么要使用CLS Log SDK

  • 异步发送：发送日志立即返回，无须等待，支持传入callback function。
  • 优雅关闭：通过调用close方法，producer会将所有其缓存的数据进行发送，防止日志丢失。
  • 感知每一条日志的成功状态： 用户可以自定义CallBack方法的实现，监控每一条日志的状态
  • 使用简单： 通过简单配置，就可以实现复杂的日志上传聚合、失败重试等逻辑
  • 失败重试： 429、500 等服务端错误，都会进行重试
  • 高性能： 得益于go语言的高并发能力

CLS Host

endpoint填写请参考可用地域中 API上传日志 Tab中的域名（也可以选择地域与网络环境类型自动生成。如：Guangzhou，Extranet）image-20230403191435319

密钥信息

AccessKeyID和AccessKeySecret为云API密钥，密钥信息获取请前往密钥获取。并请确保密钥关联的账号具有相应的SDK上传日志权限

Demo

package main

import (
	"fmt"
	"github.com/tencentcloud/tencentcloud-cls-sdk-go"
	"sync"
	"time"
)

func main() {
	producerConfig := tencentcloud_cls_sdk_go.GetDefaultAsyncProducerClientConfig()
	producerConfig.Endpoint = "ap-guangzhou.cls.tencentcs.com"
	producerConfig.AccessKeyID = ""
	producerConfig.AccessKeySecret = ""
	topicId := ""
	producerInstance, err := tencentcloud_cls_sdk_go.NewAsyncProducerClient(producerConfig)
	if err != nil {
		fmt.Println(err)
		return
	}

        // 异步发送程序，需要启动
	producerInstance.Start()
	
	var m sync.WaitGroup
	callBack := &Callback{}
	for i := 0; i < 10; i++ {
		m.Add(1)
		go func() {
			defer m.Done()
			for i := 0; i < 1000; i++ {
				log := tencentcloud_cls_sdk_go.NewCLSLog(time.Now().Unix(), map[string]string{"content": "hello world| I'm from Beijing", "content2": fmt.Sprintf("%v", i)})
				err = producerInstance.SendLog(topicId, log, callBack)
				if err != nil {
					fmt.Println(err)
					continue
				}
			}
		}()
	}
	m.Wait()
	producerInstance.Close(60000)
}

type Callback struct {
}

func (callback *Callback) Success(result *tencentcloud_cls_sdk_go.Result) {
	attemptList := result.GetReservedAttempts()
	for _, attempt := range attemptList {
		fmt.Printf("%+v \n", attempt)
	}
}

func (callback *Callback) Fail(result *tencentcloud_cls_sdk_go.Result) {
	fmt.Println(result.IsSuccessful())
	fmt.Println(result.GetErrorCode())
	fmt.Println(result.GetErrorMessage())
	fmt.Println(result.GetReservedAttempts())
	fmt.Println(result.GetRequestId())
	fmt.Println(result.GetTimeStampMs())
}

配置参数详解

参数 类型 描述
TotalSizeLnBytes Int64 实例能缓存的日志大小上限，默认为 100MB。
MaxSendWorkerCount Int64 client能并发的最多"goroutine"的数量，默认为50
MaxBlockSec Int 如果client可用空间不足，调用者在 send 方法上的最大阻塞时间，默认为 60 秒。
如果超过这个时间后所需空间仍无法得到满足，send 方法会抛出TimeoutException。如果将该值设为0，当所需空间无法得到满足时，send 方法会立即抛出 TimeoutException。如果您希望 send 方法一直阻塞直到所需空间得到满足，可将该值设为负数。
MaxBatchSize Int64 当一个Batch中缓存的日志大小大于等于 batchSizeThresholdInBytes 时，该 batch 将被发送，默认为 512 KB，最大可设置成 5MB。
MaxBatchCount Int 当一个Batch中缓存的日志条数大于等于 batchCountThreshold 时，该 batch 将被发送，默认为 4096，最大可设置成 40960。
LingerMs Int64 Batch从创建到可发送的逗留时间，默认为 2 秒，最小可设置成 100 毫秒。
Retries Int 如果某个Batch首次发送失败，能够对其重试的次数，默认为 10 次。
如果 retries 小于等于 0，该 ProducerBatch 首次发送失败后将直接进入失败队列。
MaxReservedAttempts Int 每个Batch每次被尝试发送都对应着一个Attemp，此参数用来控制返回给用户的 attempt 个数，默认只保留最近的 11 次 attempt 信息。
该参数越大能让您追溯更多的信息，但同时也会消耗更多的内存。
BaseRetryBackoffMs Int64 首次重试的退避时间，默认为 100 毫秒。 client采样指数退避算法，第 N 次重试的计划等待时间为 baseRetryBackoffMs * 2^(N-1)。
MaxRetryBackoffMs Int64 重试的最大退避时间，默认为 50 秒。

generate cls log

protoc --gofast_out=. cls.proto

三、消费 SDK 说明

本 SDK 内置了基于「消费组（Consumer Group）」的高阶日志消费模块 consumer，主要能力：

  • 消费组管理：不存在自动创建、已存在自动复用。
  • 心跳与分区分配：服务端按 PartitionStrategy=2 做动态 rebalance，多实例横向扩容自动均衡分区。
  • 多分区并发消费：每个分区独立 goroutine 拉取日志并串行调用业务 Processor。
  • Offset 自动持久化：拉取自动推进 + 60s 周期兜底 + 空闲 30s flush + 退出强制 flush，断点续传无需关心。
  • 服务端 DSL 预过滤：通过 ConsumerOption.Query 传入 DSL 表达式，命中日志才下发，节省带宽与客户端 CPU。例如 log_keep(op_and(op_gt(v("status"), 400), str_exist(v("message"), "access failed"))) 仅消费 status>400 且 message 含 access failed 的日志。详见 日志消费 DSL 过滤语法。
  • 智能停止：配置 OffsetEndTime 后所有分区追上末尾即自动退出整个 worker。
  • 优雅退出与失败自愈：InvalidOffset 自动取最新 offset、心跳超时自动重新分配分区、Process panic 不阻塞消费。

最简使用方式：实现 Processor 接口 → 构造 ConsumerOption → consumer.NewConsumerWorker(option, processor).Run(ctx) 即可。

详细使用说明、配置参数详解、并发模型、错误处理、FAQ 以及可运行示例（consumer/demo/consumer_demo.go）请参见 consumer/README.md。

Documentation

Index

Constants

View Source 
const BAD_REQUEST = "BadRequest"
View Source 
const CompressionLevelDefault = -1000
View Source 
const INTERNAL_SERVER_ERROR = "InternalError"
View Source 
const MISSING_HOST = "MissingHost"
View Source 
const MISS_ACCESS_KEY_ID = "MissAccessKeyId"
View Source 
const MISS_ACCESS_SECRET = "MissAccessKeyId"
View Source 
const MISS_ACCESS_TOKEN = "MissAccessKeyId"
View Source 
const UNKNOWN_ERROR = "UnknownError"
View Source 
const WRITE_QUOTA_EXCEED = "WriteQuotaExceed"

Variables

View Source 
var (
	ErrInvalidLengthCls        = fmt.Errorf("proto: negative length found during unmarshaling")
	ErrIntOverflowCls          = fmt.Errorf("proto: integer overflow")
	ErrUnexpectedEndOfGroupCls = fmt.Errorf("proto: unexpected end of group")
)

Functions

func DeflateCompress added in v1.0.13 

func DeflateCompress(data []byte) ([]byte, error)

DeflateCompress ...

func GenerateProducerHash added in v1.0.7 

func GenerateProducerHash(instanceID string) string

GenerateProducerHash ...

func GetLocalIP 

func GetLocalIP() (string, error)

GetLocalIP get local IP with string format

func GetLogListSize 

func GetLogListSize(logList []*Log) (int, error)

func GetLogSizeCalculate 

func GetLogSizeCalculate(log *Log) (int, error)

func GetTimeMs 

func GetTimeMs(t int64) int64

func NewCLSClient 

func NewCLSClient(options *Options) (*CLSClient, *CLSError)

func SetUserAgent added in v1.0.9 

func SetUserAgent(agent string)

SetUserAgent ...

func SetVersion added in v1.0.9 

func SetVersion(ver string)

SetVersion ...

func SignatureWithTC3 added in v1.0.15 

func SignatureWithTC3(secretID, secretKey, service, method, path string, queryString string, headers map[string]string, payload []byte, timestamp int64) (string, error)

SignatureWithTC3 generates TC3-HMAC-SHA256 signature

func ZSTDCompress added in v1.0.2 

func ZSTDCompress(params ZstdEncoderParams, dst, src []byte) ([]byte, error)

func ZSTDDecompress added in v1.0.2 

func ZSTDDecompress(params ZstdDecoderParams, dst, src []byte) ([]byte, error)

Types

type APIError added in v1.0.15 

type APIError struct {
	Code    string `json:"Code"`
	Message string `json:"Message"`
}

type Accumulator 

type Accumulator struct {
	// contains filtered or unexported fields
}

func NewAccumulator 

func NewAccumulator(config *AsyncProducerClientConfig, worker *Worker, threadPool *SendThreadPool, producer *AsyncProducerClient) *Accumulator

NewAccumulator ...

type AsyncProducerClient 

type AsyncProducerClient struct {
	Client *CLSClient
	// contains filtered or unexported fields
}

AsyncProducerClient async producer client

func NewAsyncProducerClient 

func NewAsyncProducerClient(asyncProducerClientConfig *AsyncProducerClientConfig) (*AsyncProducerClient, error)

NewAsyncProducerClient 初始化Async Producer Client

func (*AsyncProducerClient) Close 

func (producer *AsyncProducerClient) Close(timeoutMs int64) error

func (*AsyncProducerClient) SendLog 

func (producer *AsyncProducerClient) SendLog(topicId string, log *Log, callback CallBack) error

func (*AsyncProducerClient) SendLogList 

func (producer *AsyncProducerClient) SendLogList(topicId string, logList []*Log, callback CallBack) (err error)

func (*AsyncProducerClient) Start 

func (producer *AsyncProducerClient) Start()

type AsyncProducerClientConfig 

type AsyncProducerClientConfig struct {
	TotalSizeLnBytes    int64
	MaxSendWorkerCount  int64
	MaxBlockSec         int
	MaxBatchSize        int64
	MaxBatchCount       int
	LingerMs            int64
	Retries             int
	MaxReservedAttempts int
	BaseRetryBackoffMs  int64
	MaxRetryBackoffMs   int64
	Endpoint            string

	AccessKeyID     string
	AccessKeySecret string
	AccessToken     string
	Source          string
	Timeout         int
	IdleConn        int
	CompressType    string
	HostName        string
	// contains filtered or unexported fields
}

AsyncProducerClientConfig Producer Config

func GetDefaultAsyncProducerClientConfig 

func GetDefaultAsyncProducerClientConfig() *AsyncProducerClientConfig

GetDefaultAsyncProducerClientConfig 获取默认的Producer Config

func (*AsyncProducerClientConfig) SetEndpointByRegionAndNetworkType added in v1.0.12 

func (config *AsyncProducerClientConfig) SetEndpointByRegionAndNetworkType(region Region, networkType NetworkType)

SetEndpointByRegionAndNetworkType 根据地域和网络类型设置域名

type Attempt 

type Attempt struct {
	Success      bool
	RequestId    string
	ErrorCode    string
	ErrorMessage string
	TimeStampMs  int64
}

func NewAttempt 

func NewAttempt(success bool, requestId, errorCode, errorMessage string, timeStampMs int64) *Attempt

NewAttempt init attempt

type CLSClient 

type CLSClient struct {
	// contains filtered or unexported fields
}

func (*CLSClient) ResetSecretToken 

func (client *CLSClient) ResetSecretToken(secretID string, secretKEY string, secretToken string) *CLSError

func (*CLSClient) Send 

func (client *CLSClient) Send(ctx context.Context, topicId string, group ...*LogGroup) *CLSError

Send cls实际发送接口

type CLSError 

type CLSError struct {
	HTTPCode  int32  `json:"httpCode"`
	Code      string `json:"errorCode"`
	Message   string `json:"errorMessage"`
	RequestID string `json:"requestID"`
}

Error defines sls error

func NewError 

func NewError(httpCode int32, requestID, errorCode string, err error) *CLSError

NewClientError new client error

func (CLSError) Error 

func (e CLSError) Error() string

func (CLSError) String 

func (e CLSError) String() string

type CallBack 

type CallBack interface {
	Success(result *Result)
	Fail(result *Result)
}

type ConsumerGroupEntity added in v1.0.15 

type ConsumerGroupEntity struct {
	ConsumerGroup string   `json:"ConsumerGroup"`
	Timeout       int      `json:"Timeout"`
	Topics        []string `json:"Topics"`
}

type ConsumerGroupGetOffsetsResponse added in v1.0.15 

type ConsumerGroupGetOffsetsResponse struct {
	Response struct {
		ConsumerGroup             string                      `json:"ConsumerGroup"`
		TopicPartitionOffsetsInfo []TopicPartitionOffsetsInfo `json:"TopicPartitionOffsetsInfo"`
		RequestId                 string                      `json:"RequestId"`
		Error                     *APIError                   `json:"Error,omitempty"`
	} `json:"Response"`
}

ConsumerGroupGetOffsetsResponse get offset response

type ConsumerGroupHeartBeatResponse added in v1.0.15 

type ConsumerGroupHeartBeatResponse struct {
	Response struct {
		ConsumerGroup       string          `json:"ConsumerGroup,omitempty"`
		TopicPartitionsInfo []PartitionInfo `json:"TopicPartitionsInfo,omitempty"`
		RequestId           string          `json:"RequestId"`
		Error               *APIError       `json:"Error,omitempty"`
	} `json:"Response"`
}

ConsumerGroupHeartBeatResponse heartbeat response

type ConsumerGroupUpdateOffsetsResponse added in v1.0.15 

type ConsumerGroupUpdateOffsetsResponse struct {
	Response struct {
		RequestId string    `json:"RequestId"`
		Error     *APIError `json:"Error,omitempty"`
	} `json:"Response"`
}

ConsumerGroupUpdateOffsetsResponse update offset response

type CreateConsumerGroupRequest added in v1.0.15 

type CreateConsumerGroupRequest struct {
	ConsumerGroup string   `json:"ConsumerGroup"`
	Timeout       int      `json:"Timeout"`
	Topics        []string `json:"Topics"`
	LogsetId      string   `json:"LogsetId"`
}

CreateConsumerGroupRequest create consumer group request

type CreateConsumerGroupResponse added in v1.0.15 

type CreateConsumerGroupResponse struct {
	Response struct {
		RequestId     string    `json:"RequestId"`
		Error         *APIError `json:"Error,omitempty"`
		ConsumerGroup string    `json:"ConsumerGroup,omitempty"`
	} `json:"Response"`
}

type Credentials added in v1.0.6 

type Credentials struct {
	SecretID    string
	SecretKEY   string
	SecretToken string
}

type DeleteConsumerGroupRequest added in v1.0.15 

type DeleteConsumerGroupRequest struct {
	LogsetId      string `json:"LogsetId"`
	ConsumerGroup string `json:"ConsumerGroup"`
}

DeleteConsumerGroupRequest delete consumer group request

type DeleteConsumerGroupResponse added in v1.0.15 

type DeleteConsumerGroupResponse struct {
	Response struct {
		RequestId string    `json:"RequestId"`
		Error     *APIError `json:"Error,omitempty"`
	} `json:"Response"`
}

DeleteConsumerGroupResponse supports capturing all fields

type ErrorMessage 

type ErrorMessage struct {
	Code    string `json:"errorcode"`
	Message string `json:"errormessage"`
}

type Field added in v1.0.15 

type Field struct {
	Key   string
	Value interface{}
}

type GetOffsetsRequest added in v1.0.15 

type GetOffsetsRequest struct {
	ConsumerGroup string `json:"ConsumerGroup"`
	TopicId       string `json:"TopicId"`
	PartitionId   string `json:"PartitionId,omitempty"`
	LogsetId      string `json:"LogsetId"`
	From          string `json:"From,omitempty"`
}

GetOffsetsRequest get offset request body

type HeartbeatRequest added in v1.0.15 

type HeartbeatRequest struct {
	ConsumerGroup       string          `json:"ConsumerGroup"`
	Consumer            string          `json:"Consumer"`
	LogsetId            string          `json:"LogsetId"`
	TopicPartitionsInfo []PartitionInfo `json:"TopicPartitionsInfo"`
	PartitionStrategy   int             `json:"PartitionStrategy"`
}

HeartbeatRequest heartbeat request body

type ListConsumerGroupRequest added in v1.0.15 

type ListConsumerGroupRequest struct {
	LogsetId string   `json:"LogsetId"`
	Topics   []string `json:"Topics"`
}

ListConsumerGroupRequest list consumer group request

type ListConsumerGroupResponse added in v1.0.15 

type ListConsumerGroupResponse struct {
	Response struct {
		ConsumerGroupsInfo []ConsumerGroupEntity `json:"ConsumerGroupsInfo"`
		RequestId          string                `json:"RequestId"`
		Error              *APIError             `json:"Error,omitempty"`
	} `json:"Response"`
}

type Log 

type Log struct {
	Time                 *int64         `protobuf:"varint,1,req,name=time" json:"time,omitempty"`
	Contents             []*Log_Content `protobuf:"bytes,2,rep,name=contents" json:"contents,omitempty"`
	CollectTime          *int64         `protobuf:"varint,3,opt,name=collectTime" json:"collectTime,omitempty"`
	XXX_NoUnkeyedLiteral struct{}       `json:"-"`
	XXX_unrecognized     []byte         `json:"-"`
	XXX_sizecache        int32          `json:"-"`
}

func NewCLSLog 

func NewCLSLog(logTime int64, addLogMap map[string]string) *Log

NewCLSLog ...

func (*Log) Descriptor 

func (*Log) Descriptor() ([]byte, []int)

func (*Log) GetCollectTime added in v1.0.8 

func (m *Log) GetCollectTime() int64

func (*Log) GetContents 

func (m *Log) GetContents() []*Log_Content

func (*Log) GetTime 

func (m *Log) GetTime() int64

func (*Log) Marshal 

func (m *Log) Marshal() (dAtA []byte, err error)

func (*Log) MarshalTo 

func (m *Log) MarshalTo(dAtA []byte) (int, error)

func (*Log) MarshalToSizedBuffer 

func (m *Log) MarshalToSizedBuffer(dAtA []byte) (int, error)

func (*Log) ProtoMessage 

func (*Log) ProtoMessage()

func (*Log) Reset 

func (m *Log) Reset()

func (*Log) Size 

func (m *Log) Size() (n int)

func (*Log) String 

func (m *Log) String() string

func (*Log) Unmarshal 

func (m *Log) Unmarshal(dAtA []byte) error

func (*Log) XXX_DiscardUnknown 

func (m *Log) XXX_DiscardUnknown()

func (*Log) XXX_Marshal 

func (m *Log) XXX_Marshal(b []byte, deterministic bool) ([]byte, error)

func (*Log) XXX_Merge 

func (m *Log) XXX_Merge(src proto.Message)

func (*Log) XXX_Size 

func (m *Log) XXX_Size() int

func (*Log) XXX_Unmarshal 

func (m *Log) XXX_Unmarshal(b []byte) error

type LogGroup 

type LogGroup struct {
	Logs                 []*Log    `protobuf:"bytes,1,rep,name=logs" json:"logs,omitempty"`
	ContextFlow          *string   `protobuf:"bytes,2,opt,name=contextFlow" json:"contextFlow,omitempty"`
	Filename             *string   `protobuf:"bytes,3,opt,name=filename" json:"filename,omitempty"`
	Source               *string   `protobuf:"bytes,4,opt,name=source" json:"source,omitempty"`
	LogTags              []*LogTag `protobuf:"bytes,5,rep,name=logTags" json:"logTags,omitempty"`
	Hostname             *string   `protobuf:"bytes,6,opt,name=hostname" json:"hostname,omitempty"`
	XXX_NoUnkeyedLiteral struct{}  `json:"-"`
	XXX_unrecognized     []byte    `json:"-"`
	XXX_sizecache        int32     `json:"-"`
}

func (*LogGroup) Descriptor 

func (*LogGroup) Descriptor() ([]byte, []int)

func (*LogGroup) GetContextFlow 

func (m *LogGroup) GetContextFlow() string

func (*LogGroup) GetFilename 

func (m *LogGroup) GetFilename() string

func (*LogGroup) GetHostname added in v1.0.8 

func (m *LogGroup) GetHostname() string

func (*LogGroup) GetLogTags 

func (m *LogGroup) GetLogTags() []*LogTag

func (*LogGroup) GetLogs 

func (m *LogGroup) GetLogs() []*Log

func (*LogGroup) GetSource 

func (m *LogGroup) GetSource() string

func (*LogGroup) Marshal 

func (m *LogGroup) Marshal() (dAtA []byte, err error)

func (*LogGroup) MarshalTo 

func (m *LogGroup) MarshalTo(dAtA []byte) (int, error)

func (*LogGroup) MarshalToSizedBuffer 

func (m *LogGroup) MarshalToSizedBuffer(dAtA []byte) (int, error)

func (*LogGroup) ProtoMessage 

func (*LogGroup) ProtoMessage()

func (*LogGroup) Reset 

func (m *LogGroup) Reset()

func (*LogGroup) Size 

func (m *LogGroup) Size() (n int)

func (*LogGroup) String 

func (m *LogGroup) String() string

func (*LogGroup) Unmarshal 

func (m *LogGroup) Unmarshal(dAtA []byte) error

func (*LogGroup) XXX_DiscardUnknown 

func (m *LogGroup) XXX_DiscardUnknown()

func (*LogGroup) XXX_Marshal 

func (m *LogGroup) XXX_Marshal(b []byte, deterministic bool) ([]byte, error)

func (*LogGroup) XXX_Merge 

func (m *LogGroup) XXX_Merge(src proto.Message)

func (*LogGroup) XXX_Size 

func (m *LogGroup) XXX_Size() int

func (*LogGroup) XXX_Unmarshal 

func (m *LogGroup) XXX_Unmarshal(b []byte) error

type LogGroupList 

type LogGroupList struct {
	LogGroupList         []*LogGroup `protobuf:"bytes,1,rep,name=logGroupList" json:"logGroupList,omitempty"`
	XXX_NoUnkeyedLiteral struct{}    `json:"-"`
	XXX_unrecognized     []byte      `json:"-"`
	XXX_sizecache        int32       `json:"-"`
}

func (*LogGroupList) Descriptor 

func (*LogGroupList) Descriptor() ([]byte, []int)

func (*LogGroupList) GetLogGroupList 

func (m *LogGroupList) GetLogGroupList() []*LogGroup

func (*LogGroupList) Marshal 

func (m *LogGroupList) Marshal() (dAtA []byte, err error)

func (*LogGroupList) MarshalTo 

func (m *LogGroupList) MarshalTo(dAtA []byte) (int, error)

func (*LogGroupList) MarshalToSizedBuffer 

func (m *LogGroupList) MarshalToSizedBuffer(dAtA []byte) (int, error)

func (*LogGroupList) ProtoMessage 

func (*LogGroupList) ProtoMessage()

func (*LogGroupList) Reset 

func (m *LogGroupList) Reset()

func (*LogGroupList) Size 

func (m *LogGroupList) Size() (n int)

func (*LogGroupList) String 

func (m *LogGroupList) String() string

func (*LogGroupList) Unmarshal 

func (m *LogGroupList) Unmarshal(dAtA []byte) error

func (*LogGroupList) XXX_DiscardUnknown 

func (m *LogGroupList) XXX_DiscardUnknown()

func (*LogGroupList) XXX_Marshal 

func (m *LogGroupList) XXX_Marshal(b []byte, deterministic bool) ([]byte, error)

func (*LogGroupList) XXX_Merge 

func (m *LogGroupList) XXX_Merge(src proto.Message)

func (*LogGroupList) XXX_Size 

func (m *LogGroupList) XXX_Size() int

func (*LogGroupList) XXX_Unmarshal 

func (m *LogGroupList) XXX_Unmarshal(b []byte) error

type LogSendRetryQueue 

type LogSendRetryQueue struct {
	// contains filtered or unexported fields
}

LogSendRetryQueue RetryQueue cache ProducerBatch and retry latter

func NewRetryQueue 

func NewRetryQueue() *LogSendRetryQueue

NewRetryQueue ...

func (*LogSendRetryQueue) Len 

func (retryQueue *LogSendRetryQueue) Len() int

func (*LogSendRetryQueue) Less 

func (retryQueue *LogSendRetryQueue) Less(i, j int) bool

func (*LogSendRetryQueue) Pop 

func (retryQueue *LogSendRetryQueue) Pop() interface{}

func (*LogSendRetryQueue) Push 

func (retryQueue *LogSendRetryQueue) Push(x interface{})

func (*LogSendRetryQueue) Swap 

func (retryQueue *LogSendRetryQueue) Swap(i, j int)

type LogTag 

type LogTag struct {
	Key                  *string  `protobuf:"bytes,1,req,name=key" json:"key,omitempty"`
	Value                *string  `protobuf:"bytes,2,req,name=value" json:"value,omitempty"`
	XXX_NoUnkeyedLiteral struct{} `json:"-"`
	XXX_unrecognized     []byte   `json:"-"`
	XXX_sizecache        int32    `json:"-"`
}

func (*LogTag) Descriptor 

func (*LogTag) Descriptor() ([]byte, []int)

func (*LogTag) GetKey 

func (m *LogTag) GetKey() string

func (*LogTag) GetValue 

func (m *LogTag) GetValue() string

func (*LogTag) Marshal 

func (m *LogTag) Marshal() (dAtA []byte, err error)

func (*LogTag) MarshalTo 

func (m *LogTag) MarshalTo(dAtA []byte) (int, error)

func (*LogTag) MarshalToSizedBuffer 

func (m *LogTag) MarshalToSizedBuffer(dAtA []byte) (int, error)

func (*LogTag) ProtoMessage 

func (*LogTag) ProtoMessage()

func (*LogTag) Reset 

func (m *LogTag) Reset()

func (*LogTag) Size 

func (m *LogTag) Size() (n int)

func (*LogTag) String 

func (m *LogTag) String() string

func (*LogTag) Unmarshal 

func (m *LogTag) Unmarshal(dAtA []byte) error

func (*LogTag) XXX_DiscardUnknown 

func (m *LogTag) XXX_DiscardUnknown()

func (*LogTag) XXX_Marshal 

func (m *LogTag) XXX_Marshal(b []byte, deterministic bool) ([]byte, error)

func (*LogTag) XXX_Merge 

func (m *LogTag) XXX_Merge(src proto.Message)

func (*LogTag) XXX_Size 

func (m *LogTag) XXX_Size() int

func (*LogTag) XXX_Unmarshal 

func (m *LogTag) XXX_Unmarshal(b []byte) error

type Log_Content 

type Log_Content struct {
	Key                  *string  `protobuf:"bytes,1,req,name=key" json:"key,omitempty"`
	Value                *string  `protobuf:"bytes,2,req,name=value" json:"value,omitempty"`
	XXX_NoUnkeyedLiteral struct{} `json:"-"`
	XXX_unrecognized     []byte   `json:"-"`
	XXX_sizecache        int32    `json:"-"`
}

func (*Log_Content) Descriptor 

func (*Log_Content) Descriptor() ([]byte, []int)

func (*Log_Content) GetKey 

func (m *Log_Content) GetKey() string

func (*Log_Content) GetValue 

func (m *Log_Content) GetValue() string

func (*Log_Content) Marshal 

func (m *Log_Content) Marshal() (dAtA []byte, err error)

func (*Log_Content) MarshalTo 

func (m *Log_Content) MarshalTo(dAtA []byte) (int, error)

func (*Log_Content) MarshalToSizedBuffer 

func (m *Log_Content) MarshalToSizedBuffer(dAtA []byte) (int, error)

func (*Log_Content) ProtoMessage 

func (*Log_Content) ProtoMessage()

func (*Log_Content) Reset 

func (m *Log_Content) Reset()

func (*Log_Content) Size 

func (m *Log_Content) Size() (n int)

func (*Log_Content) String 

func (m *Log_Content) String() string

func (*Log_Content) Unmarshal 

func (m *Log_Content) Unmarshal(dAtA []byte) error

func (*Log_Content) XXX_DiscardUnknown 

func (m *Log_Content) XXX_DiscardUnknown()

func (*Log_Content) XXX_Marshal 

func (m *Log_Content) XXX_Marshal(b []byte, deterministic bool) ([]byte, error)

func (*Log_Content) XXX_Merge 

func (m *Log_Content) XXX_Merge(src proto.Message)

func (*Log_Content) XXX_Size 

func (m *Log_Content) XXX_Size() int

func (*Log_Content) XXX_Unmarshal 

func (m *Log_Content) XXX_Unmarshal(b []byte) error

type Logger added in v1.0.15 

type Logger interface {
	Debug(msg string, args ...Field)
	Info(msg string, args ...Field)
	Warn(msg string, args ...Field)
	Error(msg string, args ...Field)
}

type NetworkType added in v1.0.12 

type NetworkType string

NetworkType 网络类型,可以将string赋值给该类型，实现自定义网络类型 

const (
	Intranet NetworkType = "cls.tencentyun.com" //内网
	Extranet NetworkType = "cls.tencentcs.com"  //外网
)

type Options 

type Options struct {
	Host         string
	Timeout      int
	IdleConn     int
	CompressType string
	Credentials  Credentials
}

type PartitionInfo added in v1.0.15 

type PartitionInfo struct {
	TopicID    string `json:"TopicID"`
	Partitions []int  `json:"Partitions"`
}

type PartitionOffset added in v1.0.15 

type PartitionOffset struct {
	PartitionID int   `json:"PartitionId"`
	Offset      int64 `json:"Offset"`
}

PartitionOffset partition offset

type ProducerBatch 

type ProducerBatch struct {
	// contains filtered or unexported fields
}

func NewProducerBatch 

func NewProducerBatch(topicID string, config *AsyncProducerClientConfig, callBackFunc CallBack, logData interface{}, packageID string) *ProducerBatch

NewProducerBatch 初始化Producer batch

type PullLogResponse added in v1.0.15 

type PullLogResponse struct {
	NextOffset int64
	LogGroups  []*LogGroup
	Headers    map[string]string
	// contains filtered or unexported fields
}

func ParsePullLogResponse added in v1.0.15 

func ParsePullLogResponse(respBytes []byte) (*PullLogResponse, error)

ParsePullLogResponse parses the binary response of pulling logs and returns a PullLogResponse object

func (*PullLogResponse) GetLogCount added in v1.0.15 

func (r *PullLogResponse) GetLogCount() int

GetLogCount returns the number of all logs

func (*PullLogResponse) GetLogGroups added in v1.0.15 

func (r *PullLogResponse) GetLogGroups() []*LogGroup

GetLogGroups returns all log groups

func (*PullLogResponse) GetNextOffset added in v1.0.15 

func (r *PullLogResponse) GetNextOffset() int64

GetNextOffset returns the next offset

type PullLogsClient added in v1.0.15 

type PullLogsClient struct {
	Endpoint  string
	SecretId  string
	SecretKey string
	Token     string
	UserAgent string
}

func NewPullLogsClient added in v1.0.15 

func NewPullLogsClient(endpoint, secretId, secretKey, token, userAgent string) *PullLogsClient

func (*PullLogsClient) PullLogs added in v1.0.15 

func (c *PullLogsClient) PullLogs(topicId string, partitionId int, size int, startTime *int64, offset int64, endTime *int64, query string) ([]byte, error)

PullLogs pulls logs and returns snappy compressed binary data. query is an optional DSL pre-filter expression; pass "" to disable server-side filtering.

func (*PullLogsClient) PullLogsAndParse added in v1.0.15 

func (c *PullLogsClient) PullLogsAndParse(topicId string, partitionId int, size int, startTime *int64, offset int64, endTime *int64, query string) (*PullLogResponse, error)

PullLogsAndParse pulls and parses logs and outputs proto text format. query is an optional DSL pre-filter expression; pass "" to disable server-side filtering.

type PullLogsRequest added in v1.0.15 

type PullLogsRequest struct {
	StartOffset  int64  `json:"StartOffset"`
	Size         int    `json:"Size"`
	CompressType string `json:"CompressType"`
	PartitionId  int    `json:"PartitionId"`
	StartTime    int64  `json:"StartTime"`
	EndTime      *int64 `json:"EndTime,omitempty"`
	// Query is an optional DSL pre-filter expression applied on the server side
	// before logs are returned. Example:
	//   log_keep(op_and(op_gt(v("status"), 400), str_exist(v("message"), "access failed")))
	// See https://cloud.tencent.com/document/product/614/37908 for the full DSL syntax.
	Query string `json:"Query"`
}

type Region added in v1.0.12 

type Region string

Region 地域类型,可以将string赋值给该类型，实现自定义地域 

const (
	Beijing       Region = "ap-beijing"
	Guangzhou     Region = "ap-guangzhou"
	Shanghai      Region = "ap-shanghai"
	Chengdu       Region = "ap-chengdu"
	Nanjing       Region = "ap-nanjing"
	Chongqing     Region = "ap-chongqing"
	Hongkong      Region = "ap-hongkong"
	Siliconvalley Region = "na-siliconvalley"
	Ashburn       Region = "na-ashburn"
	Singapore     Region = "ap-singapore"
	Bangkok       Region = "ap-bangkok"
	Frankfurt     Region = "eu-frankfurt"
	Tokyo         Region = "ap-tokyo"
	Seoul         Region = "ap-seoul"
	Jakarta       Region = "ap-jakarta"
	Saopaulo      Region = "sa-saopaulo"
	ShenzhenFSI   Region = "ap-shenzhen-fsi"
	ShanghaiFSI   Region = "ap-shanghai-fsi"
	BeijingFSI    Region = "ap-beijing-fsi"
	ShanghaiADC   Region = "ap-shanghai-adc"
)

type Result 

type Result struct {
	// contains filtered or unexported fields
}

func NewResult 

func NewResult() *Result

NewResult init result

func (*Result) GetErrorCode 

func (result *Result) GetErrorCode() string

func (*Result) GetErrorMessage 

func (result *Result) GetErrorMessage() string

func (*Result) GetRequestId 

func (result *Result) GetRequestId() string

func (*Result) GetReservedAttempts 

func (result *Result) GetReservedAttempts() []*Attempt

func (*Result) GetTimeStampMs 

func (result *Result) GetTimeStampMs() int64

func (*Result) IsSuccessful 

func (result *Result) IsSuccessful() bool

type SendThreadPool 

type SendThreadPool struct {
	// contains filtered or unexported fields
}

func NewSendThreadPool 

func NewSendThreadPool(worker *Worker) *SendThreadPool

type SyncProducerClient added in v1.0.3 

type SyncProducerClient struct {
	// contains filtered or unexported fields
}

SyncProducerClient synchronized producer client

func NewSyncProducerClient added in v1.0.3 

func NewSyncProducerClient(config *SyncProducerClientConfig) (*SyncProducerClient, error)

NewSyncProducerClient new sync producer client

func (*SyncProducerClient) ResetSecretToken added in v1.0.3 

func (c *SyncProducerClient) ResetSecretToken(secretID string, secretKEY string, secretToken string) error

ResetSecretToken reset temporary secret info

func (*SyncProducerClient) SendLogGroupList added in v1.0.6 

func (c *SyncProducerClient) SendLogGroupList(ctx context.Context, topicID string, logGroupList []*LogGroup) error

SendLogGroupList send batched logs to cls

func (*SyncProducerClient) SendLogList added in v1.0.3 

func (c *SyncProducerClient) SendLogList(ctx context.Context, topicID string, logList []*Log) error

SendLogList send batched logs to cls

type SyncProducerClientConfig added in v1.0.3 

type SyncProducerClientConfig struct {
	Endpoint string

	AccessKeyID     string
	AccessKeySecret string
	AccessToken     string
	Timeout         int
	IdleConn        int
	CompressType    string
	NeedSource      bool
	HostName        string
	// contains filtered or unexported fields
}

SyncProducerClientConfig sync producer config

func GetDefaultSyncProducerClientConfig added in v1.0.3 

func GetDefaultSyncProducerClientConfig() *SyncProducerClientConfig

GetDefaultSyncProducerClientConfig get default sync producer config

func (*SyncProducerClientConfig) SetEndpointByRegionAndNetworkType added in v1.0.12 

func (config *SyncProducerClientConfig) SetEndpointByRegionAndNetworkType(region Region, networkType NetworkType)

SetEndpointByRegionAndNetworkType 根据地域和网络类型设置域名

type TimerSendLogTask 

type TimerSendLogTask struct {
	// contains filtered or unexported fields
}

func NewTimerSendLogTask 

func NewTimerSendLogTask(logAccumulator *Accumulator, retryQueue *LogSendRetryQueue, ioWorker *Worker, threadPool *SendThreadPool) *TimerSendLogTask

type TopicPartitionOffsetsInfo added in v1.0.15 

type TopicPartitionOffsetsInfo struct {
	TopicID          string            `json:"TopicID"`
	PartitionOffsets []PartitionOffset `json:"PartitionOffsets"`
}

TopicPartitionOffsetsInfo topic partition offset information

type UpdateConsumerGroupRequest added in v1.0.15 

type UpdateConsumerGroupRequest struct {
	LogsetId      string   `json:"LogsetId"`
	ConsumerGroup string   `json:"ConsumerGroup"`
	Topics        []string `json:"Topics,omitempty"`
	Timeout       *int     `json:"Timeout,omitempty"`
}

UpdateConsumerGroupRequest update consumer group request

type UpdateConsumerGroupResponse added in v1.0.15 

type UpdateConsumerGroupResponse struct {
	Response struct {
		RequestId string    `json:"RequestId"`
		Error     *APIError `json:"Error,omitempty"`
	} `json:"Response"`
}

UpdateConsumerGroupResponse supports capturing all fields

type UpdateOffsetsRequest added in v1.0.15 

type UpdateOffsetsRequest struct {
	TopicPartitionOffsetsInfo []TopicPartitionOffsetsInfo `json:"TopicPartitionOffsetsInfo"`
	ConsumerGroup             string                      `json:"ConsumerGroup"`
	Consumer                  string                      `json:"Consumer"`
	LogsetId                  string                      `json:"LogsetId"`
}

UpdateOffsetsRequest update offset request body

type Worker 

type Worker struct {
	// contains filtered or unexported fields
}

func NewWorker 

func NewWorker(clsClient *CLSClient, retryQueue *LogSendRetryQueue, maxSendWorkerCount int64, producer *AsyncProducerClient) *Worker

NewWorker ...

type YunApiLogClient added in v1.0.15 

type YunApiLogClient struct {
	*CLSClient
	// contains filtered or unexported fields
}

func NewYunApiLogClient added in v1.0.15 

func NewYunApiLogClient(accessKeyId, accessKey string, internal bool, securityToken, source, region string) *YunApiLogClient

NewYunApiLogClient constructor

func NewYunApiLogClientSimple added in v1.0.15 

func NewYunApiLogClientSimple(accessKeyId, accessKey string) *YunApiLogClient

NewYunApiLogClientSimple simple constructor, only requires essential parameters

func NewYunApiLogClientWithConfig added in v1.0.15 

func NewYunApiLogClientWithConfig(config YunApiLogClientConfig) *YunApiLogClient

NewYunApiLogClientWithConfig constructor using configuration struct, supports optional parameters

func (*YunApiLogClient) CreateConsumerGroup added in v1.0.15 

func (c *YunApiLogClient) CreateConsumerGroup(logsetId, consumerGroup string, timeout int, topics []string) (*CreateConsumerGroupResponse, error)

CreateConsumerGroup create consumer group

func (*YunApiLogClient) DeleteConsumerGroup added in v1.0.15 

func (c *YunApiLogClient) DeleteConsumerGroup(logsetId, consumerGroup string) (*DeleteConsumerGroupResponse, error)

DeleteConsumerGroup delete consumer group

func (*YunApiLogClient) DoRequest added in v1.0.15 

func (c *YunApiLogClient) DoRequest(
	method string,
	resource string,
	params map[string]string,
	headers map[string]string,
	body []byte,
	region string,
	action string,
	responseBodyType string,
	service string,
) ([]byte, error)

DoRequest

func (*YunApiLogClient) GetOffsets added in v1.0.15 

func (c *YunApiLogClient) GetOffsets(logsetId, consumerGroup, topicId string, partitionId int, position string) (*ConsumerGroupGetOffsetsResponse, error)

GetOffsets get offsets

func (*YunApiLogClient) Heartbeat added in v1.0.15 

func (c *YunApiLogClient) Heartbeat(logsetId, consumerGroup, consumer string, partitions []PartitionInfo) (*ConsumerGroupHeartBeatResponse, error)

Heartbeat heartbeat

func (*YunApiLogClient) ListConsumerGroup added in v1.0.15 

func (c *YunApiLogClient) ListConsumerGroup(logsetId string, topics []string) (*ListConsumerGroupResponse, error)

ListConsumerGroup list consumer groups

func (*YunApiLogClient) UpdateConsumerGroup added in v1.0.15 

func (c *YunApiLogClient) UpdateConsumerGroup(logsetId, consumerGroup string, topics []string, timeout *int) (*UpdateConsumerGroupResponse, error)

UpdateConsumerGroup update consumer group

func (*YunApiLogClient) UpdateOffsets added in v1.0.15 

func (c *YunApiLogClient) UpdateOffsets(logsetId, consumerGroup, consumer string, offsets []TopicPartitionOffsetsInfo) (*ConsumerGroupUpdateOffsetsResponse, error)

UpdateOffsets update offsets

type YunApiLogClientConfig added in v1.0.15 

type YunApiLogClientConfig struct {
	AccessKeyId   string
	AccessKey     string
	Internal      bool
	SecurityToken string
	Source        string
	Region        string
}

YunApiLogClientConfig configuration struct, supports optional parameters

type ZapLogger added in v1.0.15 

type ZapLogger struct {
	// contains filtered or unexported fields
}

func GetZapLoggerAdapter added in v1.0.15 

func GetZapLoggerAdapter() *ZapLogger

provide get logger instance function

func NewZapLogger added in v1.0.15 

func NewZapLogger(logger *zap.Logger) *ZapLogger

func (*ZapLogger) Debug added in v1.0.15 

func (z *ZapLogger) Debug(msg string, args ...Field)

func (*ZapLogger) Error added in v1.0.15 

func (z *ZapLogger) Error(msg string, args ...Field)

func (*ZapLogger) Info added in v1.0.15 

func (z *ZapLogger) Info(msg string, args ...Field)

func (*ZapLogger) Warn added in v1.0.15 

func (z *ZapLogger) Warn(msg string, args ...Field)

type ZstdDecoderParams added in v1.0.2 

type ZstdDecoderParams struct {
}

type ZstdEncoderParams added in v1.0.2 

type ZstdEncoderParams struct {
	Level int
}

Source Files

View all Source files

Directories

Path Synopsis

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.