feat: Add mezmo_analytics source

Cargo.toml

@@ -695,6 +695,7 @@ sources-logs-mezmo = [
   "sources-aws_s3",
   "sources-datadog_agent",
   "sources-demo_logs",
+  "sources-mezmo_analytics",
   "sources-mezmo_demo_logs",
   "sources-exec",
   "sources-fluent",
@@ -754,6 +755,7 @@ sources-host_metrics = ["heim/cpu", "heim/host", "heim/memory", "heim/net"]
 sources-http_client = ["sources-utils-http-client"]
 sources-http_server = ["sources-utils-http", "sources-utils-http-headers", "sources-utils-http-query"]
 sources-internal_logs = []
+sources-mezmo_analytics = ["sources-internal_logs"]
 sources-mezmo_pipeline_state_variable_change = []
 sources-mezmo_user_logs = ["sources-internal_logs"]
 sources-internal_metrics = []

lib/vector-core/Cargo.toml

@@ -53,7 +53,7 @@ snafu.workspace = true
 socket2.workspace = true
 tokio = { workspace = true, features = ["net"] }
 tokio-openssl = { version = "0.6.5", default-features = false }
-tokio-stream = { workspace = true, features = ["time"], optional = true }
+tokio-stream = { workspace = true, features = ["sync", "time"] }
 tokio-util = { version = "0.7.0", default-features = false, features = ["time"] }
 tokio-postgres = { version = "0.7.7", default-features = false, features = ["runtime", "with-chrono-0_4", "with-uuid-1", "with-serde_json-1"] }
 toml.workspace = true
@@ -104,7 +104,7 @@ vector-common = { path = "../vector-common", default-features = false, features
 
 [features]
 default = []
-lua = ["dep:mlua", "dep:tokio-stream", "vrl/lua"]
+lua = ["dep:mlua", "vrl/lua"]
 vrl = []
 test = ["vector-common/test", "proptest"]
 pgbouncer-integration-tests = ["vrl"]

lib/vector-core/src/mezmo/analytics.rs

@@ -0,0 +1,146 @@
+//! In-process delivery for Mezmo analytics records.
+
+use std::sync::OnceLock;
+
+use futures::{Stream, StreamExt, future::ready};
+use tokio::sync::broadcast::{self, Receiver, Sender};
+use tokio_stream::wrappers::{BroadcastStream, errors::BroadcastStreamRecvError};
+use tracing::warn;
+use vector_common::internal_event::{self, ComponentEventsDropped, UNINTENTIONAL};
+
+use crate::event::LogEvent;
+
+const ANALYTICS_CHANNEL_CAPACITY: usize = 1_000;
+const ANALYTICS_OUTPUT_COUNT: usize = 5;
+
+type AnalyticsSenders = [Sender<AnalyticsEventBatch>; ANALYTICS_OUTPUT_COUNT];
+
+static ANALYTICS: OnceLock<AnalyticsSenders> = OnceLock::new();
+
+/// A named output exposed by the `mezmo_analytics` source.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum AnalyticsOutput {
+    UsageMetrics,
+    UsageMetricsByAnnotations,
+    LogClusters,
+    LogClusterSamples,
+    LogClusterUsage,
+}
+
+impl AnalyticsOutput {
+    /// Returns the source output name.
+    pub const fn as_str(self) -> &'static str {
+        match self {
+            Self::UsageMetrics => "usage_metrics",
+            Self::UsageMetricsByAnnotations => "usage_metrics_by_annotations",
+            Self::LogClusters => "log_clusters",
+            Self::LogClusterSamples => "log_cluster_samples",
+            Self::LogClusterUsage => "log_cluster_usage",
+        }
+    }
+
+    /// Returns every named output in declaration order.
+    pub const fn all() -> [Self; ANALYTICS_OUTPUT_COUNT] {
+        [
+            Self::UsageMetrics,
+            Self::UsageMetricsByAnnotations,
+            Self::LogClusters,
+            Self::LogClusterSamples,
+            Self::LogClusterUsage,
+        ]
+    }
+}
+
+/// A batch of analytics records destined for one named source output.
+#[derive(Clone, Debug)]
+pub struct AnalyticsEventBatch {
+    output: AnalyticsOutput,
+    events: Vec<LogEvent>,
+}
+
+impl AnalyticsEventBatch {
+    /// Creates a batch for one named output.
+    pub fn new(output: AnalyticsOutput, events: Vec<LogEvent>) -> Self {
+        Self { output, events }
+    }
+
+    /// Returns the named output associated with this batch.
+    pub const fn output(&self) -> AnalyticsOutput {
+        self.output
+    }
+
+    /// Returns the records in this batch.
+    pub fn events(&self) -> &[LogEvent] {
+        &self.events
+    }
+
+    /// Splits the batch into its named output and records.
+    pub fn into_parts(self) -> (AnalyticsOutput, Vec<LogEvent>) {
+        (self.output, self.events)
+    }
+}
+
+fn senders() -> &'static AnalyticsSenders {
+    ANALYTICS
+        .get_or_init(|| std::array::from_fn(|_| broadcast::channel(ANALYTICS_CHANNEL_CAPACITY).0))
+}
+
+/// Builds and publishes batches if a `mezmo_analytics` source is subscribed.
+pub fn publish<F, B>(build_batches: F)
+where
+    F: FnOnce() -> B,
+    B: IntoIterator<Item = AnalyticsEventBatch>,
+{
+    let Some(senders) = ANALYTICS.get() else {
+        return;
+    };
+    if senders.iter().all(|sender| sender.receiver_count() == 0) {
+        return;
+    }
+
+    for batch in build_batches() {
+        let sender = &senders[batch.output() as usize];
+        if sender.receiver_count() > 0 {
+            let _ = sender.send(batch);
+        }
+    }
+}
+
+/// A subscription to analytics batches produced inside Vector.
+pub struct AnalyticsSubscription {
+    receiver: Receiver<AnalyticsEventBatch>,
+}
+
+impl AnalyticsSubscription {
+    /// Subscribes to analytics batches for one output produced after this call.
+    pub fn subscribe(output: AnalyticsOutput) -> Self {
+        Self {
+            receiver: senders()[output as usize].subscribe(),
+        }
+    }
+
+    /// Subscribes to analytics batches for every output.
+    pub fn subscribe_all() -> Vec<Self> {
+        AnalyticsOutput::all()
+            .into_iter()
+            .map(Self::subscribe)
+            .collect()
+    }
+
+    /// Converts this subscription into its analytics batch stream.
+    pub fn into_stream(self) -> impl Stream<Item = AnalyticsEventBatch> + Unpin {
+        BroadcastStream::new(self.receiver).filter_map(|received| {
+            ready(match received {
+                Ok(batch) => Some(batch),
+                Err(error @ BroadcastStreamRecvError::Lagged(dropped_batches)) => {
+                    warn!(message = "Mezmo analytics source lagged behind.", %error);
+                    internal_event::emit(ComponentEventsDropped::<UNINTENTIONAL> {
+                        count: usize::try_from(dropped_batches).unwrap_or(usize::MAX),
+                        reason: "Mezmo analytics source lagged behind.",
+                    });
+                    None
+                }
+            })
+        })
+    }
+}

lib/vector-core/src/mezmo/mod.rs

@@ -1 +1,2 @@
+pub mod analytics;
 pub mod postgres;

lib/vector-core/src/usage_metrics/flusher.rs

@@ -1,4 +1,4 @@
-use super::{AnnotationMap, AnnotationSet, UsageMetricsKey, UsageMetricsValue};
+use super::{AnnotationMap, AnnotationSet, UsageMetricsKey, UsageMetricsValue, processor_name};
 use crate::mezmo;
 use async_trait::async_trait;
 use chrono::Utc;
@@ -24,8 +24,6 @@ const INSERT_BILLING_QUERY: &str = "INSERT INTO usage_metrics (event_ts, account
 const INSERT_PROFILES_QUERY: &str = "INSERT INTO usage_metrics_by_annotations (ts, account_id, component_id, count, size, annotations) VALUES ($1, $2, $3, $4, $5, $6) ON CONFLICT DO NOTHING";
 
 const DB_MAX_PARALLEL_EXECUTIONS: usize = 8;
-const VECTOR_VERSION: &str = env!("CARGO_PKG_VERSION");
-
 #[async_trait]
 pub(crate) trait MetricsFlusher: Sync {
     async fn save_billing_metrics(&self, metrics: HashMap<UsageMetricsKey, UsageMetricsValue>);
@@ -67,7 +65,7 @@ impl DbFlusher {
                 DbFlusherError::QueryError
             })?;
 
-        let processor_name = format!("app=vector,pod={pod_name},version={VECTOR_VERSION}");
+        let processor_name = processor_name(pod_name);
         Ok(Self {
             conn_str,
             processor_name,
@@ -249,7 +247,7 @@ impl HttpFlusher {
         headers: HashMap<String, String>,
         max_delay: Duration,
     ) -> Self {
-        let processor_name = format!("app=vector,pod={pod_name},version={VECTOR_VERSION}");
+        let processor_name = processor_name(pod_name);
 
         HttpFlusher {
             client: Client::new(),