feat(ourlog): Add vercel log drain endpoint

CHANGELOG.md

@@ -21,6 +21,7 @@
 - Add `otp` and `two[-_]factor` to default scrubbing rules. ([#5250](https://github.com/getsentry/relay/pull/5250))
 - Add event merging logic for Playstation crashes. ([#5228](https://github.com/getsentry/relay/pull/5228))
 - Implement PII scrubbing for V2 spans. ([#5168](https://github.com/getsentry/relay/pull/5168))
+- Add vercel log drain endpoint. ([#5212](https://github.com/getsentry/relay/pull/5212))
 
 **Bug Fixes**:
 

relay-ourlogs/src/lib.rs

@@ -12,6 +12,6 @@ mod vercel_to_sentry;
 
 pub use self::otel_to_sentry::otel_to_sentry_log;
 pub use self::size::calculate_size;
-pub use self::vercel_to_sentry::vercel_log_to_sentry_log;
+pub use self::vercel_to_sentry::{VercelLog, vercel_log_to_sentry_log};
 
 pub use opentelemetry_proto::tonic::logs::v1 as otel_logs;

relay-ourlogs/src/vercel_to_sentry.rs

@@ -24,7 +24,7 @@ pub struct VercelLog {
     pub timestamp: i64,
     /// Identifier for the Vercel project.
     pub project_id: String,
-    // Log severity level.
+    /// Log severity level.
     pub level: VercelLogLevel,
     /// Log message content (may be truncated if over 256 KB).
     pub message: Option<String>,

relay-server/src/endpoints/integrations/mod.rs

@@ -1 +1,2 @@
 pub mod otlp;
+pub mod vercel;

relay-server/src/endpoints/integrations/vercel.rs

@@ -0,0 +1,51 @@
+use axum::extract::DefaultBodyLimit;
+use axum::http::StatusCode;
+use axum::response::IntoResponse;
+use axum::routing::{MethodRouter, post};
+use relay_config::Config;
+use relay_dynamic_config::Feature;
+
+use crate::endpoints::common;
+use crate::envelope::ContentType;
+use crate::extractors::{IntegrationBuilder, RawContentType};
+use crate::integrations::{LogsIntegration, VercelLogDrainFormat};
+use crate::service::ServiceState;
+
+/// All routes configured for the Vercel integration.
+///
+/// The integration currently supports the following endpoints:
+///  - Vercel Log Drain
+pub fn routes(config: &Config) -> axum::Router<ServiceState> {
+    axum::Router::new()
+        .route("/logs", logs::route(config))
+        .route("/logs/", logs::route(config))
+}
+
+mod logs {
+    use super::*;
+
+    async fn handle(
+        content_type: RawContentType,
+        state: ServiceState,
+        builder: IntegrationBuilder,
+    ) -> axum::response::Result<impl IntoResponse> {
+        let format = match ContentType::from(content_type.as_ref()) {
+            ContentType::Json => VercelLogDrainFormat::Json,
+            ContentType::NdJson => VercelLogDrainFormat::NdJson,
+            _ => return Ok(StatusCode::UNSUPPORTED_MEDIA_TYPE),
+        };
+
+        let envelope = builder
+            .with_type(LogsIntegration::VercelDrainLog { format })
+            .with_required_feature(Feature::VercelLogDrainEndpoint)
+            .build();
+
+        common::handle_envelope(&state, envelope).await?;
+
+        Ok(StatusCode::ACCEPTED)
+    }
+
+    pub fn route(config: &Config) -> MethodRouter<ServiceState> {
+        post(handle).route_layer(DefaultBodyLimit::max(config.max_envelope_size()))
+    }
+}

relay-server/src/endpoints/mod.rs

@@ -87,6 +87,7 @@ pub fn routes(config: &Config) -> Router<ServiceState>{
     // configured by users or protocols may force a specific variant.
     let integration_routes = Router::new()
         .nest("/api/{project_id}/integration/otlp", integrations::otlp::routes(config))
+        .nest("/api/{project_id}/integration/vercel", integrations::vercel::routes(config))
         .route_layer(middlewares::cors());
 
     // NOTE: If you add a new (non-experimental) route here, please also list it in

relay-server/src/envelope/content_type.rs

@@ -15,6 +15,8 @@ pub enum ContentType {
     Text,
     /// `application/json`
     Json,
+    /// `application/x-ndjson`
+    NdJson,
     /// `application/x-msgpack`
     MsgPack,
     /// `application/octet-stream`
@@ -45,6 +47,7 @@ impl ContentType {
         match self {
             Self::Text => "text/plain",
             Self::Json => "application/json",
+            Self::NdJson => "application/x-ndjson",
             Self::MsgPack => "application/x-msgpack",
             Self::OctetStream => "application/octet-stream",
             Self::Minidump => "application/x-dmp",
@@ -74,6 +77,8 @@ impl ContentType {
             Some(Self::Text)
         } else if ct.eq_ignore_ascii_case(Self::Json.as_str()) {
             Some(Self::Json)
+        } else if ct.eq_ignore_ascii_case(Self::NdJson.as_str()) {
+            Some(Self::NdJson)
         } else if ct.eq_ignore_ascii_case(Self::MsgPack.as_str()) {
             Some(Self::MsgPack)
         } else if ct.eq_ignore_ascii_case(Self::OctetStream.as_str()) {

relay-server/src/envelope/item.rs

@@ -160,6 +160,10 @@ impl Item {
                     (DataCategory::LogByte, self.len().max(1)),
                     (DataCategory::LogItem, item_count),
                 ],
+                Some(Integration::Logs(LogsIntegration::VercelDrainLog { .. })) => smallvec![
+                    (DataCategory::LogByte, self.len().max(1)),
+                    (DataCategory::LogItem, item_count),
+                ],
                 Some(Integration::Spans(SpansIntegration::OtelV1 { .. })) => {
                     smallvec![
                         (DataCategory::Span, item_count),

relay-server/src/integrations/mod.rs

@@ -42,6 +42,8 @@ define_integrations!(
     "application/vnd.sentry.integration.otel.logs+protobuf" => Integration::Logs(LogsIntegration::OtelV1 { format: OtelFormat::Protobuf }),
     "application/vnd.sentry.integration.otel.spans+json" => Integration::Spans(SpansIntegration::OtelV1 { format: OtelFormat::Json }),
     "application/vnd.sentry.integration.otel.spans+protobuf" => Integration::Spans(SpansIntegration::OtelV1 { format: OtelFormat::Protobuf }),
+    "application/vnd.sentry.integration.vercel.logs+json" => Integration::Logs(LogsIntegration::VercelDrainLog { format: VercelLogDrainFormat::Json }),
+    "application/vnd.sentry.integration.vercel.logs+ndjson" => Integration::Logs(LogsIntegration::VercelDrainLog { format: VercelLogDrainFormat::NdJson }),
 );
 
 /// An exhaustive list of all integrations supported by Relay.
@@ -74,6 +76,10 @@ pub enum LogsIntegration {
     ///
     /// Supports OTeL's [`LogsData`](opentelemetry_proto::tonic::logs::v1::LogsData).
     OtelV1 { format: OtelFormat },
+    /// The Vercel Log Drain integration.
+    ///
+    /// Supports the [`relay_ourlogs::VercelLog`] format.
+    VercelDrainLog { format: VercelLogDrainFormat },
 }
 
 /// All span integrations supported by Relay.
@@ -93,3 +99,11 @@ pub enum OtelFormat {
     /// OTeL data in a JSON container.
     Json,
 }
+
+#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
+pub enum VercelLogDrainFormat {
+    // Vercel Log Drain data in a JSON array payload
+    Json,
+    // Vercel Log Drain data in a newline delimited JSON payload
+    NdJson,
+}

relay-server/src/processing/logs/integrations/mod.rs

@@ -6,6 +6,7 @@ use crate::integrations::{Integration, LogsIntegration};
 use crate::managed::RecordKeeper;
 
 mod otel;
+mod vercel;
 
 /// Expands a list of [`Integration`] items into `result`.
 ///
@@ -43,6 +44,7 @@ pub fn expand_into(
 
         let result = match integration {
             LogsIntegration::OtelV1 { format } => otel::expand(format, &payload, produce),
+            LogsIntegration::VercelDrainLog { format } => vercel::expand(format, &payload, produce),
         };
 
         match result {

relay-server/src/processing/logs/integrations/vercel.rs

@@ -0,0 +1,52 @@
+use relay_event_schema::protocol::OurLog;
+use relay_ourlogs::VercelLog;
+
+use crate::integrations::VercelLogDrainFormat;
+use crate::processing::logs::{Error, Result};
+use crate::services::outcome::DiscardReason;
+
+/// Expands Vercel logs into the [`OurLog`] format.
+pub fn expand<F>(format: VercelLogDrainFormat, payload: &[u8], mut produce: F) -> Result<()>
+where
+    F: FnMut(OurLog),
+{
+    let mut count: i32 = 0;
+
+    match format {
+        VercelLogDrainFormat::Json => {
+            let logs = serde_json::from_slice::<Vec<VercelLog>>(payload).map_err(|e| {
+                relay_log::debug!(
+                    error = &e as &dyn std::error::Error,
+                    "Failed to parse logs data as JSON"
+                );
+                Error::Invalid(DiscardReason::InvalidJson)
+            })?;
+
+            for log in logs {
+                count += 1;
+                let ourlog = relay_ourlogs::vercel_log_to_sentry_log(log);
+                produce(ourlog);
+            }
+        }
+        VercelLogDrainFormat::NdJson => {
+            for line in payload.split(|&b| b == b'\n') {
+                if line.is_empty() {
+                    continue;
+                }
+
+                if let Ok(log) = serde_json::from_slice::<VercelLog>(line) {
+                    count += 1;
+                    let ourlog = relay_ourlogs::vercel_log_to_sentry_log(log);
+                    produce(ourlog);
+                }
+            }
+        }
+    }
+
+    if count == 0 {
+        relay_log::debug!("Failed to parse any logs from vercel log drain payload");
+        return Err(Error::Invalid(DiscardReason::InvalidJson));
+    }
+
+    Ok(())
+}

tests/integration/fixtures/__init__.py

@@ -257,6 +257,25 @@ def send_otel_logs(
 
         response.raise_for_status()
 
+    def send_vercel_logs(
+        self,
+        project_id,
+        data=None,
+        headers=None,
+        dsn_key_idx=0,
+        dsn_key=None,
+    ):
+
+        if dsn_key is None:
+            dsn_key = self.get_dsn_public_key(project_id, dsn_key_idx)
+
+        url = f"/api/{project_id}/integration/vercel/logs?sentry_key={dsn_key}"
+
+        response = self.post(url, headers=headers, data=data)
+
+        response.raise_for_status()
+        return response
+
     def send_options(self, project_id, headers=None, dsn_key_idx=0):
         headers = {
             "X-Sentry-Auth": self.get_auth_header(project_id, dsn_key_idx),