Added initial kafka support.

Author: s3rius

Cargo.lock

@@ -51,6 +51,7 @@ chrono = { version = "^0.4.38", features = ["serde"] }
 fern = { version = "^0.7.0", features = ["colored"] }
 reqwest = { version = "^0.12.9", features = ["json"] }
 derive_more = { version = "^1.0.0", features = ["display", "from", "from_str"], default-features = false }
+rdkafka = "0.37.0"
 
 [target.'cfg(not(target_env = "msvc"))'.dependencies.tikv-jemallocator]
 version = "0.6"

Cargo.toml

@@ -42,4 +42,28 @@ services:
       MINIO_DEFAULT_BUCKETS: "rustus"
       MINIO_ROOT_USER: "rustus-test"
       MINIO_ROOT_PASSWORD: "rustus-test"
-
+  kafka:
+    image: bitnami/kafka:3.9-debian-12
+    healthcheck:
+      test:
+        - CMD
+        - kafka-topics.sh
+        - --list
+        - --bootstrap-server
+        - localhost:9092
+      interval: 1s
+      timeout: 3s
+      retries: 30
+    ports:
+      - 9094:9094
+    environment:
+      KAFKA_CFG_NODE_ID: "0"
+      KAFKA_KRAFT_CLUSTER_ID: "0"
+      KAFKA_CFG_PROCESS_ROLES: "controller,broker"
+      KAFKA_CFG_LISTENERS: "PLAINTEXT://:9092,CONTROLLER://:9093,EXTERNAL://:9094"
+      KAFKA_CFG_ADVERTISED_LISTENERS: "PLAINTEXT://kafka:9092,EXTERNAL://localhost:9094"
+      KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: "CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT,PLAINTEXT:PLAINTEXT"
+      KAFKA_CFG_CONTROLLER_QUORUM_VOTERS: "0@kafka:9093"
+      KAFKA_CFG_CONTROLLER_LISTENER_NAMES: "CONTROLLER"
+      KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE: "true"
+      KAFKA_CFG_OFFSETS_TOPIC_REPLICATION_FACTOR: "1"

docker-compose.yaml

@@ -5,7 +5,7 @@ use clap::Parser;
 use crate::{
     data_storage::AvailableDataStorages,
     info_storage::AvailableInfoStorages,
-    notifiers::{Format, Hook},
+    notifiers::{impls::kafka_notifier::ExtraKafkaOptions, Format, Hook},
     protocol::extensions::Extensions,
 };
 
@@ -284,6 +284,91 @@ pub struct AMQPHooksOptions {
     pub auto_delete: bool,
 }
 
+#[derive(Parser, Debug, Clone)]
+pub struct KafkaHookOptions {
+    /// Kafka urls.
+    /// List of brokers to connect to in the format `host:port`.
+    /// If you have multiple brokers, separate them with commas.
+    /// Corresponds to `bootstrap.servers` in Kafka configuration.
+    #[arg(
+        name = "hooks-kafka-urls",
+        long,
+        env = "RUSTUS_HOOKS_KAFKA_URLS",
+        use_value_delimiter = true
+    )]
+    pub urls: String,
+    /// Kafka producer client.id.
+    #[arg(
+        name = "hooks-kafka-client-id",
+        long,
+        env = "RUSTUS_HOOKS_KAFKA_CLIENT_ID"
+    )]
+    pub client_id: Option<String>,
+    /// Kafka topic. If specified, all events will be sent to this topic.
+    #[arg(name = "hooks-kafka-topic", long, env = "RUSTUS_HOOKS_KAFKA_TOPIC")]
+    pub topic: Option<String>,
+    /// Kafka topic prefix. In case if specifeid, prefix will be added to all topics
+    /// and all events will be sent to different topics.
+    #[arg(name = "hooks-kafka-prefix", long, env = "RUSTUS_HOOKS_KAFKA_PREFIX")]
+    pub prefix: Option<String>,
+    /// Kafka required acks.
+    /// This parameter is used to configure how many replicas
+    /// must acknowledge the message.
+    ///
+    /// Corresponds to `request.required.acks` in Kafka configuration.
+    /// Possible values are:
+    /// * -1 - all replicas must acknowledge the message;
+    /// * 0 - no replicas must acknowledge the message;
+    /// * ...1000 - number of replicas that must acknowledge the message.
+    #[arg(
+        name = "hooks-kafka-required-acks",
+        long,
+        env = "RUSTUS_HOOKS_KAFKA_REQUIRED_ACKS"
+    )]
+    pub required_acks: Option<String>,
+
+    /// Compression codec.
+    /// This parameter is used to compress messages before sending them to Kafka.
+    /// Possible values are:
+    /// * none - no compression;
+    /// * gzip - gzip compression;
+    /// * snappy - snappy compression.
+    /// * lz4 - lz4 compression.
+    /// * zstd - zstd compression.
+    /// Corresponds to `compression.codec` in Kafka configuration.
+    #[arg(
+        name = "hooks-kafka-compression",
+        long,
+        env = "RUSTUS_HOOKS_KAFKA_COMPRESSION"
+    )]
+    pub compression: Option<String>,
+
+    /// Kafka idle timeout in seconds.
+    /// After this amount of time in seconds, the connection will be dropped.
+    /// Corresponds to `connections.max.idle.ms` in Kafka configuration.
+    #[arg(
+        name = "hooks-kafka-idle-timeout",
+        long,
+        env = "RUSTUS_HOOKS_KAFKA_IDLE_TIMEOUT"
+    )]
+    pub idle_timeout: Option<u64>,
+
+    /// Kafka send timeout in seconds.
+    /// After this amount of time in seconds, the message will be dropped.
+    #[arg(
+        name = "hooks-kafka-send-timeout",
+        long,
+        env = "RUSTUS_HOOKS_KAFKA_SEND_TIMEOUT"
+    )]
+    pub send_timeout: Option<u64>,
+
+    /// Extra options for Kafka.
+    /// This parameter is used to pass additional options to Kafka.
+    /// All options must be in the format `key=value`, separated by semicolon.
+    /// Example: `key1=value1;key2=value2`.
+    pub extra_kafka_opts: Option<ExtraKafkaOptions>,
+}
+
 #[derive(Parser, Debug, Clone)]
 #[allow(clippy::struct_excessive_bools)]
 pub struct NotificationsOptions {
@@ -336,6 +421,9 @@ pub struct NotificationsOptions {
 
     #[command(flatten)]
     pub amqp_hook_opts: AMQPHooksOptions,
+
+    #[command(flatten)]
+    pub kafka_hook_opts: KafkaHookOptions,
 }
 
 #[derive(Debug, Parser, Clone)]

src/config.rs

@@ -1,6 +1,8 @@
 use std::{fs::File, io::Read, path::PathBuf};
 
 use base::DataStorage;
+use std::str::FromStr;
+use strum::IntoEnumIterator;
 
 use crate::{config::RustusConf, file_info::FileInfo, from_str};
 

src/data_storage/mod.rs

@@ -45,6 +45,8 @@ pub enum RustusError {
     HookError(String),
     #[error("Unable to configure logging: {0}")]
     LogConfigError(#[from] log::SetLoggerError),
+    #[error("Kafka extra options error: {0}")]
+    KafkaExtraOptionsError(String),
     #[error("AMQP error: {0}")]
     AMQPError(#[from] lapin::Error),
     #[error("AMQP pooling error: {0}")]
@@ -73,6 +75,8 @@ pub enum RustusError {
     ParseIntError(#[from] std::num::ParseIntError),
     #[error("Can't convert int: {0}")]
     TryFromIntError(#[from] std::num::TryFromIntError),
+    #[error("Kafka error: {0}")]
+    KafkaError(#[from] rdkafka::error::KafkaError),
 }
 
 /// This conversion allows us to use `RustusError` in the `main` function.

src/errors.rs

@@ -2,6 +2,8 @@ pub mod base;
 pub mod impls;
 
 use derive_more::{Display, From};
+use std::str::FromStr;
+use strum::IntoEnumIterator;
 
 use crate::{errors::RustusResult, from_str, RustusConf};
 

src/info_storage/mod.rs

@@ -1,4 +1,4 @@
-use crate::errors::RustusResult;
+use crate::{errors::RustusResult, file_info::FileInfo};
 use actix_web::http::header::HeaderMap;
 
 use crate::notifiers::hooks::Hook;
@@ -9,6 +9,7 @@ pub trait Notifier {
         &self,
         message: String,
         hook: Hook,
+        file_info: &FileInfo,
         headers_map: &HeaderMap,
     ) -> RustusResult<()>;
 }

src/notifiers/base.rs

@@ -2,6 +2,9 @@ use crate::from_str;
 use derive_more::{Display, From};
 use strum::EnumIter;
 
+use std::str::FromStr;
+use strum::IntoEnumIterator;
+
 /// Hooks for notifications.
 #[derive(Copy, Clone, Debug, Display, From, EnumIter, Eq, PartialEq)]
 pub enum Hook {