1use crate::constants::{API_GROUP_VERSION, KIND_BIND9_CLUSTER, KIND_CLUSTER_BIND9_PROVIDER};
16use crate::context::Context;
17use crate::crd::{
18 Bind9Cluster, Bind9ClusterStatus, Bind9Instance, ClusterBind9Provider, Condition,
19};
20use crate::labels::FINALIZER_BIND9_CLUSTER;
21use crate::reconcilers::finalizers::{
22 ensure_cluster_finalizer, handle_cluster_deletion, FinalizerCleanup,
23};
24use crate::status_reasons::{
25 CONDITION_TYPE_READY, REASON_ALL_READY, REASON_NOT_READY, REASON_NO_CHILDREN,
26 REASON_PARTIALLY_READY,
27};
28use anyhow::Result;
29use chrono::Utc;
30use kube::{
31 api::{ListParams, Patch, PatchParams},
32 client::Client,
33 Api, ResourceExt,
34};
35use serde_json::json;
36use std::sync::Arc;
37use tracing::{debug, error, info, warn};
38
39#[async_trait::async_trait]
44impl FinalizerCleanup for ClusterBind9Provider {
45 async fn cleanup(&self, client: &Client) -> Result<()> {
46 use crate::labels::{
47 BINDY_CLUSTER_LABEL, BINDY_MANAGED_BY_LABEL, MANAGED_BY_CLUSTER_BIND9_PROVIDER,
48 };
49 use kube::api::DeleteParams;
50
51 let name = self.name_any();
52
53 info!(
55 "Deleting managed Bind9Cluster resources for global cluster {}",
56 name
57 );
58
59 let clusters_api: Api<Bind9Cluster> = Api::all(client.clone());
60 let all_clusters = clusters_api.list(&ListParams::default()).await?;
61
62 let managed_clusters: Vec<_> = all_clusters
64 .items
65 .iter()
66 .filter(|c| {
67 c.metadata.labels.as_ref().is_some_and(|labels| {
68 labels.get(BINDY_MANAGED_BY_LABEL)
69 == Some(&MANAGED_BY_CLUSTER_BIND9_PROVIDER.to_string())
70 && labels.get(BINDY_CLUSTER_LABEL) == Some(&name.clone())
71 })
72 })
73 .collect();
74
75 if !managed_clusters.is_empty() {
76 info!(
77 "Found {} managed Bind9Cluster resources to delete for global cluster {}",
78 managed_clusters.len(),
79 name
80 );
81
82 for managed_cluster in managed_clusters {
83 let cluster_name = managed_cluster.name_any();
84 let cluster_namespace = managed_cluster.namespace().unwrap_or_default();
85
86 info!(
87 "Deleting managed Bind9Cluster {}/{} for global cluster {}",
88 cluster_namespace, cluster_name, name
89 );
90
91 let api: Api<Bind9Cluster> = Api::namespaced(client.clone(), &cluster_namespace);
92 match api.delete(&cluster_name, &DeleteParams::default()).await {
93 Ok(_) => {
94 info!(
95 "Successfully deleted Bind9Cluster {}/{}",
96 cluster_namespace, cluster_name
97 );
98 }
99 Err(e) => {
100 if e.to_string().contains("NotFound") {
102 debug!(
103 "Bind9Cluster {}/{} already deleted",
104 cluster_namespace, cluster_name
105 );
106 } else {
107 error!(
108 "Failed to delete Bind9Cluster {}/{}: {}",
109 cluster_namespace, cluster_name, e
110 );
111 return Err(e.into());
112 }
113 }
114 }
115 }
116 }
117
118 let instances_api: Api<Bind9Instance> = Api::all(client.clone());
121 let instances = instances_api.list(&ListParams::default()).await?;
122
123 let referencing_instances: Vec<_> = instances
124 .items
125 .iter()
126 .filter(|inst| inst.spec.cluster_ref == name)
127 .collect();
128
129 if !referencing_instances.is_empty() {
130 warn!(
131 "ClusterBind9Provider {} still has {} referencing instances. \
132 These will be cleaned up by their parent Bind9Cluster finalizers.",
133 name,
134 referencing_instances.len()
135 );
136 }
137
138 Ok(())
139 }
140}
141
142pub async fn reconcile_clusterbind9provider(
164 ctx: Arc<Context>,
165 cluster: ClusterBind9Provider,
166) -> Result<()> {
167 let client = ctx.client.clone();
168 let name = cluster.name_any();
169
170 info!("Reconciling ClusterBind9Provider: {}", name);
171 debug!(
172 name = %name,
173 generation = ?cluster.metadata.generation,
174 "Starting ClusterBind9Provider reconciliation (cluster-scoped)"
175 );
176
177 if cluster.metadata.deletion_timestamp.is_some() {
179 return handle_cluster_deletion(&client, &cluster, FINALIZER_BIND9_CLUSTER).await;
180 }
181
182 ensure_cluster_finalizer(&client, &cluster, FINALIZER_BIND9_CLUSTER).await?;
184
185 let current_generation = cluster.metadata.generation;
187 let observed_generation = cluster.status.as_ref().and_then(|s| s.observed_generation);
188
189 let spec_changed =
191 crate::reconcilers::should_reconcile(current_generation, observed_generation);
192
193 let drift_detected = if spec_changed {
195 false
196 } else {
197 detect_cluster_drift(&client, &cluster).await?
198 };
199
200 if spec_changed || drift_detected {
201 if drift_detected {
202 info!(
203 "Spec unchanged but cluster drift detected for ClusterBind9Provider {}",
204 name
205 );
206 } else {
207 debug!(
208 "Reconciliation needed: current_generation={:?}, observed_generation={:?}",
209 current_generation, observed_generation
210 );
211 }
212
213 reconcile_namespace_clusters(&client, &cluster).await?;
217 } else {
218 debug!(
219 "Spec unchanged (generation={:?}) and no drift detected, skipping resource reconciliation",
220 current_generation
221 );
222 }
223
224 update_cluster_status(&client, &cluster).await?;
226
227 Ok(())
228}
229
230fn provider_target_namespace(cluster_provider: &ClusterBind9Provider) -> String {
235 cluster_provider.spec.namespace.as_ref().map_or_else(
236 || std::env::var("POD_NAMESPACE").unwrap_or_else(|_| "bindy-system".to_string()),
237 std::clone::Clone::clone,
238 )
239}
240
241#[must_use]
257pub fn expected_cluster_namespaces(
258 instances: &[Bind9Instance],
259 provider_name: &str,
260 target_namespace: &str,
261) -> std::collections::HashSet<String> {
262 let namespaces: std::collections::HashSet<String> = instances
263 .iter()
264 .filter(|inst| inst.spec.cluster_ref == provider_name)
265 .filter_map(kube::ResourceExt::namespace)
266 .collect();
267
268 if namespaces.is_empty() {
269 return std::iter::once(target_namespace.to_string()).collect();
270 }
271
272 namespaces
273}
274
275async fn compute_expected_cluster_namespaces(
283 client: &Client,
284 cluster_provider: &ClusterBind9Provider,
285) -> Result<std::collections::HashSet<String>> {
286 let cluster_provider_name = cluster_provider.name_any();
287 let target_namespace = provider_target_namespace(cluster_provider);
288
289 let instances_api: Api<Bind9Instance> = Api::all(client.clone());
290 let all_instances = instances_api.list(&ListParams::default()).await?;
291
292 Ok(expected_cluster_namespaces(
293 &all_instances.items,
294 &cluster_provider_name,
295 &target_namespace,
296 ))
297}
298
299#[allow(clippy::too_many_lines)]
314async fn reconcile_namespace_clusters(
315 client: &Client,
316 cluster_provider: &ClusterBind9Provider,
317) -> Result<()> {
318 use crate::crd::{Bind9Cluster, Bind9ClusterSpec};
319 use crate::labels::{
320 BINDY_CLUSTER_LABEL, BINDY_MANAGED_BY_LABEL, MANAGED_BY_CLUSTER_BIND9_PROVIDER,
321 };
322 use kube::api::PostParams;
323 use std::collections::BTreeMap;
324
325 let cluster_provider_name = cluster_provider.name_any();
326
327 debug!(
328 "Reconciling namespace-scoped Bind9Cluster resources for global cluster {}",
329 cluster_provider_name
330 );
331
332 let namespaces_to_reconcile =
335 compute_expected_cluster_namespaces(client, cluster_provider).await?;
336
337 debug!(
338 "Found {} namespace(s) needing Bind9Cluster for global cluster {}",
339 namespaces_to_reconcile.len(),
340 cluster_provider_name
341 );
342
343 for namespace in namespaces_to_reconcile {
345 let cluster_name = cluster_provider_name.clone();
347
348 info!(
349 "Creating/updating Bind9Cluster {}/{} for global cluster {}",
350 namespace, cluster_name, cluster_provider_name
351 );
352
353 let mut labels = BTreeMap::new();
355 labels.insert(
356 BINDY_MANAGED_BY_LABEL.to_string(),
357 MANAGED_BY_CLUSTER_BIND9_PROVIDER.to_string(),
358 );
359 labels.insert(
360 BINDY_CLUSTER_LABEL.to_string(),
361 cluster_provider_name.clone(),
362 );
363
364 let owner_ref = k8s_openapi::apimachinery::pkg::apis::meta::v1::OwnerReference {
366 api_version: API_GROUP_VERSION.to_string(),
367 kind: KIND_CLUSTER_BIND9_PROVIDER.to_string(),
368 name: cluster_provider_name.clone(),
369 uid: cluster_provider.metadata.uid.clone().unwrap_or_default(),
370 controller: Some(true),
371 block_owner_deletion: Some(true),
372 };
373
374 let cluster_spec = Bind9ClusterSpec {
376 common: cluster_provider.spec.common.clone(),
377 };
378
379 let cluster = Bind9Cluster {
380 metadata: kube::api::ObjectMeta {
381 name: Some(cluster_name.clone()),
382 namespace: Some(namespace.clone()),
383 labels: Some(labels),
384 owner_references: Some(vec![owner_ref]),
385 ..Default::default()
386 },
387 spec: cluster_spec,
388 status: None,
389 };
390
391 let api: Api<Bind9Cluster> = Api::namespaced(client.clone(), &namespace);
392
393 match api.create(&PostParams::default(), &cluster).await {
395 Ok(_) => {
396 info!(
397 "Successfully created Bind9Cluster {}/{}",
398 namespace, cluster_name
399 );
400 }
401 Err(e) => {
402 if e.to_string().contains("AlreadyExists") {
404 debug!(
405 "Bind9Cluster {}/{} already exists, patching with updated spec",
406 namespace, cluster_name
407 );
408
409 let patch = serde_json::json!({
411 "apiVersion": API_GROUP_VERSION,
412 "kind": KIND_BIND9_CLUSTER,
413 "metadata": {
414 "name": cluster_name,
415 "namespace": namespace,
416 "ownerReferences": cluster.metadata.owner_references,
417 },
418 "spec": cluster.spec,
419 });
420
421 match api
423 .patch(
424 &cluster_name,
425 &PatchParams::apply("bindy-controller").force(),
426 &Patch::Apply(&patch),
427 )
428 .await
429 {
430 Ok(_) => {
431 info!(
432 "Successfully patched Bind9Cluster {}/{} with updated spec",
433 namespace, cluster_name
434 );
435 }
436 Err(patch_err) => {
437 warn!(
438 "Failed to patch Bind9Cluster {}/{}: {}",
439 namespace, cluster_name, patch_err
440 );
441 return Err(patch_err.into());
442 }
443 }
444 } else {
445 warn!(
446 "Failed to create Bind9Cluster {}/{}: {}",
447 namespace, cluster_name, e
448 );
449 return Err(e.into());
450 }
451 }
452 }
453 }
454
455 Ok(())
456}
457
458async fn update_cluster_status(client: &Client, cluster: &ClusterBind9Provider) -> Result<()> {
464 let name = cluster.name_any();
465
466 let instances_api: Api<Bind9Instance> = Api::all(client.clone());
468 let lp = ListParams::default();
469 let all_instances = instances_api.list(&lp).await?;
470
471 let instances: Vec<_> = all_instances
473 .items
474 .into_iter()
475 .filter(|inst| inst.spec.cluster_ref == name)
476 .collect();
477
478 debug!(
479 "Found {} instances referencing ClusterBind9Provider {}",
480 instances.len(),
481 name
482 );
483
484 let new_status = calculate_cluster_status(&instances, cluster.metadata.generation);
486
487 let status_changed = cluster_status_needs_update(cluster.status.as_ref(), &new_status);
489
490 if !status_changed {
492 debug!(
493 "Status unchanged for ClusterBind9Provider {}, skipping patch",
494 name
495 );
496 return Ok(());
497 }
498
499 let api: Api<ClusterBind9Provider> = Api::all(client.clone());
501 let status_patch = json!({
502 "status": new_status
503 });
504
505 api.patch_status(&name, &PatchParams::default(), &Patch::Merge(&status_patch))
506 .await?;
507
508 debug!("Updated status for ClusterBind9Provider: {}", name);
509 Ok(())
510}
511
512#[must_use]
530pub fn cluster_status_needs_update(
531 current_status: Option<&Bind9ClusterStatus>,
532 new_status: &Bind9ClusterStatus,
533) -> bool {
534 let Some(current_status) = current_status else {
535 return true;
537 };
538
539 if current_status.instance_count != new_status.instance_count
541 || current_status.ready_instances != new_status.ready_instances
542 {
543 return true;
544 }
545
546 if current_status.observed_generation != new_status.observed_generation {
551 return true;
552 }
553
554 let Some(current_condition) = current_status.conditions.first() else {
556 return !new_status.conditions.is_empty();
558 };
559
560 match new_status.conditions.first() {
561 Some(new_cond) => {
562 current_condition.r#type != new_cond.r#type
563 || current_condition.status != new_cond.status
564 || current_condition.message != new_cond.message
565 }
566 None => true, }
568}
569
570#[must_use]
581pub fn calculate_cluster_status(
582 instances: &[Bind9Instance],
583 generation: Option<i64>,
584) -> Bind9ClusterStatus {
585 let now = Utc::now();
586
587 let ready_instances = instances
589 .iter()
590 .filter(|inst| {
591 inst.status
592 .as_ref()
593 .and_then(|s| s.conditions.iter().find(|c| c.r#type == "Ready"))
594 .is_some_and(|c| c.status == "True")
595 })
596 .count();
597
598 let total_instances = instances.len();
599
600 let (status, reason, message) = if total_instances == 0 {
602 (
603 "False",
604 REASON_NO_CHILDREN,
605 "No instances found for this cluster".to_string(),
606 )
607 } else if ready_instances == total_instances {
608 (
609 "True",
610 REASON_ALL_READY,
611 format!("All {total_instances} instances are ready"),
612 )
613 } else if ready_instances > 0 {
614 (
615 "False",
616 REASON_PARTIALLY_READY,
617 format!("{ready_instances}/{total_instances} instances are ready"),
618 )
619 } else {
620 (
621 "False",
622 REASON_NOT_READY,
623 "No instances are ready".to_string(),
624 )
625 };
626
627 let instance_names: Vec<String> = instances
629 .iter()
630 .map(|inst| {
631 let name = inst.name_any();
632 let ns = inst.namespace().unwrap_or_default();
633 format!("{ns}/{name}")
634 })
635 .collect();
636
637 Bind9ClusterStatus {
638 conditions: vec![Condition {
639 r#type: CONDITION_TYPE_READY.to_string(),
640 status: status.to_string(),
641 reason: Some(reason.to_string()),
642 message: Some(message.clone()),
643 last_transition_time: Some(now.to_rfc3339()),
644 }],
645 instances: instance_names,
646 observed_generation: generation,
647 #[allow(clippy::cast_possible_truncation, clippy::cast_possible_wrap)]
648 instance_count: Some(total_instances as i32),
649 #[allow(clippy::cast_possible_truncation, clippy::cast_possible_wrap)]
650 ready_instances: Some(ready_instances as i32),
651 }
652}
653
654async fn detect_cluster_drift(
677 client: &Client,
678 cluster_provider: &ClusterBind9Provider,
679) -> Result<bool> {
680 use crate::crd::Bind9Cluster;
681 use crate::labels::{
682 BINDY_CLUSTER_LABEL, BINDY_MANAGED_BY_LABEL, MANAGED_BY_CLUSTER_BIND9_PROVIDER,
683 };
684 use kube::api::ListParams;
685
686 let cluster_provider_name = cluster_provider.name_any();
687
688 let expected_namespaces = compute_expected_cluster_namespaces(client, cluster_provider).await?;
693
694 let expected_count_per_namespace = 1;
696
697 for namespace in &expected_namespaces {
698 let clusters_api: Api<Bind9Cluster> = Api::namespaced(client.clone(), namespace);
700 let clusters = clusters_api.list(&ListParams::default()).await?;
701
702 let managed_clusters: Vec<_> = clusters
704 .items
705 .into_iter()
706 .filter(|cluster| {
707 cluster.metadata.labels.as_ref().is_some_and(|labels| {
708 labels.get(BINDY_MANAGED_BY_LABEL)
709 == Some(&MANAGED_BY_CLUSTER_BIND9_PROVIDER.to_string())
710 && labels.get(BINDY_CLUSTER_LABEL) == Some(&cluster_provider_name.clone())
711 })
712 })
713 .collect();
714
715 let actual_count = managed_clusters.len();
717 if actual_count != expected_count_per_namespace {
718 info!(
719 "Cluster count drift detected for ClusterBind9Provider {}: expected {} Bind9Cluster in namespace {}, found {}",
720 cluster_provider_name, expected_count_per_namespace, namespace, actual_count
721 );
722 return Ok(true);
723 }
724
725 if let Some(managed_cluster) = managed_clusters.first() {
727 if cluster_provider.spec.common != managed_cluster.spec.common {
729 info!(
730 "Cluster spec drift detected for ClusterBind9Provider {} in namespace {}: \
731 managed Bind9Cluster spec differs from desired spec",
732 cluster_provider_name, namespace
733 );
734 return Ok(true);
735 }
736 }
737 }
738
739 Ok(false)
741}
742
743pub async fn delete_clusterbind9provider(
762 ctx: Arc<Context>,
763 cluster: ClusterBind9Provider,
764) -> Result<()> {
765 let name = cluster.name_any();
766 info!("Deleting ClusterBind9Provider: {}", name);
767
768 Box::pin(reconcile_clusterbind9provider(ctx, cluster)).await
770}