Add DistinctCountSketch meter

go.mod

@@ -1,3 +1,5 @@
 module github.com/Netflix/spectator-go/v2
 
 go 1.21
+
+require github.com/cespare/xxhash/v2 v2.3.0

go.sum

@@ -0,0 +1,2 @@
+github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
+github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=

spectator/meter/distinct_count_sketch.go

@@ -0,0 +1,60 @@
+package meter
+
+import (
+	"encoding/binary"
+	"strconv"
+
+	"github.com/Netflix/spectator-go/v2/spectator/writer"
+	"github.com/cespare/xxhash/v2"
+)
+
+// DistinctCountSketch estimates the number of distinct values seen during a step interval, such as
+// the number of unique users, device ids, or source IPs, using a HyperLogLog sketch. The result is
+// an estimate (~13% standard error), not an exact count.
+//
+// This client computes the xxHash64 of the recorded value locally and sends the precomputed hash
+// to SpectatorD (the 's' line type), which derives the HyperLogLog registers. The hashing matches
+// the other Spectator clients so that sketches recorded by different clients merge correctly.
+// Integers are hashed as their 8-byte little-endian representation and strings as their UTF-8
+// bytes.
+//
+// Recording into a sketch results in a fixed number of gauges (64) being published by SpectatorD,
+// so treat any additional dimensions with the same diligence as percentile timers and ensure they
+// have a small bounded cardinality. This type is safe for concurrent use.
+type DistinctCountSketch struct {
+	id         *Id
+	writer     writer.Writer
+	linePrefix string
+}
+
+// NewDistinctCountSketch generates a new distinct count sketch, using the provided meter
+// identifier.
+func NewDistinctCountSketch(id *Id, writer writer.Writer) *DistinctCountSketch {
+	return &DistinctCountSketch{id, writer, "s:" + id.spectatordId + ":"}
+}
+
+// MeterId returns the meter identifier.
+func (d *DistinctCountSketch) MeterId() *Id {
+	return d.id
+}
+
+// RecordString records a distinct string value, hashed as its UTF-8 bytes.
+func (d *DistinctCountSketch) RecordString(value string) {
+	d.recordHash(xxhash.Sum64String(value))
+}
+
+// RecordBytes records a distinct value from its raw bytes.
+func (d *DistinctCountSketch) RecordBytes(value []byte) {
+	d.recordHash(xxhash.Sum64(value))
+}
+
+// RecordInt64 records a distinct integer value, hashed as its 8-byte little-endian representation.
+func (d *DistinctCountSketch) RecordInt64(value int64) {
+	var buf [8]byte
+	binary.LittleEndian.PutUint64(buf[:], uint64(value))
+	d.recordHash(xxhash.Sum64(buf[:]))
+}
+
+func (d *DistinctCountSketch) recordHash(hash uint64) {
+	d.writer.Write(d.linePrefix + strconv.FormatUint(hash, 10))
+}

spectator/meter/distinct_count_sketch_test.go

@@ -0,0 +1,172 @@
+package meter
+
+import (
+	"bufio"
+	"encoding/binary"
+	"encoding/hex"
+	"os"
+	"strconv"
+	"strings"
+	"testing"
+
+	"github.com/Netflix/spectator-go/v2/spectator/writer"
+	"github.com/cespare/xxhash/v2"
+)
+
+const vectorFile = "testdata/distinct_count_sketch_test_vectors.txt"
+
+// hashOfVector computes the xxHash64 of an encoding-vector input the same way the sketch does for
+// each type: long as 8-byte little-endian, str as UTF-8 bytes, bytes as raw bytes. This is an
+// independent check that the Go hashing matches the hash column produced by spectator-java.
+func hashOfVector(t *testing.T, typ, input string) uint64 {
+	switch typ {
+	case "long":
+		v, err := strconv.ParseInt(input, 10, 64)
+		if err != nil {
+			t.Fatalf("bad long input %q: %v", input, err)
+		}
+		var buf [8]byte
+		binary.LittleEndian.PutUint64(buf[:], uint64(v))
+		return xxhash.Sum64(buf[:])
+	case "str":
+		return xxhash.Sum64String(unescapeVector(input))
+	case "bytes":
+		b, err := hex.DecodeString(input)
+		if err != nil {
+			t.Fatalf("bad bytes input %q: %v", input, err)
+		}
+		return xxhash.Sum64(b)
+	default:
+		t.Fatalf("unknown vector type %q", typ)
+		return 0
+	}
+}
+
+// recordVector drives a vector input through the production Record* method for its type.
+func recordVector(t *testing.T, d *DistinctCountSketch, typ, input string) {
+	switch typ {
+	case "long":
+		v, err := strconv.ParseInt(input, 10, 64)
+		if err != nil {
+			t.Fatalf("bad long input %q: %v", input, err)
+		}
+		d.RecordInt64(v)
+	case "str":
+		d.RecordString(unescapeVector(input))
+	case "bytes":
+		b, err := hex.DecodeString(input)
+		if err != nil {
+			t.Fatalf("bad bytes input %q: %v", input, err)
+		}
+		d.RecordBytes(b)
+	default:
+		t.Fatalf("unknown vector type %q", typ)
+	}
+}
+
+// unescapeVector reverses the escaping used for str values in the vector file (\\, \t, \n, \r).
+func unescapeVector(s string) string {
+	var b strings.Builder
+	for i := 0; i < len(s); i++ {
+		if s[i] == '\\' && i+1 < len(s) {
+			i++
+			switch s[i] {
+			case '\\':
+				b.WriteByte('\\')
+			case 't':
+				b.WriteByte('\t')
+			case 'n':
+				b.WriteByte('\n')
+			case 'r':
+				b.WriteByte('\r')
+			default:
+				b.WriteByte(s[i])
+			}
+		} else {
+			b.WriteByte(s[i])
+		}
+	}
+	return b.String()
+}
+
+// TestDistinctCountSketch_HashVectors verifies that the Go xxHash64 of each encoding-vector input
+// matches the hash produced by the spectator-java reference implementation. This is what the 's'
+// (precomputed hash) line type sends to SpectatorD, so matching hashes is what lets Go-recorded
+// sketches merge with sketches from the other clients.
+func TestDistinctCountSketch_HashVectors(t *testing.T) {
+	f, err := os.Open(vectorFile)
+	if err != nil {
+		t.Fatalf("cannot open vector file: %v", err)
+	}
+	defer f.Close()
+
+	inSection := false
+	checked := 0
+	scanner := bufio.NewScanner(f)
+	for scanner.Scan() {
+		line := scanner.Text()
+		if line == "" || strings.HasPrefix(line, "#") {
+			continue
+		}
+		if strings.HasPrefix(line, "[") {
+			inSection = line == "[encoding]"
+			continue
+		}
+		if !inSection {
+			continue
+		}
+		fields := strings.Split(line, "\t")
+		if len(fields) != 5 {
+			t.Fatalf("malformed encoding line: %q", line)
+		}
+		typ, input, hashHex := fields[0], fields[1], fields[2]
+		expected, err := strconv.ParseUint(hashHex, 16, 64)
+		if err != nil {
+			t.Fatalf("bad hash %q: %v", hashHex, err)
+		}
+		if got := hashOfVector(t, typ, input); got != expected {
+			t.Errorf("hash mismatch for %s %q: got %016x, want %016x", typ, input, got, expected)
+		}
+
+		// Drive the input through the production Record* methods and confirm the emitted line
+		// carries the hash from the vector file, tying the shipped code to the cross-language
+		// reference (covers negatives, min int64, multibyte strings, and empty values).
+		w := writer.MemoryWriter{}
+		d := NewDistinctCountSketch(NewId("test", nil), &w)
+		recordVector(t, d, typ, input)
+		wantLine := "s:test:" + strconv.FormatUint(expected, 10)
+		if lines := w.Lines(); len(lines) != 1 || lines[0] != wantLine {
+			t.Errorf("line mismatch for %s %q: got %v, want %q", typ, input, lines, wantLine)
+		}
+		checked++
+	}
+	if err := scanner.Err(); err != nil {
+		t.Fatalf("error reading vector file: %v", err)
+	}
+	if checked == 0 {
+		t.Fatal("no encoding vectors were checked")
+	}
+}
+
+func TestDistinctCountSketch_RecordString(t *testing.T) {
+	w := writer.MemoryWriter{}
+	d := NewDistinctCountSketch(NewId("dcs", nil), &w)
+	// "a" hashes to 0xd24ec4f1a98c6e5b == 15154266338359012955 (see the vector file).
+	d.RecordString("a")
+
+	expected := "s:dcs:15154266338359012955"
+	if lines := w.Lines(); len(lines) != 1 || lines[0] != expected {
+		t.Errorf("expected %q, got %v", expected, lines)
+	}
+}
+
+func TestDistinctCountSketch_RecordStringMatchesBytes(t *testing.T) {
+	// RecordString(s) and RecordBytes([]byte(s)) must agree, since both hash the same bytes.
+	ws := writer.MemoryWriter{}
+	wb := writer.MemoryWriter{}
+	NewDistinctCountSketch(NewId("dcs", nil), &ws).RecordString("user-12345")
+	NewDistinctCountSketch(NewId("dcs", nil), &wb).RecordBytes([]byte("user-12345"))
+	if ws.Lines()[0] != wb.Lines()[0] {
+		t.Errorf("RecordString and RecordBytes disagree: %q vs %q", ws.Lines()[0], wb.Lines()[0])
+	}
+}

spectator/meter/testdata/distinct_count_sketch_test_vectors.txt

@@ -0,0 +1,62 @@
+# Distinct count sketch cross-language test vectors.
+#
+# Generated from the spectator-java reference implementation
+# (DistinctCountSketch). Copy this file to other implementations (e.g.
+# spectatord) and verify they produce the same registers, so that sketches
+# from different clients merge correctly.
+#
+# Hash: xxHash64, seed 0. A long is hashed as its 8-byte little-endian
+# representation; a string as its UTF-8 bytes. Register index = low 6 bits of
+# the hash; rho = 1 + the number of leading zeros in the remaining 58 bits.
+#
+# [encoding]: <type>\t<input>\t<hash>\t<index>\t<rho>
+#   type  = long | str | bytes
+#   input = long: signed decimal; str: UTF-8 literal with \\, \t, \n, \r
+#           escaped; bytes: lowercase hex
+#   hash  = unsigned 64-bit hash as 16 hex digits
+#   index = register index 0..63
+#   rho   = register value 1..59
+#
+# [sketch]: <spec>\t<r0>,<r1>,...,<r63>
+#   spec = encoding-inputs (all inputs above) | longs:<start>:<end>
+#   followed by the 64 register rho values (0 = unused) after recording the
+#   spec's inputs into a single sketch.
+[encoding]
+long	0	34c96acdcadb1bbb	59	3
+long	1	9f29cb17a2a49995	21	1
+long	-1	85d136adb773c6c9	9	1
+long	2	eac73e4044e82db0	48	1
+long	7	0876cd406afde455	21	5
+long	42	b556806fb6d14353	19	1
+long	100	a1df9c5cd454307a	58	1
+long	1000	e53f2d22876c9a49	9	1
+long	1000000	f126194744ecba1e	30	1
+long	123456789	cb7c2941b198004d	13	1
+long	-123456789	6652a701f322f9c0	0	2
+long	3405691582	6b4aca62b5efd3db	27	2
+long	9223372036854775807	ff70cc60366e770c	12	1
+long	-9223372036854775808	3f425eacf01544e0	32	3
+str		ef46db3751d8e999	25	1
+str	a	d24ec4f1a98c6e5b	27	1
+str	ab	65f708ca92d04a61	33	2
+str	abc	44bc2cf5ad770999	25	2
+str	user-12345	64797fca50bfc10c	12	2
+str	192.168.0.1	87f49df2f9f76384	4	1
+str	2001:db8::1	26c33ef1c67e1031	49	3
+str	key=value,other:thing	d799484be9292024	36	1
+str	héllo	3bd06310388ebbe4	36	3
+str	naïve	c07351dc8a26afe6	38	1
+str	世界	6af6be193ab0db0f	15	2
+str	日本語	7179a19f3719f5e1	33	2
+str	😀	9025b8abaae87b80	0	1
+str	👍🏽	637d0995477b428c	12	2
+str	/api/v1/users/abcd1234//api/v1/users/abcd1234//api/v1/users/abcd1234//api/v1/users/abcd1234//api/v1/users/abcd1234//api/v1/users/abcd1234//api/v1/users/abcd1234//api/v1/users/abcd1234//api/v1/users/abcd1234//api/v1/users/abcd1234/	92c0d26c7c73d1a6	38	1
+bytes		ef46db3751d8e999	25	1
+bytes	00	e934a84adb052768	40	1
+bytes	ff	95634172a60b7544	4	1
+bytes	0001020304050607	884a173614b81b8d	13	1
+bytes	deadbeef	2ff5cfb6af9aaf68	40	3
+bytes	fffefd	622529177845a110	16	2
+[sketch]
+encoding-inputs	2,0,0,0,1,0,0,0,0,1,0,0,2,1,0,2,2,0,0,1,0,5,0,0,0,2,0,2,0,0,1,0,3,2,0,0,3,0,1,0,3,0,0,0,0,0,0,0,1,3,0,0,0,0,0,0,0,0,1,3,0,0,0,0
+longs:0:100000	15,12,13,11,10,10,13,11,11,11,12,10,9,10,13,12,12,10,9,12,11,10,12,9,12,12,11,14,9,16,17,14,11,12,9,12,11,12,13,15,12,20,12,13,10,11,11,11,11,12,11,9,15,14,10,12,14,11,13,10,13,10,9,14

spectator/protocol_parser_test.go

@@ -63,3 +63,25 @@ func TestParseGaugeWithTTL(t *testing.T) {
 		t.Errorf("Expected '1', got '%s'", value)
 	}
 }
+
+func TestParseProtocolLineDistinctCountSketch(t *testing.T) {
+	// The 's' (precomputed-hash) line carries a decimal uint64; it has no ':' so it splits cleanly.
+	line := "s:sketch:15154266338359012955"
+	meterType, meterId, value, err := ParseProtocolLine(line)
+
+	if err != nil {
+		t.Errorf("Unexpected error: %v", err)
+	}
+
+	if meterType != "s" {
+		t.Errorf("Expected 's', got '%s'", meterType)
+	}
+
+	if meterId.Name() != "sketch" {
+		t.Errorf("Unexpected meterId: %v", meterId)
+	}
+
+	if value != "15154266338359012955" {
+		t.Errorf("Expected '15154266338359012955', got '%s'", value)
+	}
+}

spectator/registry.go

@@ -27,6 +27,8 @@ type Registry interface {
 	AgeGaugeWithId(id *meter.Id) *meter.AgeGauge
 	Counter(name string, tags map[string]string) *meter.Counter
 	CounterWithId(id *meter.Id) *meter.Counter
+	DistinctCountSketch(name string, tags map[string]string) *meter.DistinctCountSketch
+	DistinctCountSketchWithId(id *meter.Id) *meter.DistinctCountSketch
 	DistributionSummary(name string, tags map[string]string) *meter.DistributionSummary
 	DistributionSummaryWithId(id *meter.Id) *meter.DistributionSummary
 	Gauge(name string, tags map[string]string) *meter.Gauge
@@ -117,6 +119,14 @@ func (r *spectatordRegistry) CounterWithId(id *meter.Id) *meter.Counter {
 	return meter.NewCounter(id, r.writer)
 }
 
+func (r *spectatordRegistry) DistinctCountSketch(name string, tags map[string]string) *meter.DistinctCountSketch {
+	return meter.NewDistinctCountSketch(r.NewId(name, tags), r.writer)
+}
+
+func (r *spectatordRegistry) DistinctCountSketchWithId(id *meter.Id) *meter.DistinctCountSketch {
+	return meter.NewDistinctCountSketch(id, r.writer)
+}
+
 func (r *spectatordRegistry) DistributionSummary(name string, tags map[string]string) *meter.DistributionSummary {
 	return meter.NewDistributionSummary(r.NewId(name, tags), r.writer)
 }

spectator/registry_test.go

@@ -72,6 +72,33 @@ func TestRegistryWithMemoryWriter_CounterWithId(t *testing.T) {
 	}
 }
 
+func TestRegistryWithMemoryWriter_DistinctCountSketch(t *testing.T) {
+	r := NewTestRegistry()
+	mw := r.GetWriter().(*writer.MemoryWriter)
+
+	sketch := r.DistinctCountSketch("test_distinctcountsketch", nil)
+	sketch.RecordString("a")
+
+	// "a" hashes to 15154266338359012955 (see the meter test vectors).
+	expected := "s:test_distinctcountsketch:15154266338359012955"
+	if len(mw.Lines()) != 1 || mw.Lines()[0] != expected {
+		t.Errorf("Expected '%s', got '%s'", expected, mw.Lines()[0])
+	}
+}
+
+func TestRegistryWithMemoryWriter_DistinctCountSketchWithId(t *testing.T) {
+	r := NewTestRegistryWithCommonTags()
+	mw := r.GetWriter().(*writer.MemoryWriter)
+
+	sketch := r.DistinctCountSketchWithId(r.NewId("test_distinctcountsketch", nil))
+	sketch.RecordString("a")
+
+	expected := "s:test_distinctcountsketch,extra-tag=foo:15154266338359012955"
+	if len(mw.Lines()) != 1 || mw.Lines()[0] != expected {
+		t.Errorf("Expected '%s', got '%s'", expected, mw.Lines()[0])
+	}
+}
+
 func TestRegistryWithMemoryWriter_DistributionSummary(t *testing.T) {
 	r := NewTestRegistry()
 	mw := r.GetWriter().(*writer.MemoryWriter)