mirror of
https://github.com/hanzoai/o11y.git
synced 2026-10-15 12:58:33 +00:00
zeekay
c3dc2f1ecf
HIP-0132 unified o11y_logs, o11y_traces and o11y_metrics into one `event`
database and dropped them. The commit that restored the reads repointed
three files and said so; it was thirty. The rest went on naming databases
that no longer exist, so trace funnels, the Kafka and queue panels, the
v3/v4 logs and traces builders and the PromQL adapter each answered
Code: 81 UNKNOWN_DATABASE, and the golden tests agreed with them because
they pinned the same dead SQL.
pkg/telemetryplane is the one address for the name, imports nothing, and
is reachable from every depth that needs it. Its test walks the tree and
fails the build on a literal that names a dropped database, tests
included — a golden encoding broken SQL is a second vote for the bug, not
coverage.
Two service views turned a failed read into an offer to instrument an
already-instrumented application. A failure now shows the failure, and
onboarding is offered on a successful empty read.
Cherry-picked from salvage/blue-wt-o11y-readers 83ecbec03c.
209 lines
7.6 KiB
Go
209 lines
7.6 KiB
Go
package telemetrymeter
|
|
|
|
import (
|
|
"time"
|
|
|
|
"github.com/hanzoai/o11y/pkg/errors"
|
|
"github.com/hanzoai/o11y/pkg/telemetryplane"
|
|
"github.com/hanzoai/o11y/pkg/types/metrictypes"
|
|
)
|
|
|
|
// DBName is NOT the event plane. o11y_meter holds METERING — the billing-grade
|
|
// usage counters — which HIP-0132 did not unify and no writer has left. The name
|
|
// is spelled in pkg/telemetryplane beside the plane it is not part of, so
|
|
// "which databases does this binary name?" has one answer.
|
|
const (
|
|
DBName = telemetryplane.MeterDBName
|
|
SamplesTableName = "distributed_samples"
|
|
SamplesLocalTableName = "samples"
|
|
SamplesAgg1dTableName = "distributed_samples_agg_1d"
|
|
SamplesAgg1dLocalTableName = "samples_agg_1d"
|
|
)
|
|
|
|
var (
|
|
oneMonthInMilliseconds = uint64(time.Hour.Milliseconds() * 24 * 30)
|
|
|
|
// when the query requests for almost 1 day, but not exactly 1 day, we need to add an offset to the end time
|
|
// to make sure that we are using the correct table
|
|
// this is because the start gets adjusted to the nearest step interval and uses the 5m table for 4m step interval
|
|
// leading to time series that doesn't best represent the rate of change.
|
|
offsetBucket = uint64(1 * time.Hour.Milliseconds())
|
|
)
|
|
|
|
// start and end are in milliseconds
|
|
// we have two tables for samples
|
|
// 1. distributed_samples
|
|
// 2. distributed_samples_agg_1d - for queries with time range above or equal to 30 days
|
|
// if the `timeAggregation` is `count_distinct` we can't use the aggregated tables because they don't support it.
|
|
func WhichSamplesTableToUse(
|
|
start, end uint64,
|
|
metricType metrictypes.Type,
|
|
timeAggregation metrictypes.TimeAggregation,
|
|
tableHints *metrictypes.MetricTableHints,
|
|
) string {
|
|
|
|
// if we have a hint for the table, we need to use it
|
|
// the hint will be used to override the default table selection logic
|
|
if tableHints != nil {
|
|
if tableHints.SamplesTableName != "" {
|
|
return tableHints.SamplesTableName
|
|
}
|
|
}
|
|
|
|
// if the time aggregation is count_distinct, we need to use the distributed_samples table
|
|
// because the aggregated tables don't support count_distinct
|
|
if timeAggregation == metrictypes.TimeAggregationCountDistinct {
|
|
return SamplesTableName
|
|
}
|
|
|
|
if end-start < oneMonthInMilliseconds+offsetBucket {
|
|
return SamplesTableName
|
|
}
|
|
return SamplesAgg1dTableName
|
|
|
|
}
|
|
|
|
func AggregationColumnForSamplesTable(
|
|
start, end uint64,
|
|
metricType metrictypes.Type,
|
|
temporality metrictypes.Temporality,
|
|
timeAggregation metrictypes.TimeAggregation,
|
|
tableHints *metrictypes.MetricTableHints,
|
|
) (string, error) {
|
|
tableName := WhichSamplesTableToUse(start, end, metricType, timeAggregation, tableHints)
|
|
var aggregationColumn string
|
|
switch temporality {
|
|
case metrictypes.Delta:
|
|
switch tableName {
|
|
case SamplesTableName:
|
|
switch timeAggregation {
|
|
case metrictypes.TimeAggregationLatest:
|
|
aggregationColumn = "anyLast(value)"
|
|
case metrictypes.TimeAggregationSum:
|
|
aggregationColumn = "sum(value)"
|
|
case metrictypes.TimeAggregationAvg:
|
|
aggregationColumn = "avg(value)"
|
|
case metrictypes.TimeAggregationMin:
|
|
aggregationColumn = "min(value)"
|
|
case metrictypes.TimeAggregationMax:
|
|
aggregationColumn = "max(value)"
|
|
case metrictypes.TimeAggregationCount:
|
|
aggregationColumn = "count(value)"
|
|
case metrictypes.TimeAggregationCountDistinct:
|
|
aggregationColumn = "countDistinct(value)"
|
|
case metrictypes.TimeAggregationRate, metrictypes.TimeAggregationIncrease: // only these two options give meaningful results
|
|
aggregationColumn = "sum(value)"
|
|
}
|
|
case SamplesAgg1dTableName:
|
|
switch timeAggregation {
|
|
case metrictypes.TimeAggregationLatest:
|
|
aggregationColumn = "anyLast(last)"
|
|
case metrictypes.TimeAggregationSum:
|
|
aggregationColumn = "sum(sum)"
|
|
case metrictypes.TimeAggregationAvg:
|
|
aggregationColumn = "sum(sum) / sum(count)"
|
|
case metrictypes.TimeAggregationMin:
|
|
aggregationColumn = "min(min)"
|
|
case metrictypes.TimeAggregationMax:
|
|
aggregationColumn = "max(max)"
|
|
case metrictypes.TimeAggregationCount:
|
|
aggregationColumn = "sum(count)"
|
|
// count_distinct is not supported in aggregated tables
|
|
case metrictypes.TimeAggregationRate, metrictypes.TimeAggregationIncrease: // only these two options give meaningful results
|
|
aggregationColumn = "sum(sum)"
|
|
}
|
|
}
|
|
case metrictypes.Cumulative:
|
|
switch tableName {
|
|
case SamplesTableName:
|
|
switch timeAggregation {
|
|
case metrictypes.TimeAggregationLatest:
|
|
aggregationColumn = "anyLast(value)"
|
|
case metrictypes.TimeAggregationSum:
|
|
aggregationColumn = "sum(value)"
|
|
case metrictypes.TimeAggregationAvg:
|
|
aggregationColumn = "avg(value)"
|
|
case metrictypes.TimeAggregationMin:
|
|
aggregationColumn = "min(value)"
|
|
case metrictypes.TimeAggregationMax:
|
|
aggregationColumn = "max(value)"
|
|
case metrictypes.TimeAggregationCount:
|
|
aggregationColumn = "count(value)"
|
|
case metrictypes.TimeAggregationCountDistinct:
|
|
aggregationColumn = "countDistinct(value)"
|
|
case metrictypes.TimeAggregationRate, metrictypes.TimeAggregationIncrease: // only these two options give meaningful results
|
|
aggregationColumn = "max(value)"
|
|
}
|
|
case SamplesAgg1dTableName:
|
|
switch timeAggregation {
|
|
case metrictypes.TimeAggregationLatest:
|
|
aggregationColumn = "anyLast(last)"
|
|
case metrictypes.TimeAggregationSum:
|
|
aggregationColumn = "sum(sum)"
|
|
case metrictypes.TimeAggregationAvg:
|
|
aggregationColumn = "sum(sum) / sum(count)"
|
|
case metrictypes.TimeAggregationMin:
|
|
aggregationColumn = "min(min)"
|
|
case metrictypes.TimeAggregationMax:
|
|
aggregationColumn = "max(max)"
|
|
case metrictypes.TimeAggregationCount:
|
|
aggregationColumn = "sum(count)"
|
|
// count_distinct is not supported in aggregated tables
|
|
case metrictypes.TimeAggregationRate, metrictypes.TimeAggregationIncrease: // only these two options give meaningful results
|
|
aggregationColumn = "max(max)"
|
|
}
|
|
}
|
|
|
|
case metrictypes.Unspecified:
|
|
switch tableName {
|
|
case SamplesTableName:
|
|
switch timeAggregation {
|
|
case metrictypes.TimeAggregationLatest:
|
|
aggregationColumn = "anyLast(value)"
|
|
case metrictypes.TimeAggregationSum:
|
|
aggregationColumn = "sum(value)"
|
|
case metrictypes.TimeAggregationAvg:
|
|
aggregationColumn = "avg(value)"
|
|
case metrictypes.TimeAggregationMin:
|
|
aggregationColumn = "min(value)"
|
|
case metrictypes.TimeAggregationMax:
|
|
aggregationColumn = "max(value)"
|
|
case metrictypes.TimeAggregationCount:
|
|
aggregationColumn = "count(value)"
|
|
case metrictypes.TimeAggregationCountDistinct:
|
|
aggregationColumn = "countDistinct(value)"
|
|
case metrictypes.TimeAggregationRate, metrictypes.TimeAggregationIncrease: // ideally, this should never happen
|
|
aggregationColumn = "sum(value)"
|
|
}
|
|
case SamplesAgg1dTableName:
|
|
switch timeAggregation {
|
|
case metrictypes.TimeAggregationLatest:
|
|
aggregationColumn = "anyLast(last)"
|
|
case metrictypes.TimeAggregationSum:
|
|
aggregationColumn = "sum(sum)"
|
|
case metrictypes.TimeAggregationAvg:
|
|
aggregationColumn = "sum(sum) / sum(count)"
|
|
case metrictypes.TimeAggregationMin:
|
|
aggregationColumn = "min(min)"
|
|
case metrictypes.TimeAggregationMax:
|
|
aggregationColumn = "max(max)"
|
|
case metrictypes.TimeAggregationCount:
|
|
aggregationColumn = "sum(count)"
|
|
// count_distinct is not supported in aggregated tables
|
|
case metrictypes.TimeAggregationRate, metrictypes.TimeAggregationIncrease: // ideally, this should never happen
|
|
aggregationColumn = "sum(sum)"
|
|
}
|
|
}
|
|
|
|
}
|
|
|
|
if aggregationColumn == "" {
|
|
return "", errors.Newf(
|
|
errors.TypeInvalidInput,
|
|
errors.CodeInvalidInput,
|
|
"invalid time aggregation, should be one of the following: [`latest`, `sum`, `avg`, `min`, `max`, `count`, `rate`, `increase`]",
|
|
)
|
|
}
|
|
return aggregationColumn, nil
|
|
}
|