jvtrudel/harmony
1
0
Fork 0
forked from NationTech/harmony
Code Pull Requests Activity

fix merge conflict

Browse Source
This commit is contained in:
Willem
2025-07-02 11:26:27 -04:00
parent 7de9860dcf ee02906ce9
commit e857efa92f
19 changed files with 520 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
harmony/src/domain/config.rs
View File

@@ -10,4 +10,6 @@ lazy_static! {
std::env::var("HARMONY_REGISTRY_URL").unwrap_or_else(|_| "hub.nationtech.io".to_string());
pub static ref REGISTRY_PROJECT: String =
std::env::var("HARMONY_REGISTRY_PROJECT").unwrap_or_else(|_| "harmony".to_string());
pub static ref DRY_RUN: bool =
std::env::var("HARMONY_DRY_RUN").map_or(true, |value| value.parse().unwrap_or(true));
}

90
harmony/src/domain/topology/k8s.rs
View File

@@ -4,15 +4,28 @@ use kube::{
Api, Client, Config, Error, Resource,
api::{Patch, PatchParams},
config::{KubeConfigOptions, Kubeconfig},
core::ErrorResponse,
};
use log::{debug, error, trace};
use serde::de::DeserializeOwned;
use similar::TextDiff;
#[derive(new)]
#[derive(new, Clone)]
pub struct K8sClient {
client: Client,
}
impl std::fmt::Debug for K8sClient {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
// This is a poor man's debug implementation for now as kube::Client does not provide much
// useful information
f.write_fmt(format_args!(
"K8sClient {{ kube client using default namespace {} }}",
self.client.default_namespace()
))
}
}
impl K8sClient {
pub async fn try_default() -> Result<Self, Error> {
Ok(Self {
@@ -48,8 +61,79 @@ impl K8sClient {
.name
.as_ref()
.expect("K8s Resource should have a name");
api.patch(name, &patch_params, &Patch::Apply(resource))
.await
if *crate::config::DRY_RUN {
match api.get(name).await {
Ok(current) => {
trace!("Received current value {current:#?}");
// The resource exists, so we calculate and display a diff.
println!("\nPerforming dry-run for resource: '{}'", name);
let mut current_yaml = serde_yaml::to_value(&current)
.expect(&format!("Could not serialize current value : {current:#?}"));
if current_yaml.is_mapping() && current_yaml.get("status").is_some() {
let map = current_yaml.as_mapping_mut().unwrap();
let removed = map.remove_entry("status");
trace!("Removed status {:?}", removed);
} else {
trace!(
"Did not find status entry for current object {}/{}",
current.meta().namespace.as_ref().unwrap_or(&"".to_string()),
current.meta().name.as_ref().unwrap_or(&"".to_string())
);
}
let current_yaml = serde_yaml::to_string(&current_yaml)
.unwrap_or_else(|_| "Failed to serialize current resource".to_string());
let new_yaml = serde_yaml::to_string(resource)
.unwrap_or_else(|_| "Failed to serialize new resource".to_string());
if current_yaml == new_yaml {
println!("No changes detected.");
// Return the current resource state as there are no changes.
return Ok(current);
}
println!("Changes detected:");
let diff = TextDiff::from_lines(&current_yaml, &new_yaml);
// Iterate over the changes and print them in a git-like diff format.
for change in diff.iter_all_changes() {
let sign = match change.tag() {
similar::ChangeTag::Delete => "-",
similar::ChangeTag::Insert => "+",
similar::ChangeTag::Equal => " ",
};
print!("{}{}", sign, change);
}
// In a dry run, we return the new resource state that would have been applied.
Ok(resource.clone())
}
Err(Error::Api(ErrorResponse { code: 404, .. })) => {
// The resource does not exist, so the "diff" is the entire new resource.
println!("\nPerforming dry-run for new resource: '{}'", name);
println!(
"Resource does not exist. It would be created with the following content:"
);
let new_yaml = serde_yaml::to_string(resource)
.unwrap_or_else(|_| "Failed to serialize new resource".to_string());
// Print each line of the new resource with a '+' prefix.
for line in new_yaml.lines() {
println!("+{}", line);
}
// In a dry run, we return the new resource state that would have been created.
Ok(resource.clone())
}
Err(e) => {
// Another API error occurred.
error!("Failed to get resource '{}': {}", name, e);
Err(e)
}
}
} else {
return api
.patch(name, &patch_params, &Patch::Apply(resource))
.await;
}
}
pub async fn apply_many<K>(&self, resource: &Vec<K>, ns: Option<&str>) -> Result<Vec<K>, Error>

33
harmony/src/domain/topology/k8s_anywhere.rs
View File

@@ -3,6 +3,7 @@ use std::{process::Command, sync::Arc};
use async_trait::async_trait;
use inquire::Confirm;
use log::{debug, info, warn};
use serde::Serialize;
use tokio::sync::OnceCell;
use crate::{
@@ -20,23 +21,25 @@ use super::{
tenant::{TenantConfig, TenantManager, k8s::K8sTenantManager},
};
#[derive(Clone, Debug)]
struct K8sState {
client: Arc<K8sClient>,
_source: K8sSource,
message: String,
}
#[derive(Debug)]
#[derive(Debug, Clone)]
enum K8sSource {
LocalK3d,
Kubeconfig,
}
#[derive(Clone, Debug)]
pub struct K8sAnywhereTopology {
k8s_state: OnceCell<Option<K8sState>>,
tenant_manager: OnceCell<K8sTenantManager>,
k8s_state: Arc<OnceCell<Option<K8sState>>>,
tenant_manager: Arc<OnceCell<K8sTenantManager>>,
config: Arc<K8sAnywhereConfig>,
tenant_manager_config: OnceCell<TenantConfig>,
config: K8sAnywhereConfig,
}
#[async_trait]
@@ -56,22 +59,31 @@ impl K8sclient for K8sAnywhereTopology {
}
}
impl Serialize for K8sAnywhereTopology {
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
where
S: serde::Serializer,
{
todo!()
}
}
impl K8sAnywhereTopology {
pub fn from_env() -> Self {
Self {
k8s_state: OnceCell::new(),
tenant_manager: OnceCell::new(),
k8s_state: Arc::new(OnceCell::new()),
tenant_manager: Arc::new(OnceCell::new()),
config: Arc::new(K8sAnywhereConfig::from_env()),
tenant_manager_config: OnceCell::new(),
config: K8sAnywhereConfig::from_env(),
}
}
pub fn with_config(config: K8sAnywhereConfig) -> Self {
Self {
k8s_state: OnceCell::new(),
tenant_manager: OnceCell::new(),
k8s_state: Arc::new(OnceCell::new()),
tenant_manager: Arc::new(OnceCell::new()),
config: Arc::new(config),
tenant_manager_config: OnceCell::new(),
config,
}
}
@@ -208,6 +220,7 @@ impl K8sAnywhereTopology {
}
}
#[derive(Clone, Debug)]
pub struct K8sAnywhereConfig {
/// The path of the KUBECONFIG file that Harmony should use to interact with the Kubernetes
/// cluster

28
harmony/src/domain/topology/tenant/k8s.rs
View File

@@ -22,7 +22,7 @@ use serde_json::json;
use super::{TenantConfig, TenantManager};
#[derive(new)]
#[derive(new, Clone, Debug)]
pub struct K8sTenantManager {
k8s_client: Arc<K8sClient>,
}
@@ -137,7 +137,8 @@ impl K8sTenantManager {
"apiVersion": "networking.k8s.io/v1",
"kind": "NetworkPolicy",
"metadata": {
"name": format!("{}-network-policy", config.name)
"name": format!("{}-network-policy", config.name),
"namespace": self.get_namespace_name(config),
},
"spec": {
"podSelector": {},
@@ -253,8 +254,29 @@ impl K8sTenantManager {
})
})
.collect();
let ports: Option<Vec<NetworkPolicyPort>> =
c.1.as_ref().map(|spec| match &spec.data {
super::PortSpecData::SinglePort(port) => vec![NetworkPolicyPort {
port: Some(IntOrString::Int(port.clone().into())),
..Default::default()
}],
super::PortSpecData::PortRange(start, end) => vec![NetworkPolicyPort {
port: Some(IntOrString::Int(start.clone().into())),
end_port: Some(end.clone().into()),
protocol: None, // Not currently supported by Harmony
}],
super::PortSpecData::ListOfPorts(items) => items
.iter()
.map(|i| NetworkPolicyPort {
port: Some(IntOrString::Int(i.clone().into())),
..Default::default()
})
.collect(),
});
let rule = serde_json::from_value::<NetworkPolicyIngressRule>(json!({
"from": cidr_list
"from": cidr_list,
"ports": ports,
}))
.map_err(|e| {
ExecutorError::ConfigurationError(format!(

42
harmony/src/modules/application/features/continuous_delivery.rs Normal file
View File

@@ -0,0 +1,42 @@
use async_trait::async_trait;
use log::info;
use crate::{modules::application::ApplicationFeature, topology::Topology};
/// ContinuousDelivery in Harmony provides this functionality :
///
/// - **Package** the application
/// - **Push** to an artifact registry
/// - **Deploy** to a testing environment
/// - **Deploy** to a production environment
///
/// It is intended to be used as an application feature passed down to an ApplicationInterpret. For
/// example :
///
/// ```rust,ignore
/// let app = RustApplicationScore {
/// name: "My Rust App".to_string(),
/// features: vec![ContinuousDelivery::default()],
/// };
/// ```
///
/// *Note :*
///
/// By default, the Harmony Opinionated Pipeline is built using these technologies :
///
/// - Gitea Action (executes pipeline steps)
/// - Docker to build an OCI container image
/// - Helm chart to package Kubernetes resources
/// - Harbor as artifact registru
/// - ArgoCD to install/upgrade/rollback/inspect k8s resources
/// - Kubernetes for runtime orchestration
#[derive(Debug, Default)]
pub struct ContinuousDelivery {}
#[async_trait]
impl<T: Topology + 'static> ApplicationFeature<T> for ContinuousDelivery {
async fn ensure_installed(&self, _topology: &T) -> Result<(), String> {
info!("Installing ContinuousDelivery feature");
todo!()
}
}

39
harmony/src/modules/application/features/endpoint.rs Normal file
View File

@@ -0,0 +1,39 @@
use async_trait::async_trait;
use log::info;
use crate::{
modules::application::ApplicationFeature,
topology::{K8sclient, Topology},
};
#[derive(Debug)]
pub struct PublicEndpoint {
application_port: u16,
}
/// Use port 3000 as default port. Harmony wants to provide "sane defaults" in general, and in this
/// particular context, using port 80 goes against our philosophy to provide production grade
/// defaults out of the box. Using an unprivileged port is a good security practice and will allow
/// for unprivileged containers to work with this out of the box.
///
/// Now, why 3000 specifically? Many popular web/network frameworks use it by default, there is no
/// perfect answer for this but many Rust and Python libraries tend to use 3000.
impl Default for PublicEndpoint {
fn default() -> Self {
Self {
application_port: 3000,
}
}
}
/// For now we only suport K8s ingress, but we will support more stuff at some point
#[async_trait]
impl<T: Topology + K8sclient + 'static> ApplicationFeature<T> for PublicEndpoint {
async fn ensure_installed(&self, _topology: &T) -> Result<(), String> {
info!(
"Making sure public endpoint is installed for port {}",
self.application_port
);
todo!()
}
}

8
harmony/src/modules/application/features/mod.rs Normal file
View File

@@ -0,0 +1,8 @@
mod endpoint;
pub use endpoint::*;
mod monitoring;
pub use monitoring::*;
mod continuous_delivery;
pub use continuous_delivery::*;

18
harmony/src/modules/application/features/monitoring.rs Normal file
View File

@@ -0,0 +1,18 @@
use async_trait::async_trait;
use log::info;
use crate::{
modules::application::ApplicationFeature,
topology::{HelmCommand, Topology},
};
#[derive(Debug, Default)]
pub struct Monitoring {}
#[async_trait]
impl<T: Topology + HelmCommand + 'static> ApplicationFeature<T> for Monitoring {
async fn ensure_installed(&self, _topology: &T) -> Result<(), String> {
info!("Ensuring monitoring is available for application");
todo!("create and execute k8s prometheus score, depends on Will's work")
}
}

67
harmony/src/modules/application/mod.rs Normal file
View File

@@ -0,0 +1,67 @@
pub mod features;
mod rust;
pub use rust::*;
use async_trait::async_trait;
use serde::Serialize;
use crate::{
data::{Id, Version},
interpret::{Interpret, InterpretError, InterpretName, InterpretStatus, Outcome},
inventory::Inventory,
topology::Topology,
};
#[derive(Debug)]
pub struct ApplicationInterpret<T: Topology + std::fmt::Debug> {
features: Vec<Box<dyn ApplicationFeature<T>>>,
}
#[async_trait]
impl<T: Topology + std::fmt::Debug> Interpret<T> for ApplicationInterpret<T> {
async fn execute(
&self,
_inventory: &Inventory,
_topology: &T,
) -> Result<Outcome, InterpretError> {
todo!()
}
fn get_name(&self) -> InterpretName {
todo!()
}
fn get_version(&self) -> Version {
todo!()
}
fn get_status(&self) -> InterpretStatus {
todo!()
}
fn get_children(&self) -> Vec<Id> {
todo!()
}
}
/// An ApplicationFeature provided by harmony, such as Backups, Monitoring, MultisiteAvailability,
/// ContinuousIntegration, ContinuousDelivery
#[async_trait]
pub trait ApplicationFeature<T: Topology>: std::fmt::Debug + Send + Sync {
async fn ensure_installed(&self, topology: &T) -> Result<(), String>;
}
impl<T: Topology> Serialize for Box<dyn ApplicationFeature<T>> {
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
where
S: serde::Serializer,
{
todo!()
}
}
impl<T: Topology> Clone for Box<dyn ApplicationFeature<T>> {
fn clone(&self) -> Self {
todo!()
}
}

25
harmony/src/modules/application/rust.rs Normal file
View File

@@ -0,0 +1,25 @@
use serde::Serialize;
use crate::{
score::Score,
topology::{Topology, Url},
};
use super::{ApplicationFeature, ApplicationInterpret};
#[derive(Debug, Serialize, Clone)]
pub struct RustWebappScore<T: Topology + Clone + Serialize> {
pub name: String,
pub domain: Url,
pub features: Vec<Box<dyn ApplicationFeature<T>>>,
}
impl<T: Topology + std::fmt::Debug + Clone + Serialize + 'static> Score<T> for RustWebappScore<T> {
fn create_interpret(&self) -> Box<dyn crate::interpret::Interpret<T>> {
Box::new(ApplicationInterpret { features: todo!() })
}
fn name(&self) -> String {
format!("{}-RustWebapp", self.name)
}
}

1
harmony/src/modules/mod.rs
View File

@@ -1,3 +1,4 @@
pub mod application;
pub mod cert_manager;
pub mod dhcp;
pub mod dns;

1
harmony/src/modules/monitoring/alert_channel/mod.rs
View File

@@ -1 +1,2 @@
pub mod discord_alert_channel;
pub mod webhook_receiver;

124
harmony/src/modules/monitoring/alert_channel/webhook_receiver.rs Normal file
View File

@@ -0,0 +1,124 @@
use async_trait::async_trait;
use serde::Serialize;
use serde_yaml::{Mapping, Value};
use crate::{
interpret::{InterpretError, Outcome},
modules::monitoring::kube_prometheus::{
prometheus::{Prometheus, PrometheusReceiver},
types::{AlertChannelConfig, AlertManagerChannelConfig},
},
topology::{Url, oberservability::monitoring::AlertReceiver},
};
#[derive(Debug, Clone, Serialize)]
pub struct WebhookReceiver {
pub name: String,
pub url: Url,
}
#[async_trait]
impl AlertReceiver<Prometheus> for WebhookReceiver {
async fn install(&self, sender: &Prometheus) -> Result<Outcome, InterpretError> {
sender.install_receiver(self).await
}
fn clone_box(&self) -> Box<dyn AlertReceiver<Prometheus>> {
Box::new(self.clone())
}
}
#[async_trait]
impl PrometheusReceiver for WebhookReceiver {
fn name(&self) -> String {
self.name.clone()
}
async fn configure_receiver(&self) -> AlertManagerChannelConfig {
self.get_config().await
}
}
#[async_trait]
impl AlertChannelConfig for WebhookReceiver {
async fn get_config(&self) -> AlertManagerChannelConfig {
let channel_global_config = None;
let channel_receiver = self.alert_channel_receiver().await;
let channel_route = self.alert_channel_route().await;
AlertManagerChannelConfig {
channel_global_config,
channel_receiver,
channel_route,
}
}
}
impl WebhookReceiver {
async fn alert_channel_route(&self) -> serde_yaml::Value {
let mut route = Mapping::new();
route.insert(
Value::String("receiver".to_string()),
Value::String(self.name.clone()),
);
route.insert(
Value::String("matchers".to_string()),
Value::Sequence(vec![Value::String("alertname!=Watchdog".to_string())]),
);
route.insert(Value::String("continue".to_string()), Value::Bool(true));
Value::Mapping(route)
}
async fn alert_channel_receiver(&self) -> serde_yaml::Value {
let mut receiver = Mapping::new();
receiver.insert(
Value::String("name".to_string()),
Value::String(self.name.clone()),
);
let mut webhook_config = Mapping::new();
webhook_config.insert(
Value::String("url".to_string()),
Value::String(self.url.to_string()),
);
receiver.insert(
Value::String("webhook_configs".to_string()),
Value::Sequence(vec![Value::Mapping(webhook_config)]),
);
Value::Mapping(receiver)
}
}
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
async fn webhook_serialize_should_match() {
let webhook_receiver = WebhookReceiver {
name: "test-webhook".to_string(),
url: Url::Url(url::Url::parse("https://webhook.i.dont.exist.com").unwrap()),
};
let webhook_receiver_receiver =
serde_yaml::to_string(&webhook_receiver.alert_channel_receiver().await).unwrap();
println!("receiver \n{:#}", webhook_receiver_receiver);
let webhook_receiver_receiver_yaml = r#"name: test-webhook
webhook_configs:
- url: https://webhook.i.dont.exist.com/
"#
.to_string();
let webhook_receiver_route =
serde_yaml::to_string(&webhook_receiver.alert_channel_route().await).unwrap();
println!("route \n{:#}", webhook_receiver_route);
let webhook_receiver_route_yaml = r#"receiver: test-webhook
matchers:
- alertname!=Watchdog
continue: true
"#
.to_string();
assert_eq!(webhook_receiver_receiver, webhook_receiver_receiver_yaml);
assert_eq!(webhook_receiver_route, webhook_receiver_route_yaml);
}
}