Mitshah2406 commented on code in PR #4617:
URL: https://github.com/apache/iggy/pull/4617#discussion_r4491268506
##########
core/integration/tests/cli/general/test_help_command.rs:
##########
@@ -245,6 +246,22 @@ Options:
{CLAP_INDENT}
[possible values: bash, elvish, fish, powershell, zsh]
+ --connectors-api-url <CONNECTORS_API_URL>
+ Connectors runtime HTTP API base URL
+{CLAP_INDENT}
+ Used by the connector commands. Falls back to the active context,
+ defaulting to http://localhost:8081.
+
+ --connectors-api-key <CONNECTORS_API_KEY>
+ Connectors runtime HTTP API key
+{CLAP_INDENT}
+ Sent as the api-key header. Falls back to IGGY_CONNECTORS_API_KEY,
+ then to the active context. When the runtime answers 401 and stdin
Review Comment:
Fixed in c62fec01b. The expected help text in the test didn't match the real
help text. They match now.
##########
core/cli/src/args/connector.rs:
##########
@@ -0,0 +1,442 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::path::PathBuf;
+
+use crate::args::common::ListMode;
+use clap::{Args, Subcommand, ValueEnum};
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, ValueEnum)]
+pub(crate) enum ConnectorKind {
+ Sink,
+ Source,
+}
+
+/// Same rule as the runtime, so a key cannot carry `/`, `?`, `#` or `..`
+/// into the request path.
+fn parse_connector_key(key: &str) -> Result<String, String> {
+ let mut chars = key.chars();
+ let valid = chars.next().is_some_and(|c| c.is_ascii_alphanumeric())
+ && chars.all(|c| c.is_ascii_alphanumeric() || matches!(c, '.' | '_' |
'-'));
+ if valid {
+ Ok(key.to_string())
+ } else {
+ Err(
+ "must start with a letter or digit and contain only letters,
digits, '.', '_' and '-'"
+ .to_string(),
+ )
+ }
+}
+
+#[derive(Debug, Clone, Subcommand)]
+pub(crate) enum ConnectorAction {
+ /// List all sinks or sources
+ ///
+ /// Examples:
+ /// iggy connector list sink
+ /// iggy connector list source --list-mode list
+ #[clap(verbatim_doc_comment, visible_alias = "l")]
+ List(ConnectorListArgs),
+ /// Get details of a single sink or source
+ ///
+ /// Examples:
+ /// iggy connector get sink my-sink
+ #[clap(verbatim_doc_comment, visible_alias = "g")]
+ Get(ConnectorGetArgs),
+ /// List the transforms of a sink or source
+ ///
+ /// Examples:
+ /// iggy connector transforms sink my-sink
+ #[clap(verbatim_doc_comment, visible_alias = "t")]
+ Transforms(ConnectorGetArgs),
+ /// Restart a sink or source
+ ///
+ /// On the local provider restart loads the newest stored configuration
+ /// version, which may differ from the active one.
+ #[clap(verbatim_doc_comment, visible_alias = "r")]
+ Restart(ConnectorGetArgs),
+ /// Configuration version operations
+ #[command(subcommand, visible_alias = "c")]
+ Config(ConnectorConfigAction),
+ /// Get connectors runtime statistics
+ ///
+ /// Examples:
+ /// iggy connector stats
+ #[clap(verbatim_doc_comment, visible_alias = "s")]
+ Stats,
+}
+
+#[derive(Debug, Clone, Args)]
+pub(crate) struct ConnectorListArgs {
+ /// Connector kind
+ pub(crate) kind: ConnectorKind,
+ /// List mode (table or list)
+ #[clap(short, long, value_enum, default_value_t = ListMode::Table)]
+ pub(crate) list_mode: ListMode,
+}
+
+#[derive(Debug, Clone, Args)]
+pub(crate) struct ConnectorGetArgs {
+ /// Connector kind
+ pub(crate) kind: ConnectorKind,
+ /// Connector key
+ #[clap(value_parser = parse_connector_key)]
+ pub(crate) key: String,
+}
+
+#[derive(Debug, Clone, Subcommand)]
+pub(crate) enum ConnectorConfigAction {
+ /// List all configuration versions of a sink or source
+ ///
+ /// The Running column marks the version the connector runs now, which
+ /// differs from the activated one until a restart. The responses
Review Comment:
Fixed in c62fec01b. A restart loads the newest config version, not the
activated one. The help now says the activated version only applies when the
runtime process starts.
##########
core/cli/src/commands/binary_connectors/stats.rs:
##########
@@ -0,0 +1,133 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use super::client::ConnectorEndpoint;
+use super::{bool_field, cell, str_field};
+use crate::commands::cli_command::{CliCommand, PRINT_TARGET};
+use anyhow::Context;
+use async_trait::async_trait;
+use comfy_table::Table;
+use iggy_common::Client;
+use tracing::{Level, event};
+
+pub struct GetConnectorStatsCmd {
+ endpoint: ConnectorEndpoint,
+}
+
+impl GetConnectorStatsCmd {
+ pub fn new(endpoint: ConnectorEndpoint) -> Self {
+ Self { endpoint }
+ }
+}
+
+#[async_trait]
+impl CliCommand for GetConnectorStatsCmd {
+ fn explain(&self) -> String {
+ "get connectors runtime statistics".to_string()
+ }
+
+ fn login_required(&self) -> bool {
+ false
+ }
+
+ fn connection_required(&self) -> bool {
+ false
+ }
+
+ async fn execute_cmd(&mut self, _client: &dyn Client) ->
anyhow::Result<(), anyhow::Error> {
+ let mut api = self.endpoint.connect()?;
+ let stats = api
+ .get_value("/stats")
+ .await
+ .with_context(|| "problem getting connectors runtime statistics")?;
+
+ let mut table = Table::new();
+ table.set_header(vec!["Property", "Value"]);
+ for key in [
+ "connectors_runtime_version",
+ "process_id",
+ "cpu_usage",
+ "total_cpu_usage",
+ "memory_usage",
+ "total_memory",
+ "available_memory",
+ "run_time",
+ "start_time",
+ "sources_total",
+ "sources_running",
+ "sinks_total",
+ "sinks_running",
+ ] {
+ if let Some(value) = stats.get(key) {
+ table.add_row(vec![key.to_string(), cell(value)]);
+ }
+ }
+ event!(target: PRINT_TARGET, Level::INFO, "{table}");
+
+ let connectors = stats
Review Comment:
Fixed in c62fec01b. `stats` and `get` now fail with a clear error if the
runtime sends back something unexpected. Before, they printed an empty result.
##########
core/cli/src/args/context.rs:
##########
@@ -138,6 +150,14 @@ impl From<ContextCreateArgs> for ContextConfig {
password: args.password,
token: args.token,
token_name: args.token_name,
+ connectors_api_url: args.connectors_api_url,
+ connectors_api_key: args.connectors_api_key,
+ // Stored absolute so the context works from any directory.
+ connectors_tls_ca_file: args.connectors_tls_ca_file.map(|path| {
+ std::fs::canonicalize(&path)
+ .map(|absolute| absolute.to_string_lossy().into_owned())
+ .unwrap_or(path)
Review Comment:
Fixed in c62fec01b. If the CA file doesn't exist yet, the path is still
saved as absolute, so the saved context works from any directory.
##########
core/cli/src/args/connector.rs:
##########
@@ -0,0 +1,442 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::path::PathBuf;
+
+use crate::args::common::ListMode;
+use clap::{Args, Subcommand, ValueEnum};
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, ValueEnum)]
+pub(crate) enum ConnectorKind {
+ Sink,
+ Source,
+}
+
+/// Same rule as the runtime, so a key cannot carry `/`, `?`, `#` or `..`
+/// into the request path.
+fn parse_connector_key(key: &str) -> Result<String, String> {
Review Comment:
Fixed in c62fec01b. Keys over 128 bytes are now rejected by the CLI,
matching the runtime's limit.
##########
core/cli/src/commands/binary_connectors/client.rs:
##########
@@ -0,0 +1,258 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use anyhow::{Context, Result, bail};
+use passterm::{Stream, isatty, prompt_password_tty};
+use reqwest::{Method, StatusCode};
+use secrecy::{ExposeSecret, SecretString};
+use serde_json::Value;
+use std::path::PathBuf;
+use std::time::Duration;
+
+static DEFAULT_API_URL: &str = "http://localhost:8081";;
+const CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
+const REQUEST_TIMEOUT: Duration = Duration::from_secs(60);
+
+/// Sink or source: separate routes, and both sides may use the same key.
+#[derive(Debug, Clone, Copy)]
+pub enum ConnectorKind {
+ Sinks,
+ Sources,
+}
+
+impl ConnectorKind {
+ pub fn path(self) -> &'static str {
+ match self {
+ ConnectorKind::Sinks => "sinks",
+ ConnectorKind::Sources => "sources",
+ }
+ }
+
+ pub fn singular(self) -> &'static str {
+ match self {
+ ConnectorKind::Sinks => "sink",
+ ConnectorKind::Sources => "source",
+ }
+ }
+}
+
+/// Where a connector command reaches the runtime: flag, context or default.
+pub struct ConnectorEndpoint {
+ pub api_url: Option<String>,
+ pub api_key: Option<String>,
+ pub tls_ca_file: Option<PathBuf>,
+}
+
+impl ConnectorEndpoint {
+ pub fn connect(&self) -> Result<ConnectorsApiClient> {
+ ConnectorsApiClient::new(
+ self.api_url.clone(),
+ self.api_key.clone(),
+ self.tls_ca_file.clone(),
+ )
+ }
+}
+
+/// HTTP client for the connectors runtime control API.
+///
+/// Responses are read as JSON values: the response types are private to the
+/// runtime binary, and pulling in the SDK `api` feature would add Avro,
+/// Protobuf and FlatBuffers to the CLI dependency tree.
+pub struct ConnectorsApiClient {
+ inner: reqwest::Client,
+ base_url: String,
+ api_key: Option<SecretString>,
+}
+
+impl ConnectorsApiClient {
+ pub fn new(
+ api_url: Option<String>,
+ api_key: Option<String>,
+ tls_ca_file: Option<PathBuf>,
+ ) -> Result<Self> {
+ // No redirects: a redirect would resend the api-key header to another
host.
+ let mut builder = reqwest::Client::builder()
+ .connect_timeout(CONNECT_TIMEOUT)
+ .timeout(REQUEST_TIMEOUT)
+ .redirect(reqwest::redirect::Policy::none());
+ if let Some(ca_file) = tls_ca_file {
+ let pem = std::fs::read(&ca_file)
+ .with_context(|| format!("problem reading CA file {}",
ca_file.display()))?;
+ // from_pem only stores the bytes, so a wrong file would pass here
and fail
+ // later at the TLS handshake. The bundle parser checks the
encoding now.
+ let certificates = reqwest::Certificate::from_pem_bundle(&pem)
+ .with_context(|| format!("problem parsing CA file {} as PEM",
ca_file.display()))?;
+ if certificates.is_empty() {
+ bail!("CA file {} contains no certificate", ca_file.display());
+ }
+ for certificate in certificates {
+ builder = builder.add_root_certificate(certificate);
+ }
+ }
+ let inner = builder
+ .build()
+ .with_context(|| "problem building connectors API client")?;
+ let base_url = api_url
+ .unwrap_or_else(|| DEFAULT_API_URL.to_string())
+ .trim_end_matches('/')
+ .to_string();
+ let api_key = api_key.map(SecretString::from);
+ Ok(Self {
+ inner,
+ base_url,
+ api_key,
+ })
+ }
+
+ pub async fn get_value(&mut self, path: &str) -> Result<Value> {
+ let response = self.execute(Method::GET, path, &[], None).await?;
+ let response = Self::ensure_success(response, path).await?;
+ response
+ .json()
+ .await
+ .with_context(|| format!("problem decoding connectors API response
{path} as JSON"))
+ }
+
+ pub async fn get_text(&mut self, path: &str, query: &[(&str, String)]) ->
Result<String> {
+ let response = self.execute(Method::GET, path, query, None).await?;
+ let response = Self::ensure_success(response, path).await?;
+ response
+ .text()
+ .await
+ .with_context(|| format!("problem reading connectors API response
{path} as text"))
+ }
+
+ pub async fn post_json(&mut self, path: &str, body: &Value) ->
Result<Value> {
+ let response = self.execute(Method::POST, path, &[],
Some(body)).await?;
+ let response = Self::ensure_success(response, path).await?;
+ response
+ .json()
+ .await
+ .with_context(|| format!("problem decoding connectors API response
{path} as JSON"))
+ }
+
+ pub async fn post_empty(&mut self, path: &str) -> Result<()> {
+ let response = self.execute(Method::POST, path, &[], None).await?;
+ Self::ensure_success(response, path).await?;
+ Ok(())
+ }
+
+ pub async fn put_json(&mut self, path: &str, body: &Value) -> Result<()> {
+ let response = self.execute(Method::PUT, path, &[], Some(body)).await?;
+ Self::ensure_success(response, path).await?;
+ Ok(())
+ }
+
+ pub async fn delete(&mut self, path: &str, query: &[(&str, String)]) ->
Result<()> {
+ let response = self.execute(Method::DELETE, path, query, None).await?;
+ Self::ensure_success(response, path).await?;
+ Ok(())
+ }
+
+ /// Sends one request, retrying once after prompting for the key.
+ ///
+ /// The prompt only happens when no key is configured at all and stdin is
+ /// a terminal: a runtime without a key never triggers it, and a wrong
+ /// configured key fails with the server's reason instead of asking again.
+ async fn execute(
+ &mut self,
+ method: Method,
+ path: &str,
+ query: &[(&str, String)],
+ body: Option<&Value>,
+ ) -> Result<reqwest::Response> {
+ for attempt in 0..2 {
+ let mut request = self
+ .inner
+ .request(method.clone(), format!("{}{path}", self.base_url));
+ if !query.is_empty() {
+ request = request.query(query);
+ }
+ if let Some(body) = body {
+ request = request.json(body);
+ }
+ if let Some(key) = &self.api_key {
+ request = request.header("api-key", key.expose_secret());
+ }
+ let response = request
+ .send()
+ .await
+ .with_context(|| format!("problem calling connectors API
{method} {path}"))?;
+ if response.status() == StatusCode::UNAUTHORIZED
+ && self.api_key.is_none()
+ && attempt == 0
+ && isatty(Stream::Stdin)
+ {
+ let key = prompt_password_tty(Some("Connectors API key: "))?;
+ self.api_key = Some(SecretString::from(key));
+ continue;
+ }
+ return Ok(response);
+ }
+ bail!("connectors API request {method} {path} was not sent")
Review Comment:
Fixed in c62fec01b. Rewrote it as "send, and on a 401 ask for the key and
send once more". The dead error branch is gone.
##########
core/cli/src/commands/binary_connectors/get_transforms.rs:
##########
@@ -0,0 +1,91 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use super::client::{ConnectorEndpoint, ConnectorKind};
+use super::{as_array, cell};
+use crate::commands::cli_command::{CliCommand, PRINT_TARGET};
+use anyhow::Context;
+use async_trait::async_trait;
+use comfy_table::Table;
+use iggy_common::Client;
+use tracing::{Level, event};
+
+pub struct GetTransformsCmd {
+ kind: ConnectorKind,
+ key: String,
+ endpoint: ConnectorEndpoint,
+}
+
+impl GetTransformsCmd {
+ pub fn new(kind: ConnectorKind, key: String, endpoint: ConnectorEndpoint)
-> Self {
+ Self {
+ kind,
+ key,
+ endpoint,
+ }
+ }
+}
+
+#[async_trait]
+impl CliCommand for GetTransformsCmd {
+ fn explain(&self) -> String {
+ format!("get transforms of {} {}", self.kind.singular(), self.key)
+ }
+
+ fn login_required(&self) -> bool {
+ false
+ }
+
+ fn connection_required(&self) -> bool {
+ false
+ }
+
+ async fn execute_cmd(&mut self, _client: &dyn Client) ->
anyhow::Result<(), anyhow::Error> {
+ let mut api = self.endpoint.connect()?;
+ let path = format!("/{}/{}/transforms", self.kind.path(), self.key);
+ let transforms = api.get_value(&path).await.with_context(|| {
+ format!(
+ "problem getting transforms of {} {}",
+ self.kind.singular(),
+ self.key
+ )
+ })?;
+ let transforms = as_array(&transforms, "transforms")?;
+
+ if transforms.is_empty() {
+ event!(target: PRINT_TARGET, Level::INFO,
+ "No transforms found for {} {}!", self.kind.singular(),
self.key);
+ return Ok(());
+ }
+
+ let mut table = Table::new();
+ table.set_header(vec!["Type", "Config"]);
+ transforms.iter().for_each(|transform| {
+ table.add_row(vec![
+ transform
Review Comment:
Fixed in c62fec01b. It now uses the existing `str_field` helper.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
