diff --git a/AGENTS.md b/AGENTS.md index 64536b6..6a9b582 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -67,9 +67,10 @@ the latter case, keep the change minimal and explain why it is necessary and whi when changing field ownership or event handling. - **Watch lifecycle:** [remote_watcher_manager.rs](src/remote_watcher_manager.rs) keys watches by resource reference and owning sync, for both local and remote resources. Preserve watcher cancellation and joining during cleanup and - shutdown. The main `ResourceSync` stream uses kube's generation predicate with UID-aware caching and a 24-hour idle - TTL. Kubeconfig Secret changes are not explicit triggers. Read [event and retry behavior](README.md#status-and-observability) - before changing configuration refresh or reconciliation scheduling. + shutdown. The main `ResourceSync` stream hashes generation, deletion state, and deletion-control annotations with + UID-aware caching and a 24-hour idle TTL. Kubeconfig Secret changes are not explicit triggers. Read + [event and retry behavior](README.md#status-and-observability) before changing configuration refresh or reconciliation + scheduling. - **Status:** In [controller.rs](src/controller.rs), compute `ResourceSyncFailing` from live status, not the reflector cache. Success clears failure, unchanged condition values retain `lastTransitionTime`, and successful deletion cleanup skips the status write because the object may already be gone. Preserve the regression coverage for these @@ -77,8 +78,10 @@ the latter case, keep the change minimal and explain why it is necessary and whi - **Deletion and ownership:** Read [Annotations and finalizers](README.md#annotations-and-finalizers) before changing cleanup. Preserve unrelated finalizers. Normal cleanup waits for target absence before stopping watches and removing Sinker's finalizer. Both APIs are resolved before cleanup; `force-delete` only bypasses failures at that resolution - step. Local targets also have an owner reference, so `disable-target-deletion` alone does not retain them. Account for - the documented local cluster-scoped ownership limitation when changing target scope or retention behavior. + step. Local namespaced targets receive a Sinker owner reference only when target deletion is enabled; local + cluster-scoped targets receive none. Changing a deletion-control annotation's effective Boolean value triggers + reconciliation. A successful target apply removes Sinker's omitted owner reference if no other manager owns it; + deletion skips that apply. Account for existing references when changing target scope or retention behavior. ## CRD generation diff --git a/README.md b/README.md index 30d7287..14bfb9c 100644 --- a/README.md +++ b/README.md @@ -176,7 +176,7 @@ The two namespace fields have different meanings: | `cluster.namespace` | Namespace of the source/target in the **referenced cluster**. | Selected kubeconfig context's namespace, or `default`. | Discovery determines whether a kind is namespaced or cluster-scoped. Cluster-scoped API requests use no resource -namespace; see the [local target ownership limitation](#annotations-and-finalizers) before using such a target. +namespace; see [target ownership and deletion](#annotations-and-finalizers) when using such a target. Create the kubeconfig Secret in an existing namespace. This example authorizes syncs in `default` to use credentials stored in `cluster-credentials`: @@ -282,17 +282,25 @@ These annotations belong to the **ResourceSync** and are enabled only by the str | Annotation | Effect | | --- | --- | -| `sinker.influxdata.io/disable-target-deletion: "true"` | Skip Sinker issuing the target delete request, stop watches, and remove Sinker's finalizer. | +| `sinker.influxdata.io/disable-target-deletion: "true"` | Skip Sinker's target delete request and owner reference, stop watches, and remove Sinker's finalizer. The owner-reference change applies when Sinker next applies a local namespaced target. | | `sinker.influxdata.io/force-delete: "true"` | During deletion, remove Sinker's finalizer if constructing either source or target API fails, for example because credentials or a remote API are unavailable. It does not bypass errors from subsequent target GET/DELETE requests or a target's finalizers. | Both source and target APIs are resolved before normal cleanup, even with `disable-target-deletion`. An unavailable source cluster can therefore block cleanup too. Force deletion can leave a target behind for manual cleanup. -Local targets (`target.cluster` omitted) also receive an owner reference to the namespaced `ResourceSync`. -**Disabling target deletion alone does not retain a local target**: Kubernetes garbage collection can still delete it. -Retention must also account for Kubernetes [orphaning behavior](https://kubernetes.io/docs/concepts/architecture/garbage-collection/#orphaned-dependents). -The controller currently adds this owner reference even to a local cluster-scoped target, which conflicts with Kubernetes +Local namespaced targets (`target.cluster` omitted) receive an owner reference to the `ResourceSync` in the same namespace +unless `sinker.influxdata.io/disable-target-deletion` is `"true"` when Sinker applies the target. Set it before creating the sync +to have Sinker skip both the delete request and its owner reference. Enabling it on an active sync triggers +reconciliation. On a successful server-side apply, omitting Sinker's previously managed owner reference removes it if +no other field manager also owns that entry; see Kubernetes [field management](https://kubernetes.io/docs/reference/using-api/server-side-apply/#field-management). +Verify the target's `metadata.ownerReferences` before deleting the sync if reconciliation fails or the sync is already +deleting, since deletion skips the normal target apply. +Existing owner references from other controllers also affect retention; see Kubernetes +[orphaning behavior](https://kubernetes.io/docs/concepts/architecture/garbage-collection/#orphaned-dependents). +Local cluster-scoped targets receive no Sinker owner reference, following Kubernetes [owner scope rules](https://kubernetes.io/docs/concepts/architecture/garbage-collection/#owners-and-dependents). +For those targets, `disable-target-deletion` skips Sinker's delete request without a Sinker owner reference triggering +garbage collection. Targets configured through a kubeconfig receive no Sinker owner reference, even if that kubeconfig points at the local cluster. ## Status and observability @@ -342,10 +350,11 @@ select that backend as the dashboard's Prometheus datasource. Metric labels iden source or target resource's namespace or cluster. Reconciliation errors retry after five seconds; object watches reconnect with backoff. A healthy sync waits for events -rather than polling on a fixed interval. `ResourceSync` generation filtering tracks name, namespace, and UID, with a -24-hour cache TTL since the last observation. An event with a new UID or an expired cache entry can trigger reconciliation -even when its generation is unchanged; cache expiration itself does not schedule reconciliation. Metadata-only edits -are normally filtered, and kubeconfig Secret changes are not explicit reconciliation triggers. +rather than polling on a fixed interval. The `ResourceSync` filter tracks generation, deletion state, +`disable-target-deletion`, and `force-delete`, with a cache keyed by name, namespace, and UID and a 24-hour idle TTL. +Changes to either annotation's effective Boolean value trigger reconciliation even if the generation is unchanged. +Status writes and unrelated metadata-only edits are filtered. A new UID or expired cache entry can also trigger reconciliation; cache expiration +itself does not schedule it. Events with no generation are forwarded. Kubeconfig Secret changes are not explicit triggers. Existing watches keep their clients until they reconnect. A retry or subsequent object event can pick up updated settings; restart the controller when credential or access changes need to take effect promptly. @@ -371,8 +380,8 @@ flowchart TD C -->|ResourceSync is deleting| D[Clean up target and remove finalizer] ``` -The controller filters `ResourceSync` events by generation to avoid loops from its own status writes. The watcher manager -keeps a watcher per resource reference and owning sync, for local and remote endpoints. Object watches use +The controller filters `ResourceSync` events by reconciliation-relevant fields to avoid loops from its own status writes. +The watcher manager keeps a watcher per resource reference and owning sync, for local and remote endpoints. Object watches use `metadata.name` selectors and managed-field timestamps to suppress changes last attributed to `sinker.influxdata.io`; events with unknown ownership trigger reconciliation. Shutdown cancels and joins the watchers. diff --git a/src/controller.rs b/src/controller.rs index 244cdc8..d9604bf 100644 --- a/src/controller.rs +++ b/src/controller.rs @@ -3,7 +3,7 @@ use k8s_openapi::apimachinery::pkg::apis::meta::v1::{Condition, OwnerReference, use k8s_openapi::jiff::Timestamp; use kube::api::DeleteParams; use kube::api::Patch::Merge; -use kube::runtime::{predicates, reflector, PredicateConfig, WatchStreamExt}; +use kube::runtime::{reflector, PredicateConfig, WatchStreamExt}; use kube::{ api::{ListParams, Patch, PatchParams}, runtime::{ @@ -13,6 +13,7 @@ use kube::{ Api, Client, Resource, ResourceExt, }; use serde_json::json; +use std::hash::{DefaultHasher, Hash, Hasher}; use std::string::ToString; use std::{sync::Arc, time::Duration}; #[allow(unused_imports)] @@ -195,26 +196,31 @@ async fn reconcile_normally( )? }; - // If the target is local then add an owner reference to it - match resource_sync.spec.target.cluster.to_owned() { - Some(_) => target, - None => { - target.owner_references_mut().push(OwnerReference { - api_version: ResourceSync::api_version(&()).to_string(), - kind: ResourceSync::kind(&()).to_string(), - name: name.to_owned(), - uid: resource_sync - .metadata - .uid - .to_owned() - .ok_or(Error::UIDRequired)?, - controller: Some(false), - block_owner_deletion: Some(true), - }); - - target - } + // Owner-based garbage collection must not override disable-target-deletion. + // A namespaced ResourceSync can own only a local target in its namespace; + // cluster-scoped and kubeconfig targets do not receive this reference. + if resource_sync.spec.target.cluster.is_none() + && !resource_sync.has_disable_target_deletion_option_enabled() + && target_api + .namespace + .as_ref() + .eq(&resource_sync.metadata.namespace.as_ref()) + { + target.owner_references_mut().push(OwnerReference { + api_version: ResourceSync::api_version(&()).to_string(), + kind: ResourceSync::kind(&()).to_string(), + name: name.to_owned(), + uid: resource_sync + .metadata + .uid + .to_owned() + .ok_or(Error::UIDRequired)?, + controller: Some(false), + block_owner_deletion: Some(true), + }); } + + target }; debug!(?target, "produced target object"); @@ -233,8 +239,6 @@ async fn reconcile_normally( Ok(Action::await_change()) } -// TODO: If secrets for remote clusters on target and source (when applicable) no longer exist then simply allow the ResourceSync to be deleted by removing the finalizer - #[expect( clippy::result_large_err, reason = "Preserve the public Error variants without boxing" @@ -437,7 +441,6 @@ fn sync_failing_transition_time(status: &Option, new_status: fn error_policy(resource_sync: Arc, error: &Error, _ctx: Arc) -> Action { let name = resource_sync.name_any(); warn!(?name, %error, "reconcile failed"); - // TODO(mkm): make error requeue duration configurable Action::requeue(Duration::from_secs(5)) } @@ -478,7 +481,7 @@ pub async fn run_with_metrics(client: Client, metrics: ControllerMetrics) -> Res .reflect(writer) .applied_objects() .predicate_filter( - predicates::generation, + resource_sync_predicate, PredicateConfig::default().ttl(RESOURCE_SYNC_PREDICATE_TTL), ); @@ -509,11 +512,38 @@ pub async fn run_with_metrics(client: Client, metrics: ControllerMetrics) -> Res Ok(()) } +#[derive(Hash)] +struct ResourceSyncPredicateFields { + generation: i64, + deleting: bool, + disable_target_deletion: bool, + force_delete: bool, +} + +/// Return a watch-cache hash of generation and metadata that affects reconciliation. +/// +/// The hash is for equality checks within the watch cache, not a persistent identifier. +/// A missing generation returns `None` so the filter forwards every such event. +pub fn resource_sync_predicate(obj: &ResourceSync) -> Option { + let fields = ResourceSyncPredicateFields { + generation: obj.meta().generation?, + deleting: obj.has_been_deleted(), + disable_target_deletion: obj.has_disable_target_deletion_option_enabled(), + force_delete: obj.has_force_delete_option_enabled(), + }; + + let mut hasher = DefaultHasher::new(); + fields.hash(&mut hasher); + + Some(hasher.finish()) +} + #[cfg(test)] mod tests { use super::{ reconcile, reconcile_deleted_resource, reconcile_helper, reconcile_normally, - reconcile_with_metrics, Context, ControllerMetrics, RemoteWatcherManager, + reconcile_with_metrics, resource_sync_predicate, Context, ControllerMetrics, + RemoteWatcherManager, }; use super::{ reconcile_status, sync_failing_transition_time, RESOURCE_SYNC_FAILING_CONDITION, @@ -536,16 +566,33 @@ mod tests { use std::{sync::Arc, time::Duration}; #[tokio::test] - async fn resource_sync_predicate_accepts_new_uids_and_generation_changes() { + async fn resource_sync_predicate_tracks_reconciliation_relevant_changes() { use super::RESOURCE_SYNC_PREDICATE_TTL; use futures::{stream, StreamExt}; - use kube::runtime::{predicates, watcher, PredicateConfig, WatchStreamExt}; + use kube::runtime::{watcher, PredicateConfig, WatchStreamExt}; let mut original = sync_fixture(); original.metadata.generation = Some(1); let mut metadata_only = original.clone(); metadata_only.metadata.resource_version = Some("2".into()); metadata_only.status = Some(ResourceSyncStatus::default()); + let mut deletion_disabled = metadata_only.clone(); + deletion_disabled.metadata.annotations = Some(std::collections::BTreeMap::from([( + crate::resources::DISABLE_TARGET_DELETION_ANNOTATION.into(), + "true".into(), + )])); + let mut force_delete = deletion_disabled.clone(); + force_delete + .metadata + .annotations + .as_mut() + .expect("annotation map") + .insert( + crate::resources::FORCE_DELETE_ANNOTATION.into(), + "true".into(), + ); + let mut deleting = force_delete.clone(); + deleting.metadata.deletion_timestamp = Some(EPOCH.clone()); let mut recreated = original.clone(); recreated.metadata.uid = Some("replacement-uid".into()); let mut updated = recreated.clone(); @@ -557,6 +604,9 @@ mod tests { Ok(original.clone()), Ok(metadata_only), Err(watcher::Error::NoResourceVersion), + Ok(deletion_disabled.clone()), + Ok(force_delete.clone()), + Ok(deleting.clone()), Ok(recreated.clone()), Ok(recreated.clone()), Ok(updated.clone()), @@ -564,13 +614,13 @@ mod tests { Ok(missing_generation.clone()), ]) .predicate_filter( - predicates::generation, + resource_sync_predicate, PredicateConfig::default().ttl(RESOURCE_SYNC_PREDICATE_TTL), ) .collect::>() .await; - assert_eq!(events.len(), 6); + assert_eq!(events.len(), 9); assert!(matches!( events.remove(1), Err(watcher::Error::NoResourceVersion) @@ -583,6 +633,9 @@ mod tests { objects, [ json!(original), + json!(deletion_disabled), + json!(force_delete), + json!(deleting), json!(recreated), json!(updated), json!(missing_generation), @@ -591,6 +644,31 @@ mod tests { ); } + #[rstest] + #[case::local_disable(false, crate::resources::DISABLE_TARGET_DELETION_ANNOTATION)] + #[case::remote_disable(true, crate::resources::DISABLE_TARGET_DELETION_ANNOTATION)] + #[case::local_force(false, crate::resources::FORCE_DELETE_ANNOTATION)] + #[case::remote_force(true, crate::resources::FORCE_DELETE_ANNOTATION)] + fn resource_sync_predicate_tracks_deletion_annotations_for_both_target_scopes( + #[case] remote: bool, + #[case] annotation: &str, + ) { + let mut before = sync_fixture(); + if remote { + before.spec.target.cluster = Some(Default::default()); + } + let mut after = before.clone(); + after.metadata.annotations = Some(std::collections::BTreeMap::from([( + annotation.into(), + "true".into(), + )])); + + assert_ne!( + resource_sync_predicate(&before), + resource_sync_predicate(&after) + ); + } + fn context(client: kube::Client) -> Arc { let (remote_watcher_manager, _events) = RemoteWatcherManager::new(client.clone()); Arc::new(Context { @@ -943,16 +1021,30 @@ mod tests { } #[rstest] - #[case::whole_resource(false, false)] - #[case::mapped_resource(true, false)] - #[case::remote_target(false, true)] - #[case::mapped_remote_target(true, true)] + #[case::whole_resource(false, false, false, true)] + #[case::mapped_resource(true, false, false, true)] + #[case::remote_target(false, true, false, true)] + #[case::mapped_remote_target(true, true, false, true)] + #[case::local_deletion_disabled(false, false, true, true)] + #[case::mapped_local_deletion_disabled_without_uid(true, false, true, false)] + #[case::remote_deletion_disabled_without_uid(false, true, true, false)] #[tokio::test] - async fn target_apply_uses_forced_field_manager_and_local_ownership( + async fn target_apply_uses_forced_field_manager_and_conditional_ownership( #[case] mapped: bool, #[case] remote: bool, + #[case] deletion_disabled: bool, + #[case] uid_present: bool, ) { let mut sync = sync_fixture(); + if deletion_disabled { + sync.metadata.annotations = Some(std::collections::BTreeMap::from([( + crate::resources::DISABLE_TARGET_DELETION_ANNOTATION.into(), + "true".into(), + )])); + } + if !uid_present { + sync.metadata.uid = None; + } if mapped { sync.spec.mappings = vec![crate::resources::Mapping { from_field_path: Some("data.original".into()), @@ -1022,7 +1114,7 @@ mod tests { json!({"original": "value"}) } ); - if remote { + if remote || deletion_disabled { assert!(patch.body()["metadata"].get("ownerReferences").is_none()); } else { assert_eq!( @@ -1034,6 +1126,76 @@ mod tests { } } + #[rstest] + #[case::with_uid(Some("sync-uid"))] + #[case::without_uid(None)] + #[tokio::test] + async fn local_cluster_scoped_target_has_no_owner_reference(#[case] uid: Option<&str>) { + let mut sync = sync_fixture(); + sync.spec.source.resource_ref.kind = "Namespace".into(); + sync.spec.target.resource_ref.kind = "Namespace".into(); + sync.metadata.uid = uid.map(str::to_string); + let source = json!({ + "apiVersion": "v1", + "kind": "Namespace", + "metadata": { + "name": "source-config", + "labels": {"example.com/source": "yes"} + } + }); + let mock = MockApi::new(vec![ + discovery_response("Namespace", "namespaces", false), + discovery_response("Namespace", "namespaces", false), + response(200, source.clone()), + response(200, source), + ]); + let source_api = sync + .spec + .source + .api_for(mock.client.clone(), "team-a") + .await + .expect("source API"); + let target_api = sync + .spec + .target + .api_for(mock.client.clone(), "team-a") + .await + .expect("target API"); + assert_eq!(target_api.namespace, None); + let ctx = context(mock.client.clone()); + let _cancelled = + crate::remote_watcher_manager::tests::park_watchers(&ctx.remote_watcher_manager, &sync) + .await; + + let result = reconcile_normally( + Arc::new(sync), + "copy-config", + source_api, + target_api, + Arc::clone(&ctx), + ) + .await; + stop_watches(&ctx).await; + assert_eq!( + result.expect("apply cluster-scoped target"), + Action::await_change() + ); + let requests = mock.finish(&[ + ("GET", "/api/v1"), + ("GET", "/api/v1"), + ("GET", "/api/v1/namespaces/source-config"), + ("PATCH", "/api/v1/namespaces/target-config"), + ]); + let patch = &requests[3]; + assert_eq!(patch.body()["metadata"]["name"], "target-config"); + assert!(patch.body()["metadata"].get("namespace").is_none()); + assert!(patch.body()["metadata"].get("ownerReferences").is_none()); + assert_eq!( + patch.body()["metadata"]["labels"], + json!({"example.com/source": "yes"}) + ); + } + #[test] fn deletion_errors_still_update_status_and_ignore_unrelated_conditions() { let mut live = status_with_condition("True").expect("status fixture"); diff --git a/src/remote_watcher.rs b/src/remote_watcher.rs index b95acba..550a14c 100644 --- a/src/remote_watcher.rs +++ b/src/remote_watcher.rs @@ -47,9 +47,6 @@ macro_rules! rv_for { }; } -// TODO: There may be some other error types that should not be triggering reconciles -// TODO: Could also process mappings to ignore changes to fields that we don't care about, but it's far more complex for a lot less benefit - impl RemoteWatcher { pub fn new( key: RemoteWatcherKey,