feat: schema client and CLI (#4105)

* feat: schema client and CLI

* chore: clarification in comment in schema command

Co-authored-by: Andrew Lamb <andrew@nerdnetworks.org>

Co-authored-by: Andrew Lamb <andrew@nerdnetworks.org>
Co-authored-by: kodiakhq[bot] <49736102+kodiakhq[bot]@users.noreply.github.com>

influxdb_iox/src/commands/schema.rs

@@ -0,0 +1,52 @@
+//! This module implements the `schema` CLI command
+
+use clap_blocks::catalog_dsn::CatalogDsnConfig;
+use influxdb_iox_client::{connection::Connection, schema};
+use thiserror::Error;
+
+#[allow(clippy::enum_variant_names)]
+#[derive(Debug, Error)]
+pub enum Error {
+    #[error("JSON Serialization error: {0}")]
+    Serde(#[from] serde_json::Error),
+
+    #[error("Client error: {0}")]
+    ClientError(#[from] influxdb_iox_client::error::Error),
+}
+
+/// Various commands for catalog schema inspection
+#[derive(Debug, clap::Parser)]
+pub struct Config {
+    #[clap(subcommand)]
+    command: Command,
+}
+
+/// Run database migrations
+#[derive(Debug, clap::Parser)]
+struct Get {
+    #[clap(flatten)]
+    catalog_dsn: CatalogDsnConfig,
+
+    // The name of the namespace for which you want to fetch the schema
+    namespace: String,
+}
+
+/// All possible subcommands for catalog
+#[derive(Debug, clap::Parser)]
+enum Command {
+    /// Fetch schema for a namespace
+    Get(Get),
+}
+
+pub async fn command(connection: Connection, config: Config) -> Result<(), Error> {
+    match config.command {
+        Command::Get(command) => {
+            let mut client = schema::Client::new(connection);
+            let schema = client.get_schema(&command.namespace).await?;
+            println!("{}", serde_json::to_string_pretty(&schema)?);
+        } // Deliberately not adding _ => so the compiler will direct people here to impl new
+          // commands
+    }
+
+    Ok(())
+}

influxdb_iox/src/main.rs

@@ -30,6 +30,7 @@ mod commands {
     pub mod operations;
     pub mod router;
     pub mod run;
+    pub mod schema;
     pub mod server;
     pub mod server_remote;
     pub mod sql;
@@ -168,6 +169,9 @@ enum Command {
     /// Router-related commands
     Router(commands::router::Config),
 
+    /// IOx schema configuration commands
+    Schema(commands::schema::Config),
+
     /// IOx server configuration commands
     Server(commands::server::Config),
 
@@ -282,6 +286,14 @@ fn main() -> Result<(), std::io::Error> {
                     std::process::exit(ReturnCode::Failure as _)
                 }
             }
+            Command::Schema(config) => {
+                let _tracing_guard = handle_init_logs(init_simple_logs(log_verbose_count));
+                let connection = connection().await;
+                if let Err(e) = commands::schema::command(connection, config).await {
+                    eprintln!("{}", e);
+                    std::process::exit(ReturnCode::Failure as _)
+                }
+            }
             Command::Sql(config) => {
                 let _tracing_guard = handle_init_logs(init_simple_logs(log_verbose_count));
                 let connection = connection().await;

influxdb_iox_client/src/client.rs

@@ -19,6 +19,9 @@ pub mod remote;
 /// Client for router API
 pub mod router;
 
+/// Client for schema API
+pub mod schema;
+
 /// Client for write API
 pub mod write;
 

influxdb_iox_client/src/client/schema.rs

@@ -0,0 +1,37 @@
+use self::generated_types::{schema_service_client::SchemaServiceClient, *};
+use ::generated_types::google::OptionalField;
+
+use crate::connection::Connection;
+use crate::error::Error;
+
+/// Re-export generated_types
+pub mod generated_types {
+    pub use generated_types::influxdata::iox::schema::v1::*;
+}
+
+/// A basic client for fetching the Schema for a Namespace.
+#[derive(Debug, Clone)]
+pub struct Client {
+    inner: SchemaServiceClient<Connection>,
+}
+
+impl Client {
+    /// Creates a new client with the provided connection
+    pub fn new(channel: Connection) -> Self {
+        Self {
+            inner: SchemaServiceClient::new(channel),
+        }
+    }
+
+    /// Get the schema for a namespace.
+    pub async fn get_schema(&mut self, namespace: &str) -> Result<NamespaceSchema, Error> {
+        let response = self
+            .inner
+            .get_schema(GetSchemaRequest {
+                namespace: namespace.to_string(),
+            })
+            .await?;
+
+        Ok(response.into_inner().schema.unwrap_field("schema")?)
+    }
+}