Indexes 9 - Adds messaging on package events to kafka

.github/workflows/coverage.yml

@@ -39,7 +39,7 @@ jobs:
       - name: Install prerequisites
         run: |
           apt-get update
-          apt-get install --yes build-essential libcap2-bin sudo cmake tcsh rsync protobuf-compiler fuse libfuse-dev curl unzip pkg-config libssl-dev git
+          apt-get install --yes build-essential libcap2-bin sudo cmake tcsh rsync protobuf-compiler fuse libfuse-dev curl unzip pkg-config libssl-dev git libcurl4-openssl-dev
           # spfs-fuse requires this option enabled
           echo user_allow_other >> /etc/fuse.conf
           FB_REL=https://github.com/google/flatbuffers/releases/

.github/workflows/rust.yml

@@ -122,7 +122,7 @@ jobs:
       - name: Prepare Container
         run: |
           apt-get update
-          apt-get install -y libcap2-bin sudo cmake tcsh rsync protobuf-compiler fuse libfuse-dev curl unzip pkg-config libssl-dev git
+          apt-get install -y build-essential libcap2-bin sudo cmake tcsh rsync protobuf-compiler fuse libfuse-dev curl unzip pkg-config libssl-dev git libcurl4-openssl-dev
           # spfs-fuse requires this option enabled
           echo user_allow_other >> /etc/fuse.conf
           FB_REL=https://github.com/google/flatbuffers/releases/
@@ -234,7 +234,7 @@ jobs:
       - name: Prepare Container
         run: |
           apt-get update
-          apt-get install -y build-essential libcap2-bin sudo cmake tcsh rsync protobuf-compiler fuse libfuse-dev curl unzip pkg-config libssl-dev git
+          apt-get install -y build-essential libcap2-bin sudo cmake tcsh rsync protobuf-compiler fuse libfuse-dev curl unzip pkg-config libssl-dev git libcurl4-openssl-dev
           # spfs-fuse requires this option enabled
           echo user_allow_other >> /etc/fuse.conf
           FB_REL=https://github.com/google/flatbuffers/releases/
@@ -244,7 +244,7 @@ jobs:
       - name: Setup cross-compilation
         if: matrix.platform == 'windows'
         run: |
-          apt-get install -y gcc-mingw-w64-x86-64
+          apt-get install -y gcc-mingw-w64-x86-64 g++-mingw-w64-x86-64-win32
           rustup target add ${{ matrix.target }}
       - name: Configure cargo
         run: |

Cargo.toml

@@ -95,6 +95,7 @@ progress_bar_derive_macro = { path = "crates/progress_bar_derive_macro" }
 proptest = "1.0.0"
 prost = "0.13"
 rand = "0.8.5"
+rdkafka = { version = "0.39.0", features = ["cmake-build"] }
 regex = "1.6"
 relative-path = "1.3"
 resolvo = "0.9.1"

crates/spk-config/src/config.rs

@@ -196,6 +196,30 @@ pub struct HostOptions {
     pub distro_rules: HashMap<String, DistroRule>,
 }
 
+/// Helper for testing default kafka timeout ms when not specified in
+/// config
+fn default_kafka_timeout_ms() -> u64 {
+    5000
+}
+
+/// Configuration for using Kafka as a message channel.
+#[derive(Clone, Debug, Deserialize, Serialize)]
+pub struct KafkaChannel {
+    /// List of brokers to connect to, specified as "hostname:port"
+    pub brokers: Vec<String>,
+    /// Topic name to push package updates to
+    pub package_updates_topic_name: Option<String>,
+    /// Message timeout in milliseconds, defaults to 5000 ms (5 seconds)
+    #[serde(default = "default_kafka_timeout_ms")]
+    pub timeout_ms: u64,
+}
+
+/// Types of message channels.
+#[derive(Clone, Debug, Deserialize, Serialize)]
+pub enum MessageChannel {
+    Kafka(KafkaChannel),
+}
+
 /// Configuration values for spk.
 #[derive(Clone, Debug, Default, Deserialize, Serialize)]
 #[serde(default)]
@@ -210,6 +234,7 @@ pub struct Config {
     pub metadata: Metadata,
     pub cli: Cli,
     pub host_options: HostOptions,
+    pub messaging: Vec<MessageChannel>,
 }
 
 impl Config {

crates/spk-storage/Cargo.toml

@@ -44,6 +44,7 @@ nom = { workspace = true }
 nom-supreme = { workspace = true }
 once_cell = { workspace = true }
 paste = { workspace = true }
+rdkafka = { workspace = true }
 regex = { workspace = true }
 relative-path = { workspace = true }
 ring = { workspace = true }
@@ -57,6 +58,7 @@ spfs = { workspace = true }
 spk-config = { workspace = true }
 spk-proto = { workspace = true }
 spk-schema = { workspace = true }
+strum = { workspace = true }
 sys-info = "0.9.0"
 tar = "0.4.45"
 tempfile = { workspace = true }

crates/spk-storage/src/storage/messaging/kafka.rs

@@ -0,0 +1,74 @@
+// Copyright (c) Contributors to the SPK project.
+// SPDX-License-Identifier: Apache-2.0
+// https://github.com/spkenv/spk
+
+use std::collections::HashMap;
+use std::time::Duration;
+
+use once_cell::sync::Lazy;
+use rdkafka::ClientConfig;
+use rdkafka::error::KafkaError;
+use rdkafka::producer::{FutureProducer, FutureRecord};
+use serde_json::json;
+use spk_config::KafkaChannel;
+use spk_schema::BuildIdent;
+use tokio::sync::Mutex;
+
+use crate::storage::messaging::PackageEvent;
+use crate::{Error, Result};
+
+type ProducersByBrokers = HashMap<Vec<String>, std::result::Result<FutureProducer, KafkaError>>;
+
+static KAFKA_PRODUCERS: Lazy<Mutex<ProducersByBrokers>> = Lazy::new(|| Mutex::new(HashMap::new()));
+
+pub(crate) async fn announce_package_event(
+    kafka_channel: &KafkaChannel,
+    event: PackageEvent,
+    to: &url::Url,
+    ident: &BuildIdent,
+) -> Result<()> {
+    let Some(topic_name) = kafka_channel.package_updates_topic_name.as_ref() else {
+        return Ok(());
+    };
+
+    let mut kafka_producers = KAFKA_PRODUCERS.lock().await;
+
+    let producer = kafka_producers
+        .entry(kafka_channel.brokers.clone())
+        .or_insert_with(|| {
+            ClientConfig::new()
+                // XXX: probably should expose all configuration parameters in
+                // spk config
+                .set("bootstrap.servers", kafka_channel.brokers.join(","))
+                .set("message.timeout.ms", kafka_channel.timeout_ms.to_string())
+                .create()
+        })
+        .as_ref()
+        .map_err(|err| {
+            Error::String(format!(
+                "failed to create kafka producer from config: {err}"
+            ))
+        })?;
+
+    producer
+        .send(
+            FutureRecord::to(topic_name)
+                // Using the package name as the key for purposes of sending
+                // all messages about the same package to the same topic
+                // partition.
+                .key(ident.name().as_str())
+                .payload(
+                    &json!({
+                        "event": event.to_string(),
+                        "repo": to.to_string(),
+                        "package": ident.to_string(),
+                    })
+                    .to_string(),
+                ),
+            Duration::from_secs(0),
+        )
+        .await
+        .map_err(|err| Error::String(format!("failed to send kafka message: {err:?}")))?;
+
+    Ok(())
+}

crates/spk-storage/src/storage/messaging/mod.rs

@@ -0,0 +1,38 @@
+// Copyright (c) Contributors to the SPK project.
+// SPDX-License-Identifier: Apache-2.0
+// https://github.com/spkenv/spk
+
+mod kafka;
+
+use spk_config::MessageChannel;
+use spk_schema::BuildIdent;
+
+use crate::Result;
+
+/// Types of events than can be reported about a package
+#[derive(Copy, Clone, Debug, strum::Display)]
+pub(crate) enum PackageEvent {
+    #[strum(to_string = "package published")]
+    Published,
+    #[strum(to_string = "package modified")]
+    Modified,
+    #[strum(to_string = "package removed")]
+    Removed,
+}
+
+pub(crate) async fn announce_package_event(
+    event: PackageEvent,
+    to: &url::Url,
+    ident: &BuildIdent,
+) -> Result<()> {
+    let config = spk_config::get_config()?;
+    for channel in &config.messaging {
+        match channel {
+            MessageChannel::Kafka(kafka_channel) => {
+                kafka::announce_package_event(kafka_channel, event, to, ident).await?
+            }
+        }
+    }
+
+    Ok(())
+}

crates/spk-storage/src/storage/mod.rs

@@ -7,6 +7,7 @@ mod flatbuffer_index;
 mod handle;
 mod indexed;
 mod mem;
+mod messaging;
 mod repository;
 mod repository_index;
 mod runtime;
@@ -17,6 +18,7 @@ pub use flatbuffer_index::FlatBufferRepoIndex;
 pub use handle::RepositoryHandle;
 pub use indexed::IndexedRepository;
 pub use mem::MemRepository;
+pub(crate) use messaging::announce_package_event;
 pub use repository::{CachePolicy, Repository, Storage};
 pub use repository_index::{RepoIndex, RepositoryIndex, RepositoryIndexMut};
 pub use runtime::{RuntimeRepository, find_path_providers, pretty_print_filepath};