Files
zeekayandHanzo Dev f337a4b21e one /v1/o11y surface: kill /api and every version segment, unbreak main, migrate consumers to the event plane
MAIN HAD NOT COMPILED SINCE 2026-07-30. Commit 5e3210f759 rewrote tables.go in
three telemetry packages and the sentrytypes Event struct without updating their
consumers, while its message claimed the read paths 'build clean' — proven false
with a detached worktree at clean HEAD. That is why the deployed backend is
SHA-pinned and stale: CI could not build main at all.

THE RENAME: /api/v1|v2|v3|v4/<rest> -> /v1/o11y/<rest> (290 replacements; the
host already says api, so the path never does, and one live generation carries
no version segments). 21 duplicate v1 registrations DELETED with their v2 twins
kept — gorilla/mux is first-match, so after the collapse the v1 line would have
shadowed v2 silently, with no error, forever: rules (6), dashboards (6),
services/top_operations/entry_point_operations/ttl (5), features, query_range
v3+v4 -> v5, metrics/dashboards. 16 orphaned handler funcs (430 lines) deleted.

CONSUMERS MIGRATED to the one-way table names, never the reverse: tables.go's
deletion of the versioned constants was deliberate (stated three times), and the
new blocks are positional 1:1 rewrites of the old, so every resolution is a
constant the same commit declared — SpanIndexV3->span, LogsV2->log,
SamplesV4->metric, TimeseriesV4->series, UpdatedMetadata->descriptor, ambiguity
on operation-vs-trace settled by tables.go's own inline comments.

sentrytypes.Frame/Event.Frames/Event.Handled RECONSTRUCTED, not guessed: pinned
independently by eventsql.go's projection, eventstore.go's scan targets and
module.go's clampLine/InApp, and asserted literally by the pre-existing
TestFramesRoundTrip. TraceSummary.Sample -> Message (event.error has no sample
column). Event.Sample deleted: a field that always read "" while claiming to
hold the occurrence JSON.

PATH MANGLERS KILLED, both deployment shapes: mount.go rewriteExternalPath+
setPath deleted (it mapped /v1/o11y/services -> /api/services => 404 on the
cloud-binary path — proven with a real test before deletion); server.go
StripPrefix removed (it reduced /v1/o11y/services -> /services => the web
catch-all answered index.html on the standalone path). versionless.go +
versionless_test.go + both call sites deleted: its regex matched 0 of the
registered routes and walked the router at every startup to register nothing —
the second spelling of one path, gone with the mechanism that needed it.

FRONTEND: prependBase '/api/v1/' -> '/v1/o11y/' (both baseURL branches), 17
/api/v1/assistant literals renamed (note: no assistant route exists in pkg/ —
client-only surface, a real product gap, not papered over), 21 MSW mocks moved
so they intercept the paths the app now requests.

GATE, all executed: go build ./... green; go test ./... green (exit 0);
0 /api/v[0-9] references in pkg code (frontend's 5 are query-builder escaping
fixtures); 0 duplicate (method,path) registrations; /v1/o11y/services survives
both mount.go and server.go untouched.

Co-authored-by: Hanzo Dev <[email protected]>
2026-07-31 14:07:44 -07:00

2675 lines
94 KiB
Go

package telemetrymetadata
import (
"context"
"fmt"
"log/slog"
"strings"
"time"
"github.com/hanzoai/o11y/pkg/datastoresql"
"github.com/hanzo-ds/sqlbuilder"
"golang.org/x/exp/maps"
"github.com/hanzoai/o11y/pkg/errors"
"github.com/hanzoai/o11y/pkg/factory"
"github.com/hanzoai/o11y/pkg/flagger"
"github.com/hanzoai/o11y/pkg/querybuilder"
"github.com/hanzoai/o11y/pkg/telemetryaudit"
"github.com/hanzoai/o11y/pkg/telemetrylogs"
"github.com/hanzoai/o11y/pkg/telemetrymetrics"
"github.com/hanzoai/o11y/pkg/telemetrystore"
"github.com/hanzoai/o11y/pkg/telemetrytraces"
"github.com/hanzoai/o11y/pkg/types/ctxtypes"
"github.com/hanzoai/o11y/pkg/types/featuretypes"
"github.com/hanzoai/o11y/pkg/types/instrumentationtypes"
"github.com/hanzoai/o11y/pkg/types/metrictypes"
qbtypes "github.com/hanzoai/o11y/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/hanzoai/o11y/pkg/types/telemetrytypes"
"github.com/hanzoai/o11y/pkg/valuer"
)
var (
ErrFailedToGetTracesKeys = errors.Newf(errors.TypeInternal, errors.CodeInternal, "failed to get traces keys")
ErrFailedToGetLogsKeys = errors.Newf(errors.TypeInternal, errors.CodeInternal, "failed to get logs keys")
ErrFailedToGetAuditKeys = errors.Newf(errors.TypeInternal, errors.CodeInternal, "failed to get audit keys")
ErrFailedToGetTblStatement = errors.Newf(errors.TypeInternal, errors.CodeInternal, "failed to get tbl statement")
ErrFailedToGetMetricsKeys = errors.Newf(errors.TypeInternal, errors.CodeInternal, "failed to get metrics keys")
ErrFailedToGetMeterKeys = errors.Newf(errors.TypeInternal, errors.CodeInternal, "failed to get meter keys")
ErrFailedToGetMeterValues = errors.Newf(errors.TypeInternal, errors.CodeInternal, "failed to get meter values")
ErrFailedToGetRelatedValues = errors.Newf(errors.TypeInternal, errors.CodeInternal, "failed to get related values")
)
type telemetryMetaStore struct {
logger *slog.Logger
telemetrystore telemetrystore.TelemetryStore
tracesDBName string
tracesFieldsTblName string
spanAttributesKeysTblName string
indexV3TblName string
metricsDBName string
metricsFieldsTblName string
meterDBName string
meterFieldsTblName string
logsDBName string
logsFieldsTblName string
logAttributeKeysTblName string
logResourceKeysTblName string
logsV2TblName string
auditDBName string
auditLogsTblName string
auditFieldsTblName string
auditAttributeKeysTblName string
auditResourceKeysTblName string
relatedMetadataDBName string
relatedMetadataTblName string
columnEvolutionMetadataTblName string
fm qbtypes.FieldMapper
conditionBuilder qbtypes.ConditionBuilder
fl flagger.Flagger
jsonColumnMetadata map[telemetrytypes.Signal]map[telemetrytypes.FieldContext]telemetrytypes.JSONColumnMetadata
}
func escapeForLike(s string) string {
return strings.ReplaceAll(strings.ReplaceAll(s, `_`, `\_`), `%`, `\%`)
}
func NewTelemetryMetaStore(
settings factory.ProviderSettings,
telemetrystore telemetrystore.TelemetryStore,
tracesDBName string,
tracesFieldsTblName string,
spanAttributesKeysTblName string,
indexV3TblName string,
metricsDBName string,
metricsFieldsTblName string,
meterDBName string,
meterFieldsTblName string,
logsDBName string,
logsV2TblName string,
logsFieldsTblName string,
logAttributeKeysTblName string,
logResourceKeysTblName string,
auditDBName string,
auditLogsTblName string,
auditFieldsTblName string,
auditAttributeKeysTblName string,
auditResourceKeysTblName string,
relatedMetadataDBName string,
relatedMetadataTblName string,
columnEvolutionMetadataTblName string,
fl flagger.Flagger,
) telemetrytypes.MetadataStore {
metadataSettings := factory.NewScopedProviderSettings(settings, "github.com/hanzoai/o11y/pkg/telemetrymetadata")
fm := NewFieldMapper()
conditionBuilder := NewConditionBuilder(fm)
t := &telemetryMetaStore{
logger: metadataSettings.Logger(),
telemetrystore: telemetrystore,
tracesDBName: tracesDBName,
tracesFieldsTblName: tracesFieldsTblName,
spanAttributesKeysTblName: spanAttributesKeysTblName,
indexV3TblName: indexV3TblName,
metricsDBName: metricsDBName,
metricsFieldsTblName: metricsFieldsTblName,
meterDBName: meterDBName,
meterFieldsTblName: meterFieldsTblName,
logsDBName: logsDBName,
logsV2TblName: logsV2TblName,
logsFieldsTblName: logsFieldsTblName,
logAttributeKeysTblName: logAttributeKeysTblName,
logResourceKeysTblName: logResourceKeysTblName,
auditDBName: auditDBName,
auditLogsTblName: auditLogsTblName,
auditFieldsTblName: auditFieldsTblName,
auditAttributeKeysTblName: auditAttributeKeysTblName,
auditResourceKeysTblName: auditResourceKeysTblName,
relatedMetadataDBName: relatedMetadataDBName,
relatedMetadataTblName: relatedMetadataTblName,
columnEvolutionMetadataTblName: columnEvolutionMetadataTblName,
jsonColumnMetadata: map[telemetrytypes.Signal]map[telemetrytypes.FieldContext]telemetrytypes.JSONColumnMetadata{
telemetrytypes.SignalLogs: {
telemetrytypes.FieldContextBody: telemetrytypes.JSONColumnMetadata{
BaseColumn: telemetrylogs.LogsV2BodyV2Column,
PromotedColumn: telemetrylogs.LogsV2BodyPromotedColumn,
},
},
},
fl: fl,
fm: fm,
conditionBuilder: conditionBuilder,
}
return t
}
// tracesTblStatementToFieldKeys returns materialised attribute/resource/scope keys from the traces table.
func (t *telemetryMetaStore) tracesTblStatementToFieldKeys(ctx context.Context) ([]*telemetrytypes.TelemetryFieldKey, error) {
ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
instrumentationtypes.TelemetrySignal: telemetrytypes.SignalTraces.StringValue(),
instrumentationtypes.CodeNamespace: "metadata",
instrumentationtypes.CodeFunctionName: "tracesTblStatementToFieldKeys",
})
query := fmt.Sprintf("SHOW CREATE TABLE %s.%s", t.tracesDBName, t.indexV3TblName)
statements := []telemetrytypes.ShowCreateTableStatement{}
err := t.telemetrystore.Datastore().Select(ctx, &statements, query)
if err != nil {
return nil, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, ErrFailedToGetTblStatement.Error())
}
materialisedKeys, err := ExtractFieldKeysFromTblStatement(statements[0].Statement)
if err != nil {
return nil, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, ErrFailedToGetTracesKeys.Error())
}
for idx := range materialisedKeys {
materialisedKeys[idx].Signal = telemetrytypes.SignalTraces
}
return materialisedKeys, nil
}
// getTracesKeys returns the keys from the spans that match the field selection criteria.
func (t *telemetryMetaStore) getTracesKeys(ctx context.Context, fieldKeySelectors []*telemetrytypes.FieldKeySelector) ([]*telemetrytypes.TelemetryFieldKey, bool, error) {
ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
instrumentationtypes.TelemetrySignal: telemetrytypes.SignalTraces.StringValue(),
instrumentationtypes.CodeNamespace: "metadata",
instrumentationtypes.CodeFunctionName: "getTracesKeys",
})
if len(fieldKeySelectors) == 0 {
return nil, true, nil
}
// pre-fetch the materialised keys from the traces table
matKeys, err := t.tracesTblStatementToFieldKeys(ctx)
if err != nil {
return nil, false, err
}
mapOfKeys := make(map[string]*telemetrytypes.TelemetryFieldKey)
for _, key := range matKeys {
mapOfKeys[key.Name+";"+key.FieldContext.StringValue()+";"+key.FieldDataType.StringValue()] = key
}
sb := sqlbuilder.Select(
"tagKey AS tag_key",
"tagType AS tag_type",
"dataType AS tag_data_type",
`CASE
// WHEN tagType = 'spanfield' THEN 1
WHEN tagType = 'resource' THEN 2
// WHEN tagType = 'scope' THEN 3
WHEN tagType = 'tag' THEN 4
ELSE 5
END as priority`,
).From(t.tracesDBName + "." + t.spanAttributesKeysTblName)
var limit int
searchTexts := []string{}
conds := []string{}
for _, fieldKeySelector := range fieldKeySelectors {
// TODO(srikanthccv): support time filtering for span attribute keys
// if fieldKeySelector.StartUnixMilli != 0 {
// conds = append(conds, sb.GE("unix_milli", fieldKeySelector.StartUnixMilli))
// }
// if fieldKeySelector.EndUnixMilli != 0 {
// conds = append(conds, sb.LE("unix_milli", fieldKeySelector.EndUnixMilli))
// }
// key part of the selector
fieldKeyConds := []string{}
if fieldKeySelector.SelectorMatchType == telemetrytypes.FieldSelectorMatchTypeExact {
fieldKeyConds = append(fieldKeyConds, sb.E("tagKey", fieldKeySelector.Name))
} else {
fieldKeyConds = append(fieldKeyConds, sb.ILike("tagKey", "%"+escapeForLike(fieldKeySelector.Name)+"%"))
}
searchTexts = append(searchTexts, fieldKeySelector.Name)
// now look at the field context
// we don't write most of intrinsic fields to keys table
// for this reason we don't want to apply tagType if the field context
// is not attribute or resource attribute
if fieldKeySelector.FieldContext != telemetrytypes.FieldContextUnspecified &&
(fieldKeySelector.FieldContext == telemetrytypes.FieldContextAttribute ||
fieldKeySelector.FieldContext == telemetrytypes.FieldContextResource) {
fieldKeyConds = append(fieldKeyConds, sb.E("tagType", fieldKeySelector.FieldContext.TagType()))
}
// now look at the field data type
if fieldKeySelector.FieldDataType != telemetrytypes.FieldDataTypeUnspecified {
fieldKeyConds = append(fieldKeyConds, sb.E("dataType", fieldKeySelector.FieldDataType.TagDataType()))
}
conds = append(conds, sb.And(fieldKeyConds...))
limit += fieldKeySelector.Limit
}
// the span_attribute_keys has historically pushed the top level column as attributes
sb.Where(sb.Or(conds...)).Where("isColumn = false")
sb.GroupBy("tagKey", "tagType", "dataType")
if limit == 0 {
limit = 1000
}
mainSb := sqlbuilder.Select("tag_key", "tag_type", "tag_data_type", "max(priority) as priority")
mainSb.From(mainSb.BuilderAs(sb, "sub_query"))
mainSb.GroupBy("tag_key", "tag_type", "tag_data_type")
mainSb.OrderBy("priority")
// query one extra to check if we hit the limit
mainSb.Limit(limit + 1)
query, args := mainSb.BuildWithFlavor(datastoresql.Flavor)
rows, err := t.telemetrystore.Datastore().Query(ctx, query, args...)
if err != nil {
return nil, false, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, ErrFailedToGetTracesKeys.Error())
}
defer rows.Close()
keys := []*telemetrytypes.TelemetryFieldKey{}
rowCount := 0
for rows.Next() {
rowCount++
// reached the limit, we know there are more results
if rowCount > limit {
break
}
var name string
var fieldContext telemetrytypes.FieldContext
var fieldDataType telemetrytypes.FieldDataType
var priority uint8
err = rows.Scan(&name, &fieldContext, &fieldDataType, &priority)
if err != nil {
return nil, false, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, ErrFailedToGetTracesKeys.Error())
}
key, ok := mapOfKeys[name+";"+fieldContext.StringValue()+";"+fieldDataType.StringValue()]
// if there is no materialised column, create a key with the field context and data type
if !ok {
key = &telemetrytypes.TelemetryFieldKey{
Name: name,
Signal: telemetrytypes.SignalTraces,
FieldContext: fieldContext,
FieldDataType: fieldDataType,
}
}
keys = append(keys, key)
mapOfKeys[name+";"+fieldContext.StringValue()+";"+fieldDataType.StringValue()] = key
}
if rows.Err() != nil {
return nil, false, errors.Wrap(rows.Err(), errors.TypeInternal, errors.CodeInternal, ErrFailedToGetTracesKeys.Error())
}
// hit the limit? (only counting DB results)
complete := rowCount <= limit
staticKeys := []string{"isRoot", "isEntryPoint"}
staticKeys = append(staticKeys, maps.Keys(telemetrytraces.IntrinsicFields)...)
staticKeys = append(staticKeys, maps.Keys(telemetrytraces.CalculatedFields)...)
// Add matching intrinsic and matching calculated fields
// These don't count towards the limit
for _, key := range staticKeys {
found := false
for _, v := range searchTexts {
if v == "" || strings.Contains(key, v) {
found = true
break
}
}
if found {
if field, exists := telemetrytraces.IntrinsicFields[key]; exists {
if _, added := mapOfKeys[field.Name+";"+field.FieldContext.StringValue()+";"+field.FieldDataType.StringValue()]; !added {
keys = append(keys, &field)
}
continue
}
if field, exists := telemetrytraces.CalculatedFields[key]; exists {
if _, added := mapOfKeys[field.Name+";"+field.FieldContext.StringValue()+";"+field.FieldDataType.StringValue()]; !added {
keys = append(keys, &field)
}
continue
}
keys = append(keys, &telemetrytypes.TelemetryFieldKey{
Name: key,
FieldContext: telemetrytypes.FieldContextSpan,
Signal: telemetrytypes.SignalTraces,
})
}
}
if err = t.updateColumnEvolutionMetadataForKeys(ctx, keys); err != nil {
return nil, false, err
}
return keys, complete, nil
}
// logsTblStatementToFieldKeys returns materialised attribute/resource/scope keys from the logs table.
func (t *telemetryMetaStore) logsTblStatementToFieldKeys(ctx context.Context) ([]*telemetrytypes.TelemetryFieldKey, error) {
ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
instrumentationtypes.TelemetrySignal: telemetrytypes.SignalLogs.StringValue(),
instrumentationtypes.CodeNamespace: "metadata",
instrumentationtypes.CodeFunctionName: "logsTblStatementToFieldKeys",
})
query := fmt.Sprintf("SHOW CREATE TABLE %s.%s", t.logsDBName, t.logsV2TblName)
statements := []telemetrytypes.ShowCreateTableStatement{}
err := t.telemetrystore.Datastore().Select(ctx, &statements, query)
if err != nil {
return nil, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, ErrFailedToGetTblStatement.Error())
}
materialisedKeys, err := ExtractFieldKeysFromTblStatement(statements[0].Statement)
if err != nil {
return nil, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, ErrFailedToGetLogsKeys.Error())
}
for idx := range materialisedKeys {
materialisedKeys[idx].Signal = telemetrytypes.SignalLogs
}
return materialisedKeys, nil
}
// getLogsKeys returns the keys from the spans that match the field selection criteria.
func (t *telemetryMetaStore) getLogsKeys(ctx context.Context, fieldKeySelectors []*telemetrytypes.FieldKeySelector) ([]*telemetrytypes.TelemetryFieldKey, bool, error) {
ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
instrumentationtypes.TelemetrySignal: telemetrytypes.SignalLogs.StringValue(),
instrumentationtypes.CodeNamespace: "metadata",
instrumentationtypes.CodeFunctionName: "getLogsKeys",
})
if len(fieldKeySelectors) == 0 {
return nil, true, nil
}
// pre-fetch the materialised keys from the logs table
matKeys, err := t.logsTblStatementToFieldKeys(ctx)
if err != nil {
return nil, false, err
}
mapOfKeys := make(map[string]*telemetrytypes.TelemetryFieldKey)
for _, key := range matKeys {
mapOfKeys[key.Name+";"+key.FieldContext.StringValue()+";"+key.FieldDataType.StringValue()] = key
}
// queries for both attribute and resource keys tables
var queries []string
var allArgs []any
// tables to query based on field selectors
queryAttributeTable := false
queryResourceTable := false
queryBodyTable := false
for _, selector := range fieldKeySelectors {
if selector.FieldContext == telemetrytypes.FieldContextUnspecified {
// unspecified context, query all tables
queryAttributeTable = true
queryResourceTable = true
queryBodyTable = true
break
} else if selector.FieldContext == telemetrytypes.FieldContextAttribute {
queryAttributeTable = true
} else if selector.FieldContext == telemetrytypes.FieldContextResource {
queryResourceTable = true
} else if selector.FieldContext == telemetrytypes.FieldContextBody {
queryBodyTable = true
}
}
// body keys are gated behind the feature flag
// TODO(Tushar): thread orgID here to evaluate correctly
queryBodyTable = queryBodyTable && t.fl.BooleanOrEmpty(ctx, flagger.FeatureUseJSONBody, featuretypes.NewFlaggerEvaluationContext(valuer.UUID{}))
// requestedFieldKeySelectors is the set of names the user explicitly asked for.
// Used to ensure a name that is both a parent path AND a directly requested field still surfaces
// in the result keys (e.g. "a.b[].c" is a parent of "a.b[].c[].d" but also a queried field).
mapOfRequestedSelectors := make(map[string]bool)
for _, sel := range fieldKeySelectors {
if sel.SelectorMatchType == telemetrytypes.FieldSelectorMatchTypeExact {
mapOfRequestedSelectors[sel.Name] = true
}
}
// pre-compute parent array path names from body selectors for JSON plan building;
// these will be fetched as a separate UNION arm filtered to ArrayJSON/ArrayDynamic only.
parentPaths := make(map[string]bool)
if queryBodyTable {
for _, sel := range fieldKeySelectors {
if sel.FieldContext != telemetrytypes.FieldContextBody &&
sel.FieldContext != telemetrytypes.FieldContextUnspecified {
continue
}
key := &telemetrytypes.TelemetryFieldKey{
Name: sel.Name,
Signal: telemetrytypes.SignalLogs,
FieldContext: telemetrytypes.FieldContextBody,
}
if !key.KeyNameContainsArray() {
continue
}
for _, parent := range key.ArrayParentPaths() {
parentPaths[parent] = true
}
}
}
// attribute and resource tables share identical schema (name/datatype columns, lower-cased select,
// TagDataType encoding) — only the table name and field context differ.
addTagTableQuery := func(tblName string, fieldContext telemetrytypes.FieldContext) {
sb := sqlbuilder.Select(
"name AS tag_key",
fmt.Sprintf("'%s' AS tag_type", fieldContext.TagType()),
"lower(datatype) AS tag_data_type", // logs had historical mixed-case data
fmt.Sprintf("%d AS priority", getPriorityForContext(fieldContext)),
).From(tblName)
conds := []string{}
for _, sel := range fieldKeySelectors {
// Include this selector if:
// 1. It has unspecified context (matches all tables)
// 2. Its context matches the current table's context
if sel.FieldContext != telemetrytypes.FieldContextUnspecified && sel.FieldContext != fieldContext {
continue
}
fieldKeyConds := []string{}
if sel.SelectorMatchType == telemetrytypes.FieldSelectorMatchTypeExact {
fieldKeyConds = append(fieldKeyConds, sb.E("name", sel.Name))
} else {
fieldKeyConds = append(fieldKeyConds, sb.ILike("name", "%"+escapeForLike(sel.Name)+"%"))
}
if sel.FieldDataType != telemetrytypes.FieldDataTypeUnspecified {
fieldKeyConds = append(fieldKeyConds, sb.E("datatype", sel.FieldDataType.TagDataType()))
}
if len(fieldKeyConds) > 0 {
conds = append(conds, sb.And(fieldKeyConds...))
}
}
if len(conds) > 0 {
sb.Where(sb.Or(conds...))
}
sb.GroupBy("name", "datatype")
query, args := sb.BuildWithFlavor(datastoresql.Flavor)
queries = append(queries, query)
allArgs = append(allArgs, args...)
}
if queryAttributeTable {
addTagTableQuery(t.logsDBName+"."+t.logAttributeKeysTblName, telemetrytypes.FieldContextAttribute)
}
if queryResourceTable {
addTagTableQuery(t.logsDBName+"."+t.logResourceKeysTblName, telemetrytypes.FieldContextResource)
}
// body context uses a different table/schema (field_name, field_data_type) and requires
// signal+context base filters. It also fetches parent array container types (ArrayJSON/ArrayDynamic)
// needed for JSON access plan building.
if queryBodyTable {
sb := sqlbuilder.Select(
"field_name AS tag_key",
fmt.Sprintf("'%s' AS tag_type", telemetrytypes.FieldContextBody.TagType()),
"field_data_type AS tag_data_type",
fmt.Sprintf("%d AS priority", getPriorityForContext(telemetrytypes.FieldContextBody)),
).From(fmt.Sprintf("%s.%s", DBName, FieldKeysTable))
sb.Where(sb.E("signal", telemetrytypes.SignalLogs.StringValue()))
sb.Where(sb.E("field_context", telemetrytypes.FieldContextBody.StringValue()))
branches := []string{}
for _, sel := range fieldKeySelectors {
if sel.FieldContext != telemetrytypes.FieldContextUnspecified && sel.FieldContext != telemetrytypes.FieldContextBody {
continue
}
fieldKeyConds := []string{}
if sel.SelectorMatchType == telemetrytypes.FieldSelectorMatchTypeExact {
fieldKeyConds = append(fieldKeyConds, sb.E("field_name", sel.Name))
} else {
fieldKeyConds = append(fieldKeyConds, sb.ILike("field_name", "%"+escapeForLike(sel.Name)+"%"))
}
if sel.FieldDataType != telemetrytypes.FieldDataTypeUnspecified {
fieldKeyConds = append(fieldKeyConds, sb.E("field_data_type", sel.FieldDataType.StringValue()))
}
if len(fieldKeyConds) > 0 {
branches = append(branches, sb.And(fieldKeyConds...))
}
}
if len(parentPaths) > 0 {
names := make([]any, 0, len(parentPaths))
for n := range parentPaths {
names = append(names, n)
}
branches = append(branches, sb.And(
sb.In("field_name", names...),
sb.In("field_data_type",
telemetrytypes.FieldDataTypeArrayDynamic.StringValue(),
telemetrytypes.FieldDataTypeArrayJSON.StringValue(),
),
))
}
if len(branches) > 0 {
sb.Where(sb.Or(branches...))
}
sb.GroupBy("field_name", "field_data_type")
query, args := sb.BuildWithFlavor(datastoresql.Flavor)
queries = append(queries, query)
allArgs = append(allArgs, args...)
}
if len(queries) == 0 {
// No matching contexts, return empty result
return []*telemetrytypes.TelemetryFieldKey{}, true, nil
}
// Combine queries with UNION ALL
var limit int
for _, fieldKeySelector := range fieldKeySelectors {
limit += fieldKeySelector.Limit
}
if limit == 0 {
limit = 1000
}
mainQuery := fmt.Sprintf(`
SELECT tag_key, tag_type, tag_data_type, max(priority) as priority
FROM (
%s
) AS combined_results
GROUP BY tag_key, tag_type, tag_data_type
ORDER BY priority
LIMIT %d
`, strings.Join(queries, " UNION ALL "), limit+1)
rows, err := t.telemetrystore.Datastore().Query(ctx, mainQuery, allArgs...)
if err != nil {
return nil, false, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, ErrFailedToGetLogsKeys.Error())
}
defer rows.Close()
keys := []*telemetrytypes.TelemetryFieldKey{}
parentTypes := make(map[string][]telemetrytypes.FieldDataType)
rowCount := 0
searchTexts := []string{}
// Collect search texts for static field matching
for _, fieldKeySelector := range fieldKeySelectors {
searchTexts = append(searchTexts, fieldKeySelector.Name)
}
for rows.Next() {
rowCount++
// reached the limit, we know there are more results
if rowCount > limit {
break
}
var name string
var fieldContext telemetrytypes.FieldContext
var fieldDataType telemetrytypes.FieldDataType
var priority uint8
err = rows.Scan(&name, &fieldContext, &fieldDataType, &priority)
if err != nil {
return nil, false, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, ErrFailedToGetLogsKeys.Error())
}
// ArrayJSON/ArrayDynamic body rows for parent paths are needed by the JSON access plan
// builder (enrichJSONKeys). Always record them in parentTypes. Only skip adding to keys
// if the user did not also directly request this name — a field like "education" can be
// both a parent of "education[].name" and an explicitly queried field in its own right.
switch fieldDataType {
case telemetrytypes.FieldDataTypeArrayJSON, telemetrytypes.FieldDataTypeArrayDynamic:
if fieldContext == telemetrytypes.FieldContextBody && parentPaths[name] {
parentTypes[name] = append(parentTypes[name], fieldDataType)
if !mapOfRequestedSelectors[name] {
continue // skip; don't register the key.
}
}
}
key, ok := mapOfKeys[name+";"+fieldContext.StringValue()+";"+fieldDataType.StringValue()]
// if there is no materialised column, create a key with the field context and data type
if !ok {
key = &telemetrytypes.TelemetryFieldKey{
Name: name,
Signal: telemetrytypes.SignalLogs,
FieldContext: fieldContext,
FieldDataType: fieldDataType,
}
}
keys = append(keys, key)
mapOfKeys[name+";"+fieldContext.StringValue()+";"+fieldDataType.StringValue()] = key
}
if rows.Err() != nil {
return nil, false, errors.Wrap(rows.Err(), errors.TypeInternal, errors.CodeInternal, ErrFailedToGetLogsKeys.Error())
}
// hit the limit? (only counting DB results)
complete := rowCount <= limit
staticKeys := []string{}
staticKeys = append(staticKeys, maps.Keys(telemetrylogs.IntrinsicFields)...)
// Add matching intrinsic and matching calculated fields
// These don't count towards the limit
for _, key := range staticKeys {
found := false
for _, v := range searchTexts {
if v == "" || strings.Contains(key, v) {
found = true
break
}
}
if found {
if field, exists := telemetrylogs.IntrinsicFields[key]; exists {
if _, added := mapOfKeys[field.Name+";"+field.FieldContext.StringValue()+";"+field.FieldDataType.StringValue()]; !added {
keys = append(keys, &field)
}
continue
}
keys = append(keys, &telemetrytypes.TelemetryFieldKey{
Name: key,
FieldContext: telemetrytypes.FieldContextLog,
Signal: telemetrytypes.SignalLogs,
})
}
}
// enrich body keys with promoted paths, indexes, and JSON access plans
// TODO(Tushar): thread orgID here to evaluate correctly
if t.fl.BooleanOrEmpty(ctx, flagger.FeatureUseJSONBody, featuretypes.NewFlaggerEvaluationContext(valuer.UUID{})) {
if err := t.enrichJSONKeys(ctx, fieldKeySelectors, keys, parentTypes); err != nil {
return nil, false, err
}
}
if err := t.updateColumnEvolutionMetadataForKeys(ctx, keys); err != nil {
return nil, false, err
}
return keys, complete, nil
}
func (t *telemetryMetaStore) auditTblStatementToFieldKeys(ctx context.Context) ([]*telemetrytypes.TelemetryFieldKey, error) {
ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
instrumentationtypes.TelemetrySignal: telemetrytypes.SignalLogs.StringValue(),
instrumentationtypes.CodeNamespace: "metadata",
instrumentationtypes.CodeFunctionName: "auditTblStatementToFieldKeys",
})
query := fmt.Sprintf("SHOW CREATE TABLE %s.%s", t.auditDBName, t.auditLogsTblName)
statements := []telemetrytypes.ShowCreateTableStatement{}
err := t.telemetrystore.Datastore().Select(ctx, &statements, query)
if err != nil {
return nil, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, ErrFailedToGetTblStatement.Error())
}
materialisedKeys, err := ExtractFieldKeysFromTblStatement(statements[0].Statement)
if err != nil {
return nil, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, ErrFailedToGetAuditKeys.Error())
}
for idx := range materialisedKeys {
materialisedKeys[idx].Signal = telemetrytypes.SignalLogs
}
return materialisedKeys, nil
}
func (t *telemetryMetaStore) getAuditKeys(ctx context.Context, fieldKeySelectors []*telemetrytypes.FieldKeySelector) ([]*telemetrytypes.TelemetryFieldKey, bool, error) {
ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
instrumentationtypes.TelemetrySignal: telemetrytypes.SignalLogs.StringValue(),
instrumentationtypes.CodeNamespace: "metadata",
instrumentationtypes.CodeFunctionName: "getAuditKeys",
})
if len(fieldKeySelectors) == 0 {
return nil, true, nil
}
matKeys, err := t.auditTblStatementToFieldKeys(ctx)
if err != nil {
return nil, false, err
}
mapOfKeys := make(map[string]*telemetrytypes.TelemetryFieldKey)
for _, key := range matKeys {
mapOfKeys[key.Name+";"+key.FieldContext.StringValue()+";"+key.FieldDataType.StringValue()] = key
}
var queries []string
var allArgs []any
queryAttributeTable := false
queryResourceTable := false
for _, selector := range fieldKeySelectors {
if selector.FieldContext == telemetrytypes.FieldContextUnspecified {
queryAttributeTable = true
queryResourceTable = true
break
} else if selector.FieldContext == telemetrytypes.FieldContextAttribute {
queryAttributeTable = true
} else if selector.FieldContext == telemetrytypes.FieldContextResource {
queryResourceTable = true
}
}
tablesToQuery := []struct {
fieldContext telemetrytypes.FieldContext
shouldQuery bool
tblName string
}{
{telemetrytypes.FieldContextAttribute, queryAttributeTable, t.auditDBName + "." + t.auditAttributeKeysTblName},
{telemetrytypes.FieldContextResource, queryResourceTable, t.auditDBName + "." + t.auditResourceKeysTblName},
}
for _, table := range tablesToQuery {
if !table.shouldQuery {
continue
}
fieldContext := table.fieldContext
tblName := table.tblName
sb := sqlbuilder.Select(
"name AS tag_key",
fmt.Sprintf("'%s' AS tag_type", fieldContext.TagType()),
"lower(datatype) AS tag_data_type",
fmt.Sprintf("%d AS priority", getPriorityForContext(fieldContext)),
).From(tblName)
var limit int
conds := []string{}
for _, fieldKeySelector := range fieldKeySelectors {
if fieldKeySelector.FieldContext != telemetrytypes.FieldContextUnspecified && fieldKeySelector.FieldContext != fieldContext {
continue
}
fieldKeyConds := []string{}
if fieldKeySelector.SelectorMatchType == telemetrytypes.FieldSelectorMatchTypeExact {
fieldKeyConds = append(fieldKeyConds, sb.E("name", fieldKeySelector.Name))
} else {
fieldKeyConds = append(fieldKeyConds, sb.ILike("name", "%"+escapeForLike(fieldKeySelector.Name)+"%"))
}
if fieldKeySelector.FieldDataType != telemetrytypes.FieldDataTypeUnspecified {
fieldKeyConds = append(fieldKeyConds, sb.E("datatype", fieldKeySelector.FieldDataType.TagDataType()))
}
if len(fieldKeyConds) > 0 {
conds = append(conds, sb.And(fieldKeyConds...))
}
limit += fieldKeySelector.Limit
}
if len(conds) > 0 {
sb.Where(sb.Or(conds...))
}
sb.GroupBy("name", "datatype")
if limit == 0 {
limit = 1000
}
query, args := sb.BuildWithFlavor(datastoresql.Flavor)
queries = append(queries, query)
allArgs = append(allArgs, args...)
}
if len(queries) == 0 {
return []*telemetrytypes.TelemetryFieldKey{}, true, nil
}
var limit int
for _, fieldKeySelector := range fieldKeySelectors {
limit += fieldKeySelector.Limit
}
if limit == 0 {
limit = 1000
}
mainQuery := fmt.Sprintf(`
SELECT tag_key, tag_type, tag_data_type, max(priority) as priority
FROM (
%s
) AS combined_results
GROUP BY tag_key, tag_type, tag_data_type
ORDER BY priority
LIMIT %d
`, strings.Join(queries, " UNION ALL "), limit+1)
rows, err := t.telemetrystore.Datastore().Query(ctx, mainQuery, allArgs...)
if err != nil {
return nil, false, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, ErrFailedToGetAuditKeys.Error())
}
defer rows.Close()
keys := []*telemetrytypes.TelemetryFieldKey{}
rowCount := 0
searchTexts := []string{}
for _, fieldKeySelector := range fieldKeySelectors {
searchTexts = append(searchTexts, fieldKeySelector.Name)
}
for rows.Next() {
rowCount++
if rowCount > limit {
break
}
var name string
var fieldContext telemetrytypes.FieldContext
var fieldDataType telemetrytypes.FieldDataType
var priority uint8
err = rows.Scan(&name, &fieldContext, &fieldDataType, &priority)
if err != nil {
return nil, false, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, ErrFailedToGetAuditKeys.Error())
}
key, ok := mapOfKeys[name+";"+fieldContext.StringValue()+";"+fieldDataType.StringValue()]
if !ok {
key = &telemetrytypes.TelemetryFieldKey{
Name: name,
Signal: telemetrytypes.SignalLogs,
FieldContext: fieldContext,
FieldDataType: fieldDataType,
}
}
keys = append(keys, key)
mapOfKeys[name+";"+fieldContext.StringValue()+";"+fieldDataType.StringValue()] = key
}
if rows.Err() != nil {
return nil, false, errors.Wrap(rows.Err(), errors.TypeInternal, errors.CodeInternal, ErrFailedToGetAuditKeys.Error())
}
complete := rowCount <= limit
// Add intrinsic audit fields (same as logs intrinsics: body, severity_text, etc.)
staticKeys := maps.Keys(telemetryaudit.IntrinsicFields)
for _, key := range staticKeys {
found := false
for _, v := range searchTexts {
if v == "" || strings.Contains(key, v) {
found = true
break
}
}
if found {
if field, exists := telemetryaudit.IntrinsicFields[key]; exists {
if _, added := mapOfKeys[field.Name+";"+field.FieldContext.StringValue()+";"+field.FieldDataType.StringValue()]; !added {
keys = append(keys, &field)
}
}
}
}
return keys, complete, nil
}
func getPriorityForContext(ctx telemetrytypes.FieldContext) int {
switch ctx {
case telemetrytypes.FieldContextLog:
return 1
case telemetrytypes.FieldContextResource:
return 2
case telemetrytypes.FieldContextScope:
return 3
case telemetrytypes.FieldContextAttribute:
return 4
default:
return 5
}
}
// getMetricsKeys returns the keys from the metrics that match the field selection criteria.
func (t *telemetryMetaStore) getMetricsKeys(ctx context.Context, fieldKeySelectors []*telemetrytypes.FieldKeySelector) ([]*telemetrytypes.TelemetryFieldKey, bool, error) {
ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
instrumentationtypes.TelemetrySignal: telemetrytypes.SignalMetrics.StringValue(),
instrumentationtypes.CodeNamespace: "metadata",
instrumentationtypes.CodeFunctionName: "getMetricsKeys",
})
if len(fieldKeySelectors) == 0 {
return nil, true, nil
}
sb := sqlbuilder.
Select("attr_name as name", "attr_type as field_context", "attr_datatype as field_data_type", `
CASE
WHEN attr_type = 'resource' THEN 1
WHEN attr_type = 'scope' THEN 2
WHEN attr_type = 'point' THEN 3
ELSE 4
END as priority`).
From(t.metricsDBName + "." + t.metricsFieldsTblName)
var limit int
conds := []string{}
for _, fieldKeySelector := range fieldKeySelectors {
fieldConds := []string{}
if fieldKeySelector.SelectorMatchType == telemetrytypes.FieldSelectorMatchTypeExact {
fieldConds = append(fieldConds, sb.E("attr_name", fieldKeySelector.Name))
} else {
fieldConds = append(fieldConds, sb.ILike("attr_name", "%"+escapeForLike(fieldKeySelector.Name)+"%"))
}
fieldConds = append(fieldConds, sb.NotLike("attr_name", "\\_\\_%"))
// note: type and datatype do not have much significance in metrics
// if fieldKeySelector.FieldContext != telemetrytypes.FieldContextUnspecified {
// fieldConds = append(fieldConds, sb.E("attr_type", fieldKeySelector.FieldContext.TagType()))
// }
// if fieldKeySelector.FieldDataType != telemetrytypes.FieldDataTypeUnspecified {
// fieldConds = append(fieldConds, sb.E("attr_datatype", fieldKeySelector.FieldDataType.TagDataType()))
// }
if fieldKeySelector.MetricContext != nil {
if fieldKeySelector.MetricContext.MetricName != "" {
fieldConds = append(fieldConds, sb.E("metric_name", fieldKeySelector.MetricContext.MetricName))
}
if fieldKeySelector.MetricContext.MetricNamespace != "" {
fieldConds = append(fieldConds, sb.Like("metric_name", escapeForLike(fieldKeySelector.MetricContext.MetricNamespace)+"%"))
}
}
conds = append(conds, sb.And(fieldConds...))
limit += fieldKeySelector.Limit
}
sb.Where(sb.Or(conds...))
sb.GroupBy("name", "field_context", "field_data_type")
if limit == 0 {
limit = 1000
}
mainSb := sqlbuilder.Select("name", "field_context", "field_data_type", "max(priority) as priority")
mainSb.From(mainSb.BuilderAs(sb, "sub_query"))
mainSb.GroupBy("name", "field_context", "field_data_type")
mainSb.OrderBy("priority")
// query one extra to check if we hit the limit
mainSb.Limit(limit + 1)
query, args := mainSb.BuildWithFlavor(datastoresql.Flavor)
rows, err := t.telemetrystore.Datastore().Query(ctx, query, args...)
if err != nil {
return nil, false, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, ErrFailedToGetMetricsKeys.Error())
}
defer rows.Close()
keys := []*telemetrytypes.TelemetryFieldKey{}
rowCount := 0
for rows.Next() {
rowCount++
// reached the limit, we know there are more results
if rowCount > limit {
break
}
var name string
var fieldContext telemetrytypes.FieldContext
var fieldDataType telemetrytypes.FieldDataType
var priority uint8
err = rows.Scan(&name, &fieldContext, &fieldDataType, &priority)
if err != nil {
return nil, false, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, ErrFailedToGetMetricsKeys.Error())
}
keys = append(keys, &telemetrytypes.TelemetryFieldKey{
Name: name,
Signal: telemetrytypes.SignalMetrics,
FieldContext: fieldContext,
FieldDataType: fieldDataType,
})
}
if rows.Err() != nil {
return nil, false, errors.Wrap(rows.Err(), errors.TypeInternal, errors.CodeInternal, ErrFailedToGetMetricsKeys.Error())
}
// hit the limit?
complete := rowCount <= limit
return keys, complete, nil
}
// getMeterKeys returns the keys from the meter metrics that match the field selection criteria.
func (t *telemetryMetaStore) getMeterSourceMetricKeys(ctx context.Context, fieldKeySelectors []*telemetrytypes.FieldKeySelector) ([]*telemetrytypes.TelemetryFieldKey, bool, error) {
ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
instrumentationtypes.TelemetrySignal: telemetrytypes.SignalMetrics.StringValue(),
instrumentationtypes.CodeNamespace: "metadata",
instrumentationtypes.CodeFunctionName: "getMeterSourceMetricKeys",
})
if len(fieldKeySelectors) == 0 {
return nil, true, nil
}
sb := sqlbuilder.Select("DISTINCT arrayJoin(JSONExtractKeys(labels)) as attr_name").From(t.meterDBName + "." + t.meterFieldsTblName)
conds := []string{}
var limit int
for _, fieldKeySelector := range fieldKeySelectors {
fieldConds := []string{}
if fieldKeySelector.SelectorMatchType == telemetrytypes.FieldSelectorMatchTypeExact {
fieldConds = append(fieldConds, sb.E("attr_name", fieldKeySelector.Name))
} else {
fieldConds = append(fieldConds, sb.Like("attr_name", "%"+fieldKeySelector.Name+"%"))
}
fieldConds = append(fieldConds, sb.NotLike("attr_name", "\\_\\_%"))
if fieldKeySelector.MetricContext != nil {
if fieldKeySelector.MetricContext.MetricName != "" {
fieldConds = append(fieldConds, sb.E("metric_name", fieldKeySelector.MetricContext.MetricName))
}
if fieldKeySelector.MetricContext.MetricNamespace != "" {
fieldConds = append(fieldConds, sb.Like("metric_name", escapeForLike(fieldKeySelector.MetricContext.MetricNamespace)+"%"))
}
}
conds = append(conds, sb.And(fieldConds...))
limit += fieldKeySelector.Limit
}
sb.Where(sb.Or(conds...))
if limit == 0 {
limit = 1000
}
sb.Limit(limit)
query, args := sb.BuildWithFlavor(datastoresql.Flavor)
rows, err := t.telemetrystore.Datastore().Query(ctx, query, args...)
if err != nil {
return nil, false, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, ErrFailedToGetMeterKeys.Error())
}
defer rows.Close()
keys := []*telemetrytypes.TelemetryFieldKey{}
rowCount := 0
for rows.Next() {
rowCount++
// reached the limit, we know there are more results
if rowCount > limit {
break
}
var name string
err = rows.Scan(&name)
if err != nil {
return nil, false, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, ErrFailedToGetMeterKeys.Error())
}
keys = append(keys, &telemetrytypes.TelemetryFieldKey{
Name: name,
Signal: telemetrytypes.SignalMetrics,
})
}
if rows.Err() != nil {
return nil, false, errors.Wrap(rows.Err(), errors.TypeInternal, errors.CodeInternal, ErrFailedToGetMeterKeys.Error())
}
// hit the limit?
complete := rowCount <= limit
return keys, complete, nil
}
// applyBackwardCompatibleKeys adds backward compatible key aliases to the map.
func applyBackwardCompatibleKeys(mapOfKeys map[string][]*telemetrytypes.TelemetryFieldKey) {
// Get backward compatible keys for all signals
backwardCompatKeysBySignal := map[telemetrytypes.Signal]BackwardCompatibleKeyMap{
telemetrytypes.SignalTraces: GetBackwardCompatKeysForSignal(telemetrytypes.SignalTraces),
telemetrytypes.SignalLogs: GetBackwardCompatKeysForSignal(telemetrytypes.SignalLogs),
telemetrytypes.SignalMetrics: GetBackwardCompatKeysForSignal(telemetrytypes.SignalMetrics),
}
// Iterate over existing keys and add aliases if they exist in backward compat mapping
for srcKey, srcKeys := range mapOfKeys {
for _, srcKeyEntry := range srcKeys {
backwardCompatKeys := backwardCompatKeysBySignal[srcKeyEntry.Signal]
if backwardCompatKeys == nil {
continue
}
if aliasKey, ok := backwardCompatKeys[srcKey]; ok {
if _, aliasExists := mapOfKeys[aliasKey]; !aliasExists {
aliasKeyEntry := &telemetrytypes.TelemetryFieldKey{
Name: aliasKey,
Signal: srcKeyEntry.Signal,
FieldContext: srcKeyEntry.FieldContext,
FieldDataType: srcKeyEntry.FieldDataType,
}
mapOfKeys[aliasKey] = []*telemetrytypes.TelemetryFieldKey{aliasKeyEntry}
}
// Found the alias for this signal, no need to check other entries
break
}
}
}
}
func enrichWithIntrinsicMetricKeys(keys map[string][]*telemetrytypes.TelemetryFieldKey, selectors []*telemetrytypes.FieldKeySelector) map[string][]*telemetrytypes.TelemetryFieldKey {
if len(selectors) == 0 {
return keys
}
for _, selector := range selectors {
if selector.Signal != telemetrytypes.SignalMetrics && selector.Signal != telemetrytypes.SignalUnspecified {
continue
}
// If metric filters are provided, do not surface intrinsic metric keys.
if selector.MetricContext != nil && (selector.MetricContext.MetricName != "" || selector.MetricContext.MetricNamespace != "") {
continue
}
for name, key := range telemetrymetrics.IntrinsicMetricFieldDefinitions {
if !selectorMatchesIntrinsicField(selector, key) {
continue
}
keyCopy := key
keys[name] = append(keys[name], &keyCopy)
}
}
return keys
}
func selectorMatchesIntrinsicField(selector *telemetrytypes.FieldKeySelector, definition telemetrytypes.TelemetryFieldKey) bool {
if selector.FieldContext != telemetrytypes.FieldContextUnspecified && selector.FieldContext != definition.FieldContext {
return false
}
if selector.FieldDataType != telemetrytypes.FieldDataTypeUnspecified && selector.FieldDataType != definition.FieldDataType {
return false
}
return matchesSelectorName(selector.Name, definition.Name, selector.SelectorMatchType)
}
func matchesSelectorName(selectorName, target string, matchType telemetrytypes.FieldSelectorMatchType) bool {
if selectorName == "" {
return true
}
switch matchType {
case telemetrytypes.FieldSelectorMatchTypeExact:
return strings.EqualFold(selectorName, target)
default:
return strings.Contains(strings.ToLower(target), strings.ToLower(selectorName))
}
}
func (t *telemetryMetaStore) GetKeys(ctx context.Context, fieldKeySelector *telemetrytypes.FieldKeySelector) (map[string][]*telemetrytypes.TelemetryFieldKey, bool, error) {
var keys []*telemetrytypes.TelemetryFieldKey
var complete = true
var err error
selectors := []*telemetrytypes.FieldKeySelector{}
if fieldKeySelector != nil {
selectors = []*telemetrytypes.FieldKeySelector{fieldKeySelector}
}
switch fieldKeySelector.Signal {
case telemetrytypes.SignalTraces:
keys, complete, err = t.getTracesKeys(ctx, selectors)
case telemetrytypes.SignalLogs:
if fieldKeySelector.Source == telemetrytypes.SourceAudit {
keys, complete, err = t.getAuditKeys(ctx, selectors)
} else {
keys, complete, err = t.getLogsKeys(ctx, selectors)
}
case telemetrytypes.SignalMetrics:
if fieldKeySelector.Source == telemetrytypes.SourceMeter {
keys, complete, err = t.getMeterSourceMetricKeys(ctx, selectors)
} else {
keys, complete, err = t.getMetricsKeys(ctx, selectors)
}
case telemetrytypes.SignalUnspecified:
// get traces keys
tracesKeys, tracesComplete, err := t.getTracesKeys(ctx, selectors)
if err != nil {
return nil, false, err
}
keys = append(keys, tracesKeys...)
// get logs keys
logsKeys, logsComplete, err := t.getLogsKeys(ctx, selectors)
if err != nil {
return nil, false, err
}
keys = append(keys, logsKeys...)
// get metrics keys
metricsKeys, metricsComplete, err := t.getMetricsKeys(ctx, selectors)
if err != nil {
return nil, false, err
}
keys = append(keys, metricsKeys...)
complete = tracesComplete && logsComplete && metricsComplete
}
if err != nil {
return nil, false, err
}
mapOfKeys := make(map[string][]*telemetrytypes.TelemetryFieldKey)
for _, key := range keys {
mapOfKeys[key.Name] = append(mapOfKeys[key.Name], key)
}
applyBackwardCompatibleKeys(mapOfKeys)
mapOfKeys = enrichWithIntrinsicMetricKeys(mapOfKeys, selectors)
return mapOfKeys, complete, nil
}
func (t *telemetryMetaStore) GetKeysMulti(ctx context.Context, fieldKeySelectors []*telemetrytypes.FieldKeySelector) (map[string][]*telemetrytypes.TelemetryFieldKey, bool, error) {
logsSelectors := []*telemetrytypes.FieldKeySelector{}
auditSelectors := []*telemetrytypes.FieldKeySelector{}
tracesSelectors := []*telemetrytypes.FieldKeySelector{}
metricsSelectors := []*telemetrytypes.FieldKeySelector{}
meterSourceMetricsSelectors := []*telemetrytypes.FieldKeySelector{}
for _, fieldKeySelector := range fieldKeySelectors {
switch fieldKeySelector.Signal {
case telemetrytypes.SignalLogs:
if fieldKeySelector.Source == telemetrytypes.SourceAudit {
auditSelectors = append(auditSelectors, fieldKeySelector)
} else {
logsSelectors = append(logsSelectors, fieldKeySelector)
}
case telemetrytypes.SignalTraces:
tracesSelectors = append(tracesSelectors, fieldKeySelector)
case telemetrytypes.SignalMetrics:
if fieldKeySelector.Source == telemetrytypes.SourceMeter {
meterSourceMetricsSelectors = append(meterSourceMetricsSelectors, fieldKeySelector)
} else {
metricsSelectors = append(metricsSelectors, fieldKeySelector)
}
case telemetrytypes.SignalUnspecified:
logsSelectors = append(logsSelectors, fieldKeySelector)
tracesSelectors = append(tracesSelectors, fieldKeySelector)
metricsSelectors = append(metricsSelectors, fieldKeySelector)
}
}
logsKeys, logsComplete, err := t.getLogsKeys(ctx, logsSelectors)
if err != nil {
return nil, false, err
}
auditKeys, auditComplete, err := t.getAuditKeys(ctx, auditSelectors)
if err != nil {
return nil, false, err
}
tracesKeys, tracesComplete, err := t.getTracesKeys(ctx, tracesSelectors)
if err != nil {
return nil, false, err
}
metricsKeys, metricsComplete, err := t.getMetricsKeys(ctx, metricsSelectors)
if err != nil {
return nil, false, err
}
meterSourceMetricsKeys, _, err := t.getMeterSourceMetricKeys(ctx, meterSourceMetricsSelectors)
if err != nil {
return nil, false, err
}
// Complete only if all queries are complete
complete := logsComplete && auditComplete && tracesComplete && metricsComplete
mapOfKeys := make(map[string][]*telemetrytypes.TelemetryFieldKey)
for _, key := range logsKeys {
mapOfKeys[key.Name] = append(mapOfKeys[key.Name], key)
}
for _, key := range auditKeys {
mapOfKeys[key.Name] = append(mapOfKeys[key.Name], key)
}
for _, key := range tracesKeys {
mapOfKeys[key.Name] = append(mapOfKeys[key.Name], key)
}
for _, key := range metricsKeys {
mapOfKeys[key.Name] = append(mapOfKeys[key.Name], key)
}
for _, key := range meterSourceMetricsKeys {
mapOfKeys[key.Name] = append(mapOfKeys[key.Name], key)
}
applyBackwardCompatibleKeys(mapOfKeys)
mapOfKeys = enrichWithIntrinsicMetricKeys(mapOfKeys, fieldKeySelectors)
return mapOfKeys, complete, nil
}
func (t *telemetryMetaStore) GetKey(ctx context.Context, fieldKeySelector *telemetrytypes.FieldKeySelector) ([]*telemetrytypes.TelemetryFieldKey, error) {
keys, _, err := t.GetKeys(ctx, fieldKeySelector)
if err != nil {
return nil, err
}
return keys[fieldKeySelector.Name], nil
}
func (t *telemetryMetaStore) getRelatedValues(ctx context.Context, fieldValueSelector *telemetrytypes.FieldValueSelector) ([]string, bool, error) {
ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
instrumentationtypes.TelemetrySignal: fieldValueSelector.Signal.StringValue(),
instrumentationtypes.CodeNamespace: "metadata",
instrumentationtypes.CodeFunctionName: "getRelatedValues",
})
// TODO(Tushar): thread orgID here to evaluate correctly
bodyJSONEnabled := t.fl.BooleanOrEmpty(ctx, flagger.FeatureUseJSONBody, featuretypes.NewFlaggerEvaluationContext(valuer.UUID{}))
// nothing to return as "related" value if there is nothing to filter on
if fieldValueSelector.ExistingQuery == "" {
return nil, true, nil
}
key := &telemetrytypes.TelemetryFieldKey{
Name: fieldValueSelector.Name,
Signal: fieldValueSelector.Signal,
FieldContext: fieldValueSelector.FieldContext,
FieldDataType: fieldValueSelector.FieldDataType,
}
selectColumn, err := t.fm.FieldFor(ctx, 0, 0, key)
if err != nil {
// we don't have a explicit column to select from the related metadata table
// so we will select either from resource_attributes or attributes table
// in that order
resourceColumn, _ := t.fm.FieldFor(ctx, 0, 0, &telemetrytypes.TelemetryFieldKey{
Name: key.Name,
FieldContext: telemetrytypes.FieldContextResource,
FieldDataType: telemetrytypes.FieldDataTypeString,
})
attributeColumn, _ := t.fm.FieldFor(ctx, 0, 0, &telemetrytypes.TelemetryFieldKey{
Name: key.Name,
FieldContext: telemetrytypes.FieldContextAttribute,
FieldDataType: telemetrytypes.FieldDataTypeString,
})
selectColumn = fmt.Sprintf("if(notEmpty(%s), %s, %s)", resourceColumn, resourceColumn, attributeColumn)
}
sb := sqlbuilder.Select("DISTINCT " + selectColumn).From(t.relatedMetadataDBName + "." + t.relatedMetadataTblName)
if len(fieldValueSelector.ExistingQuery) != 0 {
keySelectors := querybuilder.QueryStringToKeysSelectors(fieldValueSelector.ExistingQuery)
for _, keySelector := range keySelectors {
keySelector.Signal = fieldValueSelector.Signal
}
keys, _, err := t.GetKeysMulti(ctx, keySelectors)
if err != nil {
return nil, false, err
}
whereClause, err := querybuilder.PrepareWhereClause(fieldValueSelector.ExistingQuery, querybuilder.FilterExprVisitorOpts{
Context: ctx,
Logger: t.logger,
FieldMapper: t.fm,
ConditionBuilder: t.conditionBuilder,
FieldKeys: keys,
BodyJSONEnabled: bodyJSONEnabled,
})
if err != nil {
t.logger.WarnContext(ctx, "error parsing existing query for related values", errors.Attr(err))
}
if !whereClause.IsEmpty() {
sb.AddWhereClause(whereClause.WhereClause)
}
}
if fieldValueSelector.StartUnixMilli != 0 {
sb.Where(sb.GE("unix_milli", fieldValueSelector.StartUnixMilli))
}
if fieldValueSelector.EndUnixMilli != 0 {
sb.Where(sb.LE("unix_milli", fieldValueSelector.EndUnixMilli))
}
if fieldValueSelector.Value != "" {
var conds []string
if fieldValueSelector.FieldContext != telemetrytypes.FieldContextAttribute &&
fieldValueSelector.FieldContext != telemetrytypes.FieldContextResource {
origContext := key.FieldContext
// search on attributes
key.FieldContext = telemetrytypes.FieldContextAttribute
cond, err := t.conditionBuilder.ConditionFor(ctx, 0, 0, key, qbtypes.FilterOperatorContains, fieldValueSelector.Value, sb)
if err == nil {
conds = append(conds, cond)
}
// search on resource
key.FieldContext = telemetrytypes.FieldContextResource
cond, err = t.conditionBuilder.ConditionFor(ctx, 0, 0, key, qbtypes.FilterOperatorContains, fieldValueSelector.Value, sb)
if err == nil {
conds = append(conds, cond)
}
key.FieldContext = origContext
} else {
cond, err := t.conditionBuilder.ConditionFor(ctx, 0, 0, key, qbtypes.FilterOperatorContains, fieldValueSelector.Value, sb)
if err == nil {
conds = append(conds, cond)
}
}
if len(conds) != 0 {
// see `expr` in condition_builder.go, if key doesn't exist we don't check for value
// hence, this is join of conditions on resource and attributes
sb.Where(sb.And(conds...))
}
}
limit := fieldValueSelector.Limit
if limit == 0 {
limit = 50
}
// query one extra to check if we hit the limit
sb.Limit(limit + 1)
query, args := sb.BuildWithFlavor(datastoresql.Flavor)
t.logger.DebugContext(ctx, "query for related values", slog.String("query", query), slog.Any("args", args))
rows, err := t.telemetrystore.Datastore().Query(ctx, query, args...)
if err != nil {
return nil, false, ErrFailedToGetRelatedValues
}
defer rows.Close()
var attributeValues []string
rowCount := 0
for rows.Next() {
rowCount++
// reached the limit, we know there are more results
if rowCount > limit {
break
}
var value string
if err := rows.Scan(&value); err != nil {
return nil, false, ErrFailedToGetRelatedValues
}
if value != "" {
attributeValues = append(attributeValues, value)
}
}
// hit the limit?
complete := rowCount <= limit
return attributeValues, complete, nil
}
func (t *telemetryMetaStore) GetRelatedValues(ctx context.Context, fieldValueSelector *telemetrytypes.FieldValueSelector) ([]string, bool, error) {
return t.getRelatedValues(ctx, fieldValueSelector)
}
func (t *telemetryMetaStore) getSpanFieldValues(ctx context.Context, fieldValueSelector *telemetrytypes.FieldValueSelector) (*telemetrytypes.TelemetryFieldValues, bool, error) {
ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
instrumentationtypes.TelemetrySignal: telemetrytypes.SignalTraces.StringValue(),
instrumentationtypes.CodeNamespace: "metadata",
instrumentationtypes.CodeFunctionName: "getSpanFieldValues",
})
// build the query to get the keys from the spans that match the field selection criteria
limit := fieldValueSelector.Limit
if limit == 0 {
limit = 50
}
sb := sqlbuilder.Select("DISTINCT string_value, number_value").From(t.tracesDBName + "." + t.tracesFieldsTblName)
if fieldValueSelector.Name != "" {
sb.Where(sb.E("tag_key", fieldValueSelector.Name))
}
// now look at the field context
if fieldValueSelector.FieldContext != telemetrytypes.FieldContextUnspecified {
sb.Where(sb.E("tag_type", fieldValueSelector.FieldContext.TagType()))
}
// now look at the field data type
if fieldValueSelector.FieldDataType != telemetrytypes.FieldDataTypeUnspecified {
sb.Where(sb.E("tag_data_type", fieldValueSelector.FieldDataType.TagDataType()))
}
if fieldValueSelector.Value != "" {
switch fieldValueSelector.FieldDataType {
case telemetrytypes.FieldDataTypeString:
sb.Where(sb.ILike("string_value", "%"+escapeForLike(fieldValueSelector.Value)+"%"))
case telemetrytypes.FieldDataTypeNumber:
sb.Where(sb.IsNotNull("number_value"))
sb.Where(sb.ILike("toString(number_value)", "%"+escapeForLike(fieldValueSelector.Value)+"%"))
case telemetrytypes.FieldDataTypeUnspecified:
// or b/w string and number
sb.Where(sb.Or(
sb.ILike("string_value", "%"+escapeForLike(fieldValueSelector.Value)+"%"),
sb.ILike("toString(number_value)", "%"+escapeForLike(fieldValueSelector.Value)+"%"),
))
}
}
// query one extra to check if we hit the limit
sb.Limit(limit + 1)
query, args := sb.BuildWithFlavor(datastoresql.Flavor)
rows, err := t.telemetrystore.Datastore().Query(ctx, query, args...)
if err != nil {
return nil, false, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, ErrFailedToGetLogsKeys.Error())
}
defer rows.Close()
values := &telemetrytypes.TelemetryFieldValues{}
seen := make(map[string]bool)
rowCount := 0
totalCount := 0 // Track total unique values
for rows.Next() {
rowCount++
var stringValue string
var numberValue float64
if err := rows.Scan(&stringValue, &numberValue); err != nil {
return nil, false, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, ErrFailedToGetLogsKeys.Error())
}
// Only add values if we haven't hit the limit yet
if totalCount < limit {
if _, ok := seen[stringValue]; !ok && stringValue != "" {
values.StringValues = append(values.StringValues, stringValue)
seen[stringValue] = true
totalCount++
}
if _, ok := seen[fmt.Sprintf("%f", numberValue)]; !ok && numberValue != 0 && totalCount < limit {
values.NumberValues = append(values.NumberValues, numberValue)
seen[fmt.Sprintf("%f", numberValue)] = true
totalCount++
}
}
}
// hit the limit?
complete := rowCount <= limit
return values, complete, nil
}
func (t *telemetryMetaStore) getLogFieldValues(ctx context.Context, fieldValueSelector *telemetrytypes.FieldValueSelector) (*telemetrytypes.TelemetryFieldValues, bool, error) {
ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
instrumentationtypes.TelemetrySignal: telemetrytypes.SignalLogs.StringValue(),
instrumentationtypes.CodeNamespace: "metadata",
instrumentationtypes.CodeFunctionName: "getLogFieldValues",
})
// build the query to get the keys from the spans that match the field selection criteria
limit := fieldValueSelector.Limit
if limit == 0 {
limit = 50
}
sb := sqlbuilder.Select("DISTINCT string_value, number_value").From(t.logsDBName + "." + t.logsFieldsTblName)
if fieldValueSelector.Name != "" {
sb.Where(sb.E("tag_key", fieldValueSelector.Name))
}
if fieldValueSelector.FieldContext != telemetrytypes.FieldContextUnspecified {
sb.Where(sb.E("tag_type", fieldValueSelector.FieldContext.TagType()))
}
if fieldValueSelector.FieldDataType != telemetrytypes.FieldDataTypeUnspecified {
sb.Where(sb.E("tag_data_type", fieldValueSelector.FieldDataType.TagDataType()))
}
if fieldValueSelector.Value != "" {
switch fieldValueSelector.FieldDataType {
case telemetrytypes.FieldDataTypeString:
sb.Where(sb.ILike("string_value", "%"+escapeForLike(fieldValueSelector.Value)+"%"))
case telemetrytypes.FieldDataTypeNumber:
sb.Where(sb.IsNotNull("number_value"))
sb.Where(sb.ILike("toString(number_value)", "%"+escapeForLike(fieldValueSelector.Value)+"%"))
case telemetrytypes.FieldDataTypeUnspecified:
// or b/w string and number
sb.Where(sb.Or(
sb.ILike("string_value", "%"+escapeForLike(fieldValueSelector.Value)+"%"),
sb.ILike("toString(number_value)", "%"+escapeForLike(fieldValueSelector.Value)+"%"),
))
}
}
// query one extra to check if we hit the limit
sb.Limit(limit + 1)
query, args := sb.BuildWithFlavor(datastoresql.Flavor)
rows, err := t.telemetrystore.Datastore().Query(ctx, query, args...)
if err != nil {
return nil, false, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, ErrFailedToGetLogsKeys.Error())
}
defer rows.Close()
values := &telemetrytypes.TelemetryFieldValues{}
seen := make(map[string]bool)
rowCount := 0
totalCount := 0 // Track total unique values
for rows.Next() {
rowCount++
var stringValue string
var numberValue float64
if err := rows.Scan(&stringValue, &numberValue); err != nil {
return nil, false, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, ErrFailedToGetLogsKeys.Error())
}
// Only add values if we haven't hit the limit yet
if totalCount < limit {
if _, ok := seen[stringValue]; !ok && stringValue != "" {
values.StringValues = append(values.StringValues, stringValue)
seen[stringValue] = true
totalCount++
}
if _, ok := seen[fmt.Sprintf("%f", numberValue)]; !ok && numberValue != 0 && totalCount < limit {
values.NumberValues = append(values.NumberValues, numberValue)
seen[fmt.Sprintf("%f", numberValue)] = true
totalCount++
}
}
}
// hit the limit?
complete := rowCount <= limit
return values, complete, nil
}
func (t *telemetryMetaStore) getAuditFieldValues(ctx context.Context, fieldValueSelector *telemetrytypes.FieldValueSelector) (*telemetrytypes.TelemetryFieldValues, bool, error) {
ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
instrumentationtypes.TelemetrySignal: telemetrytypes.SignalLogs.StringValue(),
instrumentationtypes.CodeNamespace: "metadata",
instrumentationtypes.CodeFunctionName: "getAuditFieldValues",
})
limit := fieldValueSelector.Limit
if limit == 0 {
limit = 50
}
sb := sqlbuilder.Select("DISTINCT string_value, number_value").From(t.auditDBName + "." + t.auditFieldsTblName)
if fieldValueSelector.Name != "" {
sb.Where(sb.E("tag_key", fieldValueSelector.Name))
}
if fieldValueSelector.FieldContext != telemetrytypes.FieldContextUnspecified {
sb.Where(sb.E("tag_type", fieldValueSelector.FieldContext.TagType()))
}
if fieldValueSelector.FieldDataType != telemetrytypes.FieldDataTypeUnspecified {
sb.Where(sb.E("tag_data_type", fieldValueSelector.FieldDataType.TagDataType()))
}
if fieldValueSelector.Value != "" {
switch fieldValueSelector.FieldDataType {
case telemetrytypes.FieldDataTypeString:
sb.Where(sb.ILike("string_value", "%"+escapeForLike(fieldValueSelector.Value)+"%"))
case telemetrytypes.FieldDataTypeNumber:
sb.Where(sb.IsNotNull("number_value"))
sb.Where(sb.ILike("toString(number_value)", "%"+escapeForLike(fieldValueSelector.Value)+"%"))
case telemetrytypes.FieldDataTypeUnspecified:
sb.Where(sb.Or(
sb.ILike("string_value", "%"+escapeForLike(fieldValueSelector.Value)+"%"),
sb.ILike("toString(number_value)", "%"+escapeForLike(fieldValueSelector.Value)+"%"),
))
}
}
// fetch one extra row to detect whether the result set is complete
sb.Limit(limit + 1)
query, args := sb.BuildWithFlavor(datastoresql.Flavor)
rows, err := t.telemetrystore.Datastore().Query(ctx, query, args...)
if err != nil {
return nil, false, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, ErrFailedToGetAuditKeys.Error())
}
defer rows.Close()
values := &telemetrytypes.TelemetryFieldValues{}
seen := make(map[string]bool)
rowCount := 0
totalCount := 0
for rows.Next() {
rowCount++
var stringValue string
var numberValue float64
err = rows.Scan(&stringValue, &numberValue)
if err != nil {
return nil, false, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, ErrFailedToGetAuditKeys.Error())
}
if stringValue != "" && !seen[stringValue] {
if totalCount >= limit {
break
}
values.StringValues = append(values.StringValues, stringValue)
seen[stringValue] = true
totalCount++
}
if numberValue != 0 {
if totalCount >= limit {
break
}
if !seen[fmt.Sprintf("%f", numberValue)] {
values.NumberValues = append(values.NumberValues, numberValue)
seen[fmt.Sprintf("%f", numberValue)] = true
totalCount++
}
}
}
complete := rowCount <= limit
return values, complete, nil
}
// getMetricFieldValues returns field values and whether the result is complete.
func (t *telemetryMetaStore) getMetricFieldValues(ctx context.Context, orgID valuer.UUID, fieldValueSelector *telemetrytypes.FieldValueSelector) (*telemetrytypes.TelemetryFieldValues, bool, error) {
ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
instrumentationtypes.TelemetrySignal: telemetrytypes.SignalMetrics.StringValue(),
instrumentationtypes.CodeNamespace: "metadata",
instrumentationtypes.CodeFunctionName: "getMetricFieldValues",
})
limit := fieldValueSelector.Limit
if limit == 0 {
limit = 50
}
values, err := t.getIntrinsicMetricFieldValues(ctx, orgID, fieldValueSelector, limit)
if err != nil {
return nil, false, err
}
var remainingLimit int
if values == nil {
values = &telemetrytypes.TelemetryFieldValues{}
remainingLimit = limit
} else {
remainingLimit = limit - values.NumValues()
}
if remainingLimit == 0 {
return values, values.NumValues() < limit, nil
}
sb := sqlbuilder.
Select("DISTINCT attr_string_value").
From(t.metricsDBName + "." + t.metricsFieldsTblName)
if fieldValueSelector.Name != "" {
sb.Where(sb.E("attr_name", fieldValueSelector.Name))
}
if fieldValueSelector.FieldContext != telemetrytypes.FieldContextUnspecified {
sb.Where(sb.E("attr_type", fieldValueSelector.FieldContext.TagType()))
}
if fieldValueSelector.FieldDataType != telemetrytypes.FieldDataTypeUnspecified {
sb.Where(sb.E("attr_datatype", fieldValueSelector.FieldDataType.TagDataType()))
}
if fieldValueSelector.MetricContext != nil && fieldValueSelector.MetricContext.MetricName != "" {
sb.Where(sb.E("metric_name", fieldValueSelector.MetricContext.MetricName))
}
if fieldValueSelector.MetricContext != nil && fieldValueSelector.MetricContext.MetricNamespace != "" {
sb.Where(sb.Like("metric_name", escapeForLike(fieldValueSelector.MetricContext.MetricNamespace)+"%"))
}
if fieldValueSelector.StartUnixMilli > 0 {
sb.Where(sb.GE("last_reported_unix_milli", fieldValueSelector.StartUnixMilli))
}
if fieldValueSelector.EndUnixMilli > 0 {
sb.Where(sb.LE("first_reported_unix_milli", fieldValueSelector.EndUnixMilli))
}
if fieldValueSelector.Value != "" {
if fieldValueSelector.SelectorMatchType == telemetrytypes.FieldSelectorMatchTypeExact {
sb.Where(sb.E("attr_string_value", fieldValueSelector.Value))
} else {
sb.Where(sb.ILike("attr_string_value", "%"+escapeForLike(fieldValueSelector.Value)+"%"))
}
}
// query one extra to check if we hit the limit
sb.Limit(remainingLimit + 1)
query, args := sb.BuildWithFlavor(datastoresql.Flavor)
rows, err := t.telemetrystore.Datastore().Query(ctx, query, args...)
if err != nil {
return nil, false, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, ErrFailedToGetMetricsKeys.Error())
}
defer rows.Close()
rowCount := 0
for rows.Next() {
rowCount++
if rowCount > remainingLimit {
break
}
var stringValue string
if err := rows.Scan(&stringValue); err != nil {
return nil, false, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, ErrFailedToGetMetricsKeys.Error())
}
values.StringValues = append(values.StringValues, stringValue)
}
// hit the limit?
complete := values.NumValues() < limit // decision on total limit
return values, complete, nil
}
// getIntrinsicMetricFieldValues returns values, isSearchComplete, error.
func (t *telemetryMetaStore) getIntrinsicMetricFieldValues(ctx context.Context, orgID valuer.UUID, fieldValueSelector *telemetrytypes.FieldValueSelector, limit int) (*telemetrytypes.TelemetryFieldValues, error) {
ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
instrumentationtypes.TelemetrySignal: telemetrytypes.SignalMetrics.StringValue(),
instrumentationtypes.CodeNamespace: "metadata",
instrumentationtypes.CodeFunctionName: "getIntrinsicMetricFieldValues",
})
key, ok := telemetrymetrics.IntrinsicMetricFieldDefinitions[fieldValueSelector.Name]
if !ok {
return &telemetrytypes.TelemetryFieldValues{}, nil
}
if fieldValueSelector.Signal != telemetrytypes.SignalMetrics && fieldValueSelector.Signal != telemetrytypes.SignalUnspecified {
return &telemetrytypes.TelemetryFieldValues{}, nil
}
// if the field context or data type does not match for the requested key (and is not unspecified as well),
// we should return false to enable searching in metadata table.
if fieldValueSelector.FieldContext != telemetrytypes.FieldContextUnspecified && fieldValueSelector.FieldContext != key.FieldContext {
return &telemetrytypes.TelemetryFieldValues{}, nil
}
if fieldValueSelector.FieldDataType != telemetrytypes.FieldDataTypeUnspecified && fieldValueSelector.FieldDataType != key.FieldDataType {
return &telemetrytypes.TelemetryFieldValues{}, nil
}
// no values are surfaced for intrinsic boolean fields.
if key.FieldDataType == telemetrytypes.FieldDataTypeBool {
return &telemetrytypes.TelemetryFieldValues{}, nil
}
// Reduced metrics live in time_series_v4_reduced + the buffer (is_reduced=false), not the raw
// tables; query those only when reduction is enabled (the tables may not exist otherwise).
reductionEnabled := t.fl.BooleanOrEmpty(ctx, flagger.FeatureEnableMetricsReduction, featuretypes.NewFlaggerEvaluationContext(orgID))
sources := []struct {
tableName string
extraConds []string
}{
{telemetrymetrics.Series1wTableName, nil},
}
if reductionEnabled {
sources = append(sources,
struct {
tableName string
extraConds []string
}{telemetrymetrics.SeriesBufferTableName, []string{"is_reduced = false"}},
struct {
tableName string
extraConds []string
}{telemetrymetrics.SeriesReducedTableName, nil},
)
}
values := &telemetrytypes.TelemetryFieldValues{}
seen := make(map[string]struct{})
for _, src := range sources {
if len(values.StringValues) >= limit {
break
}
tableValues, err := t.getIntrinsicMetricFieldValuesForTable(ctx, fieldValueSelector, key, src.tableName, limit, src.extraConds...)
if err != nil {
return nil, err
}
for _, str := range tableValues {
if _, ok := seen[str]; ok {
continue
}
seen[str] = struct{}{}
values.StringValues = append(values.StringValues, str)
if len(values.StringValues) >= limit {
break
}
}
}
return values, nil
}
// getIntrinsicMetricFieldValuesForTable returns the distinct intrinsic field values for the
// given key from a single time series table. extraConds are appended verbatim (e.g. the
// buffer's "is_reduced = false" filter).
func (t *telemetryMetaStore) getIntrinsicMetricFieldValuesForTable(ctx context.Context, fieldValueSelector *telemetrytypes.FieldValueSelector, key telemetrytypes.TelemetryFieldKey, tableName string, limit int, extraConds ...string) ([]string, error) {
sb := sqlbuilder.Select(sqlbuilder.Escape(key.Name)).
From(t.metricsDBName + "." + tableName)
if fieldValueSelector.MetricContext != nil && fieldValueSelector.MetricContext.MetricName != "" {
sb.Where(sb.E("metric_name", fieldValueSelector.MetricContext.MetricName))
}
if fieldValueSelector.MetricContext != nil && fieldValueSelector.MetricContext.MetricNamespace != "" {
sb.Where(sb.Like("metric_name", escapeForLike(fieldValueSelector.MetricContext.MetricNamespace)+"%"))
}
if fieldValueSelector.StartUnixMilli > 0 {
sb.Where(sb.GE("unix_milli", fieldValueSelector.StartUnixMilli))
}
if fieldValueSelector.EndUnixMilli > 0 {
sb.Where(sb.LE("unix_milli", fieldValueSelector.EndUnixMilli))
}
if fieldValueSelector.Value != "" {
if fieldValueSelector.SelectorMatchType == telemetrytypes.FieldSelectorMatchTypeExact {
sb.Where(sb.E(key.Name, fieldValueSelector.Value))
} else {
sb.Where(sb.ILike(key.Name, "%"+escapeForLike(fieldValueSelector.Value)+"%"))
}
}
for _, cond := range extraConds {
sb.Where(cond)
}
sb.GroupBy(sqlbuilder.Escape(key.Name))
sb.Limit(limit + 1)
query, args := sb.BuildWithFlavor(datastoresql.Flavor)
rows, err := t.telemetrystore.Datastore().Query(ctx, query, args...)
if err != nil {
return nil, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, ErrFailedToGetMetricsKeys.Error())
}
defer rows.Close()
var values []string
rowCount := 0
for rows.Next() {
rowCount++
if rowCount > limit {
break
}
var str string
if err := rows.Scan(&str); err != nil {
return nil, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, ErrFailedToGetMetricsKeys.Error())
}
values = append(values, str)
}
return values, nil
}
func (t *telemetryMetaStore) getMeterSourceMetricFieldValues(ctx context.Context, fieldValueSelector *telemetrytypes.FieldValueSelector) (*telemetrytypes.TelemetryFieldValues, bool, error) {
ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
instrumentationtypes.TelemetrySignal: telemetrytypes.SignalMetrics.StringValue(),
instrumentationtypes.CodeNamespace: "metadata",
instrumentationtypes.CodeFunctionName: "getMeterSourceMetricFieldValues",
})
sb := sqlbuilder.Select("DISTINCT arrayJoin(JSONExtractKeysAndValues(labels, 'String')) AS attr").
From(t.meterDBName + "." + t.meterFieldsTblName)
if fieldValueSelector.Name != "" {
sb.Where(sb.E("attr.1", fieldValueSelector.Name))
}
sb.Where(sb.NotLike("attr.1", "\\_\\_%"))
if fieldValueSelector.MetricContext != nil && fieldValueSelector.MetricContext.MetricName != "" {
sb.Where(sb.E("metric_name", fieldValueSelector.MetricContext.MetricName))
}
if fieldValueSelector.MetricContext != nil && fieldValueSelector.MetricContext.MetricNamespace != "" {
sb.Where(sb.Like("metric_name", escapeForLike(fieldValueSelector.MetricContext.MetricNamespace)+"%"))
}
if fieldValueSelector.Value != "" {
if fieldValueSelector.SelectorMatchType == telemetrytypes.FieldSelectorMatchTypeExact {
sb.Where(sb.E("attr.2", fieldValueSelector.Value))
} else {
sb.Where(sb.Like("attr.2", "%"+fieldValueSelector.Value+"%"))
}
}
sb.Where(sb.NE("attr.2", ""))
limit := fieldValueSelector.Limit
if limit == 0 {
limit = 50
}
// query one extra to check if we hit the limit
sb.Limit(limit + 1)
query, args := sb.BuildWithFlavor(datastoresql.Flavor)
rows, err := t.telemetrystore.Datastore().Query(ctx, query, args...)
if err != nil {
return nil, false, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, ErrFailedToGetMeterValues.Error())
}
defer rows.Close()
values := &telemetrytypes.TelemetryFieldValues{}
rowCount := 0
for rows.Next() {
rowCount++
// reached the limit, we know there are more results
if rowCount > limit {
break
}
var attribute []string
if err := rows.Scan(&attribute); err != nil {
return nil, false, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, ErrFailedToGetMeterValues.Error())
}
if len(attribute) > 1 {
values.StringValues = append(values.StringValues, attribute[1])
}
}
// hit the limit?
complete := rowCount <= limit
return values, complete, nil
}
func populateAllUnspecifiedValues(allUnspecifiedValues *telemetrytypes.TelemetryFieldValues, mapOfValues map[any]bool, mapOfRelatedValues map[any]bool, values *telemetrytypes.TelemetryFieldValues, limit int) bool {
complete := true
totalCount := len(mapOfValues) + len(mapOfRelatedValues)
for _, value := range values.StringValues {
if totalCount >= limit {
complete = false
break
}
if _, ok := mapOfValues[value]; !ok {
mapOfValues[value] = true
allUnspecifiedValues.StringValues = append(allUnspecifiedValues.StringValues, value)
totalCount++
}
}
for _, value := range values.NumberValues {
if totalCount >= limit {
complete = false
break
}
if _, ok := mapOfValues[value]; !ok {
mapOfValues[value] = true
allUnspecifiedValues.NumberValues = append(allUnspecifiedValues.NumberValues, value)
totalCount++
}
}
for _, value := range values.RelatedValues {
if totalCount >= limit {
complete = false
break
}
if _, ok := mapOfRelatedValues[value]; !ok {
mapOfRelatedValues[value] = true
allUnspecifiedValues.RelatedValues = append(allUnspecifiedValues.RelatedValues, value)
totalCount++
}
}
return complete
}
// GetAllValues returns all values and whether the result is complete.
func (t *telemetryMetaStore) GetAllValues(ctx context.Context, orgID valuer.UUID, fieldValueSelector *telemetrytypes.FieldValueSelector) (*telemetrytypes.TelemetryFieldValues, bool, error) {
values := &telemetrytypes.TelemetryFieldValues{}
var complete = true
var err error
limit := fieldValueSelector.Limit
if limit == 0 {
limit = 50
}
switch fieldValueSelector.Signal {
case telemetrytypes.SignalTraces:
values, complete, err = t.getSpanFieldValues(ctx, fieldValueSelector)
case telemetrytypes.SignalLogs:
if fieldValueSelector.Source == telemetrytypes.SourceAudit {
values, complete, err = t.getAuditFieldValues(ctx, fieldValueSelector)
} else {
values, complete, err = t.getLogFieldValues(ctx, fieldValueSelector)
}
case telemetrytypes.SignalMetrics:
if fieldValueSelector.Source == telemetrytypes.SourceMeter {
values, complete, err = t.getMeterSourceMetricFieldValues(ctx, fieldValueSelector)
} else {
values, complete, err = t.getMetricFieldValues(ctx, orgID, fieldValueSelector)
}
case telemetrytypes.SignalUnspecified:
mapOfValues := make(map[any]bool)
mapOfRelatedValues := make(map[any]bool)
allUnspecifiedValues := &telemetrytypes.TelemetryFieldValues{}
tracesValues, tracesComplete, err := t.getSpanFieldValues(ctx, fieldValueSelector)
if err == nil {
populateComplete := populateAllUnspecifiedValues(allUnspecifiedValues, mapOfValues, mapOfRelatedValues, tracesValues, limit)
complete = complete && tracesComplete && populateComplete
}
logsValues, logsComplete, err := t.getLogFieldValues(ctx, fieldValueSelector)
if err == nil {
populateComplete := populateAllUnspecifiedValues(allUnspecifiedValues, mapOfValues, mapOfRelatedValues, logsValues, limit)
complete = complete && logsComplete && populateComplete
}
metricsValues, metricsComplete, err := t.getMetricFieldValues(ctx, orgID, fieldValueSelector)
if err == nil {
populateComplete := populateAllUnspecifiedValues(allUnspecifiedValues, mapOfValues, mapOfRelatedValues, metricsValues, limit)
complete = complete && metricsComplete && populateComplete
}
values = allUnspecifiedValues
}
if err != nil {
return nil, false, err
}
return values, complete, nil
}
func (t *telemetryMetaStore) FetchTemporality(ctx context.Context, orgID valuer.UUID, queryTimeRangeStartTs, queryTimeRangeEndTs uint64, metricName string) (metrictypes.Temporality, error) {
if metricName == "" {
return metrictypes.Unknown, errors.Newf(errors.TypeInternal, errors.CodeInternal, "metric name cannot be empty")
}
temporalityMap, err := t.FetchTemporalityMulti(ctx, orgID, queryTimeRangeStartTs, queryTimeRangeEndTs, metricName)
if err != nil {
return metrictypes.Unknown, err
}
temporality, ok := temporalityMap[metricName]
if !ok {
return metrictypes.Unknown, nil
}
return temporality, nil
}
func (t *telemetryMetaStore) FetchTemporalityMulti(ctx context.Context, orgID valuer.UUID, queryTimeRangeStartTs, queryTimeRangeEndTs uint64, metricNames ...string) (map[string]metrictypes.Temporality, error) {
temporalities, _, _, err := t.FetchTemporalityAndTypeMulti(ctx, orgID, queryTimeRangeStartTs, queryTimeRangeEndTs, metricNames...)
return temporalities, err
}
func (t *telemetryMetaStore) FetchTemporalityAndTypeMulti(ctx context.Context, orgID valuer.UUID, queryTimeRangeStartTs, queryTimeRangeEndTs uint64, metricNames ...string) (map[string]metrictypes.Temporality, map[string]metrictypes.Type, map[string]bool, error) {
if len(metricNames) == 0 {
return make(map[string]metrictypes.Temporality), make(map[string]metrictypes.Type), make(map[string]bool), nil
}
reductionEnabled := t.fl.BooleanOrEmpty(ctx, flagger.FeatureEnableMetricsReduction, featuretypes.NewFlaggerEvaluationContext(orgID))
temporalities := make(map[string]metrictypes.Temporality)
types := make(map[string]metrictypes.Type)
metricsTemporality, metricTypes, reduced, err := t.fetchMetricsTemporalityAndType(ctx, queryTimeRangeStartTs, queryTimeRangeEndTs, reductionEnabled, metricNames...)
if err != nil {
return nil, nil, nil, err
}
meterMetricsTemporality, meterMetricsTypes, err := t.fetchMeterSourceMetricsTemporalityAndType(ctx, metricNames...)
if err != nil {
return nil, nil, nil, err
}
// For metrics not found in the database, set to Unknown
for _, metricName := range metricNames {
if temporality, exists := metricsTemporality[metricName]; exists && len(temporality) > 0 {
if len(temporality) > 1 {
temporalities[metricName] = metrictypes.Multiple
} else {
temporalities[metricName] = temporality[0]
}
} else if temporality, exists := meterMetricsTemporality[metricName]; exists {
temporalities[metricName] = temporality
} else {
temporalities[metricName] = metrictypes.Unknown
}
if metricType, exists := metricTypes[metricName]; exists {
types[metricName] = metricType
} else if meterMetricType, exists := meterMetricsTypes[metricName]; exists {
types[metricName] = meterMetricType
} else {
types[metricName] = metrictypes.UnspecifiedType
}
}
return temporalities, types, reduced, nil
}
func (t *telemetryMetaStore) fetchMetricsTemporalityAndType(ctx context.Context, queryTimeRangeStartTs, queryTimeRangeEndTs uint64, reductionEnabled bool, metricNames ...string) (map[string][]metrictypes.Temporality, map[string]metrictypes.Type, map[string]bool, error) {
ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
instrumentationtypes.TelemetrySignal: telemetrytypes.SignalMetrics.StringValue(),
instrumentationtypes.CodeNamespace: "metadata",
instrumentationtypes.CodeFunctionName: "fetchMetricsTemporalityAndType",
})
adjustedStartTs, adjustedEndTs, tsTableName, _ := telemetrymetrics.WhichTSTableToUse(queryTimeRangeStartTs, queryTimeRangeEndTs, false, nil)
temporalities, types, err := t.fetchTemporalityTypeForTable(ctx, tsTableName, adjustedStartTs, adjustedEndTs, metricNames)
if err != nil {
return nil, nil, nil, err
}
reduced := make(map[string]bool)
if reductionEnabled {
bufferTemporalities, bufferTypes, err := t.fetchTemporalityTypeForTable(ctx, telemetrymetrics.SeriesBufferTableName, adjustedStartTs, adjustedEndTs, metricNames, "is_reduced = false")
if err != nil {
return nil, nil, nil, err
}
for metricName, bufferTemps := range bufferTemporalities {
temporalities[metricName] = unionTemporalities(temporalities[metricName], bufferTemps)
}
for metricName, metricType := range bufferTypes {
if _, ok := types[metricName]; !ok {
types[metricName] = metricType
}
}
reducedTemporalities, reducedTypes, err := t.fetchTemporalityTypeForTable(ctx, telemetrymetrics.SeriesReducedTableName, adjustedStartTs, adjustedEndTs, metricNames)
if err != nil {
return nil, nil, nil, err
}
for metricName := range reducedTypes {
reduced[metricName] = true
}
for metricName, reducedTemps := range reducedTemporalities {
if len(temporalities[metricName]) == 0 {
temporalities[metricName] = reducedTemps
}
}
for metricName, metricType := range reducedTypes {
if _, ok := types[metricName]; !ok {
types[metricName] = metricType
}
}
}
return temporalities, types, reduced, nil
}
// unionTemporalities appends the distinct temporalities from additional to existing.
func unionTemporalities(existing, additional []metrictypes.Temporality) []metrictypes.Temporality {
for _, t := range additional {
found := false
for _, e := range existing {
if e == t {
found = true
break
}
}
if !found {
existing = append(existing, t)
}
}
return existing
}
func (t *telemetryMetaStore) fetchTemporalityTypeForTable(ctx context.Context, tableName string, adjustedStartTs, adjustedEndTs uint64, metricNames []string, extraConds ...string) (map[string][]metrictypes.Temporality, map[string]metrictypes.Type, error) {
temporalities := make(map[string][]metrictypes.Temporality)
types := make(map[string]metrictypes.Type)
sb := sqlbuilder.NewSelectBuilder()
sb.Select("metric_name", "temporality", "any(type) AS type", "any(is_monotonic) as is_monotonic")
sb.From(t.metricsDBName + "." + tableName)
conds := []string{
sb.In("metric_name", metricNames),
sb.GTE("unix_milli", adjustedStartTs),
sb.LT("unix_milli", adjustedEndTs),
}
conds = append(conds, extraConds...)
sb.Where(conds...)
sb.GroupBy("metric_name", "temporality")
query, args := sb.BuildWithFlavor(datastoresql.Flavor)
t.logger.DebugContext(ctx, "fetching metric temporality", slog.String("query", query), slog.Any("args", args))
rows, err := t.telemetrystore.Datastore().Query(ctx, query, args...)
if err != nil {
return nil, nil, errors.Wrapf(err, errors.TypeInternal, errors.CodeInternal, "failed to fetch metric temporality")
}
defer rows.Close()
for rows.Next() {
var metricName string
var temporality metrictypes.Temporality
var metricType metrictypes.Type
var isMonotonic bool
if err := rows.Scan(&metricName, &temporality, &metricType, &isMonotonic); err != nil {
return nil, nil, errors.Wrapf(err, errors.TypeInternal, errors.CodeInternal, "failed to scan temporality result")
}
if temporality != metrictypes.Unknown {
temporalities[metricName] = append(temporalities[metricName], temporality)
}
if metricType == metrictypes.SumType && !isMonotonic {
metricType = metrictypes.GaugeType
}
types[metricName] = metricType
}
if err := rows.Err(); err != nil {
return nil, nil, errors.Wrapf(err, errors.TypeInternal, errors.CodeInternal, "error iterating over metrics temporality rows")
}
return temporalities, types, nil
}
func (t *telemetryMetaStore) fetchMeterSourceMetricsTemporalityAndType(ctx context.Context, metricNames ...string) (map[string]metrictypes.Temporality, map[string]metrictypes.Type, error) {
ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
instrumentationtypes.TelemetrySignal: telemetrytypes.SignalMetrics.StringValue(),
instrumentationtypes.CodeNamespace: "metadata",
instrumentationtypes.CodeFunctionName: "fetchMeterSourceMetricsTemporalityAndType",
})
temporalities := make(map[string]metrictypes.Temporality)
types := make(map[string]metrictypes.Type)
sb := sqlbuilder.Select(
"metric_name",
"argMax(temporality, unix_milli) as temporality",
"any(type) AS type",
"any(is_monotonic) as is_monotonic",
).From(t.meterDBName + "." + t.meterFieldsTblName)
// Filter by metric names (in the temporality column due to data mix-up)
sb.Where(sb.In("metric_name", metricNames))
// Group by metric name to get one temporality per metric
sb.GroupBy("metric_name")
query, args := sb.BuildWithFlavor(datastoresql.Flavor)
t.logger.DebugContext(ctx, "fetching meter metrics temporality", slog.String("query", query), slog.Any("args", args))
rows, err := t.telemetrystore.Datastore().Query(ctx, query, args...)
if err != nil {
return nil, nil, errors.Wrapf(err, errors.TypeInternal, errors.CodeInternal, "failed to fetch meter metric temporality")
}
defer rows.Close()
// Process results
for rows.Next() {
var metricName string
var temporality metrictypes.Temporality
var metricType metrictypes.Type
var isMonotonic bool
if err := rows.Scan(&metricName, &temporality, &metricType, &isMonotonic); err != nil {
return nil, nil, errors.Wrapf(err, errors.TypeInternal, errors.CodeInternal, "failed to scan temporality result")
}
if metricType == metrictypes.SumType && !isMonotonic {
metricType = metrictypes.GaugeType
}
temporalities[metricName] = temporality
types[metricName] = metricType
}
return temporalities, types, nil
}
func (k *telemetryMetaStore) fetchEvolutionEntryFromDatastore(ctx context.Context, selectors []*telemetrytypes.EvolutionSelector) ([]*telemetrytypes.EvolutionEntry, error) {
sb := sqlbuilder.NewSelectBuilder()
sb.Select("signal", "column_name", "column_type", "field_context", "field_name", "version", "release_time")
sb.From(fmt.Sprintf("%s.%s", k.relatedMetadataDBName, k.columnEvolutionMetadataTblName))
sb.OrderBy("release_time ASC")
var clauses []string
for _, selector := range selectors {
var clause string
if selector.FieldContext != telemetrytypes.FieldContextUnspecified {
clause = sb.E("field_context", selector.FieldContext)
}
clause = sb.And(clause,
sb.Or(sb.E("field_name", selector.FieldName), sb.E("field_name", "__all__")),
)
clauses = append(clauses, sb.And(sb.E("signal", selector.Signal), clause))
}
sb.Where(sb.Or(clauses...))
query, args := sb.BuildWithFlavor(datastoresql.Flavor)
var entries []*telemetrytypes.EvolutionEntry
rows, err := k.telemetrystore.Datastore().Query(ctx, query, args...)
if err != nil {
return nil, err
}
defer rows.Close()
for rows.Next() {
var entry telemetrytypes.EvolutionEntry
var releaseTimeNs float64
if err := rows.Scan(
&entry.Signal,
&entry.ColumnName,
&entry.ColumnType,
&entry.FieldContext,
&entry.FieldName,
&entry.Version,
&releaseTimeNs,
); err != nil {
return nil, err
}
// Convert nanoseconds to time.Time
releaseTime := time.Unix(0, int64(releaseTimeNs))
entry.ReleaseTime = releaseTime
entries = append(entries, &entry)
}
if err := rows.Err(); err != nil {
return nil, err
}
return entries, nil
}
// updateColumnEvolutionMetadataForKeys updates the evolution field for keys.
func (k *telemetryMetaStore) updateColumnEvolutionMetadataForKeys(ctx context.Context, keysToUpdate []*telemetrytypes.TelemetryFieldKey) error {
var metadataKeySelectors []*telemetrytypes.EvolutionSelector
for _, keySelector := range keysToUpdate {
selector := &telemetrytypes.EvolutionSelector{
Signal: keySelector.Signal,
FieldContext: keySelector.FieldContext,
FieldName: keySelector.Name,
}
metadataKeySelectors = append(metadataKeySelectors, selector)
}
evolutions, err := k.fetchEvolutionEntryFromDatastore(ctx, metadataKeySelectors)
if err != nil {
return errors.Newf(errors.TypeInternal, errors.CodeInternal, "failed to fetch evolution from datastore %s", err.Error())
}
evolutionsByUniqueKey := make(map[string][]*telemetrytypes.EvolutionEntry)
for _, evolution := range evolutions {
key := &telemetrytypes.EvolutionSelector{
Signal: evolution.Signal,
FieldContext: evolution.FieldContext,
FieldName: evolution.FieldName,
}
evolutionsByUniqueKey[key.QualifiedName()] = append(evolutionsByUniqueKey[key.QualifiedName()], evolution)
}
if len(keysToUpdate) > 0 {
for i, key := range keysToUpdate {
selector := &telemetrytypes.EvolutionSelector{
Signal: key.Signal,
FieldContext: key.FieldContext,
FieldName: "__all__",
}
// first check if there is evolutions that with field name as __all__
if keyEvolutions, ok := evolutionsByUniqueKey[selector.QualifiedName()]; ok {
keysToUpdate[i].Evolutions = keyEvolutions
}
// then check for specific field name
selector.FieldName = key.Name
if keyEvolutions, ok := evolutionsByUniqueKey[selector.QualifiedName()]; ok {
keysToUpdate[i].Evolutions = keyEvolutions
}
}
}
return nil
}
// chunkSizeFirstSeenMetricMetadata limits the number of tuples per SQL query to avoid hitting the max_query_size limit.
//
// Calculation Logic:
//
// 1. The Ceiling: 250 KB (256,000 bytes). A conservative safety limit set just below the common DB max_query_size (262KB)
// to guarantee the database does not reject the query.
// Reference: https://datastore.com/docs/operations/settings/settings#max_query_size
//
// 2. Unit Cost: ~150 bytes per tuple. The estimated "weight" of a single lookup key, summing MetricName (40B),
// AttrName (30B), AttrValue (64B), and SQL syntax overhead (16B).
//
// 3. Theoretical Max: ~1,706 tuples. The absolute maximum capacity if all keys adhere to the average size (256,000 / 150).
//
// 4. Final Limit: 1600. Rounds down to provide a ~6% safety buffer, accounting for data outliers
// (unusually long attribute values) and multi-byte character expansion.
const chunkSizeFirstSeenMetricMetadata = 1600
// GetFirstSeenFromMetricMetadata queries the metadata table to get the first_seen timestamp
// for each metric-attribute-value combination.
// Returns a map where key is `telemetrytypes.MetricMetadataLookupKey` and value is first_seen in milliseconds.
func (t *telemetryMetaStore) GetFirstSeenFromMetricMetadata(ctx context.Context, lookupKeys []telemetrytypes.MetricMetadataLookupKey) (map[telemetrytypes.MetricMetadataLookupKey]int64, error) {
ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
instrumentationtypes.TelemetrySignal: telemetrytypes.SignalMetrics.StringValue(),
instrumentationtypes.CodeNamespace: "metadata",
instrumentationtypes.CodeFunctionName: "GetFirstSeenFromMetricMetadata",
})
result := make(map[telemetrytypes.MetricMetadataLookupKey]int64)
for i := 0; i < len(lookupKeys); i += chunkSizeFirstSeenMetricMetadata {
end := i + chunkSizeFirstSeenMetricMetadata
if end > len(lookupKeys) {
end = len(lookupKeys)
}
chunk := lookupKeys[i:end]
sb := sqlbuilder.Select(
"metric_name",
"attr_name",
"attr_string_value",
"min(first_reported_unix_milli) AS first_seen",
).From(t.metricsDBName + "." + t.metricsFieldsTblName)
lookupItems := make([]interface{}, 0, len(chunk))
for _, key := range chunk {
lookupItems = append(lookupItems, sqlbuilder.Tuple(key.MetricName, key.AttributeName, key.AttributeValue))
}
sb.Where(
sb.In(
sqlbuilder.TupleNames("metric_name", "attr_name", "attr_string_value"),
lookupItems...,
),
)
sb.GroupBy("metric_name", "attr_name", "attr_string_value")
sb.OrderBy("first_seen")
query, args := sb.BuildWithFlavor(datastoresql.Flavor)
rows, err := t.telemetrystore.Datastore().Query(ctx, query, args...)
if err != nil {
return nil, errors.Wrapf(err, errors.TypeInternal, errors.CodeInternal, "failed to query metadata for first_seen")
}
for rows.Next() {
var metricName, attrName, attrValue string
var firstSeen uint64
if err := rows.Scan(&metricName, &attrName, &attrValue, &firstSeen); err != nil {
rows.Close()
return nil, errors.Wrapf(err, errors.TypeInternal, errors.CodeInternal, "failed to scan metadata first_seen result")
}
result[telemetrytypes.MetricMetadataLookupKey{
MetricName: metricName,
AttributeName: attrName,
AttributeValue: attrValue,
}] = int64(firstSeen)
}
if err := rows.Err(); err != nil {
rows.Close()
return nil, errors.Wrapf(err, errors.TypeInternal, errors.CodeInternal, "failed to iterate metadata first_seen results")
}
rows.Close()
}
return result, nil
}
func (t *telemetryMetaStore) FetchLastSeenInfoMulti(ctx context.Context, orgID valuer.UUID, metricNames ...string) (map[string]int64, error) {
lastSeenInfo, err := t.fetchLastSeenInfoForTable(ctx, telemetrymetrics.SeriesTableName, metricNames)
if err != nil {
return nil, err
}
if t.fl.BooleanOrEmpty(ctx, flagger.FeatureEnableMetricsReduction, featuretypes.NewFlaggerEvaluationContext(orgID)) {
reducedLastSeen, err := t.fetchLastSeenInfoForTable(ctx, telemetrymetrics.SeriesReducedTableName, metricNames)
if err != nil {
return nil, err
}
for metricName, ts := range reducedLastSeen {
if ts > lastSeenInfo[metricName] {
lastSeenInfo[metricName] = ts
}
}
}
return lastSeenInfo, nil
}
func (t *telemetryMetaStore) fetchLastSeenInfoForTable(ctx context.Context, tableName string, metricNames []string) (map[string]int64, error) {
sb := sqlbuilder.Select(
"metric_name",
"max(unix_milli)",
).
From(t.metricsDBName + "." + tableName)
sb.Where(sb.In("metric_name", metricNames))
sb.GroupBy("metric_name")
query, args := sb.BuildWithFlavor(datastoresql.Flavor)
t.logger.DebugContext(ctx, "fetching metric last seen timestamp", slog.String("query", query), slog.Any("args", args))
rows, err := t.telemetrystore.Datastore().Query(ctx, query, args...)
if err != nil {
return nil, errors.Wrapf(err, errors.TypeInternal, errors.CodeInternal, "failed to fetch metric last seen info")
}
defer rows.Close()
lastSeenInfo := make(map[string]int64)
for rows.Next() {
var metricName string
var unix_milli int64
if err := rows.Scan(&metricName, &unix_milli); err != nil {
return nil, errors.Wrapf(err, errors.TypeInternal, errors.CodeInternal, "failed to scan last seen info result")
}
lastSeenInfo[metricName] = unix_milli
}
if err := rows.Err(); err != nil {
return nil, errors.Wrapf(err, errors.TypeInternal, errors.CodeInternal, "error iterating over metrics temporality rows")
}
return lastSeenInfo, nil
}