Sum as first-class AggregationType (edge + vendored pdata) + google_cluster E2E harness

asap-precompute-go/config.go

@@ -124,6 +124,11 @@ type PrecomputeConfig struct {
 	// SketchType determines which Sketch implementation handles
 	// observations for this AggId.
 	SketchType SketchType
+	// AggKind is the umbrella aggregation kind stamped onto every emitted
+	// envelope (Sketch vs Sum). Unset (AggKindUnspecified) is resolved to
+	// AggKindSketch for any SketchType-bearing config, so existing sketch
+	// configs are unaffected; a Sum config sets AggKindSum.
+	AggKind AggregationKind
 	// Mode picks the windowing strategy.
 	Mode AggregationMode
 	// Window configures size / slide / lateness.

asap-precompute-go/envelope.go

@@ -48,6 +48,39 @@ func (s SketchType) String() string {
 	return "Unspecified"
 }
 
+// AggregationKind is the umbrella over WHAT kind of aggregate an envelope
+// carries: a Sketch (whose SketchType sub-tag names the algorithm —
+// DDSketch/KLL/HLL/CountSketch/CountMinSketch) or a scalar Sum. This mirrors
+// the ASAPQuery backend's AggKind { Sketch | ExactAgg } split: "sketch" is
+// ONE aggregation kind and "sum" is a sibling, NOT a SketchType.
+//
+// Backward compatibility: every producer that predates this field emits the
+// zero value (AggKindUnspecified). On decode, an Unspecified AggKind paired
+// with a real SketchType is read as AggKindSketch (see EffectiveAggKind), so
+// existing sketch envelopes stay byte-identical and decode unchanged.
+type AggregationKind uint8
+
+const (
+	// AggKindUnspecified is the proto3 zero value; resolved via
+	// EffectiveAggKind (Unspecified + a real SketchType => Sketch).
+	AggKindUnspecified AggregationKind = iota
+	// AggKindSketch: the payload is a sketch; SketchType names which.
+	AggKindSketch
+	// AggKindSum: the payload is a scalar Sum aggregate ({sum,count}).
+	AggKindSum
+)
+
+// String returns the canonical aggregation-kind name.
+func (a AggregationKind) String() string {
+	switch a {
+	case AggKindSketch:
+		return "Sketch"
+	case AggKindSum:
+		return "Sum"
+	}
+	return "Unspecified"
+}
+
 // Encoding describes how the bytes in SketchEnvelope.Payload are
 // encoded. Mirrors the design-doc §5.1 SketchEnvelope.encoding enum.
 type Encoding uint8
@@ -111,6 +144,11 @@ type SketchEnvelope struct {
 	SchemaVersion uint32
 	// SketchType identifies which sketch algorithm produced Payload.
 	SketchType SketchType
+	// AggKind is the umbrella aggregation kind (Sketch vs Sum). The zero
+	// value (AggKindUnspecified) is resolved by EffectiveAggKind to
+	// AggKindSketch whenever SketchType is set, so every pre-existing
+	// (sketch-only) producer is byte-for-byte unaffected.
+	AggKind AggregationKind
 	// AggID is the controller-plan join key. Pairs the envelope to
 	// a specific PrecomputeConfig.
 	AggID AggId
@@ -162,4 +200,25 @@ type SketchEnvelope struct {
 	// adapter encode-side reads this to set
 	// Sum.SetAggregationTemporality(...). In-process only.
 	AggregationTemporality int32
+	// RelativeAccuracy is the DDSketch alpha (relative accuracy) the
+	// producing sketch was built with — non-zero only for
+	// SketchType==DDSketch. In-process only (NOT a proto wire field): the
+	// OTel adapter's Encode stamps it onto the output pmetric.DDSketch
+	// container's relative_accuracy so the backend registers a non-zero ε.
+	// A 0.0 here leaves the container at its zero value, which the backend
+	// treats as a degenerate (exact, no-bucket) DDSketch — quantile queries
+	// then capability-miss to the archive and return empty.
+	RelativeAccuracy float64
+}
+
+// EffectiveAggKind resolves the envelope's aggregation kind, applying the
+// backward-compat default: an unset AggKind on an envelope that carries a
+// real SketchType is treated as AggKindSketch (every producer emitted before
+// AggKind existed predates the field and only ever produced sketches). A Sum
+// producer sets AggKind = AggKindSum explicitly.
+func (e *SketchEnvelope) EffectiveAggKind() AggregationKind {
+	if e.AggKind == AggKindUnspecified && e.SketchType != SketchTypeUnspecified {
+		return AggKindSketch
+	}
+	return e.AggKind
 }

asap-precompute-go/otel/decode.go

@@ -109,6 +109,28 @@ func decodeMetric(
 				Value:          precompute.EnvelopeValue(env),
 			})
 		}
+	case pmetric.MetricTypeSumAgg:
+		dps := m.SumAgg().DataPoints()
+		for i := 0; i < dps.Len(); i++ {
+			dp := dps.At(i)
+			env := &precompute.SketchEnvelope{
+				SchemaVersion:  1,
+				AggKind:        precompute.AggKindSum,
+				ResourceLabels: resourceLabels,
+				Labels:         AttributesToKeyValues(dp.Attributes()),
+				WindowStartMs:  timestampMs(dp.StartTimestamp()),
+				WindowEndMs:    timestampMs(dp.Timestamp()),
+				Encoding:       sumAggEncodingToHostNeutral(dp.Encoding()),
+				Payload:        copyBytes(dp.Sketch()),
+			}
+			*out = append(*out, precompute.Observation{
+				TimestampMs:    timestampMs(dp.Timestamp()),
+				Metric:         name,
+				ResourceLabels: resourceLabels,
+				Labels:         env.Labels,
+				Value:          precompute.EnvelopeValue(env),
+			})
+		}
 	case pmetric.MetricTypeKLLSketch:
 		dps := m.KLLSketch().DataPoints()
 		for i := 0; i < dps.Len(); i++ {
@@ -263,6 +285,18 @@ func ddSketchEncodingToHostNeutral(e pmetric.DDSketchEncoding) precompute.Encodi
 	return precompute.EncodingUnspecified
 }
 
+func sumAggEncodingToHostNeutral(e pmetric.SumAggEncoding) precompute.Encoding {
+	switch e {
+	case pmetric.SumAggEncodingProto:
+		return precompute.EncodingProtoFull
+	case pmetric.SumAggEncodingProtoDelta:
+		return precompute.EncodingProtoDelta
+	case pmetric.SumAggEncodingMsgpack, pmetric.SumAggEncodingMsgpackDelta:
+		return precompute.EncodingMsgpack
+	}
+	return precompute.EncodingUnspecified
+}
+
 // kllSketchEncodingToHostNeutral mirrors the DDSketch helper for KLL.
 func kllSketchEncodingToHostNeutral(e pmetric.KLLSketchEncoding) precompute.Encoding {
 	switch e {

asap-precompute-go/otel/decode_test.go

@@ -76,6 +76,60 @@ func TestDecode_GaugeReadAsInt(t *testing.T) {
 	}
 }
 
+func TestRoundTrip_SumAggEnvelope(t *testing.T) {
+	t.Parallel()
+	// The fixed 16-byte Sum payload SumWrapper produces (and the backend
+	// cross-language golden): float64 sum (LE) || uint64 count (LE),
+	// here sum=100, count=4.
+	payload := []byte{
+		0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x59, 0x40,
+		0x04, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+	}
+	in := &precompute.SketchEnvelope{
+		SchemaVersion: 1,
+		AggKind:       precompute.AggKindSum,
+		MetricName:    "google_cluster_2019_cpu_rate",
+		Labels:        []precompute.KeyValue{{Key: "zone", Value: "z1"}},
+		WindowStartMs: 1_000,
+		WindowEndMs:   2_000,
+		Encoding:      precompute.EncodingProtoFull,
+		Payload:       payload,
+	}
+	md, err := Encode([]*precompute.SketchEnvelope{in}, &AdapterConfig{})
+	if err != nil {
+		t.Fatalf("encode: %v", err)
+	}
+	m := md.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(0)
+	if m.Type() != pmetric.MetricTypeSumAgg {
+		t.Fatalf("encoded metric type: want SumAgg, got %v", m.Type())
+	}
+	dp := m.SumAgg().DataPoints().At(0)
+	if string(dp.Sketch()) != string(payload) {
+		t.Errorf("encoded sketch payload mismatch")
+	}
+	if dp.Encoding() != pmetric.SumAggEncodingProto {
+		t.Errorf("encoded encoding: %v", dp.Encoding())
+	}
+
+	obs, err := Decode(md, &AdapterConfig{})
+	if err != nil {
+		t.Fatalf("decode: %v", err)
+	}
+	if len(obs) != 1 {
+		t.Fatalf("obs len: want 1, got %d", len(obs))
+	}
+	env := obs[0].Value.Envelope
+	if env == nil || env.EffectiveAggKind() != precompute.AggKindSum {
+		t.Fatalf("decoded agg kind: %+v", env)
+	}
+	if string(env.Payload) != string(payload) {
+		t.Errorf("payload round-trip mismatch: want %x got %x", payload, env.Payload)
+	}
+	if env.WindowStartMs != 1_000 || env.WindowEndMs != 2_000 {
+		t.Errorf("window round-trip: [%d,%d)", env.WindowStartMs, env.WindowEndMs)
+	}
+}
+
 func TestDecode_DDSketchProducesEnvelope(t *testing.T) {
 	t.Parallel()
 	payload := []byte{0xDE, 0xAD, 0xBE, 0xEF}

asap-precompute-go/otel/encode.go

@@ -88,9 +88,30 @@ func writeMetric(out pmetric.Metric, env *precompute.SketchEnvelope, cfg *Adapte
 	startTs := pcommon.Timestamp(env.WindowStartMs * 1_000_000)
 	endTs := pcommon.Timestamp(env.WindowEndMs * 1_000_000)
 
+	// Sum is a first-class AggregationType (not a sketch): emit the
+	// modified-OTLP SumAgg metric carrying the SumState envelope bytes.
+	if env.EffectiveAggKind() == precompute.AggKindSum {
+		dst := out.SetEmptySumAgg()
+		dp := dst.DataPoints().AppendEmpty()
+		KeyValuesToAttributes(env.Labels, dp.Attributes())
+		dp.SetStartTimestamp(startTs)
+		dp.SetTimestamp(endTs)
+		dp.SetSketch(env.Payload)
+		dp.SetEncoding(hostNeutralToSumAggEncoding(env.Encoding))
+		return nil
+	}
+
 	switch env.SketchType {
 	case precompute.SketchTypeDDSketch:
 		dst := out.SetEmptyDDSketch()
+		// Stamp the container's relative_accuracy (the DDSketch alpha) so the
+		// backend registers a non-zero ε. Omitting it left the container at
+		// 0.0 — a degenerate sketch that makes quantile queries
+		// capability-miss to the archive and return empty. Mirrors the
+		// standalone ddsketchprocessor (shim_helpers.go SetRelativeAccuracy).
+		if env.RelativeAccuracy > 0 {
+			dst.SetRelativeAccuracy(env.RelativeAccuracy)
+		}
 		dp := dst.DataPoints().AppendEmpty()
 		KeyValuesToAttributes(env.Labels, dp.Attributes())
 		dp.SetStartTimestamp(startTs)
@@ -190,6 +211,18 @@ func hostNeutralToDDSketchEncoding(e precompute.Encoding) pmetric.DDSketchEncodi
 	return pmetric.DDSketchEncodingProto
 }
 
+func hostNeutralToSumAggEncoding(e precompute.Encoding) pmetric.SumAggEncoding {
+	switch e {
+	case precompute.EncodingProtoFull:
+		return pmetric.SumAggEncodingProto
+	case precompute.EncodingProtoDelta:
+		return pmetric.SumAggEncodingProtoDelta
+	case precompute.EncodingMsgpack:
+		return pmetric.SumAggEncodingMsgpack
+	}
+	return pmetric.SumAggEncodingProto
+}
+
 func hostNeutralToKLLSketchEncoding(e precompute.Encoding) pmetric.KLLSketchEncoding {
 	switch e {
 	case precompute.EncodingMsgpack:

asap-precompute-go/otel/relative_accuracy_integration_test.go

@@ -0,0 +1,95 @@
+package otel_test
+
+import (
+	"testing"
+	"time"
+
+	precompute "github.com/ProjectASAP/asap-precompute-go"
+	oteladapter "github.com/ProjectASAP/asap-precompute-go/otel"
+	"github.com/ProjectASAP/asap-precompute-go/sketches"
+	"go.opentelemetry.io/collector/pdata/pmetric"
+)
+
+// Exercises the REAL DDSketch wrapper through observe -> Drain -> Encode and
+// asserts the relative_accuracy (alpha) survives onto the emitted
+// pmetric.DDSketch container. Regression guard for the fused-edge bug where
+// the wire frame shipped relative_accuracy=0.0 (degenerate sketch -> backend
+// quantile queries capability-miss to archive).
+func TestDDSketchRelativeAccuracyReachesWire(t *testing.T) {
+	const alpha = 0.02
+	cfg := &precompute.PrecomputeConfig{
+		AggID:      1,
+		SketchType: precompute.SketchTypeDDSketch,
+		Mode:       precompute.Tumbling,
+		Window:     precompute.WindowSpec{Size: 10 * time.Second},
+		MetricName: "m",
+	}
+	factory := func() precompute.Sketch { return sketches.NewDDSketchWrapper(alpha) }
+	p := precompute.New(cfg, factory, sketches.DDSketchObserver{})
+	for i := 0; i < 5; i++ {
+		if err := p.Observe(&precompute.Observation{
+			TimestampMs: uint64(1000 * (i + 1)),
+			Metric:      "m",
+			Labels:      []precompute.KeyValue{{Key: "k", Value: "v"}},
+			Value:       precompute.FloatValue(float64(i + 1)),
+		}); err != nil {
+			t.Fatalf("observe: %v", err)
+		}
+	}
+	envs := p.Tick(10_000)
+	if len(envs) == 0 {
+		t.Fatal("no envelopes drained")
+	}
+	if envs[0].RelativeAccuracy != alpha {
+		t.Fatalf("envelope.RelativeAccuracy: want %v, got %v", alpha, envs[0].RelativeAccuracy)
+	}
+
+	md, err := oteladapter.Encode(envs, &oteladapter.AdapterConfig{})
+	if err != nil {
+		t.Fatalf("encode: %v", err)
+	}
+	var got float64 = -1
+	rms := md.ResourceMetrics()
+	for i := 0; i < rms.Len(); i++ {
+		sms := rms.At(i).ScopeMetrics()
+		for j := 0; j < sms.Len(); j++ {
+			ms := sms.At(j).Metrics()
+			for k := 0; k < ms.Len(); k++ {
+				if ms.At(k).Type() == pmetric.MetricTypeDDSketch {
+					got = ms.At(k).DDSketch().RelativeAccuracy()
+				}
+			}
+		}
+	}
+	if got != alpha {
+		t.Fatalf("in-memory pmetric.DDSketch.RelativeAccuracy: want %v, got %v", alpha, got)
+	}
+
+	// Decisive: marshal to OTLP proto bytes and back (what the gRPC exporter
+	// actually ships). If the pdata marshaler drops field 3, relative_accuracy
+	// is lost on the wire even though it was set in memory.
+	b, err := (&pmetric.ProtoMarshaler{}).MarshalMetrics(md)
+	if err != nil {
+		t.Fatalf("marshal: %v", err)
+	}
+	md2, err := (&pmetric.ProtoUnmarshaler{}).UnmarshalMetrics(b)
+	if err != nil {
+		t.Fatalf("unmarshal: %v", err)
+	}
+	var wire float64 = -1
+	rms2 := md2.ResourceMetrics()
+	for i := 0; i < rms2.Len(); i++ {
+		sms := rms2.At(i).ScopeMetrics()
+		for j := 0; j < sms.Len(); j++ {
+			ms := sms.At(j).Metrics()
+			for k := 0; k < ms.Len(); k++ {
+				if ms.At(k).Type() == pmetric.MetricTypeDDSketch {
+					wire = ms.At(k).DDSketch().RelativeAccuracy()
+				}
+			}
+		}
+	}
+	if wire != alpha {
+		t.Fatalf("AFTER proto round-trip DDSketch.RelativeAccuracy: want %v, got %v (pdata marshaler drops it)", alpha, wire)
+	}
+}

asap-precompute-go/precompute.go

@@ -588,6 +588,7 @@ func (p *precompute) serializeSeries(entry *seriesEntry, cfg *PrecomputeConfig,
 	return &SketchEnvelope{
 		SchemaVersion:          1,
 		SketchType:             cfg.SketchType,
+		AggKind:                cfg.AggKind,
 		AggID:                  cfg.AggID,
 		ResourceLabels:         entry.ResourceLabels,
 		Labels:                 labels,
@@ -598,9 +599,22 @@ func (p *precompute) serializeSeries(entry *seriesEntry, cfg *PrecomputeConfig,
 		MetricName:             cfg.MetricName,
 		Count:                  entry.Count,
 		AggregationTemporality: cfg.Temporality,
+		RelativeAccuracy:       sketchRelativeAccuracy(entry.Sketch),
 	}, nil
 }
 
+// sketchRelativeAccuracy reads the DDSketch relative-accuracy alpha off a
+// sketch instance when it exposes one (DDSketchWrapper); 0 for every other
+// family. Stamped onto the envelope so the OTel encoder can set the output
+// pmetric.DDSketch container's relative_accuracy (an ε=0 container is a
+// degenerate sketch the backend can't answer quantiles from).
+func sketchRelativeAccuracy(s Sketch) float64 {
+	if ra, ok := s.(interface{ RelativeAccuracy() float64 }); ok {
+		return ra.RelativeAccuracy()
+	}
+	return 0
+}
+
 // UpdateConfig implements Precompute.UpdateConfig.
 //
 // Currently picks the FIRST config in the set whose AggID matches