mirror of
https://github.com/hanzoai/o11y.git
synced 2026-10-15 12:58:33 +00:00
zeekayandHanzo Dev
cc111f0630
A telemetry field key name arrives as JSON on /api/v5/query_range and is
replayed out of stored dashboards and alert rules. It lands in two syntactic
positions and a bound parameter can occupy NEITHER:
- a backtick-quoted IDENTIFIER — the SELECT alias, the GROUP BY term, the
ORDER BY term;
- a single-quoted LITERAL — the accessors that read one attribute out of a
column, JSONExtractString(labels, '<name>') and friends.
Nothing quoted it. The rendering was written inline about forty times, always
as "`"+name+"`" or '%s', and the one call that sat next to a name —
sqlbuilder.Escape — replaces "$" with "$$" for the builder's own placeholder
syntax and is not SQL escaping. Measured against the real builder, a group-by
name of
le`,(select/**/grouparray(name)/**/from/**/system.tables)/**/as/**/`x
emitted that subquery as an ADDITIONAL SELECT COLUMN, whose rows come back in
the response body; "/**/" stands in for the spaces the builder strips and the
payload is all-lowercase so nothing folds it. A name carrying "'" breaks the
labels accessor the same way, and one ending in "\" eats whichever delimiter
closes it.
The request validator is not where this can be fixed. The rule evaluator builds
a QueryRangeRequest and calls the querier directly, so a stored rule's name
never meets it; and validateOrderBy, which does gate the DIRECTION strictly to
asc|desc, gates the sibling key NAME against an allowlist it builds out of the
caller's own GroupBy — so a name repeated in both fields validates itself.
Metrics settles it: a metric label is free-form, an unrecognised name falls back
to the labels column BY DESIGN, and there is no key allowlist to fail against.
So the guard goes where the name becomes text. QuoteIdent and EscapeLiteral are
the one renderer, in querybuilder, which every signal package already imports;
the ~40 inline spellings now call them. Both escape the backslash BEFORE the
delimiter, or a trailing backslash would consume the delimiter being added. A
name with none of those characters — every real attribute key, which is a dotted
identifier — renders byte for byte as it did, which is why the golden-SQL suites
in all six signal packages are unchanged.
The test asserts on emitted SQL and structurally rather than textually: the
payload MAY appear, quoted, as the inert name of a label nobody has, but it must
never appear in the executable skeleton — the text left once every quoted region
is removed. That is the property an injection breaks, so it fails if any layer
underneath stops escaping.
Co-authored-by: Hanzo Dev <[email protected]>
398 lines
14 KiB
Go
398 lines
14 KiB
Go
package telemetrymeter
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"log/slog"
|
|
|
|
"github.com/hanzoai/o11y/pkg/datastoresql"
|
|
|
|
"github.com/hanzo-ds/sqlbuilder"
|
|
"github.com/hanzoai/o11y/pkg/errors"
|
|
"github.com/hanzoai/o11y/pkg/factory"
|
|
"github.com/hanzoai/o11y/pkg/querybuilder"
|
|
"github.com/hanzoai/o11y/pkg/telemetrymetrics"
|
|
"github.com/hanzoai/o11y/pkg/types/metrictypes"
|
|
qbtypes "github.com/hanzoai/o11y/pkg/types/querybuildertypes/querybuildertypesv5"
|
|
"github.com/hanzoai/o11y/pkg/types/telemetrytypes"
|
|
)
|
|
|
|
type meterQueryStatementBuilder struct {
|
|
logger *slog.Logger
|
|
metadataStore telemetrytypes.MetadataStore
|
|
fm qbtypes.FieldMapper
|
|
cb qbtypes.ConditionBuilder
|
|
metricsStatementBuilder *telemetrymetrics.MetricQueryStatementBuilder
|
|
}
|
|
|
|
var _ qbtypes.StatementBuilder[qbtypes.MetricAggregation] = (*meterQueryStatementBuilder)(nil)
|
|
|
|
func NewMeterQueryStatementBuilder(
|
|
settings factory.ProviderSettings,
|
|
metadataStore telemetrytypes.MetadataStore,
|
|
fieldMapper qbtypes.FieldMapper,
|
|
conditionBuilder qbtypes.ConditionBuilder,
|
|
metricsStatementBuilder *telemetrymetrics.MetricQueryStatementBuilder,
|
|
) *meterQueryStatementBuilder {
|
|
metricsSettings := factory.NewScopedProviderSettings(settings, "github.com/hanzoai/o11y/pkg/telemetrymeter")
|
|
|
|
return &meterQueryStatementBuilder{
|
|
logger: metricsSettings.Logger(),
|
|
metadataStore: metadataStore,
|
|
fm: fieldMapper,
|
|
cb: conditionBuilder,
|
|
metricsStatementBuilder: metricsStatementBuilder,
|
|
}
|
|
}
|
|
|
|
func (b *meterQueryStatementBuilder) Build(
|
|
ctx context.Context,
|
|
start uint64,
|
|
end uint64,
|
|
_ qbtypes.RequestType,
|
|
query qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation],
|
|
variables map[string]qbtypes.VariableItem,
|
|
) (*qbtypes.Statement, error) {
|
|
keySelectors := telemetrymetrics.GetKeySelectors(query)
|
|
keys, _, err := b.metadataStore.GetKeysMulti(ctx, keySelectors)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
start, end = querybuilder.AdjustedMetricTimeRange(start, end, uint64(query.StepInterval.Seconds()), query)
|
|
|
|
return b.buildPipelineStatement(ctx, start, end, query, keys, variables)
|
|
}
|
|
|
|
func (b *meterQueryStatementBuilder) buildPipelineStatement(
|
|
ctx context.Context,
|
|
start, end uint64,
|
|
query qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation],
|
|
keys map[string][]*telemetrytypes.TelemetryFieldKey,
|
|
variables map[string]qbtypes.VariableItem,
|
|
) (*qbtypes.Statement, error) {
|
|
var (
|
|
cteFragments []string
|
|
cteArgs [][]any
|
|
)
|
|
|
|
if qbtypes.CanShortCircuitDelta(query.Aggregations[0]) {
|
|
// spatial_aggregation_cte directly for certain delta queries
|
|
if frag, args, err := b.buildTemporalAggDeltaFastPath(ctx, start, end, query, keys, variables); err != nil {
|
|
return nil, err
|
|
} else if frag != "" {
|
|
cteFragments = append(cteFragments, frag)
|
|
cteArgs = append(cteArgs, args)
|
|
}
|
|
} else {
|
|
// temporal_aggregation_cte
|
|
if frag, args, err := b.buildTemporalAggregationCTE(ctx, start, end, query, keys, variables); err != nil {
|
|
return nil, err
|
|
} else if frag != "" {
|
|
cteFragments = append(cteFragments, frag)
|
|
cteArgs = append(cteArgs, args)
|
|
}
|
|
|
|
// spatial_aggregation_cte
|
|
if frag, args, err := b.buildSpatialAggregationCTE(ctx, start, end, query, keys); err != nil {
|
|
return nil, err
|
|
} else if frag != "" {
|
|
cteFragments = append(cteFragments, frag)
|
|
cteArgs = append(cteArgs, args)
|
|
}
|
|
}
|
|
|
|
// final SELECT
|
|
return b.metricsStatementBuilder.BuildFinalSelect(cteFragments, cteArgs, query)
|
|
}
|
|
|
|
func (b *meterQueryStatementBuilder) buildTemporalAggDeltaFastPath(
|
|
ctx context.Context,
|
|
start, end uint64,
|
|
query qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation],
|
|
keys map[string][]*telemetrytypes.TelemetryFieldKey,
|
|
variables map[string]qbtypes.VariableItem,
|
|
) (string, []any, error) {
|
|
var filterWhere querybuilder.PreparedWhereClause
|
|
var err error
|
|
stepSec := int64(query.StepInterval.Seconds())
|
|
|
|
sb := sqlbuilder.NewSelectBuilder()
|
|
|
|
sb.SelectMore(fmt.Sprintf(
|
|
"toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(%d)) AS ts",
|
|
stepSec,
|
|
))
|
|
for _, g := range query.GroupBy {
|
|
col, err := b.fm.ColumnExpressionFor(ctx, start, end, &g.TelemetryFieldKey, keys)
|
|
if err != nil {
|
|
return "", nil, err
|
|
}
|
|
sb.SelectMore(col)
|
|
}
|
|
|
|
tbl := WhichSamplesTableToUse(start, end, query.Aggregations[0].Type, query.Aggregations[0].TimeAggregation, query.Aggregations[0].TableHints)
|
|
aggCol, err := AggregationColumnForSamplesTable(start, end, query.Aggregations[0].Type, query.Aggregations[0].Temporality, query.Aggregations[0].TimeAggregation, query.Aggregations[0].TableHints)
|
|
if err != nil {
|
|
return "", nil, err
|
|
}
|
|
if query.Aggregations[0].TimeAggregation == metrictypes.TimeAggregationRate {
|
|
aggCol = fmt.Sprintf("%s/%d", aggCol, stepSec)
|
|
}
|
|
|
|
sb.SelectMore(fmt.Sprintf("%s AS value", aggCol))
|
|
sb.From(fmt.Sprintf("%s.%s AS points", DBName, tbl))
|
|
sb.Where(
|
|
sb.In("metric_name", query.Aggregations[0].MetricName),
|
|
sb.GTE("unix_milli", start),
|
|
sb.LT("unix_milli", end),
|
|
)
|
|
if query.Filter != nil && query.Filter.Expression != "" {
|
|
filterWhere, err = querybuilder.PrepareWhereClause(query.Filter.Expression, querybuilder.FilterExprVisitorOpts{
|
|
Context: ctx,
|
|
Logger: b.logger,
|
|
FieldMapper: b.fm,
|
|
ConditionBuilder: b.cb,
|
|
FieldKeys: keys,
|
|
FullTextColumn: &telemetrytypes.TelemetryFieldKey{Name: "labels"},
|
|
Variables: variables,
|
|
StartNs: start,
|
|
EndNs: end,
|
|
})
|
|
if err != nil {
|
|
return "", nil, err
|
|
}
|
|
}
|
|
if !filterWhere.IsEmpty() {
|
|
sb.AddWhereClause(filterWhere.WhereClause)
|
|
}
|
|
|
|
if query.Aggregations[0].Temporality != metrictypes.Unknown {
|
|
sb.Where(sb.ILike("temporality", query.Aggregations[0].Temporality.StringValue()))
|
|
}
|
|
sb.GroupBy("ts")
|
|
sb.GroupBy(querybuilder.GroupByKeys(query.GroupBy)...)
|
|
|
|
q, args := sb.BuildWithFlavor(datastoresql.Flavor)
|
|
return fmt.Sprintf("__spatial_aggregation_cte AS (%s)", q), args, nil
|
|
}
|
|
|
|
func (b *meterQueryStatementBuilder) buildTemporalAggregationCTE(
|
|
ctx context.Context,
|
|
start, end uint64,
|
|
query qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation],
|
|
keys map[string][]*telemetrytypes.TelemetryFieldKey,
|
|
variables map[string]qbtypes.VariableItem,
|
|
) (string, []any, error) {
|
|
if query.Aggregations[0].Temporality == metrictypes.Delta {
|
|
return b.buildTemporalAggDelta(ctx, start, end, query, keys, variables)
|
|
}
|
|
return b.buildTemporalAggCumulativeOrUnspecified(ctx, start, end, query, keys, variables)
|
|
}
|
|
|
|
func (b *meterQueryStatementBuilder) buildTemporalAggDelta(
|
|
ctx context.Context,
|
|
start, end uint64,
|
|
query qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation],
|
|
keys map[string][]*telemetrytypes.TelemetryFieldKey,
|
|
variables map[string]qbtypes.VariableItem,
|
|
) (string, []any, error) {
|
|
var filterWhere querybuilder.PreparedWhereClause
|
|
var err error
|
|
|
|
stepSec := int64(query.StepInterval.Seconds())
|
|
sb := sqlbuilder.NewSelectBuilder()
|
|
|
|
sb.Select("fingerprint")
|
|
sb.SelectMore(fmt.Sprintf(
|
|
"toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(%d)) AS ts",
|
|
stepSec,
|
|
))
|
|
|
|
for _, g := range query.GroupBy {
|
|
col, err := b.fm.ColumnExpressionFor(ctx, start, end, &g.TelemetryFieldKey, keys)
|
|
if err != nil {
|
|
return "", nil, err
|
|
}
|
|
sb.SelectMore(col)
|
|
}
|
|
|
|
tbl := WhichSamplesTableToUse(start, end, query.Aggregations[0].Type, query.Aggregations[0].TimeAggregation, query.Aggregations[0].TableHints)
|
|
aggCol, err := AggregationColumnForSamplesTable(start, end, query.Aggregations[0].Type, query.Aggregations[0].Temporality,
|
|
query.Aggregations[0].TimeAggregation, query.Aggregations[0].TableHints)
|
|
if err != nil {
|
|
return "", nil, err
|
|
}
|
|
if query.Aggregations[0].TimeAggregation == metrictypes.TimeAggregationRate {
|
|
aggCol = fmt.Sprintf("%s/%d", aggCol, stepSec)
|
|
}
|
|
|
|
sb.SelectMore(fmt.Sprintf("%s AS per_series_value", aggCol))
|
|
|
|
sb.From(fmt.Sprintf("%s.%s AS points", DBName, tbl))
|
|
sb.Where(
|
|
sb.In("metric_name", query.Aggregations[0].MetricName),
|
|
sb.GTE("unix_milli", start),
|
|
sb.LT("unix_milli", end),
|
|
)
|
|
|
|
if query.Filter != nil && query.Filter.Expression != "" {
|
|
filterWhere, err = querybuilder.PrepareWhereClause(query.Filter.Expression, querybuilder.FilterExprVisitorOpts{
|
|
Context: ctx,
|
|
Logger: b.logger,
|
|
FieldMapper: b.fm,
|
|
ConditionBuilder: b.cb,
|
|
FieldKeys: keys,
|
|
FullTextColumn: &telemetrytypes.TelemetryFieldKey{Name: "labels"},
|
|
Variables: variables,
|
|
StartNs: start,
|
|
EndNs: end,
|
|
})
|
|
if err != nil {
|
|
return "", nil, err
|
|
}
|
|
}
|
|
if !filterWhere.IsEmpty() {
|
|
sb.AddWhereClause(filterWhere.WhereClause)
|
|
}
|
|
|
|
if query.Aggregations[0].Temporality != metrictypes.Unknown {
|
|
sb.Where(sb.ILike("temporality", query.Aggregations[0].Temporality.StringValue()))
|
|
}
|
|
|
|
sb.GroupBy("fingerprint", "ts")
|
|
sb.GroupBy(querybuilder.GroupByKeys(query.GroupBy)...)
|
|
sb.OrderBy("fingerprint", "ts")
|
|
|
|
q, args := sb.BuildWithFlavor(datastoresql.Flavor)
|
|
return fmt.Sprintf("__temporal_aggregation_cte AS (%s)", q), args, nil
|
|
}
|
|
|
|
func (b *meterQueryStatementBuilder) buildTemporalAggCumulativeOrUnspecified(
|
|
ctx context.Context,
|
|
start, end uint64,
|
|
query qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation],
|
|
keys map[string][]*telemetrytypes.TelemetryFieldKey,
|
|
variables map[string]qbtypes.VariableItem,
|
|
) (string, []any, error) {
|
|
var filterWhere querybuilder.PreparedWhereClause
|
|
var err error
|
|
stepSec := int64(query.StepInterval.Seconds())
|
|
|
|
baseSb := sqlbuilder.NewSelectBuilder()
|
|
baseSb.Select("fingerprint")
|
|
baseSb.SelectMore(fmt.Sprintf(
|
|
"toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(%d)) AS ts",
|
|
stepSec,
|
|
))
|
|
for _, g := range query.GroupBy {
|
|
col, err := b.fm.ColumnExpressionFor(ctx, start, end, &g.TelemetryFieldKey, keys)
|
|
if err != nil {
|
|
return "", nil, err
|
|
}
|
|
baseSb.SelectMore(col)
|
|
}
|
|
|
|
tbl := WhichSamplesTableToUse(start, end, query.Aggregations[0].Type, query.Aggregations[0].TimeAggregation, query.Aggregations[0].TableHints)
|
|
aggCol, err := AggregationColumnForSamplesTable(start, end, query.Aggregations[0].Type, query.Aggregations[0].Temporality, query.Aggregations[0].TimeAggregation, query.Aggregations[0].TableHints)
|
|
if err != nil {
|
|
return "", nil, err
|
|
}
|
|
baseSb.SelectMore(fmt.Sprintf("%s AS per_series_value", aggCol))
|
|
|
|
baseSb.From(fmt.Sprintf("%s.%s AS points", DBName, tbl))
|
|
baseSb.Where(
|
|
baseSb.In("metric_name", query.Aggregations[0].MetricName),
|
|
baseSb.GTE("unix_milli", start),
|
|
baseSb.LT("unix_milli", end),
|
|
)
|
|
if query.Filter != nil && query.Filter.Expression != "" {
|
|
filterWhere, err = querybuilder.PrepareWhereClause(query.Filter.Expression, querybuilder.FilterExprVisitorOpts{
|
|
Context: ctx,
|
|
Logger: b.logger,
|
|
FieldMapper: b.fm,
|
|
ConditionBuilder: b.cb,
|
|
FieldKeys: keys,
|
|
FullTextColumn: &telemetrytypes.TelemetryFieldKey{Name: "labels"},
|
|
Variables: variables,
|
|
StartNs: start,
|
|
EndNs: end,
|
|
})
|
|
if err != nil {
|
|
return "", nil, err
|
|
}
|
|
}
|
|
if !filterWhere.IsEmpty() {
|
|
baseSb.AddWhereClause(filterWhere.WhereClause)
|
|
}
|
|
|
|
if query.Aggregations[0].Temporality != metrictypes.Unknown {
|
|
baseSb.Where(baseSb.ILike("temporality", query.Aggregations[0].Temporality.StringValue()))
|
|
}
|
|
baseSb.GroupBy("fingerprint", "ts")
|
|
baseSb.GroupBy(querybuilder.GroupByKeys(query.GroupBy)...)
|
|
baseSb.OrderBy("fingerprint", "ts")
|
|
|
|
innerQuery, innerArgs := baseSb.BuildWithFlavor(datastoresql.Flavor)
|
|
|
|
switch query.Aggregations[0].TimeAggregation {
|
|
case metrictypes.TimeAggregationRate:
|
|
wrapped := sqlbuilder.NewSelectBuilder()
|
|
wrapped.Select("ts")
|
|
for _, g := range query.GroupBy {
|
|
wrapped.SelectMore(querybuilder.QuoteIdent(g.Name))
|
|
}
|
|
wrapped.SelectMore(fmt.Sprintf("%s AS per_series_value", telemetrymetrics.RateTmpl))
|
|
wrapped.From(fmt.Sprintf("(%s) WINDOW rate_window AS (PARTITION BY fingerprint ORDER BY fingerprint, ts)", innerQuery))
|
|
q, args := wrapped.BuildWithFlavor(datastoresql.Flavor, innerArgs...)
|
|
return fmt.Sprintf("__temporal_aggregation_cte AS (%s)", q), args, nil
|
|
|
|
case metrictypes.TimeAggregationIncrease:
|
|
wrapped := sqlbuilder.NewSelectBuilder()
|
|
wrapped.Select("ts")
|
|
for _, g := range query.GroupBy {
|
|
wrapped.SelectMore(querybuilder.QuoteIdent(g.Name))
|
|
}
|
|
wrapped.SelectMore(fmt.Sprintf("%s AS per_series_value", telemetrymetrics.IncreaseTmpl))
|
|
wrapped.From(fmt.Sprintf("(%s) WINDOW rate_window AS (PARTITION BY fingerprint ORDER BY fingerprint, ts)", innerQuery))
|
|
q, args := wrapped.BuildWithFlavor(datastoresql.Flavor, innerArgs...)
|
|
return fmt.Sprintf("__temporal_aggregation_cte AS (%s)", q), args, nil
|
|
default:
|
|
return fmt.Sprintf("__temporal_aggregation_cte AS (%s)", innerQuery), innerArgs, nil
|
|
}
|
|
}
|
|
|
|
func (b *meterQueryStatementBuilder) buildSpatialAggregationCTE(
|
|
_ context.Context,
|
|
_ uint64,
|
|
_ uint64,
|
|
query qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation],
|
|
_ map[string][]*telemetrytypes.TelemetryFieldKey,
|
|
) (string, []any, error) {
|
|
|
|
if query.Aggregations[0].SpaceAggregation.IsZero() {
|
|
return "", nil, errors.Newf(
|
|
errors.TypeInvalidInput,
|
|
errors.CodeInvalidInput,
|
|
"invalid space aggregation, should be one of the following: [`sum`, `avg`, `min`, `max`, `count`]",
|
|
)
|
|
}
|
|
sb := sqlbuilder.NewSelectBuilder()
|
|
|
|
sb.Select("ts")
|
|
for _, g := range query.GroupBy {
|
|
sb.SelectMore(querybuilder.QuoteIdent(g.Name))
|
|
}
|
|
sb.SelectMore(fmt.Sprintf("%s(per_series_value) AS value", query.Aggregations[0].SpaceAggregation.StringValue()))
|
|
sb.From("__temporal_aggregation_cte")
|
|
sb.Where(sb.EQ("isNaN(per_series_value)", 0))
|
|
if query.Aggregations[0].ValueFilter != nil {
|
|
sb.Where(sb.EQ("per_series_value", query.Aggregations[0].ValueFilter.Value))
|
|
}
|
|
sb.GroupBy("ts")
|
|
sb.GroupBy(querybuilder.GroupByKeys(query.GroupBy)...)
|
|
|
|
q, args := sb.BuildWithFlavor(datastoresql.Flavor)
|
|
return fmt.Sprintf("__spatial_aggregation_cte AS (%s)", q), args, nil
|
|
}
|