Merge branch 'master' into greylilac09/add-incremental-to-absolute

Author: GreyLilac09

.github/workflows/preview_site_trigger.yml

@@ -7,62 +7,44 @@ jobs:
     runs-on: ubuntu-24.04
     timeout-minutes: 5
     # Only run for PRs with 'website' in the branch name
-    if: ${{ contains(github.head_ref, 'website') && contains(github.head_ref, '-') }}
+    if: ${{ contains(github.head_ref, 'website') }}
     steps:
-      - name: Echo approval
+      # Validate branch name
+      - name: Validate branch name and set output
+        id: validate
         run: |
-          echo "Workflow has been allowed to run for PR ${{ github.event.number }}. Setting artifacts and then continuing workflow runs"
-
-      # Use GitHub Action to safely validate and store PR information
+          BRANCH="${{ github.head_ref }}"
+          if [[ ! "$BRANCH" =~ ^[a-zA-Z0-9_-]+$ ]]; then
+            echo "valid=false" >> $GITHUB_OUTPUT
+          else
+            echo "valid=true" >> $GITHUB_OUTPUT
+          fi
+
+      # Save PR information (only if branch is valid)
       - name: Validate and save PR information
+        if: steps.validate.outputs.valid == 'true'
         uses: actions/github-script@60a0d83039c74a4aee543508d2ffcb1c3799cdea # v7.0.1
         with:
           script: |
             const fs = require('fs').promises;
             const path = require('path');
             const crypto = require('crypto');
+            const prNumber = context.payload.number;
+            const branchName = context.payload.pull_request.head.ref;
 
-            async function createAndValidateArtifact() {
-              try {
-                // Create directory for artifact
-                await fs.mkdir('./pr', { recursive: true });
-
-                // Get PR number and validate
-                const prNumber = context.payload.number;
-                if (typeof prNumber !== 'number' || !Number.isInteger(prNumber) || prNumber <= 0) {
-                  core.setFailed(`Invalid PR number: ${prNumber}`);
-                  return;
-                }
-
-                // Get branch name and validate
-                const branchName = context.payload.pull_request.head.ref;
-                // Validate branch name (only allow alphanumeric, dash, and underscore)
-                const branchNameRegex = /^[a-zA-Z0-9_\-]+$/;
-                if (!branchNameRegex.test(branchName)) {
-                  core.setFailed(`Invalid branch name detected: ${branchName}`);
-                  return;
-                }
-
-                // Write validated information to files
-                await fs.writeFile('./pr/number', prNumber.toString());
-                await fs.writeFile('./pr/branch', branchName);
-
-                // Log success
-                core.info(`Successfully validated and saved PR #${prNumber} with branch ${branchName}`);
+            await fs.mkdir('./pr', { recursive: true });
+            await fs.writeFile('./pr/number', prNumber.toString());
+            await fs.writeFile('./pr/branch', branchName);
 
-                // Create hash signature of the data
-                const numberHash = crypto.createHash('sha256').update(prNumber.toString()).digest('hex');
-                const branchHash = crypto.createHash('sha256').update(branchName).digest('hex');
-                await fs.writeFile('./pr/integrity', `${numberHash}:${branchHash}`);
-              } catch (error) {
-                core.setFailed(`An error occurred: ${error.message}`);
-              }
-            }
+            const numberHash = crypto.createHash('sha256').update(prNumber.toString()).digest('hex');
+            const branchHash = crypto.createHash('sha256').update(branchName).digest('hex');
+            await fs.writeFile('./pr/integrity', `${numberHash}:${branchHash}`);
 
-            createAndValidateArtifact();
+            core.info(`Saved PR #${prNumber} and branch ${branchName}`);
 
-      # Upload the artifact using latest version
+      # Upload the artifact using latest version (only if branch is valid)
       - name: Upload PR information artifact
+        if: steps.validate.outputs.valid == 'true'
         uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4.6.2
         with:
           name: pr

changelog.d/23509_allow_headers_in_nats_jetstream_sink.enhancement.md

@@ -0,0 +1,5 @@
+The `nats` sink now supports message headers when publishing to JetStream.
+
+It introduces a configurable, templated Nats-Msg-Id header that ensures a unique ID for each message. This enables broker-level deduplication, resulting in stronger delivery guarantees and exactly-once semantics when combined with idempotent consumers.
+
+authors: benjamin-awd

changelog.d/23515_max_packet_size_mqtt.feature.md

@@ -0,0 +1,3 @@
+Added an option to set max packet size for MQTT source and sink.
+
+authors: simplepad

changelog.d/log_to_metric_pair_expansion_error.fix.md

@@ -0,0 +1,3 @@
+The `log_to_metric` transforms now emits a pair expansion error. This error was previously silently ignored.
+
+authors: pront

scripts/integration/nats/compose.yaml

@@ -43,3 +43,10 @@ services:
     - /usr/share/nats/config/nats-jwt.conf
     volumes:
     - ../../../tests/data/nats:/usr/share/nats/config
+  nats-jetstream-test:
+    image: docker.io/library/nats:${CONFIG_VERSION}
+    command:
+    - --config
+    - /usr/share/nats/config/nats-jetstream.conf
+    volumes:
+    - ../../../tests/data/nats:/usr/share/nats/config

scripts/integration/nats/test.yaml

@@ -1,7 +1,7 @@
 features:
-- nats-integration-tests
+  - nats-integration-tests
 
-test_filter: '::nats::'
+test_filter: "::nats::"
 
 env:
   NATS_ADDRESS: nats://nats:4222
@@ -11,17 +11,18 @@ env:
   NATS_TLS_CLIENT_CERT_ADDRESS: nats://nats-tls-client-cert:4222
   NATS_TOKEN_ADDRESS: nats://nats-token:4222
   NATS_USERPASS_ADDRESS: nats://nats-userpass:4222
+  NATS_JETSTREAM_ADDRESS: nats://nats-jetstream-test:4222
 
 matrix:
   version: [latest]
 
 # changes to these files/paths will invoke the integration test in CI
 # expressions are evaluated using https://github.com/micromatch/picomatch
 paths:
-- "src/internal_events/nats.rs"
-- "src/sources/nats.rs"
-- "src/sources/util/**"
-- "src/sinks/nats/**"
-- "src/sinks/util/**"
-- "src/nats.rs"
-- "scripts/integration/nats/**"
+  - "src/internal_events/nats.rs"
+  - "src/sources/nats.rs"
+  - "src/sources/util/**"
+  - "src/sinks/nats/**"
+  - "src/sinks/util/**"
+  - "src/nats.rs"
+  - "scripts/integration/nats/**"

src/common/mqtt.rs

@@ -41,6 +41,11 @@ pub struct MqttCommonConfig {
     #[derivative(Default(value = "default_keep_alive()"))]
     pub keep_alive: u16,
 
+    /// Maximum packet size
+    #[serde(default = "default_max_packet_size")]
+    #[derivative(Default(value = "default_max_packet_size()"))]
+    pub max_packet_size: usize,
+
     /// TLS configuration.
     #[configurable(derived)]
     pub tls: Option<TlsEnableableConfig>,
@@ -54,6 +59,10 @@ const fn default_keep_alive() -> u16 {
     60
 }
 
+const fn default_max_packet_size() -> usize {
+    10 * 1024
+}
+
 /// MQTT Error Types
 #[derive(Debug, Snafu)]
 #[snafu(visibility(pub))]

src/internal_events/expansion.rs

@@ -0,0 +1,46 @@
+use metrics::counter;
+use vector_lib::internal_event::{error_stage, error_type};
+use vector_lib::internal_event::{ComponentEventsDropped, InternalEvent, UNINTENTIONAL};
+
+pub struct PairExpansionError<'a> {
+    pub key: &'a str,
+    pub value: &'a str,
+    pub drop_event: bool,
+    pub error: serde_json::Error,
+}
+
+impl InternalEvent for PairExpansionError<'_> {
+    fn emit(self) {
+        let message = format!("Failed to expand key: `{}`:`{}`", self.key, self.value);
+
+        if self.drop_event {
+            error!(
+                message = %message,
+                error = %self.error,
+                error_type = error_type::PARSER_FAILED,
+                stage = error_stage::PROCESSING,
+
+            );
+
+            counter!(
+                "component_errors_total",
+                "error_type" => error_type::PARSER_FAILED,
+                "stage" => error_stage::PROCESSING,
+            )
+            .increment(1);
+
+            emit!(ComponentEventsDropped::<UNINTENTIONAL> {
+                count: 1,
+                reason: &message,
+            });
+        } else {
+            warn!(
+                message = %message,
+                error = %self.error,
+                error_type = error_type::PARSER_FAILED,
+                stage = error_stage::PROCESSING,
+
+            );
+        }
+    }
+}

src/internal_events/mod.rs

@@ -140,6 +140,11 @@ mod window;
 mod file;
 mod windows;
 
+#[cfg(any(feature = "transforms-log_to_metric", feature = "sinks-loki"))]
+mod expansion;
+#[cfg(any(feature = "transforms-log_to_metric", feature = "sinks-loki"))]
+pub use self::expansion::*;
+
 #[cfg(feature = "sources-mongodb_metrics")]
 pub(crate) use mongodb_metrics::*;
 

src/sinks/mqtt/config.rs

@@ -148,6 +148,7 @@ impl MqttSinkConfig {
             MaybeTlsSettings::from_config(self.common.tls.as_ref(), false).context(TlsSnafu)?;
         let mut options = MqttOptions::new(&client_id, &self.common.host, self.common.port);
         options.set_keep_alive(Duration::from_secs(self.common.keep_alive.into()));
+        options.set_max_packet_size(self.common.max_packet_size, self.common.max_packet_size);
         options.set_clean_session(self.clean_session);
         match (&self.common.user, &self.common.password) {
             (Some(user), Some(password)) => {