From be03d4f470ed7887405ab93f751a8591aef9b6f9 Mon Sep 17 00:00:00 2001 From: Nathan Dias Date: Wed, 23 Oct 2019 19:34:23 -0500 Subject: [PATCH] internal/telemetry/export/ocagent: attach timestamps to metrics This change attaches start timestamps to timeseries and end timestamps to the points in each timeseries. Int64Data, Float64Data, HistogramInt64Data, and HistogramFloat64Data have also had an EndTime field added to keep track of the last time the metric was updated. What works: * Start and end timestamps will now be attached to timeseries. What does not work yet: * MetricDescriptors will not have a unit attached. * No labels will be attached to timeseries. * Distributions will not have SumOfSquaredDeviation attached. Updates golang/go#33819 Change-Id: I692e1676bb1e31de26c1f799b96428fc9a55d6c7 Reviewed-on: https://go-review.googlesource.com/c/tools/+/203060 Run-TryBot: Emmanuel Odeke TryBot-Result: Gobot Gobot Reviewed-by: Emmanuel Odeke --- internal/telemetry/export/ocagent/metrics.go | 20 ++++-- .../telemetry/export/ocagent/metrics_test.go | 72 ++++++++++++++++--- internal/telemetry/export/ocagent/ocagent.go | 6 +- internal/telemetry/metric/metric.go | 43 ++++++----- 4 files changed, 102 insertions(+), 39 deletions(-) diff --git a/internal/telemetry/export/ocagent/metrics.go b/internal/telemetry/export/ocagent/metrics.go index 11c2a83a2d..6cb824df6c 100644 --- a/internal/telemetry/export/ocagent/metrics.go +++ b/internal/telemetry/export/ocagent/metrics.go @@ -6,6 +6,7 @@ package ocagent import ( "fmt" + "time" "golang.org/x/tools/internal/telemetry" "golang.org/x/tools/internal/telemetry/export/ocagent/wire" @@ -95,17 +96,18 @@ func dataToMetricDescriptorType(data telemetry.MetricData) wire.MetricDescriptor // dataToTimeseries returns a slice of *wire.TimeSeries based on the // points in data. -func dataToTimeseries(data telemetry.MetricData) []*wire.TimeSeries { +func dataToTimeseries(data telemetry.MetricData, 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{ - // TODO: attach StartTimestamp + StartTimestamp: &startTimestamp, // TODO: labels? Points: dataToPoints(data, i), }) @@ -135,21 +137,23 @@ func numRows(data telemetry.MetricData) int { func dataToPoints(data telemetry.MetricData, 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], }, - // TODO: attach Timestamp + Timestamp: ×tamp, }, } case *metric.Float64Data: + timestamp := convertTimestamp(*d.EndTime) return []*wire.Point{ { Value: wire.PointDoubleValue{ DoubleValue: d.Rows[i], }, - // TODO: attach Timestamp + Timestamp: ×tamp, }, } case *metric.HistogramInt64Data: @@ -158,10 +162,10 @@ func dataToPoints(data telemetry.MetricData, i int) []*wire.Point { for i, val := range d.Info.Buckets { bucketBounds[i] = float64(val) } - return distributionToPoints(row.Values, row.Count, float64(row.Sum), bucketBounds) + 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) + return distributionToPoints(row.Values, row.Count, row.Sum, d.Info.Buckets, *d.EndTime) } return nil @@ -170,13 +174,14 @@ func dataToPoints(data telemetry.MetricData, i int) []*wire.Point { // 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) []*wire.Point { +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{ @@ -190,6 +195,7 @@ func distributionToPoints(counts []int64, count int64, sum float64, bucketBounds }, }, }, + Timestamp: ×tamp, }, } } diff --git a/internal/telemetry/export/ocagent/metrics_test.go b/internal/telemetry/export/ocagent/metrics_test.go index 85e90dd512..033a60ed9b 100644 --- a/internal/telemetry/export/ocagent/metrics_test.go +++ b/internal/telemetry/export/ocagent/metrics_test.go @@ -3,6 +3,7 @@ package ocagent import ( "reflect" "testing" + "time" "golang.org/x/tools/internal/telemetry" "golang.org/x/tools/internal/telemetry/export/ocagent/wire" @@ -318,14 +319,22 @@ func TestDataToMetricDescriptorType(t *testing.T) { } func TestDataToTimeseries(t *testing.T) { + epoch := time.Unix(0, 0) + epochTimestamp := convertTimestamp(epoch) + + end := time.Unix(30, 0) + endTimestamp := convertTimestamp(end) + tests := []struct { - name string - data telemetry.MetricData - want []*wire.TimeSeries + name string + data telemetry.MetricData + start time.Time + want []*wire.TimeSeries }{ { "nil data", nil, + time.Time{}, nil, }, { @@ -336,28 +345,36 @@ func TestDataToTimeseries(t *testing.T) { 2, 3, }, + EndTime: &end, }, + epoch, []*wire.TimeSeries{ &wire.TimeSeries{ Points: []*wire.Point{ &wire.Point{ - Value: wire.PointInt64Value{Int64Value: 1}, + Value: wire.PointInt64Value{Int64Value: 1}, + Timestamp: &endTimestamp, }, }, + StartTimestamp: &epochTimestamp, }, &wire.TimeSeries{ Points: []*wire.Point{ &wire.Point{ - Value: wire.PointInt64Value{Int64Value: 2}, + Value: wire.PointInt64Value{Int64Value: 2}, + Timestamp: &endTimestamp, }, }, + StartTimestamp: &epochTimestamp, }, &wire.TimeSeries{ Points: []*wire.Point{ &wire.Point{ - Value: wire.PointInt64Value{Int64Value: 3}, + Value: wire.PointInt64Value{Int64Value: 3}, + Timestamp: &endTimestamp, }, }, + StartTimestamp: &epochTimestamp, }, }, }, @@ -368,21 +385,27 @@ func TestDataToTimeseries(t *testing.T) { 1.5, 4.5, }, + EndTime: &end, }, + epoch, []*wire.TimeSeries{ &wire.TimeSeries{ Points: []*wire.Point{ &wire.Point{ - Value: wire.PointDoubleValue{DoubleValue: 1.5}, + Value: wire.PointDoubleValue{DoubleValue: 1.5}, + Timestamp: &endTimestamp, }, }, + StartTimestamp: &epochTimestamp, }, &wire.TimeSeries{ Points: []*wire.Point{ &wire.Point{ - Value: wire.PointDoubleValue{DoubleValue: 4.5}, + Value: wire.PointDoubleValue{DoubleValue: 4.5}, + Timestamp: &endTimestamp, }, }, + StartTimestamp: &epochTimestamp, }, }, }, @@ -405,7 +428,9 @@ func TestDataToTimeseries(t *testing.T) { 0, 5, 10, }, }, + EndTime: &end, }, + epoch, []*wire.TimeSeries{ &wire.TimeSeries{ Points: []*wire.Point{ @@ -432,8 +457,10 @@ func TestDataToTimeseries(t *testing.T) { }, }, }, + Timestamp: &endTimestamp, }, }, + StartTimestamp: &epochTimestamp, }, }, }, @@ -455,7 +482,9 @@ func TestDataToTimeseries(t *testing.T) { 0, 5, }, }, + EndTime: &end, }, + epoch, []*wire.TimeSeries{ &wire.TimeSeries{ Points: []*wire.Point{ @@ -479,8 +508,10 @@ func TestDataToTimeseries(t *testing.T) { }, }, }, + Timestamp: &endTimestamp, }, }, + StartTimestamp: &epochTimestamp, }, }, }, @@ -488,7 +519,7 @@ func TestDataToTimeseries(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - got := dataToTimeseries(tt.data) + got := dataToTimeseries(tt.data, tt.start) if !reflect.DeepEqual(got, tt.want) { t.Fatalf("Got:\n%s\nWant:\n%s", marshaled(got), marshaled(tt.want)) } @@ -559,11 +590,15 @@ func TestNumRows(t *testing.T) { } func TestDataToPoints(t *testing.T) { + end := time.Unix(30, 0) + endTimestamp := convertTimestamp(end) + int64Data := &metric.Int64Data{ Rows: []int64{ 0, 10, }, + EndTime: &end, } float64Data := &metric.Float64Data{ @@ -571,6 +606,7 @@ func TestDataToPoints(t *testing.T) { 0.5, 0.25, }, + EndTime: &end, } histogramInt64Data := &metric.HistogramInt64Data{ @@ -599,6 +635,7 @@ func TestDataToPoints(t *testing.T) { 0, 5, 10, }, }, + EndTime: &end, } histogramFloat64Data := &metric.HistogramFloat64Data{ @@ -627,6 +664,7 @@ func TestDataToPoints(t *testing.T) { 0, 5, 10, }, }, + EndTime: &end, } tests := []struct { @@ -650,6 +688,7 @@ func TestDataToPoints(t *testing.T) { Value: wire.PointInt64Value{ Int64Value: 0, }, + Timestamp: &endTimestamp, }, }, }, @@ -662,6 +701,7 @@ func TestDataToPoints(t *testing.T) { Value: wire.PointInt64Value{ Int64Value: 10, }, + Timestamp: &endTimestamp, }, }, }, @@ -674,6 +714,7 @@ func TestDataToPoints(t *testing.T) { Value: wire.PointDoubleValue{ DoubleValue: 0.5, }, + Timestamp: &endTimestamp, }, }, }, @@ -686,6 +727,7 @@ func TestDataToPoints(t *testing.T) { Value: wire.PointDoubleValue{ DoubleValue: 0.25, }, + Timestamp: &endTimestamp, }, }, }, @@ -717,6 +759,7 @@ func TestDataToPoints(t *testing.T) { }, }, }, + Timestamp: &endTimestamp, }, }, }, @@ -748,6 +791,7 @@ func TestDataToPoints(t *testing.T) { }, }, }, + Timestamp: &endTimestamp, }, }, }, @@ -779,6 +823,7 @@ func TestDataToPoints(t *testing.T) { }, }, }, + Timestamp: &endTimestamp, }, }, }, @@ -810,6 +855,7 @@ func TestDataToPoints(t *testing.T) { }, }, }, + Timestamp: &endTimestamp, }, }, }, @@ -826,12 +872,16 @@ func TestDataToPoints(t *testing.T) { } func TestDistributionToPoints(t *testing.T) { + end := time.Unix(30, 0) + endTimestamp := convertTimestamp(end) + tests := []struct { name string counts []int64 count int64 sum float64 buckets []float64 + end time.Time want []*wire.Point }{ { @@ -846,6 +896,7 @@ func TestDistributionToPoints(t *testing.T) { buckets: []float64{ 0, 5, 10, }, + end: end, want: []*wire.Point{ { Value: wire.PointDistributionValue{ @@ -871,6 +922,7 @@ func TestDistributionToPoints(t *testing.T) { }, }, }, + Timestamp: &endTimestamp, }, }, }, @@ -878,7 +930,7 @@ func TestDistributionToPoints(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - got := distributionToPoints(tt.counts, tt.count, tt.sum, tt.buckets) + got := distributionToPoints(tt.counts, tt.count, tt.sum, tt.buckets, tt.end) if !reflect.DeepEqual(got, tt.want) { t.Fatalf("Got:\n%s\nWant:\n%s", marshaled(got), marshaled(tt.want)) } diff --git a/internal/telemetry/export/ocagent/ocagent.go b/internal/telemetry/export/ocagent/ocagent.go index d1ebfd2c43..cad7e13380 100644 --- a/internal/telemetry/export/ocagent/ocagent.go +++ b/internal/telemetry/export/ocagent/ocagent.go @@ -114,7 +114,7 @@ func (e *exporter) Log(context.Context, telemetry.Event) {} func (e *exporter) Metric(ctx context.Context, data telemetry.MetricData) { e.mu.Lock() defer e.mu.Unlock() - e.metrics = append(e.metrics, convertMetric(data)) + e.metrics = append(e.metrics, convertMetric(data, e.config.Start)) } func (e *exporter) Flush() { @@ -202,9 +202,9 @@ func convertSpan(span *telemetry.Span) *wire.Span { return result } -func convertMetric(data telemetry.MetricData) *wire.Metric { +func convertMetric(data telemetry.MetricData, start time.Time) *wire.Metric { descriptor := dataToMetricDescriptor(data) - timeseries := dataToTimeseries(data) + timeseries := dataToTimeseries(data, start) if descriptor == nil && timeseries == nil { return nil diff --git a/internal/telemetry/metric/metric.go b/internal/telemetry/metric/metric.go index 0333450d90..98f0bd8136 100644 --- a/internal/telemetry/metric/metric.go +++ b/internal/telemetry/metric/metric.go @@ -131,6 +131,8 @@ type Int64Data struct { IsGauge bool // Rows holds the per group values for the metric. Rows []int64 + // End is the last time this metric was updated. + EndTime *time.Time groups []telemetry.TagList } @@ -143,6 +145,8 @@ type Float64Data struct { IsGauge bool // Rows holds the per group values for the metric. Rows []float64 + // End is the last time this metric was updated. + EndTime *time.Time groups []telemetry.TagList } @@ -153,6 +157,8 @@ type HistogramInt64Data struct { Info *HistogramInt64 // Rows holds the per group values for the metric. Rows []*HistogramInt64Row + // End is the last time this metric was updated. + EndTime *time.Time groups []telemetry.TagList } @@ -177,6 +183,8 @@ type HistogramFloat64Data struct { Info *HistogramFloat64 // Rows holds the per group values for the metric. Rows []*HistogramFloat64Row + // End is the last time this metric was updated. + EndTime *time.Time groups []telemetry.TagList } @@ -215,7 +223,7 @@ func getGroup(ctx context.Context, g *[]telemetry.TagList, keys []interface{}) ( func (data *Int64Data) Handle() string { return data.Info.Name } func (data *Int64Data) Groups() []telemetry.TagList { return data.groups } -func (data *Int64Data) modify(ctx context.Context, f func(v int64) int64) { +func (data *Int64Data) modify(ctx context.Context, at time.Time, f func(v int64) int64) { index, insert := getGroup(ctx, &data.groups, data.Info.Keys) old := data.Rows if insert { @@ -227,34 +235,31 @@ func (data *Int64Data) modify(ctx context.Context, f func(v int64) int64) { copy(data.Rows, old) } data.Rows[index] = f(data.Rows[index]) + data.EndTime = &at frozen := *data export.Metric(ctx, &frozen) } func (data *Int64Data) countInt64(ctx context.Context, measure *stats.Int64Measure, value int64, at time.Time) { - // TODO: Use at. - data.modify(ctx, func(v int64) int64 { return v + 1 }) + data.modify(ctx, at, func(v int64) int64 { return v + 1 }) } func (data *Int64Data) countFloat64(ctx context.Context, measure *stats.Float64Measure, value float64, at time.Time) { - // TODO: Use at. - data.modify(ctx, func(v int64) int64 { return v + 1 }) + data.modify(ctx, at, func(v int64) int64 { return v + 1 }) } func (data *Int64Data) sum(ctx context.Context, measure *stats.Int64Measure, value int64, at time.Time) { - // TODO: Use at. - data.modify(ctx, func(v int64) int64 { return v + value }) + data.modify(ctx, at, func(v int64) int64 { return v + value }) } func (data *Int64Data) latest(ctx context.Context, measure *stats.Int64Measure, value int64, at time.Time) { - // TODO: Use at. - data.modify(ctx, func(v int64) int64 { return value }) + data.modify(ctx, at, func(v int64) int64 { return value }) } func (data *Float64Data) Handle() string { return data.Info.Name } func (data *Float64Data) Groups() []telemetry.TagList { return data.groups } -func (data *Float64Data) modify(ctx context.Context, f func(v float64) float64) { +func (data *Float64Data) modify(ctx context.Context, at time.Time, f func(v float64) float64) { index, insert := getGroup(ctx, &data.groups, data.Info.Keys) old := data.Rows if insert { @@ -266,23 +271,23 @@ func (data *Float64Data) modify(ctx context.Context, f func(v float64) float64) copy(data.Rows, old) } data.Rows[index] = f(data.Rows[index]) + data.EndTime = &at frozen := *data export.Metric(ctx, &frozen) } func (data *Float64Data) sum(ctx context.Context, measure *stats.Float64Measure, value float64, at time.Time) { - data.modify(ctx, func(v float64) float64 { return v + value }) + data.modify(ctx, at, func(v float64) float64 { return v + value }) } func (data *Float64Data) latest(ctx context.Context, measure *stats.Float64Measure, value float64, at time.Time) { - // TODO: Use at. - data.modify(ctx, func(v float64) float64 { return value }) + data.modify(ctx, at, func(v float64) float64 { return value }) } func (data *HistogramInt64Data) Handle() string { return data.Info.Name } func (data *HistogramInt64Data) Groups() []telemetry.TagList { return data.groups } -func (data *HistogramInt64Data) modify(ctx context.Context, f func(v *HistogramInt64Row)) { +func (data *HistogramInt64Data) modify(ctx context.Context, at time.Time, f func(v *HistogramInt64Row)) { index, insert := getGroup(ctx, &data.groups, data.Info.Keys) old := data.Rows var v HistogramInt64Row @@ -300,13 +305,13 @@ func (data *HistogramInt64Data) modify(ctx context.Context, f func(v *HistogramI copy(v.Values, oldValues) f(&v) data.Rows[index] = &v + data.EndTime = &at frozen := *data export.Metric(ctx, &frozen) } func (data *HistogramInt64Data) record(ctx context.Context, measure *stats.Int64Measure, value int64, at time.Time) { - // TODO: Use at. - data.modify(ctx, func(v *HistogramInt64Row) { + data.modify(ctx, at, func(v *HistogramInt64Row) { v.Sum += value if v.Min > value || v.Count == 0 { v.Min = value @@ -326,7 +331,7 @@ func (data *HistogramInt64Data) record(ctx context.Context, measure *stats.Int64 func (data *HistogramFloat64Data) Handle() string { return data.Info.Name } func (data *HistogramFloat64Data) Groups() []telemetry.TagList { return data.groups } -func (data *HistogramFloat64Data) modify(ctx context.Context, f func(v *HistogramFloat64Row)) { +func (data *HistogramFloat64Data) modify(ctx context.Context, at time.Time, f func(v *HistogramFloat64Row)) { index, insert := getGroup(ctx, &data.groups, data.Info.Keys) old := data.Rows var v HistogramFloat64Row @@ -344,13 +349,13 @@ func (data *HistogramFloat64Data) modify(ctx context.Context, f func(v *Histogra copy(v.Values, oldValues) f(&v) data.Rows[index] = &v + data.EndTime = &at frozen := *data export.Metric(ctx, &frozen) } func (data *HistogramFloat64Data) record(ctx context.Context, measure *stats.Float64Measure, value float64, at time.Time) { - // TODO: Use at. - data.modify(ctx, func(v *HistogramFloat64Row) { + data.modify(ctx, at, func(v *HistogramFloat64Row) { v.Sum += value if v.Min > value || v.Count == 0 { v.Min = value