Merge branch 'main' into modularize-alerts

Author: nikhilsinhaparseable

Cargo.lock

@@ -33,7 +33,7 @@ actix-web-prometheus = { version = "0.1" }
 actix-web-static-files = "4.0"
 http = "0.2.7"
 http-auth-basic = "0.3.3"
-tonic = { version = "0.12.3", features = ["tls", "transport", "gzip", "zstd"] }
+tonic = { version = "0.12.3", features = ["tls", "transport", "gzip", "zstd", "prost"] }
 tonic-web = "0.12.3"
 tower-http = { version = "0.6.1", features = ["cors"] }
 url = "2.4.0"
@@ -76,7 +76,13 @@ tokio-stream = { version = "0.1", features = ["fs"] }
 tokio-util = { version = "0.7" }
 
 # Logging and Metrics
-opentelemetry-proto = { git = "https://github.com/parseablehq/opentelemetry-rust", branch = "fix-metrics-u64-serialization" }
+opentelemetry-proto = { version = "0.30.0", features = [
+    "gen-tonic",
+    "with-serde",
+    "logs",
+    "metrics",
+    "trace",
+] }
 prometheus = { version = "0.13", features = ["process"] }
 prometheus-parse = "0.2.5"
 tracing = "0.1"
@@ -133,6 +139,7 @@ xxhash-rust = { version = "0.8", features = ["xxh3"] }
 futures-core = "0.3.31"
 tempfile = "3.20.0"
 lazy_static = "1.4.0"
+prost = "0.13.1"
 
 [build-dependencies]
 cargo_toml = "0.21"

Cargo.toml

@@ -16,24 +16,24 @@
  *
  */
 
-use async_trait::async_trait;
-use futures_util::StreamExt;
-use rdkafka::consumer::{CommitMode, Consumer};
-use serde_json::Value;
-use std::collections::HashMap;
-use std::sync::Arc;
-use tokio_stream::wrappers::ReceiverStream;
-use tracing::{debug, error};
-
 use crate::{
     connectors::common::processor::Processor,
     event::{
         Event as ParseableEvent, USER_AGENT_KEY,
         format::{EventFormat, LogSourceEntry, json},
     },
+    handlers::TelemetryType,
     parseable::PARSEABLE,
     storage::StreamType,
 };
+use async_trait::async_trait;
+use futures_util::StreamExt;
+use rdkafka::consumer::{CommitMode, Consumer};
+use serde_json::Value;
+use std::collections::HashMap;
+use std::sync::Arc;
+use tokio_stream::wrappers::ReceiverStream;
+use tracing::{debug, error};
 
 use super::{ConsumerRecord, StreamConsumer, TopicPartition, config::BufferConfig};
 
@@ -50,13 +50,13 @@ impl ParseableSinkProcessor {
             .map(|r| r.topic.as_str())
             .unwrap_or_default();
         let log_source_entry = LogSourceEntry::default();
-
         PARSEABLE
             .create_stream_if_not_exists(
                 stream_name,
                 StreamType::UserDefined,
                 None,
                 vec![log_source_entry],
+                TelemetryType::default(),
             )
             .await?;