Skip to content

[INLONG-12203][SDK] Dataproxy Go SDK support configuring request timeout in discoverer #12210

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
yfsn666 wants to merge 1 commit into
base: master
Choose a base branch
from
Open

[INLONG-12203][SDK] Dataproxy Go SDK support configuring request timeout in discoverer #12210

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-golang/dataproxy/client.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
}
Expand Down
8 changes: 6 additions & 2 deletions inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-golang/dataproxy/discoverer.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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")
}
Expand All @@ -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{}),
Expand All @@ -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
Expand Down Expand Up @@ -194,16 +196,18 @@ 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)
}
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)
Expand Down
65 changes: 35 additions & 30 deletions inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-golang/dataproxy/options.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
}
Expand Down
10 changes: 10 additions & 0 deletions inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-golang/dataproxy/options_basic.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
}
}
Loading