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]>
153 lines
4.1 KiB
Go
153 lines
4.1 KiB
Go
package querier
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"fmt"
|
|
"log/slog"
|
|
"sort"
|
|
"strings"
|
|
"text/template"
|
|
"time"
|
|
|
|
datastore "github.com/hanzo-ds/go"
|
|
"github.com/hanzoai/o11y/pkg/errors"
|
|
"github.com/hanzoai/o11y/pkg/querybuilder"
|
|
"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"
|
|
)
|
|
|
|
type chSQLQuery struct {
|
|
logger *slog.Logger
|
|
telemetryStore telemetrystore.TelemetryStore
|
|
|
|
query qbtypes.DatastoreQuery
|
|
args []any
|
|
fromMS uint64
|
|
toMS uint64
|
|
kind qbtypes.RequestType
|
|
vars map[string]qbtypes.VariableItem
|
|
}
|
|
|
|
var _ qbtypes.Query = (*chSQLQuery)(nil)
|
|
var _ qbtypes.StatementProvider = (*chSQLQuery)(nil)
|
|
|
|
func newchSQLQuery(
|
|
logger *slog.Logger,
|
|
telemetryStore telemetrystore.TelemetryStore,
|
|
query qbtypes.DatastoreQuery,
|
|
args []any,
|
|
tr qbtypes.TimeRange,
|
|
kind qbtypes.RequestType,
|
|
variables map[string]qbtypes.VariableItem,
|
|
) *chSQLQuery {
|
|
return &chSQLQuery{
|
|
logger: logger,
|
|
telemetryStore: telemetryStore,
|
|
query: query,
|
|
args: args,
|
|
fromMS: tr.From,
|
|
toMS: tr.To,
|
|
kind: kind,
|
|
vars: variables,
|
|
}
|
|
}
|
|
|
|
func (q *chSQLQuery) Fingerprint() string {
|
|
// No caching for Datastore queries for now
|
|
return ""
|
|
}
|
|
|
|
func (q *chSQLQuery) Window() (uint64, uint64) { return q.fromMS, q.toMS }
|
|
|
|
// TODO(srikanthccv): cleanup the templating logic.
|
|
func (q *chSQLQuery) renderVars(query string, vars map[string]qbtypes.VariableItem, start, end uint64) (string, error) {
|
|
varsData := map[string]any{}
|
|
for k, v := range vars {
|
|
varsData[k] = formatValueForCH(v.Value)
|
|
}
|
|
|
|
querybuilder.AssignReservedVars(varsData, start, end)
|
|
|
|
keys := make([]string, 0, len(varsData))
|
|
for k := range varsData {
|
|
keys = append(keys, k)
|
|
}
|
|
sort.Slice(keys, func(i, j int) bool {
|
|
return len(keys[i]) > len(keys[j])
|
|
})
|
|
|
|
for _, k := range keys {
|
|
query = strings.ReplaceAll(query, fmt.Sprintf("{{%s}}", k), fmt.Sprint(varsData[k]))
|
|
query = strings.ReplaceAll(query, fmt.Sprintf("[[%s]]", k), fmt.Sprint(varsData[k]))
|
|
query = strings.ReplaceAll(query, fmt.Sprintf("$%s", k), fmt.Sprint(varsData[k]))
|
|
}
|
|
|
|
tmpl := template.New("datastore-query")
|
|
tmpl, err := tmpl.Parse(query)
|
|
if err != nil {
|
|
return "", errors.WrapInternalf(err, errors.CodeInternal, "error while replacing template variables")
|
|
}
|
|
var newQuery bytes.Buffer
|
|
|
|
// replace go template variables
|
|
err = tmpl.Execute(&newQuery, varsData)
|
|
if err != nil {
|
|
return "", errors.WrapInternalf(err, errors.CodeInternal, "error while replacing template variables")
|
|
}
|
|
return newQuery.String(), nil
|
|
}
|
|
|
|
// Statement renders the SQL without executing it, for the preview path.
|
|
func (q *chSQLQuery) Statement(_ context.Context) (*qbtypes.Statement, error) {
|
|
rendered, err := q.renderVars(q.query.Query, q.vars, q.fromMS, q.toMS)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return &qbtypes.Statement{Query: rendered, Args: q.args}, nil
|
|
}
|
|
|
|
func (q *chSQLQuery) Execute(ctx context.Context) (*qbtypes.Result, error) {
|
|
ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
|
|
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
|
|
}))
|
|
|
|
query, err := q.renderVars(q.query.Query, q.vars, q.fromMS, q.toMS)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
rows, err := q.telemetryStore.Datastore().Query(ctx, query, q.args...)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
|
|
// TODO: map the errors from Datastore to our error types
|
|
payload, err := consume(rows, q.kind, nil, qbtypes.Step{}, q.query.Name)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return &qbtypes.Result{
|
|
Type: q.kind,
|
|
Value: payload,
|
|
Stats: qbtypes.ExecStats{
|
|
RowsScanned: totalRows,
|
|
BytesScanned: totalBytes,
|
|
DurationMS: uint64(elapsed.Milliseconds()),
|
|
},
|
|
}, nil
|
|
}
|