-
Notifications
You must be signed in to change notification settings - Fork 198
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat: add core commands + metadata store (#3931)
* feat: add remote metadata store * feat: add core mirroring commands * chore: remove remote private api * chore: remove upstream * test: edge basic commands tests * fix: use edge instead core to register * fix: bump minor instead of patch * fix: export should check if edge exist * fix: apply feedbacks * fix: bump only patch for fluvio-protocol * fix: avoid clonings in list remote * chore: hide core command * fix: remove k8s feature for core commands
- Loading branch information
Showing
40 changed files
with
1,035 additions
and
33 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,77 @@ | ||
| use std::sync::Arc; | ||
| use anyhow::{Context, Result}; | ||
| use clap::Parser; | ||
| use fluvio_extension_common::{target::ClusterTarget, Terminal}; | ||
| use fluvio_sc_schema::{ | ||
| edge::EdgeMetadataExport, | ||
| remote::{Core, RemoteSpec, RemoteType}, | ||
| }; | ||
| use k8_types::K8Obj; | ||
| use anyhow::anyhow; | ||
|
|
||
| use super::get_admin; | ||
|
|
||
| #[derive(Debug, Parser)] | ||
| pub struct ExportOpt { | ||
| /// id of the edge cluster to export | ||
| edge_id: String, | ||
| /// name of the file where we should put the file | ||
| #[arg(long, short = 'f')] | ||
| file: Option<String>, | ||
| /// override endpoint of the core cluster | ||
| #[arg(long, short = 'e')] | ||
| public_endpoint: Option<String>, | ||
| // id of the core cluster to share | ||
| #[arg(name = "c")] | ||
| core_id: Option<String>, | ||
| } | ||
|
|
||
| impl ExportOpt { | ||
| pub async fn execute<T: Terminal>( | ||
| self, | ||
| out: Arc<T>, | ||
| cluster_target: ClusterTarget, | ||
| ) -> Result<()> { | ||
| let public_endpoint = if let Some(public_endpoint) = self.public_endpoint { | ||
| public_endpoint | ||
| } else { | ||
| let fluvio_config = cluster_target.clone().load()?; | ||
| fluvio_config.endpoint | ||
| }; | ||
|
|
||
| let admin = get_admin(cluster_target).await?; | ||
| let all_remotes = admin.all::<RemoteSpec>().await?; | ||
| let _edge = all_remotes | ||
| .iter() | ||
| .find(|remote| match &remote.spec.remote_type { | ||
| RemoteType::Edge(edge) => edge.id == self.edge_id, | ||
| _ => false, | ||
| }) | ||
| .ok_or_else(|| anyhow!("edge cluster not found"))?; | ||
|
|
||
| let core_id = self.core_id.clone().unwrap_or_else(|| "core".to_owned()); | ||
|
|
||
| let metadata_name = format!("edge-{}", self.edge_id); | ||
| let edge_metadata = vec![K8Obj::new( | ||
| metadata_name, | ||
| RemoteSpec { | ||
| remote_type: RemoteType::Core(Core { | ||
| id: core_id, | ||
| edge_id: self.edge_id, | ||
| public_endpoint, | ||
| }), | ||
| }, | ||
| )]; | ||
|
|
||
| let metadata = EdgeMetadataExport::new(edge_metadata); | ||
|
|
||
| if let Some(filename) = self.file { | ||
| std::fs::write(filename, serde_json::to_string_pretty(&metadata)?) | ||
| .context("failed to write output file")?; | ||
| } else { | ||
| out.println(&serde_json::to_string_pretty(&metadata)?); | ||
| } | ||
|
|
||
| Ok(()) | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,115 @@ | ||
| pub use std::sync::Arc; | ||
|
|
||
| use anyhow::Result; | ||
| use clap::Parser; | ||
| use fluvio_extension_common::target::ClusterTarget; | ||
| use fluvio_extension_common::{OutputFormat, Terminal}; | ||
| use fluvio_sc_schema::remote::{RemoteSpec, RemoteStatus}; | ||
|
|
||
| use super::get_admin; | ||
|
|
||
| #[derive(Debug, Parser)] | ||
| pub struct ListOpt { | ||
| #[clap(flatten)] | ||
| output: OutputFormat, | ||
| } | ||
| impl ListOpt { | ||
| pub async fn execute<T: Terminal>( | ||
| self, | ||
| out: Arc<T>, | ||
| cluster_target: ClusterTarget, | ||
| ) -> Result<()> { | ||
| let admin = get_admin(cluster_target).await?; | ||
| let list = admin.all::<RemoteSpec>().await?; | ||
|
|
||
| let outlist: Vec<(String, String, String, String)> = list | ||
| .into_iter() | ||
| .map(|item| { | ||
| let status: RemoteStatus = item.status; | ||
| ( | ||
| item.name, | ||
| item.spec.type_name().to_string(), | ||
| status.to_string(), | ||
| status.connection_stat.last_seen.to_string(), | ||
| ) | ||
| }) | ||
| .collect(); | ||
| output::format(out, outlist, self.output.format) | ||
| } | ||
| } | ||
|
|
||
| #[allow(dead_code)] | ||
| mod output { | ||
|
|
||
| //! | ||
| //! # Fluvio list - output processing | ||
| //! | ||
| //! Format SmartModules response based on output type | ||
| use comfy_table::{Cell, Row}; | ||
| use comfy_table::CellAlignment; | ||
| use tracing::debug; | ||
| use serde::Serialize; | ||
| use anyhow::Result; | ||
|
|
||
| use fluvio_extension_common::output::OutputType; | ||
| use fluvio_extension_common::Terminal; | ||
| use fluvio_extension_common::output::TableOutputHandler; | ||
| use fluvio_extension_common::t_println; | ||
|
|
||
| type ListVec = Vec<(String, String, String, String)>; | ||
|
|
||
| #[derive(Serialize)] | ||
| struct TableList(ListVec); | ||
|
|
||
| // ----------------------------------- | ||
| // Format Output | ||
| // ----------------------------------- | ||
|
|
||
| /// Format SmartModules based on output type | ||
| pub fn format<O: Terminal>( | ||
| out: std::sync::Arc<O>, | ||
| listvec: ListVec, | ||
| output_type: OutputType, | ||
| ) -> Result<()> { | ||
| debug!("listvec: {:#?}", listvec); | ||
|
|
||
| if !listvec.is_empty() { | ||
| let rlist = TableList(listvec); | ||
| out.render_list(&rlist, output_type)?; | ||
| Ok(()) | ||
| } else { | ||
| t_println!(out, "no items"); | ||
| Ok(()) | ||
| } | ||
| } | ||
|
|
||
| // ----------------------------------- | ||
| // Output Handlers | ||
| // ----------------------------------- | ||
| impl TableOutputHandler for TableList { | ||
| /// table header implementation | ||
| fn header(&self) -> Row { | ||
| Row::from(["REMOTE ", "TYPE", "STATUS", "LAST-SEEN"]) | ||
| } | ||
|
|
||
| /// return errors in string format | ||
| fn errors(&self) -> Vec<String> { | ||
| vec![] | ||
| } | ||
|
|
||
| /// table content implementation | ||
| fn content(&self) -> Vec<Row> { | ||
| self.0 | ||
| .iter() | ||
| .map(|e| { | ||
| Row::from([ | ||
| Cell::new(&e.0).set_alignment(CellAlignment::Left), | ||
| Cell::new(&e.1).set_alignment(CellAlignment::Left), | ||
| Cell::new(&e.2).set_alignment(CellAlignment::Left), | ||
| Cell::new(&e.3).set_alignment(CellAlignment::Left), | ||
| ]) | ||
| }) | ||
| .collect() | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,53 @@ | ||
| pub mod unregister; | ||
| pub mod list; | ||
| pub mod register; | ||
| pub mod export; | ||
|
|
||
| use std::sync::Arc; | ||
| use anyhow::Result; | ||
| use clap::Parser; | ||
| use fluvio_extension_common::target::ClusterTarget; | ||
| use unregister::UnregisterOpt; | ||
| use list::ListOpt; | ||
| use register::RegisterOpt; | ||
| use fluvio::FluvioAdmin; | ||
| use fluvio_extension_common::output::Terminal; | ||
| use self::export::ExportOpt; | ||
|
|
||
| #[derive(Debug, Parser)] | ||
| pub enum CoreCmd { | ||
| /// Register a new remote cluster | ||
| #[command(name = "register")] | ||
| Register(RegisterOpt), | ||
| /// List all remote clusters | ||
| #[command(name = "list")] | ||
| List(ListOpt), | ||
| /// List all remote clusters | ||
| #[command(name = "unregister")] | ||
| Unregister(UnregisterOpt), | ||
| // /// Generate metadata file for remote cluster | ||
| #[command(name = "export")] | ||
| Export(ExportOpt), | ||
| } | ||
|
|
||
| impl CoreCmd { | ||
| pub async fn process<O: Terminal>( | ||
| self, | ||
| out: Arc<O>, | ||
| cluster_target: ClusterTarget, | ||
| ) -> Result<()> { | ||
| match self { | ||
| Self::Register(reg) => reg.execute(out, cluster_target).await, | ||
| Self::Unregister(del) => del.execute(out, cluster_target).await, | ||
| Self::List(list) => list.execute(out, cluster_target).await, | ||
| Self::Export(meta) => meta.execute(out, cluster_target).await, | ||
| } | ||
| } | ||
| } | ||
|
|
||
| pub async fn get_admin(cluster_target: ClusterTarget) -> Result<FluvioAdmin> { | ||
| let fluvio_config = cluster_target.load()?; | ||
| let flv = fluvio::Fluvio::connect_with_config(&fluvio_config).await?; | ||
| let admin = flv.admin().await; | ||
| Ok(admin) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,36 @@ | ||
| use std::sync::Arc; | ||
| use anyhow::Result; | ||
| use clap::Parser; | ||
| use fluvio_controlplane_metadata::remote::{RemoteSpec, RemoteType}; | ||
| use fluvio_extension_common::target::ClusterTarget; | ||
| use fluvio_extension_common::Terminal; | ||
| use fluvio_sc_schema::remote::Edge; | ||
|
|
||
| #[derive(Debug, Parser)] | ||
| pub struct RegisterOpt { | ||
| name: String, | ||
| } | ||
|
|
||
| impl RegisterOpt { | ||
| pub async fn execute<T: Terminal>( | ||
| self, | ||
| _out: Arc<T>, | ||
| cluster_target: ClusterTarget, | ||
| ) -> Result<()> { | ||
| let fluvio_config = cluster_target.load()?; | ||
| let admin = fluvio::Fluvio::connect_with_config(&fluvio_config) | ||
| .await? | ||
| .admin() | ||
| .await; | ||
|
|
||
| let spec = RemoteSpec { | ||
| remote_type: RemoteType::Edge(Edge { | ||
| id: self.name.clone(), | ||
| }), | ||
| }; | ||
|
|
||
| admin.create(self.name.clone(), false, spec).await?; | ||
| println!("edge cluster \"{}\" was registered", self.name); | ||
| Ok(()) | ||
| } | ||
| } |
Oops, something went wrong.