feat(horaectl): impl horaectl in rs

Cargo.toml

@@ -25,6 +25,7 @@ license = "Apache-2.0"
 resolver = "2"
 # In alphabetical order
 members = [
+    "horaectl",
     "integration_tests",
     "integration_tests/sdk/rust",
     "src/analytic_engine",
@@ -89,6 +90,7 @@ arrow = { version = "49.0.0", features = ["prettyprint"] }
 arrow_ipc = { version = "49.0.0" }
 arrow_ext = { path = "src/components/arrow_ext" }
 analytic_engine = { path = "src/analytic_engine" }
+anyhow = { version = "1.0" }
 arena = { path = "src/components/arena" }
 async-stream = "0.3.4"
 async-trait = "0.1.72"
@@ -101,7 +103,7 @@ catalog_impls = { path = "src/catalog_impls" }
 horaedbproto = { git = "https://github.com/apache/incubator-horaedb-proto.git", rev = "19ece8f771fc0b3e8e734072cc3d8040de6c74cb" }
 codec = { path = "src/components/codec" }
 chrono = "0.4"
-clap = "4.5.1"
+clap = { version = "4.5.1", features = ["derive"] }
 clru = "0.6.1"
 cluster = { path = "src/cluster" }
 criterion = "0.5"

horaectl/Cargo.toml

@@ -0,0 +1,39 @@
+# 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.
+
+[package]
+name = "horaectl"
+
+[package.license]
+workspace = true
+
+[package.version]
+workspace = true
+
+[package.edition]
+workspace = true
+
+[dependencies]
+anyhow = { workspace = true, features = ["backtrace"] }
+chrono = { workspace = true }
+clap = { workspace = true, features = ["env", "derive"] }
+lazy_static = { workspace = true }
+prettytable = "0.10.0"
+reqwest = { workspace = true }
+serde = { workspace = true }
+shell-words = "1.1.0"
+tokio = { workspace = true }

horaectl/src/cmd/cluster.rs

@@ -0,0 +1,67 @@
+// 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::Result;
+use clap::Subcommand;
+
+use crate::operation::cluster::ClusterOp;
+
+#[derive(Subcommand)]
+pub enum ClusterCommand {
+    /// List cluster
+    List,
+
+    /// Diagnose cluster
+    Diagnose,
+
+    /// Schedule cluster
+    Schedule {
+        #[clap(subcommand)]
+        cmd: Option<ScheduleCommand>,
+    },
+}
+
+#[derive(Subcommand)]
+pub enum ScheduleCommand {
+    /// Get the schedule status
+    Get,
+
+    /// Enable schedule
+    On,
+
+    /// Disable schedule
+    Off,
+}
+
+pub async fn run(cmd: ClusterCommand) -> Result<()> {
+    let op = ClusterOp::try_new()?;
+    match cmd {
+        ClusterCommand::List => op.list().await,
+        ClusterCommand::Diagnose => op.diagnose().await,
+        ClusterCommand::Schedule { cmd } => {
+            if let Some(cmd) = cmd {
+                match cmd {
+                    ScheduleCommand::Get => op.get_schedule_status().await,
+                    ScheduleCommand::On => op.update_schedule_status(true).await,
+                    ScheduleCommand::Off => op.update_schedule_status(false).await,
+                }
+            } else {
+                op.get_schedule_status().await
+            }
+        }
+    }
+}

horaectl/src/cmd/mod.rs

@@ -0,0 +1,140 @@
+// 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.
+
+mod cluster;
+use std::{io, io::Write};
+
+use anyhow::Result;
+use clap::{Args, Parser, Subcommand};
+
+use crate::{
+    cmd::cluster::ClusterCommand,
+    util::{CLUSTER_NAME, META_ADDR},
+};
+
+#[derive(Parser)]
+#[clap(name = "horaectl")]
+#[clap(about = "HoraeCTL is a command line tool for HoraeDB", version)]
+pub struct App {
+    #[clap(flatten)]
+    pub global_opts: GlobalOpts,
+
+    /// Enter interactive mode
+    #[clap(short, long, default_value_t = false)]
+    pub interactive: bool,
+
+    #[clap(subcommand)]
+    pub command: Option<SubCommand>,
+}
+
+#[derive(Debug, Args)]
+pub struct GlobalOpts {
+    /// Meta addr
+    #[clap(
+        short,
+        long = "meta",
+        global = true,
+        env = "HORAECTL_META_ADDR",
+        default_value = "127.0.0.1:8080"
+    )]
+    pub meta_addr: String,
+
+    /// Cluster name
+    #[clap(
+        short,
+        long = "cluster",
+        global = true,
+        env = "HORAECTL_CLUSTER",
+        default_value = "defaultCluster"
+    )]
+    pub cluster_name: String,
+}
+
+#[derive(Subcommand)]
+pub enum SubCommand {
+    /// Operations on cluster
+    #[clap(alias = "c")]
+    Cluster {
+        #[clap(subcommand)]
+        commands: ClusterCommand,
+    },
+}
+
+pub async fn run_command(cmd: SubCommand) -> Result<()> {
+    match cmd {
+        SubCommand::Cluster { commands } => cluster::run(commands).await,
+    }
+}
+
+pub async fn repl_loop() {
+    loop {
+        print_prompt(
+            META_ADDR.lock().unwrap().as_str(),
+            CLUSTER_NAME.lock().unwrap().as_str(),
+        );
+
+        let args = match read_args() {
+            Ok(args) => args,
+            Err(e) => {
+                println!("Read input failed, err:{}", e);
+                continue;
+            }
+        };
+
+        if let Some(cmd) = args.get(1) {
+            if ["quit", "exit", "q"].iter().any(|v| v == cmd) {
+                break;
+            }
+        }
+
+        match App::try_parse_from(args) {
+            Ok(horaectl) => {
+                if let Some(cmd) = horaectl.command {
+                    if let Err(e) = match cmd {
+                        SubCommand::Cluster { commands } => cluster::run(commands).await,
+                    } {
+                        println!("Run command failed, err:{e}");
+                    }
+                }
+            }
+            Err(e) => {
+                println!("Parse command failed, err:{e}");
+            }
+        }
+    }
+}
+
+fn read_args() -> Result<Vec<String>, String> {
+    io::stdout().flush().unwrap();
+    let mut input = String::new();
+    io::stdin()
+        .read_line(&mut input)
+        .map_err(|e| e.to_string())?;
+
+    let input = input.trim();
+    if input.is_empty() {
+        return Err("No arguments provided".into());
+    }
+
+    let mut args = vec!["horaectl".to_string()];
+    args.extend(shell_words::split(input).map_err(|e| e.to_string())?);
+    Ok(args)
+}
+
+fn print_prompt(address: &str, cluster: &str) {
+    print!("{}({}) > ", address, cluster);
+}

horaectl/src/main.rs

@@ -0,0 +1,54 @@
+// 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.
+
+mod cmd;
+mod operation;
+mod util;
+
+use clap::{CommandFactory, Parser};
+
+use crate::{
+    cmd::{repl_loop, run_command, App},
+    util::{CLUSTER_NAME, META_ADDR},
+};
+
+#[tokio::main]
+async fn main() {
+    let app = App::parse();
+    {
+        let mut meta_addr = META_ADDR.lock().unwrap();
+        *meta_addr = app.global_opts.meta_addr;
+    }
+    {
+        let mut cluster_name = CLUSTER_NAME.lock().unwrap();
+        *cluster_name = app.global_opts.cluster_name;
+    }
+
+    if app.interactive {
+        repl_loop().await;
+        return;
+    }
+
+    if let Some(cmd) = app.command {
+        if let Err(e) = run_command(cmd).await {
+            println!("Run command failed, err:{e}");
+            std::process::exit(1);
+        }
+    } else {
+        App::command().print_help().expect("print help failed");
+    }
+}