use std::sync::Arc; use std::time::Duration; use k8s_openapi::api::apps::v1::Deployment; use k8s_openapi::api::core::v1::Service; use k8s_crds_cert_manager::certificates::Certificate; use kube::api::{Api, DynamicObject, Patch, PatchParams}; use kube::core::{ApiResource, GroupVersionKind}; use kube::runtime::controller::Action; use crate::resources::certificate::build_web_certificate; use crate::resources::deployment::build_web_deployment; use crate::resources::route::build_web_route; use crate::resources::service::build_web_service; use crate::types::webservice::{Condition, WebService, WebServiceStatus}; #[derive(Debug, thiserror::Error)] pub enum Error { #[error("Kube error: {0}")] Kube(#[source] kube::Error), #[error("Missing field: {0}")] MissingField(String), #[error("Serialization error: {0}")] Serialization(#[source] serde_json::Error), } pub struct Context { pub client: kube::Client, } static FIELD_MANAGER: &str = "platform-operator"; pub async fn reconcile( ws: Arc, ctx: Arc, ) -> Result { let result = reconcile_apply(&ws, &ctx).await; if let Err(ref e) = result { let _ = update_status(&ctx.client, &ws, false, &e.to_string()).await; } result } async fn reconcile_apply(ws: &WebService, ctx: &Context) -> Result { let client = &ctx.client; let ns = ws .metadata .namespace .as_deref() .ok_or_else(|| Error::MissingField("metadata.namespace".to_string()))?; let name = ws .metadata .name .as_deref() .ok_or_else(|| Error::MissingField("metadata.name".to_string()))?; tracing::info!(webservice = name, namespace = ns, "reconciling"); if ws.spec.tls.is_some() { let cert = build_web_certificate(ws); let cert_api: Api = Api::namespaced(client.clone(), ns); let cert_name = cert.metadata.name.as_deref().unwrap(); cert_api .patch( cert_name, &PatchParams::apply(FIELD_MANAGER).force(), &Patch::Apply(&cert), ) .await .map_err(Error::Kube)?; tracing::info!(webservice = name, "applied certificate"); } let dep = build_web_deployment(ws); let dep_api: Api = Api::namespaced(client.clone(), ns); dep_api .patch( name, &PatchParams::apply(FIELD_MANAGER).force(), &Patch::Apply(&dep), ) .await .map_err(Error::Kube)?; tracing::info!(webservice = name, "applied deployment"); let svc = build_web_service(ws); let svc_api: Api = Api::namespaced(client.clone(), ns); svc_api .patch( name, &PatchParams::apply(FIELD_MANAGER).force(), &Patch::Apply(&svc), ) .await .map_err(Error::Kube)?; tracing::info!(webservice = name, "applied service"); if ws.spec.tls.is_some() { let route = build_web_route(ws); let gvk = GroupVersionKind::gvk("route.openshift.io", "v1", "Route"); let ar = ApiResource::from_gvk_with_plural(&gvk, "routes"); let route_api: Api = Api::namespaced_with(client.clone(), ns, &ar); let patch_value = serde_json::json!({ "apiVersion": "route.openshift.io/v1", "kind": "Route", "metadata": route.metadata, "spec": route.spec, }); route_api .patch( name, &PatchParams::apply(FIELD_MANAGER).force(), &Patch::Apply(&patch_value), ) .await .map_err(Error::Kube)?; tracing::info!(webservice = name, "applied route"); } update_status(client, ws, true, "all resources applied").await?; tracing::info!(webservice = name, "reconciled"); Ok(Action::requeue(Duration::from_secs(300))) } pub fn error_policy( _ws: Arc, error: &Error, _ctx: Arc, ) -> Action { tracing::warn!(%error, "reconcile failed"); Action::requeue(Duration::from_secs(60)) } async fn update_status( client: &kube::Client, ws: &WebService, ready: bool, message: &str, ) -> Result<(), Error> { let ns = ws.metadata.namespace.as_deref().unwrap_or("default"); let api = Api::::namespaced(client.clone(), ns); let name = ws.metadata.name.as_deref().unwrap(); let now = chrono::Utc::now().to_rfc3339(); let status = WebServiceStatus { conditions: vec![Condition { type_: "Ready".to_string(), status: if ready { "True" } else { "False" }.to_string(), reason: if ready { "Reconciled" } else { "Error" }.to_string(), message: message.to_string(), last_transition_time: now, }], observed_generation: ws.metadata.generation, }; let patch = serde_json::json!({ "apiVersion": "irc.josie.cloud/v1alpha1", "kind": "WebService", "status": status, }); api.patch_status(name, &PatchParams::apply(FIELD_MANAGER).force(), &Patch::Apply(&patch)) .await .map_err(Error::Kube)?; Ok(()) }