diff --git a/Cargo.toml b/Cargo.toml index 0a577cde861d9..83e28a55332de 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -904,7 +904,7 @@ sinks-mqtt = ["dep:rumqttc"] sinks-nats = ["dep:async-nats", "dep:nkeys"] sinks-new_relic_logs = ["sinks-http"] sinks-new_relic = [] -sinks-opentelemetry = ["sinks-http", "codecs-opentelemetry"] +sinks-opentelemetry = ["sinks-http", "codecs-opentelemetry", "dep:tonic", "dep:tonic-health", "dep:prost"] sinks-papertrail = ["dep:syslog"] sinks-prometheus = ["dep:base64", "dep:prost", "vector-lib/prometheus"] sinks-postgres = ["dep:sqlx"] @@ -1025,7 +1025,7 @@ mongodb_metrics-integration-tests = ["sources-mongodb_metrics"] mqtt-integration-tests = ["sinks-mqtt", "sources-mqtt"] nats-integration-tests = ["sinks-nats", "sources-nats"] nginx-integration-tests = ["sources-nginx_metrics"] -opentelemetry-integration-tests = ["sources-opentelemetry", "dep:prost"] +opentelemetry-integration-tests = ["sources-opentelemetry", "sinks-opentelemetry", "dep:prost"] postgresql_metrics-integration-tests = ["sources-postgresql_metrics"] postgres_sink-integration-tests = ["sinks-postgres"] prometheus-integration-tests = ["sinks-prometheus", "sources-prometheus", "sinks-influxdb"] diff --git a/changelog.d/opentelemetry_sink_config.breaking.md b/changelog.d/opentelemetry_sink_config.breaking.md new file mode 100644 index 0000000000000..b1cf6003d9027 --- /dev/null +++ b/changelog.d/opentelemetry_sink_config.breaking.md @@ -0,0 +1,49 @@ +Changed the `opentelemetry` sink config fields to remove `protocol.*`. `protocol.type` was replaced +by `protocol` and all fields previously nested under `protocol` now can be placed in the top level +configuration. + +Before: + +```yaml +sinks: + otel_sink: + inputs: + - in + protocol: + type: http + uri: http://otel-collector-sink:5318/v1/logs + method: post + encoding: + codec: json + framing: + method: newline_delimited + batch: + max_events: 1 + request: + headers: + content-type: application/json +``` + +After: + +```yaml +sinks: + otel_sink: + inputs: + - in + type: opentelemetry + protocol: http + uri: http://otel-collector-sink:5318/v1/logs + method: post + encoding: + codec: json + framing: + method: newline_delimited + batch: + max_events: 1 + request: + headers: + content-type: application/json +``` + +authors: thomasqueirozb diff --git a/changelog.d/opentelemetry_sink_grpc.feature.md b/changelog.d/opentelemetry_sink_grpc.feature.md new file mode 100644 index 0000000000000..78cc87a94b717 --- /dev/null +++ b/changelog.d/opentelemetry_sink_grpc.feature.md @@ -0,0 +1,3 @@ +Added gRPC transport support for the `opentelemetry` sink. Configure with `protocol = "grpc"` and set `uri` to your OTLP/gRPC endpoint (e.g. `http://localhost:4317`). Supports TLS and gzip compression. + +authors: thomasqueirozb diff --git a/lib/codecs/src/encoding/format/otlp.rs b/lib/codecs/src/encoding/format/otlp.rs index fd0ffffe47b04..bb06c5d1cf4ae 100644 --- a/lib/codecs/src/encoding/format/otlp.rs +++ b/lib/codecs/src/encoding/format/otlp.rs @@ -107,9 +107,14 @@ impl Encoder for OtlpSerializer { } else if log.contains(RESOURCE_METRICS_JSON_FIELD) { // Currently the OTLP metrics are Vector logs (not metrics). self.metrics_descriptor.encode(event, buffer) + } else if log.contains(RESOURCE_SPANS_JSON_FIELD) { + // OTLP spans can arrive as Log events when the source is configured + // without use_otlp_decoding.traces = true. + self.traces_descriptor.encode(event, buffer) } else { Err(format!( - "Log event does not contain OTLP top-level fields ({RESOURCE_LOGS_JSON_FIELD} or {RESOURCE_METRICS_JSON_FIELD})", + "Log event does not contain OTLP top-level fields \ + ({RESOURCE_LOGS_JSON_FIELD}, {RESOURCE_METRICS_JSON_FIELD}, or {RESOURCE_SPANS_JSON_FIELD})", ) .into()) } diff --git a/src/sinks/opentelemetry/grpc.rs b/src/sinks/opentelemetry/grpc.rs new file mode 100644 index 0000000000000..395454715d65b --- /dev/null +++ b/src/sinks/opentelemetry/grpc.rs @@ -0,0 +1,1055 @@ +use std::{ + fmt, + num::NonZeroUsize, + task::{Context, Poll}, +}; + +use async_trait::async_trait; +use bytes::BytesMut; +use futures::{StreamExt, TryFutureExt, future::BoxFuture, stream::BoxStream}; +use http::Uri; +use hyper::client::HttpConnector; +use hyper_openssl::HttpsConnector; +use hyper_proxy::ProxyConnector; +use prost::Message; +use snafu::Snafu; +use tokio_util::codec::Encoder; +use tonic::body::BoxBody; +use tower::{Service, ServiceBuilder}; +use vector_lib::{ + ByteSizeOf, EstimatedJsonEncodedSizeOf, + codecs::encoding::format::OtlpSerializer, + config::telemetry, + configurable::configurable_component, + internal_event::{ComponentEventsDropped, UNINTENTIONAL}, + opentelemetry::proto::{ + RESOURCE_LOGS_JSON_FIELD, RESOURCE_METRICS_JSON_FIELD, RESOURCE_SPANS_JSON_FIELD, + collector::{ + logs::v1::{ExportLogsServiceRequest, logs_service_client::LogsServiceClient}, + metrics::v1::{ + ExportMetricsServiceRequest, metrics_service_client::MetricsServiceClient, + }, + trace::v1::{ExportTraceServiceRequest, trace_service_client::TraceServiceClient}, + }, + }, + partition::Partitioner, + request_metadata::{GroupedCountByteSize, MetaDescriptive, RequestMetadata}, + stream::{BatcherSettings, DriverResponse}, +}; + +use crate::{ + config::{AcknowledgementsConfig, SinkContext, SinkHealthcheckOptions}, + event::{Event, EventFinalizers, EventStatus, Finalizable}, + http::build_proxy_connector, + internal_events::{EndpointBytesSent, SinkRequestBuildError}, + sinks::util::grpc::{HyperGrpcService, with_default_scheme}, + sinks::{ + Healthcheck, VectorSink, + util::{ + BatchConfig, RealtimeEventBasedDefaultBatchSettings, ServiceBuilderExt, SinkBuilderExt, + StreamSink, http::RequestConfig, metadata::RequestMetadataBuilder, retries::RetryLogic, + }, + }, + template::Template, + tls::{MaybeTlsSettings, TlsConfig}, +}; + +/// Compression codec for gRPC transport. +#[configurable_component] +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +#[serde(rename_all = "snake_case")] +pub enum GrpcCompression { + /// No compression. + #[default] + None, + + /// [Gzip][gzip] compression. + /// + /// [gzip]: https://www.gzip.org/ + Gzip, +} + +/// Configuration for the OpenTelemetry sink's gRPC transport. +#[configurable_component] +#[derive(Clone, Debug)] +pub(super) struct GrpcSinkConfig { + /// The URI to send gRPC requests to. + /// + /// The URI _must_ include a port. If the scheme is omitted, `http` is used unless + /// TLS options are configured, in which case `https` is used. + /// + /// # Examples + /// + /// - `http://localhost:4317` + /// - `https://otelcol.example.com:4317` + #[configurable(metadata(docs::examples = "http://localhost:4317"))] + #[configurable(metadata( + docs::warnings = "When using template syntax, the rendered URI is taken from event data. Only use dynamic URIs with trusted event sources to avoid directing Vector to unintended internal network destinations." + ))] + pub uri: Template, + + #[configurable(derived)] + #[serde(default)] + pub compression: GrpcCompression, + + #[configurable(derived)] + #[serde(default)] + pub batch: BatchConfig, + + #[configurable(derived)] + #[serde(default)] + pub request: RequestConfig, + + #[configurable(derived)] + #[serde(default)] + pub tls: Option, + + #[configurable(derived)] + #[serde( + default, + deserialize_with = "crate::serde::bool_or_struct", + skip_serializing_if = "crate::serde::is_default" + )] + pub acknowledgements: AcknowledgementsConfig, +} + +impl GrpcSinkConfig { + pub(super) async fn build(&self, cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> { + // For static URIs, parse at build time for the healthcheck. + // Dynamic URIs are rendered per-event during sink execution. + let static_uri = if self.uri.is_dynamic() { + None + } else { + Some(with_default_scheme( + self.uri + .get_ref() + .parse() + .map_err(|e| format!("invalid URI for gRPC sink: {e}"))?, + self.tls.is_some(), + )?) + }; + + // The connector is always TLS-capable so it can handle both http:// and https:// + // URIs per-request, matching the behaviour of the HTTP sink. Whether the scheme + // defaults to http or https when the URI has no explicit scheme is controlled by + // whether a `tls:` block is configured. + let tls_configured = self.tls.is_some(); + let tls = MaybeTlsSettings::tls_client(self.tls.as_ref())?; + + let use_gzip = self.compression == GrpcCompression::Gzip; + + // Split headers into static (literal values) and dynamic (template values). + // Static headers are pre-parsed once and used for the healthcheck and every export. + // Dynamic headers are rendered per-event so that templated fields (e.g. tenant IDs) + // resolve correctly at export time. + let (static_header_strings, dynamic_header_templates_raw) = self.request.split_headers(); + + // Static headers are validated at build time and any invalid key or value is a hard + // error. Silently dropping them would cause every request to be sent without the + // intended metadata (e.g. auth headers), which is worse than failing fast. + let mut static_grpc_headers: Vec<( + tonic::metadata::AsciiMetadataKey, + tonic::metadata::AsciiMetadataValue, + )> = Vec::with_capacity(static_header_strings.len()); + for (k, v) in &static_header_strings { + let k_lower = k.to_lowercase(); + let key = tonic::metadata::AsciiMetadataKey::from_bytes(k_lower.as_bytes()) + .map_err(|e| format!("invalid gRPC metadata key {k:?}: {e}"))?; + let value = tonic::metadata::AsciiMetadataValue::try_from(v.as_str()) + .map_err(|e| format!("invalid gRPC metadata value for key {k:?}: {e}"))?; + static_grpc_headers.push((key, value)); + } + + // Dynamic header key names are known at build time and validated eagerly. + // Values are templated and validated per-event at runtime. + let mut dynamic_grpc_header_templates: Vec<(tonic::metadata::AsciiMetadataKey, Template)> = + Vec::with_capacity(dynamic_header_templates_raw.len()); + for (k, t) in dynamic_header_templates_raw { + let k_lower = k.to_lowercase(); + let key = tonic::metadata::AsciiMetadataKey::from_bytes(k_lower.as_bytes()) + .map_err(|e| format!("invalid gRPC metadata key {k:?}: {e}"))?; + dynamic_grpc_header_templates.push((key, t)); + } + + let client = new_grpc_client(&tls, cx.proxy())?; + // Dynamic headers cannot be rendered without a live event, so the healthcheck cannot + // include them. Rather than running a check that omits required auth metadata (which + // would cause a false failure against a properly secured collector), skip the + // healthcheck entirely when dynamic header templates are present — unless the operator + // has supplied an explicit `healthcheck.uri` override, in which case we honour that URI + // with only the static headers (the override endpoint does not require the dynamic auth). + // Prefer an explicit healthcheck.uri override, then fall back to the static sink URI. + // with_default_scheme is applied once to the resolved value. + let raw_healthcheck_uri = cx + .healthcheck + .uri + .clone() + .map(|u| u.uri) + .or_else(|| static_uri.clone()); + let healthcheck_uri = if !dynamic_grpc_header_templates.is_empty() + && raw_healthcheck_uri.is_none() + { + warn!( + "Skipping gRPC healthcheck: dynamic (templated) headers are configured and \ + cannot be rendered at startup. Set a static `healthcheck.uri` override to \ + re-enable the healthcheck." + ); + None + } else if raw_healthcheck_uri.is_none() && self.uri.is_dynamic() { + warn!( + "Skipping gRPC healthcheck: `uri` is a dynamic template and no static \ + `healthcheck.uri` override is set. To enable healthchecking, use a static URI." + ); + None + } else { + raw_healthcheck_uri + .map(|u| with_default_scheme(u, tls_configured)) + .transpose()? + }; + let healthcheck = Box::pin(grpc_healthcheck( + client.clone(), + healthcheck_uri, + static_grpc_headers.clone(), + cx.healthcheck, + )); + let service = OtlpGrpcService::new(client, use_gzip, static_grpc_headers); + + let request_settings = self.request.tower.into_settings(); + let batch_settings = self.batch.into_batcher_settings()?; + + let service = ServiceBuilder::new() + .settings(request_settings, OtlpGrpcRetryLogic) + .service(service); + + let sink = OtlpGrpcSink { + batch_settings, + service, + dynamic_header_templates: dynamic_grpc_header_templates, + uri_template: self.uri.clone(), + tls_configured, + }; + + Ok((VectorSink::from_event_streamsink(sink), healthcheck)) + } +} + +fn new_grpc_client( + tls_settings: &MaybeTlsSettings, + proxy_config: &crate::config::ProxyConfig, +) -> crate::Result>, BoxBody>> { + let proxy = build_proxy_connector(tls_settings.clone(), proxy_config)?; + Ok(hyper::Client::builder().http2_only(true).build(proxy)) +} + +async fn grpc_healthcheck( + client: hyper::Client>, BoxBody>, + uri: Option, + headers: Vec<( + tonic::metadata::AsciiMetadataKey, + tonic::metadata::AsciiMetadataValue, + )>, + options: SinkHealthcheckOptions, +) -> crate::Result<()> { + if !options.enabled { + return Ok(()); + } + + let Some(uri) = uri else { + return Ok(()); + }; + + use tonic::Code; + use tonic_health::pb::{HealthCheckRequest, health_client::HealthClient}; + + let svc = HyperSvc::new(uri, client); + let mut health_client = HealthClient::new(svc); + + let mut req = tonic::Request::new(HealthCheckRequest { + service: String::new(), + }); + for (key, value) in headers { + req.metadata_mut().insert(key, value); + } + + match health_client.check(req).await { + Ok(response) => { + use tonic_health::pb::health_check_response::ServingStatus; + let status = response.into_inner().status; + if status == ServingStatus::Serving as i32 { + Ok(()) + } else { + Err(format!("gRPC collector reported non-serving status: {status}").into()) + } + } + // Server is reachable but does not implement the health protocol; treat as healthy. + Err(status) if status.code() == Code::Unimplemented => Ok(()), + Err(status) => Err(Box::new(OtlpGrpcError::Request { source: status })), + } +} + +// ── Retry logic ────────────────────────────────────────────────────────────── + +#[derive(Debug, Clone)] +struct OtlpGrpcRetryLogic; + +impl RetryLogic for OtlpGrpcRetryLogic { + type Error = OtlpGrpcError; + type Request = OtlpGrpcRequest; + type Response = OtlpGrpcResponse; + + fn is_retriable_error(&self, err: &Self::Error) -> bool { + use tonic::Code::*; + + match err { + OtlpGrpcError::Request { source } => !matches!( + source.code(), + // List taken from + // + NotFound + | InvalidArgument + | AlreadyExists + | PermissionDenied + | OutOfRange + | Unimplemented + | Unauthenticated + // DataLoss: per gRPC spec this means unrecoverable data corruption, not + // retriable. Note that the OTLP partial-success model can also surface + // DataLoss for a partial write; we treat it as non-retriable (consistent with + // the vector sink) to avoid sending an already-accepted partial batch twice. + | DataLoss + ), + } + } +} + +// ── Error type ─────────────────────────────────────────────────────────────── + +#[derive(Debug, Snafu)] +enum OtlpGrpcError { + #[snafu(display("gRPC request failed: {source}"))] + Request { source: tonic::Status }, +} + +// ── Request/Response ───────────────────────────────────────────────────────── + +/// A single-signal OTLP export request. One request per signal type ensures +/// that retries are atomic: a failed metrics export cannot duplicate a +/// previously-accepted logs export. +#[derive(Clone)] +struct OtlpGrpcRequest { + signal: OtlpSignal, + uri: Uri, + /// Per-event rendered values for dynamic (templated) metadata headers. + dynamic_headers: Vec<( + tonic::metadata::AsciiMetadataKey, + tonic::metadata::AsciiMetadataValue, + )>, + finalizers: EventFinalizers, + metadata: RequestMetadata, +} + +impl Finalizable for OtlpGrpcRequest { + fn take_finalizers(&mut self) -> EventFinalizers { + self.finalizers.take_finalizers() + } +} + +impl MetaDescriptive for OtlpGrpcRequest { + fn get_metadata(&self) -> &RequestMetadata { + &self.metadata + } + + fn metadata_mut(&mut self) -> &mut RequestMetadata { + &mut self.metadata + } +} + +struct OtlpGrpcResponse { + events_byte_size: GroupedCountByteSize, + /// True when the collector returned a partial-success response with one or more rejected + /// records. The entire batch is marked `Rejected` so that upstream sources are notified of + /// data loss. This may cause the accepted portion to be re-sent if the source retries, but + /// silent data loss is worse than potential duplication. + had_partial_success: bool, +} + +impl DriverResponse for OtlpGrpcResponse { + fn event_status(&self) -> EventStatus { + if self.had_partial_success { + EventStatus::Rejected + } else { + EventStatus::Delivered + } + } + + fn events_sent(&self) -> &GroupedCountByteSize { + &self.events_byte_size + } +} + +// ── Service ────────────────────────────────────────────────────────────────── + +#[derive(Clone)] +struct CachedClients { + uri: Uri, + logs: LogsServiceClient, + metrics: MetricsServiceClient, + traces: TraceServiceClient, +} + +/// Holds whichever tonic client is needed for a single `OtlpGrpcRequest`. Only the +/// relevant client is cloned out of `CachedClients`, avoiding two unnecessary clones +/// per `Service::call` invocation. +enum SignalClient { + Logs(LogsServiceClient), + Metrics(MetricsServiceClient), + Traces(TraceServiceClient), +} + +#[derive(Clone)] +struct OtlpGrpcService { + /// Tonic clients for the most-recently-seen URI; rebuilt when the rendered URI changes. + /// Each Tower concurrency-slot clone owns its cache independently — no shared mutex needed. + clients: Option, + hyper_client: hyper::Client>, BoxBody>, + compression: bool, + headers: std::sync::Arc< + Vec<( + tonic::metadata::AsciiMetadataKey, + tonic::metadata::AsciiMetadataValue, + )>, + >, +} + +impl OtlpGrpcService { + fn new( + hyper_client: hyper::Client>, BoxBody>, + compression: bool, + headers: Vec<( + tonic::metadata::AsciiMetadataKey, + tonic::metadata::AsciiMetadataValue, + )>, + ) -> Self { + Self { + clients: None, + hyper_client, + compression, + headers: std::sync::Arc::new(headers), + } + } + + /// Rebuilds cached tonic clients if the URI changed, then returns the single client + /// matching `signal`. Only that one client is cloned. + fn client_for_signal(&mut self, uri: &Uri, signal: &OtlpSignal) -> SignalClient { + if self.clients.as_ref().is_none_or(|c| &c.uri != uri) { + let svc = HyperSvc::new(uri.clone(), self.hyper_client.clone()); + let mut logs = LogsServiceClient::new(svc.clone()); + let mut metrics = MetricsServiceClient::new(svc.clone()); + let mut traces = TraceServiceClient::new(svc); + if self.compression { + logs = logs.send_compressed(tonic::codec::CompressionEncoding::Gzip); + metrics = metrics.send_compressed(tonic::codec::CompressionEncoding::Gzip); + traces = traces.send_compressed(tonic::codec::CompressionEncoding::Gzip); + } + self.clients = Some(CachedClients { + uri: uri.clone(), + logs, + metrics, + traces, + }); + } + let c = self.clients.as_ref().expect("just populated"); + match signal { + OtlpSignal::Logs(_) => SignalClient::Logs(c.logs.clone()), + OtlpSignal::Metrics(_) => SignalClient::Metrics(c.metrics.clone()), + OtlpSignal::Traces(_) => SignalClient::Traces(c.traces.clone()), + } + } +} + +impl Service for OtlpGrpcService { + type Response = OtlpGrpcResponse; + type Error = crate::Error; + type Future = BoxFuture<'static, Result>; + + fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + + fn call(&mut self, mut req: OtlpGrpcRequest) -> Self::Future { + let (protocol, endpoint) = crate::sinks::util::uri::protocol_endpoint(req.uri.clone()); + // Rebuild clients if the URI changed; clone only the client for this signal type. + let client = self.client_for_signal(&req.uri, &req.signal); + let static_headers = std::sync::Arc::clone(&self.headers); + let metadata = std::mem::take(req.metadata_mut()); + let events_byte_size = metadata.into_events_estimated_json_encoded_byte_size(); + + let future = async move { + macro_rules! export { + ($client:expr, $payload:expr) => {{ + let len = $payload.encoded_len(); + let mut grpc_req = tonic::Request::new($payload); + for (key, value) in static_headers.iter() { + grpc_req.metadata_mut().insert(key.clone(), value.clone()); + } + for (key, value) in &req.dynamic_headers { + grpc_req.metadata_mut().insert(key.clone(), value.clone()); + } + let response = $client + .export(grpc_req) + .map_err(|source| OtlpGrpcError::Request { source }) + .await?; + (len, response.into_inner()) + }}; + } + + // The signal variant and the client type are always aligned — both are derived + // from `req.signal` in `client_for_signal` — so the mixed arms are unreachable. + let (byte_size, had_partial_success) = match (req.signal, client) { + (OtlpSignal::Logs(r), SignalClient::Logs(mut c)) => { + let (len, resp) = export!(c, r); + let partial = if let Some(ps) = resp.partial_success + && ps.rejected_log_records > 0 + { + warn!( + rejected = ps.rejected_log_records, + message = ps.error_message, + "OTLP collector rejected log records" + ); + emit!(ComponentEventsDropped:: { + count: ps.rejected_log_records as usize, + reason: if ps.error_message.is_empty() { + "OTLP partial success rejection" + } else { + &ps.error_message + }, + }); + true + } else { + false + }; + (len, partial) + } + (OtlpSignal::Metrics(r), SignalClient::Metrics(mut c)) => { + let (len, resp) = export!(c, r); + let partial = if let Some(ps) = resp.partial_success + && ps.rejected_data_points > 0 + { + warn!( + rejected = ps.rejected_data_points, + message = ps.error_message, + "OTLP collector rejected metric data points" + ); + emit!(ComponentEventsDropped:: { + count: ps.rejected_data_points as usize, + reason: if ps.error_message.is_empty() { + "OTLP partial success rejection" + } else { + &ps.error_message + }, + }); + true + } else { + false + }; + (len, partial) + } + (OtlpSignal::Traces(r), SignalClient::Traces(mut c)) => { + let (len, resp) = export!(c, r); + let partial = if let Some(ps) = resp.partial_success + && ps.rejected_spans > 0 + { + warn!( + rejected = ps.rejected_spans, + message = ps.error_message, + "OTLP collector rejected trace spans" + ); + emit!(ComponentEventsDropped:: { + count: ps.rejected_spans as usize, + reason: if ps.error_message.is_empty() { + "OTLP partial success rejection" + } else { + &ps.error_message + }, + }); + true + } else { + false + }; + (len, partial) + } + _ => unreachable!("signal variant and cached client are always aligned"), + }; + + emit!(EndpointBytesSent { + byte_size, + protocol: &protocol, + endpoint: &endpoint, + }); + + Ok(OtlpGrpcResponse { + events_byte_size, + had_partial_success, + }) + }; + + Box::pin(future.map_err(|err: OtlpGrpcError| -> crate::Error { Box::new(err) })) + } +} + +type HyperSvc = HyperGrpcService; + +// ── Sink ───────────────────────────────────────────────────────────────────── + +/// Partition key for gRPC batches. Events that render to different URIs or +/// dynamic headers must be sent in separate requests, so they are batched +/// independently. The key stores rendered values as strings so it can derive +/// `Hash + Eq` without requiring those traits from `Uri` or tonic types. +#[derive(Clone, Debug, Eq, PartialEq, Hash)] +struct BatchPartitionKey { + uri: String, + /// Dynamic header (key, rendered-value) pairs, sorted for deterministic equality. + headers: Vec<(String, String)>, +} + +struct GrpcPartitioner; + +impl Partitioner for GrpcPartitioner { + type Item = OtlpEventData; + type Key = BatchPartitionKey; + + fn partition(&self, item: &OtlpEventData) -> BatchPartitionKey { + let mut headers: Vec<(String, String)> = item + .dynamic_headers + .iter() + .map(|(k, v)| (k.to_string(), v.to_str().unwrap_or("").to_owned())) + .collect(); + headers.sort_unstable(); + BatchPartitionKey { + uri: item.uri.to_string(), + headers, + } + } +} + +/// Intermediate event data extracted before batching. +struct OtlpEventData { + byte_size: usize, + json_byte_size: GroupedCountByteSize, + finalizers: EventFinalizers, + signal: OtlpSignal, + uri: Uri, + dynamic_headers: Vec<( + tonic::metadata::AsciiMetadataKey, + tonic::metadata::AsciiMetadataValue, + )>, +} + +impl ByteSizeOf for OtlpEventData { + fn size_of(&self) -> usize { + std::mem::size_of::() + self.allocated_bytes() + } + + fn allocated_bytes(&self) -> usize { + self.byte_size + } +} + +/// OTLP signal payload. Used as per-event intermediate data before batching and as the +/// request payload sent via gRPC after batching. Each variant corresponds to one signal +/// type so that requests can be retried independently. +#[derive(Clone)] +enum OtlpSignal { + Logs(ExportLogsServiceRequest), + Metrics(ExportMetricsServiceRequest), + Traces(ExportTraceServiceRequest), +} + +/// Per-signal accumulator tracking the merged proto request and its associated +/// event metadata. Kept separate so each signal can be retried independently. +struct SignalData { + request: R, + finalizers: EventFinalizers, + event_count: usize, + byte_size: usize, + json_byte_size: GroupedCountByteSize, +} + +impl SignalData { + const fn new( + request: R, + finalizers: EventFinalizers, + byte_size: usize, + json_byte_size: GroupedCountByteSize, + ) -> Self { + SignalData { + request, + finalizers, + event_count: 1, + byte_size, + json_byte_size, + } + } + + fn merge( + &mut self, + req: R, + finalizers: EventFinalizers, + byte_size: usize, + json_byte_size: GroupedCountByteSize, + merge_req: impl FnOnce(&mut R, R), + ) { + merge_req(&mut self.request, req); + self.finalizers.merge(finalizers); + self.event_count += 1; + self.byte_size += byte_size; + self.json_byte_size += json_byte_size; + } +} + +/// Accumulator for a batch of OTLP events, separated by signal type so that +/// the resulting requests can be retried independently. +#[derive(Default)] +struct OtlpBatch { + logs: Option>, + metrics: Option>, + traces: Option>, +} + +impl OtlpBatch { + fn push(&mut self, item: OtlpEventData) { + let OtlpEventData { + byte_size, + json_byte_size, + finalizers, + signal, + uri: _, + dynamic_headers: _, + } = item; + match signal { + OtlpSignal::Logs(req) => match &mut self.logs { + Some(existing) => { + existing.merge(req, finalizers, byte_size, json_byte_size, |e, r| { + e.resource_logs.extend(r.resource_logs) + }) + } + slot => *slot = Some(SignalData::new(req, finalizers, byte_size, json_byte_size)), + }, + OtlpSignal::Metrics(req) => match &mut self.metrics { + Some(existing) => { + existing.merge(req, finalizers, byte_size, json_byte_size, |e, r| { + e.resource_metrics.extend(r.resource_metrics) + }) + } + slot => *slot = Some(SignalData::new(req, finalizers, byte_size, json_byte_size)), + }, + OtlpSignal::Traces(req) => match &mut self.traces { + Some(existing) => { + existing.merge(req, finalizers, byte_size, json_byte_size, |e, r| { + e.resource_spans.extend(r.resource_spans) + }) + } + slot => *slot = Some(SignalData::new(req, finalizers, byte_size, json_byte_size)), + }, + } + } +} + +struct OtlpGrpcSink { + batch_settings: BatcherSettings, + service: S, + uri_template: Template, + /// Used only to pick the default scheme (`http` vs `https`) when the rendered URI + /// has no explicit scheme. The connector itself handles both schemes. + tls_configured: bool, + dynamic_header_templates: Vec<(tonic::metadata::AsciiMetadataKey, Template)>, +} + +impl OtlpGrpcSink +where + S: Service + Send + 'static, + S::Future: Send + 'static, + S::Response: DriverResponse + Send + 'static, + S::Error: fmt::Debug + Into + Send, +{ + async fn run_inner(self: Box, input: BoxStream<'_, Event>) -> Result<(), ()> { + let mut serializer = OtlpSerializer::new().map_err(|e| { + error!("Failed to create OTLP serializer: {}", e); + })?; + + let uri_template = self.uri_template.clone(); + let tls_configured = self.tls_configured; + let dynamic_header_templates = self.dynamic_header_templates.clone(); + + input + .filter_map(move |mut event| { + let rendered = match uri_template.render_string(&event) { + Ok(s) => s, + Err(e) => return reject_event(event.take_finalizers(), format!("failed to render gRPC URI template: {e}")), + }; + let parsed = match rendered.parse::() { + Ok(u) => u, + Err(e) => return reject_event(event.take_finalizers(), format!("failed to parse rendered gRPC URI: {e}")), + }; + let uri = match with_default_scheme(parsed, tls_configured) { + Ok(u) => u, + Err(e) => return reject_event(event.take_finalizers(), format!("invalid gRPC URI after rendering template: {e}")), + }; + + // Render dynamic headers. If any required header fails to render or produces a + // non-ASCII value, drop the entire event rather than forwarding without it. + // Silently omitting a header (e.g. X-Tenant-ID) could bypass authorization on + // the receiving collector. + let mut dynamic_headers = Vec::with_capacity(dynamic_header_templates.len()); + for (key, template) in &dynamic_header_templates { + match template.render_string(&event) { + Ok(rendered) => { + match tonic::metadata::AsciiMetadataValue::try_from(rendered.as_str()) { + Ok(value) => dynamic_headers.push((key.clone(), value)), + Err(e) => { + return reject_event(event.take_finalizers(), format!( + "gRPC metadata value for key {:?} is not valid ASCII: {e}", + key.as_str() + )); + } + } + } + Err(e) => { + return reject_event(event.take_finalizers(), format!( + "failed to render gRPC metadata template for key {:?}: {e}", + key.as_str() + )); + } + } + } + + // TODO(perf): OtlpSerializer only exposes a byte-level Encoder interface, so + // encode_event must encode to bytes and then decode back to a typed proto struct. + // Exposing a typed output from OtlpSerializer would eliminate this roundtrip. + let signal = encode_event(&mut serializer, &event); + match signal { + Ok(Some(signal)) => { + let mut json_byte_size = telemetry().create_request_count_byte_size(); + json_byte_size.add_event(&event, event.estimated_json_encoded_size_of()); + let data = OtlpEventData { + byte_size: event.size_of(), + json_byte_size, + finalizers: event.take_finalizers(), + signal, + uri, + dynamic_headers, + }; + futures::future::ready(Some(data)) + } + Ok(None) => { + // Event does not contain OTLP structure (missing resourceLogs, + // resourceMetrics, or resourceSpans field). This sink only accepts + // OTLP-structured events; plain log events from non-OTLP sources + // must be encoded with an OTLP codec before reaching this sink. + reject_event( + event.take_finalizers(), + "event is not OTLP-encoded (missing resourceLogs, resourceMetrics, \ + or resourceSpans field); this sink only accepts OTLP-structured events", + ) + } + Err(e) => reject_event(event.take_finalizers(), e.to_string()), + } + }) + // Partition by (rendered URI, dynamic headers) so that events destined for + // different collectors or with different tenant metadata are never merged into + // the same batch. Each partition is independently flushed by size or timeout. + // + // `as_byte_size_config` is intentional here: the primary flush triggers are the + // event-count limit (default 1000, from `RealtimeEventBasedDefaultBatchSettings`) + // and the 1s timeout. The byte limit defaults to `usize::MAX` (effectively + // disabled) because `MAX_BYTES = None` for that settings type. `as_byte_size_config` + // selects `ByteSizeOf` as the size metric so that an operator-configured + // `batch.max_bytes` is measured consistently with the rest of Vector. + .batched_partitioned( + GrpcPartitioner, + self.batch_settings.timeout, + |_| self.batch_settings.as_byte_size_config(), + ) + .flat_map(|(_, mut items)| { + // All items in a partition share the same URI and dynamic headers by construction; + // take them from the first item rather than re-parsing from the partition key. + let (uri, dynamic_headers) = match items.first() { + Some(first) => (first.uri.clone(), first.dynamic_headers.clone()), + None => return futures::stream::iter(Vec::new()), + }; + + // Reduce the partitioned items into per-signal accumulators. + let mut batch = OtlpBatch::default(); + for item in items.drain(..) { + batch.push(item); + } + + let mut requests = Vec::new(); + + if let Some(data) = batch.logs { + requests.push(signal_into_request(data, OtlpSignal::Logs, &uri, dynamic_headers.clone())); + } + if let Some(data) = batch.metrics { + requests.push(signal_into_request(data, OtlpSignal::Metrics, &uri, dynamic_headers.clone())); + } + if let Some(data) = batch.traces { + requests.push(signal_into_request(data, OtlpSignal::Traces, &uri, dynamic_headers.clone())); + } + + futures::stream::iter(requests) + }) + .into_driver(self.service) + .run() + .await + } +} + +#[async_trait] +impl StreamSink for OtlpGrpcSink +where + S: Service + Send + 'static, + S::Future: Send + 'static, + S::Response: DriverResponse + Send + 'static, + S::Error: fmt::Debug + Into + Send, +{ + async fn run(self: Box, input: BoxStream<'_, Event>) -> Result<(), ()> { + self.run_inner(input).await + } +} + +// ── Helpers ─────────────────────────────────────────────────────────────────── + +/// Encode a Vector event into an OTLP proto signal using [`OtlpSerializer`]. +/// +/// Returns `Ok(None)` for native Vector `Metric` events (not supported by OTLP +/// serializer). Returns `Err` for unexpected encoding failures. +fn encode_event( + serializer: &mut OtlpSerializer, + event: &Event, +) -> Result, vector_common::Error> { + let mut buf = BytesMut::new(); + // Clone the event to pass ownership to the encoder while keeping the original + // for metadata extraction in the caller. The clone is shallow for logs/traces. + match event { + Event::Log(log) if log.contains(RESOURCE_LOGS_JSON_FIELD) => { + serializer.encode(event.clone(), &mut buf)?; + Ok(Some(OtlpSignal::Logs(ExportLogsServiceRequest::decode( + buf.freeze(), + )?))) + } + Event::Log(log) if log.contains(RESOURCE_METRICS_JSON_FIELD) => { + serializer.encode(event.clone(), &mut buf)?; + Ok(Some(OtlpSignal::Metrics( + ExportMetricsServiceRequest::decode(buf.freeze())?, + ))) + } + // OTLP spans can arrive as Log events when the source does not use + // use_otlp_decoding.traces = true. + Event::Log(log) if log.contains(RESOURCE_SPANS_JSON_FIELD) => { + serializer.encode(event.clone(), &mut buf)?; + Ok(Some(OtlpSignal::Traces(ExportTraceServiceRequest::decode( + buf.freeze(), + )?))) + } + Event::Trace(trace) if trace.contains(RESOURCE_SPANS_JSON_FIELD) => { + serializer.encode(event.clone(), &mut buf)?; + Ok(Some(OtlpSignal::Traces(ExportTraceServiceRequest::decode( + buf.freeze(), + )?))) + } + _ => Ok(None), + } +} + +/// Drop an event that could not be prepared for export. +/// +/// Marks `finalizers` as [`EventStatus::Rejected`] so that upstream sources are +/// notified of the failure, then emits [`SinkRequestBuildError`] and +/// [`ComponentEventsDropped`]. Returns a ready future resolving to `None` so +/// the caller can use `return reject_event(...)` directly inside a `filter_map` +/// closure. +fn reject_event( + finalizers: EventFinalizers, + reason: impl fmt::Display, +) -> futures::future::Ready> { + let reason = reason.to_string(); + finalizers.update_status(EventStatus::Rejected); + emit!(SinkRequestBuildError { error: &reason }); + emit!(ComponentEventsDropped:: { + count: 1, + reason: &reason, + }); + futures::future::ready(None) +} + +/// Build an [`OtlpGrpcRequest`] from a per-signal accumulator. +fn signal_into_request( + data: SignalData, + make_signal: fn(R) -> OtlpSignal, + uri: &Uri, + dynamic_headers: Vec<( + tonic::metadata::AsciiMetadataKey, + tonic::metadata::AsciiMetadataValue, + )>, +) -> OtlpGrpcRequest { + let byte_size = data.request.encoded_len(); + let bytes_len = NonZeroUsize::new(byte_size.max(1)).expect("should be non-zero"); + let builder = + RequestMetadataBuilder::new(data.event_count, data.byte_size, data.json_byte_size); + OtlpGrpcRequest { + signal: make_signal(data.request), + uri: uri.clone(), + dynamic_headers, + finalizers: data.finalizers, + metadata: builder.with_request_size(bytes_len), + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::sinks::util::grpc::with_default_scheme; + + #[test] + fn generate_grpc_config() { + let config: GrpcSinkConfig = toml::from_str( + r#" + uri = "http://localhost:4317" + "#, + ) + .unwrap(); + assert_eq!(config.uri.get_ref(), "http://localhost:4317"); + assert_eq!(config.compression, GrpcCompression::default()); + } + + #[test] + fn grpc_config_with_gzip() { + let config: GrpcSinkConfig = toml::from_str( + r#" + uri = "https://otelcol.example.com:4317" + compression = "gzip" + "#, + ) + .unwrap(); + assert_eq!(config.uri.get_ref(), "https://otelcol.example.com:4317"); + assert_eq!(config.compression, GrpcCompression::Gzip); + } + + #[test] + fn with_default_scheme_adds_http() { + let uri = with_default_scheme("localhost:4317".parse().unwrap(), false).unwrap(); + assert_eq!(uri.scheme_str(), Some("http")); + } + + #[test] + fn with_default_scheme_adds_https() { + let uri = with_default_scheme("localhost:4317".parse().unwrap(), true).unwrap(); + assert_eq!(uri.scheme_str(), Some("https")); + } + + #[test] + fn with_default_scheme_preserves_existing() { + let uri = with_default_scheme("http://localhost:4317".parse().unwrap(), true).unwrap(); + assert_eq!(uri.scheme_str(), Some("http")); + } +} diff --git a/src/sinks/opentelemetry/integration_tests.rs b/src/sinks/opentelemetry/integration_tests.rs new file mode 100644 index 0000000000000..2c535ef642801 --- /dev/null +++ b/src/sinks/opentelemetry/integration_tests.rs @@ -0,0 +1,111 @@ +use futures::stream; +use prost::Message as _; +use vector_lib::{ + codecs::decoding::format::{Deserializer as _, OtlpDeserializer}, + config::LogNamespace, + opentelemetry::proto::{ + collector::logs::v1::ExportLogsServiceRequest, + logs::v1::{LogRecord, ResourceLogs, ScopeLogs}, + resource::v1::Resource, + }, +}; + +use super::OpenTelemetryConfig; +use crate::{ + config::{SinkConfig as _, SinkContext}, + test_util::{ + components::{HTTP_SINK_TAGS, run_and_assert_sink_compliance}, + wait_for_tcp, + }, +}; + +fn sink_grpc_address() -> String { + std::env::var("OTEL_GRPC_SINK_ADDRESS") + .unwrap_or_else(|_| "opentelemetry-collector:4317".to_owned()) +} + +fn otlp_log_event_with_host(host: &str) -> vector_lib::event::Event { + let mut event = otlp_log_event(); + if let vector_lib::event::Event::Log(ref mut log) = event { + log.insert("host", host.to_owned()); + } + event +} + +fn otlp_log_event() -> vector_lib::event::Event { + let req = ExportLogsServiceRequest { + resource_logs: vec![ResourceLogs { + resource: Some(Resource { + attributes: vec![], + dropped_attributes_count: 0, + }), + scope_logs: vec![ScopeLogs { + scope: None, + log_records: vec![LogRecord { + severity_text: "INFO".to_string(), + body: Some(vector_lib::opentelemetry::proto::common::v1::AnyValue { + value: Some( + vector_lib::opentelemetry::proto::common::v1::any_value::Value::StringValue( + "integration test log message".to_string(), + ), + ), + }), + ..Default::default() + }], + schema_url: String::new(), + }], + schema_url: String::new(), + }], + }; + + let bytes = bytes::Bytes::from(req.encode_to_vec()); + let mut events = OtlpDeserializer::default() + .parse(bytes, LogNamespace::Legacy) + .expect("failed to deserialize OTLP log event"); + events.remove(0) +} + +#[tokio::test] +async fn delivers_logs_via_grpc() { + let address = sink_grpc_address(); + wait_for_tcp(address.clone()).await; + + let config: OpenTelemetryConfig = toml::from_str(&format!( + r#" + protocol = "grpc" + uri = "http://{address}" + "# + )) + .unwrap(); + + let (sink, healthcheck) = config.build(SinkContext::default()).await.unwrap(); + healthcheck.await.expect("gRPC healthcheck failed"); + + let events = vec![otlp_log_event()]; + // The gRPC sink emits EndpointBytesSent with the same "endpoint" and "protocol" tags as + // HTTP sinks, so HTTP_SINK_TAGS is the correct compliance set here. + run_and_assert_sink_compliance(sink, stream::iter(events), &HTTP_SINK_TAGS).await; +} + +#[tokio::test] +async fn delivers_logs_via_grpc_template_uri() { + let host = sink_grpc_address(); + wait_for_tcp(host.clone()).await; + + let config: OpenTelemetryConfig = toml::from_str( + r#" + protocol = "grpc" + uri = "http://{{ host }}:4317" + "#, + ) + .unwrap(); + + let (sink, healthcheck) = config.build(SinkContext::default()).await.unwrap(); + // The URI is a dynamic template so there is no static healthcheck URI; the healthcheck + // skips gracefully and returns Ok(()) rather than failing. + healthcheck.await.expect("gRPC healthcheck failed unexpectedly for dynamic URI"); + + // The event carries `host` so the template renders to the collector address. + let events = vec![otlp_log_event_with_host(host.split(':').next().unwrap())]; + run_and_assert_sink_compliance(sink, stream::iter(events), &HTTP_SINK_TAGS).await; +} diff --git a/src/sinks/opentelemetry/mod.rs b/src/sinks/opentelemetry/mod.rs index 7d73e25c4030d..68adccedbad26 100644 --- a/src/sinks/opentelemetry/mod.rs +++ b/src/sinks/opentelemetry/mod.rs @@ -1,70 +1,134 @@ +mod grpc; + +#[cfg(all(test, feature = "opentelemetry-integration-tests"))] +mod integration_tests; + use indoc::indoc; use vector_config::component::GenerateConfig; -use vector_lib::{ - codecs::{ - JsonSerializerConfig, - encoding::{FramingConfig, SerializerConfig}, - }, - configurable::configurable_component, -}; +use vector_lib::configurable::configurable_component; use crate::{ - codecs::{EncodingConfigWithFraming, Transformer}, - config::{AcknowledgementsConfig, Input, SinkConfig, SinkContext}, + codecs::EncodingConfigWithFraming, + config::{AcknowledgementsConfig, DataType, Input, SinkConfig, SinkContext}, + http::Auth, sinks::{ Healthcheck, VectorSink, http::config::{HttpMethod, HttpSinkConfig}, + util::{ + BatchConfig, Compression, RealtimeEventBasedDefaultBatchSettings, + RealtimeSizeBasedDefaultBatchSettings, http::RequestConfig, + }, }, + template::Template, + tls::TlsConfig, }; -/// Configuration for the `OpenTelemetry` sink. -#[configurable_component(sink("opentelemetry", "Deliver OTLP data over HTTP."))] -#[derive(Clone, Debug, Default)] -pub struct OpenTelemetryConfig { - /// Protocol configuration - #[configurable(derived)] - protocol: Protocol, -} +pub use grpc::GrpcCompression; +use grpc::GrpcSinkConfig; -/// The protocol used to send data to OpenTelemetry. -/// Currently only HTTP is supported, but we plan to support gRPC. -/// The proto definitions are defined [here](https://github.com/vectordotdev/vector/blob/master/lib/opentelemetry-proto/src/proto/opentelemetry-proto/opentelemetry/proto/README.md). +/// Transport protocol for the OpenTelemetry sink. #[configurable_component] #[derive(Clone, Debug)] -#[serde(rename_all = "snake_case", tag = "type")] -#[configurable(metadata(docs::enum_tag_description = "The communication protocol."))] -pub enum Protocol { - /// Send data over HTTP. - Http(HttpSinkConfig), +#[serde(tag = "protocol", rename_all = "snake_case")] +#[allow(clippy::large_enum_variant)] +#[configurable(metadata(docs::enum_tag_description = "The transport protocol to use."))] +pub enum OtlpProtocol { + /// Send OTLP data over HTTP. + Http { + /// The HTTP method to use. Defaults to `post`. + #[serde(default)] + method: HttpMethod, + + #[configurable(derived)] + auth: Option, + + /// Encoding configuration. + #[configurable(derived)] + #[serde(flatten)] + encoding: EncodingConfigWithFraming, + + /// A string to prefix the payload with. + #[serde(default)] + payload_prefix: String, + + /// A string to suffix the payload with. + #[serde(default)] + payload_suffix: String, + + #[configurable(derived)] + #[serde(default)] + batch: BatchConfig, + }, + + /// Send OTLP data over gRPC. + Grpc { + #[configurable(derived)] + #[serde(default)] + batch: BatchConfig, + }, } -impl Default for Protocol { - fn default() -> Self { - Protocol::Http(HttpSinkConfig { - encoding: EncodingConfigWithFraming::new( - Some(FramingConfig::NewlineDelimited), - SerializerConfig::Json(JsonSerializerConfig::default()), - Transformer::default(), - ), - uri: Default::default(), - method: HttpMethod::Post, - auth: Default::default(), - compression: Default::default(), - payload_prefix: Default::default(), - payload_suffix: Default::default(), - batch: Default::default(), - request: Default::default(), - tls: Default::default(), - acknowledgements: Default::default(), - }) - } +/// Configuration for the `opentelemetry` sink. +#[configurable_component(sink("opentelemetry", "Deliver OTLP data over HTTP or gRPC."))] +#[derive(Clone, Debug)] +pub struct OpenTelemetryConfig { + /// The transport protocol to use. + #[configurable(derived)] + #[serde(flatten)] + pub protocol: OtlpProtocol, + + /// The URI to send requests to. + /// + /// Supports template syntax (e.g. `http://{{ host }}:4317`). Must include a scheme + /// (`http://` or `https://`) and a port. + /// + /// For the gRPC transport, the template is rendered once per batch using the first event + /// in the batch. + /// + /// # Examples + /// + /// - `http://localhost:5318/v1/logs` (HTTP) + /// - `http://localhost:4317` (gRPC) + #[configurable(metadata(docs::examples = "http://localhost:5318/v1/logs"))] + #[configurable(metadata(docs::examples = "http://localhost:4317"))] + #[configurable(metadata( + docs::warnings = "When using template syntax, the rendered URI is taken from event data. Only use dynamic URIs with trusted event sources to avoid directing Vector to unintended internal network destinations." + ))] + pub uri: Template, + + #[configurable(derived)] + #[configurable(metadata( + docs::warnings = "The `grpc` protocol only supports `none` and `gzip`. Specifying any other algorithm causes Vector to fail at startup." + ))] + #[serde(default)] + pub compression: Compression, + + #[configurable(derived)] + #[configurable(metadata( + docs::description = "Outbound request settings for retry, concurrency, timeout, and headers. \ + For the `grpc` protocol, `request.headers` entries are forwarded as gRPC metadata — use them \ + for authentication (e.g. `authorization: \"Bearer \"`) since the HTTP-only `auth` field \ + is not available for gRPC." + ))] + #[serde(default)] + pub request: RequestConfig, + + #[configurable(derived)] + pub tls: Option, + + #[configurable(derived)] + #[serde( + default, + deserialize_with = "crate::serde::bool_or_struct", + skip_serializing_if = "crate::serde::is_default" + )] + pub acknowledgements: AcknowledgementsConfig, } impl GenerateConfig for OpenTelemetryConfig { fn generate_config() -> toml::Value { toml::from_str(indoc! {r#" - [protocol] - type = "http" + protocol = "http" uri = "http://localhost:5318/v1/logs" encoding.codec = "json" "#}) @@ -77,20 +141,60 @@ impl GenerateConfig for OpenTelemetryConfig { impl SinkConfig for OpenTelemetryConfig { async fn build(&self, cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> { match &self.protocol { - Protocol::Http(config) => config.build(cx).await, + OtlpProtocol::Http { + method, + auth, + encoding, + payload_prefix, + payload_suffix, + batch, + } => { + let config = HttpSinkConfig { + uri: self.uri.clone(), + method: *method, + auth: auth.clone(), + compression: self.compression, + encoding: encoding.clone(), + payload_prefix: payload_prefix.clone(), + payload_suffix: payload_suffix.clone(), + batch: *batch, + request: self.request.clone(), + tls: self.tls.clone(), + acknowledgements: self.acknowledgements, + }; + config.build(cx).await + } + OtlpProtocol::Grpc { batch } => { + let grpc_compression = match self.compression { + Compression::None => GrpcCompression::None, + Compression::Gzip(_) => GrpcCompression::Gzip, + other => return Err(format!( + "gRPC transport only supports 'none' or 'gzip' compression, got '{other}'" + ) + .into()), + }; + let config = GrpcSinkConfig { + uri: self.uri.clone(), + compression: grpc_compression, + batch: *batch, + request: self.request.clone(), + tls: self.tls.clone(), + acknowledgements: self.acknowledgements, + }; + config.build(cx).await + } } } fn input(&self) -> Input { match &self.protocol { - Protocol::Http(config) => config.input(), + OtlpProtocol::Http { encoding, .. } => Input::new(encoding.config().1.input_type()), + OtlpProtocol::Grpc { .. } => Input::new(DataType::Log | DataType::Trace), } } fn acknowledgements(&self) -> &AcknowledgementsConfig { - match self.protocol { - Protocol::Http(ref config) => config.acknowledgements(), - } + &self.acknowledgements } } diff --git a/src/sinks/util/grpc.rs b/src/sinks/util/grpc.rs new file mode 100644 index 0000000000000..4133885dd3abb --- /dev/null +++ b/src/sinks/util/grpc.rs @@ -0,0 +1,133 @@ +use std::task::{Context, Poll}; + +use futures::future::BoxFuture; +use http::{ + Uri, + uri::{Authority, PathAndQuery, Scheme}, +}; +use hyper::client::HttpConnector; +use hyper_openssl::HttpsConnector; +use hyper_proxy::ProxyConnector; +use tonic::body::BoxBody; +use tower::Service; + +/// Adds a default scheme to a URI that lacks one, and validates that the URI has an authority. +/// +/// Returns the URI unchanged if a scheme is already present. Otherwise prepends +/// `https` when `tls` is `true`, or `http` when `false`. Also sets the +/// path-and-query to `/` if missing. +/// +/// Returns an error if the URI has no authority (host), since `HyperGrpcService` +/// requires one and would otherwise panic at request time. +pub(crate) fn with_default_scheme(uri: Uri, tls: bool) -> crate::Result { + let uri = if uri.scheme().is_none() { + let mut parts = uri.into_parts(); + parts.scheme = Some(if tls { Scheme::HTTPS } else { Scheme::HTTP }); + if parts.path_and_query.is_none() { + parts.path_and_query = Some(PathAndQuery::from_static("/")); + } + Uri::from_parts(parts)? + } else { + uri + }; + if uri.authority().is_none() { + return Err(format!( + "gRPC URI {uri:?} has no host; expected \"scheme://host:port\"" + ) + .into()); + } + Ok(uri) +} + +/// A Tower [`Service`] that routes gRPC requests through a Hyper HTTP/2 client, +/// substituting the scheme and authority from a fixed base URI while preserving +/// the path set by tonic. +/// +/// If the configured URI includes a path prefix (e.g. `https://gateway/grpc`), +/// that prefix is prepended to every tonic RPC path so that requests reach the +/// correct backend through a reverse proxy. +/// +/// Used by gRPC sinks that need to send requests to a specific endpoint using a +/// shared Hyper client. +#[derive(Clone, Debug)] +pub struct HyperGrpcService { + scheme: Scheme, + authority: Authority, + /// Path prefix extracted from the configured URI (e.g. `"/grpc"`), or empty + /// string when the URI has no meaningful path. Never ends with `/` so it can + /// be concatenated directly with tonic's `/ServiceName/Method` paths. + path_prefix: String, + pub client: hyper::Client>, BoxBody>, +} + +impl HyperGrpcService { + /// Creates a new [`HyperGrpcService`]. + /// + /// # Panics + /// + /// Panics at construction time if `uri` lacks a scheme or authority. Always supply a + /// URI produced by `with_default_scheme`, which guarantees both components are present. + /// Panicking here (once, at startup) is preferable to panicking inside the hot-path + /// `Service::call` implementation on every request. + pub fn new( + uri: Uri, + client: hyper::Client>, BoxBody>, + ) -> Self { + let scheme = uri + .scheme() + .expect("gRPC service URI must have a scheme — supply a URI from `with_default_scheme`") + .clone(); + let authority = uri + .authority() + .expect("gRPC service URI must have an authority (host:port) — supply a URI from `with_default_scheme`") + .clone(); + // Strip trailing slash so we can concatenate directly with tonic's leading-slash paths. + let path_prefix = uri.path().trim_end_matches('/').to_owned(); + Self { + scheme, + authority, + path_prefix, + client, + } + } +} + +impl Service> for HyperGrpcService { + type Response = hyper::Response; + type Error = hyper::Error; + type Future = BoxFuture<'static, Result>; + + fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + + fn call(&mut self, mut req: hyper::Request) -> Self::Future { + // Tonic sets the RPC method path (e.g. `/pkg.Service/Method`). Prepend any + // configured path prefix so requests reach the right backend through a proxy. + // Falls back to "/" if tonic omits the path (it never does in practice). + let rpc_path = req + .uri() + .path_and_query() + .map(|pq| pq.as_str()) + .unwrap_or("/"); + let full_path: PathAndQuery = if self.path_prefix.is_empty() { + rpc_path + .parse() + .unwrap_or_else(|_| PathAndQuery::from_static("/")) + } else { + format!("{}{}", self.path_prefix, rpc_path) + .parse() + .unwrap_or_else(|_| PathAndQuery::from_static("/")) + }; + let uri = Uri::builder() + .scheme(self.scheme.clone()) + .authority(self.authority.clone()) + .path_and_query(full_path) + .build() + .expect("pre-validated scheme and authority always produce a valid URI"); + + *req.uri_mut() = uri; + + Box::pin(self.client.request(req)) + } +} diff --git a/src/sinks/util/mod.rs b/src/sinks/util/mod.rs index 5d766a9b30780..8487511d605fb 100644 --- a/src/sinks/util/mod.rs +++ b/src/sinks/util/mod.rs @@ -1,5 +1,7 @@ pub mod adaptive_concurrency; pub mod auth; +#[cfg(any(feature = "sinks-opentelemetry", feature = "sinks-vector"))] +pub mod grpc; // https://github.com/mcarton/rust-derivative/issues/112 #[allow(clippy::non_canonical_clone_impl)] pub mod batch; diff --git a/src/sinks/vector/config.rs b/src/sinks/vector/config.rs index 52601115cb67a..17c2cde4fb751 100644 --- a/src/sinks/vector/config.rs +++ b/src/sinks/vector/config.rs @@ -110,7 +110,8 @@ fn default_config(address: &str) -> VectorConfig { impl SinkConfig for VectorConfig { async fn build(&self, cx: SinkContext) -> crate::Result<(VectorSinkType, Healthcheck)> { let tls = MaybeTlsSettings::from_config(self.tls.as_ref(), false)?; - let uri = with_default_scheme(&self.address, tls.is_tls())?; + let uri = + crate::sinks::util::grpc::with_default_scheme(self.address.parse()?, tls.is_tls())?; let client = new_client(&tls, cx.proxy())?; @@ -176,40 +177,6 @@ async fn healthcheck( } } -/// grpc doesn't like an address without a scheme, so we default to http or https if one isn't -/// specified in the address. -pub fn with_default_scheme(address: &str, tls: bool) -> crate::Result { - let uri: Uri = address.parse()?; - if uri.scheme().is_none() { - // Default the scheme to http or https. - let mut parts = uri.into_parts(); - - parts.scheme = if tls { - Some( - "https" - .parse() - .unwrap_or_else(|_| unreachable!("https should be valid")), - ) - } else { - Some( - "http" - .parse() - .unwrap_or_else(|_| unreachable!("http should be valid")), - ) - }; - - if parts.path_and_query.is_none() { - parts.path_and_query = Some( - "/".parse() - .unwrap_or_else(|_| unreachable!("root should be valid")), - ); - } - Ok(Uri::from_parts(parts)?) - } else { - Ok(uri) - } -} - fn new_client( tls_settings: &MaybeTlsSettings, proxy_config: &ProxyConfig, diff --git a/src/sinks/vector/mod.rs b/src/sinks/vector/mod.rs index 4cf882b1d2ee6..03846f2b538e3 100644 --- a/src/sinks/vector/mod.rs +++ b/src/sinks/vector/mod.rs @@ -49,7 +49,8 @@ mod tests { event::{BatchNotifier, BatchStatus}, }; - use super::{config::with_default_scheme, *}; + use super::*; + use crate::sinks::util::grpc::with_default_scheme; use crate::{ config::{SinkConfig as _, SinkContext}, event::Event, @@ -183,11 +184,15 @@ mod tests { #[test] fn test_with_default_scheme() { assert_eq!( - with_default_scheme("0.0.0.0", false).unwrap().to_string(), + with_default_scheme("0.0.0.0".parse().unwrap(), false) + .unwrap() + .to_string(), "http://0.0.0.0/" ); assert_eq!( - with_default_scheme("0.0.0.0", true).unwrap().to_string(), + with_default_scheme("0.0.0.0".parse().unwrap(), true) + .unwrap() + .to_string(), "https://0.0.0.0/" ); } diff --git a/src/sinks/vector/service.rs b/src/sinks/vector/service.rs index 471b3a98ed6b1..407a1cf912737 100644 --- a/src/sinks/vector/service.rs +++ b/src/sinks/vector/service.rs @@ -73,10 +73,7 @@ impl VectorService { compression: bool, ) -> Self { let (protocol, endpoint) = uri::protocol_endpoint(uri.clone()); - let mut proto_client = proto_vector::Client::new(HyperSvc { - uri, - client: hyper_client, - }); + let mut proto_client = proto_vector::Client::new(HyperSvc::new(uri, hyper_client)); if compression { proto_client = proto_client.send_compressed(tonic::codec::CompressionEncoding::Gzip); @@ -132,33 +129,4 @@ impl Service for VectorService { } } -#[derive(Clone, Debug)] -pub struct HyperSvc { - uri: Uri, - client: hyper::Client>, BoxBody>, -} - -impl Service> for HyperSvc { - type Response = hyper::Response; - type Error = hyper::Error; - type Future = BoxFuture<'static, Result>; - - // Emission of an internal event in case of errors is handled upstream by the caller. - fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll> { - Poll::Ready(Ok(())) - } - - // Emission of internal events for errors and dropped events is handled upstream by the caller. - fn call(&mut self, mut req: hyper::Request) -> Self::Future { - let uri = Uri::builder() - .scheme(self.uri.scheme().unwrap().clone()) - .authority(self.uri.authority().unwrap().clone()) - .path_and_query(req.uri().path_and_query().unwrap().clone()) - .build() - .unwrap(); - - *req.uri_mut() = uri; - - Box::pin(self.client.request(req)) - } -} +pub use crate::sinks::util::grpc::HyperGrpcService as HyperSvc; diff --git a/tests/e2e/opentelemetry-logs/config/compose.yaml b/tests/e2e/opentelemetry-logs/config/compose.yaml index 62b3d48c44aa6..d193ae5d9b467 100644 --- a/tests/e2e/opentelemetry-logs/config/compose.yaml +++ b/tests/e2e/opentelemetry-logs/config/compose.yaml @@ -60,6 +60,7 @@ services: target: /output ports: - "${OTEL_COLLECTOR_SINK_HTTP_PORT:-5318}:5318" + - "${OTEL_COLLECTOR_SINK_GRPC_PORT:-14317}:4317" vector: container_name: vector-otel-logs-e2e diff --git a/tests/e2e/opentelemetry-logs/config/test.yaml b/tests/e2e/opentelemetry-logs/config/test.yaml index 54aa7063e632c..95fd08d50aadc 100644 --- a/tests/e2e/opentelemetry-logs/config/test.yaml +++ b/tests/e2e/opentelemetry-logs/config/test.yaml @@ -11,11 +11,12 @@ runner: OTEL_COLLECTOR_SOURCE_GRPC_PORT: '4317' OTEL_COLLECTOR_SOURCE_HTTP_PORT: '4318' OTEL_COLLECTOR_SINK_HTTP_PORT: '5318' + OTEL_COLLECTOR_SINK_GRPC_PORT: '14317' matrix: # Determines which `otel/opentelemetry-collector-contrib` version to use collector_version: [ 'latest' ] - vector_config: [ 'vector_default.yaml', 'vector_otlp.yaml' ] + vector_config: [ 'vector_default.yaml', 'vector_otlp.yaml', 'vector_grpc.yaml' ] # Only trigger this integration test if relevant OTEL source/sink files change paths: diff --git a/tests/e2e/opentelemetry-logs/data/collector-sink.yaml b/tests/e2e/opentelemetry-logs/data/collector-sink.yaml index e22f62f91d558..fbc6cc87827a3 100644 --- a/tests/e2e/opentelemetry-logs/data/collector-sink.yaml +++ b/tests/e2e/opentelemetry-logs/data/collector-sink.yaml @@ -3,6 +3,8 @@ receivers: protocols: http: endpoint: "0.0.0.0:5318" + grpc: + endpoint: "0.0.0.0:4317" processors: batch: { } diff --git a/tests/e2e/opentelemetry-logs/data/vector_default.yaml b/tests/e2e/opentelemetry-logs/data/vector_default.yaml index 1f3fd9b119d09..06f515b5729ec 100644 --- a/tests/e2e/opentelemetry-logs/data/vector_default.yaml +++ b/tests/e2e/opentelemetry-logs/data/vector_default.yaml @@ -51,19 +51,18 @@ sinks: otel_sink: inputs: [ "remap_otel" ] type: opentelemetry - protocol: - type: http - uri: http://otel-collector-sink:5318/v1/logs - method: post - encoding: - codec: json - framing: - method: newline_delimited - batch: - max_events: 1 - request: - headers: - content-type: application/json + protocol: http + uri: http://otel-collector-sink:5318/v1/logs + method: post + encoding: + codec: json + framing: + method: newline_delimited + batch: + max_events: 1 + request: + headers: + content-type: application/json otel_file_sink: type: file diff --git a/tests/e2e/opentelemetry-logs/data/vector_grpc.yaml b/tests/e2e/opentelemetry-logs/data/vector_grpc.yaml new file mode 100644 index 0000000000000..18057cd60d9d7 --- /dev/null +++ b/tests/e2e/opentelemetry-logs/data/vector_grpc.yaml @@ -0,0 +1,39 @@ +sources: + source0: + type: opentelemetry + grpc: + address: 0.0.0.0:4317 + http: + address: 0.0.0.0:4318 + keepalive: + max_connection_age_jitter_factor: 0.1 + max_connection_age_secs: 300 + use_otlp_decoding: true + + internal_metrics: + type: internal_metrics + scrape_interval_secs: 60 + +sinks: + otel_sink: + inputs: + - source0.logs + type: opentelemetry + protocol: grpc + uri: http://otel-collector-sink:4317 + + otel_file_sink: + type: file + path: "/output/opentelemetry-logs/vector-file-sink.log" + inputs: + - source0.logs + encoding: + codec: json + + metrics_file_sink: + type: file + path: "/output/opentelemetry-logs/vector-internal-metrics-sink.log" + inputs: + - internal_metrics + encoding: + codec: json diff --git a/tests/e2e/opentelemetry-logs/data/vector_otlp.yaml b/tests/e2e/opentelemetry-logs/data/vector_otlp.yaml index fb2267ca131ec..b182ad502c900 100644 --- a/tests/e2e/opentelemetry-logs/data/vector_otlp.yaml +++ b/tests/e2e/opentelemetry-logs/data/vector_otlp.yaml @@ -19,11 +19,10 @@ sinks: inputs: - source0.logs type: opentelemetry - protocol: - type: http - uri: http://otel-collector-sink:5318/v1/logs - encoding: - codec: otlp + protocol: http + uri: http://otel-collector-sink:5318/v1/logs + encoding: + codec: otlp otel_file_sink: type: file diff --git a/tests/e2e/opentelemetry-metrics/data/vector_otlp.yaml b/tests/e2e/opentelemetry-metrics/data/vector_otlp.yaml index 0dec0745c3a13..2900326e7d346 100644 --- a/tests/e2e/opentelemetry-metrics/data/vector_otlp.yaml +++ b/tests/e2e/opentelemetry-metrics/data/vector_otlp.yaml @@ -19,11 +19,10 @@ sinks: inputs: - source0.metrics type: opentelemetry - protocol: - type: http - uri: http://otel-collector-sink:5318/v1/metrics - encoding: - codec: otlp + protocol: http + uri: http://otel-collector-sink:5318/v1/metrics + encoding: + codec: otlp otel_file_sink: type: file diff --git a/tests/integration/opentelemetry/config/test.yaml b/tests/integration/opentelemetry/config/test.yaml index 91a2968fdfae0..90d058a281a12 100644 --- a/tests/integration/opentelemetry/config/test.yaml +++ b/tests/integration/opentelemetry/config/test.yaml @@ -7,6 +7,7 @@ runner: env: OTEL_HEALTH_URL: http://opentelemetry-collector:13133 OTEL_OTLPHTTP_URL: http://opentelemetry-collector:9876 + OTEL_GRPC_SINK_ADDRESS: opentelemetry-collector:4317 matrix: version: [0.56.0] @@ -15,5 +16,6 @@ matrix: # expressions are evaluated using https://github.com/micromatch/picomatch paths: - "src/sources/opentelemetry/**" +- "src/sinks/opentelemetry/**" - "src/sources/util/**" - "tests/integration/opentelemetry/**" diff --git a/tests/integration/opentelemetry/data/config.yaml b/tests/integration/opentelemetry/data/config.yaml index 66c8fe34e6112..2a0c9ee30c98c 100644 --- a/tests/integration/opentelemetry/data/config.yaml +++ b/tests/integration/opentelemetry/data/config.yaml @@ -3,6 +3,8 @@ receivers: protocols: http: endpoint: 0.0.0.0:9876 + grpc: + endpoint: 0.0.0.0:4317 exporters: otlp: diff --git a/website/content/en/highlights/2025-09-23-otlp-support.md b/website/content/en/highlights/2025-09-23-otlp-support.md index fa3a1add99544..38f5a51703e1a 100644 --- a/website/content/en/highlights/2025-09-23-otlp-support.md +++ b/website/content/en/highlights/2025-09-23-otlp-support.md @@ -39,12 +39,11 @@ sinks: inputs: - source0.logs type: opentelemetry - protocol: - type: http - uri: http://otel-collector-sink:5318/v1/logs - method: post - encoding: - codec: otlp + protocol: http + uri: http://otel-collector-sink:5318/v1/logs + method: post + encoding: + codec: otlp ``` The above configuration will only work with Vector versions >= `0.51`. @@ -58,21 +57,20 @@ otel_sink: inputs: - otel.logs type: opentelemetry - protocol: - type: http - uri: http://localhost:5318/v1/logs - method: post - encoding: - codec: protobuf - protobuf: - desc_file: path/to/opentelemetry-proto.desc - message_type: opentelemetry.proto.collector.logs.v1.ExportLogsServiceRequest - use_json_names: true - framing: - method: 'bytes' - request: - headers: - content-type: 'application/x-protobuf' + protocol: http + uri: http://localhost:5318/v1/logs + method: post + encoding: + codec: protobuf + protobuf: + desc_file: path/to/opentelemetry-proto.desc + message_type: opentelemetry.proto.collector.logs.v1.ExportLogsServiceRequest + use_json_names: true + framing: + method: 'bytes' + request: + headers: + content-type: 'application/x-protobuf' ``` The `desc` file was generated with the following command: diff --git a/website/cue/reference/components/sinks/generated/opentelemetry.cue b/website/cue/reference/components/sinks/generated/opentelemetry.cue index 4505d8a9bf512..9a3a207767e14 100644 --- a/website/cue/reference/components/sinks/generated/opentelemetry.cue +++ b/website/cue/reference/components/sinks/generated/opentelemetry.cue @@ -1,1140 +1,1138 @@ package metadata -generated: components: sinks: opentelemetry: configuration: protocol: { - description: "Protocol configuration" - required: true - type: object: options: { - acknowledgements: { - description: """ - Controls how acknowledgements are handled for this sink. +generated: components: sinks: opentelemetry: configuration: { + acknowledgements: { + description: """ + Controls how acknowledgements are handled for this sink. - See [End-to-end Acknowledgements][e2e_acks] for more information on how event acknowledgement is handled. + See [End-to-end Acknowledgements][e2e_acks] for more information on how event acknowledgement is handled. - [e2e_acks]: https://vector.dev/docs/architecture/end-to-end-acknowledgements/ - """ - required: false - type: object: options: enabled: { - description: """ - Controls whether or not end-to-end acknowledgements are enabled. + [e2e_acks]: https://vector.dev/docs/architecture/end-to-end-acknowledgements/ + """ + required: false + type: object: options: enabled: { + description: """ + Controls whether or not end-to-end acknowledgements are enabled. - When enabled for a sink, any source that supports end-to-end - acknowledgements that is connected to that sink waits for events - to be acknowledged by **all connected sinks** before acknowledging them at the source. + When enabled for a sink, any source that supports end-to-end + acknowledgements that is connected to that sink waits for events + to be acknowledged by **all connected sinks** before acknowledging them at the source. - Enabling or disabling acknowledgements at the sink level takes precedence over any global - [`acknowledgements`][global_acks] configuration. + Enabling or disabling acknowledgements at the sink level takes precedence over any global + [`acknowledgements`][global_acks] configuration. - [global_acks]: https://vector.dev/docs/reference/configuration/global-options/#acknowledgements - """ - required: false - type: bool: {} - } - } - auth: { - description: """ - Configuration of the authentication strategy for HTTP requests. - - HTTP authentication should be used with HTTPS only, as the authentication credentials are passed as an - HTTP header without any additional encryption beyond what is provided by the transport itself. + [global_acks]: https://vector.dev/docs/reference/configuration/global-options/#acknowledgements """ required: false - type: object: options: { - auth: { - description: "The AWS authentication configuration." - relevant_when: "strategy = \"aws\"" - required: true - type: object: options: { - access_key_id: { - description: "The AWS access key ID." - required: true - type: string: examples: ["AKIAIOSFODNN7EXAMPLE"] - } - assume_role: { - description: """ - The ARN of an [IAM role][iam_role] to assume. - - [iam_role]: https://docs.aws.amazon.com/IAM/latest/UserGuide/id_roles.html - """ - required: true - type: string: examples: ["arn:aws:iam::123456789098:role/my_role"] - } - credentials_file: { - description: "Path to the credentials file." - required: true - type: string: examples: ["/my/aws/credentials"] - } - external_id: { - description: """ - The optional unique external ID in conjunction with role to assume. - - [external_id]: https://docs.aws.amazon.com/IAM/latest/UserGuide/id_roles_create_for-user_externalid.html - """ - required: false - type: string: examples: ["randomEXAMPLEidString"] - } - imds: { - description: "Configuration for authenticating with AWS through IMDS." - required: false - type: object: options: { - connect_timeout_seconds: { - description: "Connect timeout for IMDS." - required: false - type: uint: { - default: 1 - unit: "seconds" - } - } - max_attempts: { - description: "Number of IMDS retries for fetching tokens and metadata." - required: false - type: uint: default: 4 - } - read_timeout_seconds: { - description: "Read timeout for IMDS." - required: false - type: uint: { - default: 1 - unit: "seconds" - } + type: bool: {} + } + } + auth: { + description: """ + Configuration of the authentication strategy for HTTP requests. + + HTTP authentication should be used with HTTPS only, as the authentication credentials are passed as an + HTTP header without any additional encryption beyond what is provided by the transport itself. + """ + relevant_when: "protocol = \"http\"" + required: false + type: object: options: { + auth: { + description: "The AWS authentication configuration." + relevant_when: "strategy = \"aws\"" + required: true + type: object: options: { + access_key_id: { + description: "The AWS access key ID." + required: true + type: string: examples: ["AKIAIOSFODNN7EXAMPLE"] + } + assume_role: { + description: """ + The ARN of an [IAM role][iam_role] to assume. + + [iam_role]: https://docs.aws.amazon.com/IAM/latest/UserGuide/id_roles.html + """ + required: true + type: string: examples: ["arn:aws:iam::123456789098:role/my_role"] + } + credentials_file: { + description: "Path to the credentials file." + required: true + type: string: examples: ["/my/aws/credentials"] + } + external_id: { + description: """ + The optional unique external ID in conjunction with role to assume. + + [external_id]: https://docs.aws.amazon.com/IAM/latest/UserGuide/id_roles_create_for-user_externalid.html + """ + required: false + type: string: examples: ["randomEXAMPLEidString"] + } + imds: { + description: "Configuration for authenticating with AWS through IMDS." + required: false + type: object: options: { + connect_timeout_seconds: { + description: "Connect timeout for IMDS." + required: false + type: uint: { + default: 1 + unit: "seconds" } } - } - load_timeout_secs: { - description: """ - Timeout for successfully loading any credentials, in seconds. - - Relevant when the default credentials chain or `assume_role` is used. - """ - required: false - type: uint: { - examples: [30] - unit: "seconds" + max_attempts: { + description: "Number of IMDS retries for fetching tokens and metadata." + required: false + type: uint: default: 4 } - } - profile: { - description: """ - The credentials profile to use. - - Used to select AWS credentials from a provided credentials file. - """ - required: false - type: string: { - default: "default" - examples: ["develop"] + read_timeout_seconds: { + description: "Read timeout for IMDS." + required: false + type: uint: { + default: 1 + unit: "seconds" + } } } - region: { - description: """ - The [AWS region][aws_region] to send STS requests to. - - If not set, this defaults to the configured region - for the service itself. + } + load_timeout_secs: { + description: """ + Timeout for successfully loading any credentials, in seconds. - [aws_region]: https://docs.aws.amazon.com/general/latest/gr/rande.html#regional-endpoints - """ - required: false - type: string: examples: ["us-west-2"] - } - secret_access_key: { - description: "The AWS secret access key." - required: true - type: string: examples: ["wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"] + Relevant when the default credentials chain or `assume_role` is used. + """ + required: false + type: uint: { + examples: [30] + unit: "seconds" } - session_name: { - description: """ - The optional [RoleSessionName][role_session_name] is a unique session identifier for your assumed role. - - Should be unique per principal or reason. - If not set, the session name is autogenerated like assume-role-provider-1736428351340 + } + profile: { + description: """ + The credentials profile to use. - [role_session_name]: https://docs.aws.amazon.com/STS/latest/APIReference/API_AssumeRole.html - """ - required: false - type: string: examples: ["vector-indexer-role"] - } - session_token: { - description: """ - The AWS session token. - See [AWS temporary credentials](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_credentials_temp_use-resources.html) - """ - required: false - type: string: examples: ["AQoDYXdz...AQoDYXdz..."] + Used to select AWS credentials from a provided credentials file. + """ + required: false + type: string: { + default: "default" + examples: ["develop"] } } - } - password: { - description: "The basic authentication password." - relevant_when: "strategy = \"basic\"" - required: true - type: string: examples: ["${PASSWORD}", "password"] - } - service: { - description: "The AWS service name to use for signing." - relevant_when: "strategy = \"aws\"" - required: true - type: string: {} - } - strategy: { - description: "The authentication strategy to use." - required: true - type: string: enum: { - aws: "AWS authentication." - basic: """ - Basic authentication. + region: { + description: """ + The [AWS region][aws_region] to send STS requests to. - The username and password are concatenated and encoded using [base64][base64]. + If not set, this defaults to the configured region + for the service itself. - [base64]: https://en.wikipedia.org/wiki/Base64 - """ - bearer: """ - Bearer authentication. + [aws_region]: https://docs.aws.amazon.com/general/latest/gr/rande.html#regional-endpoints + """ + required: false + type: string: examples: ["us-west-2"] + } + secret_access_key: { + description: "The AWS secret access key." + required: true + type: string: examples: ["wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"] + } + session_name: { + description: """ + The optional [RoleSessionName][role_session_name] is a unique session identifier for your assumed role. - The bearer token value (OAuth2, JWT, etc.) is passed as-is. - """ - custom: "Custom Authorization Header Value, will be inserted into the headers as `Authorization: < value >`" + Should be unique per principal or reason. + If not set, the session name is autogenerated like assume-role-provider-1736428351340 + + [role_session_name]: https://docs.aws.amazon.com/STS/latest/APIReference/API_AssumeRole.html + """ + required: false + type: string: examples: ["vector-indexer-role"] + } + session_token: { + description: """ + The AWS session token. + See [AWS temporary credentials](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_credentials_temp_use-resources.html) + """ + required: false + type: string: examples: ["AQoDYXdz...AQoDYXdz..."] } - } - token: { - description: "The bearer authentication token." - relevant_when: "strategy = \"bearer\"" - required: true - type: string: {} - } - user: { - description: "The basic authentication username." - relevant_when: "strategy = \"basic\"" - required: true - type: string: examples: ["${USERNAME}", "username"] - } - value: { - description: "Custom string value of the Authorization header" - relevant_when: "strategy = \"custom\"" - required: true - type: string: examples: ["${AUTH_HEADER_VALUE}", "CUSTOM_PREFIX ${TOKEN}"] } } - } - batch: { - description: "Event batching behavior." - required: false - type: object: options: { - max_bytes: { - description: """ - The maximum size of a batch that is processed by a sink. + password: { + description: "The basic authentication password." + relevant_when: "strategy = \"basic\"" + required: true + type: string: examples: ["${PASSWORD}", "password"] + } + service: { + description: "The AWS service name to use for signing." + relevant_when: "strategy = \"aws\"" + required: true + type: string: {} + } + strategy: { + description: "The authentication strategy to use." + required: true + type: string: enum: { + aws: "AWS authentication." + basic: """ + Basic authentication. + + The username and password are concatenated and encoded using [base64][base64]. - This is based on the uncompressed size of the batched events, before they are - serialized or compressed. + [base64]: https://en.wikipedia.org/wiki/Base64 """ - required: false - type: uint: { - default: 10000000 - unit: "bytes" - } + bearer: """ + Bearer authentication. + + The bearer token value (OAuth2, JWT, etc.) is passed as-is. + """ + custom: "Custom Authorization Header Value, will be inserted into the headers as `Authorization: < value >`" } - max_events: { - description: "The maximum size of a batch before it is flushed." - required: false - type: uint: unit: "events" + } + token: { + description: "The bearer authentication token." + relevant_when: "strategy = \"bearer\"" + required: true + type: string: {} + } + user: { + description: "The basic authentication username." + relevant_when: "strategy = \"basic\"" + required: true + type: string: examples: ["${USERNAME}", "username"] + } + value: { + description: "Custom string value of the Authorization header" + relevant_when: "strategy = \"custom\"" + required: true + type: string: examples: ["${AUTH_HEADER_VALUE}", "CUSTOM_PREFIX ${TOKEN}"] + } + } + } + batch: { + description: "Event batching behavior." + required: false + type: object: options: { + max_bytes: { + description: """ + The maximum size of a batch that is processed by a sink. + + This is based on the uncompressed size of the batched events, before they are + serialized or compressed. + """ + required: false + type: uint: { + default: 10000000 + unit: "bytes" } - timeout_secs: { - description: "The maximum age of a batch before it is flushed." - required: false - type: float: { - default: 1.0 - unit: "seconds" - } + } + max_events: { + description: "The maximum size of a batch before it is flushed." + required: false + type: uint: unit: "events" + } + timeout_secs: { + description: "The maximum age of a batch before it is flushed." + required: false + type: float: { + default: 1.0 + unit: "seconds" } } } - compression: { - description: """ - Compression configuration. - - All compression algorithms use the default compression level unless otherwise specified. - """ - required: false - type: string: { - default: "none" - enum: { - gzip: """ - [Gzip][gzip] compression. - - [gzip]: https://www.gzip.org/ - """ - none: "No compression." - snappy: """ - [Snappy][snappy] compression. + } + compression: { + description: """ + Compression configuration. + + All compression algorithms use the default compression level unless otherwise specified. + """ + required: false + type: string: { + default: "none" + enum: { + gzip: """ + [Gzip][gzip] compression. + + [gzip]: https://www.gzip.org/ + """ + none: "No compression." + snappy: """ + [Snappy][snappy] compression. - [snappy]: https://github.com/google/snappy/blob/main/docs/README.md - """ - zlib: """ - [Zlib][zlib] compression. + [snappy]: https://github.com/google/snappy/blob/main/docs/README.md + """ + zlib: """ + [Zlib][zlib] compression. - [zlib]: https://zlib.net/ - """ - zstd: """ - [Zstandard][zstd] compression. + [zlib]: https://zlib.net/ + """ + zstd: """ + [Zstandard][zstd] compression. - [zstd]: https://facebook.github.io/zstd/ - """ - } + [zstd]: https://facebook.github.io/zstd/ + """ } } - encoding: { - description: """ - Encoding configuration. - Configures how events are encoded into raw bytes. - The selected encoding also determines which input types (logs, metrics, traces) are supported. - """ - required: true - type: object: options: { - avro: { - description: "Apache Avro-specific encoder options." - relevant_when: "codec = \"avro\"" - required: true - type: object: options: schema: { - description: "The Avro schema." - required: true - type: string: examples: ["{ \"type\": \"record\", \"name\": \"log\", \"fields\": [{ \"name\": \"message\", \"type\": \"string\" }] }"] - } + warnings: ["The `grpc` protocol only supports `none` and `gzip`. Specifying any other algorithm causes Vector to fail at startup."] + } + encoding: { + description: """ + Encoding configuration. + Configures how events are encoded into raw bytes. + The selected encoding also determines which input types (logs, metrics, traces) are supported. + """ + relevant_when: "protocol = \"http\"" + required: true + type: object: options: { + avro: { + description: "Apache Avro-specific encoder options." + relevant_when: "codec = \"avro\"" + required: true + type: object: options: schema: { + description: "The Avro schema." + required: true + type: string: examples: ["{ \"type\": \"record\", \"name\": \"log\", \"fields\": [{ \"name\": \"message\", \"type\": \"string\" }] }"] } - cef: { - description: "The CEF Serializer Options." - relevant_when: "codec = \"cef\"" - required: true - type: object: options: { - device_event_class_id: { - description: """ - Unique identifier for each event type. Identifies the type of event reported. - The value length must be less than or equal to 1023. - """ - required: true - type: string: {} - } - device_product: { - description: """ - Identifies the product of a vendor. - The part of a unique device identifier. No two products can use the same combination of device vendor and device product. - The value length must be less than or equal to 63. - """ - required: true - type: string: {} - } - device_vendor: { - description: """ - Identifies the vendor of the product. - The part of a unique device identifier. No two products can use the same combination of device vendor and device product. - The value length must be less than or equal to 63. - """ - required: true - type: string: {} - } - device_version: { - description: """ - Identifies the version of the problem. The combination of the device product, vendor, and this value make up the unique id of the device that sends messages. - The value length must be less than or equal to 31. - """ - required: true - type: string: {} - } - extensions: { - description: """ - The collection of key-value pairs. Keys are the keys of the extensions, and values are paths that point to the extension values of a log event. - The event can have any number of key-value pairs in any order. - """ - required: false - type: object: options: "*": { - description: "This is a path that points to the extension value of a log event." - required: true - type: string: {} - } - } - name: { - description: """ - This is a path that points to the human-readable description of a log event. - The value length must be less than or equal to 512. - Equals "cef.name" by default. - """ - required: true - type: string: {} - } - severity: { - description: """ - This is a path that points to the field of a log event that reflects importance of the event. - - It must point to a number from 0 to 10. - 0 = lowest_importance, 10 = highest_importance. - Set to "cef.severity" by default. - """ - required: true + } + cef: { + description: "The CEF Serializer Options." + relevant_when: "codec = \"cef\"" + required: true + type: object: options: { + device_event_class_id: { + description: """ + Unique identifier for each event type. Identifies the type of event reported. + The value length must be less than or equal to 1023. + """ + required: true + type: string: {} + } + device_product: { + description: """ + Identifies the product of a vendor. + The part of a unique device identifier. No two products can use the same combination of device vendor and device product. + The value length must be less than or equal to 63. + """ + required: true + type: string: {} + } + device_vendor: { + description: """ + Identifies the vendor of the product. + The part of a unique device identifier. No two products can use the same combination of device vendor and device product. + The value length must be less than or equal to 63. + """ + required: true + type: string: {} + } + device_version: { + description: """ + Identifies the version of the problem. The combination of the device product, vendor, and this value make up the unique id of the device that sends messages. + The value length must be less than or equal to 31. + """ + required: true + type: string: {} + } + extensions: { + description: """ + The collection of key-value pairs. Keys are the keys of the extensions, and values are paths that point to the extension values of a log event. + The event can have any number of key-value pairs in any order. + """ + required: false + type: object: options: "*": { + description: "This is a path that points to the extension value of a log event." + required: true type: string: {} } - version: { - description: """ - CEF Version. Can be either 0 or 1. - Set to "0" by default. - """ - required: true - type: string: enum: { - V0: "CEF specification version 0.1." - V1: "CEF specification version 1.x." - } + } + name: { + description: """ + This is a path that points to the human-readable description of a log event. + The value length must be less than or equal to 512. + Equals "cef.name" by default. + """ + required: true + type: string: {} + } + severity: { + description: """ + This is a path that points to the field of a log event that reflects importance of the event. + + It must point to a number from 0 to 10. + 0 = lowest_importance, 10 = highest_importance. + Set to "cef.severity" by default. + """ + required: true + type: string: {} + } + version: { + description: """ + CEF Version. Can be either 0 or 1. + Set to "0" by default. + """ + required: true + type: string: enum: { + V0: "CEF specification version 0.1." + V1: "CEF specification version 1.x." } } } - codec: { - description: "The codec to use for encoding events." - required: true - type: string: enum: { - avro: """ - Encodes an event as an [Apache Avro][apache_avro] message. + } + codec: { + description: "The codec to use for encoding events." + required: true + type: string: enum: { + avro: """ + Encodes an event as an [Apache Avro][apache_avro] message. + + [apache_avro]: https://avro.apache.org/ + """ + cef: "Encodes an event as a CEF (Common Event Format) formatted message." + csv: """ + Encodes an event as a CSV message. - [apache_avro]: https://avro.apache.org/ - """ - cef: "Encodes an event as a CEF (Common Event Format) formatted message." - csv: """ - Encodes an event as a CSV message. + This codec must be configured with fields to encode. + """ + gelf: """ + Encodes an event as a [GELF][gelf] message. - This codec must be configured with fields to encode. - """ - gelf: """ - Encodes an event as a [GELF][gelf] message. + This codec is experimental for the following reason: - This codec is experimental for the following reason: + The GELF specification is more strict than the actual Graylog receiver. + Vector's encoder currently adheres more strictly to the GELF spec, with + the exception that some characters such as `@` are allowed in field names. - The GELF specification is more strict than the actual Graylog receiver. - Vector's encoder currently adheres more strictly to the GELF spec, with - the exception that some characters such as `@` are allowed in field names. + Other GELF codecs, such as Loki's, use a [Go SDK][implementation] that is maintained + by Graylog and is much more relaxed than the GELF spec. - Other GELF codecs, such as Loki's, use a [Go SDK][implementation] that is maintained - by Graylog and is much more relaxed than the GELF spec. + Going forward, Vector will use that [Go SDK][implementation] as the reference implementation, which means + the codec might continue to relax the enforcement of the specification. - Going forward, Vector will use that [Go SDK][implementation] as the reference implementation, which means - the codec might continue to relax the enforcement of the specification. + [gelf]: https://docs.graylog.org/docs/gelf + [implementation]: https://github.com/Graylog2/go-gelf/blob/v2/gelf/reader.go + """ + json: """ + Encodes an event as [JSON][json]. - [gelf]: https://docs.graylog.org/docs/gelf - [implementation]: https://github.com/Graylog2/go-gelf/blob/v2/gelf/reader.go - """ - json: """ - Encodes an event as [JSON][json]. + [json]: https://www.json.org/ + """ + logfmt: """ + Encodes an event as a [logfmt][logfmt] message. - [json]: https://www.json.org/ - """ - logfmt: """ - Encodes an event as a [logfmt][logfmt] message. + [logfmt]: https://brandur.org/logfmt + """ + native: """ + Encodes an event in the [native Protocol Buffers format][vector_native_protobuf]. - [logfmt]: https://brandur.org/logfmt - """ - native: """ - Encodes an event in the [native Protocol Buffers format][vector_native_protobuf]. + This codec is **[experimental][experimental]**. - This codec is **[experimental][experimental]**. + [vector_native_protobuf]: https://github.com/vectordotdev/vector/blob/master/lib/vector-core/proto/event.proto + [experimental]: https://vector.dev/highlights/2022-03-31-native-event-codecs + """ + native_json: """ + Encodes an event in the [native JSON format][vector_native_json]. - [vector_native_protobuf]: https://github.com/vectordotdev/vector/blob/master/lib/vector-core/proto/event.proto - [experimental]: https://vector.dev/highlights/2022-03-31-native-event-codecs - """ - native_json: """ - Encodes an event in the [native JSON format][vector_native_json]. + This codec is **[experimental][experimental]**. - This codec is **[experimental][experimental]**. + [vector_native_json]: https://github.com/vectordotdev/vector/blob/master/lib/codecs/tests/data/native_encoding/schema.cue + [experimental]: https://vector.dev/highlights/2022-03-31-native-event-codecs + """ + otlp: """ + Encodes an event in the [OTLP (OpenTelemetry Protocol)][otlp] format. - [vector_native_json]: https://github.com/vectordotdev/vector/blob/master/lib/codecs/tests/data/native_encoding/schema.cue - [experimental]: https://vector.dev/highlights/2022-03-31-native-event-codecs - """ - otlp: """ - Encodes an event in the [OTLP (OpenTelemetry Protocol)][otlp] format. + This codec uses protobuf encoding, which is the recommended format for OTLP. + The output is suitable for sending to OTLP-compatible endpoints with + `content-type: application/x-protobuf`. - This codec uses protobuf encoding, which is the recommended format for OTLP. - The output is suitable for sending to OTLP-compatible endpoints with - `content-type: application/x-protobuf`. + [otlp]: https://opentelemetry.io/docs/specs/otlp/ + """ + protobuf: """ + Encodes an event as a [Protobuf][protobuf] message. - [otlp]: https://opentelemetry.io/docs/specs/otlp/ - """ - protobuf: """ - Encodes an event as a [Protobuf][protobuf] message. + [protobuf]: https://protobuf.dev/ + """ + raw_message: """ + No encoding. - [protobuf]: https://protobuf.dev/ - """ - raw_message: """ - No encoding. + This encoding uses the `message` field of a log event. - This encoding uses the `message` field of a log event. + Be careful if you are modifying your log events (for example, by using a `remap` + transform) and removing the message field while doing additional parsing on it, as this + could lead to the encoding emitting empty strings for the given event. + """ + syslog: """ + Syslog encoding + RFC 3164 and 5424 are supported + """ + text: """ + Plain text encoding. - Be careful if you are modifying your log events (for example, by using a `remap` - transform) and removing the message field while doing additional parsing on it, as this - could lead to the encoding emitting empty strings for the given event. - """ - syslog: """ - Syslog encoding - RFC 3164 and 5424 are supported - """ - text: """ - Plain text encoding. + This encoding uses the `message` field of a log event. For metrics, it uses an + encoding that resembles the Prometheus export format. + + Be careful if you are modifying your log events (for example, by using a `remap` + transform) and removing the message field while doing additional parsing on it, as this + could lead to the encoding emitting empty strings for the given event. + """ + } + } + csv: { + description: "The CSV Serializer Options." + relevant_when: "codec = \"csv\"" + required: true + type: object: options: { + capacity: { + description: """ + Sets the capacity (in bytes) of the internal buffer used in the CSV writer. + This defaults to 8192 bytes (8KB). + """ + required: false + type: uint: default: 8192 + } + delimiter: { + description: "The field delimiter to use when writing CSV." + required: false + type: ascii_char: default: "," + } + double_quote: { + description: """ + Enables double quote escapes. - This encoding uses the `message` field of a log event. For metrics, it uses an - encoding that resembles the Prometheus export format. + This is enabled by default, but you can disable it. When disabled, quotes in + field data are escaped instead of doubled. + """ + required: false + type: bool: default: true + } + escape: { + description: """ + The escape character to use when writing CSV. - Be careful if you are modifying your log events (for example, by using a `remap` - transform) and removing the message field while doing additional parsing on it, as this - could lead to the encoding emitting empty strings for the given event. - """ + In some variants of CSV, quotes are escaped using a special escape character + like \\ (instead of escaping quotes by doubling them). + + To use this, `double_quotes` needs to be disabled as well; otherwise, this setting is ignored. + """ + required: false + type: ascii_char: default: "\"" } - } - csv: { - description: "The CSV Serializer Options." - relevant_when: "codec = \"csv\"" - required: true - type: object: options: { - capacity: { - description: """ - Sets the capacity (in bytes) of the internal buffer used in the CSV writer. - This defaults to 8192 bytes (8KB). - """ - required: false - type: uint: default: 8192 - } - delimiter: { - description: "The field delimiter to use when writing CSV." - required: false - type: ascii_char: default: "," - } - double_quote: { - description: """ - Enables double quote escapes. - - This is enabled by default, but you can disable it. When disabled, quotes in - field data are escaped instead of doubled. - """ - required: false - type: bool: default: true - } - escape: { - description: """ - The escape character to use when writing CSV. + fields: { + description: """ + Configures the fields that are encoded, as well as the order in which they + appear in the output. - In some variants of CSV, quotes are escaped using a special escape character - like \\ (instead of escaping quotes by doubling them). + If a field is not present in the event, the output for that field is an empty string. - To use this, `double_quotes` needs to be disabled as well; otherwise, this setting is ignored. - """ - required: false - type: ascii_char: default: "\"" - } - fields: { - description: """ - Configures the fields that are encoded, as well as the order in which they - appear in the output. - - If a field is not present in the event, the output for that field is an empty string. - - Values of type `Array`, `Object`, and `Regex` are not supported, and the - output for any of these types is an empty string. - """ - required: true - type: array: items: type: string: {} - } - quote: { - description: "The quote character to use when writing CSV." - required: false - type: ascii_char: default: "\"" - } - quote_style: { - description: "The quoting style to use when writing CSV data." - required: false - type: string: { - default: "necessary" - enum: { - always: "Always puts quotes around every field." - necessary: """ - Puts quotes around fields only when necessary. - They are necessary when fields contain a quote, delimiter, or record terminator. - Quotes are also necessary when writing an empty record - (which is indistinguishable from a record with one empty field). - """ - never: "Never writes quotes, even if it produces invalid CSV data." - non_numeric: """ - Puts quotes around all fields that are non-numeric. - This means that when writing a field that does not parse as a valid float or integer, - quotes are used even if they aren't strictly necessary. - """ - } + Values of type `Array`, `Object`, and `Regex` are not supported, and the + output for any of these types is an empty string. + """ + required: true + type: array: items: type: string: {} + } + quote: { + description: "The quote character to use when writing CSV." + required: false + type: ascii_char: default: "\"" + } + quote_style: { + description: "The quoting style to use when writing CSV data." + required: false + type: string: { + default: "necessary" + enum: { + always: "Always puts quotes around every field." + necessary: """ + Puts quotes around fields only when necessary. + They are necessary when fields contain a quote, delimiter, or record terminator. + Quotes are also necessary when writing an empty record + (which is indistinguishable from a record with one empty field). + """ + never: "Never writes quotes, even if it produces invalid CSV data." + non_numeric: """ + Puts quotes around all fields that are non-numeric. + This means that when writing a field that does not parse as a valid float or integer, + quotes are used even if they aren't strictly necessary. + """ } } } } - except_fields: { - description: "List of fields that are excluded from the encoded event." + } + except_fields: { + description: "List of fields that are excluded from the encoded event." + required: false + type: array: items: type: string: {} + } + gelf: { + description: "The GELF Serializer Options." + relevant_when: "codec = \"gelf\"" + required: false + type: object: options: max_chunk_size: { + description: """ + Maximum size for each GELF chunked datagram (including 12-byte header). + Chunking starts when datagrams exceed this size. + For Graylog target, keep at or below 8192 bytes; for Vector target (`gelf` decoding with `chunked_gelf` framing), up to 65,500 bytes is recommended. + """ + required: false + type: uint: default: 8192 + } + } + json: { + description: "Options for the JsonSerializer." + relevant_when: "codec = \"json\"" + required: false + type: object: options: pretty: { + description: "Whether to use pretty JSON formatting." required: false - type: array: items: type: string: {} + type: bool: default: false } - gelf: { - description: "The GELF Serializer Options." - relevant_when: "codec = \"gelf\"" - required: false - type: object: options: max_chunk_size: { - description: """ - Maximum size for each GELF chunked datagram (including 12-byte header). - Chunking starts when datagrams exceed this size. - For Graylog target, keep at or below 8192 bytes; for Vector target (`gelf` decoding with `chunked_gelf` framing), up to 65,500 bytes is recommended. - """ - required: false - type: uint: default: 8192 + } + metric_tag_values: { + description: """ + Controls how metric tag values are encoded. + + When set to `single`, only the last non-bare value of tags are displayed with the + metric. When set to `full`, all metric tags are exposed as separate assignments. + """ + relevant_when: "codec = \"json\" or codec = \"text\"" + required: false + type: string: { + default: "single" + enum: { + full: "All tags are exposed as arrays of either string or null values." + single: """ + Tag values are exposed as single strings, the same as they were before this config + option. Tags with multiple values show the last assigned value, and null values + are ignored. + """ } } - json: { - description: "Options for the JsonSerializer." - relevant_when: "codec = \"json\"" - required: false - type: object: options: pretty: { - description: "Whether to use pretty JSON formatting." - required: false + } + only_fields: { + description: "List of fields that are included in the encoded event." + required: false + type: array: items: type: string: {} + } + protobuf: { + description: "Options for the Protobuf serializer." + relevant_when: "codec = \"protobuf\"" + required: true + type: object: options: { + desc_file: { + description: """ + The path to the protobuf descriptor set file. + + This file is the output of `protoc -I -o ` + + You can read more [here](https://buf.build/docs/reference/images/#how-buf-images-work). + """ + required: true + type: string: examples: ["/etc/vector/protobuf_descriptor_set.desc"] + } + message_type: { + description: "The name of the message type to use for serializing." + required: true + type: string: examples: ["package.Message"] + } + use_json_names: { + description: """ + Use JSON field names (camelCase) instead of protobuf field names (snake_case). + + When enabled, the serializer looks for fields using their JSON names as defined + in the `.proto` file (for example `jobDescription` instead of `job_description`). + + This is useful when working with data that has already been converted from JSON or + when interfacing with systems that use JSON naming conventions. + """ + required: false type: bool: default: false } } - metric_tag_values: { - description: """ - Controls how metric tag values are encoded. + } + syslog: { + description: "Options for the Syslog serializer." + relevant_when: "codec = \"syslog\"" + required: false + type: object: options: { + app_name: { + description: """ + Path to a field in the event to use for the app name. - When set to `single`, only the last non-bare value of tags are displayed with the - metric. When set to `full`, all metric tags are exposed as separate assignments. - """ - relevant_when: "codec = \"json\" or codec = \"text\"" - required: false - type: string: { - default: "single" - enum: { - full: "All tags are exposed as arrays of either string or null values." - single: """ - Tag values are exposed as single strings, the same as they were before this config - option. Tags with multiple values show the last assigned value, and null values - are ignored. - """ - } + If not provided, the encoder checks for a semantic "service" field. + If that is also missing, it defaults to "vector". + """ + required: false + type: string: {} } - } - only_fields: { - description: "List of fields that are included in the encoded event." - required: false - type: array: items: type: string: {} - } - protobuf: { - description: "Options for the Protobuf serializer." - relevant_when: "codec = \"protobuf\"" - required: true - type: object: options: { - desc_file: { - description: """ - The path to the protobuf descriptor set file. - - This file is the output of `protoc -I -o ` - - You can read more [here](https://buf.build/docs/reference/images/#how-buf-images-work). - """ - required: true - type: string: examples: ["/etc/vector/protobuf_descriptor_set.desc"] - } - message_type: { - description: "The name of the message type to use for serializing." - required: true - type: string: examples: ["package.Message"] - } - use_json_names: { - description: """ - Use JSON field names (camelCase) instead of protobuf field names (snake_case). - - When enabled, the serializer looks for fields using their JSON names as defined - in the `.proto` file (for example `jobDescription` instead of `job_description`). - - This is useful when working with data that has already been converted from JSON or - when interfacing with systems that use JSON naming conventions. - """ - required: false - type: bool: default: false - } + facility: { + description: "Path to a field in the event to use for the facility. Defaults to \"user\"." + required: false + type: string: {} } - } - syslog: { - description: "Options for the Syslog serializer." - relevant_when: "codec = \"syslog\"" - required: false - type: object: options: { - app_name: { - description: """ - Path to a field in the event to use for the app name. - - If not provided, the encoder checks for a semantic "service" field. - If that is also missing, it defaults to "vector". - """ - required: false - type: string: {} - } - facility: { - description: "Path to a field in the event to use for the facility. Defaults to \"user\"." - required: false - type: string: {} - } - msg_id: { - description: "Path to a field in the event to use for the msg ID." - required: false - type: string: {} - } - proc_id: { - description: "Path to a field in the event to use for the proc ID." - required: false - type: string: {} - } - rfc: { - description: "RFC to use for formatting." - required: false - type: string: { - default: "rfc5424" - enum: { - rfc3164: "The legacy RFC3164 syslog format." - rfc5424: "The modern RFC5424 syslog format." - } + msg_id: { + description: "Path to a field in the event to use for the msg ID." + required: false + type: string: {} + } + proc_id: { + description: "Path to a field in the event to use for the proc ID." + required: false + type: string: {} + } + rfc: { + description: "RFC to use for formatting." + required: false + type: string: { + default: "rfc5424" + enum: { + rfc3164: "The legacy RFC3164 syslog format." + rfc5424: "The modern RFC5424 syslog format." } } - severity: { - description: "Path to a field in the event to use for the severity. Defaults to \"informational\"." - required: false - type: string: {} - } } - } - timestamp_format: { - description: "Format used for timestamp fields." - required: false - type: string: enum: { - rfc3339: "Represent the timestamp as a RFC 3339 timestamp." - unix: "Represent the timestamp as a Unix timestamp." - unix_float: "Represent the timestamp as a Unix timestamp in floating point." - unix_ms: "Represent the timestamp as a Unix timestamp in milliseconds." - unix_ns: "Represent the timestamp as a Unix timestamp in nanoseconds." - unix_us: "Represent the timestamp as a Unix timestamp in microseconds." + severity: { + description: "Path to a field in the event to use for the severity. Defaults to \"informational\"." + required: false + type: string: {} } } } + timestamp_format: { + description: "Format used for timestamp fields." + required: false + type: string: enum: { + rfc3339: "Represent the timestamp as a RFC 3339 timestamp." + unix: "Represent the timestamp as a Unix timestamp." + unix_float: "Represent the timestamp as a Unix timestamp in floating point." + unix_ms: "Represent the timestamp as a Unix timestamp in milliseconds." + unix_ns: "Represent the timestamp as a Unix timestamp in nanoseconds." + unix_us: "Represent the timestamp as a Unix timestamp in microseconds." + } + } } - framing: { - description: "Framing configuration." - required: false - type: object: options: { - character_delimited: { - description: "Options for the character delimited encoder." - relevant_when: "method = \"character_delimited\"" - required: true - type: object: options: delimiter: { - description: "The ASCII (7-bit) character that delimits byte sequences." - required: true - type: ascii_char: {} - } + } + framing: { + description: "Framing configuration." + relevant_when: "protocol = \"http\"" + required: false + type: object: options: { + character_delimited: { + description: "Options for the character delimited encoder." + relevant_when: "method = \"character_delimited\"" + required: true + type: object: options: delimiter: { + description: "The ASCII (7-bit) character that delimits byte sequences." + required: true + type: ascii_char: {} } - length_delimited: { - description: "Options for the length delimited decoder." - relevant_when: "method = \"length_delimited\"" - required: true - type: object: options: { - length_field_is_big_endian: { - description: "Length field byte order (little or big endian)" - required: false - type: bool: default: true - } - length_field_length: { - description: "Number of bytes representing the field length" - required: false - type: uint: default: 4 - } - length_field_offset: { - description: "Number of bytes in the header before the length field" - required: false - type: uint: default: 0 - } - max_frame_length: { - description: "Maximum frame length" - required: false - type: uint: default: 8388608 - } + } + length_delimited: { + description: "Options for the length delimited decoder." + relevant_when: "method = \"length_delimited\"" + required: true + type: object: options: { + length_field_is_big_endian: { + description: "Length field byte order (little or big endian)" + required: false + type: bool: default: true + } + length_field_length: { + description: "Number of bytes representing the field length" + required: false + type: uint: default: 4 + } + length_field_offset: { + description: "Number of bytes in the header before the length field" + required: false + type: uint: default: 0 + } + max_frame_length: { + description: "Maximum frame length" + required: false + type: uint: default: 8388608 } } - max_frame_length: { - description: "Maximum frame length" - relevant_when: "method = \"varint_length_delimited\"" - required: false - type: uint: default: 8388608 - } - method: { - description: "The framing method." - required: true - type: string: enum: { - bytes: "Event data is not delimited at all." - character_delimited: "Event data is delimited by a single ASCII (7-bit) character." - length_delimited: """ - Event data is prefixed with its length in bytes. - - The prefix is a 32-bit unsigned integer, little endian. - """ - newline_delimited: "Event data is delimited by a newline (LF) character." - varint_length_delimited: """ - Event data is prefixed with its length in bytes as a varint. + } + max_frame_length: { + description: "Maximum frame length" + relevant_when: "method = \"varint_length_delimited\"" + required: false + type: uint: default: 8388608 + } + method: { + description: "The framing method." + required: true + type: string: enum: { + bytes: "Event data is not delimited at all." + character_delimited: "Event data is delimited by a single ASCII (7-bit) character." + length_delimited: """ + Event data is prefixed with its length in bytes. + + The prefix is a 32-bit unsigned integer, little endian. + """ + newline_delimited: "Event data is delimited by a newline (LF) character." + varint_length_delimited: """ + Event data is prefixed with its length in bytes as a varint. - This is compatible with protobuf's length-delimited encoding. - """ - } + This is compatible with protobuf's length-delimited encoding. + """ } } } - method: { - description: "The HTTP method to use when making the request." - required: false - type: string: { - default: "post" - enum: { - delete: "DELETE." - get: "GET." - head: "HEAD." - options: "OPTIONS." - patch: "PATCH." - post: "POST." - put: "PUT." - trace: "TRACE." - } + } + method: { + description: "The HTTP method to use. Defaults to `post`." + relevant_when: "protocol = \"http\"" + required: false + type: string: { + default: "post" + enum: { + delete: "DELETE." + get: "GET." + head: "HEAD." + options: "OPTIONS." + patch: "PATCH." + post: "POST." + put: "PUT." + trace: "TRACE." } } - payload_prefix: { - description: """ - A string to prefix the payload with. + } + payload_prefix: { + description: "A string to prefix the payload with." + relevant_when: "protocol = \"http\"" + required: false + type: string: default: "" + } + payload_suffix: { + description: "A string to suffix the payload with." + relevant_when: "protocol = \"http\"" + required: false + type: string: default: "" + } + protocol: { + description: "The transport protocol to use." + required: true + type: string: enum: { + grpc: "Send OTLP data over gRPC." + http: "Send OTLP data over HTTP." + } + } + request: { + description: "Outbound HTTP request settings." + required: false + type: object: options: { + adaptive_concurrency: { + description: """ + Configuration of adaptive concurrency parameters. - This option is ignored if the encoding is not character delimited JSON. + These parameters typically do not require changes from the default, and incorrect values can lead to meta-stable or + unstable performance and sink behavior. Proceed with caution. + """ + required: false + type: object: options: { + decrease_ratio: { + description: """ + The fraction of the current value to set the new concurrency limit when decreasing the limit. - If specified, the `payload_suffix` must also be specified and together they must produce a valid JSON object. - """ - required: false - type: string: { - default: "" - examples: ["{\"data\":"] - } - } - payload_suffix: { - description: """ - A string to suffix the payload with. + Valid values are greater than `0` and less than `1`. Smaller values cause the algorithm to scale back rapidly + when latency increases. - This option is ignored if the encoding is not character delimited JSON. + **Note**: The new limit is rounded down after applying this ratio. + """ + required: false + type: float: default: 0.9 + } + ewma_alpha: { + description: """ + The weighting of new measurements compared to older measurements. - If specified, the `payload_prefix` must also be specified and together they must produce a valid JSON object. - """ - required: false - type: string: { - default: "" - examples: ["}"] - } - } - request: { - description: "Outbound HTTP request settings." - required: false - type: object: options: { - adaptive_concurrency: { - description: """ - Configuration of adaptive concurrency parameters. + Valid values are greater than `0` and less than `1`. - These parameters typically do not require changes from the default, and incorrect values can lead to meta-stable or - unstable performance and sink behavior. Proceed with caution. - """ - required: false - type: object: options: { - decrease_ratio: { - description: """ - The fraction of the current value to set the new concurrency limit when decreasing the limit. - - Valid values are greater than `0` and less than `1`. Smaller values cause the algorithm to scale back rapidly - when latency increases. - - **Note**: The new limit is rounded down after applying this ratio. - """ - required: false - type: float: default: 0.9 - } - ewma_alpha: { - description: """ - The weighting of new measurements compared to older measurements. - - Valid values are greater than `0` and less than `1`. - - ARC uses an exponentially weighted moving average (EWMA) of past RTT measurements as a reference to compare with - the current RTT. Smaller values cause this reference to adjust more slowly, which may be useful if a service has - unusually high response variability. - """ - required: false - type: float: default: 0.4 - } - initial_concurrency: { - description: """ - The initial concurrency limit to use. If not specified, the initial limit is 1 (no concurrency). - - Datadog recommends setting this value to your service's average limit if you're seeing that it takes a - long time to ramp up adaptive concurrency after a restart. You can find this value by looking at the - `adaptive_concurrency_limit` metric. - """ - required: false - type: uint: default: 1 - } - max_concurrency_limit: { - description: """ - The maximum concurrency limit. - - The adaptive request concurrency limit does not go above this bound. This is put in place as a safeguard. - """ - required: false - type: uint: default: 200 - } - rtt_deviation_scale: { - description: """ - Scale of RTT deviations which are not considered anomalous. - - Valid values are greater than or equal to `0`, and reasonable values range from `1.0` to `3.0`. - - When calculating the past RTT average, a secondary “deviation” value is also computed that indicates how variable - those values are. That deviation is used when comparing the past RTT average to the current measurements, so we - can ignore increases in RTT that are within an expected range. This factor is used to scale up the deviation to - an appropriate range. Larger values cause the algorithm to ignore larger increases in the RTT. - """ - required: false - type: float: default: 2.5 - } + ARC uses an exponentially weighted moving average (EWMA) of past RTT measurements as a reference to compare with + the current RTT. Smaller values cause this reference to adjust more slowly, which may be useful if a service has + unusually high response variability. + """ + required: false + type: float: default: 0.4 } - } - concurrency: { - description: """ - Configuration for outbound request concurrency. + initial_concurrency: { + description: """ + The initial concurrency limit to use. If not specified, the initial limit is 1 (no concurrency). - This can be set either to one of the below enum values or to a positive integer, which denotes - a fixed concurrency limit. - """ - required: false - type: { - string: { - default: "adaptive" - enum: { - adaptive: """ - Concurrency is managed by Vector's [Adaptive Request Concurrency][arc] feature. + Datadog recommends setting this value to your service's average limit if you're seeing that it takes a + long time to ramp up adaptive concurrency after a restart. You can find this value by looking at the + `adaptive_concurrency_limit` metric. + """ + required: false + type: uint: default: 1 + } + max_concurrency_limit: { + description: """ + The maximum concurrency limit. - [arc]: https://vector.dev/docs/architecture/arc/ - """ - none: """ - A fixed concurrency of 1. + The adaptive request concurrency limit does not go above this bound. This is put in place as a safeguard. + """ + required: false + type: uint: default: 200 + } + rtt_deviation_scale: { + description: """ + Scale of RTT deviations which are not considered anomalous. - Only one request can be outstanding at any given time. - """ - } - } - uint: {} + Valid values are greater than or equal to `0`, and reasonable values range from `1.0` to `3.0`. + + When calculating the past RTT average, a secondary “deviation” value is also computed that indicates how variable + those values are. That deviation is used when comparing the past RTT average to the current measurements, so we + can ignore increases in RTT that are within an expected range. This factor is used to scale up the deviation to + an appropriate range. Larger values cause the algorithm to ignore larger increases in the RTT. + """ + required: false + type: float: default: 2.5 } } - headers: { - description: "Additional HTTP headers to add to every HTTP request." - required: false - type: object: { - examples: [{ - Accept: "text/plain" - "X-Event-Level": "{{level}}" - "X-Event-Timestamp": "{{timestamp}}" - "X-My-Custom-Header": "A-Value" - }] - options: "*": { - description: "An HTTP request header and its value. Both header names and values support templating with event data." - required: true - type: string: {} + } + concurrency: { + description: """ + Configuration for outbound request concurrency. + + This can be set either to one of the below enum values or to a positive integer, which denotes + a fixed concurrency limit. + """ + required: false + type: { + string: { + default: "adaptive" + enum: { + adaptive: """ + Concurrency is managed by Vector's [Adaptive Request Concurrency][arc] feature. + + [arc]: https://vector.dev/docs/architecture/arc/ + """ + none: """ + A fixed concurrency of 1. + + Only one request can be outstanding at any given time. + """ } } + uint: {} } - rate_limit_duration_secs: { - description: "The time window used for the `rate_limit_num` option." - required: false - type: uint: { - default: 1 - unit: "seconds" + } + headers: { + description: "Additional HTTP headers to add to every HTTP request." + required: false + type: object: { + examples: [{ + Accept: "text/plain" + "X-Event-Level": "{{level}}" + "X-Event-Timestamp": "{{timestamp}}" + "X-My-Custom-Header": "A-Value" + }] + options: "*": { + description: "An HTTP request header and its value. Both header names and values support templating with event data." + required: true + type: string: {} } } - rate_limit_num: { - description: "The maximum number of requests allowed within the `rate_limit_duration_secs` time window." - required: false - type: uint: { - default: 9223372036854775807 - unit: "requests" - } + } + rate_limit_duration_secs: { + description: "The time window used for the `rate_limit_num` option." + required: false + type: uint: { + default: 1 + unit: "seconds" } - retry_attempts: { - description: "The maximum number of retries to make for failed requests." - required: false - type: uint: { - default: 9223372036854775807 - unit: "retries" - } + } + rate_limit_num: { + description: "The maximum number of requests allowed within the `rate_limit_duration_secs` time window." + required: false + type: uint: { + default: 9223372036854775807 + unit: "requests" } - retry_initial_backoff_secs: { - description: """ - The amount of time to wait before attempting the first retry for a failed request. - - After the first retry has failed, the Fibonacci sequence is used to select future backoffs. - """ - required: false - type: uint: { - default: 1 - unit: "seconds" - } + } + retry_attempts: { + description: "The maximum number of retries to make for failed requests." + required: false + type: uint: { + default: 9223372036854775807 + unit: "retries" } - retry_jitter_mode: { - description: "The jitter mode to use for retry backoff behavior." - required: false - type: string: { - default: "Full" - enum: { - Full: """ - Full jitter. - - The random delay is anywhere from 0 up to the maximum current delay calculated by the backoff - strategy. + } + retry_initial_backoff_secs: { + description: """ + The amount of time to wait before attempting the first retry for a failed request. - Incorporating full jitter into your backoff strategy can greatly reduce the likelihood - of creating accidental denial of service (DoS) conditions against your own systems when - many clients are recovering from a failure state. - """ - None: "No jitter." - } - } + After the first retry has failed, the Fibonacci sequence is used to select future backoffs. + """ + required: false + type: uint: { + default: 1 + unit: "seconds" } - retry_max_duration_secs: { - description: "The maximum amount of time to wait between retries." - required: false - type: uint: { - default: 30 - unit: "seconds" + } + retry_jitter_mode: { + description: "The jitter mode to use for retry backoff behavior." + required: false + type: string: { + default: "Full" + enum: { + Full: """ + Full jitter. + + The random delay is anywhere from 0 up to the maximum current delay calculated by the backoff + strategy. + + Incorporating full jitter into your backoff strategy can greatly reduce the likelihood + of creating accidental denial of service (DoS) conditions against your own systems when + many clients are recovering from a failure state. + """ + None: "No jitter." } } - timeout_secs: { - description: """ - The time a request can take before being aborted. + } + retry_max_duration_secs: { + description: "The maximum amount of time to wait between retries." + required: false + type: uint: { + default: 30 + unit: "seconds" + } + } + timeout_secs: { + description: """ + The time a request can take before being aborted. - Datadog highly recommends that you do not lower this value below the service's internal timeout, as this could - create orphaned requests, pile on retries, and result in duplicate data downstream. - """ - required: false - type: uint: { - default: 60 - unit: "seconds" - } + Datadog highly recommends that you do not lower this value below the service's internal timeout, as this could + create orphaned requests, pile on retries, and result in duplicate data downstream. + """ + required: false + type: uint: { + default: 60 + unit: "seconds" } } } - tls: { - description: "TLS configuration." - required: false - type: object: options: { - alpn_protocols: { - description: """ - Sets the list of supported ALPN protocols. + } + tls: { + description: "TLS configuration." + required: false + type: object: options: { + alpn_protocols: { + description: """ + Sets the list of supported ALPN protocols. - Declare the supported ALPN protocols, which are used during negotiation with a peer. They are prioritized in the order - that they are defined. - """ - required: false - type: array: items: type: string: examples: ["h2"] - } - ca_file: { - description: """ - Absolute path to an additional CA certificate file. + Declare the supported ALPN protocols, which are used during negotiation with a peer. They are prioritized in the order + that they are defined. + """ + required: false + type: array: items: type: string: examples: ["h2"] + } + ca_file: { + description: """ + Absolute path to an additional CA certificate file. - The certificate must be in the DER or PEM (X.509) format. Additionally, the certificate can be provided as an inline string in PEM format. - """ - required: false - type: string: examples: ["/path/to/certificate_authority.crt"] - } - crt_file: { - description: """ - Absolute path to a certificate file used to identify this server. + The certificate must be in the DER or PEM (X.509) format. Additionally, the certificate can be provided as an inline string in PEM format. + """ + required: false + type: string: examples: ["/path/to/certificate_authority.crt"] + } + crt_file: { + description: """ + Absolute path to a certificate file used to identify this server. - The certificate must be in DER, PEM (X.509), or PKCS#12 format. Additionally, the certificate can be provided as - an inline string in PEM format. + The certificate must be in DER, PEM (X.509), or PKCS#12 format. Additionally, the certificate can be provided as + an inline string in PEM format. - If this is set _and_ is not a PKCS#12 archive, `key_file` must also be set. - """ - required: false - type: string: examples: ["/path/to/host_certificate.crt"] - } - key_file: { - description: """ - Absolute path to a private key file used to identify this server. + If this is set _and_ is not a PKCS#12 archive, `key_file` must also be set. + """ + required: false + type: string: examples: ["/path/to/host_certificate.crt"] + } + key_file: { + description: """ + Absolute path to a private key file used to identify this server. - The key must be in DER or PEM (PKCS#8) format. Additionally, the key can be provided as an inline string in PEM format. - """ - required: false - type: string: examples: ["/path/to/host_certificate.key"] - } - key_pass: { - description: """ - Passphrase used to unlock the encrypted key file. + The key must be in DER or PEM (PKCS#8) format. Additionally, the key can be provided as an inline string in PEM format. + """ + required: false + type: string: examples: ["/path/to/host_certificate.key"] + } + key_pass: { + description: """ + Passphrase used to unlock the encrypted key file. - This has no effect unless `key_file` is set. - """ - required: false - type: string: examples: ["${KEY_PASS_ENV_VAR}", "PassWord1"] - } - server_name: { - description: """ - Server name to use when using Server Name Indication (SNI). + This has no effect unless `key_file` is set. + """ + required: false + type: string: examples: ["${KEY_PASS_ENV_VAR}", "PassWord1"] + } + server_name: { + description: """ + Server name to use when using Server Name Indication (SNI). - Only relevant for outgoing connections. - """ - required: false - type: string: examples: ["www.example.com"] - } - verify_certificate: { - description: """ - Enables certificate verification. For components that create a server, this requires that the - client connections have a valid client certificate. For components that initiate requests, - this validates that the upstream has a valid certificate. + Only relevant for outgoing connections. + """ + required: false + type: string: examples: ["www.example.com"] + } + verify_certificate: { + description: """ + Enables certificate verification. For components that create a server, this requires that the + client connections have a valid client certificate. For components that initiate requests, + this validates that the upstream has a valid certificate. - If enabled, certificates must not be expired and must be issued by a trusted - issuer. This verification operates in a hierarchical manner, checking that the leaf certificate (the - certificate presented by the client/server) is not only valid, but that the issuer of that certificate is also valid, and - so on, until the verification process reaches a root certificate. + If enabled, certificates must not be expired and must be issued by a trusted + issuer. This verification operates in a hierarchical manner, checking that the leaf certificate (the + certificate presented by the client/server) is not only valid, but that the issuer of that certificate is also valid, and + so on, until the verification process reaches a root certificate. - Do NOT set this to `false` unless you understand the risks of not verifying the validity of certificates. - """ - required: false - type: bool: {} - } - verify_hostname: { - description: """ - Enables hostname verification. + Do NOT set this to `false` unless you understand the risks of not verifying the validity of certificates. + """ + required: false + type: bool: {} + } + verify_hostname: { + description: """ + Enables hostname verification. - If enabled, the hostname used to connect to the remote host must be present in the TLS certificate presented by - the remote host, either as the Common Name or as an entry in the Subject Alternative Name extension. + If enabled, the hostname used to connect to the remote host must be present in the TLS certificate presented by + the remote host, either as the Common Name or as an entry in the Subject Alternative Name extension. - Only relevant for outgoing connections. + Only relevant for outgoing connections. - Do NOT set this to `false` unless you understand the risks of not verifying the remote hostname. - """ - required: false - type: bool: {} - } + Do NOT set this to `false` unless you understand the risks of not verifying the remote hostname. + """ + required: false + type: bool: {} } } - type: { - description: "The communication protocol." - required: true - type: string: enum: http: "Send data over HTTP." - } - uri: { - description: """ - The full URI to make HTTP requests to. + } + uri: { + description: """ + The URI to send requests to. - This should include the protocol and host, but can also include the port, path, and any other valid part of a URI. - """ - required: true - type: string: { - examples: ["https://10.22.212.22:9000/endpoint"] - syntax: "template" - } + Supports template syntax (e.g. `http://{{ host }}:4317`). Must include a scheme + (`http://` or `https://`) and a port. + + For the gRPC transport, the template is rendered once per batch using the first event + in the batch. + + # Examples + + - `http://localhost:5318/v1/logs` (HTTP) + - `http://localhost:4317` (gRPC) + """ + required: true + type: string: { + examples: ["http://localhost:5318/v1/logs", "http://localhost:4317"] + syntax: "template" } + warnings: ["When using template syntax, the rendered URI is taken from event data. Only use dynamic URIs with trusted event sources to avoid directing Vector to unintended internal network destinations."] } } diff --git a/website/cue/reference/components/sinks/opentelemetry.cue b/website/cue/reference/components/sinks/opentelemetry.cue index ffdc1d40b67c2..5fcd812fbac1c 100644 --- a/website/cue/reference/components/sinks/opentelemetry.cue +++ b/website/cue/reference/components/sinks/opentelemetry.cue @@ -106,14 +106,14 @@ components: sinks: opentelemetry: { emit_syslog: inputs: ["remap_syslog"] type: opentelemetry - protocol: - type: http - uri: http://localhost:5318/v1/logs - method: post - encoding: - codec: json - framing: - method: newline_delimited + protocol: http + uri: http://localhost:5318/v1/logs + method: post + encoding: + codec: json + framing: + method: newline_delimited + request: headers: content-type: application/json ```