From 8eb430d03a5b75f572f8f19313d92e045d597fbd Mon Sep 17 00:00:00 2001 From: Zeying Zhu Date: Mon, 4 May 2026 21:13:51 -0400 Subject: [PATCH] chore(telegraf-patch): delete deprecated aggregator scaffolding MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The processors.allsketches unified plugin (PR #237) supersedes the older per-sketch aggregator-based attempt at Telegraf integration. Remove the dead scaffolding: - telegraf-patch/plugins/aggregators/{ddsketch,kll,countmin,gorilla,sketchutil}/ - telegraf-patch/plugins/aggregators/all/* (registration overlays) - telegraf-patch/plugins/outputs/gorilla_s3/ (was wired to deleted gorilla) Restore script and build pipeline already skip these per PR #240. Three stale reference comments updated (matched_accuracy/main.go, matched_accuracy/README.md, design-asap-edge-framework.md §7 table). Net: 2420 LoC removed. Co-Authored-By: Claude Opus 4.7 (1M context) --- docs/design-asap-edge-framework.md | 2 +- .../matched_accuracy/README.md | 7 +- .../matched_accuracy/main.go | 4 +- restore_telegraf_patches.sh | 9 +- .../plugins/aggregators/all/countmin.go | 5 - .../plugins/aggregators/all/ddsketch.go | 5 - .../plugins/aggregators/all/gorilla.go | 5 - telegraf-patch/plugins/aggregators/all/kll.go | 5 - .../plugins/aggregators/countmin/countmin.go | 262 ------------- .../plugins/aggregators/countmin/sample.conf | 27 -- .../plugins/aggregators/ddsketch/ddsketch.go | 221 ----------- .../plugins/aggregators/ddsketch/sample.conf | 18 - .../plugins/aggregators/gorilla/bitwriter.go | 76 ---- .../plugins/aggregators/gorilla/encoder.go | 171 -------- .../aggregators/gorilla/encoder_test.go | 295 -------------- .../plugins/aggregators/gorilla/gorilla.go | 366 ------------------ .../plugins/aggregators/gorilla/metric.go | 91 ----- .../plugins/aggregators/gorilla/sample.conf | 17 - telegraf-patch/plugins/aggregators/kll/kll.go | 123 ------ .../plugins/aggregators/kll/sample.conf | 13 - .../aggregators/sketchutil/countmin.go | 137 ------- .../plugins/aggregators/sketchutil/topk.go | 139 ------- .../plugins/aggregators/sketchutil/util.go | 32 -- 23 files changed, 7 insertions(+), 2023 deletions(-) delete mode 100644 telegraf-patch/plugins/aggregators/all/countmin.go delete mode 100644 telegraf-patch/plugins/aggregators/all/ddsketch.go delete mode 100644 telegraf-patch/plugins/aggregators/all/gorilla.go delete mode 100644 telegraf-patch/plugins/aggregators/all/kll.go delete mode 100644 telegraf-patch/plugins/aggregators/countmin/countmin.go delete mode 100644 telegraf-patch/plugins/aggregators/countmin/sample.conf delete mode 100644 telegraf-patch/plugins/aggregators/ddsketch/ddsketch.go delete mode 100644 telegraf-patch/plugins/aggregators/ddsketch/sample.conf delete mode 100644 telegraf-patch/plugins/aggregators/gorilla/bitwriter.go delete mode 100644 telegraf-patch/plugins/aggregators/gorilla/encoder.go delete mode 100644 telegraf-patch/plugins/aggregators/gorilla/encoder_test.go delete mode 100644 telegraf-patch/plugins/aggregators/gorilla/gorilla.go delete mode 100644 telegraf-patch/plugins/aggregators/gorilla/metric.go delete mode 100644 telegraf-patch/plugins/aggregators/gorilla/sample.conf delete mode 100755 telegraf-patch/plugins/aggregators/kll/kll.go delete mode 100755 telegraf-patch/plugins/aggregators/kll/sample.conf delete mode 100644 telegraf-patch/plugins/aggregators/sketchutil/countmin.go delete mode 100644 telegraf-patch/plugins/aggregators/sketchutil/topk.go delete mode 100644 telegraf-patch/plugins/aggregators/sketchutil/util.go diff --git a/docs/design-asap-edge-framework.md b/docs/design-asap-edge-framework.md index 53f12b47..1942e972 100644 --- a/docs/design-asap-edge-framework.md +++ b/docs/design-asap-edge-framework.md @@ -514,7 +514,7 @@ Per-platform compile-in mechanism (all Tier 2): | Platform | Mechanism | Distribution unit | |---|---|---| | **OTel Collector** | OCB (OpenTelemetry Collector Builder) reads `builder-config.yaml`, compiles in custom processors at build time. ASAPCollector already does this. | `sketchcollector` | -| **Telegraf** | `plugins/aggregators/all/asap.go` + build tag in `plugins/aggregators/all/aggregators.go`. No `aggregators.execd` exists. | `sketchtelegraf` | +| **Telegraf** | `plugins/processors/all/allsketches.go` build-tag overlay registering the unified `processors.allsketches` plugin. | `sketchtelegraf` | | **Vector** | In-tree feature flag + `inventory::submit!` registration in a path-dependency Cargo workspace. | `sketchvector` | | **OTAP Dataflow** | `linkme` distributed-slice compile-time registration; the project README states "current system is compile-time only." | `sketchotap` | diff --git a/otel_collector_benchmark/matched_accuracy/README.md b/otel_collector_benchmark/matched_accuracy/README.md index fc919a30..1945cdbd 100644 --- a/otel_collector_benchmark/matched_accuracy/README.md +++ b/otel_collector_benchmark/matched_accuracy/README.md @@ -37,11 +37,8 @@ binary marshaller — you'd ship the centroid array yourself). ### what got skipped -* **Gorilla.** A Gorilla codec exists at - `telegraf-patch/plugins/aggregators/gorilla/`, but it's - 1. wrapped in Telegraf's plugin interface (needs `telegraf.Metric`, - `time.Time`, etc.), and - 2. **a (timestamp, value) lossless compressor, not a quantile sketch.** +* **Gorilla.** Gorilla is **a (timestamp, value) lossless compressor, + not a quantile sketch.** In a matched-p99-error table Gorilla either has to store every sample (in which case its wire size is necessarily larger than DDSketch / KLL by diff --git a/otel_collector_benchmark/matched_accuracy/main.go b/otel_collector_benchmark/matched_accuracy/main.go index 338cc8cb..166986e0 100644 --- a/otel_collector_benchmark/matched_accuracy/main.go +++ b/otel_collector_benchmark/matched_accuracy/main.go @@ -21,9 +21,7 @@ // 6. Raw samples – inline (count = N*8 bytes) // 7. Gorilla – SKIPPED. See rationale below. // -// On Gorilla: the only Gorilla codec available locally lives in -// telegraf-patch/plugins/aggregators/gorilla/ and is wrapped in a Telegraf -// aggregator. More importantly, Gorilla is a (timestamp,value) lossless +// On Gorilla: Gorilla is a (timestamp,value) lossless // compression scheme for sequential time-series points; it does not estimate // quantiles. Including it in a "matched p99 error" comparison would mean // either feeding it synthetic timestamps and storing every sample (in which diff --git a/restore_telegraf_patches.sh b/restore_telegraf_patches.sh index 59ee3a8f..7202e07a 100755 --- a/restore_telegraf_patches.sh +++ b/restore_telegraf_patches.sh @@ -17,12 +17,9 @@ # binary (Telegraf is a single-module repo). The script therefore # excludes go.mod / go.sum when staging into the submodule. # -# Pre-`allsketches` experimental patches under telegraf-patch/plugins/ -# (aggregators/ddsketch, aggregators/kll, outputs/gorilla_s3, …) are -# superseded by the unified `allsketches` plugin per the design doc and -# are intentionally NOT copied — their dependencies were never wired -# into Telegraf's go.mod and they no longer build. They remain in -# telegraf-patch/ as archaeology. +# Pre-`allsketches` experimental aggregator/output patches were +# deleted in the cleanup that landed alongside this script — the +# unified `allsketches` plugin supersedes them per the design doc. # # Idempotent: re-running re-copies files; no-ops if patches don't differ. set -euo pipefail diff --git a/telegraf-patch/plugins/aggregators/all/countmin.go b/telegraf-patch/plugins/aggregators/all/countmin.go deleted file mode 100644 index ea091640..00000000 --- a/telegraf-patch/plugins/aggregators/all/countmin.go +++ /dev/null @@ -1,5 +0,0 @@ -//go:build !custom || aggregators || aggregators.countmin - -package all - -import _ "github.com/influxdata/telegraf/plugins/aggregators/countmin" // register plugin diff --git a/telegraf-patch/plugins/aggregators/all/ddsketch.go b/telegraf-patch/plugins/aggregators/all/ddsketch.go deleted file mode 100644 index bf57db98..00000000 --- a/telegraf-patch/plugins/aggregators/all/ddsketch.go +++ /dev/null @@ -1,5 +0,0 @@ -//go:build !custom || aggregators || aggregators.ddsketch - -package all - -import _ "github.com/influxdata/telegraf/plugins/aggregators/ddsketch" // register plugin diff --git a/telegraf-patch/plugins/aggregators/all/gorilla.go b/telegraf-patch/plugins/aggregators/all/gorilla.go deleted file mode 100644 index f9d6e65c..00000000 --- a/telegraf-patch/plugins/aggregators/all/gorilla.go +++ /dev/null @@ -1,5 +0,0 @@ -//go:build !custom || aggregators || aggregators.gorilla - -package all - -import _ "github.com/influxdata/telegraf/plugins/aggregators/gorilla" // register plugin diff --git a/telegraf-patch/plugins/aggregators/all/kll.go b/telegraf-patch/plugins/aggregators/all/kll.go deleted file mode 100644 index 3cd4c0c3..00000000 --- a/telegraf-patch/plugins/aggregators/all/kll.go +++ /dev/null @@ -1,5 +0,0 @@ -//go:build !custom || aggregators || aggregators.kll - -package all - -import _ "github.com/influxdata/telegraf/plugins/aggregators/kll" // register plugin diff --git a/telegraf-patch/plugins/aggregators/countmin/countmin.go b/telegraf-patch/plugins/aggregators/countmin/countmin.go deleted file mode 100644 index d1601407..00000000 --- a/telegraf-patch/plugins/aggregators/countmin/countmin.go +++ /dev/null @@ -1,262 +0,0 @@ -//go:generate ../../../tools/readme_config_includer/generator -package countmin - -import ( - _ "embed" - "encoding/binary" - "encoding/json" - "fmt" - "io" - "sort" - "strings" - - "github.com/cespare/xxhash/v2" - "github.com/influxdata/telegraf" - "github.com/influxdata/telegraf/plugins/aggregators" - sketchutil "github.com/influxdata/telegraf/plugins/aggregators/sketchutil" -) - -//go:embed sample.conf -var sampleConfig string - -const ( - defaultRows = 3 - defaultColumns = 4096 -) - -type CountMinSketchAggregator struct { - Measurement string `toml:"measurement"` - TagKeys []string `toml:"tag_keys"` - GroupBy []string `toml:"group_by"` - Rows int `toml:"rows"` - Columns int `toml:"columns"` - Seed uint64 `toml:"seed"` - TopK int `toml:"top_k"` - - Log telegraf.Logger `toml:"-"` - - cache map[string]*aggregate - groupByKeys map[string]struct{} -} - -type aggregate struct { - measurement string - groupTags map[string]string - sketches map[string]*sketchutil.CountMinSketch -} - -func (*CountMinSketchAggregator) SampleConfig() string { - return sampleConfig -} - -func (c *CountMinSketchAggregator) Init() error { - if c.Measurement == "" { - c.Measurement = "countmin" - } - if c.Rows <= 0 { - c.Rows = defaultRows - } - if c.Columns <= 0 { - c.Columns = defaultColumns - } - if c.Seed == 0 { - c.Seed = 0x9e3779b185ebca87 - } - if c.TopK < 0 { - return fmt.Errorf("countmin: top_k must be >= 0") - } - c.groupByKeys = make(map[string]struct{}, len(c.GroupBy)) - for _, key := range c.GroupBy { - c.groupByKeys[key] = struct{}{} - } - if c.cache == nil { - c.cache = make(map[string]*aggregate) - } - return nil -} - -/* -PromQL query: Spatial aggregation -sum(rate(http_requests_total[5m])) without (instance, pod) - -Example results: -{job="api", method="GET"} 12 -{job="api", method="POST"} 3 -{job="frontend", method="GET"} 9 -*/ - -func (c *CountMinSketchAggregator) Add(m telegraf.Metric) { - if c.cache == nil { - c.cache = make(map[string]*aggregate) - } - - tags := m.Tags() - groupTags := make(map[string]string, len(c.groupByKeys)) - if len(c.groupByKeys) > 0 { - for key := range c.groupByKeys { - if value, ok := tags[key]; ok { - groupTags[key] = value - } - } - } - - cacheKey := aggregateKey(m.Name(), groupTags) - agg, ok := c.cache[cacheKey] - if !ok { - agg = &aggregate{ - measurement: m.Name(), - groupTags: copyTags(groupTags), - sketches: make(map[string]*sketchutil.CountMinSketch), - } - c.cache[cacheKey] = agg - } - - keys := c.effectiveTagKeys(tags) - if len(keys) == 0 { - return - } - valueKey := joinTagValues(tags, keys) - - // Sketches Per Subpopulation; Can be optimized with Hydra sketch later - // Count-Min Sketch for topk(k, multidimensional metrics e.g., http_requests_total by (tag1, tag2, ...)) - // Example PromQL: topk(3, node_memory_Active_bytes by (instance)) - // Noted that for topk(k, q), "by" clause is inside the function - for _, tagKey := range keys { - sk, ok := agg.sketches[tagKey] - if !ok { - derived := deriveSeed(c.Seed, agg.measurement, tagKey) - var err error - sk, err = sketchutil.NewCountMinSketch(c.Rows, c.Columns, derived, c.TopK) - if err != nil { - if c.Log != nil { - c.Log.Errorf("countmin: create sketch for %s/%s failed: %v", agg.measurement, tagKey, err) - } - continue - } - agg.sketches[tagKey] = sk - } - - // Assume FieldList only has one field for now; like Prometheus client protocol - for _, field := range m.FieldList() { - value, ok := sketchutil.ToFloat(field.Value) - if !ok { - continue - } - sk.Insert(valueKey, value) - } - - } -} - -func (c *CountMinSketchAggregator) Push(acc telegraf.Accumulator) { - for _, agg := range c.cache { - for tagKey, sketch := range agg.sketches { - payload, err := sketch.MarshalBinary() - if err != nil { - acc.AddError(fmt.Errorf("countmin: serialize sketch for %s/%s: %w", agg.measurement, tagKey, err)) - continue - } - fields := map[string]interface{}{ - "rows": int64(sketch.Rows()), - "columns": int64(sketch.Columns()), - "count": sketch.Total(), - "countmin": payload, - } - if top := sketch.TopKEntries(); len(top) > 0 { - if encoded, err := json.Marshal(top); err != nil { - acc.AddError(fmt.Errorf("countmin: serialize topk for %s/%s: %w", agg.measurement, tagKey, err)) - } else { - fields["topk"] = encoded - } - } - tags := copyTags(agg.groupTags) - tags["source_measurement"] = agg.measurement - tags["tag_key"] = tagKey - acc.AddFields(c.Measurement, fields, tags) - } - } -} - -func (c *CountMinSketchAggregator) Reset() { - for k := range c.cache { - delete(c.cache, k) - } -} - -func (c *CountMinSketchAggregator) effectiveTagKeys(all map[string]string) []string { - if len(c.TagKeys) > 0 { - return c.TagKeys - } - keys := make([]string, 0, len(all)) - for key := range all { - if _, skip := c.groupByKeys[key]; skip { - continue - } - keys = append(keys, key) - } - sort.Strings(keys) - return keys -} - -func aggregateKey(measurement string, tags map[string]string) string { - if len(tags) == 0 { - return measurement - } - keys := make([]string, 0, len(tags)) - for k := range tags { - keys = append(keys, k) - } - sort.Strings(keys) - var b strings.Builder - b.WriteString(measurement) - for _, k := range keys { - b.WriteString(",") - b.WriteString(k) - b.WriteString("=") - b.WriteString(tags[k]) - } - return b.String() -} - -func copyTags(tags map[string]string) map[string]string { - out := make(map[string]string, len(tags)) - for k, v := range tags { - out[k] = v - } - return out -} - -func joinTagValues(tags map[string]string, keys []string) string { - if len(keys) == 0 { - return "" - } - var b strings.Builder - for i, key := range keys { - if i > 0 { - b.WriteString(",") - } - b.WriteString(key) - b.WriteString("=") - if v, ok := tags[key]; ok { - b.WriteString(v) - } - } - return b.String() -} - -func deriveSeed(base uint64, parts ...string) uint64 { - h := xxhash.New() - var buf [8]byte - binary.LittleEndian.PutUint64(buf[:], base) - h.Write(buf[:]) - for _, part := range parts { - io.WriteString(h, part) - } - return h.Sum64() -} - -func init() { - aggregators.Add("countmin", func() telegraf.Aggregator { - return &CountMinSketchAggregator{} - }) -} diff --git a/telegraf-patch/plugins/aggregators/countmin/sample.conf b/telegraf-patch/plugins/aggregators/countmin/sample.conf deleted file mode 100644 index d1cca071..00000000 --- a/telegraf-patch/plugins/aggregators/countmin/sample.conf +++ /dev/null @@ -1,27 +0,0 @@ -# Approximate per-tag frequencies using Count-Min sketches over the aggregation period. -[[aggregators.countmin]] - ## Configure the aggregation window using the standard [[aggregators]] settings. - # period = "1m" - # grace = "0s" - # drop_original = false - - ## Measurement written by the aggregator when it emits sketches. - # measurement = "countmin" - - ## Sketch dimensions (rows = hash functions, columns = buckets per row). - # rows = 3 - # columns = 4096 - - ## Base seed used to derive row-specific salts. Keep this stable to allow sketch merging. - # seed = 11400714819323198485 - - ## Maintain a heap of the Top-K keys per sketch (set to 0 to disable heap tracking). - # top_k = 20 - - ## Limit the tags inserted into each sketch. When empty, every tag except those listed - ## in group_by is inserted once per metric. - # tag_keys = ["machineid", "tenant"] - - ## Optional set of tags used to partition sketches. Each unique combination of values - ## across these tags produces an independent sketch. - # group_by = ["scrape_url"] diff --git a/telegraf-patch/plugins/aggregators/ddsketch/ddsketch.go b/telegraf-patch/plugins/aggregators/ddsketch/ddsketch.go deleted file mode 100644 index fa865fba..00000000 --- a/telegraf-patch/plugins/aggregators/ddsketch/ddsketch.go +++ /dev/null @@ -1,221 +0,0 @@ -//go:generate ../../../tools/readme_config_includer/generator -package ddsketch - -import ( - _ "embed" - "fmt" - "math" - "sort" - "sync" - - "github.com/DataDog/sketches-go/ddsketch" - sketchpb "github.com/DataDog/sketches-go/ddsketch/pb/sketchpb" - "github.com/influxdata/telegraf" - "github.com/influxdata/telegraf/plugins/aggregators" - sketchutil "github.com/influxdata/telegraf/plugins/aggregators/sketchutil" - "google.golang.org/protobuf/proto" -) - -//go:embed sample.conf -var sampleConfig string - -type DDSketchAggregator struct { - Measurement string `toml:"measurement"` - Accuracy float64 `toml:"accuracy"` - Quantiles []float64 `toml:"quantiles"` - - Log telegraf.Logger `toml:"-"` - - cache map[uint64]*aggregate -} - -type aggregate struct { - name string - tags map[string]string - fields map[string]*seriesData -} - -type zeroPool struct { - pool sync.Pool -} - -func newZeroPool() *zeroPool { - zp := &zeroPool{} - zp.pool.New = func() interface{} { - return &aggregate{ - fields: make(map[string]*seriesData), - } - } - return zp -} - -func (p *zeroPool) Get() *aggregate { - return p.pool.Get().(*aggregate) -} - -func (p *zeroPool) Put(v *aggregate) { - p.pool.Put(v) -} - -var aggregatePool = newZeroPool() - -type seriesData struct { - sketch *ddsketch.DDSketch - count int64 - sum float64 - min float64 - max float64 -} - -func (*DDSketchAggregator) SampleConfig() string { - return sampleConfig -} - -func (d *DDSketchAggregator) Init() error { - if d.Measurement == "" { - d.Measurement = "ddsketch" - } - if d.Accuracy <= 0 { - d.Accuracy = 0.01 - } - if len(d.Quantiles) == 0 { - d.Quantiles = []float64{0.5, 0.9, 0.99} - } - sort.Float64s(d.Quantiles) - d.cache = make(map[uint64]*aggregate) - return nil -} - -func (d *DDSketchAggregator) Add(m telegraf.Metric) { - if d.cache == nil { - d.cache = make(map[uint64]*aggregate) - } - id := m.HashID() - agg, ok := d.cache[id] - if !ok { - agg = acquireAggregate(m.Name(), m.Tags()) - d.cache[id] = agg - } - - for _, field := range m.FieldList() { - value, ok := sketchutil.ToFloat(field.Value) - if !ok { - continue - } - series, ok := agg.fields[field.Key] - if !ok { - sk, err := ddsketch.NewDefaultDDSketch(d.Accuracy) - if err != nil { - d.Log.Errorf("ddsketch: cannot create sketch: %v", err) - continue - } - series = &seriesData{ - sketch: sk, - min: math.Inf(1), - max: math.Inf(-1), - } - agg.fields[field.Key] = series - } - series.sketch.Add(value) - series.count++ - series.sum += value - if value < series.min { - series.min = value - } - if value > series.max { - series.max = value - } - } -} - -func (d *DDSketchAggregator) Push(acc telegraf.Accumulator) { - for _, agg := range d.cache { - for fieldName, data := range agg.fields { - if data.count == 0 { - continue - } - fields := make(map[string]interface{}) - fields["count"] = data.count - fields["sum"] = data.sum - fields["min"] = data.min - fields["max"] = data.max - if data.count > 0 { - fields["mean"] = data.sum / float64(data.count) - } - - for _, q := range d.Quantiles { - if q <= 0 || q >= 1 { - continue - } - value, err := data.sketch.GetValueAtQuantile(q) - if err != nil { - d.Log.Errorf("ddsketch: quantile %.4f failed: %v", q, err) - continue - } - key := fmt.Sprintf("p%g", q*100) - fields[key] = value - } - - if bytes, err := serializeSketch(data.sketch); err != nil { - d.Log.Errorf("ddsketch: serialize: %v", err) - } else { - fields["ddsketch"] = bytes - } - - tags := copyTags(agg.tags) - tags["source_measurement"] = agg.name - tags["field"] = fieldName - acc.AddFields(d.Measurement, fields, tags) - } - } -} - -func (d *DDSketchAggregator) Reset() { - if d.cache == nil { - d.cache = make(map[uint64]*aggregate) - return - } - for id, agg := range d.cache { - releaseAggregate(agg) - delete(d.cache, id) - } -} - -func serializeSketch(sk *ddsketch.DDSketch) ([]byte, error) { - var protoSketch *sketchpb.DDSketch = sk.ToProto() - return proto.Marshal(protoSketch) -} - -func acquireAggregate(name string, tags map[string]string) *aggregate { - agg := aggregatePool.Get() - agg.name = name - agg.tags = copyTags(tags) - return agg -} - -func releaseAggregate(agg *aggregate) { - for k := range agg.fields { - delete(agg.fields, k) - } - agg.name = "" - agg.tags = nil - aggregatePool.Put(agg) -} - -func copyTags(tags map[string]string) map[string]string { - out := make(map[string]string, len(tags)) - for k, v := range tags { - out[k] = v - } - return out -} - -func newDDSketchAggregator() telegraf.Aggregator { - return &DDSketchAggregator{} -} - -func init() { - aggregators.Add("ddsketch", func() telegraf.Aggregator { - return newDDSketchAggregator() - }) -} diff --git a/telegraf-patch/plugins/aggregators/ddsketch/sample.conf b/telegraf-patch/plugins/aggregators/ddsketch/sample.conf deleted file mode 100644 index 325d8418..00000000 --- a/telegraf-patch/plugins/aggregators/ddsketch/sample.conf +++ /dev/null @@ -1,18 +0,0 @@ -# Encode numeric series into DDSketch structures over each aggregation period. -[[aggregators.ddsketch]] - ## Configure a tumbling window unique to this aggregator. - # period = "1m" - # grace = "0s" - # drop_original = false - - ## The aggregator period is configured via the global [[aggregators]] settings. - ## This plugin simply summarizes each series within that tumbling window. - - ## Measurement name for emitted sketch metrics. - # measurement = "ddsketch" - - ## Relative accuracy target passed to sketches-go. - # accuracy = 0.01 - - ## Optional quantiles to evaluate (0 < q < 1). Defaults to [0.5, 0.9, 0.99]. - # quantiles = [0.5, 0.9, 0.99] diff --git a/telegraf-patch/plugins/aggregators/gorilla/bitwriter.go b/telegraf-patch/plugins/aggregators/gorilla/bitwriter.go deleted file mode 100644 index 01bf8387..00000000 --- a/telegraf-patch/plugins/aggregators/gorilla/bitwriter.go +++ /dev/null @@ -1,76 +0,0 @@ -package gorilla - -import ( - "bytes" - "math/bits" -) - -// bitWriter packs bits into a byte buffer. -type bitWriter struct { - buf bytes.Buffer - curByte byte - nbits uint8 // number of bits filled in curByte (0..7) -} - -func newBitWriter() *bitWriter { - return &bitWriter{} -} - -func (w *bitWriter) writeBit(bit uint8) { - if bit != 0 { - w.curByte |= 1 << (7 - w.nbits) - } - w.nbits++ - if w.nbits == 8 { - w.buf.WriteByte(w.curByte) - w.curByte = 0 - w.nbits = 0 - } -} - -// writeBits writes the lower n bits of v (n in 1..64), MSB-first. -func (w *bitWriter) writeBits(v uint64, n uint8) { - for i := int(n) - 1; i >= 0; i-- { - bit := uint8((v >> uint(i)) & 1) - w.writeBit(bit) - } -} - -// writeByteAlign pads with zeros to the next byte boundary. -func (w *bitWriter) writeByteAlign() { - if w.nbits == 0 { - return - } - for w.nbits != 0 { - w.writeBit(0) - } -} - -func (w *bitWriter) bytes() []byte { - // Important: create a copy if caller may retain - w.writeByteAlign() - return w.buf.Bytes() -} - -func leadingZeros64(x uint64) uint8 { - if x == 0 { - return 64 - } - return uint8(bits.LeadingZeros64(x)) -} - -func trailingZeros64(x uint64) uint8 { - if x == 0 { - return 64 - } - return uint8(bits.TrailingZeros64(x)) -} - -func fitsInSignedBits(v int64, n uint8) bool { - if n == 0 || n >= 64 { - return true - } - min := -(int64(1) << (n - 1)) - max := (int64(1) << (n - 1)) - 1 - return v >= min && v <= max -} diff --git a/telegraf-patch/plugins/aggregators/gorilla/encoder.go b/telegraf-patch/plugins/aggregators/gorilla/encoder.go deleted file mode 100644 index 47a8a093..00000000 --- a/telegraf-patch/plugins/aggregators/gorilla/encoder.go +++ /dev/null @@ -1,171 +0,0 @@ -package gorilla - -import ( - "math" - "sort" -) - -type point struct { - ts int64 - v float64 -} - -type seriesKey struct { - measurement string - field string - // tagsKey is a canonicalized, sorted tag string: k1=v1|k2=v2 - tagsKey string -} - -type seriesMeta struct { - Measurement string `json:"measurement"` - Field string `json:"field"` - Tags map[string]string `json:"tags"` - StartTS int64 `json:"start_ts"` - EndTS int64 `json:"end_ts"` - PointCount int `json:"point_count"` -} - -// gorillaTimestampEncoder encodes timestamps using delta-of-delta with Gorilla-style buckets. -type gorillaTimestampEncoder struct { - bw *bitWriter - prevTS int64 - prevDelta int64 - firstSet bool -} - -func newTsEncoder() *gorillaTimestampEncoder { - return &gorillaTimestampEncoder{bw: newBitWriter()} -} - -func (e *gorillaTimestampEncoder) push(ts int64) { - if !e.firstSet { - // First timestamp is stored outside the bitstream (in header) - e.prevTS = ts - e.prevDelta = 0 - e.firstSet = true - return - } - delta := ts - e.prevTS - dd := delta - e.prevDelta - // Gorilla encoding for delta-of-delta - switch { - case dd == 0: - // 0 - e.bw.writeBit(0) - case fitsInSignedBits(dd, 7): - // 10 + 7 bits - e.bw.writeBits(0b10, 2) - e.bw.writeBits(uint64(uint64(dd)&((1<<7)-1)), 7) - case fitsInSignedBits(dd, 9): - // 110 + 9 bits - e.bw.writeBits(0b110, 3) - e.bw.writeBits(uint64(uint64(dd)&((1<<9)-1)), 9) - case fitsInSignedBits(dd, 12): - // 1110 + 12 bits - e.bw.writeBits(0b1110, 4) - e.bw.writeBits(uint64(uint64(dd)&((1<<12)-1)), 12) - default: - // 1111 + 64 bits (fallback) - e.bw.writeBits(0b1111, 4) - e.bw.writeBits(uint64(dd), 64) - } - e.prevTS = ts - e.prevDelta = delta -} - -func (e *gorillaTimestampEncoder) bytes() ([]byte, uint32) { - b := e.bw.bytes() - return b, uint32(len(b) * 8) -} - -// gorillaValueEncoder encodes float64 values using XOR scheme. -type gorillaValueEncoder struct { - bw *bitWriter - prev uint64 - prevSet bool - leadingZeros uint8 - trailingZeros uint8 - havePrevWindow bool -} - -func newValEncoder() *gorillaValueEncoder { - return &gorillaValueEncoder{bw: newBitWriter()} -} - -func (e *gorillaValueEncoder) push(v float64) { - vb := math.Float64bits(v) - if !e.prevSet { - // First value stored outside the bitstream - e.prev = vb - e.prevSet = true - e.leadingZeros = 0 - e.trailingZeros = 0 - e.havePrevWindow = false - return - } - x := e.prev ^ vb - if x == 0 { - e.bw.writeBit(0) // control bit 0 - e.prev = vb - return - } - e.bw.writeBit(1) // control bit 1 - - lz := leadingZeros64(x) - tz := trailingZeros64(x) - sig := 64 - lz - tz - - if e.havePrevWindow && lz >= e.leadingZeros && tz >= e.trailingZeros { - // Use previous window - e.bw.writeBit(0) - e.bw.writeBits(x>>uint(e.trailingZeros), uint8(64-int(e.leadingZeros)-int(e.trailingZeros))) - } else { - // Define new window - e.bw.writeBit(1) - // 5 bits for leading zeros (0..31), clamp at 31 - lz5 := lz - if lz5 > 31 { - lz5 = 31 - } - e.bw.writeBits(uint64(lz5), 5) - // 6 bits for significant bits length minus 1 (1..64) - if sig == 0 { - sig = 64 // degenerate, but write 64 bits - } - sig6 := uint8(sig - 1) - e.bw.writeBits(uint64(sig6), 6) - e.bw.writeBits(x>>uint(tz), uint8(sig)) - e.leadingZeros = lz - e.trailingZeros = tz - e.havePrevWindow = true - } - e.prev = vb -} - -func (e *gorillaValueEncoder) bytes() ([]byte, uint32) { - b := e.bw.bytes() - return b, uint32(len(b) * 8) -} - -// sortAndEncode encodes a series of points into container pieces. -func sortAndEncode(points []point) (firstTS int64, firstValBits uint64, tsBits []byte, tsBitsLen uint32, valBits []byte, valBitsLen uint32) { - if len(points) == 0 { - return 0, 0, nil, 0, nil, 0 - } - sort.Slice(points, func(i, j int) bool { return points[i].ts < points[j].ts }) - // First timestamp and value - firstTS = points[0].ts - firstValBits = math.Float64bits(points[0].v) - - tsEnc := newTsEncoder() - valEnc := newValEncoder() - - for _, p := range points { - tsEnc.push(p.ts) - valEnc.push(p.v) - } - tsBits, tsBitsLen = tsEnc.bytes() - valBits, valBitsLen = valEnc.bytes() - return -} diff --git a/telegraf-patch/plugins/aggregators/gorilla/encoder_test.go b/telegraf-patch/plugins/aggregators/gorilla/encoder_test.go deleted file mode 100644 index 342d496f..00000000 --- a/telegraf-patch/plugins/aggregators/gorilla/encoder_test.go +++ /dev/null @@ -1,295 +0,0 @@ -package gorilla - -import ( - "math" - "math/rand" - "testing" - "time" -) - -type bitReader struct { - b []byte - bitlen uint32 - off uint32 // bit offset -} - -func newBitReader(b []byte, bitlen uint32) *bitReader { - return &bitReader{b: b, bitlen: bitlen, off: 0} -} - -func (r *bitReader) readBit() (uint8, bool) { - if r.off >= r.bitlen { - return 0, false - } - byteIdx := r.off / 8 - bitIdx := r.off % 8 - bit := (r.b[byteIdx] >> (7 - bitIdx)) & 1 - r.off++ - return bit, true -} - -func (r *bitReader) readBits(n uint8) (uint64, bool) { - if n == 0 { - return 0, true - } - var v uint64 - for i := uint8(0); i < n; i++ { - bit, ok := r.readBit() - if !ok { - return 0, false - } - v = (v << 1) | uint64(bit) - } - return v, true -} - -func signExtend(v uint64, n uint8) int64 { - if n == 0 || n >= 64 { - return int64(v) - } - shift := 64 - n - return int64(v<> shift -} - -func decodeTimestamps(firstTS int64, count int, bits []byte, bitlen uint32) ([]int64, bool) { - out := make([]int64, count) - out[0] = firstTS - prevTS := firstTS - prevDelta := int64(0) - r := newBitReader(bits, bitlen) - for i := 1; i < count; i++ { - b, ok := r.readBit() - if !ok { - return nil, false - } - var dd int64 - if b == 0 { - dd = 0 - } else { - // read selector - b2, ok := r.readBit() - if !ok { - return nil, false - } - if b2 == 0 { - v, ok := r.readBits(7) - if !ok { - return nil, false - } - dd = signExtend(v, 7) - } else { - b3, ok := r.readBit() - if !ok { - return nil, false - } - if b3 == 0 { - v, ok := r.readBits(9) - if !ok { - return nil, false - } - dd = signExtend(v, 9) - } else { - b4, ok := r.readBit() - if !ok { - return nil, false - } - if b4 == 0 { - v, ok := r.readBits(12) - if !ok { - return nil, false - } - dd = signExtend(v, 12) - } else { - v, ok := r.readBits(64) - if !ok { - return nil, false - } - dd = int64(v) - } - } - } - } - delta := prevDelta + dd - ts := prevTS + delta - out[i] = ts - prevTS = ts - prevDelta = delta - } - return out, true -} - -func decodeValues(firstValBits uint64, count int, bits []byte, bitlen uint32) ([]float64, bool) { - out := make([]float64, count) - out[0] = math.Float64frombits(firstValBits) - prev := firstValBits - var lz, tz uint8 - haveWindow := false - r := newBitReader(bits, bitlen) - for i := 1; i < count; i++ { - c, ok := r.readBit() - if !ok { - return nil, false - } - var vb uint64 - if c == 0 { - vb = prev - } else { - c2, ok := r.readBit() - if !ok { - return nil, false - } - if c2 == 0 { - // reuse current window - if !haveWindow { - return nil, false - } - sigLen := 64 - lz - tz - sig, ok := r.readBits(uint8(sigLen)) - if !ok { - return nil, false - } - x := sig << tz - vb = prev ^ x - } else { - // new window - lz5, ok := r.readBits(5) - if !ok { - return nil, false - } - lz = uint8(lz5) - sigm1, ok := r.readBits(6) - if !ok { - return nil, false - } - sigLen := uint8(sigm1) + 1 - sig, ok := r.readBits(sigLen) - if !ok { - return nil, false - } - if sigLen == 64 { - tz = 0 - } else { - tz = 64 - lz - sigLen - } - x := sig << tz - vb = prev ^ x - haveWindow = true - } - } - out[i] = math.Float64frombits(vb) - prev = vb - } - return out, true -} - -func makePoints(n int, start int64, step int64, f func(i int) float64) []point { - pts := make([]point, n) - t := start - for i := 0; i < n; i++ { - pts[i] = point{ts: t, v: f(i)} - t += step - } - return pts -} - -func TestRoundTripConstant(t *testing.T) { - n := 200 - pts := makePoints(n, time.Now().UnixNano(), int64(time.Second), func(i int) float64 { return 42 }) - firstTS, firstValBits, tsBits, tsBitsLen, valBits, valBitsLen := sortAndEncode(pts) - dts, ok := decodeTimestamps(firstTS, n, tsBits, tsBitsLen) - if !ok { - t.Fatalf("failed to decode timestamps") - } - dvals, ok := decodeValues(firstValBits, n, valBits, valBitsLen) - if !ok { - t.Fatalf("failed to decode values") - } - for i := 0; i < n; i++ { - if dts[i] != pts[i].ts { - t.Fatalf("timestamp mismatch at %d: got %d want %d", i, dts[i], pts[i].ts) - } - if math.Float64bits(dvals[i]) != math.Float64bits(pts[i].v) { - t.Fatalf("value mismatch at %d: got %v want %v", i, dvals[i], pts[i].v) - } - } -} - -func TestRoundTripLinear(t *testing.T) { - n := 512 - pts := makePoints(n, 1_700_000_000_000_000_000, int64(250*time.Millisecond), func(i int) float64 { return float64(i) }) - firstTS, firstValBits, tsBits, tsBitsLen, valBits, valBitsLen := sortAndEncode(pts) - dts, ok := decodeTimestamps(firstTS, n, tsBits, tsBitsLen) - if !ok { - t.Fatalf("failed to decode timestamps") - } - dvals, ok := decodeValues(firstValBits, n, valBits, valBitsLen) - if !ok { - t.Fatalf("failed to decode values") - } - for i := 0; i < n; i++ { - if dts[i] != pts[i].ts || math.Float64bits(dvals[i]) != math.Float64bits(pts[i].v) { - t.Fatalf("mismatch at %d", i) - } - } -} - -func TestRoundTripRandom(t *testing.T) { - n := 1000 - rnd := rand.New(rand.NewSource(1234)) - base := time.Now().UnixNano() - step := int64(time.Second) - pts := make([]point, n) - ts := base - val := 0.0 - for i := 0; i < n; i++ { - // Random walk for timestamp jitter and values - jitter := rnd.Int63n(int64(5 * time.Millisecond)) - ts += step + jitter - val += rnd.NormFloat64()*0.1 + 0.5 - pts[i] = point{ts: ts, v: val} - } - firstTS, firstValBits, tsBits, tsBitsLen, valBits, valBitsLen := sortAndEncode(pts) - dts, ok := decodeTimestamps(firstTS, n, tsBits, tsBitsLen) - if !ok { - t.Fatalf("failed to decode timestamps") - } - dvals, ok := decodeValues(firstValBits, n, valBits, valBitsLen) - if !ok { - t.Fatalf("failed to decode values") - } - for i := 0; i < n; i++ { - if dts[i] != pts[i].ts || math.Float64bits(dvals[i]) != math.Float64bits(pts[i].v) { - t.Fatalf("mismatch at %d", i) - } - } -} - -// go test github.com/influxdata/telegraf/plugins/aggregators/gorilla -run TestEncodeSingleThreadThroughput -v - -func TestEncodeSingleThreadThroughput(t *testing.T) { - if testing.Short() { - t.Skip("skipping throughput test in short mode") - } - const ( - seriesCount = 1000 - pointsPerSeries = 20000 - minThroughput = 20000.0 // pts/sec – low bar to catch severe regressions - ) - - base := time.Now().UnixNano() - pts := makePoints(pointsPerSeries, base, int64(200*time.Millisecond), func(i int) float64 { - return math.Sin(float64(i) * 0.01) - }) - - totalPoints := seriesCount * pointsPerSeries - start := time.Now() - for i := 0; i < seriesCount; i++ { - sortAndEncode(pts) - } - elapsed := time.Since(start) - throughput := float64(totalPoints) / elapsed.Seconds() - t.Logf("encoded %d points in %s (~%.0f pts/s)", totalPoints, elapsed, throughput) - if throughput < minThroughput { - t.Fatalf("encode throughput too low: got %.0f pts/s, want >= %.0f pts/s", throughput, minThroughput) - } -} diff --git a/telegraf-patch/plugins/aggregators/gorilla/gorilla.go b/telegraf-patch/plugins/aggregators/gorilla/gorilla.go deleted file mode 100644 index 80061262..00000000 --- a/telegraf-patch/plugins/aggregators/gorilla/gorilla.go +++ /dev/null @@ -1,366 +0,0 @@ -//go:generate ../../../tools/readme_config_includer/generator -package gorilla - -import ( - "bytes" - _ "embed" - "encoding/binary" - "encoding/json" - "fmt" - "math" - "sort" - "strconv" - "strings" - "time" - - "github.com/influxdata/telegraf" - "github.com/influxdata/telegraf/config" - "github.com/influxdata/telegraf/metric" - "github.com/influxdata/telegraf/plugins/aggregators" -) - -//go:embed sample.conf -var sampleConfig string - -type Gorilla struct { - Measurement string `toml:"measurement"` - BlockInterval config.Duration `toml:"block_interval"` - MaxObjectBytes int64 `toml:"max_object_bytes"` - - Log telegraf.Logger `toml:"-"` - - series map[seriesKey]*seriesBuffer - blockStart time.Time - blockEnd time.Time -} - -type seriesBuffer struct { - tags map[string]string - points []point -} - -type gorillaChunk struct { - buf []byte - points int - rawBytes int64 -} - -type gorillaObject struct { - data []byte - seriesCount int - points int - rawBytes int64 -} - -func (*Gorilla) SampleConfig() string { - return sampleConfig -} - -func (g *Gorilla) Init() error { - if g.Measurement == "" { - g.Measurement = "gorilla_block" - } - if time.Duration(g.BlockInterval) <= 0 { - g.BlockInterval = config.Duration(10 * time.Minute) - } - if g.series == nil { - g.series = make(map[seriesKey]*seriesBuffer) - } - return nil -} - -func (g *Gorilla) Add(m telegraf.Metric) { - if len(m.FieldList()) == 0 { - return - } - if g.series == nil { - g.series = make(map[seriesKey]*seriesBuffer) - } - ts := m.Time() - if g.blockStart.IsZero() || ts.Before(g.blockStart) { - g.blockStart = ts - } - if ts.After(g.blockEnd) { - g.blockEnd = ts - } - - tags := m.Tags() - tagKey := canonicalizeTags(tags) - for _, field := range m.FieldList() { - val, ok := convertNumeric(field.Value) - if !ok { - continue - } - sk := seriesKey{ - measurement: m.Name(), - field: field.Key, - tagsKey: tagKey, - } - buf, ok := g.series[sk] - if !ok { - buf = &seriesBuffer{ - tags: tags, - points: make([]point, 0, 128), - } - g.series[sk] = buf - } - buf.points = append(buf.points, point{ts: ts.UnixNano(), v: val}) - } -} - -func (g *Gorilla) Push(acc telegraf.Accumulator) { - if len(g.series) == 0 { - return - } - objects, err := g.buildObjects() - if err != nil { - acc.AddError(err) - return - } - if len(objects) == 0 { - return - } - - blockStart := g.blockStart - blockEnd := g.blockEnd - now := time.Now() - if blockStart.IsZero() { - blockStart = now - } - if blockEnd.IsZero() { - blockEnd = now - } - blockInterval := time.Duration(g.BlockInterval) - - for idx, obj := range objects { - var ratio float64 - if obj.rawBytes > 0 { - ratio = float64(len(obj.data)) / float64(obj.rawBytes) - } - fields := map[string]interface{}{ - "series_count": int64(obj.seriesCount), - "point_count": int64(obj.points), - "estimated_raw_bytes": obj.rawBytes, - "compressed_bytes": int64(len(obj.data)), - } - if obj.rawBytes > 0 { - fields["compression_ratio"] = ratio - } - tags := map[string]string{ - "object_index": strconv.Itoa(idx), - "block_start": blockStart.UTC().Format(time.RFC3339Nano), - "block_end": blockEnd.UTC().Format(time.RFC3339Nano), - } - if blockInterval > 0 { - tags["block_interval"] = blockInterval.String() - } - - base := metric.New(g.Measurement, tags, fields, blockEnd) - acc.AddMetric(newGorillaBlockMetric(base, obj.data)) - if g.Log != nil { - g.Log.Infof( - "gorilla aggregator block=%d series=%d points=%d raw_bytes=%d compressed_bytes=%d ratio=%.4f", - idx, - obj.seriesCount, - obj.points, - obj.rawBytes, - len(obj.data), - ratio, - ) - } - } -} - -func (g *Gorilla) Reset() { - g.series = make(map[seriesKey]*seriesBuffer) - g.blockStart = time.Time{} - g.blockEnd = time.Time{} -} - -func (g *Gorilla) buildObjects() ([]gorillaObject, error) { - keys := make([]seriesKey, 0, len(g.series)) - for k := range g.series { - keys = append(keys, k) - } - sort.Slice(keys, func(i, j int) bool { - if keys[i].measurement != keys[j].measurement { - return keys[i].measurement < keys[j].measurement - } - if keys[i].field != keys[j].field { - return keys[i].field < keys[j].field - } - return keys[i].tagsKey < keys[j].tagsKey - }) - - chunks := make([]gorillaChunk, 0, len(keys)) - for _, key := range keys { - buf := g.series[key] - if buf == nil || len(buf.points) == 0 { - continue - } - firstTS, firstValBits, tsBits, tsBitsLen, valBits, valBitsLen := sortAndEncode(buf.points) - meta := seriesMeta{ - Measurement: key.measurement, - Field: key.field, - Tags: buf.tags, - StartTS: buf.points[0].ts, - EndTS: buf.points[len(buf.points)-1].ts, - PointCount: len(buf.points), - } - mb, err := json.Marshal(meta) - if err != nil { - return nil, fmt.Errorf("gorilla aggregator: marshal metadata: %w", err) - } - if len(mb) > math.MaxUint16 { - return nil, fmt.Errorf("gorilla aggregator: metadata too large for series %s %s", key.measurement, key.field) - } - var sb bytes.Buffer - _ = binary.Write(&sb, binary.LittleEndian, uint16(len(mb))) - sb.Write(mb) - _ = binary.Write(&sb, binary.LittleEndian, uint32(len(buf.points))) - _ = binary.Write(&sb, binary.LittleEndian, uint64(firstTS)) - _ = binary.Write(&sb, binary.LittleEndian, uint64(firstValBits)) - _ = binary.Write(&sb, binary.LittleEndian, tsBitsLen) - sb.Write(tsBits) - _ = binary.Write(&sb, binary.LittleEndian, valBitsLen) - sb.Write(valBits) - - chunks = append(chunks, gorillaChunk{ - buf: sb.Bytes(), - points: len(buf.points), - rawBytes: int64(len(buf.points)) * 16, - }) - } - - if len(chunks) == 0 { - return nil, nil - } - return assembleObjects(chunks, g.MaxObjectBytes), nil -} - -func assembleObjects(chunks []gorillaChunk, maxBytes int64) []gorillaObject { - if len(chunks) == 0 { - return nil - } - const headerOverhead = 8 + 1 + 4 - - var objects []gorillaObject - var cur bytes.Buffer - curSeries := 0 - curPoints := 0 - curRaw := int64(0) - curSize := int64(0) - - writeHeader := func() { - cur.Reset() - cur.WriteString("GORILLA1") - cur.WriteByte(1) - _ = binary.Write(&cur, binary.LittleEndian, uint32(0)) - curSeries = 0 - curPoints = 0 - curRaw = 0 - curSize = headerOverhead - } - writeHeader() - - flush := func() { - if curSeries == 0 { - return - } - buf := cur.Bytes() - binary.LittleEndian.PutUint32(buf[9:13], uint32(curSeries)) - data := make([]byte, len(buf)) - copy(data, buf) - objects = append(objects, gorillaObject{ - data: data, - seriesCount: curSeries, - points: curPoints, - rawBytes: curRaw, - }) - writeHeader() - } - - for _, chunk := range chunks { - if maxBytes > 0 && curSeries > 0 && curSize+int64(len(chunk.buf)) > maxBytes { - flush() - } - cur.Write(chunk.buf) - curSeries++ - curPoints += chunk.points - curRaw += chunk.rawBytes - curSize += int64(len(chunk.buf)) - - if maxBytes > 0 && curSeries > 0 && curSize >= maxBytes { - flush() - } - } - flush() - - return objects -} - -func canonicalizeTags(tags map[string]string) string { - if len(tags) == 0 { - return "" - } - keys := make([]string, 0, len(tags)) - for k := range tags { - keys = append(keys, k) - } - sort.Strings(keys) - var b strings.Builder - for i, k := range keys { - if i > 0 { - b.WriteString("|") - } - b.WriteString(k) - b.WriteString("=") - b.WriteString(tags[k]) - } - return b.String() -} - -func convertNumeric(v interface{}) (float64, bool) { - switch val := v.(type) { - case float64: - return val, true - case float32: - return float64(val), true - case int64: - return float64(val), true - case int32: - return float64(val), true - case int16: - return float64(val), true - case int8: - return float64(val), true - case int: - return float64(val), true - case uint64: - return float64(val), true - case uint32: - return float64(val), true - case uint16: - return float64(val), true - case uint8: - return float64(val), true - case uint: - return float64(val), true - default: - return 0, false - } -} - -func newGorilla() *Gorilla { - return &Gorilla{ - Measurement: "gorilla_block", - BlockInterval: config.Duration(10 * time.Minute), - series: make(map[seriesKey]*seriesBuffer), - } -} - -func init() { - aggregators.Add("gorilla", func() telegraf.Aggregator { - return newGorilla() - }) -} diff --git a/telegraf-patch/plugins/aggregators/gorilla/metric.go b/telegraf-patch/plugins/aggregators/gorilla/metric.go deleted file mode 100644 index 5cd3801c..00000000 --- a/telegraf-patch/plugins/aggregators/gorilla/metric.go +++ /dev/null @@ -1,91 +0,0 @@ -package gorilla - -import ( - "github.com/influxdata/telegraf" -) - -// gorillaBlockMetric wraps an existing metric and exposes an additional binary field named "payload". -type gorillaBlockMetric struct { - telegraf.Metric - payload []byte -} - -func newGorillaBlockMetric(base telegraf.Metric, payload []byte) telegraf.Metric { - return &gorillaBlockMetric{ - Metric: base, - payload: payload, - } -} - -func (m *gorillaBlockMetric) Fields() map[string]interface{} { - fields := m.Metric.Fields() - if m.payload != nil { - fields["payload"] = m.payload - } - return fields -} - -func (m *gorillaBlockMetric) FieldList() []*telegraf.Field { - base := m.Metric.FieldList() - if m.payload == nil { - return base - } - out := make([]*telegraf.Field, len(base)+1) - copy(out, base) - out[len(base)] = &telegraf.Field{Key: "payload", Value: m.payload} - return out -} - -func (m *gorillaBlockMetric) HasField(key string) bool { - if key == "payload" { - return m.payload != nil - } - return m.Metric.HasField(key) -} - -func (m *gorillaBlockMetric) GetField(key string) (interface{}, bool) { - if key == "payload" { - if m.payload == nil { - return nil, false - } - return m.payload, true - } - return m.Metric.GetField(key) -} - -func (m *gorillaBlockMetric) AddField(key string, value interface{}) { - if key == "payload" { - if value == nil { - m.payload = nil - return - } - switch v := value.(type) { - case []byte: - m.payload = append([]byte(nil), v...) - return - case string: - m.payload = []byte(v) - return - } - } - m.Metric.AddField(key, value) -} - -func (m *gorillaBlockMetric) RemoveField(key string) { - if key == "payload" { - m.payload = nil - return - } - m.Metric.RemoveField(key) -} - -func (m *gorillaBlockMetric) Copy() telegraf.Metric { - var payload []byte - if m.payload != nil { - payload = append([]byte(nil), m.payload...) - } - return &gorillaBlockMetric{ - Metric: m.Metric.Copy(), - payload: payload, - } -} diff --git a/telegraf-patch/plugins/aggregators/gorilla/sample.conf b/telegraf-patch/plugins/aggregators/gorilla/sample.conf deleted file mode 100644 index 68ef089a..00000000 --- a/telegraf-patch/plugins/aggregators/gorilla/sample.conf +++ /dev/null @@ -1,17 +0,0 @@ -# Encode numeric series into Gorilla blocks before shipping them to gorilla_s3. -[[aggregators.gorilla]] - ## Flush one Gorilla block per aggregation period. Set the global `period` - ## and this block interval to the same duration (e.g. 10m) and enable - ## drop_original to avoid forwarding uncompressed metrics. - # period = "10m" - # drop_original = true - - ## Measurement name that carries the encoded blocks. - # measurement = "gorilla_block" - - ## Duration of metrics per Gorilla block. Should match the aggregator period. - # block_interval = "10m" - - ## Optional: split a block if the encoded payload would exceed this many - ## bytes. Set to 0 to keep all series in a single object. - # max_object_bytes = 0 diff --git a/telegraf-patch/plugins/aggregators/kll/kll.go b/telegraf-patch/plugins/aggregators/kll/kll.go deleted file mode 100755 index 43bf403f..00000000 --- a/telegraf-patch/plugins/aggregators/kll/kll.go +++ /dev/null @@ -1,123 +0,0 @@ -//go:generate ../../../tools/readme_config_includer/generator -package kll - -import ( - _ "embed" - "fmt" - "slices" - "github.com/influxdata/telegraf" - "github.com/influxdata/telegraf/plugins/aggregators" - "github.com/zzylol/go-kll" -) - -type quantile struct { - seen []float64; // for debugging, save the seen values - sketch *kll.Sketch; -} - -type metric struct { - name string; - fields map[string]*quantile; -} - -type KLL struct { - K int `toml:"k"`; - Quantiles []float64 `toml:"quantiles"`; - WriteSeen bool `toml:"write_seen"`; - - cache map[uint64]*metric; // state for each metric/field, key is metric.HashID() - suffixes map[float64]string; // suffix to attach to output quantiles, e.g.. _p50, _p99, ... -} - -//go:embed sample.conf -var sampleConfig string -func (*KLL) SampleConfig() string { return sampleConfig; } - -func (k *KLL) Init() error { - if k.K < 2 { return fmt.Errorf("Invalid Argument. k must be >= 2 (k=%d)", k.K); } - - k.cache = make(map[uint64]*metric); - k.suffixes = make(map[float64]string); - for _, q := range k.Quantiles { - if q < 0 || q > 1 { return fmt.Errorf("Invalid Argument. Quantiles must be in [0, 1] (q=%f)", q); } - k.suffixes[q] = fmt.Sprintf("_p%d", int(q * 100)); - } - - return nil; -} - -// for each numeric field in each metric, update the backing KLL sketch -func (k *KLL) Add(in telegraf.Metric) { - var id uint64 = in.HashID(); - - // get saved metric - m, ok := k.cache[id]; - if !ok { - k.cache[id] = &metric{name: in.Name(), fields: make(map[string]*quantile)}; - m = k.cache[id]; - } - - // for each field, get associated sketch - var fields []*telegraf.Field = in.FieldList(); - for _, field := range fields { - var val float64; - - // conversion referenced from minmax aggregator - switch field.Value.(type) { - case float64: - val = field.Value.(float64); - case int64: - val = float64(field.Value.(int64)); - case uint64: - val = float64(field.Value.(uint64)); - default: - continue; - } - - // get sketch - sketch, ok := m.fields[field.Key]; - if !ok { - m.fields[field.Key] = &quantile{seen: nil, sketch: kll.New(k.K)}; - if k.WriteSeen { m.fields[field.Key].seen = make([]float64, 0); } - - sketch = m.fields[field.Key]; - } - - if k.WriteSeen { sketch.seen = append(sketch.seen, val); } - - sketch.sketch.Update(val); - } -} - -func (k *KLL) Push(acc telegraf.Accumulator) { - for _, m := range k.cache { - out := make(map[string]any); - - fields := m.fields; - for name, sketch := range fields { - // get the desired quantile - cdf := sketch.sketch.CDF(); - for q, str := range k.suffixes { out[name + str] = cdf.Query(q); } - - if k.WriteSeen { - slices.Sort(sketch.seen) - out[name + "_seen"] = fmt.Sprintf("%v", sketch.seen); - } - } - - acc.AddSummary(m.name + "_KLL", out, nil); - } -} - -func (k *KLL) Reset() { - for _, m := range k.cache { - for key := range m.fields { - if k.WriteSeen { clear(m.fields[key].seen); } - m.fields[key].sketch = kll.New(k.K); - } - } -} - -func init() { - aggregators.Add("kll", func() telegraf.Aggregator { return &KLL{}; }); -} diff --git a/telegraf-patch/plugins/aggregators/kll/sample.conf b/telegraf-patch/plugins/aggregators/kll/sample.conf deleted file mode 100755 index 524117ea..00000000 --- a/telegraf-patch/plugins/aggregators/kll/sample.conf +++ /dev/null @@ -1,13 +0,0 @@ -# Outputs quantiles for each numeric field for each metric -# Aggregated metric has name _KLL -[[aggregators.kll]] - ## Size of topmost compactor - ## Total memory used by sketch is 3*k + lg(n) - k = 256 - - ## Quantiles to retrieve. Must be in range [0, 1] and float (e.g. 0 must be 0.0) - ## Values will be output as field_p, e.g. cpu_p50 - quantiles = [0.5, 0.99] - - ## Debugging option to also output the sorted ingested values (with suffix _seen) - write_seen = false diff --git a/telegraf-patch/plugins/aggregators/sketchutil/countmin.go b/telegraf-patch/plugins/aggregators/sketchutil/countmin.go deleted file mode 100644 index 09aff4d6..00000000 --- a/telegraf-patch/plugins/aggregators/sketchutil/countmin.go +++ /dev/null @@ -1,137 +0,0 @@ -package sketchutil - -import ( - "bytes" - "encoding/binary" - "fmt" - "hash" - "io" - "math" - - "github.com/cespare/xxhash/v2" -) - -type CountMinSketch struct { - rows int - cols int - total float64 - table []float32 - salts []uint64 - hashes []hash.Hash64 - topk *TopKHeap -} - -func NewCountMinSketch(rows, cols int, seed uint64, topk int) (*CountMinSketch, error) { - if rows <= 0 || cols <= 0 { - return nil, fmt.Errorf("countmin: invalid dimensions rows=%d cols=%d", rows, cols) - } - cms := &CountMinSketch{ - rows: rows, - cols: cols, - table: make([]float32, rows*cols), - salts: make([]uint64, rows), - hashes: make([]hash.Hash64, rows), - topk: NewTopKHeap(topk), - } - for i := 0; i < rows; i++ { - cms.salts[i] = mixSeed(seed, uint64(i)) - cms.hashes[i] = xxhash.New() - } - return cms, nil -} - -func (c *CountMinSketch) Insert(key string, weight float64) { - if weight == 0 { - return - } - estimate := math.MaxFloat64 - for row := 0; row < c.rows; row++ { - h := c.hashes[row] - h.Reset() - var saltBuf [8]byte - binary.LittleEndian.PutUint64(saltBuf[:], c.salts[row]) - h.Write(saltBuf[:]) - io.WriteString(h, key) - idx := int(h.Sum64() % uint64(c.cols)) - c.table[row*c.cols+idx] += float32(weight) - value := float64(c.table[row*c.cols+idx]) - if value < estimate { - estimate = value - } - } - c.total += weight - if estimate == math.MaxFloat64 { - estimate = 0 - } - if c.topk != nil { - c.topk.Update(key, estimate) - } -} - -func (c *CountMinSketch) Rows() int { - return c.rows -} - -func (c *CountMinSketch) Columns() int { - return c.cols -} - -func (c *CountMinSketch) Total() float64 { - return c.total -} - -func (c *CountMinSketch) TopKEntries() []TopKEntry { - if c.topk == nil { - return nil - } - return c.topk.Entries() -} - -func (c *CountMinSketch) MarshalBinary() ([]byte, error) { - buf := &bytes.Buffer{} - if err := binary.Write(buf, binary.BigEndian, uint32(countMinMagic)); err != nil { - return nil, err - } - if err := binary.Write(buf, binary.BigEndian, uint16(countMinVersion)); err != nil { - return nil, err - } - if err := binary.Write(buf, binary.BigEndian, uint16(c.rows)); err != nil { - return nil, err - } - if err := binary.Write(buf, binary.BigEndian, uint32(c.cols)); err != nil { - return nil, err - } - if err := binary.Write(buf, binary.BigEndian, c.total); err != nil { - return nil, err - } - if err := binary.Write(buf, binary.BigEndian, uint16(len(c.salts))); err != nil { - return nil, err - } - for _, salt := range c.salts { - if err := binary.Write(buf, binary.BigEndian, salt); err != nil { - return nil, err - } - } - for _, v := range c.table { - if err := binary.Write(buf, binary.BigEndian, v); err != nil { - return nil, err - } - } - return buf.Bytes(), nil -} - -const ( - countMinMagic = 0x434d5331 // "CMS1" - countMinVersion = 1 -) - -func mixSeed(base uint64, row uint64) uint64 { - const prime uint64 = 0x100000001b3 - value := base ^ (row * prime) - value ^= value >> 33 - value *= 0xff51afd7ed558ccd - value ^= value >> 33 - value *= 0xc4ceb9fe1a85ec53 - value ^= value >> 33 - return value -} diff --git a/telegraf-patch/plugins/aggregators/sketchutil/topk.go b/telegraf-patch/plugins/aggregators/sketchutil/topk.go deleted file mode 100644 index 284b225c..00000000 --- a/telegraf-patch/plugins/aggregators/sketchutil/topk.go +++ /dev/null @@ -1,139 +0,0 @@ -package sketchutil - -import "sort" - -// TopKEntry represents a heavy hitter entry. -type TopKEntry struct { - Key string `json:"key"` - Count float64 `json:"count"` -} - -type heapItem struct { - key string - count float64 -} - -// TopKHeap maintains approximate heavy hitters using a bounded min-heap -// structure inspired by github.com/Froot-NetSys/promsketch. -type TopKHeap struct { - heap []heapItem - limit int -} - -// NewTopKHeap creates a heap limited to k entries. Returns nil if k <= 0. -func NewTopKHeap(k int) *TopKHeap { - if k <= 0 { - return nil - } - return &TopKHeap{ - heap: make([]heapItem, 0, k), - limit: k, - } -} - -func (h *TopKHeap) leftChild(i int) int { - return 2*i + 1 -} - -func (h *TopKHeap) rightChild(i int) int { - return 2*i + 2 -} - -func (h *TopKHeap) parent(i int) int { - return (i - 1) / 2 -} - -func (h *TopKHeap) swap(i, j int) { - h.heap[i], h.heap[j] = h.heap[j], h.heap[i] -} - -func (h *TopKHeap) updateOrderDown(i int) bool { - n := len(h.heap) - orig := i - for i < n { - l := h.leftChild(i) - r := h.rightChild(i) - smallest := i - if l < n && h.heap[l].count < h.heap[smallest].count { - smallest = l - } - if r < n && h.heap[r].count < h.heap[smallest].count { - smallest = r - } - if smallest == i { - break - } - h.swap(i, smallest) - i = smallest - } - return i != orig -} - -func (h *TopKHeap) updateOrderUp(i int) { - for i > 0 { - p := h.parent(i) - if h.heap[p].count > h.heap[i].count { - h.swap(p, i) - i = p - } else { - break - } - } -} - -// Update inserts or updates the given key with its latest approximate count. -// For efficiency we linearly scan to find an existing key (as in promsketch) -// to avoid auxiliary maps and their GC overhead. -func (h *TopKHeap) Update(key string, count float64) { - if h == nil || count <= 0 { - return - } - if idx, ok := h.find(key); ok { - h.heap[idx].count = count - if !h.updateOrderDown(idx) { - h.updateOrderUp(idx) - } - return - } - if len(h.heap) < h.limit { - h.heap = append(h.heap, heapItem{ - key: key, - count: count, - }) - h.updateOrderUp(len(h.heap) - 1) - return - } - if len(h.heap) == 0 || h.heap[0].count >= count { - return - } - h.heap[0].key = key - h.heap[0].count = count - h.updateOrderDown(0) -} - -func (h *TopKHeap) find(key string) (int, bool) { - for i, item := range h.heap { - if item.key == key { - return i, true - } - } - return -1, false -} - -// Entries returns the heap contents sorted descending by count. -func (h *TopKHeap) Entries() []TopKEntry { - if h == nil || len(h.heap) == 0 { - return nil - } - entries := make([]TopKEntry, len(h.heap)) - for i, item := range h.heap { - entries[i] = TopKEntry{ - Key: item.key, - Count: item.count, - } - } - sort.Slice(entries, func(i, j int) bool { - return entries[i].Count > entries[j].Count - }) - return entries -} diff --git a/telegraf-patch/plugins/aggregators/sketchutil/util.go b/telegraf-patch/plugins/aggregators/sketchutil/util.go deleted file mode 100644 index e55d2152..00000000 --- a/telegraf-patch/plugins/aggregators/sketchutil/util.go +++ /dev/null @@ -1,32 +0,0 @@ -package sketchutil - -func ToFloat(v interface{}) (float64, bool) { - switch value := v.(type) { - case float64: - return value, true - case float32: - return float64(value), true - case int64: - return float64(value), true - case int32: - return float64(value), true - case int16: - return float64(value), true - case int8: - return float64(value), true - case int: - return float64(value), true - case uint64: - return float64(value), true - case uint32: - return float64(value), true - case uint16: - return float64(value), true - case uint8: - return float64(value), true - case uint: - return float64(value), true - default: - return 0, false - } -}