From 6adf6cefe2488f776bb0013e0aba8a7564c0a8c0 Mon Sep 17 00:00:00 2001 From: Jeffrey Aven Date: Mon, 7 Sep 2026 19:49:01 +1000 Subject: [PATCH 1/2] output: add otel format for inventory snapshots; add SHOW CONTRIBUTORS; bump minor to 0.11 (#738, #320) --output otel (#738, phase 1): one OTLP/JSON LogsData per result row plus a completion record per statement, via pkg/sink.NewOTelSink. Every record of a statement shares the snapshot instant (statement start), snapshot id and span; one trace per invocation honouring TRACEPARENT. Rows carry the full row JSON as body, typed column attributes, stackql.* context and a sha256 fingerprint. Resource attributes derive cloud.* from the statement and honour OTEL_SERVICE_NAME / OTEL_RESOURCE_ATTRIBUTES. Own scope with a pinned attribute schema. The statement text and start time reach the writer via OutputContext, stamped in driver.ProcessQuery. SHOW [EXTENDED] CONTRIBUTORS (#320): embedded contributors.csv in a new leaf package, wired through the same three SHOW touch points as VERSION. The CSV is generated by stackql itself (contributors joined to stargazers, humans only) and refreshed by the contributors workflow, which calls the binary directly because stackql-exec cannot lift the page cap the stargazer join needs. cicd/version.txt: MinorVersion 10 -> 11. Co-Authored-By: Claude Fable 5.1 --- .github/workflows/build.yml | 2 + .github/workflows/contributors.yml | 80 +++++ cicd/version.txt | 2 +- go.mod | 2 +- internal/stackql/cmd/root.go | 2 +- .../stackql/contributors/contributors.csv | 18 + internal/stackql/contributors/contributors.go | 65 ++++ .../stackql/contributors/contributors_test.go | 33 ++ internal/stackql/driver/driver.go | 2 + internal/stackql/handler/handler.go | 10 + .../internaldto/output_ctx.go | 6 + internal/stackql/output/otel.go | 313 ++++++++++++++++++ internal/stackql/output/otel_test.go | 248 ++++++++++++++ internal/stackql/output/output.go | 2 + .../stackql/primitivebuilder/shortcuts.go | 20 ++ .../primitivegenerator/primitive_generator.go | 2 +- .../primitivegenerator/statement_analyzer.go | 2 +- .../responsehandler/responsehandler.go | 4 + pkg/sink/otel.go | 40 ++- .../stackql_test_tooling/StackQLInterfaces.py | 57 ++++ test/robot/functional/mcp.robot | 21 ++ .../functional/stackql_from_cmd_line.robot | 20 ++ .../stackql_mocked_from_cmd_line.robot | 54 +++ 23 files changed, 990 insertions(+), 15 deletions(-) create mode 100644 .github/workflows/contributors.yml create mode 100644 internal/stackql/contributors/contributors.csv create mode 100644 internal/stackql/contributors/contributors.go create mode 100644 internal/stackql/contributors/contributors_test.go create mode 100644 internal/stackql/output/otel.go create mode 100644 internal/stackql/output/otel_test.go diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index 744125da..a6a8906c 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -26,6 +26,7 @@ on: - 'packaging/**' - '.github/workflows/mcp-packaging.yml' - '.github/workflows/mcp-registry-publish.yml' + - '.github/workflows/contributors.yml' pull_request: branches: - main @@ -49,6 +50,7 @@ on: - 'packaging/**' - '.github/workflows/mcp-packaging.yml' - '.github/workflows/mcp-registry-publish.yml' + - '.github/workflows/contributors.yml' env: GOTESTCMD: "go test -timeout 1200s --tags \"sqlite_stackql\" -v ./..." TESTSCRIPT: "test/deprecated/python/main.py" diff --git a/.github/workflows/contributors.yml b/.github/workflows/contributors.yml new file mode 100644 index 00000000..33ca718d --- /dev/null +++ b/.github/workflows/contributors.yml @@ -0,0 +1,80 @@ +name: contributors + +# Regenerates internal/stackql/contributors/contributors.csv, the leaderboard +# behind SHOW CONTRIBUTORS (issue #320), using stackql itself: human +# contributors who have starred the repo, contributions descending. Opens a +# PR so every change to the credit list is reviewed. PRs opened with the +# workflow token do not trigger CI; close and reopen the PR to run it. + +on: + schedule: + - cron: '17 3 1 * *' + workflow_dispatch: + +permissions: + contents: write + pull-requests: write + +env: + CSV_PATH: internal/stackql/contributors/contributors.csv + # Machine accounts operating as regular user accounts (PAT-driven CI + # identities, AI coding agents with user accounts). GitHub Apps are + # excluded generically by the type and [bot] filters in the query. + DENYLIST: "'claude'" + # An API failure or truncated result must not replace a healthy list. + MIN_ROWS: 10 + +jobs: + contributors: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v6 + + - uses: stackql/setup-stackql@v2.3.1 + + - name: Query contributors + env: + GITHUB_TOKEN: ${{ github.token }} + run: | + stackql exec "REGISTRY PULL github;" + stackql exec \ + --auth='{"github":{"type":"bearer","credentialsenvvar":"GITHUB_TOKEN"}}' \ + --http.response.pageLimit=-1 --http.response.maxResults=100 \ + --output csv \ + "SELECT c.login AS contributor, c.contributions + FROM github.repos.contributors c + INNER JOIN github.activity.repo_stargazers s ON s.login = c.login + WHERE c.owner = 'stackql' AND c.repo = 'stackql' + AND s.owner = 'stackql' AND s.repo = 'stackql' + AND c.type = 'User' + AND c.login NOT LIKE '%[bot]' + AND c.login NOT IN (${DENYLIST}) + ORDER BY c.contributions DESC, c.login ASC" > "${CSV_PATH}" + + - name: Guard contributor count + run: | + count=$(tail -n +2 "${CSV_PATH}" | wc -l) + if [ "${count}" -le "${MIN_ROWS}" ]; then + echo "::error::contributor query returned ${count} rows (<= ${MIN_ROWS}), refusing to update ${CSV_PATH}" + exit 1 + fi + echo "${count} contributors" + + - name: Open PR on change + env: + GH_TOKEN: ${{ github.token }} + run: | + if git diff --quiet -- "${CSV_PATH}"; then + echo "leaderboard unchanged" + exit 0 + fi + branch="contributors/$(date -u +%Y%m%d)" + git config user.name "github-actions[bot]" + git config user.email "41898282+github-actions[bot]@users.noreply.github.com" + git checkout -b "${branch}" + git add "${CSV_PATH}" + git commit -m "contributors: refresh leaderboard" + git push -u origin "${branch}" + gh pr create --base main --head "${branch}" \ + --title "contributors: refresh leaderboard" \ + --body "Regenerated by the contributors workflow. Human contributors who have starred the repo, contributions descending." diff --git a/cicd/version.txt b/cicd/version.txt index c783f880..16cce635 100644 --- a/cicd/version.txt +++ b/cicd/version.txt @@ -1,2 +1,2 @@ MajorVersion=0 -MinorVersion=10 +MinorVersion=11 diff --git a/go.mod b/go.mod index dd815fb5..64316177 100644 --- a/go.mod +++ b/go.mod @@ -7,6 +7,7 @@ require ( github.com/chzyer/readline v0.0.0-20180603132655-2972be24d48e github.com/getkin/kin-openapi v0.88.0 github.com/google/go-jsonnet v0.17.0 + github.com/google/uuid v1.6.0 github.com/jackc/pgtype v1.10.0 github.com/jackc/pgx/v5 v5.0.4 github.com/lib/pq v1.10.4 @@ -93,7 +94,6 @@ require ( github.com/google/flatbuffers v24.12.23+incompatible // indirect github.com/google/go-cmp v0.7.0 // indirect github.com/google/jsonschema-go v0.4.3 // indirect - github.com/google/uuid v1.6.0 // indirect github.com/gorilla/mux v1.8.0 // indirect github.com/gsterjov/go-libsecret v0.0.0-20161001094733-a6f4afe4910c // indirect github.com/hashicorp/hcl v1.0.0 // indirect diff --git a/internal/stackql/cmd/root.go b/internal/stackql/cmd/root.go index a5ded201..5480d274 100644 --- a/internal/stackql/cmd/root.go +++ b/internal/stackql/cmd/root.go @@ -212,7 +212,7 @@ func init() { rootCmd.PersistentFlags().BoolVarP(&runtimeCtx.VerboseFlag, dto.VerboseFlagKey, "v", false, "Verbose flag") rootCmd.PersistentFlags().BoolVar(&runtimeCtx.DryRunFlag, dto.DryRunFlagKey, false, "dryrun flag; preprocessor only will run and output returned") rootCmd.PersistentFlags().BoolVarP(&runtimeCtx.CSVHeadersDisable, dto.CSVHeadersDisableKey, "H", false, "Disable CSV headers flag") - rootCmd.PersistentFlags().StringVarP(&runtimeCtx.OutputFormat, dto.OutputFormatKey, "o", "table", "Output format, must be (json | jsonl | table | csv)") + rootCmd.PersistentFlags().StringVarP(&runtimeCtx.OutputFormat, dto.OutputFormatKey, "o", "table", "Output format, must be (json | jsonl | table | csv | text | otel)") rootCmd.PersistentFlags().StringVarP(&runtimeCtx.OutfilePath, dto.OutfilePathKey, "f", "stdout", "Output file into which results are written") rootCmd.PersistentFlags().StringVarP(&runtimeCtx.InfilePath, dto.InfilePathKey, "i", "stdin", "Input file from which queries are read") rootCmd.PersistentFlags().StringVarP(&runtimeCtx.TemplateCtxFilePath, dto.TemplateCtxFilePathKey, "q", "", "Context file for templating") diff --git a/internal/stackql/contributors/contributors.csv b/internal/stackql/contributors/contributors.csv new file mode 100644 index 00000000..35221a38 --- /dev/null +++ b/internal/stackql/contributors/contributors.csv @@ -0,0 +1,18 @@ +contributor,contributions +general-kroll-4-life,627 +jeffreyaven,50 +Iqbalabdi,4 +Sneha41sb,4 +jthegedus,4 +derek10cloud,3 +nirmalpatel008,2 +Racso-3141,1 +adityagiri3600,1 +datwiz,1 +kieranrimmer,1 +ryuichi-maeda,1 +ryukinoz,1 +sanketmp,1 +soysaucewaso,1 +tomekz,1 +yunchengyang515,1 diff --git a/internal/stackql/contributors/contributors.go b/internal/stackql/contributors/contributors.go new file mode 100644 index 00000000..3182134e --- /dev/null +++ b/internal/stackql/contributors/contributors.go @@ -0,0 +1,65 @@ +// Package contributors credits the humans behind stackql: the vendored +// leaderboard behind SHOW CONTRIBUTORS (issue #320). contributors.csv is +// regenerated by the contributors workflow and compiled in, so the build +// stays hermetic and each release freezes its credits. +package contributors + +import ( + _ "embed" + "encoding/csv" + "strconv" + "strings" + "sync" +) + +//go:embed contributors.csv +var contributorsCSV string //nolint:gochecknoglobals // embed target + +// csvFields is the column count of contributors.csv: contributor,contributions. +const csvFields = 2 + +// Contributor is one leaderboard entry. +type Contributor interface { + GetLogin() string + GetContributions() int +} + +type contributor struct { + login string + contributions int +} + +func (c contributor) GetLogin() string { return c.login } +func (c contributor) GetContributions() int { return c.contributions } + +//nolint:gochecknoglobals // parsed once +var ( + parseOnce sync.Once + parsed []Contributor +) + +// List returns the leaderboard in vendored order (contributions descending). +func List() []Contributor { + parseOnce.Do(func() { parsed = parse(contributorsCSV) }) + return parsed +} + +// parse skips the header and any malformed line. +func parse(raw string) []Contributor { + records, err := csv.NewReader(strings.NewReader(raw)).ReadAll() + if err != nil || len(records) == 0 { + return nil + } + rv := make([]Contributor, 0, len(records)-1) + for _, rec := range records[1:] { + if len(rec) != csvFields { + continue + } + n, convErr := strconv.Atoi(strings.TrimSpace(rec[1])) + if convErr != nil || strings.TrimSpace(rec[0]) == "" { + continue + } + rv = append(rv, contributor{login: strings.TrimSpace(rec[0]), contributions: n}) + } + return rv +} diff --git a/internal/stackql/contributors/contributors_test.go b/internal/stackql/contributors/contributors_test.go new file mode 100644 index 00000000..28d3e62b --- /dev/null +++ b/internal/stackql/contributors/contributors_test.go @@ -0,0 +1,33 @@ +package contributors_test + +import ( + "testing" + + "github.com/stackql/stackql/internal/stackql/contributors" +) + +// The embedded leaderboard must be non-empty, ordered by contributions +// descending, and free of blank or automated logins. +func TestList_EmbeddedLeaderboard(t *testing.T) { + list := contributors.List() + if len(list) == 0 { + t.Fatal("embedded contributors.csv parsed to an empty leaderboard") + } + for i, c := range list { + if c.GetLogin() == "" || c.GetContributions() <= 0 { + t.Fatalf("entry %d is malformed: %q %d", i, c.GetLogin(), c.GetContributions()) + } + if i > 0 && c.GetContributions() > list[i-1].GetContributions() { + t.Fatalf("entry %d (%s) breaks descending order", i, c.GetLogin()) + } + } + if list[0].GetLogin() != "general-kroll-4-life" { + t.Fatalf("expected the project founder on top, got %q", list[0].GetLogin()) + } +} + +func TestList_IsStable(t *testing.T) { + if a, b := contributors.List(), contributors.List(); len(a) != len(b) || &a[0] != &b[0] { + t.Fatal("List() must parse once and return the same leaderboard") + } +} diff --git a/internal/stackql/driver/driver.go b/internal/stackql/driver/driver.go index f7fb090d..8dc99609 100644 --- a/internal/stackql/driver/driver.go +++ b/internal/stackql/driver/driver.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "fmt" + "time" "github.com/stackql/any-sdk/pkg/logging" "github.com/stackql/any-sdk/public/sqlengine" @@ -124,6 +125,7 @@ func (dr *basicStackQLDriver) ProcessDryRun(query string) { func (dr *basicStackQLDriver) ProcessQuery(query string) { clonedCtx := dr.handlerCtx.Clone() clonedCtx.SetRawQuery(query) + clonedCtx.SetQueryStartTime(time.Now()) responses, ok := dr.processQueryOrQueries(clonedCtx) if ok { for _, r := range responses { diff --git a/internal/stackql/handler/handler.go b/internal/stackql/handler/handler.go index 0900027e..9a22e4fa 100644 --- a/internal/stackql/handler/handler.go +++ b/internal/stackql/handler/handler.go @@ -9,6 +9,7 @@ import ( "path" "strings" "sync" + "time" "github.com/stackql/any-sdk/pkg/constants" "github.com/stackql/any-sdk/pkg/db/sqlcontrol" @@ -57,6 +58,7 @@ type HandlerContext interface { //nolint:revive // don't mind stuttering this on // GetRawQuery() string GetQuery() string + GetQueryStartTime() time.Time GetRuntimeContext() dto.RuntimeCtx GetProviders() map[string]provider.IProvider GetControlAttributes() sqlcontrol.ControlAttributes @@ -83,6 +85,7 @@ type HandlerContext interface { //nolint:revive // don't mind stuttering this on SetCurrentProvider(string) SetQuery(string) SetRawQuery(string) + SetQueryStartTime(time.Time) // SetOutErrFile(io.Writer) // @@ -126,6 +129,7 @@ type standardHandlerContext struct { providersMapMutex *sync.Mutex rawQuery string query string + queryStartTime time.Time runtimeContext dto.RuntimeCtx providers map[string]provider.IProvider controlAttributes sqlcontrol.ControlAttributes @@ -214,6 +218,10 @@ func (hc *standardHandlerContext) SetRawQuery(rq string) { hc.rawQuery = rq } +func (hc *standardHandlerContext) SetQueryStartTime(t time.Time) { + hc.queryStartTime = t +} + func (hc *standardHandlerContext) SetQuery(q string) { hc.query = q } @@ -223,6 +231,7 @@ func (hc *standardHandlerContext) SetTxnCounterMgr(mgr txncounter.Manager) { } func (hc *standardHandlerContext) GetRawQuery() string { return hc.rawQuery } +func (hc *standardHandlerContext) GetQueryStartTime() time.Time { return hc.queryStartTime } func (hc *standardHandlerContext) GetQuery() string { return hc.query } func (hc *standardHandlerContext) GetRuntimeContext() dto.RuntimeCtx { return hc.runtimeContext } func (hc *standardHandlerContext) GetProviders() map[string]provider.IProvider { return hc.providers } @@ -568,6 +577,7 @@ func (hc *standardHandlerContext) Clone() HandlerContext { providersMapMutex: hc.providersMapMutex, drmConfig: hc.drmConfig, rawQuery: hc.rawQuery, + queryStartTime: hc.queryStartTime, runtimeContext: hc.runtimeContext, currentProvider: hc.currentProvider, providers: hc.providers, diff --git a/internal/stackql/internal_data_transfer/internaldto/output_ctx.go b/internal/stackql/internal_data_transfer/internaldto/output_ctx.go index d67832f1..2491f6d8 100644 --- a/internal/stackql/internal_data_transfer/internaldto/output_ctx.go +++ b/internal/stackql/internal_data_transfer/internaldto/output_ctx.go @@ -1,6 +1,8 @@ package internaldto import ( + "time" + "github.com/stackql/any-sdk/pkg/dto" "github.com/stackql/psql-wire/pkg/sqldata" ) @@ -8,4 +10,8 @@ import ( type OutputContext struct { RuntimeContext dto.RuntimeCtx Result sqldata.ISQLResultStream + // Query and StartTime describe the submitted statement; the otel writer + // stamps them on every record. + Query string + StartTime time.Time } diff --git a/internal/stackql/output/otel.go b/internal/stackql/output/otel.go new file mode 100644 index 00000000..fce4f502 --- /dev/null +++ b/internal/stackql/output/otel.go @@ -0,0 +1,313 @@ +package output + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io" + "net/url" + "os" + "strings" + "sync" + "time" + + "github.com/google/uuid" + "github.com/stackql/psql-wire/pkg/sqldata" + "github.com/stackql/stackql-parser/go/vt/sqlparser" + "github.com/stackql/stackql/internal/stackql/buildinfo" + "github.com/stackql/stackql/pkg/sink" +) + +// The emitted attribute set is a versioned interface (issue #738): bump +// otelAttributeSchemaVersion when it changes and update the schema test. +const ( + otelFormatStr = "otel" + otelAttributeSchemaVersion = "1.0.0" + otelScopeName = "github.com/stackql/stackql/internal/stackql/output" + otelServiceName = "stackql" + otelTraceParentEnv = "TRACEPARENT" + otelServiceNameEnv = "OTEL_SERVICE_NAME" + otelResourceAttrsEnv = "OTEL_RESOURCE_ATTRIBUTES" + otelTraceIDBytes = 16 + otelSpanIDBytes = 8 + otelTraceParentParts = 4 +) + +// otelInvocationTraceID is shared by every statement of one process, +// honouring a TRACEPARENT supplied by the environment. +// +//nolint:gochecknoglobals // per-process trace identity +var otelInvocationTraceID = sync.OnceValue(func() string { + parts := strings.Split(os.Getenv(otelTraceParentEnv), "-") + if len(parts) == otelTraceParentParts && len(parts[1]) == 2*otelTraceIDBytes { + return parts[1] + } + return sink.OTelRandomID(otelTraceIDBytes) +}) + +// OTelWriter emits one OTLP/JSON LogsData per result row plus a completion +// record per statement, so a result set can be shipped as a timestamped +// inventory snapshot. Every record of a statement shares the snapshot +// instant, snapshot id and span id. +type OTelWriter struct { + errWriter io.Writer + encoder sink.Sink + startTime time.Time + snapshotID string + spanID string + query string + queryHash string + statement otelStatementContext +} + +// otelStatementContext is what a lightweight parse of the statement yields: +// the primary table and any cloud scoping parameters. +type otelStatementContext struct { + provider string + service string + resource string + cloud []sink.OTelAttribute +} + +// otelRecordPayload adapts a prepared record to the sink decorator. +type otelRecordPayload struct { + record sink.OTelLogRecord +} + +func (p otelRecordPayload) OTelLogRecords() []sink.OTelLogRecord { + return []sink.OTelLogRecord{p.record} +} + +// writerSink is the io.Writer-backed sink the decorator writes through: one +// compact JSON document per line, flushed per record like the jsonl writer. +type writerSink struct { + writer io.Writer +} + +func (s writerSink) Record(_ context.Context, payload any) error { + b, err := json.Marshal(payload) + if err != nil { + return err + } + if _, err = s.writer.Write(append(b, '\n')); err != nil { + return err + } + if flusher, ok := s.writer.(writeFlusher); ok { + return flusher.Flush() + } + return nil +} + +func (writerSink) Close() error { return nil } + +// NewOTelWriter returns the writer for one statement: query is the submitted +// text and startTime the snapshot instant stamped on every record. +func NewOTelWriter(writer io.Writer, errWriter io.Writer, query string, startTime time.Time) IOutputWriter { + if startTime.IsZero() { + startTime = time.Now() + } + statement := newOTelStatementContext(query) + queryHash := sha256.Sum256([]byte(strings.TrimSpace(query))) + return &OTelWriter{ + errWriter: errWriter, + encoder: sink.NewOTelSink(writerSink{writer: writer}, otelResource(statement), otelScope()), + startTime: startTime, + snapshotID: uuid.NewString(), + spanID: sink.OTelRandomID(otelSpanIDBytes), + query: query, + queryHash: hex.EncodeToString(queryHash[:]), + statement: statement, + } +} + +func otelScope() sink.OTelScope { + return sink.OTelScope{Name: otelScopeName, Version: otelAttributeSchemaVersion} +} + +// otelResource is service.*, the derivable cloud.* conventions, then the +// standard OTEL_RESOURCE_ATTRIBUTES pairs. +func otelResource(statement otelStatementContext) sink.OTelResource { + serviceName := os.Getenv(otelServiceNameEnv) + if serviceName == "" { + serviceName = otelServiceName + } + attrs := append([]sink.OTelAttribute{}, statement.cloud...) + for _, pair := range strings.Split(os.Getenv(otelResourceAttrsEnv), ",") { + key, value, found := strings.Cut(pair, "=") + if !found || strings.TrimSpace(key) == "" { + continue + } + if unescaped, err := url.PathUnescape(value); err == nil { + value = unescaped + } + attrs = append(attrs, sink.OTelString(strings.TrimSpace(key), strings.TrimSpace(value))) + } + return sink.OTelResource{ + ServiceName: serviceName, + ServiceVersion: buildinfo.Get().GetSemVersion(), + Attributes: attrs, + } +} + +// newOTelStatementContext parses a single-table SELECT for its +// provider.service.resource and the equality parameters that map onto the +// cloud resource conventions; anything more complex yields nothing. +func newOTelStatementContext(query string) otelStatementContext { + var rv otelStatementContext + stmt, err := sqlparser.Parse(query) + if err != nil { + return rv + } + sel, ok := stmt.(*sqlparser.Select) + if !ok || len(sel.From) != 1 { + return rv + } + aliased, ok := sel.From[0].(*sqlparser.AliasedTableExpr) + if !ok { + return rv + } + tableName, ok := aliased.Expr.(sqlparser.TableName) + if !ok || tableName.QualifierSecond.IsEmpty() { + return rv + } + rv.provider = tableName.QualifierSecond.GetRawVal() + rv.service = tableName.Qualifier.GetRawVal() + rv.resource = tableName.Name.GetRawVal() + params := map[string]string{} + if sel.Where != nil { + otelCollectEqualities(sel.Where.Expr, params) + } + rv.cloud = otelCloudAttributes(rv.provider, params) + return rv +} + +// otelCollectEqualities gathers `column = 'literal'` terms of an AND chain. +func otelCollectEqualities(expr sqlparser.Expr, out map[string]string) { + switch node := expr.(type) { + case *sqlparser.AndExpr: + otelCollectEqualities(node.Left, out) + otelCollectEqualities(node.Right, out) + case *sqlparser.ComparisonExpr: + col, isCol := node.Left.(*sqlparser.ColName) + val, isVal := node.Right.(*sqlparser.SQLVal) + if node.Operator == sqlparser.EqualStr && isCol && isVal && val.Type == sqlparser.StrVal { + out[strings.ToLower(col.Name.GetRawVal())] = string(val.Val) + } + } +} + +func otelCloudAttributes(providerName string, params map[string]string) []sink.OTelAttribute { + var attrs []sink.OTelAttribute + switch { + case strings.HasPrefix(providerName, "google"): + attrs = append(attrs, sink.OTelString("cloud.provider", "gcp")) + case strings.HasPrefix(providerName, "aws"): + attrs = append(attrs, sink.OTelString("cloud.provider", "aws")) + case strings.HasPrefix(providerName, "azure"): + attrs = append(attrs, sink.OTelString("cloud.provider", "azure")) + case providerName == "oci": + attrs = append(attrs, sink.OTelString("cloud.provider", "oracle_cloud")) + } + for param, key := range map[string]string{ + "project": "cloud.account.id", + "subscriptionid": "cloud.account.id", + "region": "cloud.region", + "zone": "cloud.availability_zone", + } { + if v := params[param]; v != "" { + attrs = append(attrs, sink.OTelString(key, v)) + } + } + return attrs +} + +// contextAttributes are the stackql.* attributes common to every record of +// the statement. +func (ow *OTelWriter) contextAttributes() []sink.OTelAttribute { + return []sink.OTelAttribute{ + sink.OTelString("stackql.snapshot.id", ow.snapshotID), + sink.OTelString("stackql.query", ow.query), + sink.OTelString("stackql.query.hash", ow.queryHash), + sink.OTelString("stackql.provider", ow.statement.provider), + sink.OTelString("stackql.service", ow.statement.service), + sink.OTelString("stackql.resource", ow.statement.resource), + } +} + +func (ow *OTelWriter) record(severity int, body string, attrs []sink.OTelAttribute) error { + return ow.encoder.Record(context.Background(), otelRecordPayload{sink.OTelLogRecord{ + Time: ow.startTime, + Severity: severity, + Body: body, + Attributes: attrs, + TraceParent: fmt.Sprintf("00-%s-%s-01", otelInvocationTraceID(), ow.spanID), + }}) +} + +// writeRow emits one row: the full row JSON as the body, scalar columns as +// typed attributes, nested columns as JSON strings, plus the row's index and +// a fingerprint of the canonical row JSON so drift is one comparison. +func (ow *OTelWriter) writeRow(row map[string]interface{}, index int) error { + body, err := json.Marshal(row) + if err != nil { + return err + } + attrs, _ := sink.OTelAttributesFromJSON(body) + fingerprint := sha256.Sum256(body) + attrs = append(attrs, ow.contextAttributes()...) + attrs = append(attrs, + sink.OTelInt("stackql.row.index", int64(index)), + sink.OTelString("stackql.row.fingerprint", "sha256:"+hex.EncodeToString(fingerprint[:])), + ) + return ow.record(sink.OTelSeverityInfo, string(body), attrs) +} + +// writeCompletion marks the snapshot whole; a zero-row statement still emits +// it so an absent resource is visible as drift rather than as silence. +func (ow *OTelWriter) writeCompletion(rows int) error { + body := "snapshot complete" + if ow.statement.provider != "" { + body += " " + strings.Join([]string{ow.statement.provider, ow.statement.service, ow.statement.resource}, ".") + } + attrs := append(ow.contextAttributes(), + sink.OTelBool("stackql.snapshot.complete", true), + sink.OTelInt("stackql.rows_returned", int64(rows)), + sink.OTelInt("stackql.duration_ms", time.Since(ow.startTime).Milliseconds()), + ) + return ow.record(sink.OTelSeverityInfo, body, attrs) +} + +func (ow *OTelWriter) Write(res sqldata.ISQLResultStream) error { + rows := 0 + for { + r, err := res.Read() + if err != nil && !errors.Is(err, io.EOF) { + return err + } + if r != nil { + for _, row := range r.ToArr() { + if rowErr := ow.writeRow(row, rows); rowErr != nil { + return rowErr + } + rows++ + } + } + if err != nil { + return ow.writeCompletion(rows) + } + } +} + +func (ow *OTelWriter) WriteError(err error, errorPresentation string) error { + if errorPresentation == stderrPressentationStr { + return writeStderrError(ow.errWriter, err) + } + attrs := append(ow.contextAttributes(), + sink.OTelString("error.type", fmt.Sprintf("%T", err)), + sink.OTelString("error.message", err.Error()), + ) + return ow.record(sink.OTelSeverityError, err.Error(), attrs) +} diff --git a/internal/stackql/output/otel_test.go b/internal/stackql/output/otel_test.go new file mode 100644 index 00000000..c5fa7220 --- /dev/null +++ b/internal/stackql/output/otel_test.go @@ -0,0 +1,248 @@ +package output //nolint:testpackage // do not care + +import ( + "bytes" + "encoding/json" + "errors" + "reflect" + "strings" + "testing" + "time" + + "github.com/stackql/any-sdk/pkg/dto" + "github.com/stackql/psql-wire/pkg/sqldata" + "github.com/stackql/stackql/internal/stackql/internal_data_transfer/internaldto" + "github.com/stackql/stackql/pkg/sink" +) + +const otelTestQuery = "select name, region, disabled, spec from google.compute.firewalls " + + "where project = 'p1' and zone = 'z1'" + +func otelTestStream(rows ...[]interface{}) sqldata.ISQLResultStream { + table := sqldata.NewSQLTable(0, "t") + cols := []sqldata.ISQLColumn{} + for i, name := range []string{"name", "region", "disabled", "spec"} { + cols = append(cols, sqldata.NewSQLColumn(table, name, int16(i), 0, 0, 0, "text")) //nolint:gosec // test + } + sqlRows := make([]sqldata.ISQLRow, 0, len(rows)) + for _, r := range rows { + sqlRows = append(sqlRows, sqldata.NewSQLRow(r)) + } + return sqldata.NewSimpleSQLResultStream(sqldata.NewSQLResult(cols, 0, 0, sqlRows)) +} + +// otelLines decodes one LogsData per line and returns its single record +// with the resource and scope it was wrapped in. +func otelLines(t *testing.T, out string) []sink.OTelResourceLogs { + t.Helper() + var rv []sink.OTelResourceLogs + for _, line := range strings.Split(strings.TrimSpace(out), "\n") { + var data sink.OTelLogsData + if err := json.Unmarshal([]byte(line), &data); err != nil { + t.Fatalf("line is not LogsData: %v: %q", err, line) + } + if len(data.ResourceLogs) != 1 || len(data.ResourceLogs[0].ScopeLogs) != 1 || + len(data.ResourceLogs[0].ScopeLogs[0].LogRecords) != 1 { + t.Fatalf("expected one record per line, got %q", line) + } + rv = append(rv, data.ResourceLogs[0]) + } + return rv +} + +func otelRecord(rl sink.OTelResourceLogs) sink.OTelWireLogRecord { + return rl.ScopeLogs[0].LogRecords[0] +} + +func otelAttr(attrs []sink.OTelAttribute, key string) (sink.OTelAnyValue, bool) { + for _, a := range attrs { + if a.Key == key { + return a.Value, true + } + } + return sink.OTelAnyValue{}, false +} + +func otelString(t *testing.T, attrs []sink.OTelAttribute, key string) string { + t.Helper() + v, ok := otelAttr(attrs, key) + if !ok || v.StringValue == nil { + t.Fatalf("missing string attribute %q in %+v", key, attrs) + } + return *v.StringValue +} + +func otelKeys(attrs []sink.OTelAttribute) []string { + keys := make([]string, 0, len(attrs)) + for _, a := range attrs { + keys = append(keys, a.Key) + } + return keys +} + +// otelSchemaFixture writes two rows and returns the decoded lines. +func otelSchemaFixture(t *testing.T) []sink.OTelResourceLogs { + t.Helper() + var out bytes.Buffer + start := time.Date(2026, 9, 7, 0, 0, 0, 0, time.UTC) + w := NewOTelWriter(&out, &bytes.Buffer{}, otelTestQuery, start) + stream := otelTestStream( + []interface{}{"fw-a", "us-east1", false, map[string]interface{}{"priority": 1000}}, + []interface{}{"fw-b", "us-east1", true, nil}, + ) + if err := w.Write(stream); err != nil { + t.Fatalf("Write() error = %v", err) + } + lines := otelLines(t, out.String()) + if len(lines) != 3 { + t.Fatalf("expected 2 rows + completion, got %d lines", len(lines)) + } + return lines +} + +func TestOTelWriter_AttributeSchema(t *testing.T) { + lines := otelSchemaFixture(t) + wantRow := []string{ + "disabled", "name", "region", "spec", + "stackql.snapshot.id", "stackql.query", "stackql.query.hash", + "stackql.provider", "stackql.service", "stackql.resource", + "stackql.row.index", "stackql.row.fingerprint", + } + if got := otelKeys(otelRecord(lines[0]).Attributes); !reflect.DeepEqual(got, wantRow) { + t.Fatalf("row attribute schema drift:\n got %v\nwant %v", got, wantRow) + } + wantCompletion := []string{ + "stackql.snapshot.id", "stackql.query", "stackql.query.hash", + "stackql.provider", "stackql.service", "stackql.resource", + "stackql.snapshot.complete", "stackql.rows_returned", "stackql.duration_ms", + } + if got := otelKeys(otelRecord(lines[2]).Attributes); !reflect.DeepEqual(got, wantCompletion) { + t.Fatalf("completion attribute schema drift:\n got %v\nwant %v", got, wantCompletion) + } + for _, rl := range lines { + if scope := rl.ScopeLogs[0].Scope; scope.Name != otelScopeName || scope.Version != otelAttributeSchemaVersion { + t.Fatalf("scope: %+v", scope) + } + } +} + +func TestOTelWriter_RowValuesAndBody(t *testing.T) { + lines := otelSchemaFixture(t) + row := otelRecord(lines[0]) + if v, _ := otelAttr(row.Attributes, "disabled"); v.BoolValue == nil || *v.BoolValue { + t.Fatalf("disabled should be boolValue false: %+v", v) + } + if otelString(t, row.Attributes, "spec") != `{"priority":1000}` { + t.Fatalf("nested column must be a JSON string: %+v", row.Attributes) + } + if _, ok := otelAttr(otelRecord(lines[1]).Attributes, "spec"); ok { + t.Fatalf("null column must be dropped from attributes") + } + if row.Body.StringValue == nil || *row.Body.StringValue != `{"disabled":false,"name":"fw-a","region":"us-east1","spec":{"priority":1000}}` { + t.Fatalf("body must be the full row JSON: %+v", row.Body) + } + if otelString(t, row.Attributes, "stackql.provider") != "google" || + otelString(t, row.Attributes, "stackql.service") != "compute" || + otelString(t, row.Attributes, "stackql.resource") != "firewalls" || + otelString(t, row.Attributes, "stackql.query") != otelTestQuery || + len(otelString(t, row.Attributes, "stackql.query.hash")) != 64 { + t.Fatalf("statement context: %+v", row.Attributes) + } +} + +func TestOTelWriter_CompletionRecord(t *testing.T) { + completion := otelRecord(otelSchemaFixture(t)[2]) + if v, _ := otelAttr(completion.Attributes, "stackql.rows_returned"); v.IntValue == nil || *v.IntValue != "2" { + t.Fatalf("rows_returned: %+v", v) + } + if v, _ := otelAttr(completion.Attributes, "stackql.snapshot.complete"); v.BoolValue == nil || !*v.BoolValue { + t.Fatalf("snapshot.complete: %+v", v) + } + if completion.Body.StringValue == nil || *completion.Body.StringValue != "snapshot complete google.compute.firewalls" { + t.Fatalf("completion body: %+v", completion.Body) + } +} + +// Every record of a statement shares the snapshot instant, id, trace and span. +func TestOTelWriter_StatementIdentity(t *testing.T) { + lines := otelSchemaFixture(t) + first := otelRecord(lines[0]) + snapshotID := otelString(t, first.Attributes, "stackql.snapshot.id") + for i, rl := range lines { + rec := otelRecord(rl) + if rec.TimeUnixNano != "1788739200000000000" || rec.TraceID != otelInvocationTraceID() || + rec.SpanID != first.SpanID || otelString(t, rec.Attributes, "stackql.snapshot.id") != snapshotID { + t.Fatalf("record %d does not share the statement identity: %+v", i, rec) + } + } +} + +func TestOTelWriter_ResourceAttributes(t *testing.T) { + resource := otelSchemaFixture(t)[0].Resource.Attributes + if otelString(t, resource, "service.name") != "stackql" || + otelString(t, resource, "cloud.provider") != "gcp" || + otelString(t, resource, "cloud.account.id") != "p1" || + otelString(t, resource, "cloud.availability_zone") != "z1" { + t.Fatalf("resource attributes: %+v", resource) + } +} + +func TestOTelWriter_ZeroRowsEmitsOnlyCompletion(t *testing.T) { + var out bytes.Buffer + w := NewOTelWriter(&out, &bytes.Buffer{}, otelTestQuery, time.Time{}) + if err := w.Write(otelTestStream()); err != nil { + t.Fatalf("Write() error = %v", err) + } + lines := otelLines(t, out.String()) + if len(lines) != 1 { + t.Fatalf("expected exactly one completion record, got %d", len(lines)) + } + if v, _ := otelAttr(otelRecord(lines[0]).Attributes, "stackql.rows_returned"); v.IntValue == nil || *v.IntValue != "0" { + t.Fatalf("rows_returned: %+v", v) + } +} + +func TestOTelWriter_FingerprintTracksRowContent(t *testing.T) { + fingerprint := func(disabled bool) string { + var out bytes.Buffer + w := NewOTelWriter(&out, &bytes.Buffer{}, otelTestQuery, time.Time{}) + if err := w.Write(otelTestStream([]interface{}{"fw-a", "us-east1", disabled, nil})); err != nil { + t.Fatalf("Write() error = %v", err) + } + return otelString(t, otelRecord(otelLines(t, out.String())[0]).Attributes, "stackql.row.fingerprint") + } + if a, b := fingerprint(false), fingerprint(false); a != b || !strings.HasPrefix(a, "sha256:") { + t.Fatalf("identical rows must fingerprint identically: %s vs %s", a, b) + } + if fingerprint(false) == fingerprint(true) { + t.Fatalf("a changed column value must change the fingerprint") + } +} + +func TestOTelWriter_WriteError(t *testing.T) { + var out, errOut bytes.Buffer + w := NewOTelWriter(&out, &errOut, "select 1", time.Time{}) + if err := w.WriteError(errors.New("boom"), "record"); err != nil { + t.Fatalf("WriteError() error = %v", err) + } + rec := otelRecord(otelLines(t, out.String())[0]) + if rec.SeverityNumber != sink.OTelSeverityError || rec.SeverityText != "ERROR" || + otelString(t, rec.Attributes, "error.message") != "boom" || + otelString(t, rec.Attributes, "error.type") != "*errors.errorString" { + t.Fatalf("error record: %+v", rec) + } + if err := w.WriteError(errors.New("boom"), stderrPressentationStr); err != nil || strings.TrimSpace(errOut.String()) != "boom" { + t.Fatalf("stderr presentation: err=%v out=%q", err, errOut.String()) + } +} + +func TestGetOutputWriter_OTel(t *testing.T) { + ctx := internaldto.OutputContext{RuntimeContext: dto.RuntimeCtx{OutputFormat: "otel"}, Query: "select 1"} + w, err := GetOutputWriter(&bytes.Buffer{}, &bytes.Buffer{}, ctx) + if err != nil { + t.Fatalf("GetOutputWriter() error = %v", err) + } + if _, ok := w.(*OTelWriter); !ok { + t.Fatalf("expected *OTelWriter, got %T", w) + } +} diff --git a/internal/stackql/output/output.go b/internal/stackql/output/output.go index 9af75352..7f8c47d1 100644 --- a/internal/stackql/output/output.go +++ b/internal/stackql/output/output.go @@ -63,6 +63,8 @@ func GetOutputWriter( case "jsonl", "ndjson": jsonlWriter := NewJSONLWriter(writer, errWriter) return jsonlWriter, nil + case otelFormatStr: + return NewOTelWriter(writer, errWriter, outputCtx.Query, outputCtx.StartTime), nil case constants.TableStr: tablewriter := TableWriter{ AbstractTabularWriter{ diff --git a/internal/stackql/primitivebuilder/shortcuts.go b/internal/stackql/primitivebuilder/shortcuts.go index c26d325f..3d472cc9 100644 --- a/internal/stackql/primitivebuilder/shortcuts.go +++ b/internal/stackql/primitivebuilder/shortcuts.go @@ -11,6 +11,7 @@ import ( "github.com/stackql/any-sdk/public/formulation" "github.com/stackql/stackql-parser/go/vt/sqlparser" "github.com/stackql/stackql/internal/stackql/buildinfo" + "github.com/stackql/stackql/internal/stackql/contributors" "github.com/stackql/stackql/internal/stackql/handler" "github.com/stackql/stackql/internal/stackql/internal_data_transfer/internaldto" "github.com/stackql/stackql/internal/stackql/metadatavisitors" @@ -251,6 +252,8 @@ func NewShowInstructionExecutor( keys = convertProviderServicesToMap(services, extended) case "VERSION": columnOrder, keys = buildVersionShowOutput(extended) + case "CONTRIBUTORS": + columnOrder, keys = buildContributorsShowOutput(extended) } return util.PrepareResultSet(internaldto.NewPrepareResultSetDTO(nil, keys, columnOrder, nil, err, nil, handlerCtx.GetTypingConfig())) @@ -340,6 +343,23 @@ func buildVersionShowOutput(extended bool) ([]string, map[string]map[string]inte } } +// buildContributorsShowOutput renders SHOW [EXTENDED] CONTRIBUTORS; keys are +// zero-padded so the sorted result set keeps leaderboard order. +func buildContributorsShowOutput(extended bool) ([]string, map[string]map[string]interface{}) { + keys := make(map[string]map[string]interface{}) + for i, c := range contributors.List() { + row := map[string]interface{}{"contributor": c.GetLogin()} + if extended { + row["contributions"] = c.GetContributions() + } + keys[fmt.Sprintf("%06d", i)] = row + } + if extended { + return []string{"contributor", "contributions"}, keys + } + return []string{"contributor"}, keys +} + //nolint:errcheck // future proofing func filterResources( resources map[string]formulation.Resource, diff --git a/internal/stackql/primitivegenerator/primitive_generator.go b/internal/stackql/primitivegenerator/primitive_generator.go index 8696caea..f1a070b0 100644 --- a/internal/stackql/primitivegenerator/primitive_generator.go +++ b/internal/stackql/primitivegenerator/primitive_generator.go @@ -272,7 +272,7 @@ func (pb *standardPrimitiveGenerator) inferProviderForShow( pb.PrimitiveComposer.SetProvider(prov) case "PROVIDERS": // no provider, might create some dummy object dunno - case "VERSION": + case "VERSION", "CONTRIBUTORS": // no provider needed case "RESOURCES": prov, err := handlerCtx.GetProvider(node.OnTable.Qualifier.GetRawVal()) diff --git a/internal/stackql/primitivegenerator/statement_analyzer.go b/internal/stackql/primitivegenerator/statement_analyzer.go index e276b0bf..d51d0f5a 100644 --- a/internal/stackql/primitivegenerator/statement_analyzer.go +++ b/internal/stackql/primitivegenerator/statement_analyzer.go @@ -1668,7 +1668,7 @@ func (pb *standardPrimitiveGenerator) analyzeShow( return nil case "PROVIDERS": // TODO - case "VERSION": + case "VERSION", "CONTRIBUTORS": // no further analysis required case "RESOURCES": prov, err := handlerCtx.GetProvider(node.OnTable.Qualifier.GetRawVal()) diff --git a/internal/stackql/responsehandler/responsehandler.go b/internal/stackql/responsehandler/responsehandler.go index dcb22c9d..30e06483 100644 --- a/internal/stackql/responsehandler/responsehandler.go +++ b/internal/stackql/responsehandler/responsehandler.go @@ -37,6 +37,8 @@ func HandleResponse(handlerCtx handler.HandlerContext, response internaldto.Exec internaldto.OutputContext{ RuntimeContext: handlerCtx.GetRuntimeContext(), Result: sqlResult, + Query: handlerCtx.GetRawQuery(), + StartTime: handlerCtx.GetQueryStartTime(), }, ) if outputWriter == nil || err != nil { @@ -51,6 +53,8 @@ func HandleResponse(handlerCtx handler.HandlerContext, response internaldto.Exec internaldto.OutputContext{ RuntimeContext: handlerCtx.GetRuntimeContext(), Result: sqlResult, + Query: handlerCtx.GetRawQuery(), + StartTime: handlerCtx.GetQueryStartTime(), }, ) if outputWriter == nil || err != nil { diff --git a/pkg/sink/otel.go b/pkg/sink/otel.go index 7608feb0..cdae697a 100644 --- a/pkg/sink/otel.go +++ b/pkg/sink/otel.go @@ -1,6 +1,7 @@ package sink import ( + "bytes" "context" "crypto/rand" "encoding/hex" @@ -222,25 +223,37 @@ func genericOTelLogRecord(payload any, now time.Time) (OTelLogRecord, error) { return OTelLogRecord{}, fmt.Errorf("otel sink: marshal payload: %w", err) } record := OTelLogRecord{Time: now, Severity: OTelSeverityInfo, Body: string(raw)} + // Not a JSON object (scalar or array): the body alone carries it. + record.Attributes, _ = OTelAttributesFromJSON(raw) + return record, nil +} + +// OTelAttributesFromJSON maps a JSON object's top-level fields to typed +// attributes in key order; nested values are carried as JSON strings and +// nulls as empty strings (dropped on encode). ok is false when raw is not +// an object. +func OTelAttributesFromJSON(raw []byte) ([]OTelAttribute, bool) { var fields map[string]json.RawMessage - if unmarshalErr := json.Unmarshal(raw, &fields); unmarshalErr != nil { - // Not a JSON object (scalar or array): the body alone carries it. - return record, nil //nolint:nilerr // intentional: non-object payloads are still valid records + if err := json.Unmarshal(raw, &fields); err != nil { + return nil, false } keys := make([]string, 0, len(fields)) for k := range fields { keys = append(keys, k) } sort.Strings(keys) + attrs := make([]OTelAttribute, 0, len(keys)) for _, k := range keys { - record.Attributes = append(record.Attributes, otelAttributeFromJSON(k, fields[k])) + attrs = append(attrs, otelAttributeFromJSON(k, fields[k])) } - return record, nil + return attrs, true } func otelAttributeFromJSON(key string, raw json.RawMessage) OTelAttribute { + dec := json.NewDecoder(bytes.NewReader(raw)) + dec.UseNumber() var v any - if err := json.Unmarshal(raw, &v); err != nil { + if err := dec.Decode(&v); err != nil { return OTelString(key, string(raw)) } switch t := v.(type) { @@ -248,11 +261,18 @@ func otelAttributeFromJSON(key string, raw json.RawMessage) OTelAttribute { return OTelString(key, t) case bool: return OTelBool(key, t) - case float64: - if t == float64(int64(t)) { - return OTelInt(key, int64(t)) + case json.Number: + if i, intErr := t.Int64(); intErr == nil { + return OTelInt(key, i) + } + f, floatErr := t.Float64() + if floatErr != nil { + return OTelString(key, t.String()) + } + if f == float64(int64(f)) { + return OTelInt(key, int64(f)) } - return OTelDouble(key, t) + return OTelDouble(key, f) case nil: return OTelString(key, "") default: diff --git a/test/python/stackql_test_tooling/StackQLInterfaces.py b/test/python/stackql_test_tooling/StackQLInterfaces.py index 2dd8b0b2..2c83fd55 100644 --- a/test/python/stackql_test_tooling/StackQLInterfaces.py +++ b/test/python/stackql_test_tooling/StackQLInterfaces.py @@ -1080,6 +1080,63 @@ def _parse_jsonl(self, payload :str, label :str) -> list: rows.append(json.dumps(parsed, sort_keys=True)) return sorted(rows) + @keyword + def should_stackql_exec_otel_snapshot( + self, + stackql_exe :str, + okta_secret_str :str, + github_secret_str :str, + k8s_secret_str :str, + registry_cfg :RegistryCfg, + auth_cfg_str :str, + sql_backend_cfg_str :str, + query :str, + expected_row_count :int, + *args, + **cfg + ) -> typing.List[str]: + """ + Run with `--output otel` and check the snapshot invariants (issue #738): + one LogsData per row plus one completion record, every record sharing the + snapshot instant, snapshot id and span. Returns the row fingerprints so a + repeat run can be compared. + """ + result = self._run_stackql_exec_command( + stackql_exe, okta_secret_str, github_secret_str, k8s_secret_str, + registry_cfg, auth_cfg_str, sql_backend_cfg_str, query, '-o=otel', *args, **cfg + ) + records = [] + for line in result.stdout.splitlines(): + if not line.strip(): + continue + scope_logs = json.loads(line)['resourceLogs'][0]['scopeLogs'][0] + if scope_logs['scope']['name'] != 'github.com/stackql/stackql/internal/stackql/output': + raise Exception(f'unexpected scope: {scope_logs["scope"]}') + for rec in scope_logs['logRecords']: + rec['attrs'] = {a['key']: next(iter(a['value'].values())) for a in rec['attributes']} + records.append(rec) + rows = [r for r in records if 'stackql.row.index' in r['attrs']] + completions = [r for r in records if r['attrs'].get('stackql.snapshot.complete') is True] + if len(rows) != int(expected_row_count) or len(completions) != 1 or len(records) != len(rows) + 1: + raise Exception( + f'expected {expected_row_count} row records and one completion record, ' + f'got {len(rows)} rows and {len(completions)} completions in {len(records)} records; ' + f'stderr: {(result.stderr or "").strip()[-2000:]}' + ) + for key in ('timeUnixNano', 'spanId', 'traceId'): + if len({r[key] for r in records}) != 1: + raise Exception(f'{key} differs across the records of one statement') + if len({r['attrs']['stackql.snapshot.id'] for r in records}) != 1: + raise Exception('stackql.snapshot.id differs across the records of one statement') + if [r['attrs']['stackql.row.index'] for r in rows] != [str(i) for i in range(len(rows))]: + raise Exception(f'row indices are not sequential: {[r["attrs"]["stackql.row.index"] for r in rows]}') + if completions[0]['attrs']['stackql.rows_returned'] != str(len(rows)): + raise Exception(f'completion rows_returned {completions[0]["attrs"]["stackql.rows_returned"]} != {len(rows)}') + fingerprints = [r['attrs']['stackql.row.fingerprint'] for r in rows] + if any(not fp.startswith('sha256:') or len(fp) != 71 for fp in fingerprints): + raise Exception(f'malformed fingerprints: {fingerprints}') + return fingerprints + @keyword def should_stackql_exec_inline_equal_both_streams( self, diff --git a/test/robot/functional/mcp.robot b/test/robot/functional/mcp.robot index 815fb72f..2cdeb270 100644 --- a/test/robot/functional/mcp.robot +++ b/test/robot/functional/mcp.robot @@ -328,6 +328,27 @@ PG Server Show Version Extended Should Match Regexp ${psql_client_result.stdout} \\d+\\.\\d+\\.\\d+ Should Be Equal As Integers ${psql_client_result.rc} 0 +PG Server Show Contributors + [Documentation] Issue #320: the embedded leaderboard is served over the + ... wire protocol from the same implementation. + Pass Execution If "%{IS_SKIP_MCP_TEST=false}" == "true" Some platforms do not have the MCP client available + Sleep 5s + ${posixInput} = Catenate + ... "${PSQL_EXE}" -d postgres://stackql:stackql@127.0.0.1:5665 -c + ... "SHOW EXTENDED CONTRIBUTORS;" + ${windowsInput} = Catenate + ... & ${posixInput} + ${input} = Set Variable If "${IS_WINDOWS}" == "1" ${windowsInput} ${posixInput} + ${shellExe} = Set Variable If "${IS_WINDOWS}" == "1" powershell sh + ${psql_client_result}= Run Process + ... ${shellExe} \-c ${input} + ... stdout=${CURDIR}${/}tmp${/}PG-Server-Show-Contributors-psql.txt + ... stderr=${CURDIR}${/}tmp${/}PG-Server-Show-Contributors-psql-stderr.txt + Should Contain ${psql_client_result.stdout} contributor + Should Contain ${psql_client_result.stdout} contributions + Should Match Regexp ${psql_client_result.stdout} (?s)general-kroll-4-life.*jeffreyaven + Should Be Equal As Integers ${psql_client_result.rc} 0 + MCP HTTP Server Query Tool Pass Execution If "%{IS_SKIP_MCP_TEST=false}" == "true" Some platforms do not have the MCP client available Sleep 5s diff --git a/test/robot/functional/stackql_from_cmd_line.robot b/test/robot/functional/stackql_from_cmd_line.robot index 0ded8ec8..b69eeb05 100644 --- a/test/robot/functional/stackql_from_cmd_line.robot +++ b/test/robot/functional/stackql_from_cmd_line.robot @@ -810,6 +810,26 @@ Show Version Extended ${captured}= Get File ${CURDIR}/tmp/Show-Version-Extended.tmp Should Match Regexp ${captured} \\d+\\.\\d+\\.\\d+ +Show Contributors + [Documentation] Issue #320: SHOW CONTRIBUTORS renders the embedded + ... leaderboard, contributions descending, founder on top. + Should StackQL Exec Contain + ... SHOW CONTRIBUTORS; + ... contributor + ... stdout=${CURDIR}/tmp/Show-Contributors.tmp + ${captured}= Get File ${CURDIR}/tmp/Show-Contributors.tmp + Should Match Regexp ${captured} (?s)general-kroll-4-life.*jeffreyaven + Should Not Contain ${captured} contributions + +Show Contributors Extended + [Documentation] SHOW EXTENDED CONTRIBUTORS adds the contributions count. + Should StackQL Exec Contain + ... SHOW EXTENDED CONTRIBUTORS; + ... contributions + ... stdout=${CURDIR}/tmp/Show-Contributors-Extended.tmp + ${captured}= Get File ${CURDIR}/tmp/Show-Contributors-Extended.tmp + Should Match Regexp ${captured} (?s)general-kroll-4-life.*jeffreyaven + *** Keywords *** Should StackQL Exec Equal [Arguments] ${_EXEC_CMD_STR} ${_EXEC_CMD_EXPECTED_OUTPUT} @{varargs} &{kwargs} diff --git a/test/robot/functional/stackql_mocked_from_cmd_line.robot b/test/robot/functional/stackql_mocked_from_cmd_line.robot index 470384eb..48eed218 100644 --- a/test/robot/functional/stackql_mocked_from_cmd_line.robot +++ b/test/robot/functional/stackql_mocked_from_cmd_line.robot @@ -11017,3 +11017,57 @@ Unstable Google Kms Key Rings Jsonl Row Set Matches Expectation ... --preview\=${preview} ... stdout=${CURDIR}${/}tmp${/}Unstable-Google-Kms-Key-Rings.tmp ... stderr=${CURDIR}${/}tmp${/}Unstable-Google-Kms-Key-Rings-stderr.tmp + +OTel Output Emits One Record Per Row Plus Completion + [Documentation] Issue #738: --output otel writes one OTLP/JSON LogsData per + ... row and a completion record. Every record of the statement + ... shares the snapshot instant, snapshot id and span, and an + ... identical re-run yields identical row fingerprints. + ${query} = Catenate SEPARATOR=${SPACE} + ... select login, contributions from github.repos.contributors + ... where owner = 'dummyorg' and repo = 'dummyapp.io'; + ${first} = Should StackQL Exec OTel Snapshot + ... ${STACKQL_EXE} + ... ${OKTA_SECRET_STR} + ... ${GITHUB_SECRET_STR} + ... ${K8S_SECRET_STR} + ... ${REGISTRY_NO_VERIFY_CFG_STR} + ... ${AUTH_CFG_STR} + ... ${SQL_BACKEND_CFG_STR_CANONICAL} + ... ${query} + ... 8 + ... stdout=${CURDIR}${/}tmp${/}OTel-Output-Rows-First.tmp + ... stderr=${CURDIR}${/}tmp${/}OTel-Output-Rows-First-stderr.tmp + ${second} = Should StackQL Exec OTel Snapshot + ... ${STACKQL_EXE} + ... ${OKTA_SECRET_STR} + ... ${GITHUB_SECRET_STR} + ... ${K8S_SECRET_STR} + ... ${REGISTRY_NO_VERIFY_CFG_STR} + ... ${AUTH_CFG_STR} + ... ${SQL_BACKEND_CFG_STR_CANONICAL} + ... ${query} + ... 8 + ... stdout=${CURDIR}${/}tmp${/}OTel-Output-Rows-Second.tmp + ... stderr=${CURDIR}${/}tmp${/}OTel-Output-Rows-Second-stderr.tmp + Should Be Equal ${first} ${second} + +OTel Output Zero Row Statement Emits Only Completion + [Documentation] Issue #738: a statement returning no rows still emits its + ... completion record, so an absent resource is visible as + ... drift rather than as silence. + ${query} = Catenate SEPARATOR=${SPACE} + ... select login from github.repos.contributors + ... where owner = 'dummyorg' and repo = 'dummyapp.io' and login = 'nobody'; + Should StackQL Exec OTel Snapshot + ... ${STACKQL_EXE} + ... ${OKTA_SECRET_STR} + ... ${GITHUB_SECRET_STR} + ... ${K8S_SECRET_STR} + ... ${REGISTRY_NO_VERIFY_CFG_STR} + ... ${AUTH_CFG_STR} + ... ${SQL_BACKEND_CFG_STR_CANONICAL} + ... ${query} + ... 0 + ... stdout=${CURDIR}${/}tmp${/}OTel-Output-Zero-Rows.tmp + ... stderr=${CURDIR}${/}tmp${/}OTel-Output-Zero-Rows-stderr.tmp From ef993850df0e5419adfc2a835d9875463a6ab483 Mon Sep 17 00:00:00 2001 From: Jeffrey Aven Date: Mon, 7 Sep 2026 21:41:47 +1000 Subject: [PATCH 2/2] robot: assert known otel attribute values; add otel sibling of the preview streaming scenario Should StackQL Exec OTel Snapshot now takes the expected values as JSON (context on every record, rows in order, completion) and decodes OTLP values to native types; dynamic ids, timestamps and fingerprints stay unlisted. Adds Preview Rows Are Emitted Throughout The Run As OTel Records: 79 rows plus completion streamed incrementally, as for jsonl. Co-Authored-By: Claude Fable 5.1 --- .../stackql_test_tooling/StackQLInterfaces.py | 36 ++++++++--- .../stackql_mocked_from_cmd_line.robot | 61 +++++++++++++++++-- 2 files changed, 83 insertions(+), 14 deletions(-) diff --git a/test/python/stackql_test_tooling/StackQLInterfaces.py b/test/python/stackql_test_tooling/StackQLInterfaces.py index 2c83fd55..f6466976 100644 --- a/test/python/stackql_test_tooling/StackQLInterfaces.py +++ b/test/python/stackql_test_tooling/StackQLInterfaces.py @@ -1091,16 +1091,20 @@ def should_stackql_exec_otel_snapshot( auth_cfg_str :str, sql_backend_cfg_str :str, query :str, - expected_row_count :int, + expected :str, *args, **cfg ) -> typing.List[str]: """ Run with `--output otel` and check the snapshot invariants (issue #738): one LogsData per row plus one completion record, every record sharing the - snapshot instant, snapshot id and span. Returns the row fingerprints so a - repeat run can be compared. + snapshot instant, snapshot id and span. `expected` is a JSON object with + the known attribute values: `context` is verified on every record, `rows` + (in order) on the row records and `completion` on the completion record; + dynamic values (ids, timestamps, fingerprints) are simply left unlisted. + Returns the row fingerprints so a repeat run can be compared. """ + expectation = json.loads(expected) result = self._run_stackql_exec_command( stackql_exe, okta_secret_str, github_secret_str, k8s_secret_str, registry_cfg, auth_cfg_str, sql_backend_cfg_str, query, '-o=otel', *args, **cfg @@ -1113,30 +1117,46 @@ def should_stackql_exec_otel_snapshot( if scope_logs['scope']['name'] != 'github.com/stackql/stackql/internal/stackql/output': raise Exception(f'unexpected scope: {scope_logs["scope"]}') for rec in scope_logs['logRecords']: - rec['attrs'] = {a['key']: next(iter(a['value'].values())) for a in rec['attributes']} + rec['attrs'] = {a['key']: self._otel_value(a['value']) for a in rec['attributes']} records.append(rec) rows = [r for r in records if 'stackql.row.index' in r['attrs']] completions = [r for r in records if r['attrs'].get('stackql.snapshot.complete') is True] - if len(rows) != int(expected_row_count) or len(completions) != 1 or len(records) != len(rows) + 1: + expected_rows = expectation.get('rows', []) + if len(rows) != len(expected_rows) or len(completions) != 1 or len(records) != len(rows) + 1: raise Exception( - f'expected {expected_row_count} row records and one completion record, ' + f'expected {len(expected_rows)} row records and one completion record, ' f'got {len(rows)} rows and {len(completions)} completions in {len(records)} records; ' f'stderr: {(result.stderr or "").strip()[-2000:]}' ) + for i, (rec, want) in enumerate(zip(rows + completions, expected_rows + [expectation.get('completion', {})])): + label = f'row {i}' if i < len(rows) else 'completion' + for key, value in {**expectation.get('context', {}), **want}.items(): + if key not in rec['attrs']: + raise Exception(f'{label}: attribute {key!r} missing; have {sorted(rec["attrs"])}') + if rec['attrs'][key] != value: + raise Exception(f'{label}: attribute {key!r} = {rec["attrs"][key]!r}, want {value!r}') for key in ('timeUnixNano', 'spanId', 'traceId'): if len({r[key] for r in records}) != 1: raise Exception(f'{key} differs across the records of one statement') if len({r['attrs']['stackql.snapshot.id'] for r in records}) != 1: raise Exception('stackql.snapshot.id differs across the records of one statement') - if [r['attrs']['stackql.row.index'] for r in rows] != [str(i) for i in range(len(rows))]: + if [r['attrs']['stackql.row.index'] for r in rows] != list(range(len(rows))): raise Exception(f'row indices are not sequential: {[r["attrs"]["stackql.row.index"] for r in rows]}') - if completions[0]['attrs']['stackql.rows_returned'] != str(len(rows)): + if completions[0]['attrs']['stackql.rows_returned'] != len(rows): raise Exception(f'completion rows_returned {completions[0]["attrs"]["stackql.rows_returned"]} != {len(rows)}') fingerprints = [r['attrs']['stackql.row.fingerprint'] for r in rows] if any(not fp.startswith('sha256:') or len(fp) != 71 for fp in fingerprints): raise Exception(f'malformed fingerprints: {fingerprints}') return fingerprints + @staticmethod + def _otel_value(any_value :dict): + """Decode an OTLP/JSON AnyValue to the natural python type.""" + kind, raw = next(iter(any_value.items())) + if kind == 'intValue': + return int(raw) + return raw + @keyword def should_stackql_exec_inline_equal_both_streams( self, diff --git a/test/robot/functional/stackql_mocked_from_cmd_line.robot b/test/robot/functional/stackql_mocked_from_cmd_line.robot index 48eed218..5c7182d2 100644 --- a/test/robot/functional/stackql_mocked_from_cmd_line.robot +++ b/test/robot/functional/stackql_mocked_from_cmd_line.robot @@ -10881,6 +10881,36 @@ Preview Rows Are Emitted Throughout The Run ... stdout=${CURDIR}${/}tmp${/}Preview-Rows-Are-Emitted-Throughout-The-Run.tmp ... stderr=${CURDIR}${/}tmp${/}Preview-Rows-Are-Emitted-Throughout-The-Run-stderr.tmp +Preview Rows Are Emitted Throughout The Run As OTel Records + [Documentation] Issue #738: the otel writer streams like jsonl, one + ... LogsData per row as the upstream produces it, then the + ... completion record; 79 rows plus completion is 80 lines. + [Setup] Write Gcp Service Account ${OMNISDK_MOCK_GCP_SA_HOST} + [Teardown] Remove Preview Mock Environment + ${mock} = Set Variable {"scheme":"http","host":"${LOCAL_HOST_ALIAS}","port":"${MOCKSERVER_PORT_OMNISDK}"} + ${preview} = Catenate SEPARATOR= + ... {"batchSize":10,"flushInterval":"50ms","endpoint":{"aws.s3":${mock}}} + Set Environment Variable AWS_ACCESS_KEY_ID AK + Set Environment Variable AWS_SECRET_ACCESS_KEY SK + ${query} = Catenate SEPARATOR=${SPACE} + ... select * from stackql_preview.audit.aws_s3_buckets + ... where region = 'us-east-1' and method = 'list'; + Should StackQL Exec Stream Incrementally + ... ${STACKQL_EXE} + ... ${OKTA_SECRET_STR} + ... ${GITHUB_SECRET_STR} + ... ${K8S_SECRET_STR} + ... ${REGISTRY_NO_VERIFY_CFG_STR} + ... ${AUTH_CFG_STR} + ... ${SQL_BACKEND_CFG_STR_CANONICAL} + ... ${query} + ... 80 + ... 0.2 + ... \-o\=otel + ... --preview\=${preview} + ... stdout=${CURDIR}${/}tmp${/}Preview-Rows-Are-Emitted-Throughout-The-Run-As-OTel-Records.tmp + ... stderr=${CURDIR}${/}tmp${/}Preview-Rows-Are-Emitted-Throughout-The-Run-As-OTel-Records-stderr.tmp + Preview Order By Is Refused Rather Than Ignored [Documentation] A streamed relation never reaches the SQL backend, so ... ORDER BY cannot be applied. It is refused outright rather @@ -11020,12 +11050,26 @@ Unstable Google Kms Key Rings Jsonl Row Set Matches Expectation OTel Output Emits One Record Per Row Plus Completion [Documentation] Issue #738: --output otel writes one OTLP/JSON LogsData per - ... row and a completion record. Every record of the statement - ... shares the snapshot instant, snapshot id and span, and an - ... identical re-run yields identical row fingerprints. + ... row and a completion record with the row values as typed + ... attributes. Every record of the statement shares the + ... snapshot instant, snapshot id and span, and an identical + ... re-run yields identical row fingerprints. ${query} = Catenate SEPARATOR=${SPACE} ... select login, contributions from github.repos.contributors ... where owner = 'dummyorg' and repo = 'dummyapp.io'; + ${expected} = Catenate SEPARATOR= + ... {"context":{"stackql.provider":"github","stackql.service":"repos","stackql.resource":"contributors", + ... "stackql.query":"${query}"}, + ... "rows":[ + ... {"stackql.row.index":0,"login":"octo-lead","contributions":"500"}, + ... {"stackql.row.index":1,"login":"octo-dev-one","contributions":"120"}, + ... {"stackql.row.index":2,"login":"octo-dev-two","contributions":"120"}, + ... {"stackql.row.index":3,"login":"octo-maintainer","contributions":"80"}, + ... {"stackql.row.index":4,"login":"octo-helper-one","contributions":"45"}, + ... {"stackql.row.index":5,"login":"octo-helper-two","contributions":"45"}, + ... {"stackql.row.index":6,"login":"octo-casual","contributions":"12"}, + ... {"stackql.row.index":7,"login":"octo-drive-by","contributions":"3"}], + ... "completion":{"stackql.snapshot.complete":true,"stackql.rows_returned":8}} ${first} = Should StackQL Exec OTel Snapshot ... ${STACKQL_EXE} ... ${OKTA_SECRET_STR} @@ -11035,7 +11079,7 @@ OTel Output Emits One Record Per Row Plus Completion ... ${AUTH_CFG_STR} ... ${SQL_BACKEND_CFG_STR_CANONICAL} ... ${query} - ... 8 + ... ${expected} ... stdout=${CURDIR}${/}tmp${/}OTel-Output-Rows-First.tmp ... stderr=${CURDIR}${/}tmp${/}OTel-Output-Rows-First-stderr.tmp ${second} = Should StackQL Exec OTel Snapshot @@ -11047,7 +11091,7 @@ OTel Output Emits One Record Per Row Plus Completion ... ${AUTH_CFG_STR} ... ${SQL_BACKEND_CFG_STR_CANONICAL} ... ${query} - ... 8 + ... ${expected} ... stdout=${CURDIR}${/}tmp${/}OTel-Output-Rows-Second.tmp ... stderr=${CURDIR}${/}tmp${/}OTel-Output-Rows-Second-stderr.tmp Should Be Equal ${first} ${second} @@ -11059,6 +11103,11 @@ OTel Output Zero Row Statement Emits Only Completion ${query} = Catenate SEPARATOR=${SPACE} ... select login from github.repos.contributors ... where owner = 'dummyorg' and repo = 'dummyapp.io' and login = 'nobody'; + ${expected} = Catenate SEPARATOR= + ... {"context":{"stackql.provider":"github","stackql.service":"repos","stackql.resource":"contributors", + ... "stackql.query":"${query}"}, + ... "rows":[], + ... "completion":{"stackql.snapshot.complete":true,"stackql.rows_returned":0}} Should StackQL Exec OTel Snapshot ... ${STACKQL_EXE} ... ${OKTA_SECRET_STR} @@ -11068,6 +11117,6 @@ OTel Output Zero Row Statement Emits Only Completion ... ${AUTH_CFG_STR} ... ${SQL_BACKEND_CFG_STR_CANONICAL} ... ${query} - ... 0 + ... ${expected} ... stdout=${CURDIR}${/}tmp${/}OTel-Output-Zero-Rows.tmp ... stderr=${CURDIR}${/}tmp${/}OTel-Output-Zero-Rows-stderr.tmp