mirror of
https://github.com/hanzoai/traces.git
synced 2026-10-30 18:20:41 +00:00
* Add metrics-generator component * Add some initial logic to make the generator module optional * Rename tempo_distributor_metrics_generator_clients metric * Periodical check in: add remote-write setup * jsonnet: rename generator -> metrics_generator * Periodical check in: add instance, copy servicegraphprocessor over, add scrape endpoint * Fix make test * Support spanmetrics via OTel collector Generator is now able to run an embedded OTel collector. This allows to run any number of processors, which includes spanmetrics. This collector uses a custom metrics exporter and a remote write client. Metrics generated by the spanmetrics processor are converted to Prometheus and appended to the remote write client's WAL, which will export the metric data after commit. * Implement internal spanmetrics processor * Add latency metrics * Only loop one for latency metrics * Improve how metrics are collected, some bug fixes Changes: - Remote write was constantly failing with “out of order samples”. Fix: apparently the timestamp has to be in milliseconds, not in seconds. There is not a lot of documentation on the Prometheus interfaces, but the cortex proto has a field timestampMs. - After this remote write was failing with label name \"le\" is not unique: invalid sample. Fix: there was a bug in the append logic of the labels array. Since you were reusing the labels variable in a for-loop, the le label was added multiple times. Initially I thought pushing metrics on every push request was causing the out-of-order issues, so I refactored the processor a bit to have both a PushSpans and a CollectMetrics method. CollectMetrics is called every 15s with a storage.Appender. This way we can also use a single appender for multiple processors, This wasn’t necessary to get it working, but I think this is the logic we want to end up with (i.e. don’t send samples too often to keep DPM low). * make vendor-check * Add spanmetrics_test.go * jsonnet: add metrics-generator service * Add distributor.enable_metrics_generator_ring; some bug fixes * Remove unused Registerer and scrape endpoint code; sprinkle some more TODO's around * Add crude mechanism to trim active series in span metrics processor * span metrics: add delete_after_last_update * Admit config parameters in spanmetrics processor Histogram buckets and additional dimensions can be configured * Initial implementation of service graphs processor * Hook up service graph processor Store - make sure tests pass - fix copy-paste error in shouldEvictHead Service graphs - ensure metric names align with agent - server latencies were overwriting client latencies - expose operational metrics * Rename metrics from generator_processor_... to metrics_generator_processor_... * Set up BasicLifecycler, load balance spans across metrics-generator * Replace processor.CollectMetrics with a custom prometheus.Registerer Changes: Service graphs processor - replaced code to manually build metrics with CounterVec and HistogramVec - aligned metric names with the agent - commented out code around collectCh, this was leaking memory Span metrics processor - replaced code to manually build metrics with CounterVec and HistogramVec Introduced Registry to bridge prometheus.Registerer and storage.Appender Add support for configuring external labels Fix generator_test.go * make vendor-check * Refactor remote-write structures, add tests * Add add_instance_id_label config * Split write requests up to a max msg size Builds write requests up to a max size and sends them sequentially. * Service graphs: collect edges when completed When an edge is completed, it's sent to be collected by a number of goroutines that call collectEdge. Edges can also be collected when expired. * Store: interface: give paramaters names * Service graphs: Close closech * Simplify evict method name * Move remote_write to its own method * Add metrics_generator_processors to overrides, dynamically create/remove processors * Add tests for split requests in remote write appender * Refactor distributor code a bit * Make collection interval configurable * Add concurrency test for instance * Minor tweaks * Add metrics to track failed active processors updates * Tweak remote write metrics * Add exemplars support * Fix latency measurement in spanmetrics processor * Fix typo * Change metrics generator enabled config var * Return ready when generator is registered in the ring * Add read-only mode during generator shutdown * Configure docker-compose distributed to work with metrics generator * Enable pushing bytes to metrics-generator * Remove cortex from distributed example * Replace cortexpb -> prompb, move all remote-write stuff into remotewrite package * Set context timeout when collecting metrics, remove hola * Protect readOnly from concurrency * Update vendor * Rename metrics prefix span_metrics to spanmetrics This matches the naming from OTel spanmetrics processor * Always add le="+Inf" bucket when not set already * Fix lint errors * Fix compactor test Test was writing tempodb.Trace to the WAL, instead of tempodb.TraceBytes * Move expire goroutine to collector workers * Move metrics_generator_enabled to the top-level config * Update docker-compose distributor example * Use snappy from klauspost/compress * Fix distributor test after moving metrics_generator_enabled flag * Reorganize how processors are updated, simplify use of locks * Stub time in processor tests * Fix how label slices are handled in registry * jsonnet: disable metrics-generator by default * Regenerate kube manifests * Pin time.Now in test with registry * Update CHANGELOG.md Co-authored-by: Koenraad Verheyden <[email protected]>
139 lines
2.7 KiB
Go
139 lines
2.7 KiB
Go
package units
|
|
|
|
import (
|
|
"errors"
|
|
"fmt"
|
|
"strings"
|
|
)
|
|
|
|
var (
|
|
siUnits = []string{"", "K", "M", "G", "T", "P", "E"}
|
|
)
|
|
|
|
func ToString(n int64, scale int64, suffix, baseSuffix string) string {
|
|
mn := len(siUnits)
|
|
out := make([]string, mn)
|
|
for i, m := range siUnits {
|
|
if n%scale != 0 || i == 0 && n == 0 {
|
|
s := suffix
|
|
if i == 0 {
|
|
s = baseSuffix
|
|
}
|
|
out[mn-1-i] = fmt.Sprintf("%d%s%s", n%scale, m, s)
|
|
}
|
|
n /= scale
|
|
if n == 0 {
|
|
break
|
|
}
|
|
}
|
|
return strings.Join(out, "")
|
|
}
|
|
|
|
// Below code ripped straight from http://golang.org/src/pkg/time/format.go?s=33392:33438#L1123
|
|
var errLeadingInt = errors.New("units: bad [0-9]*") // never printed
|
|
|
|
// leadingInt consumes the leading [0-9]* from s.
|
|
func leadingInt(s string) (x int64, rem string, err error) {
|
|
i := 0
|
|
for ; i < len(s); i++ {
|
|
c := s[i]
|
|
if c < '0' || c > '9' {
|
|
break
|
|
}
|
|
if x >= (1<<63-10)/10 {
|
|
// overflow
|
|
return 0, "", errLeadingInt
|
|
}
|
|
x = x*10 + int64(c) - '0'
|
|
}
|
|
return x, s[i:], nil
|
|
}
|
|
|
|
func ParseUnit(s string, unitMap map[string]float64) (int64, error) {
|
|
// [-+]?([0-9]*(\.[0-9]*)?[a-z]+)+
|
|
orig := s
|
|
f := float64(0)
|
|
neg := false
|
|
|
|
// Consume [-+]?
|
|
if s != "" {
|
|
c := s[0]
|
|
if c == '-' || c == '+' {
|
|
neg = c == '-'
|
|
s = s[1:]
|
|
}
|
|
}
|
|
// Special case: if all that is left is "0", this is zero.
|
|
if s == "0" {
|
|
return 0, nil
|
|
}
|
|
if s == "" {
|
|
return 0, errors.New("units: invalid " + orig)
|
|
}
|
|
for s != "" {
|
|
g := float64(0) // this element of the sequence
|
|
|
|
var x int64
|
|
var err error
|
|
|
|
// The next character must be [0-9.]
|
|
if !(s[0] == '.' || ('0' <= s[0] && s[0] <= '9')) {
|
|
return 0, errors.New("units: invalid " + orig)
|
|
}
|
|
// Consume [0-9]*
|
|
pl := len(s)
|
|
x, s, err = leadingInt(s)
|
|
if err != nil {
|
|
return 0, errors.New("units: invalid " + orig)
|
|
}
|
|
g = float64(x)
|
|
pre := pl != len(s) // whether we consumed anything before a period
|
|
|
|
// Consume (\.[0-9]*)?
|
|
post := false
|
|
if s != "" && s[0] == '.' {
|
|
s = s[1:]
|
|
pl := len(s)
|
|
x, s, err = leadingInt(s)
|
|
if err != nil {
|
|
return 0, errors.New("units: invalid " + orig)
|
|
}
|
|
scale := 1.0
|
|
for n := pl - len(s); n > 0; n-- {
|
|
scale *= 10
|
|
}
|
|
g += float64(x) / scale
|
|
post = pl != len(s)
|
|
}
|
|
if !pre && !post {
|
|
// no digits (e.g. ".s" or "-.s")
|
|
return 0, errors.New("units: invalid " + orig)
|
|
}
|
|
|
|
// Consume unit.
|
|
i := 0
|
|
for ; i < len(s); i++ {
|
|
c := s[i]
|
|
if c == '.' || ('0' <= c && c <= '9') {
|
|
break
|
|
}
|
|
}
|
|
u := s[:i]
|
|
s = s[i:]
|
|
unit, ok := unitMap[u]
|
|
if !ok {
|
|
return 0, errors.New("units: unknown unit " + u + " in " + orig)
|
|
}
|
|
|
|
f += g * unit
|
|
}
|
|
|
|
if neg {
|
|
f = -f
|
|
}
|
|
if f < float64(-1<<63) || f > float64(1<<63-1) {
|
|
return 0, errors.New("units: overflow parsing unit")
|
|
}
|
|
return int64(f), nil
|
|
}
|