Merge 5fc09ea into 3b55ac7

Author: jmacd

CHANGELOG.md

@@ -8,6 +8,8 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 
 ## Unreleased
 
+- Add Exemplar support in the Lightstep Metrics SDK. [#576](https://github.com/lightstep/otel-launcher-go/pull/576)
+
 ## [1.23.0](https://github.com/lightstep/otel-launcher-go/releases/tag/v1.23.0) - 2023-12-20)
 
 - Update to OTel-Arrow v0.13.0 dependencies. [#591](https://github.com/lightstep/otel-launcher-go/pull/591)

Makefile

@@ -136,7 +136,7 @@ lint: $(TOOLS_DIR)/golangci-lint $(TOOLS_DIR)/misspell
 	set -e; for dir in $(ALL_GO_MOD_DIRS) $(TOOLS_MOD_DIR); do \
 	  echo "go mod tidy in $${dir}"; \
 	  (cd "$${dir}" && \
-	    go mod tidy); \
+	    GOWORK=off go mod tidy); \
 	done
 	set -e; for dir in $(ALL_GO_MOD_DIRS); do \
 	  echo "golangci-lint in $${dir}"; \

go.mod

@@ -46,6 +46,7 @@ require (
 	github.com/lightstep/otel-launcher-go/lightstep/instrumentation v1.23.0 // indirect
 	github.com/lightstep/otel-launcher-go/lightstep/sdk/internal v1.23.0 // indirect
 	github.com/lightstep/otel-launcher-go/lightstep/sdk/trace v1.23.0 // indirect
+	github.com/lightstep/varopt v1.4.0 // indirect
 	github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect
 	github.com/minio/asm2plan9s v0.0.0-20200509001527-cdd76441f9d8 // indirect
 	github.com/minio/c2goasm v0.0.0-20190812172519-36a3d3bbc4f3 // indirect

go.sum

@@ -782,6 +782,8 @@ github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
 github.com/lightstep/go-expohisto v1.0.0 h1:UPtTS1rGdtehbbAF7o/dhkWLTDI73UifG8LbfQI7cA4=
 github.com/lightstep/go-expohisto v1.0.0/go.mod h1:xDXD0++Mu2FOaItXtdDfksfgxfV0z1TMPa+e/EUd0cs=
+github.com/lightstep/varopt v1.4.0 h1:MCpQouffyrj0xGe7pQRP0urgMHG04gHjE9v5FhpODzo=
+github.com/lightstep/varopt v1.4.0/go.mod h1:8XCrfUxO78WYWeFHSFD1j1ePNhRsGXd44YTfn+l3kjs=
 github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 h1:6E+4a0GO5zZEnZ81pIr0yLvtUWk2if982qA3F3QD6H4=
 github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0/go.mod h1:zJYVVT2jmtg6P3p1VtQj7WsuWi/y4VnjVBn7F8KPB3I=
 github.com/lyft/protoc-gen-star v0.6.0/go.mod h1:TGAoBVkt8w7MPG72TrKIu85MIdXwDuzJYeZuUPFPNwA=

lightstep/sdk/internal/common.go

@@ -15,6 +15,7 @@
 package internal
 
 import (
+	"fmt"
 	"sync"
 
 	"go.opentelemetry.io/collector/pdata/pcommon"
@@ -61,43 +62,46 @@ func (rm *ResourceMap) Get(in *resource.Resource) pcommon.Resource {
 
 func CopyAttributes(dest pcommon.Map, src attribute.Set) {
 	for iter := src.Iter(); iter.Next(); {
-		inA := iter.Attribute()
-		key := string(inA.Key)
-		switch inA.Value.Type() {
-		case attribute.BOOL:
-			dest.PutBool(key, inA.Value.AsBool())
-		case attribute.INT64:
-			dest.PutInt(key, inA.Value.AsInt64())
-		case attribute.FLOAT64:
-			dest.PutDouble(key, inA.Value.AsFloat64())
-		case attribute.STRING:
-			dest.PutStr(key, inA.Value.AsString())
-		case attribute.BOOLSLICE:
-			sl := dest.PutEmptySlice(key)
-			sl.EnsureCapacity(len(inA.Value.AsBoolSlice()))
-			for _, v := range inA.Value.AsBoolSlice() {
-				sl.AppendEmpty().SetBool(v)
-			}
-		case attribute.INT64SLICE:
-			sl := dest.PutEmptySlice(key)
-			sl.EnsureCapacity(len(inA.Value.AsInt64Slice()))
-			for _, v := range inA.Value.AsInt64Slice() {
-				sl.AppendEmpty().SetInt(v)
-			}
-		case attribute.FLOAT64SLICE:
-			sl := dest.PutEmptySlice(key)
-			sl.EnsureCapacity(len(inA.Value.AsFloat64Slice()))
-			for _, v := range inA.Value.AsFloat64Slice() {
-				sl.AppendEmpty().SetDouble(v)
-			}
-		case attribute.STRINGSLICE:
-			sl := dest.PutEmptySlice(key)
-			sl.EnsureCapacity(len(inA.Value.AsStringSlice()))
-			for _, v := range inA.Value.AsStringSlice() {
-				sl.AppendEmpty().SetStr(v)
-			}
-		default:
-			panic("unhandled case")
+		CopyAttribute(dest, iter.Attribute())
+	}
+}
+
+func CopyAttribute(dest pcommon.Map, inA attribute.KeyValue) {
+	key := string(inA.Key)
+	switch inA.Value.Type() {
+	case attribute.BOOL:
+		dest.PutBool(key, inA.Value.AsBool())
+	case attribute.INT64:
+		dest.PutInt(key, inA.Value.AsInt64())
+	case attribute.FLOAT64:
+		dest.PutDouble(key, inA.Value.AsFloat64())
+	case attribute.STRING:
+		dest.PutStr(key, inA.Value.AsString())
+	case attribute.BOOLSLICE:
+		sl := dest.PutEmptySlice(key)
+		sl.EnsureCapacity(len(inA.Value.AsBoolSlice()))
+		for _, v := range inA.Value.AsBoolSlice() {
+			sl.AppendEmpty().SetBool(v)
+		}
+	case attribute.INT64SLICE:
+		sl := dest.PutEmptySlice(key)
+		sl.EnsureCapacity(len(inA.Value.AsInt64Slice()))
+		for _, v := range inA.Value.AsInt64Slice() {
+			sl.AppendEmpty().SetInt(v)
+		}
+	case attribute.FLOAT64SLICE:
+		sl := dest.PutEmptySlice(key)
+		sl.EnsureCapacity(len(inA.Value.AsFloat64Slice()))
+		for _, v := range inA.Value.AsFloat64Slice() {
+			sl.AppendEmpty().SetDouble(v)
+		}
+	case attribute.STRINGSLICE:
+		sl := dest.PutEmptySlice(key)
+		sl.EnsureCapacity(len(inA.Value.AsStringSlice()))
+		for _, v := range inA.Value.AsStringSlice() {
+			sl.AppendEmpty().SetStr(v)
 		}
+	default:
+		panic(fmt.Errorf("unhandled case: %v", inA.Value.Type()))
 	}
 }

lightstep/sdk/metric/README.md

@@ -199,3 +199,90 @@ dynamically configure allowed cardinality values.
 This limit is used to truncate attribute key and string values to a
 reasonable size.  The default limit is 8kB.  Zero is not a valid
 limit.
+
+#### Exemplars
+
+**Status**: Experimental
+
+Exemplars are sample measurements associated with synchronous metric
+instruments.  When OpenTelemetry tracing is used in conjunction with
+this Metrics SDK, exemplars will be annotated with the TraceID and
+SpanID of the traced context.
+
+Collection of metric exemplars are off by default.  The
+`sdkinstrument.Performance.ExemplarsEnabled` field can be used to
+enable exemplars by default.  This field may be set to a number of
+exemplars to collect by default for all Counter and Histogram
+instruments.
+
+Exemplars can also be configured using the `aggregator.Config.Exemplar`
+structure, or with a hint like:
+
+```
+{
+  "description": "measurement of ...",
+  "config": {
+    "exemplar": {
+      "size": 10,
+	  "filter": "trace_based",
+    }
+  }
+}
+```
+
+Like the OpenTelemetry specification, the supported filters are
+"always_off", "always_on", and "trace_based".  Unlike the
+OpenTelemetry specification, this SDK has two reservoir
+implementations:
+
+- "Last": always chooses the last metric event as the exemplar.  This
+  method is automatically selected when the size is 1.
+- "Weighted": uses a weighted sampling technique.
+
+With weighted sampling, an unbiased sampler is used such that the
+distribution of values can be estimated from the exemplars.  Each
+exemplar includes a `sample.weight` attribute indicating its
+contribution to the aggregate value.
+
+Note that this weighted sampling property does not apply to
+UpDownCounter instruments, because they allow negative measurements.
+however these instruments can still generate exemplars.
+
+As a simple example, consider a counter instrument with two input
+attribute values counting blue and yellow items, i.e.,
+
+```
+counter := meter.Int64Counter("...")
+
+for _ := range BLUE {
+  ctx, span := tracer.Start(...)
+  counter.Add(ctx, 1, attribute.String("color", "yellow")
+  span.End()
+}
+
+for _ := range YELLOW {
+  ctx, span := tracer.Start(...)
+  counter.Add(ctx, 1, attribute.String("color", "blue")
+  span.End()
+}
+```
+
+Suppose the attribute is filtered, so that a single timeseries is
+generated.  The aggregate sum equals `len(YELLOW) + len(BLUE)`.
+
+Each exemplar will have one of the filtered attributes, `color=yellow`
+or `color=blue`, Each exemplar will have a value of 1, in this case
+(i.e., the original measurement).  The ratio of exemplars with
+`color=yellow` or `color=blue` will match the ratio of counts
+associated with each.
+
+If the number of BLUE items is 3000, and the number of YELLOW items is
+7000, and the number of exemplars is 1000, then:
+
+- We expect 300 BLUE examplars
+- We expect 700 YELLOW examplars
+- Each exemplar has a `sample.weight` of 10, the ratio of total count to exemplar count.
+
+Note that we expect the sum of `sample.weight` for the exemplars to
+equal the total number of input events (i.e., 3000 BLUE, 7000 YELLOW).
+

lightstep/sdk/metric/aggregator/aggregator.go

@@ -24,6 +24,8 @@ import (
 	"github.com/lightstep/otel-launcher-go/lightstep/sdk/metric/number"
 	"github.com/lightstep/otel-launcher-go/lightstep/sdk/metric/sdkinstrument"
 	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/trace"
 )
 
 // Sentinel errors for Aggregator interface.
@@ -33,6 +35,28 @@ var (
 	ErrInfInput      = fmt.Errorf("±Inf value is an invalid input")
 )
 
+// ExemplarFilterKind determines which events are eligible for
+// becoming exemplars.
+type ExemplarFilterKind int
+
+const (
+	// AlwaysOffKind is the default when aggregator.Config{} is
+	// used with a zero value.  This is a good default because
+	// exemplars require additional synchronization.
+	AlwaysOffKind ExemplarFilterKind = iota
+
+	// AlwaysOnKind considers all events for exemplar sampling.
+	AlwaysOnKind
+
+	// WhenTracedKind considers events only in sampled trace
+	// contexts for exemplar sampling.
+	WhenTracedKind
+)
+
+// DefaultExemplarReservoirSize determines how many exemplars will be
+// selected per instrument.
+const DefaultExemplarReservoirSize = 10
+
 // RangeTest is a common routine for testing for valid input values.
 // This rejects NaN and Inf values.  This rejects negative values when the
 // aggregation does not support negative values, including
@@ -68,6 +92,20 @@ func RangeTest[N number.Any, Traits number.Traits[N]](num N, desc sdkinstrument.
 	return true
 }
 
+// ExemplarConfig configures exemplar selection.
+type ExemplarConfig struct {
+	// Filter determines which contexts are selected.
+	Filter ExemplarFilterKind
+	// Size determines limits how many exemplars per timeseries.
+	Size uint32
+}
+
+// JSONExemplarConfig configures exemplar selection.
+type JSONExemplarConfig struct {
+	Filter string `json:"filter"`
+	Size   uint32 `json:"size"`
+}
+
 // JSONHistogramConfig configures the exponential histogram.
 type JSONHistogramConfig struct {
 	MaxSize int32 `json:"max_size"`
@@ -77,6 +115,7 @@ type JSONHistogramConfig struct {
 type JSONConfig struct {
 	Histogram        JSONHistogramConfig `json:"histogram"`
 	CardinalityLimit uint32              `json:"cardinality_limit"`
+	Exemplar         JSONExemplarConfig  `json:"exemplar"`
 }
 
 // Config supports the configuration for all aggregators in a single struct.
@@ -87,6 +126,9 @@ type Config struct {
 	// CardinalityLimit limits the number of instances of this
 	// aggregator in a given view.
 	CardinalityLimit uint32
+
+	// ExemplarFilter enables or disables exemplars
+	Exemplar ExemplarConfig
 }
 
 // Valid returns true for valid configurations.
@@ -140,7 +182,7 @@ type Methods[N number.Any, Storage any] interface {
 
 	// Update modifies Storage concurrently with respect to
 	// concurrent Move(), Copy(), and Update() operations.
-	Update(ptr *Storage, number N)
+	Update(ptr *Storage, number N, ex ExemplarBits)
 
 	// Move atomically copies `input` to `output` and resets
 	// `input` to the zero state.  The change to `input` is
@@ -181,7 +223,44 @@ type Methods[N number.Any, Storage any] interface {
 	// aggregation.  If the instrument is asynchronous, this will
 	// be called after subtraction.  Not synchronized.
 	HasChange(ptr *Storage) bool
+
+	// Exemplars returns sample points included in this aggregation.
+	Exemplars(ptr *Storage, in []WeightedExemplarBits) []WeightedExemplarBits
+
+	// Weight is the sample weight.  It is 1 for histogram and
+	// gauge aggregations, and it is the value for sum data
+	// aggregations.
+	Weight(number N) float64
 }
 
 // ConfigSelector is a per-instrument-kind, per-number-kind Config choice.
 type ConfigSelector func(sdkinstrument.Kind) (int64Config, float64Config Config)
+
+// ExemplarBits conducts extra information into the aggregation pipeline.
+//
+// Note: we could opt for an allocation instead of copying this struct
+// by value through the pipeline.
+type ExemplarBits struct {
+	// Time of the event.
+	Time time.Time
+
+	// Attributes are the complete original set of attributes.
+	Attributes []attribute.KeyValue
+
+	// Span has a reference to the span context, which has the 24
+	// bytes of ID.  We keep a span reference here because it is
+	// slightly smaller.
+	Span trace.Span
+
+	// Number is the input value.
+	Number number.Number
+}
+
+// WeightedExemplarBits are the exemplar and its calculated sample weight.
+type WeightedExemplarBits struct {
+	// ExemplarBits calculated at the event.
+	ExemplarBits
+
+	// Weight calculated by aggregation pipeline.
+	Weight float64
+}

lightstep/sdk/metric/aggregator/gauge/gauge.go

@@ -137,7 +137,7 @@ func (Methods[N, Traits]) Copy(from, to *State[N, Traits]) {
 	to.seq = from.seq
 }
 
-func (Methods[N, Traits]) Update(state *State[N, Traits], number N) {
+func (Methods[N, Traits]) Update(state *State[N, Traits], number N, _ aggregator.ExemplarBits) {
 	newSeq := atomic.AddUint64(&sequenceVar, 1)
 
 	state.lock.Lock()
@@ -169,3 +169,11 @@ func (Methods[N, Traits]) ToStorage(aggr aggregation.Aggregation) (*State[N, Tra
 func (Methods[N, Traits]) SubtractSwap(operand, argument *State[N, Traits]) {
 	panic("not used for non-temporal metrics")
 }
+
+func (Methods[N, Traits]) Exemplars(ptr *State[N, Traits], in []aggregator.WeightedExemplarBits) []aggregator.WeightedExemplarBits {
+	return in
+}
+
+func (Methods[N, Traits]) Weight(_ N) float64 {
+	return 1
+}