mirror of
https://github.com/SigNoz/signoz.git
synced 2026-10-28 20:10:43 +00:00
Compare commits
13 Commits
refactor/s
...
tvats-quer
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
b63261c6b9 | ||
|
|
fc77ee804b | ||
|
|
4c29aa1f54 | ||
|
|
4f8447fb7d | ||
|
|
a1515ab548 | ||
|
|
1137ea2274 | ||
|
|
bddb64baac | ||
|
|
3558177799 | ||
|
|
212561bfa8 | ||
|
|
80f5017773 | ||
|
|
95f15cfedf | ||
|
|
9de634e494 | ||
|
|
b932bd9b84 |
@@ -44,6 +44,7 @@ import (
|
||||
"github.com/SigNoz/signoz/pkg/ruler/signozruler"
|
||||
"github.com/SigNoz/signoz/pkg/signoz"
|
||||
"github.com/SigNoz/signoz/pkg/sqlstore"
|
||||
"github.com/SigNoz/signoz/pkg/statementbuilder/noopauditstatementbuilder"
|
||||
"github.com/SigNoz/signoz/pkg/subscription"
|
||||
"github.com/SigNoz/signoz/pkg/subscription/noopsubscription"
|
||||
"github.com/SigNoz/signoz/pkg/telemetrystore"
|
||||
@@ -121,6 +122,10 @@ func runServer(ctx context.Context, config signoz.Config, logger *slog.Logger) e
|
||||
func(_ context.Context, _ factory.ProviderSettings, _ flagger.Flagger, _ licensing.Licensing, _ telemetrystore.TelemetryStore, _ retention.Getter, _ organization.Getter, _ zeus.Zeus) (factory.NamedMap[factory.ProviderFactory[meterreporter.Reporter, meterreporter.Config]], string) {
|
||||
return signoz.NewMeterReporterProviderFactories(), "noop"
|
||||
},
|
||||
[]signoz.LogSourceQuery{{
|
||||
Source: telemetrytypes.SourceAudit,
|
||||
StatementBuilderFactory: noopauditstatementbuilder.NewFactory,
|
||||
}},
|
||||
func(ps factory.ProviderSettings, q querier.Querier, a analytics.Analytics) querier.Handler {
|
||||
return querier.NewHandler(ps, q, a)
|
||||
},
|
||||
|
||||
@@ -28,7 +28,9 @@ import (
|
||||
eequerier "github.com/SigNoz/signoz/ee/querier"
|
||||
enterpriseapp "github.com/SigNoz/signoz/ee/query-service/app"
|
||||
eerules "github.com/SigNoz/signoz/ee/query-service/rules"
|
||||
"github.com/SigNoz/signoz/ee/statementbuilder/auditstatementbuilder"
|
||||
"github.com/SigNoz/signoz/ee/subscription/httpsubscription"
|
||||
"github.com/SigNoz/signoz/ee/telemetrymetadata/audittelemetrymetadata"
|
||||
enterprisezeus "github.com/SigNoz/signoz/ee/zeus"
|
||||
"github.com/SigNoz/signoz/ee/zeus/httpzeus"
|
||||
"github.com/SigNoz/signoz/pkg/alertmanager"
|
||||
@@ -172,6 +174,13 @@ func runServer(ctx context.Context, config signoz.Config, logger *slog.Logger) e
|
||||
|
||||
return factories, "http"
|
||||
},
|
||||
[]signoz.LogSourceQuery{{
|
||||
Source: telemetrytypes.SourceAudit,
|
||||
Resolver: func(ts telemetrystore.TelemetryStore) telemetrytypes.FieldKeyResolver {
|
||||
return audittelemetrymetadata.New(ts)
|
||||
},
|
||||
StatementBuilderFactory: auditstatementbuilder.NewFactory,
|
||||
}},
|
||||
func(ps factory.ProviderSettings, q querier.Querier, a analytics.Analytics) querier.Handler {
|
||||
communityHandler := querier.NewHandler(ps, q, a)
|
||||
return eequerier.NewHandler(ps, q, communityHandler)
|
||||
|
||||
@@ -6,6 +6,7 @@ import (
|
||||
"log/slog"
|
||||
"strings"
|
||||
|
||||
"github.com/SigNoz/signoz/ee/telemetryschema/audittelemetryschema"
|
||||
"github.com/SigNoz/signoz/pkg/clickhousesql"
|
||||
"github.com/SigNoz/signoz/pkg/errors"
|
||||
"github.com/SigNoz/signoz/pkg/factory"
|
||||
@@ -13,7 +14,6 @@ import (
|
||||
"github.com/SigNoz/signoz/pkg/querybuilder"
|
||||
"github.com/SigNoz/signoz/pkg/statementbuilder"
|
||||
"github.com/SigNoz/signoz/pkg/statementbuilder/resourcefilter"
|
||||
"github.com/SigNoz/signoz/pkg/telemetryschema/audittelemetryschema"
|
||||
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
|
||||
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
|
||||
"github.com/SigNoz/signoz/pkg/valuer"
|
||||
@@ -58,7 +58,7 @@ func NewAuditQueryStatementBuilder(
|
||||
fullTextColumn *telemetrytypes.TelemetryFieldKey,
|
||||
flagger flagger.Flagger,
|
||||
) *auditQueryStatementBuilder {
|
||||
auditSettings := factory.NewScopedProviderSettings(settings, "github.com/SigNoz/signoz/pkg/telemetryschema/audittelemetryschema")
|
||||
auditSettings := factory.NewScopedProviderSettings(settings, "github.com/SigNoz/signoz/ee/telemetryschema/audittelemetryschema")
|
||||
|
||||
resourceFilterStmtBuilder := resourcefilter.New[qbtypes.LogAggregation](
|
||||
settings,
|
||||
@@ -5,10 +5,10 @@ import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/SigNoz/signoz/ee/telemetryschema/audittelemetryschema"
|
||||
"github.com/SigNoz/signoz/pkg/flagger/flaggertest"
|
||||
"github.com/SigNoz/signoz/pkg/instrumentation/instrumentationtest"
|
||||
"github.com/SigNoz/signoz/pkg/querybuilder"
|
||||
"github.com/SigNoz/signoz/pkg/telemetryschema/audittelemetryschema"
|
||||
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
|
||||
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
|
||||
"github.com/SigNoz/signoz/pkg/types/telemetrytypes/telemetrytypestest"
|
||||
333
ee/telemetrymetadata/audittelemetrymetadata/resolver.go
Normal file
333
ee/telemetrymetadata/audittelemetrymetadata/resolver.go
Normal file
@@ -0,0 +1,333 @@
|
||||
package audittelemetrymetadata
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"strings"
|
||||
|
||||
"github.com/huandu/go-sqlbuilder"
|
||||
"golang.org/x/exp/maps"
|
||||
|
||||
"github.com/SigNoz/signoz/ee/telemetryschema/audittelemetryschema"
|
||||
"github.com/SigNoz/signoz/pkg/clickhousesql"
|
||||
"github.com/SigNoz/signoz/pkg/errors"
|
||||
"github.com/SigNoz/signoz/pkg/telemetrymetadata/parser"
|
||||
"github.com/SigNoz/signoz/pkg/telemetrymetadata/resolvercore"
|
||||
"github.com/SigNoz/signoz/pkg/telemetrystore"
|
||||
"github.com/SigNoz/signoz/pkg/types/ctxtypes"
|
||||
"github.com/SigNoz/signoz/pkg/types/instrumentationtypes"
|
||||
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
|
||||
)
|
||||
|
||||
var errFailedToGetAuditKeys = errors.Newf(errors.TypeInternal, errors.CodeInternal, "failed to get audit keys")
|
||||
|
||||
// Resolver answers key and value lookups for the audit log tables. It is
|
||||
// registered against telemetrytypes.SourceAudit; without it the metadata store
|
||||
// carries no audit knowledge.
|
||||
type Resolver struct {
|
||||
store telemetrystore.TelemetryStore
|
||||
}
|
||||
|
||||
var _ telemetrytypes.FieldKeyResolver = (*Resolver)(nil)
|
||||
|
||||
func New(store telemetrystore.TelemetryStore) *Resolver {
|
||||
return &Resolver{store: store}
|
||||
}
|
||||
|
||||
func (r *Resolver) tblStatementToFieldKeys(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", audittelemetryschema.DBName, audittelemetryschema.AuditLogsTableName)
|
||||
statements := []telemetrytypes.ShowCreateTableStatement{}
|
||||
err := r.store.ClickhouseDB().Select(ctx, &statements, query)
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, resolvercore.ErrFailedToGetTblStatement.Error())
|
||||
}
|
||||
|
||||
materialisedKeys, err := parser.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 (r *Resolver) GetKeys(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 := r.tblStatementToFieldKeys(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, audittelemetryschema.DBName + "." + audittelemetryschema.LogAttributeKeysTblName},
|
||||
{telemetrytypes.FieldContextResource, queryResourceTable, audittelemetryschema.DBName + "." + audittelemetryschema.LogResourceKeysTblName},
|
||||
}
|
||||
|
||||
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", resolvercore.PriorityForContext(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", "%"+clickhousesql.LikePattern(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(sqlbuilder.ClickHouse)
|
||||
queries = append(queries, query)
|
||||
allArgs = append(allArgs, args...)
|
||||
}
|
||||
|
||||
var limit int
|
||||
for _, fieldKeySelector := range fieldKeySelectors {
|
||||
limit += fieldKeySelector.Limit
|
||||
}
|
||||
if limit == 0 {
|
||||
limit = 1000
|
||||
}
|
||||
|
||||
keys := []*telemetrytypes.TelemetryFieldKey{}
|
||||
rowCount := 0
|
||||
|
||||
// the log and scope contexts have no keys table; they are served by the
|
||||
// static fields appended below
|
||||
if len(queries) > 0 {
|
||||
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 := r.store.ClickhouseDB().Query(ctx, mainQuery, allArgs...)
|
||||
if err != nil {
|
||||
return nil, false, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, errFailedToGetAuditKeys.Error())
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
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 the matching intrinsic audit fields (same as logs intrinsics: body, severity_text, etc.)
|
||||
for _, field := range maps.Values(audittelemetryschema.IntrinsicFields) {
|
||||
if !resolvercore.StaticFieldMatchesAny(field, fieldKeySelectors) {
|
||||
continue
|
||||
}
|
||||
if _, added := mapOfKeys[field.Name+";"+field.FieldContext.StringValue()+";"+field.FieldDataType.StringValue()]; added {
|
||||
continue
|
||||
}
|
||||
keys = append(keys, &field)
|
||||
}
|
||||
|
||||
return keys, complete, nil
|
||||
}
|
||||
|
||||
func (r *Resolver) GetAllValues(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(audittelemetryschema.DBName + "." + audittelemetryschema.TagAttributesTableName)
|
||||
|
||||
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", "%"+clickhousesql.LikePattern(fieldValueSelector.Value)+"%"))
|
||||
case telemetrytypes.FieldDataTypeNumber:
|
||||
sb.Where(sb.IsNotNull("number_value"))
|
||||
sb.Where(sb.ILike("toString(number_value)", "%"+clickhousesql.LikePattern(fieldValueSelector.Value)+"%"))
|
||||
case telemetrytypes.FieldDataTypeUnspecified:
|
||||
sb.Where(sb.Or(
|
||||
sb.ILike("string_value", "%"+clickhousesql.LikePattern(fieldValueSelector.Value)+"%"),
|
||||
sb.ILike("toString(number_value)", "%"+clickhousesql.LikePattern(fieldValueSelector.Value)+"%"),
|
||||
))
|
||||
}
|
||||
}
|
||||
|
||||
// fetch one extra row to detect whether the result set is complete
|
||||
sb.Limit(limit + 1)
|
||||
|
||||
query, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
|
||||
|
||||
rows, err := r.store.ClickhouseDB().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
|
||||
}
|
||||
32
pkg/querier/log_source.go
Normal file
32
pkg/querier/log_source.go
Normal file
@@ -0,0 +1,32 @@
|
||||
package querier
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/SigNoz/signoz/pkg/errors"
|
||||
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
|
||||
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
|
||||
"github.com/SigNoz/signoz/pkg/valuer"
|
||||
)
|
||||
|
||||
// logStmtBuilderFor returns the statement builder for a log query's source. A
|
||||
// source this build registered no builder for resolves to one that reports it
|
||||
// unavailable, so the query fails with that instead of silently reading the
|
||||
// plain logs tables.
|
||||
func (q *querier) logStmtBuilderFor(source telemetrytypes.Source) qbtypes.StatementBuilder[qbtypes.LogAggregation] {
|
||||
if source == telemetrytypes.SourceUnspecified {
|
||||
return q.logStmtBuilder
|
||||
}
|
||||
if builder, ok := q.sourceLogStmtBuilders[source]; ok {
|
||||
return builder
|
||||
}
|
||||
return unavailableSourceStmtBuilder{source: source}
|
||||
}
|
||||
|
||||
type unavailableSourceStmtBuilder struct {
|
||||
source telemetrytypes.Source
|
||||
}
|
||||
|
||||
func (b unavailableSourceStmtBuilder) Build(_ context.Context, _ valuer.UUID, _, _ uint64, _ qbtypes.RequestType, _ qbtypes.QueryBuilderQuery[qbtypes.LogAggregation], _ map[string]qbtypes.VariableItem) (*qbtypes.Statement, error) {
|
||||
return nil, errors.Newf(errors.TypeInvalidInput, errors.CodeUnsupported, "log source %q is not available in this build", b.source.StringValue())
|
||||
}
|
||||
@@ -54,7 +54,7 @@ type querier struct {
|
||||
traceStmtBuilder qbtypes.StatementBuilder[qbtypes.TraceAggregation]
|
||||
aiTraceStmtBuilder qbtypes.StatementBuilder[qbtypes.TraceAggregation]
|
||||
logStmtBuilder qbtypes.StatementBuilder[qbtypes.LogAggregation]
|
||||
auditStmtBuilder qbtypes.StatementBuilder[qbtypes.LogAggregation]
|
||||
sourceLogStmtBuilders map[telemetrytypes.Source]qbtypes.StatementBuilder[qbtypes.LogAggregation]
|
||||
metricStmtBuilder qbtypes.StatementBuilder[qbtypes.MetricAggregation]
|
||||
meterStmtBuilder qbtypes.StatementBuilder[qbtypes.MetricAggregation]
|
||||
traceOperatorStmtBuilder qbtypes.TraceOperatorStatementBuilder
|
||||
@@ -74,7 +74,7 @@ func New(
|
||||
traceStmtBuilder qbtypes.StatementBuilder[qbtypes.TraceAggregation],
|
||||
aiTraceStmtBuilder qbtypes.StatementBuilder[qbtypes.TraceAggregation],
|
||||
logStmtBuilder qbtypes.StatementBuilder[qbtypes.LogAggregation],
|
||||
auditStmtBuilder qbtypes.StatementBuilder[qbtypes.LogAggregation],
|
||||
sourceLogStmtBuilders map[telemetrytypes.Source]qbtypes.StatementBuilder[qbtypes.LogAggregation],
|
||||
metricStmtBuilder qbtypes.StatementBuilder[qbtypes.MetricAggregation],
|
||||
meterStmtBuilder qbtypes.StatementBuilder[qbtypes.MetricAggregation],
|
||||
traceOperatorStmtBuilder qbtypes.TraceOperatorStatementBuilder,
|
||||
@@ -96,7 +96,7 @@ func New(
|
||||
traceStmtBuilder: traceStmtBuilder,
|
||||
aiTraceStmtBuilder: aiTraceStmtBuilder,
|
||||
logStmtBuilder: logStmtBuilder,
|
||||
auditStmtBuilder: auditStmtBuilder,
|
||||
sourceLogStmtBuilders: sourceLogStmtBuilders,
|
||||
metricStmtBuilder: metricStmtBuilder,
|
||||
meterStmtBuilder: meterStmtBuilder,
|
||||
traceOperatorStmtBuilder: traceOperatorStmtBuilder,
|
||||
@@ -266,10 +266,7 @@ func (q *querier) buildQueries(
|
||||
case qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]:
|
||||
spec.ShiftBy = extractShiftFromBuilderQuery(spec)
|
||||
timeRange := adjustTimeRangeForShift(spec, qbtypes.TimeRange{From: req.Start, To: req.End}, req.RequestType)
|
||||
stmtBuilder := q.logStmtBuilder
|
||||
if spec.Source == telemetrytypes.SourceAudit {
|
||||
stmtBuilder = q.auditStmtBuilder
|
||||
}
|
||||
stmtBuilder := q.logStmtBuilderFor(spec.Source)
|
||||
bq := newBuilderQuery(q.logger, q.telemetryStore, orgID, stmtBuilder, query.Type, spec, timeRange, req.RequestType, tmplVars, q.builderConfig)
|
||||
queries[spec.Name] = bq
|
||||
steps[spec.Name] = spec.StepInterval
|
||||
@@ -602,10 +599,7 @@ func (q *querier) QueryRawStream(ctx context.Context, orgID valuer.UUID, req *qb
|
||||
case <-tick:
|
||||
// timestamp end is not specified here
|
||||
timeRange := adjustTimeRangeForShift(spec, qbtypes.TimeRange{From: tsStart}, req.RequestType)
|
||||
liveTailStmtBuilder := q.logStmtBuilder
|
||||
if spec.Source == telemetrytypes.SourceAudit {
|
||||
liveTailStmtBuilder = q.auditStmtBuilder
|
||||
}
|
||||
liveTailStmtBuilder := q.logStmtBuilderFor(spec.Source)
|
||||
bq := newBuilderQuery(q.logger, q.telemetryStore, orgID, liveTailStmtBuilder, query.Type, spec, timeRange, req.RequestType, map[string]qbtypes.VariableItem{
|
||||
"id": {
|
||||
Value: updatedLogID,
|
||||
@@ -937,10 +931,7 @@ func (q *querier) createRangedQuery(_ valuer.UUID, originalQuery qbtypes.Query,
|
||||
specCopy := qt.spec.Copy()
|
||||
specCopy.ShiftBy = extractShiftFromBuilderQuery(specCopy)
|
||||
adjustedTimeRange := adjustTimeRangeForShift(specCopy, timeRange, qt.kind)
|
||||
shiftStmtBuilder := q.logStmtBuilder
|
||||
if qt.spec.Source == telemetrytypes.SourceAudit {
|
||||
shiftStmtBuilder = q.auditStmtBuilder
|
||||
}
|
||||
shiftStmtBuilder := q.logStmtBuilderFor(qt.spec.Source)
|
||||
return newBuilderQuery(q.logger, q.telemetryStore, qt.orgID, shiftStmtBuilder, qt.queryType, specCopy, adjustedTimeRange, qt.kind, qt.variables, q.builderConfig)
|
||||
|
||||
case *builderQuery[qbtypes.MetricAggregation]:
|
||||
|
||||
@@ -22,7 +22,7 @@ func NewFactory(
|
||||
traceStmtBuilder qbtypes.StatementBuilder[qbtypes.TraceAggregation],
|
||||
aiTraceStmtBuilder qbtypes.StatementBuilder[qbtypes.TraceAggregation],
|
||||
logStmtBuilder qbtypes.StatementBuilder[qbtypes.LogAggregation],
|
||||
auditStmtBuilder qbtypes.StatementBuilder[qbtypes.LogAggregation],
|
||||
sourceLogStmtBuilders map[telemetrytypes.Source]qbtypes.StatementBuilder[qbtypes.LogAggregation],
|
||||
metricStmtBuilder qbtypes.StatementBuilder[qbtypes.MetricAggregation],
|
||||
meterStmtBuilder qbtypes.StatementBuilder[qbtypes.MetricAggregation],
|
||||
traceOperatorStmtBuilder qbtypes.TraceOperatorStatementBuilder,
|
||||
@@ -44,7 +44,7 @@ func NewFactory(
|
||||
traceStmtBuilder,
|
||||
aiTraceStmtBuilder,
|
||||
logStmtBuilder,
|
||||
auditStmtBuilder,
|
||||
sourceLogStmtBuilders,
|
||||
metricStmtBuilder,
|
||||
meterStmtBuilder,
|
||||
traceOperatorStmtBuilder,
|
||||
|
||||
@@ -20,7 +20,6 @@ import (
|
||||
"github.com/SigNoz/signoz/pkg/sqlstore/sqlstoretest"
|
||||
"github.com/SigNoz/signoz/pkg/statementbuilder"
|
||||
"github.com/SigNoz/signoz/pkg/statementbuilder/aistatementbuilder"
|
||||
"github.com/SigNoz/signoz/pkg/statementbuilder/auditstatementbuilder"
|
||||
"github.com/SigNoz/signoz/pkg/statementbuilder/logsstatementbuilder"
|
||||
"github.com/SigNoz/signoz/pkg/statementbuilder/meterstatementbuilder"
|
||||
"github.com/SigNoz/signoz/pkg/statementbuilder/metricsstatementbuilder"
|
||||
@@ -113,7 +112,7 @@ func NewTestManager(t *testing.T, testOpts *TestManagerOptions) *Manager {
|
||||
}
|
||||
|
||||
// Create querier with test values
|
||||
metadataStore := telemetrymetadata.NewTelemetryMetaStore(providerSettings, telemetryStore, flagger)
|
||||
metadataStore := telemetrymetadata.NewTelemetryMetaStore(providerSettings, telemetryStore, flagger, nil)
|
||||
cfg := statementbuilder.Config{}
|
||||
ctx := context.Background()
|
||||
traceStmtBuilder, err := tracesstatementbuilder.NewFactory(telemetryStore, metadataStore, flagger).New(ctx, providerSettings, cfg)
|
||||
@@ -124,14 +123,12 @@ func NewTestManager(t *testing.T, testOpts *TestManagerOptions) *Manager {
|
||||
require.NoError(t, err)
|
||||
logStmtBuilder, err := logsstatementbuilder.NewFactory(telemetryStore, metadataStore, flagger).New(ctx, providerSettings, cfg)
|
||||
require.NoError(t, err)
|
||||
auditStmtBuilder, err := auditstatementbuilder.NewFactory(metadataStore, flagger).New(ctx, providerSettings, cfg)
|
||||
require.NoError(t, err)
|
||||
metricStmtBuilder, err := metricsstatementbuilder.NewFactory(metadataStore, flagger).New(ctx, providerSettings, cfg)
|
||||
require.NoError(t, err)
|
||||
meterStmtBuilder, err := meterstatementbuilder.NewFactory(metadataStore, flagger).New(ctx, providerSettings, cfg)
|
||||
require.NoError(t, err)
|
||||
bucketCache := querier.NewBucketCache(providerSettings, cache, 0, 0)
|
||||
providerFactory := signozquerier.NewFactory(telemetryStore, prometheus, metadataStore, traceStmtBuilder, aiTraceStmtBuilder, logStmtBuilder, auditStmtBuilder, metricStmtBuilder, meterStmtBuilder, traceOperatorStmtBuilder, bucketCache, flagger)
|
||||
providerFactory := signozquerier.NewFactory(telemetryStore, prometheus, metadataStore, traceStmtBuilder, aiTraceStmtBuilder, logStmtBuilder, nil, metricStmtBuilder, meterStmtBuilder, traceOperatorStmtBuilder, bucketCache, flagger)
|
||||
mockQuerier, err := providerFactory.New(context.Background(), providerSettings, querier.Config{})
|
||||
require.NoError(t, err)
|
||||
|
||||
|
||||
20
pkg/signoz/log_source_query.go
Normal file
20
pkg/signoz/log_source_query.go
Normal file
@@ -0,0 +1,20 @@
|
||||
package signoz
|
||||
|
||||
import (
|
||||
"github.com/SigNoz/signoz/pkg/factory"
|
||||
"github.com/SigNoz/signoz/pkg/flagger"
|
||||
"github.com/SigNoz/signoz/pkg/statementbuilder"
|
||||
"github.com/SigNoz/signoz/pkg/telemetrystore"
|
||||
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
|
||||
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
|
||||
)
|
||||
|
||||
// LogSourceQuery supplies the query stack for one source within the logs
|
||||
// signal. The resolver is built before the metadata store and the statement
|
||||
// builder after it, so they arrive as two functions rather than two values.
|
||||
// Resolver may be nil for a source that serves no key or value suggestions.
|
||||
type LogSourceQuery struct {
|
||||
Source telemetrytypes.Source
|
||||
Resolver func(telemetrystore.TelemetryStore) telemetrytypes.FieldKeyResolver
|
||||
StatementBuilderFactory func(telemetrytypes.MetadataStore, flagger.Flagger) factory.ProviderFactory[qbtypes.StatementBuilder[qbtypes.LogAggregation], statementbuilder.Config]
|
||||
}
|
||||
@@ -324,9 +324,9 @@ func NewStatsReporterProviderFactories(aggregator statsreporter.Aggregator, orgG
|
||||
)
|
||||
}
|
||||
|
||||
func NewQuerierProviderFactories(telemetryStore telemetrystore.TelemetryStore, prometheus prometheus.Prometheus, metadataStore telemetrytypes.MetadataStore, traceStmtBuilder qbtypes.StatementBuilder[qbtypes.TraceAggregation], aiTraceStmtBuilder qbtypes.StatementBuilder[qbtypes.TraceAggregation], logStmtBuilder qbtypes.StatementBuilder[qbtypes.LogAggregation], auditStmtBuilder qbtypes.StatementBuilder[qbtypes.LogAggregation], metricStmtBuilder qbtypes.StatementBuilder[qbtypes.MetricAggregation], meterStmtBuilder qbtypes.StatementBuilder[qbtypes.MetricAggregation], traceOperatorStmtBuilder qbtypes.TraceOperatorStatementBuilder, bucketCache querier.BucketCache, flagger flagger.Flagger) factory.NamedMap[factory.ProviderFactory[querier.Querier, querier.Config]] {
|
||||
func NewQuerierProviderFactories(telemetryStore telemetrystore.TelemetryStore, prometheus prometheus.Prometheus, metadataStore telemetrytypes.MetadataStore, traceStmtBuilder qbtypes.StatementBuilder[qbtypes.TraceAggregation], aiTraceStmtBuilder qbtypes.StatementBuilder[qbtypes.TraceAggregation], logStmtBuilder qbtypes.StatementBuilder[qbtypes.LogAggregation], sourceLogStmtBuilders map[telemetrytypes.Source]qbtypes.StatementBuilder[qbtypes.LogAggregation], metricStmtBuilder qbtypes.StatementBuilder[qbtypes.MetricAggregation], meterStmtBuilder qbtypes.StatementBuilder[qbtypes.MetricAggregation], traceOperatorStmtBuilder qbtypes.TraceOperatorStatementBuilder, bucketCache querier.BucketCache, flagger flagger.Flagger) factory.NamedMap[factory.ProviderFactory[querier.Querier, querier.Config]] {
|
||||
return factory.MustNewNamedMap(
|
||||
signozquerier.NewFactory(telemetryStore, prometheus, metadataStore, traceStmtBuilder, aiTraceStmtBuilder, logStmtBuilder, auditStmtBuilder, metricStmtBuilder, meterStmtBuilder, traceOperatorStmtBuilder, bucketCache, flagger),
|
||||
signozquerier.NewFactory(telemetryStore, prometheus, metadataStore, traceStmtBuilder, aiTraceStmtBuilder, logStmtBuilder, sourceLogStmtBuilders, metricStmtBuilder, meterStmtBuilder, traceOperatorStmtBuilder, bucketCache, flagger),
|
||||
)
|
||||
}
|
||||
|
||||
|
||||
@@ -50,7 +50,6 @@ import (
|
||||
"github.com/SigNoz/signoz/pkg/sqlschema"
|
||||
"github.com/SigNoz/signoz/pkg/sqlstore"
|
||||
"github.com/SigNoz/signoz/pkg/statementbuilder/aistatementbuilder"
|
||||
"github.com/SigNoz/signoz/pkg/statementbuilder/auditstatementbuilder"
|
||||
"github.com/SigNoz/signoz/pkg/statementbuilder/logsstatementbuilder"
|
||||
"github.com/SigNoz/signoz/pkg/statementbuilder/meterstatementbuilder"
|
||||
"github.com/SigNoz/signoz/pkg/statementbuilder/metricsstatementbuilder"
|
||||
@@ -103,7 +102,8 @@ type SigNoz struct {
|
||||
|
||||
// newQueryStack assembles the query stack once and returns, in order: the shared
|
||||
// telemetry metadata store (reused elsewhere in signoz.New), the per-signal
|
||||
// statement builders (trace, ai-trace, log, audit, metric, meter, trace-operator),
|
||||
// statement builders (trace, ai-trace, log, per-source log, metric, meter,
|
||||
// trace-operator),
|
||||
// and the bucket cache. It is the only place that imports the concrete
|
||||
// statement-builder sub-packages.
|
||||
func newQueryStack(
|
||||
@@ -113,19 +113,27 @@ func newQueryStack(
|
||||
telemetryStore telemetrystore.TelemetryStore,
|
||||
cache cache.Cache,
|
||||
fl flagger.Flagger,
|
||||
logSources []LogSourceQuery,
|
||||
) (
|
||||
telemetrytypes.MetadataStore,
|
||||
qbtypes.StatementBuilder[qbtypes.TraceAggregation],
|
||||
qbtypes.StatementBuilder[qbtypes.TraceAggregation],
|
||||
qbtypes.StatementBuilder[qbtypes.LogAggregation],
|
||||
qbtypes.StatementBuilder[qbtypes.LogAggregation],
|
||||
map[telemetrytypes.Source]qbtypes.StatementBuilder[qbtypes.LogAggregation],
|
||||
qbtypes.StatementBuilder[qbtypes.MetricAggregation],
|
||||
qbtypes.StatementBuilder[qbtypes.MetricAggregation],
|
||||
qbtypes.TraceOperatorStatementBuilder,
|
||||
querier.BucketCache,
|
||||
error,
|
||||
) {
|
||||
metadataStore := telemetrymetadata.NewTelemetryMetaStore(settings, telemetryStore, fl)
|
||||
resolvers := map[telemetrytypes.Source]telemetrytypes.FieldKeyResolver{}
|
||||
for _, logSource := range logSources {
|
||||
if logSource.Resolver == nil {
|
||||
continue
|
||||
}
|
||||
resolvers[logSource.Source] = logSource.Resolver(telemetryStore)
|
||||
}
|
||||
metadataStore := telemetrymetadata.NewTelemetryMetaStore(settings, telemetryStore, fl, resolvers)
|
||||
|
||||
cfg := config.Querier.Config
|
||||
traceStmtBuilder, err := tracesstatementbuilder.NewFactory(telemetryStore, metadataStore, fl).New(ctx, settings, cfg)
|
||||
@@ -144,9 +152,13 @@ func newQueryStack(
|
||||
if err != nil {
|
||||
return nil, nil, nil, nil, nil, nil, nil, nil, nil, err
|
||||
}
|
||||
auditStmtBuilder, err := auditstatementbuilder.NewFactory(metadataStore, fl).New(ctx, settings, cfg)
|
||||
if err != nil {
|
||||
return nil, nil, nil, nil, nil, nil, nil, nil, nil, err
|
||||
sourceLogStmtBuilders := map[telemetrytypes.Source]qbtypes.StatementBuilder[qbtypes.LogAggregation]{}
|
||||
for _, logSource := range logSources {
|
||||
builder, err := logSource.StatementBuilderFactory(metadataStore, fl).New(ctx, settings, cfg)
|
||||
if err != nil {
|
||||
return nil, nil, nil, nil, nil, nil, nil, nil, nil, err
|
||||
}
|
||||
sourceLogStmtBuilders[logSource.Source] = builder
|
||||
}
|
||||
metricStmtBuilder, err := metricsstatementbuilder.NewFactory(metadataStore, fl).New(ctx, settings, cfg)
|
||||
if err != nil {
|
||||
@@ -159,7 +171,7 @@ func newQueryStack(
|
||||
|
||||
bucketCache := querier.NewBucketCache(settings, cache, config.Querier.CacheTTL, config.Querier.FluxInterval)
|
||||
|
||||
return metadataStore, traceStmtBuilder, aiTraceStmtBuilder, logStmtBuilder, auditStmtBuilder, metricStmtBuilder, meterStmtBuilder, traceOperatorStmtBuilder, bucketCache, nil
|
||||
return metadataStore, traceStmtBuilder, aiTraceStmtBuilder, logStmtBuilder, sourceLogStmtBuilders, metricStmtBuilder, meterStmtBuilder, traceOperatorStmtBuilder, bucketCache, nil
|
||||
}
|
||||
|
||||
func New(
|
||||
@@ -182,6 +194,7 @@ func New(
|
||||
gatewayProviderFactory func(licensing.Licensing) factory.ProviderFactory[gateway.Gateway, gateway.Config],
|
||||
auditorProviderFactories func(licensing.Licensing) factory.NamedMap[factory.ProviderFactory[auditor.Auditor, auditor.Config]],
|
||||
meterReporterProviderFactories func(context.Context, factory.ProviderSettings, flagger.Flagger, licensing.Licensing, telemetrystore.TelemetryStore, retention.Getter, organization.Getter, zeus.Zeus) (factory.NamedMap[factory.ProviderFactory[meterreporter.Reporter, meterreporter.Config]], string),
|
||||
logSources []LogSourceQuery,
|
||||
querierHandlerCallback func(factory.ProviderSettings, querier.Querier, analytics.Analytics) querier.Handler,
|
||||
cloudIntegrationCallback func(sqlstore.SQLStore, dashboard.Module, global.Global, zeus.Zeus, gateway.Gateway, licensing.Licensing, serviceaccount.Module, cloudintegration.Config) (cloudintegration.Module, error),
|
||||
metricReductionRuleModuleCallback func(sqlstore.SQLStore, telemetrystore.TelemetryStore, dashboard.Module, queryparser.QueryParser, licensing.Licensing, flagger.Flagger, telemetrytypes.MetadataStore, factory.ProviderSettings, int) metricreductionrule.Module,
|
||||
@@ -323,7 +336,7 @@ func New(
|
||||
|
||||
// Assemble the query stack (metadata store, statement builders, bucket cache) once,
|
||||
// and reuse the single metadata store everywhere downstream.
|
||||
telemetryMetadataStore, traceStmtBuilder, aiTraceStmtBuilder, logStmtBuilder, auditStmtBuilder, metricStmtBuilder, meterStmtBuilder, traceOperatorStmtBuilder, bucketCache, err := newQueryStack(ctx, providerSettings, config, telemetrystore, cache, flagger)
|
||||
telemetryMetadataStore, traceStmtBuilder, aiTraceStmtBuilder, logStmtBuilder, sourceLogStmtBuilders, metricStmtBuilder, meterStmtBuilder, traceOperatorStmtBuilder, bucketCache, err := newQueryStack(ctx, providerSettings, config, telemetrystore, cache, flagger, logSources)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -333,7 +346,7 @@ func New(
|
||||
ctx,
|
||||
providerSettings,
|
||||
config.Querier,
|
||||
NewQuerierProviderFactories(telemetrystore, prometheus, telemetryMetadataStore, traceStmtBuilder, aiTraceStmtBuilder, logStmtBuilder, auditStmtBuilder, metricStmtBuilder, meterStmtBuilder, traceOperatorStmtBuilder, bucketCache, flagger),
|
||||
NewQuerierProviderFactories(telemetrystore, prometheus, telemetryMetadataStore, traceStmtBuilder, aiTraceStmtBuilder, logStmtBuilder, sourceLogStmtBuilders, metricStmtBuilder, meterStmtBuilder, traceOperatorStmtBuilder, bucketCache, flagger),
|
||||
config.Querier.Provider(),
|
||||
)
|
||||
if err != nil {
|
||||
|
||||
@@ -0,0 +1,32 @@
|
||||
package noopauditstatementbuilder
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/SigNoz/signoz/pkg/errors"
|
||||
"github.com/SigNoz/signoz/pkg/factory"
|
||||
"github.com/SigNoz/signoz/pkg/flagger"
|
||||
"github.com/SigNoz/signoz/pkg/statementbuilder"
|
||||
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
|
||||
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
|
||||
"github.com/SigNoz/signoz/pkg/valuer"
|
||||
)
|
||||
|
||||
// NewFactory builds the audit statement builder for builds that do not ship
|
||||
// audit logging. Audit records are written by an enterprise-only auditor, so
|
||||
// there is nothing for a query to read and it is refused rather than served
|
||||
// from the plain logs tables.
|
||||
func NewFactory(_ telemetrytypes.MetadataStore, _ flagger.Flagger) factory.ProviderFactory[qbtypes.StatementBuilder[qbtypes.LogAggregation], statementbuilder.Config] {
|
||||
return factory.NewProviderFactory(
|
||||
factory.MustNewName("noop"),
|
||||
func(_ context.Context, _ factory.ProviderSettings, _ statementbuilder.Config) (qbtypes.StatementBuilder[qbtypes.LogAggregation], error) {
|
||||
return statementBuilder{}, nil
|
||||
},
|
||||
)
|
||||
}
|
||||
|
||||
type statementBuilder struct{}
|
||||
|
||||
func (statementBuilder) Build(_ context.Context, _ valuer.UUID, _, _ uint64, _ qbtypes.RequestType, _ qbtypes.QueryBuilderQuery[qbtypes.LogAggregation], _ map[string]qbtypes.VariableItem) (*qbtypes.Statement, error) {
|
||||
return nil, errors.Newf(errors.TypeInvalidInput, errors.CodeUnsupported, "audit logs are not available in this build")
|
||||
}
|
||||
@@ -15,45 +15,6 @@ import (
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
// The flagger provider registration is process-global and keyed by provider
|
||||
// name, so each flagger must be used before the next one is created.
|
||||
func TestFamilyValueNames(t *testing.T) {
|
||||
selector := &telemetrytypes.FieldValueSelector{
|
||||
FieldKeySelector: &telemetrytypes.FieldKeySelector{Name: "deployment.environment"},
|
||||
}
|
||||
|
||||
off := &telemetryMetaStore{fl: flaggertest.WithBooleanFlags(t, map[string]bool{})}
|
||||
assert.Equal(t,
|
||||
[]string{"deployment.environment"},
|
||||
off.familyValueNames(context.Background(), valuer.UUID{}, telemetrytypes.SignalLogs, selector),
|
||||
"the flag default keeps values literal")
|
||||
|
||||
on := &telemetryMetaStore{fl: flaggertest.WithBooleanFlags(t, map[string]bool{
|
||||
flagger.FeatureResolveSemconvFamilies.String(): true,
|
||||
})}
|
||||
assert.Equal(t,
|
||||
[]string{"deployment.environment.name", "deployment.environment"},
|
||||
on.familyValueNames(context.Background(), valuer.UUID{}, telemetrytypes.SignalLogs, selector),
|
||||
"values for one spelling must cover the whole family")
|
||||
assert.Equal(t,
|
||||
[]string{"deployment.environment.name", "deployment.environment"},
|
||||
on.familyValueNames(context.Background(), valuer.UUID{}, telemetrytypes.SignalMetrics, selector),
|
||||
"metric values cover the family members")
|
||||
spanMetric := &telemetrytypes.FieldValueSelector{
|
||||
FieldKeySelector: &telemetrytypes.FieldKeySelector{
|
||||
Name: "deployment.environment",
|
||||
MetricContext: &telemetrytypes.MetricContext{MetricName: "signoz_calls_total"},
|
||||
},
|
||||
}
|
||||
assert.Equal(t,
|
||||
[]string{
|
||||
"deployment.environment.name", "resource_deployment.environment.name",
|
||||
"deployment.environment", "resource_deployment.environment",
|
||||
},
|
||||
on.familyValueNames(context.Background(), valuer.UUID{}, telemetrytypes.SignalMetrics, spanMetric),
|
||||
"span-metrics values cover the resource_ spellings too")
|
||||
}
|
||||
|
||||
// A family condition on the related values table follows the shared guard
|
||||
// rule: the operator applies to the current-first merge, a positive operator
|
||||
// takes the presence guard, and a negative operator keeps the keyless rows.
|
||||
|
||||
356
pkg/telemetrymetadata/logs.go
Normal file
356
pkg/telemetrymetadata/logs.go
Normal file
@@ -0,0 +1,356 @@
|
||||
package telemetrymetadata
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"strings"
|
||||
|
||||
"github.com/huandu/go-sqlbuilder"
|
||||
"golang.org/x/exp/maps"
|
||||
|
||||
"github.com/SigNoz/signoz/pkg/clickhousesql"
|
||||
"github.com/SigNoz/signoz/pkg/errors"
|
||||
"github.com/SigNoz/signoz/pkg/flagger"
|
||||
"github.com/SigNoz/signoz/pkg/telemetrymetadata/parser"
|
||||
"github.com/SigNoz/signoz/pkg/telemetrymetadata/resolvercore"
|
||||
"github.com/SigNoz/signoz/pkg/telemetryschema/logstelemetryschema"
|
||||
"github.com/SigNoz/signoz/pkg/types/ctxtypes"
|
||||
"github.com/SigNoz/signoz/pkg/types/featuretypes"
|
||||
"github.com/SigNoz/signoz/pkg/types/instrumentationtypes"
|
||||
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
|
||||
"github.com/SigNoz/signoz/pkg/valuer"
|
||||
)
|
||||
|
||||
// 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.ClickhouseDB().Select(ctx, &statements, query)
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, resolvercore.ErrFailedToGetTblStatement.Error())
|
||||
}
|
||||
|
||||
materialisedKeys, err := parser.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, orgID valuer.UUID, 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
|
||||
queryBodyTable = queryBodyTable && t.fl.BooleanOrEmpty(ctx, flagger.FeatureUseJSONBody, featuretypes.NewFlaggerEvaluationContext(orgID))
|
||||
|
||||
// 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", resolvercore.PriorityForContext(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", "%"+clickhousesql.LikePattern(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(sqlbuilder.ClickHouse)
|
||||
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", resolvercore.PriorityForContext(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", "%"+clickhousesql.LikePattern(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(sqlbuilder.ClickHouse)
|
||||
queries = append(queries, query)
|
||||
allArgs = append(allArgs, args...)
|
||||
}
|
||||
|
||||
var limit int
|
||||
for _, fieldKeySelector := range fieldKeySelectors {
|
||||
limit += fieldKeySelector.Limit
|
||||
}
|
||||
if limit == 0 {
|
||||
limit = 1000
|
||||
}
|
||||
|
||||
keys := []*telemetrytypes.TelemetryFieldKey{}
|
||||
parentTypes := make(map[string][]telemetrytypes.FieldDataType)
|
||||
rowCount := 0
|
||||
|
||||
// the log and scope contexts have no keys table; they are served by the
|
||||
// static fields appended below
|
||||
if len(queries) > 0 {
|
||||
// Combine queries with UNION ALL
|
||||
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.ClickhouseDB().Query(ctx, mainQuery, allArgs...)
|
||||
if err != nil {
|
||||
return nil, false, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, ErrFailedToGetLogsKeys.Error())
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
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
|
||||
|
||||
// Add the matching intrinsic columns. These don't count towards the limit
|
||||
for _, field := range maps.Values(logstelemetryschema.IntrinsicFields) {
|
||||
if !resolvercore.StaticFieldMatchesAny(field, fieldKeySelectors) {
|
||||
continue
|
||||
}
|
||||
if _, added := mapOfKeys[field.Name+";"+field.FieldContext.StringValue()+";"+field.FieldDataType.StringValue()]; added {
|
||||
continue
|
||||
}
|
||||
keys = append(keys, &field)
|
||||
}
|
||||
|
||||
// enrich body keys with promoted paths, indexes, and JSON access plans
|
||||
if t.fl.BooleanOrEmpty(ctx, flagger.FeatureUseJSONBody, featuretypes.NewFlaggerEvaluationContext(orgID)) {
|
||||
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) getLogFieldValues(ctx context.Context, orgID valuer.UUID, 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",
|
||||
})
|
||||
|
||||
knownBool := resolvercore.IsKnownBoolField(fieldValueSelector, logstelemetryschema.IntrinsicFields)
|
||||
names := resolvercore.FamilyValueNames(ctx, orgID, t.fl, telemetrytypes.SignalLogs, fieldValueSelector)
|
||||
// unix_milli is the hour the log was ingested, not the log's own timestamp
|
||||
return resolvercore.TagTableValues(ctx, t.telemetrystore, t.logsDBName+"."+t.logsFieldsTblName, fieldValueSelector, names, knownBool)
|
||||
}
|
||||
File diff suppressed because it is too large
Load Diff
@@ -38,6 +38,7 @@ func TestGetFirstSeenFromMetricMetadata(t *testing.T) {
|
||||
instrumentationtest.New().ToProviderSettings(),
|
||||
mockTelemetryStore,
|
||||
flaggertest.New(t),
|
||||
nil,
|
||||
)
|
||||
|
||||
lookupKeys := []telemetrytypes.MetricMetadataLookupKey{
|
||||
|
||||
215
pkg/telemetrymetadata/meter.go
Normal file
215
pkg/telemetrymetadata/meter.go
Normal file
@@ -0,0 +1,215 @@
|
||||
package telemetrymetadata
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log/slog"
|
||||
|
||||
"github.com/huandu/go-sqlbuilder"
|
||||
|
||||
"github.com/SigNoz/signoz/pkg/clickhousesql"
|
||||
"github.com/SigNoz/signoz/pkg/errors"
|
||||
"github.com/SigNoz/signoz/pkg/types/ctxtypes"
|
||||
"github.com/SigNoz/signoz/pkg/types/instrumentationtypes"
|
||||
"github.com/SigNoz/signoz/pkg/types/metrictypes"
|
||||
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
|
||||
)
|
||||
|
||||
// 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", clickhousesql.LikePattern(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(sqlbuilder.ClickHouse)
|
||||
|
||||
rows, err := t.telemetrystore.ClickhouseDB().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())
|
||||
}
|
||||
// meter labels are stored as strings in the labels JSON and have no
|
||||
// attribute context, so only the data type is known
|
||||
keys = append(keys, &telemetrytypes.TelemetryFieldKey{
|
||||
Name: name,
|
||||
Signal: telemetrytypes.SignalMetrics,
|
||||
FieldDataType: telemetrytypes.FieldDataTypeString,
|
||||
})
|
||||
}
|
||||
|
||||
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
|
||||
|
||||
}
|
||||
|
||||
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", clickhousesql.LikePattern(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(sqlbuilder.ClickHouse)
|
||||
rows, err := t.telemetrystore.ClickhouseDB().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 (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",
|
||||
"argMax(is_monotonic, unix_milli) 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(sqlbuilder.ClickHouse)
|
||||
|
||||
t.logger.DebugContext(ctx, "fetching meter metrics temporality", slog.String("query", query), slog.Any("args", args))
|
||||
|
||||
rows, err := t.telemetrystore.ClickhouseDB().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")
|
||||
}
|
||||
metricType = resolveMetricType(metricType, isMonotonic, temporality)
|
||||
temporalities[metricName] = temporality
|
||||
types[metricName] = metricType
|
||||
}
|
||||
|
||||
return temporalities, types, nil
|
||||
}
|
||||
651
pkg/telemetrymetadata/metrics.go
Normal file
651
pkg/telemetrymetadata/metrics.go
Normal file
@@ -0,0 +1,651 @@
|
||||
package telemetrymetadata
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log/slog"
|
||||
|
||||
"github.com/huandu/go-sqlbuilder"
|
||||
|
||||
"github.com/SigNoz/signoz/pkg/clickhousesql"
|
||||
"github.com/SigNoz/signoz/pkg/errors"
|
||||
"github.com/SigNoz/signoz/pkg/flagger"
|
||||
"github.com/SigNoz/signoz/pkg/querybuilder"
|
||||
"github.com/SigNoz/signoz/pkg/telemetrymetadata/resolvercore"
|
||||
"github.com/SigNoz/signoz/pkg/telemetryschema/metricstelemetryschema"
|
||||
"github.com/SigNoz/signoz/pkg/types/ctxtypes"
|
||||
"github.com/SigNoz/signoz/pkg/types/featuretypes"
|
||||
"github.com/SigNoz/signoz/pkg/types/instrumentationtypes"
|
||||
"github.com/SigNoz/signoz/pkg/types/metrictypes"
|
||||
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
|
||||
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
|
||||
"github.com/SigNoz/signoz/pkg/valuer"
|
||||
)
|
||||
|
||||
// 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://clickhouse.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
|
||||
|
||||
// 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", "%"+clickhousesql.LikePattern(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", clickhousesql.LikePattern(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(sqlbuilder.ClickHouse)
|
||||
|
||||
rows, err := t.telemetrystore.ClickhouseDB().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
|
||||
}
|
||||
|
||||
// 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 != "" {
|
||||
if names := resolvercore.FamilyValueNames(ctx, orgID, t.fl, telemetrytypes.SignalMetrics, fieldValueSelector); len(names) > 1 {
|
||||
sb.Where(sb.In("attr_name", sqlbuilder.List(names)))
|
||||
} else {
|
||||
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 != "" {
|
||||
if metricNames := querybuilder.FamilyMetricNames(ctx, orgID, t.fl, fieldValueSelector.MetricContext.MetricName); len(metricNames) > 1 {
|
||||
sb.Where(sb.In("metric_name", sqlbuilder.List(metricNames)))
|
||||
} else {
|
||||
sb.Where(sb.E("metric_name", fieldValueSelector.MetricContext.MetricName))
|
||||
}
|
||||
}
|
||||
if fieldValueSelector.MetricContext != nil && fieldValueSelector.MetricContext.MetricNamespace != "" {
|
||||
sb.Where(sb.Like("metric_name", clickhousesql.LikePattern(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", "%"+clickhousesql.LikePattern(fieldValueSelector.Value)+"%"))
|
||||
}
|
||||
}
|
||||
// query one extra to check if we hit the limit
|
||||
sb.Limit(remainingLimit + 1)
|
||||
|
||||
query, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
|
||||
|
||||
rows, err := t.telemetrystore.ClickhouseDB().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 := metricstelemetryschema.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
|
||||
}{
|
||||
{metricstelemetryschema.TimeseriesV41weekTableName, nil},
|
||||
}
|
||||
if reductionEnabled {
|
||||
sources = append(sources,
|
||||
struct {
|
||||
tableName string
|
||||
extraConds []string
|
||||
}{metricstelemetryschema.TimeseriesV4BufferTableName, []string{"is_reduced = false"}},
|
||||
struct {
|
||||
tableName string
|
||||
extraConds []string
|
||||
}{metricstelemetryschema.TimeseriesV4ReducedTableName, 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", clickhousesql.LikePattern(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, "%"+clickhousesql.LikePattern(fieldValueSelector.Value)+"%"))
|
||||
}
|
||||
}
|
||||
|
||||
for _, cond := range extraConds {
|
||||
sb.Where(cond)
|
||||
}
|
||||
|
||||
sb.GroupBy(sqlbuilder.Escape(key.Name))
|
||||
sb.Limit(limit + 1)
|
||||
|
||||
query, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
|
||||
rows, err := t.telemetrystore.ClickhouseDB().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) 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, _ := metricstelemetryschema.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, metricstelemetryschema.TimeseriesV4BufferTableName, 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, metricstelemetryschema.TimeseriesV4ReducedTableName, 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", "argMax(is_monotonic, unix_milli) 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(sqlbuilder.ClickHouse)
|
||||
t.logger.DebugContext(ctx, "fetching metric temporality", slog.String("query", query), slog.Any("args", args))
|
||||
|
||||
rows, err := t.telemetrystore.ClickhouseDB().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)
|
||||
}
|
||||
metricType = resolveMetricType(metricType, isMonotonic, temporality)
|
||||
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
|
||||
}
|
||||
|
||||
// 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(sqlbuilder.ClickHouse)
|
||||
|
||||
rows, err := t.telemetrystore.ClickhouseDB().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, metricstelemetryschema.TimeseriesV4TableName, metricNames)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if t.fl.BooleanOrEmpty(ctx, flagger.FeatureEnableMetricsReduction, featuretypes.NewFlaggerEvaluationContext(orgID)) {
|
||||
reducedLastSeen, err := t.fetchLastSeenInfoForTable(ctx, metricstelemetryschema.TimeseriesV4ReducedTableName, 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(sqlbuilder.ClickHouse)
|
||||
|
||||
t.logger.DebugContext(ctx, "fetching metric last seen timestamp", slog.String("query", query), slog.Any("args", args))
|
||||
|
||||
rows, err := t.telemetrystore.ClickhouseDB().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
|
||||
}
|
||||
|
||||
// containsConditions compiles a contains search over the key and its family
|
||||
// spellings, which stand as their own metadata.
|
||||
func (t *telemetryMetaStore) containsConditions(ctx context.Context, q qbtypes.QueryInfo, key *telemetrytypes.TelemetryFieldKey, names []string, value string, sb *sqlbuilder.SelectBuilder) ([]string, error) {
|
||||
// A family forms only over string keys with a signal. The table stores
|
||||
// only string maps, so an unspecified data type is a string here.
|
||||
dataType := key.FieldDataType
|
||||
if dataType == telemetrytypes.FieldDataTypeUnspecified {
|
||||
dataType = telemetrytypes.FieldDataTypeString
|
||||
}
|
||||
fieldKeys := make(map[string][]*telemetrytypes.TelemetryFieldKey, len(names))
|
||||
for _, name := range names {
|
||||
fieldKeys[name] = []*telemetrytypes.TelemetryFieldKey{{
|
||||
Name: name,
|
||||
Signal: key.Signal,
|
||||
FieldContext: key.FieldContext,
|
||||
FieldDataType: dataType,
|
||||
}}
|
||||
}
|
||||
conds, _, err := querybuilder.Conditions(ctx, q, t.storage, key, qbtypes.FilterOperatorContains, value, fieldKeys, false, sb)
|
||||
return conds, err
|
||||
}
|
||||
@@ -1,15 +1,15 @@
|
||||
package telemetrymetadata
|
||||
package parser
|
||||
|
||||
import (
|
||||
"strings"
|
||||
|
||||
"github.com/AfterShip/clickhouse-sql-parser/parser"
|
||||
chparser "github.com/AfterShip/clickhouse-sql-parser/parser"
|
||||
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
|
||||
)
|
||||
|
||||
// TelemetryFieldVisitor is an AST visitor for extracting telemetry fields.
|
||||
type TelemetryFieldVisitor struct {
|
||||
parser.DefaultASTVisitor
|
||||
chparser.DefaultASTVisitor
|
||||
Fields []*telemetrytypes.TelemetryFieldKey
|
||||
}
|
||||
|
||||
@@ -20,14 +20,14 @@ func NewTelemetryFieldVisitor() *TelemetryFieldVisitor {
|
||||
}
|
||||
|
||||
// VisitColumnDef is called when visiting a column definition.
|
||||
func (v *TelemetryFieldVisitor) VisitColumnDef(expr *parser.ColumnDef) error {
|
||||
func (v *TelemetryFieldVisitor) VisitColumnDef(expr *chparser.ColumnDef) error {
|
||||
// Check if this is a materialized column with DEFAULT expression
|
||||
if expr.DefaultExpr == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
// Parse column name to extract context and data type
|
||||
columnName := parser.Format(expr.Name)
|
||||
columnName := chparser.Format(expr.Name)
|
||||
|
||||
// Remove backticks if present
|
||||
columnName = strings.TrimPrefix(columnName, "`")
|
||||
@@ -75,7 +75,7 @@ func (v *TelemetryFieldVisitor) VisitColumnDef(expr *parser.ColumnDef) error {
|
||||
// Extract field name from the DEFAULT expression
|
||||
// The DEFAULT expression should be something like: resources_string['k8s.cluster.name']
|
||||
// We need to extract the key inside the square brackets
|
||||
defaultExprStr := parser.Format(expr.DefaultExpr)
|
||||
defaultExprStr := chparser.Format(expr.DefaultExpr)
|
||||
|
||||
// Look for the pattern: map['key']
|
||||
startIdx := strings.Index(defaultExprStr, "['")
|
||||
@@ -101,7 +101,7 @@ func (v *TelemetryFieldVisitor) VisitColumnDef(expr *parser.ColumnDef) error {
|
||||
|
||||
func ExtractFieldKeysFromTblStatement(statement string) ([]*telemetrytypes.TelemetryFieldKey, error) {
|
||||
// Parse the CREATE TABLE statement using the ClickHouse parser
|
||||
p := parser.NewParser(statement)
|
||||
p := chparser.NewParser(statement)
|
||||
stmts, err := p.ParseStmts()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@@ -113,7 +113,7 @@ func ExtractFieldKeysFromTblStatement(statement string) ([]*telemetrytypes.Telem
|
||||
// Visit each statement
|
||||
for _, stmt := range stmts {
|
||||
// We're looking for CreateTable statements
|
||||
createTable, ok := stmt.(*parser.CreateTable)
|
||||
createTable, ok := stmt.(*chparser.CreateTable)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
@@ -1,4 +1,4 @@
|
||||
package telemetrymetadata
|
||||
package parser
|
||||
|
||||
import (
|
||||
"slices"
|
||||
@@ -1,10 +1,9 @@
|
||||
package telemetrymetadata
|
||||
package resolvercore
|
||||
|
||||
import (
|
||||
"strconv"
|
||||
"strings"
|
||||
|
||||
"github.com/SigNoz/signoz/pkg/telemetryschema/tracestelemetryschema"
|
||||
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
|
||||
)
|
||||
|
||||
@@ -21,27 +20,10 @@ func boolFieldValues(searchText string) *telemetrytypes.TelemetryFieldValues {
|
||||
return values
|
||||
}
|
||||
|
||||
// spanSearchScopeFieldValues is the suggestion set for a search-scope selector
|
||||
// (isRoot, isEntryPoint), which only filters with true. ok is false for any
|
||||
// other name.
|
||||
func spanSearchScopeFieldValues(name, searchText string) (*telemetrytypes.TelemetryFieldValues, bool) {
|
||||
for scopeName := range tracestelemetryschema.SpanSearchScopeFields {
|
||||
if !strings.EqualFold(scopeName, name) {
|
||||
continue
|
||||
}
|
||||
values := &telemetrytypes.TelemetryFieldValues{}
|
||||
if needle := strings.ToLower(searchText); needle == "" || strings.Contains("true", needle) {
|
||||
values.BoolValues = []bool{true}
|
||||
}
|
||||
return values, true
|
||||
}
|
||||
return nil, false
|
||||
}
|
||||
|
||||
// isKnownBoolField is true when the caller asked for the bool data type, or
|
||||
// IsKnownBoolField is true when the caller asked for the bool data type, or
|
||||
// when the name is one of the signal's static bool fields and the requested
|
||||
// context does not rule that static field out.
|
||||
func isKnownBoolField(selector *telemetrytypes.FieldValueSelector, staticFields ...map[string]telemetrytypes.TelemetryFieldKey) bool {
|
||||
func IsKnownBoolField(selector *telemetrytypes.FieldValueSelector, staticFields ...map[string]telemetrytypes.TelemetryFieldKey) bool {
|
||||
if selector.FieldDataType == telemetrytypes.FieldDataTypeBool {
|
||||
return true
|
||||
}
|
||||
29
pkg/telemetrymetadata/resolvercore/family_values.go
Normal file
29
pkg/telemetrymetadata/resolvercore/family_values.go
Normal file
@@ -0,0 +1,29 @@
|
||||
package resolvercore
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/SigNoz/signoz/pkg/flagger"
|
||||
"github.com/SigNoz/signoz/pkg/querybuilder"
|
||||
"github.com/SigNoz/signoz/pkg/semconv"
|
||||
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
|
||||
"github.com/SigNoz/signoz/pkg/valuer"
|
||||
)
|
||||
|
||||
// FamilyValueNames returns the spellings whose values merge into the
|
||||
// suggestions. With the flag off, the requested name alone.
|
||||
func FamilyValueNames(ctx context.Context, orgID valuer.UUID, fl flagger.Flagger, signal telemetrytypes.Signal, fieldValueSelector *telemetrytypes.FieldValueSelector) []string {
|
||||
if !querybuilder.SemconvFamiliesEnabled(ctx, orgID, fl) {
|
||||
return []string{fieldValueSelector.Name}
|
||||
}
|
||||
selector := telemetrytypes.FieldKeySelector{
|
||||
Name: fieldValueSelector.Name,
|
||||
Signal: signal,
|
||||
FieldContext: fieldValueSelector.FieldContext,
|
||||
MetricContext: fieldValueSelector.MetricContext,
|
||||
}
|
||||
if signal == telemetrytypes.SignalMetrics {
|
||||
return querybuilder.MetricLabelSpellings(selector)
|
||||
}
|
||||
return semconv.Members(semconv.KindAttribute, selector)
|
||||
}
|
||||
52
pkg/telemetrymetadata/resolvercore/family_values_test.go
Normal file
52
pkg/telemetrymetadata/resolvercore/family_values_test.go
Normal file
@@ -0,0 +1,52 @@
|
||||
package resolvercore
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
|
||||
"github.com/SigNoz/signoz/pkg/flagger"
|
||||
"github.com/SigNoz/signoz/pkg/flagger/flaggertest"
|
||||
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
|
||||
"github.com/SigNoz/signoz/pkg/valuer"
|
||||
)
|
||||
|
||||
// The flagger provider registration is process-global and keyed by provider
|
||||
// name, so each flagger must be used before the next one is created.
|
||||
func TestFamilyValueNames(t *testing.T) {
|
||||
selector := &telemetrytypes.FieldValueSelector{
|
||||
FieldKeySelector: &telemetrytypes.FieldKeySelector{Name: "deployment.environment"},
|
||||
}
|
||||
|
||||
off := flaggertest.WithBooleanFlags(t, map[string]bool{})
|
||||
assert.Equal(t,
|
||||
[]string{"deployment.environment"},
|
||||
FamilyValueNames(context.Background(), valuer.UUID{}, off, telemetrytypes.SignalLogs, selector),
|
||||
"the flag default keeps values literal")
|
||||
|
||||
on := flaggertest.WithBooleanFlags(t, map[string]bool{
|
||||
flagger.FeatureResolveSemconvFamilies.String(): true,
|
||||
})
|
||||
assert.Equal(t,
|
||||
[]string{"deployment.environment.name", "deployment.environment"},
|
||||
FamilyValueNames(context.Background(), valuer.UUID{}, on, telemetrytypes.SignalLogs, selector),
|
||||
"values for one spelling must cover the whole family")
|
||||
assert.Equal(t,
|
||||
[]string{"deployment.environment.name", "deployment.environment"},
|
||||
FamilyValueNames(context.Background(), valuer.UUID{}, on, telemetrytypes.SignalMetrics, selector),
|
||||
"metric values cover the family members")
|
||||
spanMetric := &telemetrytypes.FieldValueSelector{
|
||||
FieldKeySelector: &telemetrytypes.FieldKeySelector{
|
||||
Name: "deployment.environment",
|
||||
MetricContext: &telemetrytypes.MetricContext{MetricName: "signoz_calls_total"},
|
||||
},
|
||||
}
|
||||
assert.Equal(t,
|
||||
[]string{
|
||||
"deployment.environment.name", "resource_deployment.environment.name",
|
||||
"deployment.environment", "resource_deployment.environment",
|
||||
},
|
||||
FamilyValueNames(context.Background(), valuer.UUID{}, on, telemetrytypes.SignalMetrics, spanMetric),
|
||||
"span-metrics values cover the resource_ spellings too")
|
||||
}
|
||||
25
pkg/telemetrymetadata/resolvercore/priority.go
Normal file
25
pkg/telemetrymetadata/resolvercore/priority.go
Normal file
@@ -0,0 +1,25 @@
|
||||
package resolvercore
|
||||
|
||||
import (
|
||||
"github.com/SigNoz/signoz/pkg/errors"
|
||||
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
|
||||
)
|
||||
|
||||
var ErrFailedToGetTblStatement = errors.Newf(errors.TypeInternal, errors.CodeInternal, "failed to get tbl statement")
|
||||
|
||||
// PriorityForContext orders the contexts a key can resolve to when the same
|
||||
// name exists in more than one, lowest first.
|
||||
func PriorityForContext(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
|
||||
}
|
||||
}
|
||||
@@ -1,4 +1,4 @@
|
||||
package telemetrymetadata
|
||||
package resolvercore
|
||||
|
||||
import (
|
||||
"strings"
|
||||
@@ -6,7 +6,7 @@ import (
|
||||
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
|
||||
)
|
||||
|
||||
func staticFieldMatchesAny(field telemetrytypes.TelemetryFieldKey, selectors []*telemetrytypes.FieldKeySelector) bool {
|
||||
func StaticFieldMatchesAny(field telemetrytypes.TelemetryFieldKey, selectors []*telemetrytypes.FieldKeySelector) bool {
|
||||
for _, selector := range selectors {
|
||||
if staticFieldMatches(field, selector) {
|
||||
return true
|
||||
161
pkg/telemetrymetadata/resolvercore/tag_table.go
Normal file
161
pkg/telemetrymetadata/resolvercore/tag_table.go
Normal file
@@ -0,0 +1,161 @@
|
||||
package resolvercore
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
|
||||
"github.com/huandu/go-sqlbuilder"
|
||||
|
||||
"github.com/SigNoz/signoz/pkg/clickhousesql"
|
||||
"github.com/SigNoz/signoz/pkg/errors"
|
||||
"github.com/SigNoz/signoz/pkg/telemetrystore"
|
||||
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
|
||||
)
|
||||
|
||||
// errFailedToGetTagValues keeps the message the tag table lookups have always
|
||||
// wrapped with, which names logs keys for every signal.
|
||||
var errFailedToGetTagValues = errors.Newf(errors.TypeInternal, errors.CodeInternal, "failed to get logs keys")
|
||||
|
||||
// TagTableValues returns the string and number values of the key from a
|
||||
// tag table, and true and false when the key is a known bool field or the
|
||||
// table holds bool rows for it. Bool rows do not count towards the limit.
|
||||
func TagTableValues(ctx context.Context, store telemetrystore.TelemetryStore, table string, fieldValueSelector *telemetrytypes.FieldValueSelector, names []string, knownBool bool) (*telemetrytypes.TelemetryFieldValues, bool, error) {
|
||||
limit := fieldValueSelector.Limit
|
||||
if limit == 0 {
|
||||
limit = 50
|
||||
}
|
||||
|
||||
values := &telemetrytypes.TelemetryFieldValues{}
|
||||
if knownBool {
|
||||
values.BoolValues = boolFieldValues(fieldValueSelector.Value).BoolValues
|
||||
if fieldValueSelector.FieldDataType == telemetrytypes.FieldDataTypeBool {
|
||||
return values, true, nil
|
||||
}
|
||||
} else if fieldValueSelector.FieldDataType == telemetrytypes.FieldDataTypeUnspecified {
|
||||
hasBoolRows, err := tagTableHasBoolRows(ctx, store, table, fieldValueSelector, names)
|
||||
if err != nil {
|
||||
return nil, false, err
|
||||
}
|
||||
if hasBoolRows {
|
||||
values.BoolValues = boolFieldValues(fieldValueSelector.Value).BoolValues
|
||||
}
|
||||
}
|
||||
|
||||
sb := sqlbuilder.Select("DISTINCT string_value, number_value").From(table)
|
||||
|
||||
if fieldValueSelector.Name != "" {
|
||||
sb.Where(tagKeyCondition(sb, fieldValueSelector.Name, names))
|
||||
}
|
||||
sb.Where(sb.NE("tag_data_type", telemetrytypes.FieldDataTypeBool.TagDataType()))
|
||||
|
||||
tagTableSinceDay(sb, fieldValueSelector.StartUnixMilli)
|
||||
|
||||
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", "%"+clickhousesql.LikePattern(fieldValueSelector.Value)+"%"))
|
||||
case telemetrytypes.FieldDataTypeNumber:
|
||||
sb.Where(sb.IsNotNull("number_value"))
|
||||
sb.Where(sb.ILike("toString(number_value)", "%"+clickhousesql.LikePattern(fieldValueSelector.Value)+"%"))
|
||||
case telemetrytypes.FieldDataTypeUnspecified:
|
||||
// or b/w string and number
|
||||
sb.Where(sb.Or(
|
||||
sb.ILike("string_value", "%"+clickhousesql.LikePattern(fieldValueSelector.Value)+"%"),
|
||||
sb.ILike("toString(number_value)", "%"+clickhousesql.LikePattern(fieldValueSelector.Value)+"%"),
|
||||
))
|
||||
}
|
||||
}
|
||||
|
||||
// query one extra to check if we hit the limit
|
||||
sb.Limit(limit + 1)
|
||||
|
||||
query, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
|
||||
|
||||
rows, err := store.ClickhouseDB().Query(ctx, query, args...)
|
||||
if err != nil {
|
||||
return nil, false, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, errFailedToGetTagValues.Error())
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
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, errFailedToGetTagValues.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
|
||||
}
|
||||
|
||||
// tagKeyCondition matches the requested key, or every spelling of its
|
||||
// family when there is more than one.
|
||||
func tagKeyCondition(sb *sqlbuilder.SelectBuilder, name string, names []string) string {
|
||||
if len(names) > 1 {
|
||||
return sb.In("tag_key", sqlbuilder.List(names))
|
||||
}
|
||||
return sb.E("tag_key", name)
|
||||
}
|
||||
|
||||
// tagTableSinceDay restricts rows to the tag table's day partitions from the
|
||||
// start's day on. Partitions are toDate(unix_milli / 1000) in the server's
|
||||
// timezone, and a value's surviving row within a day carries whichever hour
|
||||
// was inserted last, so the day is the finest safe unit.
|
||||
func tagTableSinceDay(sb *sqlbuilder.SelectBuilder, startUnixMilli int64) {
|
||||
if startUnixMilli != 0 {
|
||||
sb.Where(fmt.Sprintf("toDate(unix_milli / 1000) >= toDate(%d)", startUnixMilli/1000))
|
||||
}
|
||||
}
|
||||
|
||||
// tagTableHasBoolRows reports whether the tag table holds a bool row for the
|
||||
// key. Bool rows carry no value, so one row is enough to know the key takes
|
||||
// the values true and false.
|
||||
func tagTableHasBoolRows(ctx context.Context, store telemetrystore.TelemetryStore, table string, selector *telemetrytypes.FieldValueSelector, names []string) (bool, error) {
|
||||
sb := sqlbuilder.Select("1").From(table)
|
||||
sb.Where(tagKeyCondition(sb, selector.Name, names))
|
||||
sb.Where(sb.E("tag_data_type", telemetrytypes.FieldDataTypeBool.TagDataType()))
|
||||
if selector.FieldContext != telemetrytypes.FieldContextUnspecified {
|
||||
sb.Where(sb.E("tag_type", selector.FieldContext.TagType()))
|
||||
}
|
||||
tagTableSinceDay(sb, selector.StartUnixMilli)
|
||||
sb.Limit(1)
|
||||
|
||||
query, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
|
||||
rows, err := store.ClickhouseDB().Query(ctx, query, args...)
|
||||
if err != nil {
|
||||
return false, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, errFailedToGetTagValues.Error())
|
||||
}
|
||||
defer rows.Close()
|
||||
return rows.Next(), rows.Err()
|
||||
}
|
||||
235
pkg/telemetrymetadata/traces.go
Normal file
235
pkg/telemetrymetadata/traces.go
Normal file
@@ -0,0 +1,235 @@
|
||||
package telemetrymetadata
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"strings"
|
||||
|
||||
"github.com/huandu/go-sqlbuilder"
|
||||
"golang.org/x/exp/maps"
|
||||
|
||||
"github.com/SigNoz/signoz/pkg/clickhousesql"
|
||||
"github.com/SigNoz/signoz/pkg/errors"
|
||||
"github.com/SigNoz/signoz/pkg/telemetrymetadata/parser"
|
||||
"github.com/SigNoz/signoz/pkg/telemetrymetadata/resolvercore"
|
||||
"github.com/SigNoz/signoz/pkg/telemetryschema/tracestelemetryschema"
|
||||
"github.com/SigNoz/signoz/pkg/types/ctxtypes"
|
||||
"github.com/SigNoz/signoz/pkg/types/instrumentationtypes"
|
||||
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
|
||||
"github.com/SigNoz/signoz/pkg/valuer"
|
||||
)
|
||||
|
||||
// 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.ClickhouseDB().Select(ctx, &statements, query)
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, resolvercore.ErrFailedToGetTblStatement.Error())
|
||||
}
|
||||
|
||||
materialisedKeys, err := parser.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
|
||||
|
||||
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", "%"+clickhousesql.LikePattern(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, resource attribute or scope
|
||||
switch fieldKeySelector.FieldContext {
|
||||
case telemetrytypes.FieldContextAttribute, telemetrytypes.FieldContextResource, telemetrytypes.FieldContextScope:
|
||||
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(sqlbuilder.ClickHouse)
|
||||
|
||||
rows, err := t.telemetrystore.ClickhouseDB().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
|
||||
|
||||
// Add the matching static fields: the span scope selectors, the intrinsic
|
||||
// columns and the calculated columns. These don't count towards the limit
|
||||
staticFields := maps.Values(tracestelemetryschema.SpanSearchScopeFields)
|
||||
staticFields = append(staticFields, maps.Values(tracestelemetryschema.IntrinsicFields)...)
|
||||
staticFields = append(staticFields, maps.Values(tracestelemetryschema.CalculatedFields)...)
|
||||
|
||||
for _, field := range staticFields {
|
||||
if !resolvercore.StaticFieldMatchesAny(field, fieldKeySelectors) {
|
||||
continue
|
||||
}
|
||||
if _, added := mapOfKeys[field.Name+";"+field.FieldContext.StringValue()+";"+field.FieldDataType.StringValue()]; added {
|
||||
continue
|
||||
}
|
||||
keys = append(keys, &field)
|
||||
}
|
||||
|
||||
if err = t.updateColumnEvolutionMetadataForKeys(ctx, keys); err != nil {
|
||||
return nil, false, err
|
||||
}
|
||||
|
||||
return keys, complete, nil
|
||||
}
|
||||
|
||||
func (t *telemetryMetaStore) getSpanFieldValues(ctx context.Context, orgID valuer.UUID, 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",
|
||||
})
|
||||
|
||||
if values, ok := spanSearchScopeFieldValues(fieldValueSelector.Name, fieldValueSelector.Value); ok {
|
||||
return values, true, nil
|
||||
}
|
||||
knownBool := resolvercore.IsKnownBoolField(fieldValueSelector, tracestelemetryschema.IntrinsicFields, tracestelemetryschema.CalculatedFields)
|
||||
names := resolvercore.FamilyValueNames(ctx, orgID, t.fl, telemetrytypes.SignalTraces, fieldValueSelector)
|
||||
// unix_milli is the hour of the span start
|
||||
return resolvercore.TagTableValues(ctx, t.telemetrystore, t.tracesDBName+"."+t.tracesFieldsTblName, fieldValueSelector, names, knownBool)
|
||||
}
|
||||
|
||||
// spanSearchScopeFieldValues is the suggestion set for a search-scope selector
|
||||
// (isRoot, isEntryPoint), which only filters with true. ok is false for any
|
||||
// other name.
|
||||
func spanSearchScopeFieldValues(name, searchText string) (*telemetrytypes.TelemetryFieldValues, bool) {
|
||||
for scopeName := range tracestelemetryschema.SpanSearchScopeFields {
|
||||
if !strings.EqualFold(scopeName, name) {
|
||||
continue
|
||||
}
|
||||
values := &telemetrytypes.TelemetryFieldValues{}
|
||||
if needle := strings.ToLower(searchText); needle == "" || strings.Contains("true", needle) {
|
||||
values.BoolValues = []bool{true}
|
||||
}
|
||||
return values, true
|
||||
}
|
||||
return nil, false
|
||||
}
|
||||
49
pkg/types/telemetrytypes/resolver.go
Normal file
49
pkg/types/telemetrytypes/resolver.go
Normal file
@@ -0,0 +1,49 @@
|
||||
package telemetrytypes
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/SigNoz/signoz/pkg/types/metrictypes"
|
||||
"github.com/SigNoz/signoz/pkg/valuer"
|
||||
)
|
||||
|
||||
// FieldKeyResolver resolves keys and values for one source within a signal.
|
||||
// Implementations never dispatch; a MetadataStore owns the routing, registers
|
||||
// them by Source and merges what they return. Keys come back flat rather than
|
||||
// grouped by name because the store does the grouping across sources.
|
||||
type FieldKeyResolver interface {
|
||||
GetKeys(ctx context.Context, fieldKeySelectors []*FieldKeySelector) ([]*TelemetryFieldKey, bool, error)
|
||||
|
||||
GetAllValues(ctx context.Context, fieldValueSelector *FieldValueSelector) (*TelemetryFieldValues, bool, error)
|
||||
}
|
||||
|
||||
// MetricMetadata covers the lookups that only apply to the metrics signal.
|
||||
type MetricMetadata interface {
|
||||
// FetchTemporality fetches the temporality for metric
|
||||
FetchTemporality(ctx context.Context, orgID valuer.UUID, queryTimeRangeStartTs, queryTimeRangeEndTs uint64, metricName string) (metrictypes.Temporality, error)
|
||||
|
||||
// FetchTemporalityMulti fetches the temporality for multiple metrics
|
||||
FetchTemporalityMulti(ctx context.Context, orgID valuer.UUID, queryTimeRangeStartTs, queryTimeRangeEndTs uint64, metricNames ...string) (map[string]metrictypes.Temporality, error)
|
||||
|
||||
FetchTemporalityAndTypeMulti(ctx context.Context, orgID valuer.UUID, queryTimeRangeStartTs, queryTimeRangeEndTs uint64, metricNames ...string) (map[string]metrictypes.Temporality, map[string]metrictypes.Type, map[string]bool, error)
|
||||
|
||||
// GetFirstSeenFromMetricMetadata gets the first seen timestamp for a metric metadata lookup key.
|
||||
GetFirstSeenFromMetricMetadata(ctx context.Context, lookupKeys []MetricMetadataLookupKey) (map[MetricMetadataLookupKey]int64, error)
|
||||
|
||||
FetchLastSeenInfoMulti(ctx context.Context, orgID valuer.UUID, metricNames ...string) (map[string]int64, error)
|
||||
}
|
||||
|
||||
// LogsMetadata covers the JSON column evolution lookups, which today only the
|
||||
// logs signal materialises.
|
||||
type LogsMetadata interface {
|
||||
// ListJSONIndexes lists the per-path JSON skip indexes of the given source.
|
||||
ListJSONIndexes(ctx context.Context, lookup JSONIndexLookup, filters ...string) ([]TelemetryFieldKeySkipIndex, error)
|
||||
|
||||
// GetPromotedPaths lists the promoted paths recorded in the column
|
||||
// evolution table for the entry's signal, column and field context.
|
||||
GetPromotedPaths(ctx context.Context, entry EvolutionEntry, paths ...string) (map[string]bool, error)
|
||||
|
||||
// PromotePaths records promoted paths in the column evolution table as
|
||||
// rows templated by entry; FieldName and ReleaseTime are set per path.
|
||||
PromotePaths(ctx context.Context, entry EvolutionEntry, paths ...string) error
|
||||
}
|
||||
@@ -3,11 +3,12 @@ package telemetrytypes
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/SigNoz/signoz/pkg/types/metrictypes"
|
||||
"github.com/SigNoz/signoz/pkg/valuer"
|
||||
)
|
||||
|
||||
// MetadataStore is the interface for the telemetry metadata store.
|
||||
// MetadataStore is the interface for the telemetry metadata store. The key and
|
||||
// value lookups route to a per-signal FieldKeyResolver; the embedded interfaces
|
||||
// are signal-scoped and pass straight through.
|
||||
type MetadataStore interface {
|
||||
// GetKeys returns a map of field keys types.TelemetryFieldKey by name, there can be multiple keys with the same name
|
||||
// if they have different types or data types.
|
||||
@@ -26,32 +27,12 @@ type MetadataStore interface {
|
||||
// GetAllValues returns a list of all values.
|
||||
GetAllValues(ctx context.Context, orgID valuer.UUID, fieldValueSelector *FieldValueSelector) (*TelemetryFieldValues, bool, error)
|
||||
|
||||
// FetchTemporality fetches the temporality for metric
|
||||
FetchTemporality(ctx context.Context, orgID valuer.UUID, queryTimeRangeStartTs, queryTimeRangeEndTs uint64, metricName string) (metrictypes.Temporality, error)
|
||||
|
||||
// FetchTemporalityMulti fetches the temporality for multiple metrics
|
||||
FetchTemporalityMulti(ctx context.Context, orgID valuer.UUID, queryTimeRangeStartTs, queryTimeRangeEndTs uint64, metricNames ...string) (map[string]metrictypes.Temporality, error)
|
||||
|
||||
FetchTemporalityAndTypeMulti(ctx context.Context, orgID valuer.UUID, queryTimeRangeStartTs, queryTimeRangeEndTs uint64, metricNames ...string) (map[string]metrictypes.Temporality, map[string]metrictypes.Type, map[string]bool, error)
|
||||
|
||||
// ListJSONIndexes lists the per-path JSON skip indexes of the given source.
|
||||
ListJSONIndexes(ctx context.Context, lookup JSONIndexLookup, filters ...string) ([]TelemetryFieldKeySkipIndex, error)
|
||||
|
||||
// GetPromotedPaths lists the promoted paths recorded in the column
|
||||
// evolution table for the entry's signal, column and field context.
|
||||
GetPromotedPaths(ctx context.Context, entry EvolutionEntry, paths ...string) (map[string]bool, error)
|
||||
|
||||
// PromotePaths records promoted paths in the column evolution table as
|
||||
// rows templated by entry; FieldName and ReleaseTime are set per path.
|
||||
PromotePaths(ctx context.Context, entry EvolutionEntry, paths ...string) error
|
||||
|
||||
// GetMaterializedKeys lists the keys materialized as dedicated columns.
|
||||
GetMaterializedKeys(ctx context.Context, signal Signal) ([]*TelemetryFieldKey, error)
|
||||
|
||||
// GetFirstSeenFromMetricMetadata gets the first seen timestamp for a metric metadata lookup key.
|
||||
GetFirstSeenFromMetricMetadata(ctx context.Context, lookupKeys []MetricMetadataLookupKey) (map[MetricMetadataLookupKey]int64, error)
|
||||
MetricMetadata
|
||||
|
||||
FetchLastSeenInfoMulti(ctx context.Context, orgID valuer.UUID, metricNames ...string) (map[string]int64, error)
|
||||
LogsMetadata
|
||||
}
|
||||
|
||||
type MetricMetadataLookupKey struct {
|
||||
|
||||
Reference in New Issue
Block a user