feat: add deduplicate parameter to create topic using cli

crates/fluvio-cli/src/client/hub/mod.rs

@@ -2,6 +2,7 @@ pub use cmd::HubCmd;
 
 mod connector;
 mod smartmodule;
+pub use smartmodule::{download_local, download_cluster};
 
 mod cmd {
     use std::sync::Arc;

crates/fluvio-cli/src/client/hub/smartmodule/download.rs

@@ -5,11 +5,11 @@ use std::path::PathBuf;
 
 use async_trait::async_trait;
 use clap::Parser;
+use fluvio::FluvioAdmin;
 use tracing::info;
 use anyhow::Result;
 
 use fluvio::Fluvio;
-use fluvio::FluvioConfig;
 use fluvio::metadata::smartmodule::SmartModuleSpec;
 use fluvio_controlplane_metadata::smartmodule::{SmartModuleMetadata, SmartModuleWasm};
 use fluvio_extension_common::Terminal;
@@ -50,10 +50,13 @@ impl ClientCmd for SmartModuleDownloadHubOpts {
         _out: Arc<O>,
         _fluvio: &Fluvio,
     ) -> Result<()> {
+        let config = self.target.load()?;
+        println!("trying connection to fluvio {}", config.endpoint);
+        let fluvio = Fluvio::connect_with_config(&config).await?;
+        let admin = fluvio.admin().await;
         if self.ipkg {
             // pkgname is a package file
-            let fluvio_config = self.target.load()?;
-            download_cluster(fluvio_config, &self.pkgname).await?;
+            download_cluster(&admin, &self.pkgname).await?;
             return Ok(());
         }
         let access = get_hub_access(&self.remote)?;
@@ -63,15 +66,14 @@ impl ClientCmd for SmartModuleDownloadHubOpts {
             return Ok(());
         }
 
-        let fluvio_config = self.target.load()?;
-        download_cluster(fluvio_config, &pkgfile).await?;
+        download_cluster(&admin, &pkgfile).await?;
         Ok(())
     }
 }
 
 /// download smartmodule from hub to local fs
 /// returns path of downloaded of package
-async fn download_local(
+pub async fn download_local(
     pkgname: &str,
     access: &HubAccess,
     output: Option<PathBuf>,
@@ -105,7 +107,7 @@ async fn download_local(
 }
 
 // download smartmodule from pkg to cluster
-async fn download_cluster(config: FluvioConfig, pkgfile: &str) -> Result<()> {
+pub async fn download_cluster(admin: &FluvioAdmin, pkgfile: &str) -> Result<()> {
     println!("... checking package");
     let pm = hubutil::package_get_meta(pkgfile)
         .map_err(|_| CliError::PackageError(format!("accessing metadata in {pkgfile}")))?;
@@ -143,10 +145,6 @@ async fn download_cluster(config: FluvioConfig, pkgfile: &str) -> Result<()> {
         ..Default::default()
     };
 
-    println!("trying connection to fluvio {}", config.endpoint);
-    let fluvio = Fluvio::connect_with_config(&config).await?;
-
-    let admin = fluvio.admin().await;
     admin.create(sm_id, false, spec).await?;
     println!("... cluster smartmodule install complete");
     std::fs::remove_file(pkgfile)

crates/fluvio-cli/src/client/hub/smartmodule/mod.rs

@@ -1,7 +1,7 @@
 mod list;
 pub use list::SmartModuleHubListOpts;
 mod download;
-pub use download::SmartModuleDownloadHubOpts;
+pub use download::{SmartModuleDownloadHubOpts, download_local, download_cluster};
 
 use std::sync::Arc;
 use std::fmt::Debug;

crates/fluvio-cli/src/client/topic/create.rs

@@ -6,6 +6,7 @@
 use std::path::PathBuf;
 use std::time::Duration;
 
+use fluvio_sc_schema::smartmodule::SmartModuleSpec;
 use tracing::debug;
 use clap::Parser;
 use humantime::parse_duration;
@@ -24,12 +25,22 @@ use fluvio_sc_schema::shared::validate_resource_name;
 use fluvio_sc_schema::mirror::MirrorSpec;
 use fluvio_sc_schema::topic::HomeMirrorConfig;
 use fluvio_sc_schema::topic::MirrorConfig;
+use fluvio_sc_schema::topic::Bounds;
+use fluvio_sc_schema::topic::Deduplication;
+use fluvio_sc_schema::topic::Filter;
+use fluvio_sc_schema::topic::Transform;
+use fluvio_hub_util as hubutil;
+use hubutil::cmd::get_hub_access;
 
 use fluvio::Fluvio;
 use fluvio::FluvioAdmin;
 use fluvio::metadata::topic::TopicSpec;
+use crate::client::hub::download_cluster;
+use crate::client::hub::download_local;
 use crate::CliError;
 
+const DEFAULT_DEDUP_FILTER: &str = "fluvio/[email protected]";
+
 #[derive(Debug, Parser)]
 pub struct CreateTopicOpt {
     /// The name of the Topic to create
@@ -229,6 +240,26 @@ impl CreateTopicOpt {
             topic_spec.set_compression_type(compression_type);
         }
 
+        if self.setting.dedup {
+            let sm = admin
+                .list::<SmartModuleSpec, _>(vec![DEFAULT_DEDUP_FILTER.to_string()])
+                .await?
+                .into_iter()
+                .next();
+
+            if sm.is_none() {
+                println!("deduplication filter not found, downloading");
+                let access = get_hub_access(&None)?;
+                let pkgname = DEFAULT_DEDUP_FILTER;
+                let pkgfile = download_local(pkgname, &access, None).await?;
+                download_cluster(admin, &pkgfile).await?;
+            }
+
+            let deduplication =
+                create_deduplication(self.setting.dedup_count, Some(self.setting.dedup_age));
+            topic_spec.set_deduplication(Some(deduplication));
+        }
+
         topic_spec.set_system(self.setting.system);
 
         if self.setting.segment_size.is_some() || self.setting.max_partition_size.is_some() {
@@ -258,6 +289,21 @@ fn validate(name: &str, _spec: &TopicSpec) -> Result<()> {
     Ok(())
 }
 
+fn create_deduplication(dedup_count: u64, dedup_age: Option<Duration>) -> Deduplication {
+    Deduplication {
+        bounds: Bounds {
+            count: dedup_count,
+            age: dedup_age,
+        },
+        filter: Filter {
+            transform: Transform {
+                uses: DEFAULT_DEDUP_FILTER.to_string(),
+                with: Default::default(),
+            },
+        },
+    }
+}
+
 #[derive(Debug, Parser)]
 #[group(id = "config-arg")]
 pub struct TopicConfigOpt {
@@ -280,6 +326,19 @@ pub struct TopicConfigOpt {
     #[arg(long, value_name = "bytes")]
     max_partition_size: Option<bytesize::ByteSize>,
 
+    /// Deduplicate records in the topic
+    #[arg(long)]
+    dedup: bool,
+
+    /// Number of records to keep in deduplication filter
+    #[arg(long, value_name = "integer", requires = "dedup", default_value = "5")]
+    dedup_count: u64,
+
+    /// Age of records to keep in deduplication filter
+    /// Ex: '1h', '2d 10s', '7 days' (default)
+    #[arg(long, value_name = "time", value_parser=parse_duration, requires = "dedup", default_value = "5s")]
+    dedup_age: Duration,
+
     /// Flag to create a system topic
     /// System topics are for internal operations
     #[arg(long, short = 's', hide = true)]

crates/fluvio-controlplane-metadata/Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "fluvio-controlplane-metadata"
 edition = "2021"
-version = "0.30.1"
+version = "0.30.2"
 authors = ["Fluvio Contributors <[email protected]>"]
 description = "Metadata definition for Fluvio control plane"
 repository = "https://github.com/infinyon/fluvio"

crates/fluvio-controlplane-metadata/src/topic/config.rs

@@ -236,7 +236,7 @@ deduplication:
     age: 1m
   filter:
     transform:
-      uses: infinyon/[email protected]
+      uses: fluvio/dedup-bloom[email protected]
 "#;
 
         //when
@@ -373,7 +373,7 @@ compression:
             },
             filter: Filter {
                 transform: Transform {
-                    uses: "infinyon/[email protected]".to_string(),
+                    uses: "fluvio/dedup-bloom[email protected]".to_string(),
                     with: Default::default(),
                 },
             },

crates/fluvio-sc/src/services/public_api/topic/create.rs

@@ -130,7 +130,7 @@ async fn validate_topic_request<C: MetadataItem>(
                 sm_name.to_string(),
                 ErrorCode::DeduplicationSmartModuleNotLoaded,
                 Some(format!(
-                    "{}\nHint: try `fluvio hub download {sm_name}` and repeat this operation",
+                    "{}\nHint: try `fluvio hub sm download {sm_name}` and repeat this operation",
                     ErrorCode::DeduplicationSmartModuleNotLoaded
                 )),
             );

tests/cli/fluvio_smoke_tests/topic-basic.bats

@@ -23,7 +23,7 @@ setup_file() {
     TOPIC_NAME_SYSTEM=$(random_string)
     export TOPIC_NAME_SYSTEM
 
-    DEDUP_FILTER_NAME="dedup-filter"
+    DEDUP_FILTER_NAME="dedup-bloom-filter"
     export DEDUP_FILTER_NAME
 
     cat <<EOF >$TOPIC_CONFIG_PATH