fix: K8sTenantManager is responsible for concrete implementation. K8sAnywhere should delegate

Reviewed-on: https://git.nationtech.io/NationTech/harmony/pulls/75
Reviewed-by: taha <taha@noreply.git.nationtech.io>

examples/monitoring/src/main.rs

@@ -7,7 +7,13 @@ use harmony::{
         monitoring::{
             alert_channel::discord_alert_channel::DiscordWebhook,
             alert_rule::prometheus_alert_rule::AlertManagerRuleGroup,
-            kube_prometheus::helm_prometheus_alert_score::HelmPrometheusAlertingScore,
+            kube_prometheus::{
+                helm_prometheus_alert_score::HelmPrometheusAlertingScore,
+                types::{
+                    HTTPScheme, MatchExpression, Operator, Selector, ServiceMonitor,
+                    ServiceMonitorEndpoint,
+                },
+            },
         },
         prometheus::alerts::{
             infra::dell_server::{

examples/monitoring_with_tenant/src/main.rs

@@ -1,3 +1,5 @@
+use std::collections::HashMap;
+
 use harmony::{
     data::Id,
     inventory::Inventory,
@@ -6,7 +8,13 @@ use harmony::{
         monitoring::{
             alert_channel::discord_alert_channel::DiscordWebhook,
             alert_rule::prometheus_alert_rule::AlertManagerRuleGroup,
-            kube_prometheus::helm_prometheus_alert_score::HelmPrometheusAlertingScore,
+            kube_prometheus::{
+                helm_prometheus_alert_score::HelmPrometheusAlertingScore,
+                types::{
+                    HTTPScheme, MatchExpression, Operator, Selector, ServiceMonitor,
+                    ServiceMonitorEndpoint,
+                },
+            },
         },
         prometheus::alerts::k8s::pvc::high_pvc_fill_rate_over_two_days,
         tenant::TenantScore,
@@ -44,9 +52,31 @@ async fn main() {
     let additional_rules =
         AlertManagerRuleGroup::new("pvc-alerts", vec![high_pvc_fill_rate_over_two_days_alert]);
 
+    let service_monitor_endpoint = ServiceMonitorEndpoint {
+        port: Some("80".to_string()),
+        path: "/metrics".to_string(),
+        scheme: HTTPScheme::HTTP,
+        ..Default::default()
+    };
+
+    let service_monitor = ServiceMonitor {
+        name: "test-service-monitor".to_string(),
+        selector: Selector {
+            match_labels: HashMap::new(),
+            match_expressions: vec![MatchExpression {
+                key: "test".to_string(),
+                operator: Operator::In,
+                values: vec!["test-service".to_string()],
+            }],
+        },
+        endpoints: vec![service_monitor_endpoint],
+        ..Default::default()
+    };
+
     let alerting_score = HelmPrometheusAlertingScore {
         receivers: vec![Box::new(discord_receiver)],
         rules: vec![Box::new(additional_rules)],
+        service_monitors: vec![service_monitor],
     };
     let mut maestro = Maestro::<K8sAnywhereTopology>::initialize(
         Inventory::autoload(),

harmony/src/domain/topology/installable.rs

@@ -4,7 +4,7 @@ use crate::{interpret::InterpretError, inventory::Inventory};
 
 #[async_trait]
 pub trait Installable<T>: Send + Sync {
-    fn configure(&self, inventory: &Inventory, topology: &T) -> Result<(), InterpretError>;
+    async fn configure(&self, inventory: &Inventory, topology: &T) -> Result<(), InterpretError>;
 
     async fn ensure_installed(
         &self,

harmony/src/domain/topology/k8s_anywhere.rs

@@ -39,7 +39,6 @@ pub struct K8sAnywhereTopology {
     k8s_state: Arc<OnceCell<Option<K8sState>>>,
     tenant_manager: Arc<OnceCell<K8sTenantManager>>,
     config: Arc<K8sAnywhereConfig>,
-    tenant_manager_config: OnceCell<TenantConfig>,
 }
 
 #[async_trait]
@@ -74,7 +73,6 @@ impl K8sAnywhereTopology {
             k8s_state: Arc::new(OnceCell::new()),
             tenant_manager: Arc::new(OnceCell::new()),
             config: Arc::new(K8sAnywhereConfig::from_env()),
-            tenant_manager_config: OnceCell::new(),
         }
     }
 
@@ -83,7 +81,6 @@ impl K8sAnywhereTopology {
             k8s_state: Arc::new(OnceCell::new()),
             tenant_manager: Arc::new(OnceCell::new()),
             config: Arc::new(config),
-            tenant_manager_config: OnceCell::new(),
         }
     }
 
@@ -199,16 +196,10 @@ impl K8sAnywhereTopology {
                 let k8s_client = self.k8s_client().await?;
                 Ok(K8sTenantManager::new(k8s_client))
             })
-            .await
-            .unwrap();
+            .await?;
 
         Ok(())
     }
-    async fn store_tenant_config(&self, config: TenantConfig) {
-        self.tenant_manager_config
-            .get_or_init(|| async { config })
-            .await;
-    }
 
     fn get_k8s_tenant_manager(&self) -> Result<&K8sTenantManager, ExecutorError> {
         match self.tenant_manager.get() {
@@ -289,13 +280,15 @@ impl HelmCommand for K8sAnywhereTopology {}
 #[async_trait]
 impl TenantManager for K8sAnywhereTopology {
     async fn provision_tenant(&self, config: &TenantConfig) -> Result<(), ExecutorError> {
-        self.store_tenant_config(config.clone()).await;
         self.get_k8s_tenant_manager()?
             .provision_tenant(config)
             .await
     }
 
-    fn get_tenant_config(&self) -> Option<TenantConfig> {
-        self.tenant_manager_config.get().cloned()
+    async fn get_tenant_config(&self) -> Option<TenantConfig> {
+        self.get_k8s_tenant_manager()
+            .ok()?
+            .get_tenant_config()
+            .await
     }
 }

harmony/src/domain/topology/oberservability/monitoring.rs

@@ -27,7 +27,7 @@ impl<S: AlertSender + Installable<T>, T: Topology> Interpret<T> for AlertingInte
         inventory: &Inventory,
         topology: &T,
     ) -> Result<Outcome, InterpretError> {
-        self.sender.configure(inventory, topology)?;
+        self.sender.configure(inventory, topology).await?;
         for receiver in self.receivers.iter() {
             receiver.install(&self.sender).await?;
         }

harmony/src/domain/topology/tenant/k8s.rs

@@ -5,10 +5,9 @@ use crate::{
     topology::k8s::{ApplyStrategy, K8sClient},
 };
 use async_trait::async_trait;
-use derive_new::new;
 use k8s_openapi::{
     api::{
-        core::v1::{Namespace, ResourceQuota},
+        core::v1::{LimitRange, Namespace, ResourceQuota},
         networking::v1::{
             NetworkPolicy, NetworkPolicyEgressRule, NetworkPolicyIngressRule, NetworkPolicyPort,
         },
@@ -19,12 +18,23 @@ use kube::Resource;
 use log::{debug, info, warn};
 use serde::de::DeserializeOwned;
 use serde_json::json;
+use tokio::sync::OnceCell;
 
 use super::{TenantConfig, TenantManager};
 
-#[derive(new, Clone, Debug)]
+#[derive(Clone, Debug)]
 pub struct K8sTenantManager {
     k8s_client: Arc<K8sClient>,
+    k8s_tenant_config: Arc<OnceCell<TenantConfig>>,
+}
+
+impl K8sTenantManager {
+    pub fn new(client: Arc<K8sClient>) -> Self {
+        Self {
+            k8s_client: client,
+            k8s_tenant_config: Arc::new(OnceCell::new()),
+        }
+    }
 }
 
 impl K8sTenantManager {
@@ -132,6 +142,43 @@ impl K8sTenantManager {
         })
     }
 
+    fn build_limit_range(&self, config: &TenantConfig) -> Result<LimitRange, ExecutorError> {
+        let limit_range = json!({
+          "apiVersion": "v1",
+          "kind": "LimitRange",
+          "metadata": {
+            "name": format!("{}-defaults", config.name),
+            "labels": {
+              "harmony.nationtech.io/tenant.id": config.id.to_string(),
+              "harmony.nationtech.io/tenant.name": config.name,
+            },
+            "namespace": self.get_namespace_name(config),
+          },
+          "spec": {
+            "limits": [
+              {
+                "type": "Container",
+                "default": {
+                  "cpu": "500m",
+                  "memory": "500Mi"
+                },
+                "defaultRequest": {
+                  "cpu": "100m",
+                  "memory": "100Mi"
+                },
+              }
+            ]
+          }
+        });
+
+        serde_json::from_value(limit_range).map_err(|e| {
+            ExecutorError::ConfigurationError(format!(
+                "Could not build TenantManager LimitRange. {}",
+                e
+            ))
+        })
+    }
+
     fn build_network_policy(&self, config: &TenantConfig) -> Result<NetworkPolicy, ExecutorError> {
         let network_policy = json!({
           "apiVersion": "networking.k8s.io/v1",
@@ -354,6 +401,9 @@ impl K8sTenantManager {
 
         Ok(network_policy)
     }
+    fn store_config(&self, config: &TenantConfig) {
+        let _ = self.k8s_tenant_config.set(config.clone());
+    }
 }
 
 #[async_trait]
@@ -362,6 +412,7 @@ impl TenantManager for K8sTenantManager {
         let namespace = self.build_namespace(config)?;
         let resource_quota = self.build_resource_quota(config)?;
         let network_policy = self.build_network_policy(config)?;
+        let resource_limit_range = self.build_limit_range(config)?;
 
         self.ensure_constraints(&namespace)?;
 
@@ -371,6 +422,9 @@ impl TenantManager for K8sTenantManager {
         debug!("Creating resource_quota for tenant {}", config.name);
         self.apply_resource(resource_quota, config).await?;
 
+        debug!("Creating limit_range for tenant {}", config.name);
+        self.apply_resource(resource_limit_range, config).await?;
+
         debug!("Creating network_policy for tenant {}", config.name);
         self.apply_resource(network_policy, config).await?;
 
@@ -378,9 +432,10 @@ impl TenantManager for K8sTenantManager {
             "Success provisionning K8s tenant id {} name {}",
             config.id, config.name
         );
+        self.store_config(config);
         Ok(())
     }
-    fn get_tenant_config(&self) -> Option<TenantConfig> {
-        todo!()
+    async fn get_tenant_config(&self) -> Option<TenantConfig> {
+        self.k8s_tenant_config.get().cloned()
     }
 }

harmony/src/domain/topology/tenant/manager.rs

@@ -16,5 +16,5 @@ pub trait TenantManager {
     /// * `config`: The desired configuration for the new tenant.
     async fn provision_tenant(&self, config: &TenantConfig) -> Result<(), ExecutorError>;
 
-    fn get_tenant_config(&self) -> Option<TenantConfig>;
+    async fn get_tenant_config(&self) -> Option<TenantConfig>;
 }

harmony/src/modules/application/feature.rs

@@ -0,0 +1,41 @@
+use async_trait::async_trait;
+use serde::Serialize;
+
+use crate::topology::Topology;
+/// An ApplicationFeature provided by harmony, such as Backups, Monitoring, MultisiteAvailability,
+/// ContinuousIntegration, ContinuousDelivery
+#[async_trait]
+pub trait ApplicationFeature<T: Topology>:
+    std::fmt::Debug + Send + Sync + ApplicationFeatureClone<T>
+{
+    async fn ensure_installed(&self, topology: &T) -> Result<(), String>;
+    fn name(&self) -> String;
+}
+
+trait ApplicationFeatureClone<T: Topology> {
+    fn clone_box(&self) -> Box<dyn ApplicationFeature<T>>;
+}
+
+impl<A, T: Topology> ApplicationFeatureClone<T> for A
+where
+    A: ApplicationFeature<T> + Clone + 'static,
+{
+    fn clone_box(&self) -> Box<dyn ApplicationFeature<T>> {
+        Box::new(self.clone())
+    }
+}
+
+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 {
+        self.clone_box()
+    }
+}

harmony/src/modules/monitoring/kube_prometheus/helm/kube_prometheus_helm_chart.rs

@@ -11,7 +11,9 @@ use std::{
 use crate::modules::{
     helm::chart::HelmChartScore,
     monitoring::kube_prometheus::types::{
-        AlertGroup, AlertManager, AlertManagerAdditionalPromRules, AlertManagerConfig, AlertManagerRoute, AlertManagerSpec, AlertManagerValues, ConfigReloader, Limits, PrometheusConfig, Requests, Resources
+        AlertGroup, AlertManager, AlertManagerAdditionalPromRules, AlertManagerConfig,
+        AlertManagerRoute, AlertManagerSpec, AlertManagerValues, ConfigReloader, Limits,
+        PrometheusConfig, Requests, Resources,
     },
 };
 

harmony/src/modules/monitoring/kube_prometheus/prometheus.rs

@@ -35,8 +35,8 @@ impl AlertSender for Prometheus {
 
 #[async_trait]
 impl<T: Topology + HelmCommand + TenantManager> Installable<T> for Prometheus {
-    fn configure(&self, _inventory: &Inventory, topology: &T) -> Result<(), InterpretError> {
-        self.configure_with_topology(topology);
+    async fn configure(&self, _inventory: &Inventory, topology: &T) -> Result<(), InterpretError> {
+        self.configure_with_topology(topology).await;
         Ok(())
     }
 
@@ -62,9 +62,10 @@ impl Prometheus {
         }
     }
 
-    pub fn configure_with_topology<T: TenantManager>(&self, topology: &T) {
+    pub async fn configure_with_topology<T: TenantManager>(&self, topology: &T) {
         let ns = topology
             .get_tenant_config()
+            .await
             .map(|cfg| cfg.name.clone())
             .unwrap_or_else(|| "monitoring".to_string());
         error!("This must be refactored, see comments in pr #74");