instrument pubsub ack and mod-ack-deadline calls

Change-Id: I84c37f34b41099efad6407034f4882e175229731

pubsub/service.go

diff --git a/pubsub/service.go b/pubsub/service.go
index 328fe48..0b5cccd 100644
--- a/pubsub/service.go
+++ b/pubsub/service.go
@@ -17,6 +17,7 @@
 import (
 	"fmt"
 	"io"
+	"log"
 	"math"
 	"sync"
 	"time"
@@ -24,6 +25,7 @@
 	"cloud.google.com/go/iam"
 	"cloud.google.com/go/internal/version"
 	vkit "cloud.google.com/go/pubsub/apiv1"
+	"github.com/DataDog/datadog-go/statsd"
 	"golang.org/x/net/context"
 	"google.golang.org/api/option"
 	pb "google.golang.org/genproto/googleapis/pubsub/v1"
@@ -31,6 +33,8 @@
 	"google.golang.org/grpc/codes"
 )
 
+const sampleRate = 1.0
+
 type nextStringFunc func() (string, error)
 
 // service provides an internal abstraction to isolate the generated
@@ -72,8 +76,9 @@
 }
 
 type apiService struct {
-	pubc *vkit.PublisherClient
-	subc *vkit.SubscriberClient
+	pubc  *vkit.PublisherClient
+	subc  *vkit.SubscriberClient
+	statc *statsd.Client
 }
 
 func newPubSubService(ctx context.Context, opts []option.ClientOption) (*apiService, error) {
@@ -88,7 +93,14 @@
 	}
 	pubc.SetGoogleClientInfo("gccl", version.Repo)
 	subc.SetGoogleClientInfo("gccl", version.Repo)
-	return &apiService{pubc: pubc, subc: subc}, nil
+	statc, err := statsd.NewBuffered("127.0.0.1:8125", 1000)
+	if err != nil {
+		_ = pubc.Close()
+		_ = subc.Close()
+		return nil, err
+	}
+	statc.Namespace = "gopubsub"
+	return &apiService{pubc: pubc, subc: subc, statc: statc}, nil
 }
 
 func (s *apiService) close() error {
@@ -205,11 +217,25 @@
 }
 
 func (s *apiService) modifyAckDeadline(ctx context.Context, subName string, deadline time.Duration, ackIDs []string) error {
-	return s.subc.ModifyAckDeadline(ctx, &pb.ModifyAckDeadlineRequest{
+	start := time.Now()
+	err := s.subc.ModifyAckDeadline(ctx, &pb.ModifyAckDeadlineRequest{
 		Subscription:       subName,
 		AckIds:             ackIDs,
 		AckDeadlineSeconds: trunc32(int64(deadline.Seconds())),
 	})
+	elapsed := time.Since(start)
+	s.count("rpc.mod_ack_deadline.count", 1)
+	if deadline == 0 {
+		s.count("rpc.mod_ack_deadline.zeroes", 1)
+	}
+	s.histogram("rpc.mod_ack_deadline.time", elapsed.Seconds())
+	if err != nil {
+		s.count("rpc.mod_ack_deadline.errs", 1)
+	} else {
+		s.count("rpc.mod_ack_deadline.ids", len(ackIDs))
+	}
+	return err
+
 }
 
 // maxPayload is the maximum number of bytes to devote to actual ids in
@@ -242,10 +268,32 @@
 }
 
 func (s *apiService) acknowledge(ctx context.Context, subName string, ackIDs []string) error {
-	return s.subc.Acknowledge(ctx, &pb.AcknowledgeRequest{
+	start := time.Now()
+	err := s.subc.Acknowledge(ctx, &pb.AcknowledgeRequest{
 		Subscription: subName,
 		AckIds:       ackIDs,
 	})
+	elapsed := time.Since(start)
+	s.count("rpc.ack.count", 1)
+	s.histogram("rpc.ack.time", elapsed.Seconds())
+	if err != nil {
+		s.count("rpc.ack.errs", 1)
+	} else {
+		s.count("rpc.ack.ids", len(ackIDs))
+	}
+	return err
+}
+
+func (s *apiService) count(name string, val int) {
+	if err := s.statc.Count(name, val, nil, sampleRate); err != nil {
+		log.Printf("statsd error: %v", err)
+	}
+}
+
+func (s *apiService) histogram(name string, val float64) {
+	if err := s.statc.Histogram(name, val, nil, sampleRate); err != nil {
+		log.Printf("statsd error: %v", err)
+	}
 }
 
 func (s *apiService) fetchMessages(ctx context.Context, subName string, maxMessages int32) ([]*Message, error) {