) -> anyhow::Result<()> {
+ let len = records.len();
+ debug!("Processing {} records", len);
+
+ self.build_event_from_chunk(&records)
+ .await?
+ .process()
+ .await?;
+
+ debug!("Processed {} records", len);
+ Ok(())
+ }
+}
+
+#[derive(Clone)]
+pub struct StreamWorker
+where
+ P: Processor, ()>,
+{
+ processor: Arc,
+ consumer: Arc,
+ buffer_config: BufferConfig,
+}
+
+impl StreamWorker
+where
+ P: Processor, ()> + Send + Sync + 'static,
+{
+ pub fn new(processor: Arc, consumer: Arc) -> Self {
+ let buffer_config = consumer
+ .context()
+ .config()
+ .consumer()
+ .expect("Consumer config is missing")
+ .buffer_config();
+
+ Self {
+ processor,
+ consumer,
+ buffer_config,
+ }
+ }
+
+ pub async fn process_partition(
+ &self,
+ tp: TopicPartition,
+ record_stream: ReceiverStream,
+ ) -> anyhow::Result<()> {
+ let chunked_stream = tokio_stream::StreamExt::chunks_timeout(
+ record_stream,
+ self.buffer_config.buffer_size,
+ self.buffer_config.buffer_timeout,
+ );
+
+ chunked_stream
+ .for_each_concurrent(None, |records| async {
+ if let Some(last_record) = records.iter().max_by_key(|r| r.offset) {
+ let tpl = last_record.offset_to_commit().unwrap();
+
+ if let Err(e) = self.processor.process(records).await {
+ error!("Failed to process records for {:?}: {:?}", tp, e);
+ }
+
+ //CommitMode::Async race condition.
+ //@see https://github.com/confluentinc/librdkafka/issues/4534
+ //@see https://github.com/confluentinc/librdkafka/issues/4059
+ if let Err(e) = self.consumer.commit(&tpl, CommitMode::Sync) {
+ error!(error = %e, "Failed to commit offsets for {:?}", tpl);
+ } else {
+ debug!("Committed offsets for {:?}", tpl);
+ }
+ }
+ })
+ .await;
+
+ self.processor.post_stream().await?;
+
+ Ok(())
+ }
+}
diff --git a/src/connectors/kafka/rebalance_listener.rs b/src/connectors/kafka/rebalance_listener.rs
new file mode 100644
index 000000000..8372e1b5b
--- /dev/null
+++ b/src/connectors/kafka/rebalance_listener.rs
@@ -0,0 +1,87 @@
+/*
+ * Parseable Server (C) 2022 - 2024 Parseable, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see .
+ *
+ */
+
+use crate::connectors::common::shutdown::Shutdown;
+use crate::connectors::kafka::state::StreamState;
+use crate::connectors::kafka::RebalanceEvent;
+use std::sync::Arc;
+use tokio::sync::RwLock;
+use tokio::{runtime::Handle, sync::mpsc::Receiver};
+use tracing::{info, warn};
+
+pub struct RebalanceListener {
+ rebalance_rx: Receiver,
+ stream_state: Arc>,
+ shutdown_handle: Shutdown,
+}
+
+impl RebalanceListener {
+ pub fn new(
+ rebalance_rx: Receiver,
+ stream_state: Arc>,
+ shutdown_handle: Shutdown,
+ ) -> Self {
+ Self {
+ rebalance_rx,
+ stream_state,
+ shutdown_handle,
+ }
+ }
+
+ pub fn start(self) {
+ let mut rebalance_receiver = self.rebalance_rx;
+ let stream_state = self.stream_state.clone();
+ let shutdown_handle = self.shutdown_handle.clone();
+ let tokio_runtime_handle = Handle::current();
+
+ std::thread::Builder::new().name("rebalance-listener-thread".to_string()).spawn(move || {
+ tokio_runtime_handle.block_on(async move {
+ loop {
+ tokio::select! {
+ rebalance = rebalance_receiver.recv() => {
+ match rebalance {
+ Some(RebalanceEvent::Assign(tpl)) => info!("RebalanceEvent Assign: {:?}", tpl),
+ Some(RebalanceEvent::Revoke(tpl, callback)) => {
+ info!("RebalanceEvent Revoke: {:?}", tpl);
+ if let Ok(mut stream_state) = stream_state.try_write() {
+ stream_state.terminate_partition_streams(tpl).await;
+ drop(stream_state);
+ } else {
+ warn!("Stream state lock is busy, skipping rebalance revoke for {:?}", tpl);
+ }
+ if let Err(err) = callback.send(()) {
+ warn!("Error during sending response to context. Cause: {:?}", err);
+ }
+ info!("Finished Rebalance Revoke");
+ }
+ None => {
+ info!("Rebalance event sender is closed!");
+ break
+ }
+ }
+ },
+ _ = shutdown_handle.recv() => {
+ info!("Gracefully stopping rebalance listener!");
+ break;
+ },
+ }
+ }
+ })
+ }).expect("Failed to start rebalance listener thread");
+ }
+}
diff --git a/src/connectors/kafka/sink.rs b/src/connectors/kafka/sink.rs
new file mode 100644
index 000000000..447955686
--- /dev/null
+++ b/src/connectors/kafka/sink.rs
@@ -0,0 +1,94 @@
+/*
+ * Parseable Server (C) 2022 - 2024 Parseable, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see .
+ *
+ */
+use crate::connectors::common::build_runtime;
+use crate::connectors::common::processor::Processor;
+use crate::connectors::kafka::consumer::KafkaStreams;
+use crate::connectors::kafka::processor::StreamWorker;
+use crate::connectors::kafka::ConsumerRecord;
+use anyhow::Result;
+use futures_util::StreamExt;
+use rdkafka::consumer::Consumer;
+use std::sync::Arc;
+use tokio::runtime::Runtime;
+use tracing::{error, info};
+
+pub struct KafkaSinkConnector
+where
+ P: Processor, ()>,
+{
+ streams: KafkaStreams,
+ stream_processor: Arc>,
+ runtime: Runtime,
+}
+
+impl KafkaSinkConnector
+where
+ P: Processor, ()> + Send + Sync + 'static,
+{
+ pub fn new(kafka_streams: KafkaStreams, processor: P) -> Self {
+ let consumer = kafka_streams.consumer();
+ let stream_processor = Arc::new(StreamWorker::new(
+ Arc::new(processor),
+ Arc::clone(&consumer),
+ ));
+
+ let runtime = build_runtime(
+ consumer.context().config.partition_listener_concurrency,
+ "kafka-sink-worker",
+ )
+ .expect("Failed to build runtime");
+ let _ = runtime.enter();
+
+ Self {
+ streams: kafka_streams,
+ stream_processor,
+ runtime,
+ }
+ }
+
+ pub async fn run(self) -> Result<()> {
+ self.streams
+ .partitioned()
+ .map(|partition_stream| {
+ let worker = Arc::clone(&self.stream_processor);
+ let tp = partition_stream.topic_partition().clone();
+ self.runtime.spawn(async move {
+ partition_stream
+ .run_drain(|partition_records| async {
+ info!("Starting task for partition: {:?}", tp);
+
+ worker
+ .process_partition(tp.clone(), partition_records)
+ .await
+ .unwrap();
+ })
+ .await;
+
+ info!("Task completed for partition: {:?}", tp);
+ })
+ })
+ .for_each_concurrent(None, |task| async {
+ if let Err(e) = task.await {
+ error!("Task failed: {:?}", e);
+ }
+ })
+ .await;
+
+ Ok(())
+ }
+}
diff --git a/src/connectors/kafka/state.rs b/src/connectors/kafka/state.rs
new file mode 100644
index 000000000..7b20ae762
--- /dev/null
+++ b/src/connectors/kafka/state.rs
@@ -0,0 +1,68 @@
+/*
+ * Parseable Server (C) 2022 - 2024 Parseable, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see .
+ *
+ */
+
+use crate::connectors::kafka::partition_stream::PartitionStreamSender;
+use crate::connectors::kafka::{TopicPartition, TopicPartitionList};
+use std::collections::HashMap;
+use tracing::info;
+
+pub struct StreamState {
+ partition_senders: HashMap,
+}
+
+impl StreamState {
+ pub fn new(capacity: usize) -> Self {
+ Self {
+ partition_senders: HashMap::with_capacity(capacity),
+ }
+ }
+
+ pub fn insert_partition_sender(
+ &mut self,
+ tp: TopicPartition,
+ sender: PartitionStreamSender,
+ ) -> Option {
+ self.partition_senders.insert(tp, sender)
+ }
+
+ pub fn get_partition_sender(&self, tp: &TopicPartition) -> Option<&PartitionStreamSender> {
+ self.partition_senders.get(tp)
+ }
+
+ pub async fn terminate_partition_streams(&mut self, tpl: TopicPartitionList) {
+ info!("Terminating streams: {:?}", tpl);
+
+ for tp in tpl.tpl {
+ if let Some(sender) = self.partition_senders.remove(&tp) {
+ info!("Terminating stream for {:?}", tp);
+ sender.terminate();
+ drop(sender);
+ info!("Stream terminated for {:?}", tp);
+ } else {
+ info!("Stream already completed for {:?}", tp);
+ }
+ }
+
+ info!("All streams terminated!");
+ }
+
+ pub fn clear(&mut self) {
+ info!("Clearing all stream states...");
+ self.partition_senders.clear();
+ }
+}
diff --git a/src/connectors/mod.rs b/src/connectors/mod.rs
new file mode 100644
index 000000000..757c49a97
--- /dev/null
+++ b/src/connectors/mod.rs
@@ -0,0 +1,100 @@
+/*
+ * Parseable Server (C) 2022 - 2024 Parseable, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see .
+ *
+ */
+
+use crate::connectors::common::processor::Processor;
+use crate::connectors::common::shutdown::Shutdown;
+use crate::connectors::kafka::config::KafkaConfig;
+use crate::connectors::kafka::consumer::KafkaStreams;
+use crate::connectors::kafka::metrics::KafkaMetricsCollector;
+use crate::connectors::kafka::processor::ParseableSinkProcessor;
+use crate::connectors::kafka::rebalance_listener::RebalanceListener;
+use crate::connectors::kafka::sink::KafkaSinkConnector;
+use crate::connectors::kafka::state::StreamState;
+use crate::connectors::kafka::{ConsumerRecord, KafkaContext};
+use crate::option::{Mode, CONFIG};
+use actix_web_prometheus::PrometheusMetrics;
+use prometheus::Registry;
+use std::sync::Arc;
+use tokio::sync::RwLock;
+use tracing::{info, warn};
+
+pub mod common;
+pub mod kafka;
+
+pub async fn init(prometheus: &PrometheusMetrics) -> anyhow::Result<()> {
+ if matches!(CONFIG.options.mode, Mode::Ingest | Mode::All) {
+ match CONFIG.kafka_config.validate() {
+ Err(e) => {
+ warn!("Kafka connector configuration invalid. {}", e);
+ }
+ Ok(_) => {
+ let config = CONFIG.kafka_config.clone();
+ let shutdown_handle = Shutdown::default();
+ let registry = prometheus.registry.clone();
+ let processor = ParseableSinkProcessor;
+
+ tokio::spawn({
+ let shutdown_handle = shutdown_handle.clone();
+ async move {
+ shutdown_handle.signal_listener().await;
+ info!("Connector received shutdown signal!");
+ }
+ });
+
+ run_kafka2parseable(config, registry, processor, shutdown_handle).await?;
+ }
+ }
+ }
+
+ Ok(())
+}
+
+async fn run_kafka2parseable(
+ config: KafkaConfig,
+ registry: Registry,
+ processor: P,
+ shutdown_handle: Shutdown,
+) -> anyhow::Result<()>
+where
+ P: Processor, ()> + Send + Sync + 'static,
+{
+ info!("Initializing KafkaSink connector...");
+
+ let kafka_config = Arc::new(config.clone());
+ let (kafka_context, rebalance_rx) = KafkaContext::new(kafka_config);
+
+ //TODO: fetch topics metadata from kafka then give dynamic value to StreamState
+ let stream_state = Arc::new(RwLock::new(StreamState::new(60)));
+ let rebalance_listener = RebalanceListener::new(
+ rebalance_rx,
+ Arc::clone(&stream_state),
+ shutdown_handle.clone(),
+ );
+
+ let kafka_streams = KafkaStreams::init(kafka_context, stream_state, shutdown_handle.clone())?;
+
+ let stats = kafka_streams.statistics();
+ registry.register(Box::new(KafkaMetricsCollector::new(stats)?))?;
+
+ let kafka_parseable_sink_connector = KafkaSinkConnector::new(kafka_streams, processor);
+
+ rebalance_listener.start();
+ kafka_parseable_sink_connector.run().await?;
+
+ Ok(())
+}
diff --git a/src/handlers/http/modal/ingest_server.rs b/src/handlers/http/modal/ingest_server.rs
index bafb58080..4cdfb6428 100644
--- a/src/handlers/http/modal/ingest_server.rs
+++ b/src/handlers/http/modal/ingest_server.rs
@@ -31,7 +31,6 @@ use crate::handlers::http::logstream;
use crate::handlers::http::middleware::DisAllowRootUser;
use crate::handlers::http::middleware::RouteExt;
use crate::handlers::http::role;
-use crate::metrics;
use crate::migration;
use crate::migration::metadata_migration::migrate_ingester_metadata;
use crate::rbac::role::Action;
@@ -44,9 +43,10 @@ use crate::sync;
use crate::utils::get_ingestor_id;
use crate::utils::get_url;
use crate::{handlers::http::base_path, option::CONFIG};
+
use actix_web::web;
-use actix_web::web::resource;
use actix_web::Scope;
+use actix_web_prometheus::PrometheusMetrics;
use async_trait::async_trait;
use base64::Engine;
use bytes::Bytes;
@@ -177,8 +177,8 @@ impl ParseableServer for IngestServer {
// parseable can't use local storage for persistence when running a distributed setup
if CONFIG.get_storage_mode_string() == "Local drive" {
return Err(anyhow::Error::msg(
- "This instance of the Parseable server has been configured to run in a distributed setup, it doesn't support local storage.",
- ));
+ "This instance of the Parseable server has been configured to run in a distributed setup, it doesn't support local storage.",
+ ));
}
// check for querier state. Is it there, or was it there in the past
@@ -190,9 +190,12 @@ impl ParseableServer for IngestServer {
}
/// configure the server and start an instance to ingest data
- async fn init(&self, shutdown_rx: oneshot::Receiver<()>) -> anyhow::Result<()> {
- let prometheus = metrics::build_metrics_handler();
- CONFIG.storage().register_store_metrics(&prometheus);
+ async fn init(
+ &self,
+ prometheus: &PrometheusMetrics,
+ shutdown_rx: oneshot::Receiver<()>,
+ ) -> anyhow::Result<()> {
+ CONFIG.storage().register_store_metrics(prometheus);
migration::run_migration(&CONFIG).await?;
@@ -207,7 +210,7 @@ impl ParseableServer for IngestServer {
set_ingestor_metadata().await?;
// Ingestors shouldn't have to deal with OpenId auth flow
- let app = self.start(shutdown_rx, prometheus, None);
+ let app = self.start(shutdown_rx, prometheus.clone(), None);
tokio::pin!(app);
loop {
@@ -237,7 +240,7 @@ impl ParseableServer for IngestServer {
(remote_sync_handler, remote_sync_outbox, remote_sync_inbox) = sync::object_store_sync().await;
}
- };
+ }
}
}
}
@@ -258,21 +261,21 @@ impl IngestServer {
pub fn get_user_role_webscope() -> Scope {
web::scope("/role")
// GET Role List
- .service(resource("").route(web::get().to(role::list).authorize(Action::ListRole)))
+ .service(web::resource("").route(web::get().to(role::list).authorize(Action::ListRole)))
.service(
// PUT and GET Default Role
- resource("/default")
+ web::resource("/default")
.route(web::put().to(role::put_default).authorize(Action::PutRole))
.route(web::get().to(role::get_default).authorize(Action::GetRole)),
)
.service(
// PUT, GET, DELETE Roles
- resource("/{name}")
+ web::resource("/{name}")
.route(web::delete().to(role::delete).authorize(Action::DeleteRole))
.route(web::get().to(role::get).authorize(Action::GetRole)),
)
.service(
- resource("/{name}/sync")
+ web::resource("/{name}/sync")
.route(web::put().to(ingestor_role::put).authorize(Action::PutRole)),
)
}
diff --git a/src/handlers/http/modal/mod.rs b/src/handlers/http/modal/mod.rs
index d36372b2b..f791d16b2 100644
--- a/src/handlers/http/modal/mod.rs
+++ b/src/handlers/http/modal/mod.rs
@@ -68,7 +68,11 @@ pub trait ParseableServer {
async fn load_metadata(&self) -> anyhow::Result>;
/// code that describes starting and setup procedures for each type of server
- async fn init(&self, shutdown_rx: oneshot::Receiver<()>) -> anyhow::Result<()>;
+ async fn init(
+ &self,
+ prometheus: &PrometheusMetrics,
+ shutdown_rx: oneshot::Receiver<()>,
+ ) -> anyhow::Result<()>;
/// configure the server
async fn start(
diff --git a/src/handlers/http/modal/query_server.rs b/src/handlers/http/modal/query_server.rs
index 4c77e226c..83b4206dd 100644
--- a/src/handlers/http/modal/query_server.rs
+++ b/src/handlers/http/modal/query_server.rs
@@ -30,9 +30,10 @@ use crate::rbac::role::Action;
use crate::sync;
use crate::users::dashboards::DASHBOARDS;
use crate::users::filters::FILTERS;
-use crate::{analytics, metrics, migration, storage};
+use crate::{analytics, migration, storage};
use actix_web::web::{resource, ServiceConfig};
use actix_web::{web, Scope};
+use actix_web_prometheus::PrometheusMetrics;
use async_trait::async_trait;
use bytes::Bytes;
use tokio::sync::oneshot;
@@ -89,9 +90,12 @@ impl ParseableServer for QueryServer {
}
/// initialize the server, run migrations as needed and start an instance
- async fn init(&self, shutdown_rx: oneshot::Receiver<()>) -> anyhow::Result<()> {
- let prometheus = metrics::build_metrics_handler();
- CONFIG.storage().register_store_metrics(&prometheus);
+ async fn init(
+ &self,
+ prometheus: &PrometheusMetrics,
+ shutdown_rx: oneshot::Receiver<()>,
+ ) -> anyhow::Result<()> {
+ CONFIG.storage().register_store_metrics(prometheus);
migration::run_migration(&CONFIG).await?;
@@ -135,7 +139,7 @@ impl ParseableServer for QueryServer {
sync::object_store_sync().await;
tokio::spawn(airplane::server());
- let app = self.start(shutdown_rx, prometheus, CONFIG.options.openid());
+ let app = self.start(shutdown_rx, prometheus.clone(), CONFIG.options.openid());
tokio::pin!(app);
loop {
diff --git a/src/handlers/http/modal/server.rs b/src/handlers/http/modal/server.rs
index c7ee3963f..0b66a743a 100644
--- a/src/handlers/http/modal/server.rs
+++ b/src/handlers/http/modal/server.rs
@@ -39,6 +39,7 @@ use actix_web::web;
use actix_web::web::resource;
use actix_web::Resource;
use actix_web::Scope;
+use actix_web_prometheus::PrometheusMetrics;
use actix_web_static_files::ResourceFiles;
use async_trait::async_trait;
use bytes::Bytes;
@@ -99,9 +100,12 @@ impl ParseableServer for Server {
}
// configure the server and start an instance of the single server setup
- async fn init(&self, shutdown_rx: oneshot::Receiver<()>) -> anyhow::Result<()> {
- let prometheus = metrics::build_metrics_handler();
- CONFIG.storage().register_store_metrics(&prometheus);
+ async fn init(
+ &self,
+ prometheus: &PrometheusMetrics,
+ shutdown_rx: oneshot::Receiver<()>,
+ ) -> anyhow::Result<()> {
+ CONFIG.storage().register_store_metrics(prometheus);
migration::run_migration(&CONFIG).await?;
@@ -138,7 +142,7 @@ impl ParseableServer for Server {
tokio::spawn(handlers::livetail::server());
tokio::spawn(handlers::airplane::server());
- let app = self.start(shutdown_rx, prometheus, CONFIG.options.openid());
+ let app = self.start(shutdown_rx, prometheus.clone(), CONFIG.options.openid());
tokio::pin!(app);
diff --git a/src/kafka.rs b/src/kafka.rs
deleted file mode 100644
index 423d227ee..000000000
--- a/src/kafka.rs
+++ /dev/null
@@ -1,255 +0,0 @@
-/*
- * Parseable Server (C) 2022 - 2024 Parseable, Inc.
- *
- * This program is free software: you can redistribute it and/or modify
- * it under the terms of the GNU Affero General Public License as
- * published by the Free Software Foundation, either version 3 of the
- * License, or (at your option) any later version.
- *
- * This program is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
- * GNU Affero General Public License for more details.
- *
- * You should have received a copy of the GNU Affero General Public License
- * along with this program. If not, see .
- *
- */
-
-use arrow_schema::Field;
-use chrono::Utc;
-use futures_util::StreamExt;
-use itertools::Itertools;
-use rdkafka::config::ClientConfig;
-use rdkafka::consumer::stream_consumer::StreamConsumer;
-use rdkafka::consumer::Consumer;
-use rdkafka::error::{KafkaError as NativeKafkaError, RDKafkaError};
-use rdkafka::message::BorrowedMessage;
-use rdkafka::util::Timeout;
-use rdkafka::{Message, TopicPartitionList};
-use serde::{Deserialize, Serialize};
-use std::num::ParseIntError;
-use std::sync::Arc;
-use std::{collections::HashMap, fmt::Debug};
-use tracing::{debug, error, info, warn};
-
-use crate::audit::AuditLogBuilder;
-use crate::event::format::LogSource;
-use crate::option::CONFIG;
-use crate::{
- event::{
- self,
- error::EventError,
- format::{self, EventFormat},
- },
- handlers::http::ingest::{create_stream_if_not_exists, PostError},
- metadata::{error::stream_info::MetadataError, STREAM_INFO},
- storage::StreamType,
-};
-
-#[derive(Debug, Deserialize, Serialize, Clone, Copy)]
-pub enum SslProtocol {
- Plaintext,
- Ssl,
- SaslPlaintext,
- SaslSsl,
-}
-
-#[derive(Debug, thiserror::Error)]
-pub enum KafkaError {
- #[error("Please set env var {0} (To use Kafka integration env vars P_KAFKA_TOPICS, P_KAFKA_HOST, and P_KAFKA_GROUP are mandatory)")]
- NoVarError(&'static str),
-
- #[error("Kafka error {0}")]
- NativeError(#[from] NativeKafkaError),
- #[error("RDKafka error {0}")]
- RDKError(#[from] RDKafkaError),
-
- #[error("Error parsing int {1} for environment variable {0}")]
- ParseIntError(&'static str, ParseIntError),
- #[error("Error parsing duration int {1} for environment variable {0}")]
- ParseDurationError(&'static str, ParseIntError),
-
- #[error("Stream not found: #{0}")]
- StreamNotFound(String),
- #[error("Post error: #{0}")]
- PostError(#[from] PostError),
- #[error("Metadata error: #{0}")]
- MetadataError(#[from] MetadataError),
- #[error("Event error: #{0}")]
- EventError(#[from] EventError),
- #[error("JSON error: #{0}")]
- JsonError(#[from] serde_json::Error),
- #[error("Invalid group offset storage: #{0}")]
- InvalidGroupOffsetStorage(String),
-
- #[error("Invalid SSL protocol: #{0}")]
- InvalidSslProtocolError(String),
- #[error("Invalid unicode for environment variable {0}")]
- EnvNotUnicode(&'static str),
- #[error("")]
- DoNotPrintError,
-}
-
-fn setup_consumer() -> Result<(StreamConsumer, Vec), KafkaError> {
- if let Some(topics) = &CONFIG.options.kafka_topics {
- // topics can be a comma separated list of topics to subscribe to
- let topics = topics.split(',').map(|v| v.to_owned()).collect_vec();
-
- let host = if CONFIG.options.kafka_host.is_some() {
- CONFIG.options.kafka_host.as_ref()
- } else {
- return Err(KafkaError::NoVarError("P_KAKFA_HOST"));
- };
-
- let group = if CONFIG.options.kafka_group.is_some() {
- CONFIG.options.kafka_group.as_ref()
- } else {
- return Err(KafkaError::NoVarError("P_KAKFA_GROUP"));
- };
-
- let mut conf = ClientConfig::new();
- conf.set("bootstrap.servers", host.unwrap());
- conf.set("group.id", group.unwrap());
-
- if let Some(val) = CONFIG.options.kafka_client_id.as_ref() {
- conf.set("client.id", val);
- }
-
- if let Some(ssl_protocol) = CONFIG.options.kafka_security_protocol.as_ref() {
- conf.set("security.protocol", serde_json::to_string(&ssl_protocol)?);
- }
-
- let consumer: StreamConsumer = conf.create()?;
- consumer.subscribe(&topics.iter().map(|v| v.as_str()).collect_vec())?;
-
- if let Some(vals_raw) = CONFIG.options.kafka_partitions.as_ref() {
- // partitions is a comma separated pairs of topic:partitions
- let mut topic_partition_pairs = Vec::new();
- let mut set = true;
- for vals in vals_raw.split(',') {
- let intermediate = vals.split(':').collect_vec();
- if intermediate.len() != 2 {
- warn!(
- "Value for P_KAFKA_PARTITIONS is incorrect! Skipping setting partitions!"
- );
- set = false;
- break;
- }
- topic_partition_pairs.push(intermediate);
- }
-
- if set {
- let mut parts = TopicPartitionList::new();
- for pair in topic_partition_pairs {
- let topic = pair[0];
- match pair[1].parse::() {
- Ok(partition) => {
- parts.add_partition(topic, partition);
- }
- Err(_) => warn!("Skipping setting partition for topic- {topic}"),
- }
- }
- consumer.seek_partitions(parts, Timeout::Never)?;
- }
- }
- Ok((consumer, topics.clone()))
- } else {
- // if the user hasn't even set KAFKA_TOPICS
- // then they probably don't want to use the integration
- // send back the DoNotPrint error
- Err(KafkaError::DoNotPrintError)
- }
-}
-
-fn resolve_schema(stream_name: &str) -> Result>, KafkaError> {
- let hash_map = STREAM_INFO.read().unwrap();
- let raw = hash_map
- .get(stream_name)
- .ok_or_else(|| KafkaError::StreamNotFound(stream_name.to_owned()))?;
- Ok(raw.schema.clone())
-}
-
-async fn ingest_message(msg: BorrowedMessage<'_>) -> Result<(), KafkaError> {
- let Some(payload) = msg.payload() else {
- debug!("No payload received");
- return Ok(());
- };
-
- let msg = msg.detach();
- let stream_name = msg.topic();
-
- // stream should get created only if there is an incoming event, not before that
- create_stream_if_not_exists(stream_name, StreamType::UserDefined, LogSource::default()).await?;
-
- let schema = resolve_schema(stream_name)?;
- let event = format::json::Event {
- data: serde_json::from_slice(payload)?,
- };
-
- let time_partition = STREAM_INFO.get_time_partition(stream_name)?;
- let static_schema_flag = STREAM_INFO.get_static_schema_flag(stream_name)?;
- let schema_version = STREAM_INFO.get_schema_version(stream_name)?;
-
- let (rb, is_first) = event
- .into_recordbatch(
- &schema,
- static_schema_flag,
- time_partition.as_ref(),
- schema_version,
- )
- .map_err(|err| KafkaError::PostError(PostError::CustomError(err.to_string())))?;
-
- event::Event {
- rb,
- stream_name: stream_name.to_string(),
- origin_format: "json",
- origin_size: payload.len() as u64,
- is_first_event: is_first,
- parsed_timestamp: Utc::now().naive_utc(),
- time_partition: None,
- custom_partition_values: HashMap::new(),
- stream_type: StreamType::UserDefined,
- }
- .process()
- .await?;
-
- Ok(())
-}
-
-pub async fn setup_integration() {
- let (consumer, stream_names) = match setup_consumer() {
- Ok(c) => c,
- Err(err) => {
- match err {
- KafkaError::DoNotPrintError => {
- debug!("P_KAFKA_TOPICS not set, skipping kafka integration");
- }
- _ => {
- error!("{err}");
- }
- }
- return;
- }
- };
-
- info!("Setup kafka integration for {stream_names:?}");
- let mut stream = consumer.stream();
-
- while let Ok(curr) = stream.next().await.unwrap() {
- // TODO: maybe we should not constructs an audit log for each kafka message, but do so at the batch level
- let log_builder = AuditLogBuilder::default()
- .with_host(CONFIG.options.kafka_host.as_deref().unwrap_or(""))
- .with_user_agent("Kafka Client")
- .with_protocol("Kafka")
- .with_stream(curr.topic());
-
- let Err(err) = ingest_message(curr).await else {
- log_builder.with_status(200).send().await;
- continue;
- };
- error!("Unable to ingest incoming kafka message- {err}");
-
- log_builder.with_status(500).with_error(err).send().await;
- }
-}
diff --git a/src/lib.rs b/src/lib.rs
index 51b0d45ef..2a20b2bb0 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -23,15 +23,12 @@ pub mod audit;
pub mod banner;
mod catalog;
mod cli;
+#[cfg(feature = "kafka")]
+pub mod connectors;
pub mod correlation;
mod event;
pub mod handlers;
pub mod hottier;
-#[cfg(any(
- all(target_os = "linux", target_arch = "x86_64"),
- all(target_os = "macos", target_arch = "aarch64")
-))]
-pub mod kafka;
mod livetail;
mod metadata;
pub mod metrics;
diff --git a/src/main.rs b/src/main.rs
index 069306951..82776ed43 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -15,29 +15,24 @@
* along with this program. If not, see .
*
*/
-
+#[cfg(feature = "kafka")]
+use parseable::connectors;
use parseable::{
- banner,
+ banner, metrics,
option::{Mode, CONFIG},
rbac, storage, IngestServer, ParseableServer, QueryServer, Server,
};
use tokio::signal::ctrl_c;
use tokio::sync::oneshot;
-use tracing::info;
-use tracing_subscriber::EnvFilter;
-
-#[cfg(any(
- all(target_os = "linux", target_arch = "x86_64"),
- all(target_os = "macos", target_arch = "aarch64")
-))]
-use parseable::kafka;
+use tracing::level_filters::LevelFilter;
+use tracing::{info, warn};
+use tracing_subscriber::layer::SubscriberExt;
+use tracing_subscriber::util::SubscriberInitExt;
+use tracing_subscriber::{fmt, EnvFilter, Registry};
#[actix_web::main]
async fn main() -> anyhow::Result<()> {
- tracing_subscriber::fmt()
- .with_env_filter(EnvFilter::from_default_env())
- .compact()
- .init();
+ init_logger(LevelFilter::DEBUG);
// these are empty ptrs so mem footprint should be minimal
let server: Box = match CONFIG.options.mode {
@@ -55,30 +50,53 @@ async fn main() -> anyhow::Result<()> {
// keep metadata info in mem
metadata.set_global();
- #[cfg(any(
- all(target_os = "linux", target_arch = "x86_64"),
- all(target_os = "macos", target_arch = "aarch64")
- ))]
- // load kafka server
- if CONFIG.options.mode != Mode::Query {
- tokio::task::spawn(kafka::setup_integration());
- }
-
// Spawn a task to trigger graceful shutdown on appropriate signal
let (shutdown_trigger, shutdown_rx) = oneshot::channel::<()>();
tokio::spawn(async move {
block_until_shutdown_signal().await;
// Trigger graceful shutdown
- println!("Received shutdown signal, notifying server to shut down...");
+ warn!("Received shutdown signal, notifying server to shut down...");
shutdown_trigger.send(()).unwrap();
});
- server.init(shutdown_rx).await?;
+ let prometheus = metrics::build_metrics_handler();
+ // Start servers
+ #[cfg(feature = "kafka")]
+ {
+ let parseable_server = server.init(&prometheus, shutdown_rx);
+ let connectors = connectors::init(&prometheus);
+
+ tokio::try_join!(parseable_server, connectors)?;
+ }
+
+ #[cfg(not(feature = "kafka"))]
+ {
+ let parseable_server = server.init(&prometheus, shutdown_rx);
+ parseable_server.await?;
+ }
Ok(())
}
+pub fn init_logger(default_level: LevelFilter) {
+ let filter_layer = EnvFilter::try_from_default_env()
+ .unwrap_or_else(|_| EnvFilter::new(default_level.to_string()));
+
+ let fmt_layer = fmt::layer()
+ .with_thread_names(true)
+ .with_thread_ids(true)
+ .with_line_number(true)
+ .with_timer(tracing_subscriber::fmt::time::UtcTime::rfc_3339())
+ .with_target(true)
+ .compact();
+
+ Registry::default()
+ .with(filter_layer)
+ .with(fmt_layer)
+ .init();
+}
+
#[cfg(windows)]
/// Asynchronously blocks until a shutdown signal is received
pub async fn block_until_shutdown_signal() {
diff --git a/src/metadata.rs b/src/metadata.rs
index 66fae17a3..e98d7e52a 100644
--- a/src/metadata.rs
+++ b/src/metadata.rs
@@ -163,6 +163,21 @@ impl StreamInfo {
Ok(Arc::new(schema))
}
+ #[allow(dead_code)]
+ pub fn schema_raw(
+ &self,
+ stream_name: &str,
+ ) -> Result>, MetadataError> {
+ let map = self.read().expect(LOCK_EXPECT);
+
+ let schema = map
+ .get(stream_name)
+ .ok_or(MetadataError::StreamMetaNotFound(stream_name.to_string()))
+ .map(|metadata| metadata.schema.clone())?;
+
+ Ok(schema)
+ }
+
pub fn set_retention(
&self,
stream_name: &str,
diff --git a/src/option.rs b/src/option.rs
index c6f185397..11aea57a4 100644
--- a/src/option.rs
+++ b/src/option.rs
@@ -17,6 +17,8 @@
*/
use crate::cli::{Cli, Options, StorageOptions, DEFAULT_PASSWORD, DEFAULT_USERNAME};
+#[cfg(feature = "kafka")]
+use crate::connectors::kafka::config::KafkaConfig;
use crate::storage::object_storage::parseable_json_path;
use crate::storage::{ObjectStorageError, ObjectStorageProvider};
use bytes::Bytes;
@@ -37,6 +39,8 @@ pub struct Config {
pub options: Options,
storage: Arc,
pub storage_name: &'static str,
+ #[cfg(feature = "kafka")]
+ pub kafka_config: KafkaConfig,
}
impl Config {
@@ -63,17 +67,23 @@ impl Config {
options: args.options,
storage: Arc::new(args.storage),
storage_name: "drive",
+ #[cfg(feature = "kafka")]
+ kafka_config: args.kafka,
}
}
StorageOptions::S3(args) => Config {
options: args.options,
storage: Arc::new(args.storage),
storage_name: "s3",
+ #[cfg(feature = "kafka")]
+ kafka_config: args.kafka,
},
StorageOptions::Blob(args) => Config {
options: args.options,
storage: Arc::new(args.storage),
storage_name: "blob_store",
+ #[cfg(feature = "kafka")]
+ kafka_config: args.kafka,
},
}
}
@@ -198,12 +208,6 @@ pub mod validation {
use path_clean::PathClean;
- #[cfg(any(
- all(target_os = "linux", target_arch = "x86_64"),
- all(target_os = "macos", target_arch = "aarch64")
- ))]
- use crate::kafka::SslProtocol;
-
use super::{Compression, Mode};
pub fn file_path(s: &str) -> Result {
@@ -248,20 +252,6 @@ pub mod validation {
url::Url::parse(s).map_err(|_| "Invalid URL provided".to_string())
}
- #[cfg(any(
- all(target_os = "linux", target_arch = "x86_64"),
- all(target_os = "macos", target_arch = "aarch64")
- ))]
- pub fn kafka_security_protocol(s: &str) -> Result {
- match s {
- "plaintext" => Ok(SslProtocol::Plaintext),
- "ssl" => Ok(SslProtocol::Ssl),
- "sasl_plaintext" => Ok(SslProtocol::SaslPlaintext),
- "sasl_ssl" => Ok(SslProtocol::SaslSsl),
- _ => Err("Invalid Kafka Security Protocol provided".to_string()),
- }
- }
-
pub fn mode(s: &str) -> Result {
match s {
"query" => Ok(Mode::Query),