mirror of
https://github.com/hanzoai/o11y.git
synced 2026-10-15 12:58:33 +00:00
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]>
2675 lines
94 KiB
Go
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
|
|
}
|