feat: add support for cluster annotations (#3628)

Add support for custom metadata in the cluster configuration.

```toml
[cluster.my_favorite_cluster]
endpoint = "127.0.0.1:9003"

[cluster.my_favorite_cluster.metadata.installation]
"fluvio.io/cluster-type" = "local-k8"
```

This will be important for other tasks, such as #3620.

These changes are retrocompatible.

Closes #3627

crates/fluvio/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "fluvio"
-version = "0.21.1"
+version = "0.21.2"
 edition = "2021"
 license = "Apache-2.0"
 authors = ["Fluvio Contributors <[email protected]>"]

crates/fluvio/src/config/cluster.rs

@@ -11,7 +11,7 @@ use super::ConfigFile;
 
 /// Fluvio Cluster Target Configuration
 /// This is part of profile
-#[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize)]
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 #[non_exhaustive]
 pub struct FluvioConfig {
     /// The address to connect to the Fluvio cluster
@@ -29,6 +29,9 @@ pub struct FluvioConfig {
     #[serde(default)]
     pub tls: TlsPolicy,
 
+    /// Cluster custom metadata
+    pub metadata: Option<toml::Value>,
+
     /// This is not part of profile and doesn't persist.
     /// It is purely to override client id when creating ClientConfig
     #[serde(skip)]
@@ -49,6 +52,7 @@ impl FluvioConfig {
             endpoint: addr.into(),
             use_spu_local_address: false,
             tls: TlsPolicy::Disabled,
+            metadata: None,
             client_id: None,
         }
     }

crates/fluvio/src/config/config.rs

@@ -191,7 +191,7 @@ impl ConfigFile {
 pub const LOCAL_PROFILE: &str = "local";
 const CONFIG_VERSION: &str = "2.0";
 
-#[derive(Debug, Default, Eq, PartialEq, Serialize, Deserialize)]
+#[derive(Debug, Default, Serialize, Deserialize)]
 pub struct Config {
     version: String,
     current_profile: Option<String>,
@@ -559,14 +559,6 @@ pub mod test {
         );
     }
 
-    /*
-    #[test]
-    fn test_topic_config() {
-        let conf_file = ConfigFile::load(Some("test-data/profiles/config.toml".to_owned())).expect("parse failed");
-        let config = conf_file.config().resolve_replica_config("test3",0);
-    }
-    */
-
     #[test]
     fn test_local_cluster() {
         let config = Config::new_with_local_cluster("localhost:9003".to_owned());
@@ -575,4 +567,37 @@ pub mod test {
         let cluster = config.current_cluster().expect("cluster should exists");
         assert_eq!(cluster.endpoint, "localhost:9003");
     }
+
+    #[test]
+    fn test_profile_with_metadata() {
+        use std::collections::BTreeMap;
+
+        let config_file = ConfigFile::load(Some("test-data/profiles/config.toml".to_owned()))
+            .expect("could not parse config file");
+        let config = config_file.config();
+
+        let cluster = config
+            .cluster("extra")
+            .expect("could not find `extra` cluster in test file");
+
+        // Table({"key": String("custom field")})
+        let key = BTreeMap::from_iter([("key", "custom field")]);
+        let key = toml::Value::from(key);
+
+        // Table({"example": Table({"key": String("custom field")})})
+        let example = BTreeMap::from_iter([("example", key)]);
+        let example = toml::Value::from(example);
+
+        // Table({"nesting": Table({"example": Table({"key": String("custom field")})})})
+        let nesting = BTreeMap::from_iter([("nesting", example)]);
+        let nesting = toml::Value::from(nesting);
+
+        // Table({"type": String("local")}
+        let cluster_type = BTreeMap::from_iter([("type", "local")]);
+        let cluster_type = toml::Value::from(cluster_type);
+
+        let metadata = BTreeMap::from_iter([("installation", cluster_type), ("deep", nesting)]);
+        let metadata = toml::Value::from(metadata);
+        assert_eq!(cluster.metadata, Some(metadata));
+    }
 }

crates/fluvio/test-data/profiles/config.toml

@@ -40,3 +40,13 @@ isolation = "read_committed"
 # only for topic test=3,replication=2
 [topic.test4.2]
 isolation = "uncommitted"
+
+# cluster metadata
+[cluster.extra]
+endpoint = "127.0.0.1:9003"
+
+[cluster.extra.metadata.deep.nesting.example]
+key = "custom field"
+
+[cluster.extra.metadata.installation]
+type = "local"