mirror of
https://github.com/hanzoai/o11y.git
synced 2026-10-15 12:58:33 +00:00
zeekayandHanzo Dev
2965c80547
The five claim verbs need the address an operation will answer at, which is a root joined with a path. They used to ask the target for its prefix through a struct written to answer with one — and that struct existed for a second reason that has since gone: an operation with no declared id took a shape-derived one qualified by the prefix of the occurrence it answered under, so a plain group would have renamed every undeclared id. All 319 operations state their own id now, so a declared id survives composition verbatim and the group is just a group. The roots are this package's own constants, so each group carries the one it was made with and nothing can disagree about the address. The escape hatches name their whole address and register through Raw, which says which method each one answers. The test that measured the gap reached zero and said so; it now holds the property from the other side, that no operation goes undeclared. Co-authored-by: Hanzo Dev <[email protected]>
111 lines
3.1 KiB
Go
111 lines
3.1 KiB
Go
package querier
|
|
|
|
import (
|
|
"context"
|
|
"time"
|
|
|
|
datastore "github.com/hanzo-ds/go"
|
|
"github.com/hanzoai/o11y/pkg/telemetrystore"
|
|
"github.com/hanzoai/o11y/pkg/types/ctxtypes"
|
|
"github.com/hanzoai/o11y/pkg/types/instrumentationtypes"
|
|
qbtypes "github.com/hanzoai/o11y/pkg/types/querybuildertypes/querybuildertypesv5"
|
|
"github.com/hanzoai/o11y/pkg/types/telemetrytypes"
|
|
)
|
|
|
|
type traceOperatorQuery struct {
|
|
telemetryStore telemetrystore.TelemetryStore
|
|
stmtBuilder qbtypes.TraceOperatorStatementBuilder
|
|
spec qbtypes.QueryBuilderTraceOperator
|
|
compositeQuery *qbtypes.CompositeQuery
|
|
fromMS uint64
|
|
toMS uint64
|
|
kind qbtypes.RequestType
|
|
}
|
|
|
|
var _ qbtypes.Query = (*traceOperatorQuery)(nil)
|
|
var _ qbtypes.StatementProvider = (*traceOperatorQuery)(nil)
|
|
|
|
func (q *traceOperatorQuery) Fingerprint() string {
|
|
return ""
|
|
}
|
|
|
|
func (q *traceOperatorQuery) Window() (uint64, uint64) {
|
|
return q.fromMS, q.toMS
|
|
}
|
|
|
|
// Statement renders the SQL without executing it, for the preview path.
|
|
func (q *traceOperatorQuery) Statement(ctx context.Context) (*qbtypes.Statement, error) {
|
|
return q.stmtBuilder.Build(ctx, q.fromMS, q.toMS, q.kind, q.spec, q.compositeQuery)
|
|
}
|
|
|
|
func (q *traceOperatorQuery) Execute(ctx context.Context) (*qbtypes.Result, error) {
|
|
stmt, err := q.stmtBuilder.Build(
|
|
ctx,
|
|
q.fromMS,
|
|
q.toMS,
|
|
q.kind,
|
|
q.spec,
|
|
q.compositeQuery,
|
|
)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
// Execute the query with proper context
|
|
result, err := q.executeWithContext(ctx, stmt.Query, stmt.Args)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
result.Warnings = stmt.Warnings
|
|
return result, nil
|
|
}
|
|
|
|
func (q *traceOperatorQuery) executeWithContext(ctx context.Context, query string, args []any) (*qbtypes.Result, error) {
|
|
ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
|
|
instrumentationtypes.TelemetrySignal: telemetrytypes.SignalTraces.StringValue(),
|
|
instrumentationtypes.QueryDuration: instrumentationtypes.DurationBucket(q.fromMS, q.toMS),
|
|
})
|
|
|
|
totalRows := uint64(0)
|
|
totalBytes := uint64(0)
|
|
elapsed := time.Duration(0)
|
|
|
|
ctx = datastore.Context(ctx, datastore.WithProgress(func(p *datastore.Progress) {
|
|
totalRows += p.Rows
|
|
totalBytes += p.Bytes
|
|
elapsed += p.Elapsed
|
|
}))
|
|
|
|
rows, err := q.telemetryStore.Datastore().Query(ctx, query, args...)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
|
|
// Pass query window and step for partial value detection
|
|
queryWindow := &qbtypes.TimeRange{From: q.fromMS, To: q.toMS}
|
|
|
|
// Use the consume function like builderQuery does
|
|
payload, err := consume(rows, q.kind, queryWindow, q.spec.StepInterval, q.spec.Name)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
// TODO: This should move to readAsRaw function in consume.go but for now we can keep it here since it's only relevant for traces
|
|
if raw, ok := payload.(*qbtypes.RawData); ok {
|
|
for _, rr := range raw.Rows {
|
|
mergeSpanAttributeColumns(rr.Data)
|
|
}
|
|
}
|
|
|
|
return &qbtypes.Result{
|
|
Type: q.kind,
|
|
Value: payload,
|
|
Stats: qbtypes.ExecStats{
|
|
RowsScanned: totalRows,
|
|
BytesScanned: totalBytes,
|
|
DurationMS: uint64(elapsed.Milliseconds()),
|
|
},
|
|
}, nil
|
|
}
|