Expose SLI metrics (#209)

Author: jkohen

cmd/stackdriver-prometheus-sidecar/main.go

@@ -55,6 +55,8 @@ import (
 	"github.com/prometheus/prometheus/promql"
 	"go.opencensus.io/plugin/ocgrpc"
 	"go.opencensus.io/plugin/ochttp"
+	"go.opencensus.io/resource"
+	"go.opencensus.io/stats"
 	"go.opencensus.io/stats/view"
 	"go.opencensus.io/tag"
 	metric_pb "google.golang.org/genproto/googleapis/api/metric"
@@ -66,6 +68,14 @@ import (
 var (
 	sizeDistribution    = view.Distribution(0, 1024, 2048, 4096, 16384, 65536, 262144, 1048576, 4194304, 33554432)
 	latencyDistribution = view.Distribution(0, 1, 2, 5, 10, 15, 25, 50, 100, 200, 400, 800, 1500, 3000, 6000)
+
+	// VersionTag identifies the version of this binary.
+	VersionTag = tag.MustNewKey("version")
+	// UptimeMeasure is a cumulative metric.
+	UptimeMeasure = stats.Int64(
+		"agent.googleapis.com/agent/uptime",
+		"uptime of the Stackdriver Prometheus collector",
+		stats.UnitSeconds)
 )
 
 func init() {
@@ -108,6 +118,15 @@ func init() {
 	); err != nil {
 		panic(err)
 	}
+	if err := view.Register(
+		&view.View{
+			Measure:     UptimeMeasure,
+			TagKeys:     []tag.Key{VersionTag},
+			Aggregation: view.Sum(),
+		},
+	); err != nil {
+		panic(err)
+	}
 }
 
 type kubernetesConfig struct {
@@ -276,6 +295,21 @@ func main() {
 	level.Info(logger).Log("host_details", Uname())
 	level.Info(logger).Log("fd_limits", FdLimits())
 
+	// We instantiate a context here since the tailer is used by two other components.
+	// The context will be used in the lifecycle of prometheusReader further down.
+	ctx, cancel := context.WithCancel(context.Background())
+
+	go func() {
+		uptimeUpdateTime := time.Now()
+		c := time.Tick(60 * time.Second)
+		for now := range c {
+			stats.RecordWithTags(ctx,
+				[]tag.Mutator{tag.Upsert(VersionTag, fmt.Sprintf("stackdriver-prometheus-sidecar/%s", version.Version))},
+				UptimeMeasure.M(int64(now.Sub(uptimeUpdateTime).Seconds())))
+			uptimeUpdateTime = now
+		}
+	}()
+
 	httpClient := &http.Client{Transport: &ochttp.Transport{}}
 
 	if *projectId == "" {
@@ -294,14 +328,30 @@ func main() {
 			}
 			view.RegisterExporter(promExporter)
 		case "stackdriver":
-			sd, err := oc_stackdriver.NewExporter(oc_stackdriver.Options{ProjectID: *projectId})
+			const reportingInterval = 60 * time.Second
+			sd, err := oc_stackdriver.NewExporter(oc_stackdriver.Options{
+				ProjectID: *projectId,
+				// If the OpenCensus resource environment variables aren't set, the monitored resource will likely fall back to `generic_task`.
+				ResourceDetector:  resource.FromEnv,
+				ReportingInterval: reportingInterval,
+				// Disable default `opencensus_task` label.
+				DefaultMonitoringLabels: &oc_stackdriver.Labels{},
+				GetMetricType: func(v *view.View) string {
+					// Curated metrics produced by this process.
+					if strings.Contains(v.Name, "agent.googleapis.com") {
+						return v.Name
+					}
+					// Default OpenCensus behavior.
+					return path.Join("custom.googleapis.com", "opencensus", v.Name)
+				},
+			})
 			if err != nil {
 				level.Error(logger).Log("msg", "Creating Stackdriver exporter failed", "err", err)
 				os.Exit(1)
 			}
 			defer sd.Flush()
 			view.RegisterExporter(sd)
-			view.SetReportingPeriod(60 * time.Second)
+			view.SetReportingPeriod(reportingInterval)
 		default:
 			level.Error(logger).Log("msg", "Unknown monitoring backend", "backend", backend)
 			os.Exit(1)
@@ -356,10 +406,6 @@ func main() {
 	}
 	metadataCache := metadata.NewCache(httpClient, metadataURL, cfg.StaticMetadata)
 
-	// We instantiate a context here since the tailer is used by two other components.
-	// The context will be used in the lifecycle of prometheusReader further down.
-	ctx, cancel := context.WithCancel(context.Background())
-
 	tailer, err := tail.Tail(ctx, cfg.WALDirectory)
 	if err != nil {
 		level.Error(logger).Log("msg", "Tailing WAL failed", "err", err)

opencensus/doc.go

@@ -0,0 +1,2 @@
+// Package opencensus for the Stackdriver Prometheus collector.
+package opencensus

opencensus/test_exporter.go

@@ -0,0 +1,98 @@
+package opencensus
+
+import (
+	"context"
+	"fmt"
+	"sort"
+	"strings"
+	"time"
+
+	"go.opencensus.io/metric/metricdata"
+	"go.opencensus.io/metric/metricexport"
+	"go.opencensus.io/stats/view"
+)
+
+// TestExporter keeps exported metric data in memory to aid in testing the instrumentation.
+//
+// Metrics can be retrieved with `GetPoint()`. In order to deterministically retrieve the most recent values, you must first invoke `ReadAndExport()`.
+type TestExporter struct {
+	// points is a map from a label signature to the latest value for the time series represented by the signature.
+	// Use function `labelSignature` to get a signature from a `metricdata.Metric`.
+	points       map[string]metricdata.Point
+	metricReader *metricexport.Reader
+}
+
+// NewTestExporter returns a new exporter.
+func NewTestExporter(metricReader *metricexport.Reader) *TestExporter {
+	return &TestExporter{points: make(map[string]metricdata.Point), metricReader: metricReader}
+}
+
+// ExportMetrics records the view data.
+func (e *TestExporter) ExportMetrics(ctx context.Context, data []*metricdata.Metric) error {
+	for _, metric := range data {
+		for _, ts := range metric.TimeSeries {
+			signature := labelSignature(metric.Descriptor.Name, labelObjectsToKeyValue(metric.Descriptor.LabelKeys, ts.LabelValues))
+			e.points[signature] = ts.Points[len(ts.Points)-1]
+		}
+	}
+	return nil
+}
+
+// GetPoint returns the latest point for the time series identified by the given labels.
+func (e *TestExporter) GetPoint(metricName string, labels map[string]string) (metricdata.Point, bool) {
+	v, ok := e.points[labelSignature(metricName, labelMapToKeyValue(labels))]
+	return v, ok
+}
+
+// ReadAndExport reads the current values for all metrics and makes them available to this exporter.
+func (e *TestExporter) ReadAndExport() {
+	// The next line forces the view worker to process all stats.Record* calls that
+	// happened within Store() before the call to ReadAndExport below. This abuses the
+	// worker implementation to work around lack of synchronization.
+	// TODO(jkohen,rghetia): figure out a clean way to make this deterministic.
+	view.SetReportingPeriod(time.Minute)
+	e.metricReader.ReadAndExport(e)
+}
+
+func (e *TestExporter) String() string {
+	return fmt.Sprintf("points{%v}", e.points)
+}
+
+type keyValue struct {
+	Key   string
+	Value string
+}
+
+func sortKeyValue(kv []keyValue) {
+	sort.Slice(kv, func(i, j int) bool { return kv[i].Key < kv[j].Key })
+}
+
+func labelMapToKeyValue(labels map[string]string) []keyValue {
+	kv := make([]keyValue, 0, len(labels))
+	for k, v := range labels {
+		kv = append(kv, keyValue{Key: k, Value: v})
+	}
+	sortKeyValue(kv)
+	return kv
+}
+
+func labelObjectsToKeyValue(keys []metricdata.LabelKey, values []metricdata.LabelValue) []keyValue {
+	kv := make([]keyValue, 0, len(values))
+	for i := range keys {
+		if values[i].Present {
+			kv = append(kv, keyValue{Key: keys[i].Key, Value: values[i].Value})
+		}
+	}
+	sortKeyValue(kv)
+	return kv
+}
+
+// labelSignature returns a string that uniquely identifies the list of labels given in the input.
+func labelSignature(metricName string, kv []keyValue) string {
+	var builder strings.Builder
+	for _, x := range kv {
+		builder.WriteString(x.Key)
+		builder.WriteString(x.Value)
+	}
+	return fmt.Sprintf("%s{%s}", metricName, builder.String())
+}

stackdriver/client.go

@@ -24,6 +24,9 @@ import (
 	"time"
 
 	"go.opencensus.io/plugin/ocgrpc"
+	"go.opencensus.io/stats"
+	"go.opencensus.io/stats/view"
+	"go.opencensus.io/tag"
 	monitoring "google.golang.org/genproto/googleapis/monitoring/v3"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/balancer/roundrobin"
@@ -43,6 +46,27 @@ const (
 	MonitoringWriteScope      = "https://www.googleapis.com/auth/monitoring.write"
 )
 
+var (
+	// StatusTag is the google3 canonical status code: google3/google/rpc/code.proto
+	StatusTag = tag.MustNewKey("status")
+
+	// PointCount is a metric.
+	PointCount = stats.Int64("agent.googleapis.com/agent/monitoring/point_count",
+		"count of metric points written to Stackdriver", stats.UnitDimensionless)
+)
+
+func init() {
+	if err := view.Register(
+		&view.View{
+			Measure:     PointCount,
+			TagKeys:     []tag.Key{StatusTag},
+			Aggregation: view.Sum(),
+		},
+	); err != nil {
+		panic(err)
+	}
+}
+
 // Client allows reading and writing from/to a remote gRPC endpoint. The
 // implementation may hit a single backend, so the application should create a
 // number of these clients.
@@ -169,7 +193,12 @@ func (c *Client) Store(req *monitoring.CreateTimeSeriesRequest) error {
 				TimeSeries: req.TimeSeries[begin:end],
 			}
 			_, err := service.CreateTimeSeries(ctx, req_copy)
-			if err != nil {
+			if err == nil {
+				// The response is empty if all points were successfully written.
+				stats.RecordWithTags(ctx,
+					[]tag.Mutator{tag.Upsert(StatusTag, "0")},
+					PointCount.M(int64(end-begin)))
+			} else {
 				level.Debug(c.logger).Log(
 					"msg", "Partial failure calling CreateTimeSeries",
 					"err", err)
@@ -179,6 +208,19 @@ func (c *Client) Store(req *monitoring.CreateTimeSeriesRequest) error {
 					errors <- err
 					return
 				}
+				for _, details := range status.Details() {
+					if summary, ok := details.(*monitoring.CreateTimeSeriesSummary); ok {
+						level.Debug(c.logger).Log("summary", summary)
+						stats.RecordWithTags(ctx,
+							[]tag.Mutator{tag.Upsert(StatusTag, "0")},
+							PointCount.M(int64(summary.SuccessPointCount)))
+						for _, e := range summary.Errors {
+							stats.RecordWithTags(ctx,
+								[]tag.Mutator{tag.Upsert(StatusTag, fmt.Sprint(uint32(e.Status.Code)))},
+								PointCount.M(int64(e.PointCount)))
+						}
+					}
+				}
 				switch status.Code() {
 				// codes.DeadlineExceeded:
 				//   It is safe to retry