streams API: support per-request env overrides for config templating

[g-hurst]

Closes #481

Summary

Adds an optional envelope request form to the streams-mode HTTP API's config payloads, so a caller can supply per-request environment variable overrides instead of pre-rendering ${FOO}-style templates themselves or relying on real OS environment variables.

{
  "env": { "TOPIC": "orders", "BROKER": "localhost:9092" },
  "template": "input:\n  kafka:\n    addresses: [ \"${BROKER}\" ]\n    topics: [ \"${TOPIC}\" ]\noutput:\n  stdout: {}\n"
}

Covered endpoints:

• POST/PUT/PATCH /streams/{id} (HandleStreamCRUD)
• POST /resources/{type}/{id} (HandleResourceCRUD)

Bulk POST /streams is out of scope, as described in the issue — it has no env substitution today, so adding overrides there means building that support for the first time rather than reusing it. Its endpoint description now says so explicitly, since its two neighbours advertise the envelope.

No new templating syntax is introduced. This only changes where the existing envLookupFunc seam gets its answers from.

The request shape differs from the sketch in #481

The issue proposed env as a sibling of the config fields:

{ "env": { ... }, "input": { ... }, "output": { ... } }

That shape forces the server to parse the config in order to find and strip env before linting — and then re-serialise what is left. A config body is a template, not a document, and a round trip through go-yaml's emitter is free to:

• Re-resolve implicit scalar types — 2024-01-02 comes back as a timestamp, 0123456 as octal.
• Drop the caller's quoting around a ${VAR} — which is precisely what stops an interpolated value containing YAML metacharacters from restructuring the document.
• Rewrite comments, anchors and style.

Worse, the sibling shape requires the body to parse at all before substitution has run, which is not guaranteed: the documented ${VAR: default} form puts a : inside an otherwise plain scalar.

Carrying the config as a string under template removes the parse entirely. Reading a string node hands back exactly the bytes the caller wrote, and the envelope itself parses whatever the template contains — as a JSON string or a YAML block scalar, ${VAR: default} is just text.

Implementation

internal/stream/manager/api.go — decodeConfigBody

Splits a request body into overrides and the config document to act on. The envelope form is a mapping whose only keys are env and template, where template holds the config as a string. Anything that does not match that shape exactly — a repeated key, any third key, a missing or non-string template — falls back to treating the body as the config itself, returned unchanged (the same backing slice; a test pins this with assert.Same). No request that worked before this existed can behave differently.

Past the point where the body is unambiguously an envelope, a malformed env is reported as a 400 rather than silently reinterpreted. Value typing is validated from the YAML node tag, so a quoted "5" is accepted and a bare 5 is rejected — a decode into map[string]string could not tell those apart.

internal/config/reader.go — OptAddEnvLookupOverrides

A reader option that composes with the reader's existing lookup func rather than replacing it: overrides are checked first, and any name absent from them falls through to whatever the reader already had (by default os.LookupEnv). An override therefore shadows a same-named OS env var, while every other variable continues to resolve from the OS as usual.

Wrapping the lookup func — rather than pre-substituting the document — is what keeps the rest of the interpolation contract intact. ${FOO:default} still falls back to its default when FOO is neither overridden nor set, and a ${{FOO}} escape is still left alone, because ReplaceEnvVariables remains the single thing doing the parsing. An earlier two-pass approach broke both of those.

Everything downstream (lint, ParsedConfigFromAny, stream.FromParsed, chilled/ErrMissingEnvVars handling) is untouched.

internal/stream/manager/api.go — patchConfig

PATCH is a documented verb of /streams/{id}, so it decodes the body the same way POST and PUT do and applies the envelope's template as the patch document. Without this the envelope's own keys merged into the stored config as ordinary fields — a patch is neither linted nor parsed strictly, so the request returned a 200 while persisting stray env/template keys visible on the next GET.

A patch performs no environment variable substitution (it never has, and this PR does not change that), so overrides could only ever be a no-op there. A non-empty env is rejected with a 400 rather than accepting a request whose whole point is silently dropped:

PATCH /streams/{id} body	Result
{"input": { ... }}	200, merged as the patch — unchanged from today
{"template": "<config>"}	200, template applied as the patch
{"env": {}, "template": "<config>"}	200, template applied
{"env": {"FOO": "x"}, "template": "<config>"}	400 field env: not supported by PATCH
{"env": {"FOO": 3}, "template": "<config>"}	400 value of 'FOO' must be a string

Acceptance criteria

• An env map resolves ${VAR} placeholders in the config
• A request env value takes precedence over a same-named OS environment variable
• Omitting the envelope behaves exactly as today — no behavior change for existing callers
• A missing variable still produces today's ErrMissingEnvVars/chilled behavior
• A non-string env value is rejected with a 400, not coerced
• Test coverage for both /streams/{id} and /resources/{type}/{id}

Tests

Added in internal/stream/manager/api_test.go, internal/stream/manager/api_internal_test.go and internal/config/env_vars_test.go:

• TestDecodeConfigBody — 24 cases covering envelope detection and passthrough: raw JSON and YAML configs, a body that only parses after substitution, empty and bare-scalar bodies, template holding a mapping or a number, template alongside a config field, env with no template, duplicated keys, block scalars, and the full set of non-string env value rejections
• TestEnvLookupOverrides — override precedence, fallthrough to the wrapped lookup, defaults preserved, ${{...}} escapes untouched
• TestTypeAPIStreamEnvOverrides / TestResourceAPIEnvOverrides — override applied, precedence over a real OS env var, fallback to OS env when no override is given, mixed resolution from both sources, missing-var behavior unchanged, non-string rejected with 400, and a no-envelope regression check
• TestTypeAPIStreamEnvOverridesPreserveDocument / ...PreserveScalars — a template's quoting and implicit scalar types survive the request path unchanged
• TestTypeAPIPatchEnvelope — the envelope's template is applied as the patch and its keys do not leak into the stored config, null and empty env are accepted, a populated or non-string env is a 400 that leaves the stream untouched, and a raw patch body still applies as before

Verification

• make test — full suite passes
• make lint — 0 issues
• make fmt — no changes

The endpoint description strings for /streams/{id}, /resources/{type}/{id} and /streams were updated; no generated docs are affected.

Known gaps

One behaviour found in review that is not addressed here, called out so it is a deliberate choice rather than an oversight:

• An envelope carrying a valid env but a missing or non-string template falls through to passthrough, silently discarding the overrides. Under ?chilled=true this yields a 200 and a running stream built from the envelope itself. Since env is not a valid top-level field on either endpoint, its presence is unambiguous envelope intent and arguably should be a 400.

Happy to fold that into this PR if reviewers would rather not land the feature without it.

🤖 Generated with Claude Code

https://claude.ai/code/session_01Lw4vixL5ecJAvuKAEhCdGr

[CLAassistant]

All committers have signed the CLA.

[g-hurst]

Hi @josephwoodward — no rush at all on this one, just flagging it since the workflows here need a maintainer to kick them off for first-time contributors, and I don't think they've been approved to run yet (only the CLA check has reported so far). Whenever you have a spare moment, would you mind approving a CI run? Happy to fix anything that falls out.

For context on the motivation: I'm a happy Connect user, and this came out of actually running into the limitation rather than from reading the API surface. In a streams-mode deployment the process env is shared by every stream in it, so there's no way to post the same config template twice with different parameters — you either pre-render the template yourself before posting or split things across separate processes. Making the templating per-request removes that, and since Connect builds on Benthos the ergonomic win lands downstream in Connect, which is the outcome I'm really after here.

If you get a chance to look at the approach itself later on, I'd really value your feedback. I've just added a short "Divergences from the sketch in #481" section to the description covering the two places this departs from what I originally proposed in the issue, including one where I'd happily take the smaller-surface option if you prefer it.

Thanks for all the review work you do on this repo — much appreciated.

[g-hurst]

Commits

1. fc04aa48b bundles CHANGELOG.md with the code change. Documentation changes must be in a separate commit from code changes in a multi-commit PR. 9ea9494dd and d44af54bf do this correctly; the first commit does not.
2. The series is not a progression of self-contained changes — it carries three superseded approaches. 05322f86c ("strip the env field at the yaml node level") is replaced by 9dcf50863's envelope decoder; 993c05043 then deletes 73 lines of api.go added two commits earlier, and b263144da rewrites 249 lines of a test file added earlier in the same series. fc04aa48b..993c05043 should be squashed down to the approach that actually landed.
3. 10ed88ddd (+4 lines) and f10b36bf4 are unsquashed fixups — they patch problems introduced by earlier commits in the same PR, and belong folded into those commits.
4. Every subject uses stream manager:, with a space in the system name. The convention in this repo is a single token or a path (websocket:, http:, config:, batch:, processor/try_catch:); stream/manager: or streams: would match.

Message wording itself is fine throughout — lowercase, imperative, and accurate to the contents.

Review

The core design holds up. Applying overrides by wrapping envLookupFunc rather than pre-substituting the document is what keeps ${FOO:default} and the ${{FOO}} escape working, since ReplaceEnvVariables stays the only thing parsing; and carrying the config as a string under template avoids a parse/re-serialise round trip on a body that is only guaranteed to parse after substitution. Envelope detection rejects repeated keys, third keys and non-!!str templates by falling back to the pre-existing path, decode errors reach requestErr and surface as 400 on both handlers, and the bulk POST /streams exclusion is documented at the endpoint. Test coverage is real: a 24-case table for decodeConfigBody, precedence/fallthrough/escape cases, and a PATCH envelope-leak regression.

One issue, already disclosed in the PR description's "Known gaps" and repeated here so it is tracked against the code:

1. benthos/internal/stream/manager/api.go

   Lines 165 to 175 in d44af54

   	if templateNode == nil {
   	return nil, raw, nil
   	}
   	if templateNode.Kind == yaml.AliasNode {
   	templateNode = templateNode.Alias
   	}
   	if templateNode.Kind != yaml.ScalarNode || templateNode.Tag != "!!str" {
   	return nil, raw, nil
   	}

   — an envelope carrying a valid env but a missing or non-string template falls through to passthrough, silently discarding the overrides. Under ?chilled=true lints are skipped, so the request returns 200 and starts a stream built from the envelope body itself (defaults for input and output) rather than from the caller's config. Since env is not a valid top-level field on either endpoint, its presence is unambiguous envelope intent; once env is seen as a top-level key the body should be committed to the envelope path and a missing or non-string template rejected with a 400, instead of falling back.

Sub-threshold, not blocking:

benthos/internal/stream/manager/api_test.go

Lines 616 to 630 in d44af54

	var names []string
	assert.Eventually(t, func() bool {
	entries, err := os.ReadDir(".")
	if err != nil {
	return false
	}
	names = names[:0]
	for _, e := range entries {
	names = append(names, e.Name())
	}
	return len(names) > 0
	}, time.Second*5, time.Millisecond*10)
	assert.Equal(t, []string{"2024-01-02"}, names)
	}

builds names inside the assert.Eventually condition and asserts on it after the call returns, with the condition satisfied by len(names) > 0. Comparing inside the condition, or using assert.EventuallyWithT, keeps the assertion and the polled state on the same goroutine.

[g-hurst]

Commits

1. fc04aa48b bundles CHANGELOG.md with the code change. Documentation changes must be in a separate commit from code changes in a multi-commit PR. 9ea9494dd and d44af54bf do this correctly; the first commit does not.
2. The series is not a progression of self-contained changes — it carries three superseded approaches. 05322f86c ("strip the env field at the yaml node level") is replaced by 9dcf50863's envelope decoder; 993c05043 then deletes 73 lines of api.go added two commits earlier, and b263144da rewrites 249 lines of a test file added earlier in the same series. fc04aa48b..993c05043 should be squashed down to the approach that actually landed.
3. 10ed88ddd (+4 lines) and f10b36bf4 are unsquashed fixups — they patch problems introduced by earlier commits in the same PR, and belong folded into those commits.
4. Every subject uses stream manager:, with a space in the system name. The convention in this repo is a single token or a path (websocket:, http:, config:, batch:, processor/try_catch:); stream/manager: or streams: would match.

Message wording itself is fine throughout — lowercase, imperative, and accurate to the contents.

Review

The core design holds up. Applying overrides by wrapping envLookupFunc rather than pre-substituting the document is what keeps ${FOO:default} and the ${{FOO}} escape working, since ReplaceEnvVariables stays the only thing parsing; and carrying the config as a string under template avoids a parse/re-serialise round trip on a body that is only guaranteed to parse after substitution. Envelope detection rejects repeated keys, third keys and non-!!str templates by falling back to the pre-existing path, decode errors reach requestErr and surface as 400 on both handlers, and the bulk POST /streams exclusion is documented at the endpoint. Test coverage is real: a 24-case table for decodeConfigBody, precedence/fallthrough/escape cases, and a PATCH envelope-leak regression.

One issue, already disclosed in the PR description's "Known gaps" and repeated here so it is tracked against the code:

1. benthos/internal/stream/manager/api.go

   Lines 165 to 175 in d44af54

   	if templateNode == nil {
   	return nil, raw, nil
   	}
   	if templateNode.Kind == yaml.AliasNode {
   	templateNode = templateNode.Alias
   	}
   	if templateNode.Kind != yaml.ScalarNode || templateNode.Tag != "!!str" {
   	return nil, raw, nil
   	}

   
   — an envelope carrying a valid env but a missing or non-string template falls through to passthrough, silently discarding the overrides. Under ?chilled=true lints are skipped, so the request returns 200 and starts a stream built from the envelope body itself (defaults for input and output) rather than from the caller's config. Since env is not a valid top-level field on either endpoint, its presence is unambiguous envelope intent; once env is seen as a top-level key the body should be committed to the envelope path and a missing or non-string template rejected with a 400, instead of falling back.

Sub-threshold, not blocking:

benthos/internal/stream/manager/api_test.go

Lines 616 to 630 in d44af54

	var names []string
	assert.Eventually(t, func() bool {
	entries, err := os.ReadDir(".")
	if err != nil {
	return false
	}
	names = names[:0]
	for _, e := range entries {
	names = append(names, e.Name())
	}
	return len(names) > 0
	}, time.Second*5, time.Millisecond*10)
	assert.Equal(t, []string{"2024-01-02"}, names)
	}

builds names inside the assert.Eventually condition and asserts on it after the call returns, with the condition satisfied by len(names) > 0. Comparing inside the condition, or using assert.EventuallyWithT, keeps the assertion and the polled state on the same goroutine.

Commits

1. fc04aa48b bundles CHANGELOG.md with the code change. Documentation changes must be in a separate commit from code changes in a multi-commit PR. 9ea9494dd and d44af54bf do this correctly; the first commit does not.
2. The series is not a progression of self-contained changes — it carries three superseded approaches. 05322f86c ("strip the env field at the yaml node level") is replaced by 9dcf50863's envelope decoder; 993c05043 then deletes 73 lines of api.go added two commits earlier, and b263144da rewrites 249 lines of a test file added earlier in the same series. fc04aa48b..993c05043 should be squashed down to the approach that actually landed.
3. 10ed88ddd (+4 lines) and f10b36bf4 are unsquashed fixups — they patch problems introduced by earlier commits in the same PR, and belong folded into those commits.
4. Every subject uses stream manager:, with a space in the system name. The convention in this repo is a single token or a path (websocket:, http:, config:, batch:, processor/try_catch:); stream/manager: or streams: would match.

Message wording itself is fine throughout — lowercase, imperative, and accurate to the contents.

Review

The core design holds up. Applying overrides by wrapping envLookupFunc rather than pre-substituting the document is what keeps ${FOO:default} and the ${{FOO}} escape working, since ReplaceEnvVariables stays the only thing parsing; and carrying the config as a string under template avoids a parse/re-serialise round trip on a body that is only guaranteed to parse after substitution. Envelope detection rejects repeated keys, third keys and non-!!str templates by falling back to the pre-existing path, decode errors reach requestErr and surface as 400 on both handlers, and the bulk POST /streams exclusion is documented at the endpoint. Test coverage is real: a 24-case table for decodeConfigBody, precedence/fallthrough/escape cases, and a PATCH envelope-leak regression.

One issue, already disclosed in the PR description's "Known gaps" and repeated here so it is tracked against the code:

1. benthos/internal/stream/manager/api.go

   Lines 165 to 175 in d44af54

   	if templateNode == nil {
   	return nil, raw, nil
   	}
   	if templateNode.Kind == yaml.AliasNode {
   	templateNode = templateNode.Alias
   	}
   	if templateNode.Kind != yaml.ScalarNode || templateNode.Tag != "!!str" {
   	return nil, raw, nil
   	}

   
   — an envelope carrying a valid env but a missing or non-string template falls through to passthrough, silently discarding the overrides. Under ?chilled=true lints are skipped, so the request returns 200 and starts a stream built from the envelope body itself (defaults for input and output) rather than from the caller's config. Since env is not a valid top-level field on either endpoint, its presence is unambiguous envelope intent; once env is seen as a top-level key the body should be committed to the envelope path and a missing or non-string template rejected with a 400, instead of falling back.

Sub-threshold, not blocking:

benthos/internal/stream/manager/api_test.go

Lines 616 to 630 in d44af54

	var names []string
	assert.Eventually(t, func() bool {
	entries, err := os.ReadDir(".")
	if err != nil {
	return false
	}
	names = names[:0]
	for _, e := range entries {
	names = append(names, e.Name())
	}
	return len(names) > 0
	}, time.Second*5, time.Millisecond*10)
	assert.Equal(t, []string{"2024-01-02"}, names)
	}

builds names inside the assert.Eventually condition and asserts on it after the call returns, with the condition satisfied by len(names) > 0. Comparing inside the condition, or using assert.EventuallyWithT, keeps the assertion and the polled state on the same goroutine.

Hi @josephwoodward, thanks for triggering the CI this morning!

I ran the prompt for the AI review locally this morning, and it brought up some issues that led me to rethink my approach. I updated the PR with a revised approach this afternoon, ran a few of my review agents, and wrapped things up with a review agent with the CI prompt.

Let me know what you think of this revised approach as well as if you have any thoughts on the one area left open!