Compare commits

..

3 Commits

Author SHA1 Message Date
6933280575 feat(helm): refactor helm execution to use topology-specific commands
Some checks failed
Run Check Script / check (pull_request) Failing after 8s
Refactors the `HelmChartInterpret` to move away from the `helm-wrapper-rs` crate in favor of a custom command builder pattern. This allows the `HelmCommand` trait to provide topology-specific configurations, such as `kubeconfig` and `kube-context`, directly to the `helm` CLI.

- Implements `get_helm_command` for `K8sAnywhereTopology` to inject configuration flags.
- Replaces `DefaultHelmExecutor` with a manual `Command` construction in `run_helm_command`.
- Updates `HelmChartInterpret` to pass the topology through to repository and installation logic.
- Cleans up unused imports and removes the temporary `HelmCommand` implementation for `LocalhostTopology`.
2026-01-08 23:42:54 -05:00
77583a1ad1 wip: nats multi cluster, fixing helm command to follow multiple k8s config by providing the helm command from the topology itself, fix cli_logger that can now be initialized multiple times, some more stuff 2026-01-08 16:03:15 -05:00
f7404bed36 wip: initial setup for installing nats helm chart score 2026-01-07 16:14:58 -05:00
11 changed files with 369 additions and 140 deletions

View File

@@ -3,15 +3,27 @@ use std::str::FromStr;
use harmony::{
inventory::Inventory,
modules::helm::chart::{HelmChartScore, HelmRepository, NonBlankString},
topology::K8sAnywhereTopology,
topology::{HelmCommand, K8sAnywhereConfig, K8sAnywhereTopology, TlsRouter, Topology},
};
use harmony_macros::hurl;
use log::info;
#[tokio::main]
async fn main() {
// env_logger::init();
let values_yaml = Some(
deploy_nats(K8sAnywhereTopology::with_config(
K8sAnywhereConfig::remote_k8s_from_env_var("HARMONY_NATS_SITE_1"),
))
.await;
deploy_nats(K8sAnywhereTopology::with_config(
K8sAnywhereConfig::remote_k8s_from_env_var("HARMONY_NATS_SITE_2"),
))
.await;
}
async fn deploy_nats<T: Topology + HelmCommand + TlsRouter + 'static>(topology: T) {
topology.ensure_ready().await.unwrap();
let values_yaml = Some(format!(
r#"config:
cluster:
enabled: true
@@ -25,6 +37,14 @@ async fn main() {
leafnodes:
enabled: false
# port: 7422
websocket:
enabled: true
ingress:
enabled: true
className: openshift-default
pathType: Prefix
hosts:
- nats-ws.{}
gateway:
enabled: false
# name: my-gateway
@@ -32,9 +52,9 @@ async fn main() {
natsBox:
container:
image:
tag: nonroot"#
.to_string(),
);
tag: nonroot"#,
topology.get_internal_domain().await.unwrap().unwrap(),
));
let namespace = "nats";
let nats = HelmChartScore {
namespace: Some(NonBlankString::from_str(namespace).unwrap()),
@@ -52,14 +72,9 @@ natsBox:
)),
};
harmony_cli::run(
Inventory::autoload(),
K8sAnywhereTopology::from_env(),
vec![Box::new(nats)],
None,
)
.await
.unwrap();
harmony_cli::run(Inventory::autoload(), topology, vec![Box::new(nats)], None)
.await
.unwrap();
info!(
"Enjoy! You can test your nats cluster by running : `kubectl exec -n {namespace} -it deployment/nats-box -- nats pub test hi`"

View File

@@ -1 +1,5 @@
pub trait HelmCommand {}
use std::process::Command;
pub trait HelmCommand {
fn get_helm_command(&self) -> Command;
}

View File

@@ -35,6 +35,7 @@ use crate::{
service_monitor::ServiceMonitor,
},
},
okd::crd::ingresses_config::Ingress as IngressResource,
okd::route::OKDTlsPassthroughScore,
prometheus::{
k8s_prometheus_alerting_score::K8sPrometheusCRDAlertingScore,
@@ -107,8 +108,32 @@ impl K8sclient for K8sAnywhereTopology {
#[async_trait]
impl TlsRouter for K8sAnywhereTopology {
async fn get_wildcard_domain(&self) -> Result<Option<String>, String> {
todo!()
async fn get_internal_domain(&self) -> Result<Option<String>, String> {
match self.get_k8s_distribution().await.map_err(|e| {
format!(
"Could not get internal domain, error getting k8s distribution : {}",
e.to_string()
)
})? {
KubernetesDistribution::OpenshiftFamily => {
let client = self.k8s_client().await?;
if let Some(ingress_config) = client
.get_resource::<IngressResource>("cluster", None)
.await
.map_err(|e| {
format!("Error attempting to get ingress config : {}", e.to_string())
})?
{
debug!("Found ingress config {:?}", ingress_config.spec);
Ok(ingress_config.spec.domain.clone())
} else {
warn!("Could not find a domain configured in this cluster");
Ok(None)
}
}
KubernetesDistribution::K3sFamily => todo!(),
KubernetesDistribution::Default => todo!(),
}
}
/// Returns the port that this router exposes externally.
@@ -1087,7 +1112,21 @@ impl MultiTargetTopology for K8sAnywhereTopology {
}
}
impl HelmCommand for K8sAnywhereTopology {}
impl HelmCommand for K8sAnywhereTopology {
fn get_helm_command(&self) -> Command {
let mut cmd = Command::new("helm");
if let Some(k) = &self.config.kubeconfig {
cmd.args(["--kubeconfig", k]);
}
if let Some(c) = &self.config.k8s_context {
cmd.args(["--kube-context", c]);
}
info!("Using helm command {cmd:?}");
cmd
}
}
#[async_trait]
impl TenantManager for K8sAnywhereTopology {
@@ -1108,7 +1147,7 @@ impl TenantManager for K8sAnywhereTopology {
#[async_trait]
impl Ingress for K8sAnywhereTopology {
async fn get_domain(&self, service: &str) -> Result<String, PreparationError> {
use log::{debug, trace, warn};
use log::{trace, warn};
let client = self.k8s_client().await?;

View File

@@ -2,7 +2,7 @@ use async_trait::async_trait;
use derive_new::new;
use serde::{Deserialize, Serialize};
use super::{HelmCommand, PreparationError, PreparationOutcome, Topology};
use super::{PreparationError, PreparationOutcome, Topology};
#[derive(new, Clone, Debug, Serialize, Deserialize)]
pub struct LocalhostTopology;
@@ -19,6 +19,3 @@ impl Topology for LocalhostTopology {
})
}
}
// TODO: Delete this, temp for test
impl HelmCommand for LocalhostTopology {}

View File

@@ -112,12 +112,13 @@ pub trait TlsRouter: Send + Sync {
/// HAProxy frontend→backend \"postgres-upstream\".
async fn install_route(&self, config: TlsRoute) -> Result<(), String>;
/// Gets the base domain that can be used to deploy applications that will be automatically
/// routed to this cluster.
/// Gets the base domain of this cluster. On openshift family clusters, this is the domain
/// used by default for all components, including the default ingress controller that
/// transforms ingress to routes.
///
/// For example, if we have *.apps.nationtech.io pointing to a public load balancer, then this
/// function would install route apps.nationtech.io
async fn get_wildcard_domain(&self) -> Result<Option<String>, String>;
/// For example, get_internal_domain on a cluster that has `console-openshift-console.apps.mycluster.something`
/// will return `apps.mycluster.something`
async fn get_internal_domain(&self) -> Result<Option<String>, String>;
/// Returns the port that this router exposes externally.
async fn get_router_port(&self) -> u16;

View File

@@ -6,15 +6,11 @@ use crate::topology::{HelmCommand, Topology};
use async_trait::async_trait;
use harmony_types::id::Id;
use harmony_types::net::Url;
use helm_wrapper_rs;
use helm_wrapper_rs::blocking::{DefaultHelmExecutor, HelmExecutor};
use log::{debug, info, warn};
pub use non_blank_string_rs::NonBlankString;
use serde::Serialize;
use std::collections::HashMap;
use std::path::Path;
use std::process::{Command, Output, Stdio};
use std::str::FromStr;
use std::process::{Output, Stdio};
use temp_file::TempFile;
#[derive(Debug, Clone, Serialize)]
@@ -65,7 +61,7 @@ pub struct HelmChartInterpret {
pub score: HelmChartScore,
}
impl HelmChartInterpret {
fn add_repo(&self) -> Result<(), InterpretError> {
fn add_repo<T: HelmCommand>(&self, topology: &T) -> Result<(), InterpretError> {
let repo = match &self.score.repository {
Some(repo) => repo,
None => {
@@ -84,7 +80,7 @@ impl HelmChartInterpret {
add_args.push("--force-update");
}
let add_output = run_helm_command(&add_args)?;
let add_output = run_helm_command(topology, &add_args)?;
let full_output = format!(
"{}\n{}",
String::from_utf8_lossy(&add_output.stdout),
@@ -100,23 +96,19 @@ impl HelmChartInterpret {
}
}
fn run_helm_command(args: &[&str]) -> Result<Output, InterpretError> {
let command_str = format!("helm {}", args.join(" "));
debug!(
"Got KUBECONFIG: `{}`",
std::env::var("KUBECONFIG").unwrap_or("".to_string())
);
debug!("Running Helm command: `{}`", command_str);
fn run_helm_command<T: HelmCommand>(topology: &T, args: &[&str]) -> Result<Output, InterpretError> {
let mut helm_cmd = topology.get_helm_command();
helm_cmd.args(args);
let output = Command::new("helm")
.args(args)
debug!("Running Helm command: `{:?}`", helm_cmd);
let output = helm_cmd
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.output()
.map_err(|e| {
InterpretError::new(format!(
"Failed to execute helm command '{}': {}. Is helm installed and in PATH?",
command_str, e
"Failed to execute helm command '{helm_cmd:?}': {e}. Is helm installed and in PATH?",
))
})?;
@@ -124,13 +116,13 @@ fn run_helm_command(args: &[&str]) -> Result<Output, InterpretError> {
let stdout = String::from_utf8_lossy(&output.stdout);
let stderr = String::from_utf8_lossy(&output.stderr);
warn!(
"Helm command `{}` failed with status: {}\nStdout:\n{}\nStderr:\n{}",
command_str, output.status, stdout, stderr
"Helm command `{helm_cmd:?}` failed with status: {}\nStdout:\n{stdout}\nStderr:\n{stderr}",
output.status
);
} else {
debug!(
"Helm command `{}` finished successfully. Status: {}",
command_str, output.status
"Helm command `{helm_cmd:?}` finished successfully. Status: {}",
output.status
);
}
@@ -142,7 +134,7 @@ impl<T: Topology + HelmCommand> Interpret<T> for HelmChartInterpret {
async fn execute(
&self,
_inventory: &Inventory,
_topology: &T,
topology: &T,
) -> Result<Outcome, InterpretError> {
let ns = self
.score
@@ -150,98 +142,62 @@ impl<T: Topology + HelmCommand> Interpret<T> for HelmChartInterpret {
.as_ref()
.unwrap_or_else(|| todo!("Get namespace from active kubernetes cluster"));
let tf: TempFile;
let yaml_path: Option<&Path> = match self.score.values_yaml.as_ref() {
Some(yaml_str) => {
tf = temp_file::with_contents(yaml_str.as_bytes());
debug!(
"values yaml string for chart {} :\n {yaml_str}",
self.score.chart_name
);
Some(tf.path())
}
None => None,
self.add_repo(topology)?;
let mut args = if self.score.install_only {
vec!["install"]
} else {
vec!["upgrade", "--install"]
};
self.add_repo()?;
let helm_executor = DefaultHelmExecutor::new_with_opts(
&NonBlankString::from_str("helm").unwrap(),
None,
900,
false,
false,
);
let mut helm_options = Vec::new();
if self.score.create_namespace {
helm_options.push(NonBlankString::from_str("--create-namespace").unwrap());
}
if self.score.install_only {
let chart_list = match helm_executor.list(Some(ns)) {
Ok(charts) => charts,
Err(e) => {
return Err(InterpretError::new(format!(
"Failed to list scores in namespace {:?} because of error : {}",
self.score.namespace, e
)));
}
};
if chart_list
.iter()
.any(|item| item.name == self.score.release_name.to_string())
{
info!(
"Release '{}' already exists in namespace '{}'. Skipping installation as install_only is true.",
self.score.release_name, ns
);
return Ok(Outcome::success(format!(
"Helm Chart '{}' already installed to namespace {ns} and install_only=true",
self.score.release_name
)));
} else {
info!(
"Release '{}' not found in namespace '{}'. Proceeding with installation.",
self.score.release_name, ns
);
}
}
let res = helm_executor.install_or_upgrade(
ns,
args.extend(vec![
&self.score.release_name,
&self.score.chart_name,
self.score.chart_version.as_ref(),
self.score.values_overrides.as_ref(),
yaml_path,
Some(&helm_options),
);
"--namespace",
&ns,
]);
let status = match res {
Ok(status) => status,
Err(err) => return Err(InterpretError::new(err.to_string())),
};
if self.score.create_namespace {
args.push("--create-namespace");
}
match status {
helm_wrapper_rs::HelmDeployStatus::Deployed => Ok(Outcome::success(format!(
if let Some(version) = &self.score.chart_version {
args.push("--version");
args.push(&version);
}
let tf: TempFile;
if let Some(yaml_str) = &self.score.values_yaml {
tf = temp_file::with_contents(yaml_str.as_bytes());
args.push("--values");
args.push(tf.path().to_str().unwrap());
}
let overrides_strings: Vec<String>;
if let Some(overrides) = &self.score.values_overrides {
overrides_strings = overrides
.iter()
.map(|(key, value)| format!("{key}={value}"))
.collect();
for o in overrides_strings.iter() {
args.push("--set");
args.push(&o);
}
}
let output = run_helm_command(topology, &args)?;
if output.status.success() {
Ok(Outcome::success(format!(
"Helm Chart {} deployed",
self.score.release_name
))),
helm_wrapper_rs::HelmDeployStatus::PendingInstall => Ok(Outcome::running(format!(
"Helm Chart {} pending install...",
self.score.release_name
))),
helm_wrapper_rs::HelmDeployStatus::PendingUpgrade => Ok(Outcome::running(format!(
"Helm Chart {} pending upgrade...",
self.score.release_name
))),
helm_wrapper_rs::HelmDeployStatus::Failed => Err(InterpretError::new(format!(
"Helm Chart {} installation failed",
self.score.release_name
))),
)))
} else {
Err(InterpretError::new(format!(
"Helm Chart {} installation failed: {}",
self.score.release_name,
String::from_utf8_lossy(&output.stderr)
)))
}
}

View File

@@ -5,7 +5,7 @@ use crate::topology::{FailoverTopology, TlsRoute, TlsRouter};
#[async_trait]
impl<T: TlsRouter> TlsRouter for FailoverTopology<T> {
async fn get_wildcard_domain(&self) -> Result<Option<String>, String> {
async fn get_internal_domain(&self) -> Result<Option<String>, String> {
todo!()
}

View File

@@ -0,0 +1,214 @@
use k8s_openapi::apimachinery::pkg::apis::meta::v1::{ListMeta, ObjectMeta};
use k8s_openapi::{ClusterResourceScope, Resource};
use serde::{Deserialize, Serialize};
#[derive(Deserialize, Serialize, Clone, Debug)]
#[serde(rename_all = "camelCase")]
pub struct Ingress {
#[serde(skip_serializing_if = "Option::is_none")]
pub api_version: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub kind: Option<String>,
pub metadata: ObjectMeta,
pub spec: IngressSpec,
#[serde(skip_serializing_if = "Option::is_none")]
pub status: Option<IngressStatus>,
}
impl Resource for Ingress {
const API_VERSION: &'static str = "config.openshift.io/v1";
const GROUP: &'static str = "config.openshift.io";
const VERSION: &'static str = "v1";
const KIND: &'static str = "Ingress";
const URL_PATH_SEGMENT: &'static str = "ingresses";
type Scope = ClusterResourceScope;
}
impl k8s_openapi::Metadata for Ingress {
type Ty = ObjectMeta;
fn metadata(&self) -> &Self::Ty {
&self.metadata
}
fn metadata_mut(&mut self) -> &mut Self::Ty {
&mut self.metadata
}
}
impl Default for Ingress {
fn default() -> Self {
Ingress {
api_version: Some("config.openshift.io/v1".to_string()),
kind: Some("Ingress".to_string()),
metadata: ObjectMeta::default(),
spec: IngressSpec::default(),
status: None,
}
}
}
#[derive(Deserialize, Serialize, Clone, Debug)]
#[serde(rename_all = "camelCase")]
pub struct IngressList {
pub metadata: ListMeta,
pub items: Vec<Ingress>,
}
impl Default for IngressList {
fn default() -> Self {
Self {
metadata: ListMeta::default(),
items: Vec::new(),
}
}
}
impl Resource for IngressList {
const API_VERSION: &'static str = "config.openshift.io/v1";
const GROUP: &'static str = "config.openshift.io";
const VERSION: &'static str = "v1";
const KIND: &'static str = "IngressList";
const URL_PATH_SEGMENT: &'static str = "ingresses";
type Scope = ClusterResourceScope;
}
impl k8s_openapi::Metadata for IngressList {
type Ty = ListMeta;
fn metadata(&self) -> &Self::Ty {
&self.metadata
}
fn metadata_mut(&mut self) -> &mut Self::Ty {
&mut self.metadata
}
}
#[derive(Deserialize, Serialize, Clone, Debug, Default)]
#[serde(rename_all = "camelCase")]
pub struct IngressSpec {
#[serde(skip_serializing_if = "Option::is_none")]
pub apps_domain: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub component_routes: Option<Vec<ComponentRouteSpec>>,
#[serde(skip_serializing_if = "Option::is_none")]
pub domain: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub load_balancer: Option<LoadBalancer>,
#[serde(skip_serializing_if = "Option::is_none")]
pub required_hsts_policies: Option<Vec<RequiredHSTSPolicy>>,
}
#[derive(Deserialize, Serialize, Clone, Debug)]
#[serde(rename_all = "camelCase")]
pub struct ComponentRouteSpec {
pub hostname: String,
pub name: String,
pub namespace: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub serving_cert_key_pair_secret: Option<SecretNameReference>,
}
#[derive(Deserialize, Serialize, Clone, Debug)]
#[serde(rename_all = "camelCase")]
pub struct SecretNameReference {
pub name: String,
}
#[derive(Deserialize, Serialize, Clone, Debug, Default)]
#[serde(rename_all = "camelCase")]
pub struct LoadBalancer {
#[serde(skip_serializing_if = "Option::is_none")]
pub platform: Option<IngressPlatform>,
}
#[derive(Deserialize, Serialize, Clone, Debug, Default)]
#[serde(rename_all = "camelCase")]
pub struct IngressPlatform {
#[serde(skip_serializing_if = "Option::is_none")]
pub aws: Option<AWSPlatformLoadBalancer>,
#[serde(skip_serializing_if = "Option::is_none")]
pub r#type: Option<String>,
}
#[derive(Deserialize, Serialize, Clone, Debug)]
#[serde(rename_all = "camelCase")]
pub struct AWSPlatformLoadBalancer {
pub r#type: String,
}
#[derive(Deserialize, Serialize, Clone, Debug)]
#[serde(rename_all = "camelCase")]
pub struct RequiredHSTSPolicy {
pub domain_patterns: Vec<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub include_sub_domains_policy: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub max_age: Option<MaxAgePolicy>,
#[serde(skip_serializing_if = "Option::is_none")]
pub namespace_selector: Option<k8s_openapi::apimachinery::pkg::apis::meta::v1::LabelSelector>,
#[serde(skip_serializing_if = "Option::is_none")]
pub preload_policy: Option<String>,
}
#[derive(Deserialize, Serialize, Clone, Debug, Default)]
#[serde(rename_all = "camelCase")]
pub struct MaxAgePolicy {
#[serde(skip_serializing_if = "Option::is_none")]
pub largest_max_age: Option<i32>,
#[serde(skip_serializing_if = "Option::is_none")]
pub smallest_max_age: Option<i32>,
}
#[derive(Deserialize, Serialize, Clone, Debug, Default)]
#[serde(rename_all = "camelCase")]
pub struct IngressStatus {
#[serde(skip_serializing_if = "Option::is_none")]
pub component_routes: Option<Vec<ComponentRouteStatus>>,
#[serde(skip_serializing_if = "Option::is_none")]
pub default_placement: Option<String>,
}
#[derive(Deserialize, Serialize, Clone, Debug)]
#[serde(rename_all = "camelCase")]
pub struct ComponentRouteStatus {
#[serde(skip_serializing_if = "Option::is_none")]
pub conditions: Option<Vec<k8s_openapi::apimachinery::pkg::apis::meta::v1::Condition>>,
#[serde(skip_serializing_if = "Option::is_none")]
pub consuming_users: Option<Vec<String>>,
#[serde(skip_serializing_if = "Option::is_none")]
pub current_hostnames: Option<Vec<String>>,
pub default_hostname: String,
pub name: String,
pub namespace: String,
pub related_objects: Vec<ObjectReference>,
}
#[derive(Deserialize, Serialize, Clone, Debug)]
#[serde(rename_all = "camelCase")]
pub struct ObjectReference {
pub group: String,
pub name: String,
pub namespace: String,
pub resource: String,
}

View File

@@ -1,2 +1,3 @@
pub mod nmstate;
pub mod route;
pub mod ingresses_config;

View File

@@ -1,5 +1,4 @@
use k8s_openapi::apimachinery::pkg::apis::meta::v1::{ListMeta, ObjectMeta, Time};
use k8s_openapi::apimachinery::pkg::util::intstr::IntOrString;
use k8s_openapi::{NamespaceResourceScope, Resource};
use serde::{Deserialize, Serialize};

View File

@@ -7,11 +7,14 @@ use harmony::{
};
use log::{error, info, log_enabled};
use std::io::Write;
use std::sync::Mutex;
use std::sync::{Mutex, OnceLock};
pub fn init() {
configure_logger();
handle_events();
static INITIALIZED: OnceLock<()> = OnceLock::new();
INITIALIZED.get_or_init(|| {
configure_logger();
handle_events();
});
}
fn configure_logger() {