Metrics stdout export pipeline

example/basic/go.sum

@@ -1,5 +1,6 @@
 cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
+github.com/DataDog/sketches-go v0.0.0-20190923095040-43f19ad77ff7 h1:qELHH0AWCvf98Yf+CNIJx9vOZOfHFDDzgDRYsnNk/vs=
 github.com/DataDog/sketches-go v0.0.0-20190923095040-43f19ad77ff7/go.mod h1:Q5DbzQ+3AkgGwymQO7aZFNP7ns2lZKGtvRBzRXfdi60=
 github.com/OneOfOne/xxhash v1.2.2/go.mod h1:HSdplMjZKSmBqAxg5vPj2TmRDmfkzw+cTzAElWljhcU=
 github.com/OpenPeeDeeP/depguard v1.0.1/go.mod h1:xsIw86fROiiwelg+jB2uM9PiKihMMmUx/1V+TNhjQvM=
@@ -21,6 +22,7 @@ github.com/coreos/pkg v0.0.0-20180928190104-399ea9e2e55f/go.mod h1:E3G3o1h8I7cfc
 github.com/cpuguy83/go-md2man v1.0.10/go.mod h1:SmD6nW6nTyfqj6ABTjUi3V3JVMnlJmwcJI5acqYI6dE=
 github.com/creack/pty v1.1.7/go.mod h1:lj5s0c3V2DBrqTV7llrYr5NG6My20zk30Fl46Y7DoTY=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/dgrijalva/jwt-go v3.2.0+incompatible/go.mod h1:E3ru+11k8xSBh+hMPgOLZmtrrCbhqsmaPHjLKYnJCaQ=
 github.com/dgryski/go-sip13 v0.0.0-20181026042036-e10d5fee7954/go.mod h1:vAd38F8PWV+bWy6jNmig1y/TA+kYO4g3RSRF0IAv0no=
@@ -82,6 +84,7 @@ github.com/google/btree v1.0.0/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ
 github.com/google/go-cmp v0.2.0/go.mod h1:oXzfMopK8JAjlY9xF4vHSVASa0yLyX7SntLO5aqRK0M=
 github.com/google/go-cmp v0.3.1 h1:Xye71clBPdm5HgqGwUkwhbynsUJZhDbS20FvLhQ2izg=
 github.com/google/go-cmp v0.3.1/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
+github.com/google/gofuzz v1.0.0 h1:A8PeW59pxE9IoFRqBp37U+mSNaQoZ46F1f0f863XSXw=
 github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
 github.com/google/renameio v0.1.0/go.mod h1:KWCgfxg9yswjAJkECMjeO8J8rahYeXnNhOm40UhjYkI=
 github.com/gorilla/websocket v1.4.0/go.mod h1:E7qHFY5m1UJ88s3WnNqhKjPHQ0heANvMoAMk2YaljkQ=
@@ -90,6 +93,7 @@ github.com/gostaticanalysis/analysisutil v0.0.3/go.mod h1:eEOZF4jCKGi+aprrirO9e7
 github.com/grpc-ecosystem/go-grpc-middleware v1.0.0/go.mod h1:FiyG127CGDf3tlThmgyCl78X/SZQqEOJBCDaAfeWzPs=
 github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0/go.mod h1:8NvIoxWQoOIhqOTXgfV/d3M/q6VIi02HzZEHgUlZvzk=
 github.com/grpc-ecosystem/grpc-gateway v1.9.0/go.mod h1:vNeuVxBJEsws4ogUvrchl83t/GYV9WGTSLVdBhOQFDY=
+github.com/hashicorp/golang-lru v0.5.3 h1:YPkqC67at8FYaadspW/6uE0COsBxS2656RLEr8Bppgk=
 github.com/hashicorp/golang-lru v0.5.3/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4=
 github.com/hashicorp/hcl v1.0.0/go.mod h1:E5yfLk+7swimpb2L/Alb/PJmXilQ/rhwaUYs4T20WEQ=
 github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU=
@@ -106,9 +110,11 @@ github.com/klauspost/cpuid v1.2.0/go.mod h1:Pj4uuM528wm8OyEC2QMXAi2YiTZ96dNQPGgo
 github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
 github.com/konsorten/go-windows-terminal-sequences v1.0.2/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
 github.com/kr/logfmt v0.0.0-20140226030751-b84e30acd515/go.mod h1:+0opPa2QZZtGFBFZlji/RkVcI2GknAs/DXo4wKdlNEc=
+github.com/kr/pretty v0.1.0 h1:L/CwN0zerZDmRFUapSPitk6f+Q3+0za1rQkzVuMiMFI=
 github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
 github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
 github.com/kr/pty v1.1.8/go.mod h1:O1sed60cT9XZ5uDucP5qwvh+TE3NnUj51EiZO/lmSfw=
+github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
 github.com/lib/pq v1.2.0/go.mod h1:5WUZQaWbwv1U+lTReE5YruASi9Al49XbQIvNi/34Woo=
 github.com/logrusorgru/aurora v0.0.0-20181002194514-a7b3b318ed4e/go.mod h1:7rIyQOR62GCctdiQpZ/zOJlFyk6y+94wXzv6RNZgaR4=
@@ -135,6 +141,7 @@ github.com/pelletier/go-toml v1.2.0/go.mod h1:5z9KED0ma1S8pY6P1sdut58dfprrGBbd/9
 github.com/pelletier/go-toml v1.5.0/go.mod h1:5N711Q9dKgbdkxHL+MEfF31hpT7l0S0s/t2kKREewys=
 github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/prometheus/client_golang v0.9.1/go.mod h1:7SWBe2y4D6OKWSNQJUaRYU/AaXPKyh/dDVn+NZz0KFw=
 github.com/prometheus/client_golang v0.9.3/go.mod h1:/TN21ttK/J9q6uSwhBd54HahCDft0ttaMvbicHlPoso=
@@ -176,6 +183,7 @@ github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+
 github.com/stretchr/objx v0.2.0/go.mod h1:qt09Ya8vawLte6SNmTgCsAVtYtaKzEcn8ATUoHMkEqE=
 github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
+github.com/stretchr/testify v1.4.0 h1:2E4SXV/wtOkTonXsotYi4li6zVWxYlZuYNCXe9XRJyk=
 github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
 github.com/timakin/bodyclose v0.0.0-20190930140734-f7f2e9bca95e/go.mod h1:Qimiffbc6q9tBWlVV6x0P9sat/ao1xEkREYPPj9hphk=
 github.com/tmc/grpc-websocket-proxy v0.0.0-20190109142713-0ad062ec5ee5/go.mod h1:ncp9v5uamzpCO7NfCPTXjqaC+bZgJeR0sMTm6dMHP7U=
@@ -273,6 +281,7 @@ google.golang.org/grpc v1.24.0/go.mod h1:XDChyiUovWa60DnaeDeZmSW86xtLtjtZbwvSiRn
 gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 h1:YR8cESwS4TdDjEe65xsg0ogRM/Nc3DYOhEAlW+xobZo=
 gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/errgo.v2 v2.1.0/go.mod h1:hNsd1EY+bozCKY1Ytp96fpM3vjJbqLJn88ws8XvfDNI=
 gopkg.in/fsnotify.v1 v1.4.7/go.mod h1:Tz8NjZHkW78fSQdbUxIjBTcgA1z1m8ZHf0WmKUhAMys=
@@ -281,6 +290,7 @@ gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7/go.mod h1:dt/ZhP58zS4L8KSrWD
 gopkg.in/yaml.v2 v2.0.0-20170812160011-eb3733d160e7/go.mod h1:JAlM8MvJe8wmxCU4Bli9HhUf9+ttbYbLASfIpnQbh74=
 gopkg.in/yaml.v2 v2.2.1/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
+gopkg.in/yaml.v2 v2.2.4 h1:/eiJrUcujPVeJ3xlSWaiNi3uSVmDGBK1pDHUHAnao1I=
 gopkg.in/yaml.v2 v2.2.4/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
 honnef.co/go/tools v0.0.0-20190523083050-ea95bdfd59fc/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=

example/basic/main.go

@@ -16,17 +16,21 @@ package main
 
 import (
 	"context"
+	"time"
 
 	"go.opentelemetry.io/api/distributedcontext"
 	"go.opentelemetry.io/api/key"
 	"go.opentelemetry.io/api/metric"
 	"go.opentelemetry.io/api/trace"
+	"go.opentelemetry.io/exporter/metric/stdout"
 	"go.opentelemetry.io/global"
+	"go.opentelemetry.io/sdk/metric/batcher/stateful"
+	"go.opentelemetry.io/sdk/metric/controller/push"
+	"go.opentelemetry.io/sdk/metric/selector/simple"
 )
 
 var (
 	tracer = global.TraceProvider().GetTracer("ex.com/basic")
-	meter  = global.MeterProvider().GetMeter("ex.com/basic") // TODO: should share resources ^^^?
 
 	fooKey     = key.New("ex.com/foo")
 	barKey     = key.New("ex.com/bar")
@@ -35,6 +39,19 @@ var (
 )
 
 func main() {
+	selector := simple.New()
+	batcher := stateful.New(selector)
+	exporter := stdout.New(stdout.Options{PrettyPrint: true})
+	pusher := push.New(batcher, exporter, time.Second)
+	pusher.Start()
+	defer pusher.Stop()
+
+	global.SetMeterProvider(pusher)
+
+	// Note: Have to get the meter after the global is
+	// initialized.  See OTEP 0005.
+	meter := global.MeterProvider().GetMeter("ex.com/basic")
+
 	oneMetric := meter.NewFloat64Gauge("ex.com.one",
 		metric.WithKeys(fooKey, barKey, lemonsKey),
 		metric.WithDescription("A gauge set to 1.0"),
@@ -91,7 +108,4 @@ func main() {
 	if err != nil {
 		panic(err)
 	}
-
-	// TODO: How to flush?
-	// loader.Flush()
 }

exporter/metric/stdout/stdout.go

@@ -0,0 +1,138 @@
+// Copyright 2019, OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package stdout
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"os"
+	"strings"
+	"time"
+
+	"go.opentelemetry.io/api/core"
+	"go.opentelemetry.io/sdk/export"
+	"go.opentelemetry.io/sdk/metric/aggregator"
+)
+
+type Exporter struct {
+	options Options
+}
+
+// Options are the options to be used when initializing a stdout export.
+type Options struct {
+	// File is the destination.  If not set, os.Stdout is used.
+	File *os.File
+
+	// PrettyPrint will pretty the json representation of the span,
+	// making it print "pretty". Default is false.
+	PrettyPrint bool
+
+	// Quantiles are the desired aggregation quantiles for measure
+	// metric data, used when the configured aggregator supports
+	// quantiles.
+	Quantiles []float64
+}
+
+type expoBatch struct {
+	Timestamp time.Time  `json:"time,omitempty"`
+	Updates   []expoLine `json:"updates,omitempty"`
+}
+
+type expoLine struct {
+	Name      string      `json:"name"`
+	Max       interface{} `json:"max,omitempty"`
+	Sum       interface{} `json:"sum,omitempty"`
+	Count     interface{} `json:"count,omitempty"`
+	LastValue interface{} `json:"last,omitempty"`
+	Timestamp time.Time   `json:"time,omitempty"`
+}
+
+var _ export.MetricExporter = &Exporter{}
+
+func New(options Options) *Exporter {
+	if options.File == nil {
+		options.File = os.Stdout
+	}
+	return &Exporter{
+		options: options,
+	}
+}
+
+func (e *Exporter) Export(_ context.Context, producer export.MetricProducer) {
+	var batch expoBatch
+	producer.Foreach(func(agg export.MetricAggregator, desc *export.Descriptor, labelValues []core.Value) {
+		var expose expoLine
+		if sum, ok := agg.(aggregator.Sum); ok {
+			expose.Sum = sum.Sum().Emit(desc.NumberKind())
+
+		} else if lv, ok := agg.(aggregator.LastValue); ok {
+			expose.LastValue = lv.LastValue().Emit(desc.NumberKind())
+			expose.Timestamp = lv.Timestamp()
+
+		} else if msc, ok := agg.(aggregator.MaxSumCount); ok {
+			expose.Max = msc.Max().Emit(desc.NumberKind())
+			expose.Sum = msc.Sum().Emit(desc.NumberKind())
+			expose.Count = msc.Count().Emit(desc.NumberKind())
+
+		} else if dist, ok := agg.(aggregator.Distribution); ok {
+			expose.Max = dist.Max().Emit(desc.NumberKind())
+			expose.Sum = dist.Sum().Emit(desc.NumberKind())
+			expose.Count = dist.Count().Emit(desc.NumberKind())
+
+			// TODO print one configured quantile per line
+		}
+
+		var sb strings.Builder
+
+		sb.WriteString(desc.Name())
+
+		if len(desc.Keys()) > 0 {
+			sb.WriteRune('{')
+		}
+
+		for i, k := range desc.Keys() {
+			if i > 0 {
+				sb.WriteRune(',')
+			}
+			sb.WriteString(string(k))
+			sb.WriteRune('=')
+			sb.WriteString(labelValues[i].Emit())
+		}
+
+		if len(desc.Keys()) > 0 {
+			sb.WriteRune('}')
+		}
+
+		expose.Name = sb.String()
+
+		batch.Updates = append(batch.Updates, expose)
+	})
+
+	var data []byte
+	var err error
+	if e.options.PrettyPrint {
+		data, err = json.MarshalIndent(batch, "", "\t")
+	} else {
+		data, err = json.Marshal(batch)
+	}
+
+	if err != nil {
+		fmt.Fprintf(e.options.File, "JSON encode error: %v\n", err)
+		return
+	}
+
+	fmt.Fprintln(e.options.File, string(data))
+}

sdk/export/exporter.go

@@ -46,18 +46,18 @@ type SpanBatcher interface {
 // Multiple-exporters could be implemented by implementing this interface
 // for a group of MetricBatcher.
 type MetricBatcher interface {
-	// AggregatorFor should return the kind of aggregator
-	// suited to the requested export.  Returning `nil`
-	// indicates to ignore the metric update.
-	//
-	// Note: This is context-free because the handle should not be
-	// bound to the incoming context.  This call should not block.
-	AggregatorFor(MetricRecord) MetricAggregator
+	// MetricAggregationSelector is responsible for selecting the
+	// concrete type of aggregation used for a metric in the SDK.
+	MetricAggregationSelector
 
-	// Export receives pairs of records and aggregators
+	// Process receives pairs of records and aggregators
 	// during the SDK Collect().  Exporter implementations
 	// must access the specific aggregator to receive the
 	// exporter data, since the format of the data varies
 	// by aggregation.
-	Export(context.Context, MetricRecord, MetricAggregator)
+	Process(context.Context, MetricRecord, MetricAggregator)
+
+	// ReadCheckpoint is the interface used by exporters to access
+	// aggregate checkpoints after collection.
+	ReadCheckpoint() MetricProducer
 }

sdk/export/metric.go

@@ -21,6 +21,20 @@ import (
 	"go.opentelemetry.io/api/unit"
 )
 
+// MetricAggregationSelector supports selecting the kind of aggregator
+// to use at runtime for a specific metric instrument.
+type MetricAggregationSelector interface {
+	// AggregatorFor should return the kind of aggregator suited
+	// to the requested export.  Returning `nil` indicates to
+	// ignore this metric instrument.  Although it is not
+	// required, this should return a consistent type to avoid
+	// confusion in later stages of the metrics export process.
+	//
+	// Note: This is context-free because the handle should not be
+	// bound to the incoming context.  This call should not block.
+	AggregatorFor(MetricRecord) MetricAggregator
+}
+
 // MetricAggregator implements a specific aggregation behavior, e.g.,
 // a counter, a gauge, a histogram.
 type MetricAggregator interface {
@@ -49,6 +63,19 @@ type MetricRecord interface {
 	Labels() []core.KeyValue
 }
 
+// MetricExporter handles presentation of the checkpoint of aggregate
+// metrics.  This is the final stage of a metrics export pipeline,
+// where metric data are formatted for a specific system.
+type MetricExporter interface {
+	Export(context.Context, MetricProducer)
+}
+
+// MetricProducer allows a MetricExporter to access a checkpoint of
+// aggregated metrics one at a time.
+type MetricProducer interface {
+	Foreach(func(MetricAggregator, *Descriptor, []core.Value))
+}
+
 // MetricKind describes the kind of instrument.
 type MetricKind int8
 

sdk/metric/aggregator/api.go

@@ -0,0 +1,58 @@
+// Copyright 2019, OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package aggregator
+
+import (
+	"time"
+
+	"go.opentelemetry.io/api/core"
+)
+
+// TODO: Add Min() support to maxsumcount?  It's the same as
+// Quantile(0) but cheap to compute like Max().
+
+type (
+	Sum interface {
+		Sum() core.Number
+	}
+
+	Count interface {
+		Count() core.Number
+	}
+
+	Max interface {
+		Max() core.Number
+	}
+
+	Quantile interface {
+		Quantile() core.Number
+	}
+
+	LastValue interface {
+		LastValue() core.Number
+		Timestamp() time.Time
+	}
+
+	MaxSumCount interface {
+		Sum
+		Count
+		Max
+	}
+
+	Distribution interface {
+		MaxSumCount
+		Quantile
+	}
+)