Compare commits

...

6 Commits

Author SHA1 Message Date
b5b77cf1ac fix: modified discordwebhookreceiver to use a generic initialize_alert_receiver function from k8sanywhere topology
All checks were successful
Run Check Script / check (push) Successful in 1m47s
Run Check Script / check (pull_request) Successful in 1m47s
2025-06-05 14:04:05 -04:00
dda8e29843 fix: removed unnecessary DiscordWebhookRecevierInerpret and its impl
All checks were successful
Run Check Script / check (push) Successful in 1m47s
Run Check Script / check (pull_request) Successful in 1m47s
2025-06-04 16:15:27 -04:00
b33650e9d5 fix: cargo fmt
All checks were successful
Run Check Script / check (push) Successful in 1m48s
Run Check Script / check (pull_request) Successful in 1m47s
2025-06-04 15:52:31 -04:00
ea4709a409 fix:binding with the ensure_ready impementation of K8sAnywhe
Some checks failed
Run Check Script / check (push) Failing after 45s
Run Check Script / check (pull_request) Failing after 43s
2025-06-04 15:47:48 -04:00
cd40660350 fix: Ensure idempotent deployment of Discord adapters
All checks were successful
Run Check Script / check (push) Successful in 1m49s
Run Check Script / check (pull_request) Successful in 1m46s
Updates  to use  for managing
Discord webhook adapter deployments. This prevents redundant installations
by ensuring that the deployment interpret for a given adapter
configuration is executed only once.

The internal state now uses a  keyed by the alertreceiver ID to
track and initialize each receiver's .
2025-06-03 16:07:11 -04:00
2ca732cecd feat: added the steps to install discord-webhook-receiver for k8s anywhere topology if not already installed
All checks were successful
Run Check Script / check (push) Successful in 1m45s
Run Check Script / check (pull_request) Successful in 1m48s
2025-06-03 14:14:41 -04:00
3 changed files with 149 additions and 26 deletions

View File

@ -1,10 +1,11 @@
use std::{process::Command, sync::Arc};
use std::{collections::HashMap, process::Command, sync::Arc};
use async_trait::async_trait;
use inquire::Confirm;
use log::{info, warn};
use tokio::sync::OnceCell;
use tokio::sync::{Mutex, OnceCell};
use crate::score::Score;
use crate::{
executors::ExecutorError,
interpret::{InterpretError, Outcome},
@ -17,6 +18,7 @@ use crate::{
use super::{
HelmCommand, K8sclient, Topology,
k8s::K8sClient,
oberservability::monitoring::{AlertReceiver, AlertReceiverProvision},
tenant::{
ResourceLimits, TenantConfig, TenantManager, TenantNetworkPolicy, k8s::K8sTenantManager,
},
@ -37,6 +39,7 @@ enum K8sSource {
pub struct K8sAnywhereTopology {
k8s_state: OnceCell<Option<K8sState>>,
tenant_manager: OnceCell<K8sTenantManager>,
pub alert_receivers: Mutex<HashMap<String, OnceCell<AlertReceiver>>>,
}
#[async_trait]
@ -61,9 +64,29 @@ impl K8sAnywhereTopology {
Self {
k8s_state: OnceCell::new(),
tenant_manager: OnceCell::new(),
alert_receivers: Mutex::new(HashMap::new()),
}
}
pub async fn initialize_alert_receiver<C>(
&self,
config: &C,
inventory: &Inventory,
) -> Result<AlertReceiver, InterpretError>
where
Self: Topology + HelmCommand,
C: AlertReceiverProvision<Self> + Send + Sync,
{
let score = config.get_deployment_score();
let interpret = score.create_interpret();
interpret.execute(inventory, self).await?;
Ok(AlertReceiver {
receiver_id: config.alert_receiver_id(),
receiver_installed: true,
})
}
fn is_helm_available(&self) -> Result<(), String> {
let version_result = Command::new("helm")
.arg("version")

View File

@ -1,14 +1,18 @@
use async_trait::async_trait;
use dyn_clone::DynClone;
use serde::Serialize;
use std::fmt::Debug;
use url::Url;
use crate::interpret::InterpretError;
use crate::inventory::Inventory;
use crate::score::Score;
use crate::topology::HelmCommand;
use crate::{interpret::Outcome, topology::Topology};
/// Represents an entity responsible for collecting and organizing observability data
/// from various telemetry sources
/// from various telemetry sources such as Prometheus or Datadog
/// A `Monitor` abstracts the logic required to scrape, aggregate, and structure
/// monitoring data, enabling consistent processing regardless of the underlying data source.
#[async_trait]
@ -26,6 +30,26 @@ pub trait Monitor<T: Topology>: Debug + Send + Sync {
) -> Result<Outcome, InterpretError>;
}
#[async_trait]
pub trait EnsureAlertReceiver<T: Topology>: Debug + DynClone + Send + Sync {
async fn ensure_alert_receiver(
&self,
inventory: Inventory,
topology: &T,
) -> Result<Outcome, InterpretError>;
}
dyn_clone::clone_trait_object!(<T> EnsureAlertReceiver<T>);
#[derive(Debug, Clone, Serialize)]
pub struct AlertReceiver {
pub receiver_id: String,
pub receiver_installed: bool,
}
/// Provides the ability to turn an alert config into an executable score
/// for the topology
pub trait AlertReceiverProvision<T: Topology + HelmCommand> {
fn get_deployment_score(&self) -> Box<dyn Score<T>>;
fn alert_receiver_id(&self) -> String;
}

View File

@ -1,32 +1,104 @@
use super::discord_alert_manager::discord_alert_manager_score;
use async_trait::async_trait;
use serde_json::Value;
use serde::Serialize;
use serde_yaml::Value;
use tokio::sync::OnceCell;
use url::Url;
use crate::{
interpret::{InterpretError, Outcome},
topology::K8sAnywhereTopology,
interpret::{Interpret, InterpretError, Outcome},
inventory::Inventory,
score::Score,
topology::{
HelmCommand, K8sAnywhereTopology, Topology,
oberservability::monitoring::{AlertReceiverProvision, EnsureAlertReceiver},
},
};
#[derive(Debug, Clone)]
#[derive(Debug, Clone, Serialize)]
pub struct DiscordWebhookConfig {
pub webhook_url: Url,
pub name: String,
pub send_resolved_notifications: bool,
}
pub trait DiscordWebhookReceiver {
fn deploy_discord_webhook_receiver(
#[async_trait]
impl<T: Topology + DiscordWebhookReceiver> EnsureAlertReceiver<T> for DiscordWebhookConfig {
async fn ensure_alert_receiver(
&self,
_notification_adapter_id: &str,
) -> Result<Outcome, InterpretError>;
inventory: Inventory,
topology: &T,
) -> Result<Outcome, InterpretError> {
topology
.ensure_discord_webhook_receiver(&inventory, self.clone())
.await
}
}
impl<T: Topology + HelmCommand> AlertReceiverProvision<T> for DiscordWebhookConfig {
fn get_deployment_score(&self) -> Box<dyn Score<T>> {
Box::new(DiscordWebhookReceiverScore {
config: self.clone(),
})
}
fn alert_receiver_id(&self) -> String {
self.name.clone()
}
}
#[async_trait]
pub trait DiscordWebhookReceiver {
async fn ensure_discord_webhook_receiver(
&self,
inventory: &Inventory,
config: DiscordWebhookConfig,
) -> Result<Outcome, InterpretError>;
fn delete_discord_webhook_receiver(
&self,
_notification_adapter_id: &str,
config: DiscordWebhookConfig,
) -> Result<Outcome, InterpretError>;
}
// trait used to generate alert manager config values impl<T: Topology + AlertManagerConfig> Monitor for KubePrometheus
#[async_trait]
impl DiscordWebhookReceiver for K8sAnywhereTopology {
async fn ensure_discord_webhook_receiver(
&self,
inventory: &Inventory,
config: DiscordWebhookConfig,
) -> Result<Outcome, InterpretError> {
let receiver_key = config.name.clone();
let mut adapters_map_guard = self.alert_receivers.lock().await;
let cell = adapters_map_guard
.entry(receiver_key.clone())
.or_insert_with(OnceCell::new);
if let Some(initialized_receiver) = cell.get() {
return Ok(Outcome::success(format!(
"Discord Webhook adapter for '{}' already initialized.",
initialized_receiver.receiver_id
)));
}
let final_state = cell
.get_or_try_init(|| async { self.initialize_alert_receiver(&config, inventory).await })
.await?;
Ok(Outcome::success(format!(
"Discord Webhook Receiver for '{}' ensured/initialized.",
final_state.receiver_id
)))
}
fn delete_discord_webhook_receiver(
&self,
_config: DiscordWebhookConfig,
) -> Result<Outcome, InterpretError> {
todo!()
}
}
pub trait AlertManagerConfig<T> {
fn get_alert_manager_config(&self) -> Result<Value, InterpretError>;
}
@ -38,18 +110,22 @@ impl<T: DiscordWebhookReceiver> AlertManagerConfig<T> for DiscordWebhookConfig {
}
}
#[async_trait]
impl DiscordWebhookReceiver for K8sAnywhereTopology {
fn deploy_discord_webhook_receiver(
&self,
_notification_adapter_id: &str,
) -> Result<Outcome, InterpretError> {
todo!()
#[derive(Debug, Clone, Serialize)]
struct DiscordWebhookReceiverScore {
config: DiscordWebhookConfig,
}
impl<T: Topology + HelmCommand> Score<T> for DiscordWebhookReceiverScore {
fn create_interpret(&self) -> Box<dyn Interpret<T>> {
discord_alert_manager_score(
self.config.webhook_url.clone(),
self.config.name.clone(),
self.config.name.clone(),
)
.create_interpret()
}
fn delete_discord_webhook_receiver(
&self,
_notification_adapter_id: &str,
) -> Result<Outcome, InterpretError> {
todo!()
fn name(&self) -> String {
"DiscordWebhookReceiverScore".to_string()
}
}