diff --git a/docker-compose-testing.yml b/docker-compose-testing.yml index 239d5543..d908df29 100644 --- a/docker-compose-testing.yml +++ b/docker-compose-testing.yml @@ -155,6 +155,7 @@ services: - "1071:1071/tcp" environment: IS_DOCKER: "${IS_DOCKER:-false}" + OMNISDK_MOCK_DELAY_MS: "50" command: - bash - -c diff --git a/internal/stackql/cmd/root.go b/internal/stackql/cmd/root.go index 8a1032c5..a313a336 100644 --- a/internal/stackql/cmd/root.go +++ b/internal/stackql/cmd/root.go @@ -27,6 +27,7 @@ import ( "github.com/stackql/stackql/internal/stackql/buildinfo" "github.com/stackql/stackql/internal/stackql/config" "github.com/stackql/stackql/internal/stackql/envfile" + "github.com/stackql/stackql/internal/stackql/intrinsic" "github.com/magiconair/properties" "github.com/spf13/cobra" @@ -69,6 +70,12 @@ var ( replicateCtrMgr bool = false //nolint:unused // TODO: investigate and test then remove if possible ) +// previewCfgRaw is the raw --preview argument; cobra binds it here and +// initConfig hands it to the intrinsic package once. +// +//nolint:gochecknoglobals // cobra binds flags to package scope +var previewCfgRaw string + // rootCmd represents the base command when called without any subcommands. // //nolint:gochecknoglobals // global vars are a pattern for this lib @@ -148,6 +155,7 @@ func init() { rootCmd.PersistentFlags().StringVar(&runtimeCtx.StoreTxnCfgRaw, dto.StoreTxnCfgRawKey, "{}", "JSON / YAML string representing Txn store config") rootCmd.PersistentFlags().StringVar(&runtimeCtx.GCCfgRaw, dto.GCCfgRawKey, "{}", "JSON / YAML string representing GC config") rootCmd.PersistentFlags().StringVar(&runtimeCtx.ACIDCfgRaw, dto.ACIDCfgRawKey, "{}", "JSON / YAML string representing ACID config") + rootCmd.PersistentFlags().StringVar(&previewCfgRaw, intrinsic.CfgRawKey, "{}", "JSON string configuring the "+intrinsic.ProviderName+" provider backend; keys: batchSize, flushInterval, endpoint") rootCmd.PersistentFlags().StringVar(&runtimeCtx.SessionCtxRaw, dto.SessionCtxKey, "{}", "JSON / YAML string representing session config") rootCmd.PersistentFlags().IntVar(&runtimeCtx.APIRequestTimeout, dto.APIRequestTimeoutKey, 45, "API request timeout in seconds, 0 for no timeout.") //nolint:mnd // TODO: investigate rootCmd.PersistentFlags().StringVar(&dummyString, dto.ColorSchemeKey, "", "DEPRECATED: color schems no longer active") @@ -242,6 +250,8 @@ func mergeConfigFromFile(runtimeCtx *dto.RuntimeCtx, flagSet pflag.FlagSet) { func initConfig() { mergeConfigFromFile(&runtimeCtx, *rootCmd.PersistentFlags()) + intrinsic.Init(previewCfgRaw) + // An absent --env.file is created empty (issue #691) so packaged installs // have a credential store to populate; creation failure is non-fatal // because a missing file is already tolerated by Source. diff --git a/internal/stackql/intrinsic/intrinsic.go b/internal/stackql/intrinsic/intrinsic.go index e5a5df51..213408b6 100644 --- a/internal/stackql/intrinsic/intrinsic.go +++ b/internal/stackql/intrinsic/intrinsic.go @@ -27,6 +27,7 @@ const ( type column struct { name string + sourceName string description string dataType string } diff --git a/internal/stackql/intrinsic/omnisdk.go b/internal/stackql/intrinsic/omnisdk.go index 87b189a8..2b5e4bb6 100644 --- a/internal/stackql/intrinsic/omnisdk.go +++ b/internal/stackql/intrinsic/omnisdk.go @@ -2,16 +2,19 @@ package intrinsic import ( "context" + "encoding/json" "errors" "fmt" "io" - "os" "sort" "strconv" "strings" + "sync" + "time" "github.com/lib/pq/oid" "github.com/stackql-labs/omnisdk/pkg/omnisdk" + "github.com/stackql/any-sdk/pkg/dto" "github.com/stackql/psql-wire/pkg/sqldata" "github.com/stackql/stackql/internal/stackql/internal_data_transfer/internaldto" @@ -20,9 +23,9 @@ import ( const methodPredicate = "method" -// endpointEnvVar retargets omnisdk at a local mock. Transport configuration, so -// it stays out of the query. -const endpointEnvVar = "STACKQL_PREVIEW_ENDPOINT" +const defaultBatchSize = 100 + +const defaultFlushInterval = 50 * time.Millisecond func relationName(path string) string { return strings.ReplaceAll(path, ".", "_") @@ -145,15 +148,17 @@ func lastSegment(path string) string { } func openStream( - ctx queryContext, resourcePath string, params map[string]string) (*rowStream, error) { + ctx queryContext, resourcePath string, params map[string]string, + exprs sqlparser.SelectExprs) (*rowStream, error) { method, err := pickMethod(resourcePath, params) if err != nil { return nil, err } + input := previewCfg args := omnisdk.Args{ Params: params, - Auth: omnisdkAuth(ctx, resourcePath), - Endpoint: os.Getenv(endpointEnvVar), + Auth: omnisdkAuth(providerAuthContext(ctx, resourcePath)), + Endpoint: input.getEndpoint(), } plan, err := omnisdk.Default().New(method.Path, args) if err != nil { @@ -163,15 +168,28 @@ func openStream( if err != nil { return nil, err } - return &rowStream{rows: rows, columns: schemaColumns(method.Schema)}, nil + selected, projectionErr := projection(exprs, schemaColumns(method.Schema)) + if projectionErr != nil { + return nil, projectionErr + } + return &rowStream{ + rows: rows, + columns: selected, + batchSize: input.getBatchSize(), + flushInterval: input.getFlushInterval(), + }, nil } type rowStream struct { - rows omnisdk.Rows - columns []column - table sqldata.ISQLTable - typCfg columnFactory - done bool + rows omnisdk.Rows + batchSize int + flushInterval time.Duration + produced chan omnisdk.Row + producerOnce sync.Once + columns []column + table sqldata.ISQLTable + typCfg columnFactory + done bool } type columnFactory interface { @@ -182,15 +200,62 @@ func (rs *rowStream) Read() (sqldata.ISQLResult, error) { if rs.done { return rs.result(nil), io.EOF } - rs.done = true - var batch []omnisdk.Row - for rs.rows.Next() { - batch = append(batch, rs.rows.Row()) + rs.startProducer() + size := rs.batchSize + if size < 1 { + size = defaultBatchSize + } + batch := make([]omnisdk.Row, 0, size) + // Block for the first row, then take whatever else has arrived within the + // flush interval. A batch is therefore a cap, not a threshold: a result + // smaller than the batch still reaches the caller promptly. + row, ok := <-rs.produced + if !ok { + rs.done = true + if err := rs.rows.Err(); err != nil { + return rs.result(nil), err + } + return rs.result(nil), io.EOF + } + batch = append(batch, row) + deadline := time.After(rs.flushIntervalOrDefault()) + for len(batch) < size { + select { + case next, more := <-rs.produced: + if !more { + rs.done = true + if err := rs.rows.Err(); err != nil { + return rs.result(batch), err + } + return rs.result(batch), io.EOF + } + batch = append(batch, next) + case <-deadline: + return rs.result(batch), nil + } } - if err := rs.rows.Err(); err != nil { - return rs.result(nil), err + return rs.result(batch), nil +} + +func (rs *rowStream) flushIntervalOrDefault() time.Duration { + if rs.flushInterval <= 0 { + return defaultFlushInterval } - return rs.result(batch), io.EOF + return rs.flushInterval +} + +// startProducer pulls the cursor on its own goroutine, so a read can bound how +// long it waits for a batch to fill without abandoning rows already produced. +func (rs *rowStream) startProducer() { + rs.producerOnce.Do(func() { + rs.produced = make(chan omnisdk.Row) + go func() { + defer close(rs.produced) + for rs.rows.Next() { + rs.produced <- rs.rows.Row() + } + }() + }) } func (rs *rowStream) result(batch []omnisdk.Row) sqldata.ISQLResult { @@ -207,7 +272,7 @@ func (rs *rowStream) result(batch []omnisdk.Row) sqldata.ISQLResult { for _, row := range batch { values := make([]interface{}, 0, len(rs.columns)) for _, col := range rs.columns { - values = append(values, textValue(row[col.name])) + values = append(values, textValue(row[col.sourceKey()])) } rows = append(rows, sqldata.NewSQLRow(values)) } @@ -255,9 +320,34 @@ func selectFunc( if !ok { return nil, false } - params := equalityPredicates(node.Where) + if unsupported := unsupportedClauses(node); len(unsupported) > 0 { + return func() internaldto.ExecutorOutput { + return internaldto.NewErroneousExecutorOutput(fmt.Errorf( + "relation '%s.%s.%s' streams its rows, so %s cannot be applied; remove %s from the query", + ProviderName, auditService, relationName(resource.Path), + strings.Join(unsupported, ", "), pluralClause(len(unsupported)))) + }, true + } + _, projectionErr := projection(node.SelectExprs, schemaColumns(resource.Schema)) + if projectionErr != nil { + return func() internaldto.ExecutorOutput { + return internaldto.NewErroneousExecutorOutput(fmt.Errorf( + "relation '%s.%s.%s' streams its rows: %w", + ProviderName, auditService, relationName(resource.Path), projectionErr)) + }, true + } + params, badPredicates := equalityPredicates(node.Where) + if len(badPredicates) > 0 { + return func() internaldto.ExecutorOutput { + return internaldto.NewErroneousExecutorOutput(fmt.Errorf( + "relation '%s.%s.%s' streams its rows, so only equality predicates are applied; "+ + "%s cannot be honoured", + ProviderName, auditService, relationName(resource.Path), + strings.Join(badPredicates, ", "))) + }, true + } return func() internaldto.ExecutorOutput { - stream, err := openStream(ctx, resource.Path, params) + stream, err := openStream(ctx, resource.Path, params, node.SelectExprs) if err != nil { return internaldto.NewErroneousExecutorOutput(err) } @@ -271,10 +361,78 @@ func selectFunc( }, true } -func equalityPredicates(where *sqlparser.Where) map[string]string { +// unsupportedClauses names the parts of a select that the streaming path cannot +// honour. Rows never reach the SQL backend, so anything the backend would have +// applied has to be refused rather than quietly dropped. +func unsupportedClauses(node *sqlparser.Select) []string { + var out []string + if len(node.OrderBy) > 0 { + out = append(out, "ORDER BY") + } + if len(node.GroupBy) > 0 { + out = append(out, "GROUP BY") + } + if node.Having != nil { + out = append(out, "HAVING") + } + if node.Distinct { + out = append(out, "DISTINCT") + } + if node.Limit != nil { + out = append(out, "LIMIT") + } + return out +} + +// projection resolves the select list against the relation's columns. A star +// selects them all; named columns are emitted in the order asked for. Anything +// else - an aggregate, a function, a literal - needs the SQL backend, which +// streamed rows never reach, so it is refused rather than quietly dropped. +func projection(exprs sqlparser.SelectExprs, available []column) ([]column, error) { + if len(exprs) == 1 { + if _, isStar := exprs[0].(*sqlparser.StarExpr); isStar { + return available, nil + } + } + byName := make(map[string]column, len(available)) + for _, col := range available { + byName[strings.ToLower(col.name)] = col + } + out := make([]column, 0, len(exprs)) + for _, expr := range exprs { + aliased, isAliased := expr.(*sqlparser.AliasedExpr) + if !isAliased { + return nil, fmt.Errorf("'%s' cannot be applied to a streamed relation", sqlparser.String(expr)) + } + colName, isCol := aliased.Expr.(*sqlparser.ColName) + if !isCol { + return nil, fmt.Errorf("'%s' cannot be applied to a streamed relation", sqlparser.String(expr)) + } + found, ok := byName[strings.ToLower(colName.Name.GetRawVal())] + if !ok { + return nil, fmt.Errorf("column '%s' does not exist", colName.Name.GetRawVal()) + } + if alias := aliased.As.GetRawVal(); alias != "" { + found.name = alias + found.sourceName = colName.Name.GetRawVal() + } + out = append(out, found) + } + return out, nil +} + +func pluralClause(n int) string { + if n == 1 { + return "it" + } + return "them" +} + +func equalityPredicates(where *sqlparser.Where) (map[string]string, []string) { params := map[string]string{} + var bad []string if where == nil { - return params + return params, bad } var walk func(expr sqlparser.Expr) walk = func(expr sqlparser.Expr) { @@ -283,19 +441,19 @@ func equalityPredicates(where *sqlparser.Where) map[string]string { walk(node.Left) walk(node.Right) case *sqlparser.ComparisonExpr: - if node.Operator != sqlparser.EqualStr { - return - } col, isCol := node.Left.(*sqlparser.ColName) val, isVal := node.Right.(*sqlparser.SQLVal) - if !isCol || !isVal { + if !isCol || !isVal || node.Operator != sqlparser.EqualStr { + bad = append(bad, sqlparser.String(expr)) return } params[col.Name.GetRawVal()] = string(val.Val) + default: + bad = append(bad, sqlparser.String(expr)) } } walk(where.Expr) - return params + return params, bad } type relationMethod struct { @@ -353,14 +511,23 @@ var cloudProviders = map[string]string{ //nolint:gochecknoglobals // fixed mappi "azure": "azure", } -func omnisdkAuth(ctx queryContext, resourcePath string) *omnisdk.Auth { +// providerAuthContext is the stackql auth context for the cloud behind a +// resource. It carries both the credentials and the tuning values. +func providerAuthContext(ctx queryContext, resourcePath string) *dto.AuthCtx { cloud, _, _ := strings.Cut(resourcePath, ".") providerName, ok := cloudProviders[cloud] if !ok { return nil } authCtx, err := ctx.GetAuthContext(providerName) - if err != nil || authCtx == nil { + if err != nil { + return nil + } + return authCtx +} + +func omnisdkAuth(authCtx *dto.AuthCtx) *omnisdk.Auth { + if authCtx == nil { return nil } auth := &omnisdk.Auth{ @@ -470,3 +637,87 @@ func textValue(value any) any { return fmt.Sprintf("%v", typed) } } + +// backendInput carries the tunables for a streaming run. It is read once, at +// construction, and only read thereafter. +type backendInput interface { + getBatchSize() int + getEndpoint() string + getFlushInterval() time.Duration +} + +type standardBackendInput struct { + batchSize int + endpoint string + flushInterval time.Duration +} + +// previewCfg is the parsed --preview argument. Cobra binds the raw string in +// internal/stackql/cmd, which calls Init exactly once; nothing else writes it. +// +//nolint:gochecknoglobals // set once from the CLI, read thereafter +var previewCfg = newBackendInput(previewCfgDTO{}) + +// CfgRawKey is the CLI argument that configures this provider's backend. +const CfgRawKey = "preview" + +// previewCfgDTO is the wire shape of the --preview argument. Endpoint accepts +// either form omnisdk does: a base URL for every service, or an object of +// service to override. Both ride through as the string omnisdk parses. +type previewCfgDTO struct { + BatchSize int `json:"batchSize"` + FlushInterval string `json:"flushInterval"` + Endpoint json.RawMessage `json:"endpoint"` +} + +func (c previewCfgDTO) endpoint() string { + if len(c.Endpoint) == 0 { + return "" + } + var asURL string + if err := json.Unmarshal(c.Endpoint, &asURL); err == nil { + return asURL + } + return string(c.Endpoint) +} + +// Init records the --preview argument. It is called once, from cmd, before any +// query runs. +func Init(raw string) { + var cfg previewCfgDTO + if strings.TrimSpace(raw) != "" { + //nolint:errcheck // a malformed argument leaves the defaults in place + _ = json.Unmarshal([]byte(raw), &cfg) + } + previewCfg = newBackendInput(cfg) +} + +func newBackendInput(cfg previewCfgDTO) backendInput { + rv := &standardBackendInput{ + batchSize: defaultBatchSize, + endpoint: cfg.endpoint(), + flushInterval: defaultFlushInterval, + } + if cfg.BatchSize > 0 { + rv.batchSize = cfg.BatchSize + } + if parsed, err := time.ParseDuration(cfg.FlushInterval); err == nil && parsed > 0 { + rv.flushInterval = parsed + } + return rv +} + +func (b *standardBackendInput) getBatchSize() int { return b.batchSize } + +func (b *standardBackendInput) getEndpoint() string { return b.endpoint } + +func (b *standardBackendInput) getFlushInterval() time.Duration { return b.flushInterval } + +// sourceKey is the row key a column reads from: its own name, unless an alias +// renamed it. +func (c column) sourceKey() string { + if c.sourceName != "" { + return c.sourceName + } + return c.name +} diff --git a/internal/stackql/intrinsic/omnisdk_test.go b/internal/stackql/intrinsic/omnisdk_test.go index 8a6c2f80..b06bb3e7 100644 --- a/internal/stackql/intrinsic/omnisdk_test.go +++ b/internal/stackql/intrinsic/omnisdk_test.go @@ -5,7 +5,10 @@ import ( "errors" "fmt" "io" + "strings" + "sync" "testing" + "time" "github.com/lib/pq/oid" "github.com/stackql-labs/omnisdk/pkg/omnisdk" @@ -186,3 +189,117 @@ func TestReportedTypeUsesStackqlVocabulary(t *testing.T) { } } } + +// pagedRows serves a first page, then blocks until released, then a second. +// It models an upstream whose next page is slow to arrive. +type pagedRows struct { + first []omnisdk.Row + second []omnisdk.Row + release chan struct{} + idx int + released bool +} + +func (p *pagedRows) Next() bool { + if p.idx < len(p.first) { + p.idx++ + return true + } + if !p.released { + <-p.release + p.released = true + } + if p.idx < len(p.first)+len(p.second) { + p.idx++ + return true + } + return false +} + +func (p *pagedRows) Row() omnisdk.Row { + if p.idx <= len(p.first) { + return p.first[p.idx-1] + } + return p.second[p.idx-len(p.first)-1] +} + +func (p *pagedRows) Err() error { return nil } +func (p *pagedRows) Close() error { return nil } + +// The first page must reach the output writer before the second page is even +// available upstream. Anything that buffers the whole cursor fails this. +func TestRowsReachOutputBeforeNextPage(t *testing.T) { + first := []omnisdk.Row{{"a": "p1-r1"}, {"a": "p1-r2"}, {"a": "p1-r3"}} + second := []omnisdk.Row{{"a": "p2-r1"}} + cursor := &pagedRows{first: first, second: second, release: make(chan struct{})} + stream := &rowStream{ + rows: cursor, + batchSize: 1, + columns: []column{{name: "a"}}, + table: sqldata.NewSQLTable(0, "t"), + typCfg: fakeColumnFactory{}, + } + + var mu sync.Mutex + var written []string + sink := writerFunc(func(b []byte) (int, error) { + mu.Lock() + defer mu.Unlock() + written = append(written, string(b)) + return len(b), nil + }) + + var errBuf bytes.Buffer + writer, err := output.GetOutputWriter(sink, &errBuf, internaldto.OutputContext{ + RuntimeContext: dto.RuntimeCtx{OutputFormat: "jsonl", Delimiter: ","}, + Result: stream, + }) + if err != nil { + t.Fatalf("writer: %v", err) + } + + done := make(chan error, 1) + go func() { done <- writer.Write(stream) }() + + // The whole first page must be written while the cursor is still stalled. + deadline := time.Now().Add(5 * time.Second) + for { + mu.Lock() + count := len(written) + mu.Unlock() + if count >= len(first) { + break + } + if time.Now().After(deadline) { + t.Fatalf("only %d rows reached the writer before the next page; rows are being buffered", count) + } + time.Sleep(5 * time.Millisecond) + } + + mu.Lock() + beforeRelease := strings.Join(written, "") + mu.Unlock() + if strings.Contains(beforeRelease, "p2-r1") { + t.Fatal("second page appeared before it was released") + } + for _, want := range []string{"p1-r1", "p1-r2", "p1-r3"} { + if !strings.Contains(beforeRelease, want) { + t.Fatalf("row %s did not reach the writer before the next page: %q", want, beforeRelease) + } + } + + close(cursor.release) + if writeErr := <-done; writeErr != nil { + t.Fatalf("write: %v", writeErr) + } + mu.Lock() + all := strings.Join(written, "") + mu.Unlock() + if !strings.Contains(all, "p2-r1") { + t.Fatalf("second page missing from output: %q", all) + } +} + +type writerFunc func([]byte) (int, error) + +func (f writerFunc) Write(b []byte) (int, error) { return f(b) } diff --git a/test/python/stackql_test_tooling/StackQLInterfaces.py b/test/python/stackql_test_tooling/StackQLInterfaces.py index 7af014b1..67677a0a 100644 --- a/test/python/stackql_test_tooling/StackQLInterfaces.py +++ b/test/python/stackql_test_tooling/StackQLInterfaces.py @@ -246,6 +246,7 @@ def _docker_transform_args(self, *args) -> typing.Iterable: rv = [ f"--export.alias='{b[15:]}'" if type(b) == str and b.startswith('--export.alias=') else b for b in list(rv) ] rv = [ f"--http.log.enabled='{b[19:]}'" if type(b) == str and b.startswith('--http.log.enabled=') else b for b in list(rv) ] rv = [ f"--approot='{b[10:]}'" if type(b) == str and b.startswith('--approot=') else b for b in list(rv) ] + rv = [ f"--preview='{b[10:]}'" if type(b) == str and b.startswith('--preview=') else b for b in list(rv) ] return rv def _run_stackql_exec_command_docker( @@ -931,6 +932,76 @@ def write_gcp_service_account(self, path :str): 'token_uri': 'https://oauth2.googleapis.com/token', }, fh) + @keyword + def should_stackql_exec_stream_incrementally( + 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, + min_spread_seconds :float = 0.2, + *args, + **cfg + ): + """ + Assert rows reach the output stream while the query is still running. + + The query is run with stdout redirected to a file, and that file is sampled + while the process is alive. A run that buffers its rows writes the file in + one go at the end, so the samples show a single jump; a run that streams + shows the file growing at several distinct times. + """ + import threading + stdout_path = cfg.get('stdout') + if not stdout_path: + raise Exception('should_stackql_exec_stream_incrementally requires stdout=') + if os.path.exists(stdout_path): + os.remove(stdout_path) + result_holder = {} + + def _run(): + try: + result_holder['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, *args, **cfg + ) + except Exception as exc: + result_holder['error'] = exc + + worker = threading.Thread(target=_run) + worker.start() + samples = [] + while worker.is_alive(): + size = os.path.getsize(stdout_path) if os.path.exists(stdout_path) else 0 + if not samples or size != samples[-1][1]: + samples.append((time.time(), size)) + time.sleep(0.02) + worker.join() + if 'error' in result_holder: + raise result_holder['error'] + + growth = [s for s in samples if s[1] > 0] + with open(stdout_path) as fh: + rows = [line for line in fh.read().splitlines() if line.strip()] + if len(rows) != int(expected_row_count): + raise Exception(f'expected {expected_row_count} rows, got {len(rows)}') + if len(growth) < 2: + raise Exception( + f'output appeared in a single write, so rows were buffered rather than streamed; ' + f'samples={samples}' + ) + spread = growth[-1][0] - growth[0][0] + if spread < float(min_spread_seconds): + raise Exception( + f'output arrived over {spread:.3f}s, under the {min_spread_seconds}s expected for a ' + f'streamed result; rows are probably being buffered' + ) + @keyword def should_stackql_exec_inline_jsonl_set_equal( self, diff --git a/test/python/stackql_test_tooling/flask/README.md b/test/python/stackql_test_tooling/flask/README.md index f5fcfecc..692d1a83 100644 --- a/test/python/stackql_test_tooling/flask/README.md +++ b/test/python/stackql_test_tooling/flask/README.md @@ -144,3 +144,58 @@ COPYFILE_DISABLE=1 tar -czvf .tgz # eg: COPYFILE_DISABLE=1 tar -cvzf google-v0.2.1.tgz v0.2.1 ``` + +## The `omnisdk` mock and the streaming robot test + +`omnisdk/` is a vendored copy of the flask mock bundled with the pinned +`github.com/stackql-labs/omnisdk` module (the version is recorded in +`omnisdk/OMNISDK_VERSION.txt`). It is vendored rather than resolved from the Go +module cache because the `dockertest` and `wsltest` CI jobs download prebuilt +binaries and have no Go toolchain. + +It serves the AWS, Azure and GCP legs the `stackql_preview.audit.*` relations +call, and honours one local addition: `OMNISDK_MOCK_DELAY_MS` delays each +per-bucket request, so a test can observe rows being emitted while the upstream +is still producing. It defaults to zero, and is set to `50` where the mock is +started (`docker-compose-testing.yml` and `web_service_keywords.py`). + +To run the streaming test on its own: + +```bash +PYTHONPATH="${PWD}/test/python" robot \ + --variable EXECUTION_PLATFORM:native \ + --variable SQL_BACKEND:sqlite_embedded \ + --test "Preview Rows Are Emitted Throughout The Run" \ + test/robot/functional +``` + +Every `stackql_preview` scenario, including the multi-cloud row set: + +```bash +PYTHONPATH="${PWD}/test/python" robot \ + --variable EXECUTION_PLATFORM:native \ + --variable SQL_BACKEND:sqlite_embedded \ + --test "Preview*" \ + test/robot/functional +``` + +The streaming test asserts that stdout grows at several distinct times rather +than in one write at the end. The batch size is a provider input, not an environment variable: it rides on +the auth context values, eg +`--auth='{"aws":{"values":{"batch_size":["10"]}}}'`. It caps how many rows a +read gathers; a flush interval (`flush_interval`, default `50ms`) bounds how +long a read waits, so a result smaller than the batch still streams. + +To drive the mock by hand: + +```bash +cd test/python/stackql_test_tooling/flask/omnisdk +OMNISDK_MOCK_DELAY_MS=50 PORT=8085 python3 app.py + +# then, in another shell +export AWS_ACCESS_KEY_ID=AK AWS_SECRET_ACCESS_KEY=SK +export STACKQL_PREVIEW_ENDPOINT='{"aws.s3":{"scheme":"http","host":"127.0.0.1","port":"8085"}}' +./build/stackql exec \ + "select * from stackql_preview.audit.aws_s3_buckets where region = 'us-east-1' and method = 'list';" \ + -o=jsonl +``` diff --git a/test/python/stackql_test_tooling/flask/omnisdk/app.py b/test/python/stackql_test_tooling/flask/omnisdk/app.py index 859679fa..aa805eef 100644 --- a/test/python/stackql_test_tooling/flask/omnisdk/app.py +++ b/test/python/stackql_test_tooling/flask/omnisdk/app.py @@ -15,6 +15,7 @@ """ import json +import time import os import pathlib @@ -54,8 +55,16 @@ def ec2_query(): return Response("InvalidAction", status=400, mimetype=XML) +# Local addition: an opt-in per-request delay, so a test can observe rows being +# emitted while the upstream is still producing. Zero by default, which is the +# upstream behaviour. +_DELAY_SECONDS = float(os.environ.get("OMNISDK_MOCK_DELAY_MS", "0")) / 1000.0 + + @app.get("/") def bucket_op(bucket): + if _DELAY_SECONDS > 0: + time.sleep(_DELAY_SECONDS) if "versioning" in request.args: return Response("Enabled", mimetype=XML) if "publicAccessBlock" in request.args: diff --git a/test/python/stackql_test_tooling/web_service_keywords.py b/test/python/stackql_test_tooling/web_service_keywords.py index f5125f8a..589023e9 100644 --- a/test/python/stackql_test_tooling/web_service_keywords.py +++ b/test/python/stackql_test_tooling/web_service_keywords.py @@ -192,6 +192,8 @@ def create_omnisdk_web_service( stdout=os.path.abspath(os.path.join(self._log_root, f'omnisdk-server-{port}-stdout.txt')), stderr=os.path.abspath(os.path.join(self._log_root, f'omnisdk-server-{port}-stderr.txt')), cwd=self._cwd, + # a delay per request, so a test can observe rows emitted mid-run + env=dict(os.environ, OMNISDK_MOCK_DELAY_MS='50'), ) @keyword diff --git a/test/robot/functional/stackql.resource b/test/robot/functional/stackql.resource index aa155cad..f62c73ce 100644 --- a/test/robot/functional/stackql.resource +++ b/test/robot/functional/stackql.resource @@ -196,3 +196,13 @@ Stackql Per Test Teardown # Should Be Equal As Integers ${restwo.rc} 0 END +Remove Preview Mock Environment + [Documentation] Set Environment Variable is process wide, so the mock + ... credentials must not outlive the test that needs them. + FOR ${name} IN + ... AWS_ACCESS_KEY_ID AWS_SECRET_ACCESS_KEY + ... AZURE_TENANT_ID AZURE_CLIENT_ID AZURE_CLIENT_SECRET + ... GOOGLE_APPLICATION_CREDENTIALS STACKQL_PREVIEW_ENDPOINT + Remove Environment Variable ${name} + END + diff --git a/test/robot/functional/stackql_mocked_from_cmd_line.robot b/test/robot/functional/stackql_mocked_from_cmd_line.robot index a22f8fa0..adfc834b 100644 --- a/test/robot/functional/stackql_mocked_from_cmd_line.robot +++ b/test/robot/functional/stackql_mocked_from_cmd_line.robot @@ -10769,7 +10769,8 @@ Preview Omni Storage Buckets Jsonl Row Set Matches Expectation ... {"aws.s3":${mock}, ... "azure.login":${mock},"azure.mgmt":${mock}, ... "gcp.oauth":${mock},"gcp.storage":${mock},"gcp.crm":${mock}} - Set Environment Variable STACKQL_PREVIEW_ENDPOINT ${endpoints} + ${preview} = Catenate SEPARATOR= + ... {"endpoint":${endpoints}} ${query} = Catenate SEPARATOR=${SPACE} ... select * from stackql_preview.audit.omni_storage_buckets ... where region = 'us-east-1' and google_org = '123456789'; @@ -10783,6 +10784,59 @@ Preview Omni Storage Buckets Jsonl Row Set Matches Expectation ... ${SQL_BACKEND_CFG_STR_CANONICAL} ... ${query} ... ${expected} + ... --preview\=${preview} ... stdout=${CURDIR}${/}tmp${/}Preview-Omni-Storage-Buckets-Jsonl-Row-Set-Matches-Expectation.tmp ... stderr=${CURDIR}${/}tmp${/}Preview-Omni-Storage-Buckets-Jsonl-Row-Set-Matches-Expectation-stderr.tmp +Preview Rows Are Emitted Throughout The Run + [Documentation] A streamed relation must put rows on the output stream as + ... the upstream produces them, not accumulate and flush at + ... the end. The mock delays each per-bucket request, so a + ... buffered implementation writes stdout in one go while a + ... streaming one grows it steadily. + [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} + ... 79 + ... 0.2 + ... \-o\=jsonl + ... --preview\=${preview} + ... stdout=${CURDIR}${/}tmp${/}Preview-Rows-Are-Emitted-Throughout-The-Run.tmp + ... stderr=${CURDIR}${/}tmp${/}Preview-Rows-Are-Emitted-Throughout-The-Run-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 + ... than silently dropped, which would return rows in an + ... order the caller did not ask for. + ${query} = Catenate SEPARATOR=${SPACE} + ... select name, provider, encryption_class from stackql_preview.audit.omni_storage_buckets + ... where region = 'us-east-1' and google_org = '123456789' order by name; + Should StackQL Exec Inline Contain Stderr + ... ${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} + ... streams its rows, so ORDER BY cannot be applied; remove it from the query + ... stdout=${CURDIR}${/}tmp${/}Preview-Order-By-Is-Refused-Rather-Than-Ignored.tmp + ... stderr=${CURDIR}${/}tmp${/}Preview-Order-By-Is-Refused-Rather-Than-Ignored-stderr.tmp