Merge pull request #89 from xushiwei/q

x/event/export/ocagent

event/export/ocagent/README.md

@@ -0,0 +1,139 @@
+# Exporting Metrics and Traces with OpenCensus, Zipkin, and Prometheus
+
+This tutorial provides a minimum example to verify that metrics and traces
+can be exported to OpenCensus from Go tools.
+
+## Setting up oragent
+
+1. Ensure you have [docker](https://www.docker.com/get-started) and [docker-compose](https://docs.docker.com/compose/install/).
+2. Clone [oragent](https://github.com/orijtech/oragent).
+3. In the oragent directory, start the services:
+```bash
+docker-compose up
+```
+If everything goes well, you should see output resembling the following:
+```
+Starting oragent_zipkin_1 ... done
+Starting oragent_oragent_1 ... done
+Starting oragent_prometheus_1 ... done
+...
+```
+* You can check the status of the OpenCensus agent using zPages at http://localhost:55679/debug/tracez.
+* You can now access the Prometheus UI at http://localhost:9445.
+* You can now access the Zipkin UI at http://localhost:9444.
+4. To shut down oragent, hit Ctrl+C in the terminal.
+5. You can also start oragent in detached mode by running `docker-compose up -d`. To stop oragent while detached, run `docker-compose down`.
+
+## Exporting Metrics and Traces
+1. Clone the [tools](https://golang.org/x/tools) subrepository.
+1. Inside `internal`, create a file named `main.go` with the following contents:
+```go
+package main
+
+import (
+	"context"
+	"fmt"
+	"math/rand"
+	"net/http"
+	"time"
+
+	"golang.org/x/tools/internal/event"
+	"golang.org/x/tools/internal/event/export"
+	"golang.org/x/tools/internal/event/export/metric"
+	"golang.org/x/tools/internal/event/export/ocagent"
+)
+
+type testExporter struct {
+	metrics metric.Exporter
+	ocagent *ocagent.Exporter
+}
+
+func (e *testExporter) ProcessEvent(ctx context.Context, ev event.Event) (context.Context, event.Event) {
+	ctx, ev = export.Tag(ctx, ev)
+	ctx, ev = export.ContextSpan(ctx, ev)
+	ctx, ev = e.metrics.ProcessEvent(ctx, ev)
+	ctx, ev = e.ocagent.ProcessEvent(ctx, ev)
+	return ctx, ev
+}
+
+func main() {
+	exporter := &testExporter{}
+
+	exporter.ocagent = ocagent.Connect(&ocagent.Config{
+		Start:   time.Now(),
+		Address: "http://127.0.0.1:55678",
+		Service: "go-tools-test",
+		Rate:    5 * time.Second,
+		Client:  &http.Client{},
+	})
+	event.SetExporter(exporter)
+
+	ctx := context.TODO()
+	mLatency := event.NewFloat64Key("latency", "the latency in milliseconds")
+	distribution := metric.HistogramFloat64Data{
+		Info: &metric.HistogramFloat64{
+			Name:        "latencyDistribution",
+			Description: "the various latencies",
+			Buckets:     []float64{0, 10, 50, 100, 200, 400, 800, 1000, 1400, 2000, 5000, 10000, 15000},
+		},
+	}
+
+	distribution.Info.Record(&exporter.metrics, mLatency)
+
+	for {
+		sleep := randomSleep()
+		_, end := event.StartSpan(ctx, "main.randomSleep()")
+		time.Sleep(time.Duration(sleep) * time.Millisecond)
+		end()
+		event.Record(ctx, mLatency.Of(float64(sleep)))
+
+		fmt.Println("Latency: ", float64(sleep))
+	}
+}
+
+func randomSleep() int64 {
+	var max int64
+	switch modulus := time.Now().Unix() % 5; modulus {
+	case 0:
+		max = 17001
+	case 1:
+		max = 8007
+	case 2:
+		max = 917
+	case 3:
+		max = 87
+	case 4:
+		max = 1173
+	}
+	return rand.Int63n(max)
+}
+
+```
+3. Run the new file from within the tools repository:
+```bash
+go run internal/main.go
+```
+4. After about 5 seconds, OpenCensus should start receiving your new metrics, which you can see at http://localhost:8844/metrics. This page will look similar to the following:
+```
+# HELP promdemo_latencyDistribution the various latencies
+# TYPE promdemo_latencyDistribution histogram
+promdemo_latencyDistribution_bucket{vendor="otc",le="0"} 0
+promdemo_latencyDistribution_bucket{vendor="otc",le="10"} 2
+promdemo_latencyDistribution_bucket{vendor="otc",le="50"} 9
+promdemo_latencyDistribution_bucket{vendor="otc",le="100"} 22
+promdemo_latencyDistribution_bucket{vendor="otc",le="200"} 35
+promdemo_latencyDistribution_bucket{vendor="otc",le="400"} 49
+promdemo_latencyDistribution_bucket{vendor="otc",le="800"} 63
+promdemo_latencyDistribution_bucket{vendor="otc",le="1000"} 78
+promdemo_latencyDistribution_bucket{vendor="otc",le="1400"} 93
+promdemo_latencyDistribution_bucket{vendor="otc",le="2000"} 108
+promdemo_latencyDistribution_bucket{vendor="otc",le="5000"} 123
+promdemo_latencyDistribution_bucket{vendor="otc",le="10000"} 138
+promdemo_latencyDistribution_bucket{vendor="otc",le="15000"} 153
+promdemo_latencyDistribution_bucket{vendor="otc",le="+Inf"} 15
+promdemo_latencyDistribution_sum{vendor="otc"} 1641
+promdemo_latencyDistribution_count{vendor="otc"} 15
+```
+5. After a few more seconds, Prometheus should start displaying your new metrics. You can view the distribution at http://localhost:9445/graph?g0.range_input=5m&g0.stacked=1&g0.expr=rate(oragent_latencyDistribution_bucket%5B5m%5D)&g0.tab=0.
+
+6. Zipkin should also start displaying traces. You can view them at http://localhost:9444/zipkin/?limit=10&lookback=300000&serviceName=go-tools-test.

event/export/ocagent/metrics.go

@@ -0,0 +1,213 @@
+// Copyright 2019 The Go Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package ocagent
+
+import (
+	"time"
+
+	"github.com/qiniu/x/event/export/metric"
+	"github.com/qiniu/x/event/export/ocagent/wire"
+	"github.com/qiniu/x/event/label"
+)
+
+// dataToMetricDescriptor return a *wire.MetricDescriptor based on data.
+func dataToMetricDescriptor(data metric.Data) *wire.MetricDescriptor {
+	if data == nil {
+		return nil
+	}
+	descriptor := &wire.MetricDescriptor{
+		Name:        data.Handle(),
+		Description: getDescription(data),
+		// TODO: Unit?
+		Type:      dataToMetricDescriptorType(data),
+		LabelKeys: getLabelKeys(data),
+	}
+
+	return descriptor
+}
+
+// getDescription returns the description of data.
+func getDescription(data metric.Data) string {
+	switch d := data.(type) {
+	case *metric.Int64Data:
+		return d.Info.Description
+
+	case *metric.Float64Data:
+		return d.Info.Description
+
+	case *metric.HistogramInt64Data:
+		return d.Info.Description
+
+	case *metric.HistogramFloat64Data:
+		return d.Info.Description
+	}
+
+	return ""
+}
+
+// getLabelKeys returns a slice of *wire.LabelKeys based on the keys
+// in data.
+func getLabelKeys(data metric.Data) []*wire.LabelKey {
+	switch d := data.(type) {
+	case *metric.Int64Data:
+		return infoKeysToLabelKeys(d.Info.Keys)
+
+	case *metric.Float64Data:
+		return infoKeysToLabelKeys(d.Info.Keys)
+
+	case *metric.HistogramInt64Data:
+		return infoKeysToLabelKeys(d.Info.Keys)
+
+	case *metric.HistogramFloat64Data:
+		return infoKeysToLabelKeys(d.Info.Keys)
+	}
+
+	return nil
+}
+
+// dataToMetricDescriptorType returns a wire.MetricDescriptor_Type based on the
+// underlying type of data.
+func dataToMetricDescriptorType(data metric.Data) wire.MetricDescriptor_Type {
+	switch d := data.(type) {
+	case *metric.Int64Data:
+		if d.IsGauge {
+			return wire.MetricDescriptor_GAUGE_INT64
+		}
+		return wire.MetricDescriptor_CUMULATIVE_INT64
+
+	case *metric.Float64Data:
+		if d.IsGauge {
+			return wire.MetricDescriptor_GAUGE_DOUBLE
+		}
+		return wire.MetricDescriptor_CUMULATIVE_DOUBLE
+
+	case *metric.HistogramInt64Data:
+		return wire.MetricDescriptor_CUMULATIVE_DISTRIBUTION
+
+	case *metric.HistogramFloat64Data:
+		return wire.MetricDescriptor_CUMULATIVE_DISTRIBUTION
+	}
+
+	return wire.MetricDescriptor_UNSPECIFIED
+}
+
+// dataToTimeseries returns a slice of *wire.TimeSeries based on the
+// points in data.
+func dataToTimeseries(data metric.Data, start time.Time) []*wire.TimeSeries {
+	if data == nil {
+		return nil
+	}
+
+	numRows := numRows(data)
+	startTimestamp := convertTimestamp(start)
+	timeseries := make([]*wire.TimeSeries, 0, numRows)
+
+	for i := 0; i < numRows; i++ {
+		timeseries = append(timeseries, &wire.TimeSeries{
+			StartTimestamp: &startTimestamp,
+			// TODO: labels?
+			Points: dataToPoints(data, i),
+		})
+	}
+
+	return timeseries
+}
+
+// numRows returns the number of rows in data.
+func numRows(data metric.Data) int {
+	switch d := data.(type) {
+	case *metric.Int64Data:
+		return len(d.Rows)
+	case *metric.Float64Data:
+		return len(d.Rows)
+	case *metric.HistogramInt64Data:
+		return len(d.Rows)
+	case *metric.HistogramFloat64Data:
+		return len(d.Rows)
+	}
+
+	return 0
+}
+
+// dataToPoints returns an array of *wire.Points based on the point(s)
+// in data at index i.
+func dataToPoints(data metric.Data, i int) []*wire.Point {
+	switch d := data.(type) {
+	case *metric.Int64Data:
+		timestamp := convertTimestamp(d.EndTime)
+		return []*wire.Point{
+			{
+				Value: wire.PointInt64Value{
+					Int64Value: d.Rows[i],
+				},
+				Timestamp: &timestamp,
+			},
+		}
+	case *metric.Float64Data:
+		timestamp := convertTimestamp(d.EndTime)
+		return []*wire.Point{
+			{
+				Value: wire.PointDoubleValue{
+					DoubleValue: d.Rows[i],
+				},
+				Timestamp: &timestamp,
+			},
+		}
+	case *metric.HistogramInt64Data:
+		row := d.Rows[i]
+		bucketBounds := make([]float64, len(d.Info.Buckets))
+		for i, val := range d.Info.Buckets {
+			bucketBounds[i] = float64(val)
+		}
+		return distributionToPoints(row.Values, row.Count, float64(row.Sum), bucketBounds, d.EndTime)
+	case *metric.HistogramFloat64Data:
+		row := d.Rows[i]
+		return distributionToPoints(row.Values, row.Count, row.Sum, d.Info.Buckets, d.EndTime)
+	}
+
+	return nil
+}
+
+// distributionToPoints returns an array of *wire.Points containing a
+// wire.PointDistributionValue representing a distribution with the
+// supplied counts, count, and sum.
+func distributionToPoints(counts []int64, count int64, sum float64, bucketBounds []float64, end time.Time) []*wire.Point {
+	buckets := make([]*wire.Bucket, len(counts))
+	for i := 0; i < len(counts); i++ {
+		buckets[i] = &wire.Bucket{
+			Count: counts[i],
+		}
+	}
+	timestamp := convertTimestamp(end)
+	return []*wire.Point{
+		{
+			Value: wire.PointDistributionValue{
+				DistributionValue: &wire.DistributionValue{
+					Count: count,
+					Sum:   sum,
+					// TODO: SumOfSquaredDeviation?
+					Buckets: buckets,
+					BucketOptions: &wire.BucketOptionsExplicit{
+						Bounds: bucketBounds,
+					},
+				},
+			},
+			Timestamp: &timestamp,
+		},
+	}
+}
+
+// infoKeysToLabelKeys returns an array of *wire.LabelKeys containing the
+// string values of the elements of labelKeys.
+func infoKeysToLabelKeys(infoKeys []label.Key) []*wire.LabelKey {
+	labelKeys := make([]*wire.LabelKey, 0, len(infoKeys))
+	for _, key := range infoKeys {
+		labelKeys = append(labelKeys, &wire.LabelKey{
+			Key: key.Name(),
+		})
+	}
+
+	return labelKeys
+}