feat(gateway): add Datadog webhook source (#456)

Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>

Author: yordis

devops/docker/compose/.env.example

@@ -38,6 +38,9 @@ TROGON_GATEWAY_LOCAL_GITHUB_WEBHOOK_SECRET=local-dev-secret
 # --- Sentry Source ---
 # SENTRY_PRIMARY_CLIENT_SECRET=
 
+# --- Datadog Source ---
+# DATADOG_PRIMARY_WEBHOOK_TOKEN=
+
 # --- Discord Source ---
 # DISCORD_BOT_TOKEN=
 

devops/docker/compose/services/trogon-gateway/README.md

@@ -19,6 +19,7 @@ named TOML integrations, each with its own literal or env-backed secret.
 | Microsoft Graph change notifications | `/sources/microsoft-graph/{integration}/webhook` | `client_state` |
 | Notion | `/sources/notion/{integration}/webhook` | `verification_token` |
 | Sentry | `/sources/sentry/{integration}/webhook` | `client_secret` |
+| Datadog | `/sources/datadog/{integration}/webhook` | `webhook_token` |
 
 The gateway port is configured via `TROGON_GATEWAY_PORT` (default `8080`).
 Liveness and readiness probes are available at `GET /-/liveness` and `GET /-/readiness`.

devops/docker/compose/services/trogon-gateway/gateway.toml

@@ -81,6 +81,17 @@ webhook_secret = { env = "TROGON_GATEWAY_LOCAL_GITHUB_WEBHOOK_SECRET" }
 # [sources.sentry.integrations.primary.webhook]
 # client_secret = { env = "SENTRY_PRIMARY_CLIENT_SECRET" }
 #
+# [sources.datadog.integrations.primary]
+# subject_prefix = "datadog-primary"
+# stream_name = "DATADOG_PRIMARY"
+# stream_max_age_secs = 604800
+# nats_ack_timeout_secs = 10
+#
+# [sources.datadog.integrations.primary.webhook]
+# webhook_token = { env = "DATADOG_PRIMARY_WEBHOOK_TOKEN" }
+# webhook_token_header = "X-Datadog-Webhook-Token"  # optional; override the custom header name
+# timestamp_tolerance_secs = 300  # optional anti-replay window; requires "timestamp": "$DATE_POSIX" in the payload
+#
 # [sources.discord]
 # bot_token = { env = "DISCORD_BOT_TOKEN" }
 # gateway_intents = "guilds,guild_members,guild_messages,guild_message_reactions,direct_messages,message_content,guild_voice_states"

otel/semconv/registry/gateway.yaml

@@ -21,7 +21,7 @@ groups:
       - id: event_id
         type: string
         stability: development
-        brief: Webhook event identifier (Slack, Notion).
+        brief: Webhook event identifier (Slack, Notion, Datadog).
       - id: event
         type: string
         stability: development
@@ -157,3 +157,12 @@ groups:
     attributes:
       - ref: notification_count
       - ref: subject
+  - id: span.datadog.webhook
+    type: span
+    span_kind: server
+    stability: development
+    brief: Datadog webhook ingestion.
+    attributes:
+      - ref: event_type
+      - ref: event_id
+      - ref: subject

rsworkspace/crates/trogon-gateway/README.md

@@ -44,6 +44,7 @@ configured as a named integration and mounted at its integration path.
 | Microsoft Graph change notifications | `/sources/microsoft-graph/{integration}/webhook` |
 | Notion | `/sources/notion/{integration}/webhook` |
 | Sentry | `/sources/sentry/{integration}/webhook` |
+| Datadog | `/sources/datadog/{integration}/webhook` |
 
 Integration IDs must contain only ASCII letters, numbers, `_`, or `-`. They cannot be empty, contain path
 separators, or exceed 64 characters.
@@ -70,6 +71,7 @@ or explicit environment reference shape.
 | Microsoft Graph change notifications | `client_state` |
 | Notion | `verification_token` |
 | Sentry | `client_secret` |
+| Datadog | `webhook_token` |
 
 ## Core configuration
 
@@ -103,6 +105,8 @@ Source-specific extras:
 - Telegram `webhook_registration_mode = "startup"` attempts registration on startup and requires `bot_token` plus `public_webhook_url`
 - Linear `timestamp_tolerance_secs` (default: `60`, `0` disables tolerance)
 - Twitter/X `consumer_secret` is used for both CRC responses and `x-twitter-webhooks-signature` validation
+- Datadog webhooks are not signed. `webhook_token` is a shared secret the operator sends as a custom header (configured in the Datadog webhook's *Custom Headers*), verified in constant time. The header name defaults to `X-Datadog-Webhook-Token` and can be overridden per integration with `webhook_token_header`. Routing keys off an `event_type` field in the payload template, falling back to `.unroutable` when it is absent or invalid; an optional `id` field is used as the NATS message ID for deduplication
+- Datadog `timestamp_tolerance_secs` (optional, disabled by default) enables a freshness/anti-replay check. Datadog sends no timestamp header, so the operator must template a `timestamp` field carrying `$DATE_POSIX` (POSIX seconds) into the payload. When enabled, requests whose `timestamp` is outside the window are rejected (`401`), and requests missing/with an unparseable `timestamp` are rejected (`400`). `0` or unset disables the check
 
 ## Config file shape
 
@@ -163,8 +167,34 @@ verification_token = "notion-verification-token-example"
 
 [sources.sentry.integrations.primary.webhook]
 client_secret = "sentry-client-secret"
+
+[sources.datadog.integrations.primary.webhook]
+webhook_token = { env = "DATADOG_PRIMARY_WEBHOOK_TOKEN" }
+# webhook_token_header = "X-Datadog-Webhook-Token"  # optional; override the custom header name
+# timestamp_tolerance_secs = 300  # optional anti-replay window; requires "timestamp": "$DATE_POSIX" in the payload
+```
+
+Configure the matching Datadog webhook (Integrations > Webhooks) with a custom header and a payload template, for example:
+
+```json
+{
+  "X-Datadog-Webhook-Token": "$WEBHOOK_TOKEN"
+}
 ```
 
+```json
+{
+  "id": "$ID",
+  "event_type": "$EVENT_TYPE",
+  "timestamp": "$DATE_POSIX",
+  "title": "$EVENT_TITLE",
+  "body": "$EVENT_MSG",
+  "alert_transition": "$ALERT_TRANSITION"
+}
+```
+
+`timestamp` is only required when `timestamp_tolerance_secs` is set; `$DATE_POSIX` is POSIX epoch seconds (`$DATE` is epoch milliseconds, which this source does not accept).
+
 ## Microsoft Graph change notifications
 
 This source receives Microsoft Graph change notifications. It does not implement

rsworkspace/crates/trogon-gateway/src/config.rs

@@ -1,6 +1,7 @@
 use std::collections::BTreeMap;
 use std::path::Path;
 
+use crate::source::datadog::DatadogWebhookToken;
 use crate::source::discord::config::DiscordBotToken;
 use crate::source::github::config::GitHubWebhookSecret;
 use crate::source::gitlab::GitLabSigningToken;
@@ -19,6 +20,7 @@ use crate::source::telegram::config::{
 };
 use crate::source::twitter::config::TwitterConsumerSecret;
 use crate::source_integration_id::{SourceIntegrationId, SourceIntegrationIdError};
+use axum::http::HeaderName;
 use confique::Config;
 #[cfg(test)]
 use trogon_nats::NatsAuth;
@@ -350,6 +352,8 @@ struct SourcesConfig {
     notion: NotionConfig,
     #[config(nested)]
     sentry: SentryConfig,
+    #[config(nested)]
+    datadog: DatadogConfig,
 }
 
 #[derive(Config)]
@@ -448,6 +452,14 @@ struct SentryConfig {
     integrations: BTreeMap<String, SourceIntegrationInput<SentryWebhookConfig>>,
 }
 
+#[derive(Config)]
+#[config(layer_attr(serde(deny_unknown_fields)))]
+struct DatadogConfig {
+    status: Option<String>,
+    #[config(default = {})]
+    integrations: BTreeMap<String, SourceIntegrationInput<DatadogWebhookConfig>>,
+}
+
 #[derive(serde::Deserialize)]
 #[serde(deny_unknown_fields)]
 struct SourceIntegrationInput<T> {
@@ -544,6 +556,14 @@ struct SentryWebhookConfig {
     client_secret: Option<SecretInput>,
 }
 
+#[derive(serde::Deserialize)]
+#[serde(deny_unknown_fields)]
+struct DatadogWebhookConfig {
+    webhook_token: Option<SecretInput>,
+    webhook_token_header: Option<String>,
+    timestamp_tolerance_secs: Option<u64>,
+}
+
 pub struct ResolvedHttpServerConfig {
     pub port: u16,
 }
@@ -573,6 +593,7 @@ pub struct ResolvedConfig {
     pub microsoft_graph: Vec<SourceIntegration<crate::source::microsoft_graph::MicrosoftGraphConfig>>,
     pub notion: Vec<SourceIntegration<crate::source::notion::NotionConfig>>,
     pub sentry: Vec<SourceIntegration<crate::source::sentry::SentryConfig>>,
+    pub datadog: Vec<SourceIntegration<crate::source::datadog::DatadogConfig>>,
 }
 
 impl ResolvedConfig {
@@ -588,6 +609,7 @@ impl ResolvedConfig {
             || !self.microsoft_graph.is_empty()
             || !self.notion.is_empty()
             || !self.sentry.is_empty()
+            || !self.datadog.is_empty()
     }
 }
 
@@ -620,6 +642,7 @@ fn resolve(cfg: GatewayConfig, nats_overrides: &NatsArgs) -> Result<ResolvedConf
     let microsoft_graph = resolve_microsoft_graph_integrations(cfg.sources.microsoft_graph, &env, &mut errors);
     let notion = resolve_notion_integrations(cfg.sources.notion, &env, &mut errors);
     let sentry = resolve_sentry_integrations(cfg.sources.sentry, &env, &mut errors);
+    let datadog = resolve_datadog_integrations(cfg.sources.datadog, &env, &mut errors);
 
     if !errors.is_empty() {
         return Err(ConfigError::Validation(ValidationErrors(errors)));
@@ -641,6 +664,7 @@ fn resolve(cfg: GatewayConfig, nats_overrides: &NatsArgs) -> Result<ResolvedConf
         microsoft_graph,
         notion,
         sentry,
+        datadog,
     })
 }
 
@@ -1703,6 +1727,97 @@ fn resolve_sentry_integrations(
     integrations
 }
 
+fn resolve_datadog_integrations(
+    section: DatadogConfig,
+    env: &impl ReadEnv,
+    errors: &mut Vec<ConfigValidationError>,
+) -> Vec<SourceIntegration<crate::source::datadog::DatadogConfig>> {
+    let DatadogConfig {
+        status,
+        integrations: configured_integrations,
+    } = section;
+    let mut integrations = Vec::new();
+    if !resolve_source_status("datadog", status.as_deref(), errors) {
+        return integrations;
+    }
+    for (raw_id, integration) in configured_integrations {
+        let Some(id) = resolve_integration_id("datadog", raw_id, errors) else {
+            continue;
+        };
+        if !resolve_integration_source_status("datadog", &id, integration.status.as_deref(), errors) {
+            continue;
+        }
+        let Some(webhook) = integration.webhook else {
+            errors.push(ConfigValidationError::missing_integration("datadog", &id, "webhook"));
+            continue;
+        };
+        let Some(token) =
+            require_integration_value("datadog", &id, "webhook_token", webhook.webhook_token, env, errors)
+        else {
+            continue;
+        };
+        let webhook_token = match DatadogWebhookToken::new(token) {
+            Ok(token) => token,
+            Err(error) => {
+                errors.push(ConfigValidationError::invalid_integration(
+                    "datadog",
+                    &id,
+                    "webhook_token",
+                    error,
+                ));
+                continue;
+            }
+        };
+        let Some((subject_prefix, stream_name, stream_max_age, nats_ack_timeout)) = resolve_common_integration_fields(
+            CommonIntegrationFieldsInput {
+                source: "datadog",
+                id: &id,
+                subject_source_prefix: "datadog",
+                stream_source_prefix: "DATADOG",
+                subject_prefix: integration.subject_prefix,
+                stream_name: integration.stream_name,
+                stream_max_age_secs: integration.stream_max_age_secs,
+                nats_ack_timeout_secs: integration.nats_ack_timeout_secs,
+                default_nats_ack_timeout_secs: DEFAULT_NATS_ACK_TIMEOUT_SECS,
+            },
+            errors,
+        ) else {
+            continue;
+        };
+        let webhook_token_header = match webhook.webhook_token_header {
+            Some(raw) => match HeaderName::try_from(raw.trim()) {
+                Ok(header) => header,
+                Err(error) => {
+                    errors.push(ConfigValidationError::invalid_integration(
+                        "datadog",
+                        &id,
+                        "webhook_token_header",
+                        error,
+                    ));
+                    continue;
+                }
+            },
+            None => HeaderName::from_static(crate::source::datadog::constants::DEFAULT_HEADER_WEBHOOK_TOKEN),
+        };
+        let timestamp_tolerance = webhook
+            .timestamp_tolerance_secs
+            .and_then(|secs| NonZeroDuration::from_secs(secs).ok());
+        integrations.push(SourceIntegration::new(
+            id,
+            crate::source::datadog::DatadogConfig {
+                webhook_token,
+                webhook_token_header,
+                subject_prefix,
+                stream_name,
+                stream_max_age,
+                nats_ack_timeout,
+                timestamp_tolerance,
+            },
+        ));
+    }
+    integrations
+}
+
 fn resolve_integration_id(
     source: &'static str,
     raw_id: String,