diff --git a/go.mod b/go.mod index f5ef074..5bdd999 100644 --- a/go.mod +++ b/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 diff --git a/go.sum b/go.sum index e69de29..1987830 100644 --- a/go.sum +++ b/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= diff --git a/spectator/meter/distinct_count_sketch.go b/spectator/meter/distinct_count_sketch.go new file mode 100644 index 0000000..1ff06c6 --- /dev/null +++ b/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)) +} diff --git a/spectator/meter/distinct_count_sketch_test.go b/spectator/meter/distinct_count_sketch_test.go new file mode 100644 index 0000000..efc7bc4 --- /dev/null +++ b/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]) + } +} diff --git a/spectator/meter/testdata/distinct_count_sketch_test_vectors.txt b/spectator/meter/testdata/distinct_count_sketch_test_vectors.txt new file mode 100644 index 0000000..7294175 --- /dev/null +++ b/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]: \t\t\t\t +# 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]: \t,,..., +# spec = encoding-inputs (all inputs above) | longs:: +# 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 diff --git a/spectator/protocol_parser_test.go b/spectator/protocol_parser_test.go index 00b22c5..5112871 100644 --- a/spectator/protocol_parser_test.go +++ b/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) + } +} diff --git a/spectator/registry.go b/spectator/registry.go index 0b75fb4..464c685 100644 --- a/spectator/registry.go +++ b/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) } diff --git a/spectator/registry_test.go b/spectator/registry_test.go index bc510b7..dbef91c 100644 --- a/spectator/registry_test.go +++ b/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)