feat(sinks): add validate_structure() for early template confinement checks - #25959
feat(sinks): add validate_structure() for early template confinement checks#25959thomasqueirozb wants to merge 10 commits into
Conversation
github-actions
Bot
added
the
domain: sinks
thomasqueirozb
added
the
no-changelog
🎉 All green!🧪 All tests passed 🔗 Commit SHA: d2530d6 | Docs | Datadog PR Page | Give us feedback! |
thomasqueirozb
removed
the
no-changelog
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: d172a7cfa6
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 45539ea4cc
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
…y, axiom, humio) This adds validate_structure() implementations for wrapper sinks that delegate to HTTP or HEC sinks: - opentelemetry: Delegates to HttpSinkConfig::validate_structure() - axiom: Builds HttpSinkConfig and delegates validation - humio_logs: Delegates to HecLogsSinkConfig::validate_structure() - humio_metrics: Builds HumioLogsConfig and delegates to HEC validation These wrapper sinks were using the default validate_structure() which returns Ok(()), allowing unconfined templates to pass validation even though they would fail at boot. Now they properly delegate to their inner sink's validation. Also adds unit tests for each wrapper sink's confinement validation.
45539ea to
5dd8b9b
Compare
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 5dd8b9b0f6
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
| // Validate labels confinement (both keys and values) | ||
| for (k, v) in self.labels.clone().into_iter() { |
There was a problem hiding this comment.
When labels is empty or a static label key is invalid, for example bad-label, this new structural pass only confines the key/value templates and returns Ok; build() still rejects those same configs before doing any environmental work. That leaves vector validate --no-environment accepting Loki configs that normal startup/reload immediately rejects, so please include the existing empty-label and valid_label_name checks here.
Useful? React with 👍 / 👎.
| fn validate_structure(&self) -> std::result::Result<(), Vec<String>> { | ||
| let mut errors = Vec::new(); | ||
|
|
||
| if let Some(tenant_id) = self.tenant_id.clone() | ||
| && let Err(e) = tenant_id.confine(&self.confinement, Self::NAME, "tenant_id") |
There was a problem hiding this comment.
For prometheus_remote_write configs with a malformed static endpoint or invalid/conflicting headers such as Authorization together with auth, this method succeeds because it only checks tenant_id; build() parses the endpoint and calls validate_headers() before creating the client. Please mirror those non-environment checks so vector validate --no-environment does not pass configs that startup rejects.
Useful? React with 👍 / 👎.
| fn validate_structure(&self) -> std::result::Result<(), Vec<String>> { | ||
| let mut errors = Vec::new(); | ||
|
|
||
| if let Err(e) = | ||
| self.blob_prefix | ||
| .clone() | ||
| .confine(&self.confinement, Self::NAME, "blob_prefix") |
There was a problem hiding this comment.
When an Azure Blob sink omits all of connection_string, account_name, and blob_endpoint, supplies more than one of them, or uses account_name/blob_endpoint without auth, this structural validator still returns Ok after checking only blob_prefix; build() rejects each of those cases before contacting Azure. Please carry those option-combination checks into validate_structure() so no-environment validation matches startup.
Useful? React with 👍 / 👎.
| if let Err(e) = self | ||
| .topic | ||
| .clone() | ||
| .confine(&self.confinement, Self::NAME, "topic") |
There was a problem hiding this comment.
When a Kafka config sets batch.max_events while also setting librdkafka_options.batch.num.messages (or the analogous timeout/max-bytes pairs), KafkaSink::new() calls to_rdkafka() and rejects the duplicate producer setting before creating the producer, but this structural pass only confines topic. Please mirror those conflict checks here so vector validate --no-environment does not accept Kafka configs that startup rejects.
Useful? React with 👍 / 👎.
| if let Err(e) = self | ||
| .key | ||
| .clone() | ||
| .confine(&self.confinement, Self::NAME, "key") |
There was a problem hiding this comment.
For a Redis sink with key = "", Template::confine() succeeds because there are no dynamic fields, so this new validator returns Ok; normal startup still rejects the same config immediately via self.key.is_empty(). Please add the existing empty-key check before the confinement call so no-environment validation matches startup.
Useful? React with 👍 / 👎.
| if let Err(e) = self | ||
| .topic | ||
| .clone() | ||
| .confine(&self.confinement, Self::NAME, "topic") |
There was a problem hiding this comment.
When MQTT is configured with an empty client_id or with only one of user/password, this structural pass succeeds after confining topic, while build_connector() rejects those static settings before any broker interaction. Please run the same connector precondition checks here so no-environment validation reports the error early.
Useful? React with 👍 / 👎.
| if let Err(e) = self | ||
| .table | ||
| .clone() | ||
| .confine(&self.confinement, Self::NAME, "table") |
There was a problem hiding this comment.
For ClickHouse configs that set batch_encoding with a non-arrow_stream format, or use Arrow batch encoding with a dynamic but prefixed table/database template, this validator still returns Ok because it only confines the templates; resolve_strategy()/resolve_arrow_schema() reject those static constraints during startup. Please mirror those checks here so vector validate --no-environment catches the invalid combination.
Useful? React with 👍 / 👎.
| fn validate_structure(&self) -> std::result::Result<(), Vec<String>> { | ||
| let mut errors = Vec::new(); | ||
|
|
||
| // Validate URI confinement | ||
| if let Err(e) = self |
There was a problem hiding this comment.
For an HTTP sink with an invalid batch setting such as batch.max_events = 0, this method returns Ok after only checking URI/header confinement, but build_with_component_type() immediately calls self.batch.validate()? and rejects the config before creating the service. Please run the same batch validation here so vector validate --no-environment reports the out-of-range batch values.
Useful? React with 👍 / 👎.
| if let Err(e) = self | ||
| .subject | ||
| .clone() | ||
| .confine(&self.confinement, Self::NAME, "subject") |
There was a problem hiding this comment.
When url contains a malformed NATS server address, this structural validator still succeeds because it only confines subject; publisher() later calls parse_server_addresses() and fails before making a connection attempt. Please parse the comma-separated server addresses here as well so no-environment validation catches static URL mistakes.
Useful? React with 👍 / 👎.
| match Template::try_from(self.key_prefix.as_deref().unwrap_or("date=%F/")) { | ||
| Ok(tpl) => { | ||
| if let Err(e) = tpl.confine(&self.confinement, Self::NAME, "key_prefix") { |
There was a problem hiding this comment.
For gcp_cloud_storage configs with malformed static header values such as a newline in content_type, content_encoding, cache_control, or metadata, this validator returns Ok because it only checks key_prefix; RequestSettings::new() later converts those fields to HeaderValue and rejects the sink during startup. Please mirror those header-value checks here so vector validate --no-environment fails for the same invalid config.
Useful? React with 👍 / 👎.
- Parse static URIs during validation to catch invalid URI syntax - Validate HTTP headers before splitting templates (malformed headers, auth conflicts) - Reject raw endpoint with auto_extract_timestamp in HEC logs during validation These validations catch errors at time that would otherwise fail at boot.
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 35bb233832
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
| if errors.is_empty() { | ||
| Ok(()) |
There was a problem hiding this comment.
When the HTTP sink is configured with JSON encoding and comma-delimited framing, invalid payload_prefix/payload_suffix pairs (for example a prefix of {"data": without a closing suffix) still pass this structural pass because it stops after URI/header confinement. build_with_component_type() immediately calls validate_payload_wrapper() after build_encoder() and rejects the same config before any environment access, so vector validate --no-environment remains inconsistent; please run the wrapper check here too.
Useful? React with 👍 / 👎.
| errors.push(e.to_string()); | ||
| } | ||
|
|
||
| if errors.is_empty() { |
There was a problem hiding this comment.
For aws_cloudwatch_logs, configs with invalid batch limits such as batch.max_events = 0 or batch.timeout_secs = 0 now pass this structural validator because it only confines group_name and stream_name. CloudwatchLogsSinkConfig::build() calls self.batch.into_batcher_settings()? before creating the AWS client, so normal startup rejects the same config without needing environment access; please include the batch validation in this structural pass.
Useful? React with 👍 / 👎.
| errors.push(e.to_string()); | ||
| } | ||
|
|
||
| if errors.is_empty() { |
There was a problem hiding this comment.
When a Redis sink uses an empty endpoint list or a malformed endpoint string, this method still returns success after checking only key, but build_connection() immediately rejects an empty list and parses the URL via redis::Client::open/Sentinel::build before any Redis connection is made. That leaves vector validate --no-environment accepting Redis configs that startup cannot build; please mirror the endpoint non-empty and parse checks here.
Useful? React with 👍 / 👎.
| errors.push(e.to_string()); | ||
| } | ||
|
|
||
| if errors.is_empty() { |
There was a problem hiding this comment.
When max_channels = 0, this structural pass still succeeds because it only checks the exchange and routing-key templates, but AmqpSink::new() calls new_channel_pool() and rejects the config with max_channels must be positive before opening any AMQP connection. That leaves vector validate --no-environment accepting an AMQP config that startup cannot build; please mirror the positive max_channels check here.
Useful? React with 👍 / 👎.
| } | ||
| } | ||
|
|
||
| if errors.is_empty() { |
There was a problem hiding this comment.
When an Elasticsearch sink omits both endpoint and endpoints, or sets both the deprecated endpoint and endpoints, this validator returns Ok after only confining the active mode templates. ElasticsearchCommon::parse_many() rejects those endpoint-shape errors before any cluster/version request, so vector validate --no-environment can still validate a config that normal startup immediately rejects; please mirror the endpoint required/exclusive checks here.
Useful? React with 👍 / 👎.
Add check for empty endpoints during structural validation so that `vector validate --no-environment` catches the same error that would fail at startup.
Validate content_type, content_encoding, cache_control, and metadata header values during structural validation to catch malformed headers that would otherwise fail at startup in RequestSettings::new().
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: e6077b6057
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
| fn validate_structure(&self) -> std::result::Result<(), Vec<String>> { | ||
| let mut errors = Vec::new(); |
There was a problem hiding this comment.
When greptimedb_logs.endpoint is malformed or omits the scheme, for example endpoint = "localhost:4000", this structural pass still succeeds because it only confines the table/database/pipeline templates. The request path later calls url::Url::parse(...).unwrap() in prepare_log_ingester_url, so a config that vector validate --no-environment accepts can still fail or panic before sending; parse the static endpoint here as part of validation.
Useful? React with 👍 / 👎.
| if let Some(metadata) = &self.metadata { | ||
| for (name, value) in metadata { | ||
| if let Err(e) = HeaderValue::from_str(value) { | ||
| errors.push(format!("metadata.{name}: invalid header value: {e}")); |
There was a problem hiding this comment.
When metadata contains an invalid header name such as "bad\nname", this new validator only checks HeaderValue::from_str(value), so vector validate --no-environment succeeds even though RequestSettings::new() later calls make_header() and rejects the same config via HeaderName::from_bytes(name.as_bytes()) before any GCS request. Please validate the metadata keys alongside the values.
Useful? React with 👍 / 👎.
| fn validate_structure(&self) -> std::result::Result<(), Vec<String>> { | ||
| use http::header::HeaderValue; | ||
|
|
||
| let mut errors = Vec::new(); |
There was a problem hiding this comment.
When gcp_cloud_storage.endpoint is malformed, for example endpoint = "http://%", this validator still succeeds after checking only key/header fields. Startup later formats the endpoint into base_url and build_sink() immediately calls base_url.parse::<Uri>().unwrap(), so vector validate --no-environment can accept a config that panics before the sink is created; parse the configured endpoint/base URL here and return the error instead.
Useful? React with 👍 / 👎.
| fn validate_structure(&self) -> std::result::Result<(), Vec<String>> { | ||
| let mut errors = Vec::new(); |
There was a problem hiding this comment.
When the Loki endpoint embeds basic auth, such as http://u:p@host, and auth is also configured, this structural pass only confines templates and returns Ok; build() immediately calls self.auth.choose_one(&self.endpoint.auth)? and rejects the same config before constructing the sink. Please mirror that credential exclusivity check so no-environment validation does not accept configs startup cannot run.
Useful? React with 👍 / 👎.
| fn validate_structure(&self) -> std::result::Result<(), Vec<String>> { | ||
| let mut errors = Vec::new(); |
There was a problem hiding this comment.
When pulsar.auth is present but sets only one of name/token, leaves both empty, or mixes oauth2 with name/token, this structural pass still succeeds after confining only topic. create_pulsar_client() rejects those static combinations before it attempts to build/connect the client, so please add the same auth-shape check here so vector validate --no-environment reports the error early.
Useful? React with 👍 / 👎.
- Check for empty labels (mirrors build() check) - Validate label names using valid_label_name() (mirrors build() check) - Add tests for both validation cases
- azure_blob: validate credential shape (connection_string/account_name/blob_endpoint combinations) - clickhouse: validate batch_encoding compatibility and arrow schema requirements - doris: check for empty endpoints - gcp_cloud_storage: validate header values (content_type, content_encoding, cache_control, metadata) - http: validate batch settings (max_events, max_bytes, timeout) - kafka: check batch option conflicts with librdkafka_options - loki: validate empty labels and label name format - mqtt: check empty client_id and user/password pairing - nats: parse server URLs during validation - prometheus_remote_write: validate endpoint URI and headers - redis: reject empty keys All validations mirror checks in build() so `vector validate --no-environment` catches errors that would otherwise fail at boot.
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: d2530d6247
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
| // Validate static URIs can be parsed | ||
| if !self.uri.is_dynamic() | ||
| && let Err(e) = self.uri.get_ref().parse::<http::Uri>() |
There was a problem hiding this comment.
When the HTTP URI is static and embeds basic auth while auth is also configured, for example uri = "https://u:p@example.com" with bearer auth, this new check only parses the string as http::Uri and succeeds. HttpSinkRequestBuilder::build() later parses the rendered URI with UriSerde and calls self.auth.choose_one(&uri_auth), so every request fails despite vector validate --no-environment passing; please use the same URI-auth extraction and exclusivity check for static URIs here.
Useful? React with 👍 / 👎.
| fn validate_structure(&self) -> std::result::Result<(), Vec<String>> { | ||
| let mut errors = Vec::new(); |
There was a problem hiding this comment.
When a ClickHouse endpoint includes basic auth and auth is also configured, this structural validator succeeds because it only checks batch/table/database constraints. Normal startup reaches ClickhouseConfig::build() and immediately calls self.auth.choose_one(&self.endpoint.auth), rejecting the same static config before any connection is made, so vector validate --no-environment remains inconsistent for this sink; please mirror that credential exclusivity check here.
Useful? React with 👍 / 👎.
| fn validate_structure(&self) -> std::result::Result<(), Vec<String>> { | ||
| let mut errors = Vec::new(); |
There was a problem hiding this comment.
When an aws_s3 config has an invalid batch setting such as batch.max_events = 0, this structural pass still returns success because it only checks the key templates. With otherwise valid AWS settings, startup reaches build_processor() and rejects the same config via self.batch.into_batcher_settings()? before constructing the sink, so vector validate --no-environment can still pass a config that cannot run; please include the batch validation here.
Useful? React with 👍 / 👎.
| // Mirror credential shape checks from build() | ||
| match ( | ||
| &self.connection_string, | ||
| &self.account_name, | ||
| &self.blob_endpoint, | ||
| ) { |
There was a problem hiding this comment.
When connection_string already carries credentials, such as a SAS or shared key, and auth is also configured, this match accepts the shape because only connection_string is present. build_client() then parses the connection string and rejects the duplicate credential sources before creating the Azure client, so vector validate --no-environment still succeeds for a config normal startup rejects; please mirror the parsed connection-string auth conflict here.
Useful? React with 👍 / 👎.
| fn validate_structure(&self) -> std::result::Result<(), Vec<String>> { | ||
| let mut errors = Vec::new(); |
There was a problem hiding this comment.
When gcp_stackdriver_logs is configured with invalid batch limits, for example batch.max_events = 0 or batch.max_bytes above the 10 MB API cap, this method only confines templates and returns Ok. build() runs self.batch.validate()?.limit_max_bytes(MAX_BATCH_PAYLOAD_SIZE)? before constructing the HTTP service, so no-environment validation remains inconsistent with startup; please run the same batch checks in this structural pass.
Useful? React with 👍 / 👎.
| fn validate_structure(&self) -> std::result::Result<(), Vec<String>> { | ||
| let mut errors = Vec::new(); |
There was a problem hiding this comment.
When prometheus_remote_write has invalid batch settings such as batch.max_events = 0 or batch.timeout_secs = 0, this structural pass still succeeds after checking endpoint, headers, and tenant templates. RemoteWriteSink construction later calls self.batch.batch_settings.validate()?.into_batcher_settings()?, so vector validate --no-environment can accept a config that startup rejects before running the sink; please mirror that batch validation here.
Useful? React with 👍 / 👎.
| fn validate_structure(&self) -> std::result::Result<(), Vec<String>> { | ||
| let mut errors = Vec::new(); |
There was a problem hiding this comment.
When splunk_hec_logs has an invalid batch setting such as batch.max_events = 0, this method only checks the raw-endpoint incompatibility and metadata templates, so no-environment validation succeeds. build_processor() later calls self.batch.into_batcher_settings()? before constructing the sink, and humio_logs/humio_metrics delegate through this same HEC config, so please run the same batch validation in this structural pass.
Useful? React with 👍 / 👎.
| fn validate_structure(&self) -> std::result::Result<(), Vec<String>> { | ||
| let mut errors = Vec::new(); |
There was a problem hiding this comment.
When splunk_hec_metrics has invalid batch limits such as batch.max_events = 0 or batch.timeout_secs = 0, this structural validator only confines optional metadata templates and returns success. build_processor() rejects the same config via self.batch.into_batcher_settings()? before creating the sink, so vector validate --no-environment remains inconsistent; please include the batch validation here as well.
Useful? React with 👍 / 👎.
| fn validate_structure(&self) -> std::result::Result<(), Vec<String>> { | ||
| let mut errors = Vec::new(); |
There was a problem hiding this comment.
When an Elasticsearch endpoint embeds basic auth and top-level basic auth is also configured, this method still succeeds because it only validates routing templates. ElasticsearchCommon::parse_config() parses the endpoint and extract_auth() calls choose_one() for basic auth before building services, rejecting that same static config; please mirror this credential exclusivity check so no-environment validation does not pass a startup failure.
Useful? React with 👍 / 👎.
| fn validate_structure(&self) -> std::result::Result<(), Vec<String>> { | ||
| let mut errors = Vec::new(); |
There was a problem hiding this comment.
When a Doris endpoint contains basic auth credentials and the top-level auth option is also set, this validator accepts the config after only checking that endpoints are non-empty and confining templates. DorisCommon::parse_config() then calls config.auth.choose_one(&endpoint.auth)? before building the client and rejects the duplicate credentials, so please mirror that check here to keep vector validate --no-environment aligned with startup.
Useful? React with 👍 / 👎.
1 participant
Summary
Implements RFC 2026-07-15 migration for sinks (parallel to PR #25953 which did transforms). Adds
validate_structure()toSinkConfigtrait and wires it into validation so thatvector validate --no-environmentcatches structural errors like template confinement violations that would otherwise silently pass and only fail at boot.Vector configuration
NA
How did you test this PR?
make check-clippy- passesmake fmt- passescargo nextest run -p vector --lib "confinement"- all 69 tests passbuild()confinement logicIs this a breaking change?
Does this PR include user facing changes?
no-changeloglabel to this PR.References
vector validate --no-environmentsilently passing unconfined templates