github/weave-scope
1
0
Fork 0
mirror of https://github.com/weaveworks/scope.git synced 2026-07-19 05:19:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add CLI flag for SQS RPC timeout

Browse Source
This commit is contained in:
Marcus Cobden
2018-05-03 17:02:19 +01:00
parent d19b0293f6
commit ba81924278
3 changed files with 9 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
app/multitenant/sqs_control_router.go
View File

@@ -22,7 +22,6 @@ import (
var (
longPollTime = aws.Int64(10)
rpcTimeout = time.Minute
sqsRequestDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{
Namespace: "scope",
Name: "sqs_request_duration_seconds",
@@ -45,6 +44,7 @@ type sqsControlRouter struct {
responseQueueURL *string
userIDer UserIDer
prefix string
rpcTimeout time.Duration
mtx sync.Mutex
responses map[string]chan xfer.Response
@@ -63,12 +63,13 @@ type sqsResponseMessage struct {
}
// NewSQSControlRouter the harbinger of death
func NewSQSControlRouter(config *aws.Config, userIDer UserIDer, prefix string) app.ControlRouter {
func NewSQSControlRouter(config *aws.Config, userIDer UserIDer, prefix string, rpcTimeout time.Duration) app.ControlRouter {
result := &sqsControlRouter{
service: sqs.New(session.New(config)),
responseQueueURL: nil,
userIDer: userIDer,
prefix: prefix,
rpcTimeout: rpcTimeout,
responses: map[string]chan xfer.Response{},
probeWorkers: map[int64]*probeWorker{},
}
@@ -257,7 +258,7 @@ func (cr *sqsControlRouter) Handle(ctx context.Context, probeID string, req xfer
select {
case response := <-waiter:
return response, nil
case <-time.After(rpcTimeout):
case <-time.After(cr.rpcTimeout):
return xfer.Response{}, fmt.Errorf("request timed out")
}
}

6
prog/app.go
View File

@@ -147,7 +147,7 @@ func emitterFactory(collector app.Collector, clientCfg billing.Config, userIDer
)
}
func controlRouterFactory(userIDer multitenant.UserIDer, controlRouterURL string) (app.ControlRouter, error) {
func controlRouterFactory(userIDer multitenant.UserIDer, controlRouterURL string, controlRPCTimeout time.Duration) (app.ControlRouter, error) {
if controlRouterURL == "local" {
return app.NewLocalControlRouter(), nil
}
@@ -163,7 +163,7 @@ func controlRouterFactory(userIDer multitenant.UserIDer, controlRouterURL string
if err != nil {
return nil, err
}
return multitenant.NewSQSControlRouter(sqsConfig, userIDer, prefix), nil
return multitenant.NewSQSControlRouter(sqsConfig, userIDer, prefix, controlRPCTimeout), nil
}
return nil, fmt.Errorf("Invalid control router '%s'", controlRouterURL)
@@ -239,7 +239,7 @@ func appMain(flags appFlags) {
collector = billingEmitter
}
controlRouter, err := controlRouterFactory(userIDer, flags.controlRouterURL)
controlRouter, err := controlRouterFactory(userIDer, flags.controlRouterURL, flags.controlRPCTimeout)
if err != nil {
log.Fatalf("Error creating control router: %v", err)
return

2
prog/main.go
View File

@@ -152,6 +152,7 @@ type appFlags struct {
collectorURL string
s3URL string
controlRouterURL string
controlRPCTimeout time.Duration
pipeRouterURL string
natsHostname string
memcachedHostname string
@@ -348,6 +349,7 @@ func setupFlags(flags *flags) {
flag.StringVar(&flags.app.collectorURL, "app.collector", "local", "Collector to use (local, dynamodb, or file/directory)")
flag.StringVar(&flags.app.s3URL, "app.collector.s3", "local", "S3 URL to use (when collector is dynamodb)")
flag.StringVar(&flags.app.controlRouterURL, "app.control.router", "local", "Control router to use (local or sqs)")
flag.DurationVar(&flags.app.controlRPCTimeout, "app.control.rpctimeout", time.Minute, "Timeout for control RPC")
flag.StringVar(&flags.app.pipeRouterURL, "app.pipe.router", "local", "Pipe router to use (local)")
flag.StringVar(&flags.app.natsHostname, "app.nats", "", "Hostname for NATS service to use for shortcut reports. If empty, shortcut reporting will be disabled.")
flag.StringVar(&flags.app.memcachedHostname, "app.memcached.hostname", "", "Hostname for memcached service to use when caching reports. If empty, no memcached will be used.")