feat: log connector name and version on startup (#3356)

Provides support to log connector name and version on startup.

---

## Demo

```bash
2023-06-28T17:34:59.069598Z  INFO json_test_connector: Starting Processing [email protected]
```

> `<pkg_name>@<version>`

```bash
➜  json-test-connector git:(3334-log-connector-version-string-on-connector-startup) ✗ cdk deploy log --name my-json-test-connector
2023-06-28T17:34:59.068577Z  INFO json_test_connector: Using EnvSecretStore
2023-06-28T17:34:59.068703Z  INFO json_test_connector: Reading config file from: /json-test-connector/sample-config.yaml
2023-06-28T17:34:59.069598Z  INFO json_test_connector: Starting Processing [email protected]
2023-06-28T17:34:59.071519Z  INFO fluvio::config::tls: Using verified TLS with inline certificates domain="bitter-dawn-1012b4378e492dd6c3b4f99911f5a371.c.dev.infinyon.cloud"
2023-06-28T17:34:59.075361Z  INFO fluvio::fluvio: Connecting to Fluvio cluster fluvio_crate_version="0.19.2" fluvio_git_hash="7432c94b606b0a3a820a2f499d4d397728e96c47"
2023-06-28T17:35:00.193867Z  INFO connect: fluvio_socket::versioned: connect to socket add=router.dev.infinyon.cloud:9003
2023-06-28T17:35:00.759571Z  INFO connect:connect_with_config: fluvio::config::tls: Using verified TLS with inline certificates domain="bitter-dawn-1012b4378e492dd6c3b4f99911f5a371.c.dev.infinyon.cloud"
2023-06-28T17:35:01.469555Z  INFO connect:connect_with_config:connect: fluvio_socket::versioned: connect to socket add=router.dev.infinyon.cloud:9003
2023-06-28T17:35:03.255467Z  INFO dispatcher_loop{self=MultiplexDisp(8)}: fluvio_socket::multiplexing: multiplexer terminated
2023-06-28T17:35:04.095361Z  INFO fluvio_connector_common::monitoring: using default metric path: /tmp/fluvio-connector.sock
2023-06-28T17:35:04.095688Z  INFO fluvio_connector_common::monitoring: metric file already exists, deleting: /tmp/fluvio-connector.sock
producing a value: {"template":"test"}
2023-06-28T17:35:04.097975Z  INFO fluvio_connector_common::monitoring: monitoring started
2023-06-28T17:35:05.096034Z  INFO run:create_serial_socket_from_leader{leader_id=0}:connect_to_leader{leader=0}:connect: fluvio_socket::versioned: connect to socket add=router.dev.infinyon.cloud:9005
producing a value: {"template":"test"}
producing a value: {"template":"test"}
2023-06-28T17:35:19.098178Z  INFO run: fluvio::producer::partition_producer: partition producer end event received
```

.gitignore

@@ -9,4 +9,7 @@ VERSION
 .env
 .vscode/
 vendor/
-*.tmp
+*.tmp
+
+# Test Connectors Logs
+connector/**/*.log

crates/fluvio-connector-derive/src/generator.rs

@@ -125,7 +125,7 @@ fn init_and_parse_config(config_type_path: &Path) -> TokenStream {
 
         let user_config: #config_type_path = ::fluvio_connector_common::config::from_value(config_value, Some(#config_type_path::__config_name()))?;
 
-        ::fluvio_connector_common::tracing::info!("starting processing");
+        ::fluvio_connector_common::tracing::info!(conn_type=common_config.r#type(), conn_name=common_config.name(), conn_version=common_config.version(), "Starting Processing");
     }
 }
 

crates/fluvio-connector-package/src/config/mod.rs

@@ -324,6 +324,18 @@ impl ConnectorConfig {
     pub fn image(&self) -> String {
         self.meta().image()
     }
+
+    pub fn name(&self) -> String {
+        self.meta().name.clone()
+    }
+
+    pub fn version(&self) -> String {
+        self.meta().version.clone()
+    }
+
+    pub fn r#type(&self) -> String {
+        self.meta().type_.clone()
+    }
 }
 
 #[cfg(test)]
@@ -695,4 +707,31 @@ mod tests {
             println!("{tfile}: {connector_cfg:?}");
         }
     }
+
+    #[test]
+    fn retrieves_name_and_version() {
+        let have = ConnectorConfig::V0_1_0(ConnectorConfigV1 {
+            meta: MetaConfig {
+                name: "my-test-mqtt".to_string(),
+                type_: "mqtt-source".to_string(),
+                topic: "my-mqtt".to_string(),
+                version: "0.1.0".to_string(),
+                producer: Some(ProducerParameters {
+                    linger: None,
+                    compression: None,
+                    batch_size: Some(ByteSize::b(1600)),
+                }),
+                consumer: Some(ConsumerParameters {
+                    max_bytes: Some(ByteSize::b(1400)),
+                    partition: None,
+                }),
+                secrets: None,
+            },
+            transforms: None,
+        });
+
+        assert_eq!(have.name(), "my-test-mqtt");
+        assert_eq!(have.version(), "0.1.0");
+        assert_eq!(have.r#type(), "mqtt-source");
+    }
 }