From 77365643372c7ceb7b837ef9bef4e357a539513d Mon Sep 17 00:00:00 2001 From: Paul Bellamy Date: Tue, 5 Jul 2016 11:35:43 +0100 Subject: [PATCH] refactor some timing helpers into a common lib --- app/multitenant/aws_collector.go | 13 +++++++------ app/multitenant/memcache_client.go | 6 ++++-- app/multitenant/s3_client.go | 5 +++-- app/multitenant/sqs_control_router.go | 15 ++++++++------- .../instrument/instrument.go | 19 ++++++++++--------- 5 files changed, 32 insertions(+), 26 deletions(-) rename app/multitenant/common.go => common/instrument/instrument.go (59%) diff --git a/app/multitenant/aws_collector.go b/app/multitenant/aws_collector.go index 477e2261b..eae1038d3 100644 --- a/app/multitenant/aws_collector.go +++ b/app/multitenant/aws_collector.go @@ -19,6 +19,7 @@ import ( "golang.org/x/net/context" "github.com/weaveworks/scope/app" + "github.com/weaveworks/scope/common/instrument" "github.com/weaveworks/scope/report" ) @@ -213,7 +214,7 @@ func (c *awsCollector) CreateTables() error { func (c *awsCollector) getReportKeys(userid string, row int64, start, end time.Time) ([]string, error) { rowKey := fmt.Sprintf("%s-%s", userid, strconv.FormatInt(row, 10)) var resp *dynamodb.QueryOutput - err := timeRequest("Query", dynamoRequestDuration, func() error { + err := instrument.TimeRequest("Query", dynamoRequestDuration, func() error { var err error resp, err = c.db.Query(&dynamodb.QueryInput{ TableName: aws.String(c.tableName), @@ -372,7 +373,7 @@ func (c *awsCollector) Add(ctx context.Context, rep report.Report) error { Add(float64(len(s3Key))) var resp *dynamodb.PutItemOutput - err = timeRequest("PutItem", dynamoRequestDuration, func() error { + err = instrument.TimeRequest("PutItem", dynamoRequestDuration, func() error { var err error resp, err = c.db.PutItem(&dynamodb.PutItemInput{ TableName: aws.String(c.tableName), @@ -401,7 +402,7 @@ func (c *awsCollector) Add(ctx context.Context, rep report.Report) error { if rep.Shortcut && c.nats != nil { err := c.nats.Publish(userid, []byte(s3Key)) - natsRequests.WithLabelValues("Publish", errorCode(err)).Add(1) + natsRequests.WithLabelValues("Publish", instrument.ErrorCode(err)).Add(1) if err != nil { log.Errorf("Error sending shortcut report: %v", err) } @@ -422,7 +423,7 @@ func (c *awsCollector) WaitOn(ctx context.Context, waiter chan struct{}) { } sub, err := c.nats.SubscribeSync(userid) - natsRequests.WithLabelValues("SubscribeSync", errorCode(err)).Add(1) + natsRequests.WithLabelValues("SubscribeSync", instrument.ErrorCode(err)).Add(1) if err != nil { log.Errorf("Error subscribing for shortcuts: %v", err) return @@ -438,7 +439,7 @@ func (c *awsCollector) WaitOn(ctx context.Context, waiter chan struct{}) { if err == nats.ErrTimeout { continue } - natsRequests.WithLabelValues("NextMsg", errorCode(err)).Add(1) + natsRequests.WithLabelValues("NextMsg", instrument.ErrorCode(err)).Add(1) if err != nil { log.Debugf("NextMsg error: %v", err) return @@ -469,7 +470,7 @@ func (c *awsCollector) UnWait(ctx context.Context, waiter chan struct{}) { c.waitersLock.Unlock() err = sub.Unsubscribe() - natsRequests.WithLabelValues("Unsubscribe", errorCode(err)).Add(1) + natsRequests.WithLabelValues("Unsubscribe", instrument.ErrorCode(err)).Add(1) if err != nil { log.Errorf("Error on unsubscribe: %v", err) } diff --git a/app/multitenant/memcache_client.go b/app/multitenant/memcache_client.go index 1a4738369..52a2a79d7 100644 --- a/app/multitenant/memcache_client.go +++ b/app/multitenant/memcache_client.go @@ -11,6 +11,8 @@ import ( log "github.com/Sirupsen/logrus" "github.com/bradfitz/gomemcache/memcache" "github.com/prometheus/client_golang/prometheus" + + "github.com/weaveworks/scope/common/instrument" "github.com/weaveworks/scope/report" ) @@ -136,7 +138,7 @@ func memcacheStatusCode(err error) string { // FetchReports gets reports from memcache. func (c *MemcacheClient) FetchReports(keys []string) (map[string]report.Report, []string, error) { var found map[string]*memcache.Item - err := timeRequestStatus("Get", memcacheRequestDuration, memcacheStatusCode, func() error { + err := instrument.TimeRequestStatus("Get", memcacheRequestDuration, memcacheStatusCode, func() error { var err error found, err = c.client.GetMulti(keys) return err @@ -186,7 +188,7 @@ func (c *MemcacheClient) FetchReports(keys []string) (map[string]report.Report, // StoreBytes stores a report, expecting the report to be serialized already. func (c *MemcacheClient) StoreBytes(key string, content []byte) error { - return timeRequestStatus("Put", memcacheRequestDuration, memcacheStatusCode, func() error { + return instrument.TimeRequestStatus("Put", memcacheRequestDuration, memcacheStatusCode, func() error { item := memcache.Item{Key: key, Value: content, Expiration: c.expiration} return c.client.Set(&item) }) diff --git a/app/multitenant/s3_client.go b/app/multitenant/s3_client.go index 4cb9acd0e..e812ed016 100644 --- a/app/multitenant/s3_client.go +++ b/app/multitenant/s3_client.go @@ -8,6 +8,7 @@ import ( "github.com/aws/aws-sdk-go/service/s3" "github.com/prometheus/client_golang/prometheus" + "github.com/weaveworks/scope/common/instrument" "github.com/weaveworks/scope/report" ) @@ -68,7 +69,7 @@ func (store *S3Store) FetchReports(keys []string) (map[string]report.Report, []s func (store *S3Store) fetchReport(key string) (*report.Report, error) { var resp *s3.GetObjectOutput - err := timeRequest("Get", s3RequestDuration, func() error { + err := instrument.TimeRequest("Get", s3RequestDuration, func() error { var err error resp, err = store.s3.GetObject(&s3.GetObjectInput{ Bucket: aws.String(store.bucketName), @@ -85,7 +86,7 @@ func (store *S3Store) fetchReport(key string) (*report.Report, error) { // StoreBytes stores a report in S3, expecting the report to be serialized // already. func (store *S3Store) StoreBytes(key string, content []byte) error { - return timeRequest("Put", s3RequestDuration, func() error { + return instrument.TimeRequest("Put", s3RequestDuration, func() error { _, err := store.s3.PutObject(&s3.PutObjectInput{ Body: bytes.NewReader(content), Bucket: aws.String(store.bucketName), diff --git a/app/multitenant/sqs_control_router.go b/app/multitenant/sqs_control_router.go index 43fbb92d4..722b2ce98 100644 --- a/app/multitenant/sqs_control_router.go +++ b/app/multitenant/sqs_control_router.go @@ -16,6 +16,7 @@ import ( "golang.org/x/net/context" "github.com/weaveworks/scope/app" + "github.com/weaveworks/scope/common/instrument" "github.com/weaveworks/scope/common/xfer" ) @@ -94,7 +95,7 @@ func (cr *sqsControlRouter) getOrCreateQueue(name string) (*string, error) { // CreateQueue creates a queue or if it already exists, returns url of said queue var createQueueRes *sqs.CreateQueueOutput var err error - err = timeRequestStatus("CreateQueue", sqsRequestDuration, nil, func() error { + err = instrument.TimeRequestStatus("CreateQueue", sqsRequestDuration, nil, func() error { createQueueRes, err = cr.service.CreateQueue(&sqs.CreateQueueInput{ QueueName: aws.String(name), }) @@ -127,7 +128,7 @@ func (cr *sqsControlRouter) loop() { for { var res *sqs.ReceiveMessageOutput var err error - err = timeRequestStatus("ReceiveMessage", sqsRequestDuration, nil, func() error { + err = instrument.TimeRequestStatus("ReceiveMessage", sqsRequestDuration, nil, func() error { res, err = cr.service.ReceiveMessage(&sqs.ReceiveMessageInput{ QueueUrl: responseQueueURL, WaitTimeSeconds: longPollTime, @@ -157,7 +158,7 @@ func (cr *sqsControlRouter) deleteMessages(queueURL *string, messages []*sqs.Mes Id: message.MessageId, }) } - return timeRequestStatus("DeleteMessageBatch", sqsRequestDuration, nil, func() error { + return instrument.TimeRequestStatus("DeleteMessageBatch", sqsRequestDuration, nil, func() error { _, err := cr.service.DeleteMessageBatch(&sqs.DeleteMessageBatchInput{ QueueUrl: queueURL, Entries: entries, @@ -193,7 +194,7 @@ func (cr *sqsControlRouter) sendMessage(queueURL *string, message interface{}) e } log.Infof("sendMessage to %s: %s", *queueURL, buf.String()) - return timeRequestStatus("SendMessage", sqsRequestDuration, nil, func() error { + return instrument.TimeRequestStatus("SendMessage", sqsRequestDuration, nil, func() error { _, err := cr.service.SendMessage(&sqs.SendMessageInput{ QueueUrl: queueURL, MessageBody: aws.String(buf.String()), @@ -216,7 +217,7 @@ func (cr *sqsControlRouter) Handle(ctx context.Context, probeID string, req xfer } var probeQueueURL *sqs.GetQueueUrlOutput - err = timeRequestStatus("GetQueueUrl", sqsRequestDuration, nil, func() error { + err = instrument.TimeRequestStatus("GetQueueUrl", sqsRequestDuration, nil, func() error { probeQueueName := fmt.Sprintf("%sprobe-%s-%s", cr.prefix, userID, probeID) probeQueueURL, err = cr.service.GetQueueUrl(&sqs.GetQueueUrlInput{ QueueName: aws.String(probeQueueName), @@ -240,7 +241,7 @@ func (cr *sqsControlRouter) Handle(ctx context.Context, probeID string, req xfer }() // Next, send the request to that queue - if err := timeRequestStatus("SendMessage", sqsRequestDuration, nil, func() error { + if err := instrument.TimeRequestStatus("SendMessage", sqsRequestDuration, nil, func() error { return cr.sendMessage(probeQueueURL.QueueUrl, sqsRequestMessage{ ID: id, Request: req, @@ -323,7 +324,7 @@ func (pw *probeWorker) loop() { var res *sqs.ReceiveMessageOutput var err error - err = timeRequestStatus("ReceiveMessage", sqsRequestDuration, nil, func() error { + err = instrument.TimeRequestStatus("ReceiveMessage", sqsRequestDuration, nil, func() error { res, err = pw.router.service.ReceiveMessage(&sqs.ReceiveMessageInput{ QueueUrl: pw.requestQueueURL, WaitTimeSeconds: longPollTime, diff --git a/app/multitenant/common.go b/common/instrument/instrument.go similarity index 59% rename from app/multitenant/common.go rename to common/instrument/instrument.go index c3c1f6130..ce56f8f70 100644 --- a/app/multitenant/common.go +++ b/common/instrument/instrument.go @@ -1,4 +1,4 @@ -package multitenant +package instrument import ( "time" @@ -6,31 +6,32 @@ import ( "github.com/prometheus/client_golang/prometheus" ) -func errorCode(err error) string { +// ErrorCode converts an error in to an http-style error-code. +func ErrorCode(err error) string { if err == nil { return "200" } return "500" } -// timeRequest runs 'f' and records how long it took in the given Prometheus +// TimeRequest runs 'f' and records how long it took in the given Prometheus // metric. If 'f' returns successfully, record a "200". Otherwise, record // "500". // // If you want more complicated logic for translating errors into statuses, -// use 'timeRequestStatus'. -func timeRequest(method string, metric *prometheus.SummaryVec, f func() error) error { - return timeRequestStatus(method, metric, errorCode, f) +// use 'TimeRequestStatus'. +func TimeRequest(method string, metric *prometheus.SummaryVec, f func() error) error { + return TimeRequestStatus(method, metric, ErrorCode, f) } -// timeRequestStatus runs 'f' and records how long it took in the given +// TimeRequestStatus runs 'f' and records how long it took in the given // Prometheus metric. // // toStatusCode is a function that translates errors returned by 'f' into // HTTP-like status codes. -func timeRequestStatus(method string, metric *prometheus.SummaryVec, toStatusCode func(error) string, f func() error) error { +func TimeRequestStatus(method string, metric *prometheus.SummaryVec, toStatusCode func(error) string, f func() error) error { if toStatusCode == nil { - toStatusCode = errorCode + toStatusCode = ErrorCode } startTime := time.Now() err := f()