diff --git a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-golang/dataproxy/client.go b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-golang/dataproxy/client.go index f06b249a6b..d4ca344437 100755 --- a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-golang/dataproxy/client.go +++ b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-golang/dataproxy/client.go @@ -122,7 +122,7 @@ func (c *client) initAll() error { } func (c *client) initDiscoverer() error { - dis, err := NewDiscoverer(c.options.URL, c.options.GroupID, c.options.UpdateInterval, c.options.Logger, c.options.Auth) + dis, err := NewDiscoverer(c.options.URL, c.options.GroupID, c.options.UpdateInterval, c.options.DiscovererRequestTimeout, c.options.Logger, c.options.Auth) if err != nil { return err } diff --git a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-golang/dataproxy/discoverer.go b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-golang/dataproxy/discoverer.go index 5356777ee8..a5ecca6331 100755 --- a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-golang/dataproxy/discoverer.go +++ b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-golang/dataproxy/discoverer.go @@ -39,7 +39,7 @@ type Auth interface { } // NewDiscoverer news a DataProxy discoverer -func NewDiscoverer(url, groupID string, lookupInterval time.Duration, log logger.Logger, auth Auth) (discoverer.Discoverer, error) { +func NewDiscoverer(url, groupID string, lookupInterval, requestTimeout time.Duration, log logger.Logger, auth Auth) (discoverer.Discoverer, error) { if url == "" { return nil, errors.New("URL is not given") } @@ -57,6 +57,7 @@ func NewDiscoverer(url, groupID string, lookupInterval time.Duration, log logger url: url, groupID: groupID, lookupInterval: lookupInterval, + requestTimeout: requestTimeout, endpointList: make([]discoverer.Endpoint, 0), endpointListMap: make(map[string]discoverer.Endpoint), eventHandlers: make(map[discoverer.EventHandler]struct{}), @@ -75,6 +76,7 @@ type dataProxyDiscoverer struct { url string groupID string lookupInterval time.Duration + requestTimeout time.Duration endpointList discoverer.EndpointList endpointListStr string endpointListMap map[string]discoverer.Endpoint @@ -194,9 +196,10 @@ func (d *dataProxyDiscoverer) update() { // get gets endpoint list from DataProxy service registry func (d *dataProxyDiscoverer) get(retry int) (*cluster, error) { reqURL := fmt.Sprintf("%s/%s?protocolType=tcp", d.url, d.groupID) - client := resty.New().SetTLSClientConfig(&tls.Config{InsecureSkipVerify: true}) + client := resty.New().SetTLSClientConfig(&tls.Config{InsecureSkipVerify: true}).SetTimeout(d.requestTimeout) req := client.R() if d.auth != nil { + d.log.Debug("start to get auth token") key, token, err := d.auth.GetToken(context.Background(), d.groupID) if err != nil { return nil, fmt.Errorf("failed to get auth token. %w", err) @@ -204,6 +207,7 @@ func (d *dataProxyDiscoverer) get(retry int) (*cluster, error) { req = req.SetHeader(key, token) } + d.log.Debug("start to get server endpoint list") httpRsp, err := req.Post(reqURL) if err != nil { d.log.Error("get server endpoint list failed:", err) diff --git a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-golang/dataproxy/options.go b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-golang/dataproxy/options.go index 200096d8ce..f2033f7b3f 100755 --- a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-golang/dataproxy/options.go +++ b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-golang/dataproxy/options.go @@ -49,36 +49,37 @@ func init() { // Options is the DataProxy go client configs type Options struct { - GroupID string // InLong group ID - URL string // the Manager URL for discovering the DataProxy cluster - UpdateInterval time.Duration // interval to refresh the endpoint list, default: 5m - ConnTimeout time.Duration // connection timeout: default: 3000ms - WriteBufferSize int // write buffer size in bytes, default: 8M - ReadBufferSize int // read buffer size in bytes, default: 1M - SocketSendBufferSize int // socket send buffer size in bytes, default: 8M - SocketRecvBufferSize int // socket receive buffer size in bytes, default: 1M - BufferPool bufferpool.BufferPool // encoding/decoding buffer pool, if not given, SDK will init a new one - BytePool bufferpool.BytePool // encoding/decoding byte pool, if not given, SDK will init a new one - BufferPoolSize int // buffer pool size, default: 409600 - BytePoolSize int // byte pool size, default: 409600 - BytePoolWidth int // byte pool width, default: equals to BatchingMaxSize - Logger logger.Logger // debug logger, default: stdout - MetricsName string // the unique metrics name of this SDK, used to isolate metrics in the case that more than 1 client are initialized in one process - MetricsRegistry prometheus.Registerer // metrics registry, default: prometheus.DefaultRegisterer - WorkerNum int // worker number, default: 8 - SendTimeout time.Duration // send timeout, default: 30000ms - MaxRetries int // max retry count, default: 2 - RetryOnServerError bool // whether to retry on server error, default: false - RetryInitialInterval time.Duration // initial retry interval for exponential backoff, default: 100ms - BatchingMaxPublishDelay time.Duration // the time period within which the messages sent will be batched, default: 20ms - BatchingMaxMessages int // the maximum number of messages permitted in a batch, default: 50 - BatchingMaxSize int // the maximum number of bytes permitted in a batch, default: 40K - MaxPendingMessages int // the max size of the queue holding the messages pending to receive an acknowledgment from the broker, default: 204800 - BlockIfQueueIsFull bool // whether Send and SendAsync block if producer's message queue is full, default: false - AddColumns map[string]string // addition columns to add to the message, for example: __addcol1__worldid=xxx&__addcol2__ip=yyy, all the message will be added 2 more columns with worldid=xxx and ip=yyy - addColumnStr string // the string format of the AddColumns, just a cache, used internal - Auth Auth // dataproxy authentication interface - MaxConnLifetime time.Duration // connection max lifetime, default: 0, set to 5m/10m when the servers provide service though CLBs (Cloud Load Balancers) + GroupID string // InLong group ID + URL string // the Manager URL for discovering the DataProxy cluster + UpdateInterval time.Duration // interval to refresh the endpoint list, default: 5m + ConnTimeout time.Duration // connection timeout: default: 3000ms + WriteBufferSize int // write buffer size in bytes, default: 8M + ReadBufferSize int // read buffer size in bytes, default: 1M + SocketSendBufferSize int // socket send buffer size in bytes, default: 8M + SocketRecvBufferSize int // socket receive buffer size in bytes, default: 1M + BufferPool bufferpool.BufferPool // encoding/decoding buffer pool, if not given, SDK will init a new one + BytePool bufferpool.BytePool // encoding/decoding byte pool, if not given, SDK will init a new one + BufferPoolSize int // buffer pool size, default: 409600 + BytePoolSize int // byte pool size, default: 409600 + BytePoolWidth int // byte pool width, default: equals to BatchingMaxSize + Logger logger.Logger // debug logger, default: stdout + MetricsName string // the unique metrics name of this SDK, used to isolate metrics in the case that more than 1 client are initialized in one process + MetricsRegistry prometheus.Registerer // metrics registry, default: prometheus.DefaultRegisterer + WorkerNum int // worker number, default: 8 + SendTimeout time.Duration // send timeout, default: 30000ms + MaxRetries int // max retry count, default: 2 + RetryOnServerError bool // whether to retry on server error, default: false + RetryInitialInterval time.Duration // initial retry interval for exponential backoff, default: 100ms + BatchingMaxPublishDelay time.Duration // the time period within which the messages sent will be batched, default: 20ms + BatchingMaxMessages int // the maximum number of messages permitted in a batch, default: 50 + BatchingMaxSize int // the maximum number of bytes permitted in a batch, default: 40K + MaxPendingMessages int // the max size of the queue holding the messages pending to receive an acknowledgment from the broker, default: 204800 + BlockIfQueueIsFull bool // whether Send and SendAsync block if producer's message queue is full, default: false + AddColumns map[string]string // addition columns to add to the message, for example: __addcol1__worldid=xxx&__addcol2__ip=yyy, all the message will be added 2 more columns with worldid=xxx and ip=yyy + addColumnStr string // the string format of the AddColumns, just a cache, used internal + Auth Auth // dataproxy authentication interface + MaxConnLifetime time.Duration // connection max lifetime, default: 0, set to 5m/10m when the servers provide service though CLBs (Cloud Load Balancers) + DiscovererRequestTimeout time.Duration // HTTP request timeout when the discoverer fetches endpoint list from the Manager, default: 30s } // ValidateAndSetDefault validates an options and set up the default values @@ -101,6 +102,10 @@ func (options *Options) ValidateAndSetDefault() error { options.ConnTimeout = 3 * time.Second } + if options.DiscovererRequestTimeout <= 0 { + options.DiscovererRequestTimeout = 30 * time.Second + } + if options.BatchingMaxPublishDelay <= 0 { options.BatchingMaxPublishDelay = 20 * time.Millisecond } diff --git a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-golang/dataproxy/options_basic.go b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-golang/dataproxy/options_basic.go index de3c228769..664d5d22d8 100755 --- a/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-golang/dataproxy/options_basic.go +++ b/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-golang/dataproxy/options_basic.go @@ -189,3 +189,13 @@ func WithMaxConnLifetime(lifetime time.Duration) Option { o.MaxConnLifetime = lifetime } } + +// WithDiscovererRequestTimeout sets DiscovererRequestTimeout +func WithDiscovererRequestTimeout(t time.Duration) Option { + return func(o *Options) { + if t <= 0 { + return + } + o.DiscovererRequestTimeout = t + } +}