Add deletes to change feeds#6058
Open
deads2k wants to merge 1 commit into
Open
Conversation
Contributor
There was a problem hiding this comment.
Pull request overview
This PR updates Cosmos-backed informers/controllers and supporting tooling to propagate resource deletions through “latest version” Cosmos change feeds by implementing soft-delete (deletionTimestamp + TTL) and by treating instanceVersion as the informer resourceVersion.
Changes:
- Implement soft-delete semantics in Cosmos CRUD (deletionTimestamp + short TTL) and filter soft-deleted docs from Get/List paths.
- Introduce/extend change-feed client plumbing across DB clients, informers, mocks, and integration tests to deliver timely Deleted events.
- Update tooling and logging to understand changefeed log entries and gate event application by instanceVersion; bump azcosmos dependency.
Reviewed changes
Copilot reviewed 130 out of 141 changed files in this pull request and generated 5 comments.
Show a summary per file
| File | Description |
|---|---|
| tooling/hcpctl/cmd/datadump-to-git/cmd.go | Parse changefeed logs; instanceVersion gating |
| tooling/cleanup-sweeper/go.sum | Dependency checksum updates |
| test/go.sum | Dependency checksum updates |
| test/go.mod | Bump azcosmos dependency |
| test-integration/kube-applier/framework/framework.go | Update informer wiring signatures |
| test-integration/go.sum | Dependency checksum updates |
| test-integration/go.mod | Bump azcosmos dependency |
| test-integration/frontend/informer_test.go | Update informer wiring signatures |
| test-integration/backend/unionkubeapplierinformers/integration_test.go | Update informer factory signature usage |
| test-integration/backend/launch/notification_test.go | Update backend informer constructor usage |
| test-integration/backend/controllers/mismatches/delete_orphaned_cosmos_test.go | Wrapper implements new QueueForInformers |
| sessiongate/go.sum | Dependency checksum updates |
| mgmt-agent/go.sum | Dependency checksum updates |
| kube-applier/pkg/controllers/read_desire_kubernetes/controller.go | Add informer ObjectDescription |
| kube-applier/pkg/app/kube_applier.go | Pass changefeed client into informers |
| kube-applier/pkg/app/cosmos_wiring.go | Remove unused ctx parameter |
| kube-applier/go.sum | Dependency checksum updates |
| kube-applier/go.mod | Bump azcosmos dependency |
| kube-applier/cmd/root.go | Remove ctx plumbed into options wiring |
| internal/validation/validators.go | Guard nil CP versions during validation |
| internal/utils/context.go | Add AddResourceTypes helper |
| internal/go.sum | Dependency checksum updates |
| internal/go.mod | Bump azcosmos dependency |
| internal/databasetesting/mock_resources_global_lister.go | Filter soft-deleted docs in mocks |
| internal/databasetesting/mock_resources_db_client.go | Add mock changefeed support |
| internal/databasetesting/mock_resources_crud.go | Soft-delete behavior in mock CRUD |
| internal/databasetesting/mock_resources_changefeed.go | In-memory changefeed implementation |
| internal/databasetesting/mock_kube_applier_client.go | Add changefeed support to kube-applier mock |
| internal/database/unioninformers/kubeapplier/union_informers_test.go | Update informer constructor signature usage |
| internal/database/unioninformers/kubeapplier/factory.go | Pass changefeed client into informers |
| internal/database/unioninformers/kubeapplier/controller_test.go | Update informer constructor signature usage |
| internal/database/types_typeddocument.go | Add deletionTimestamp to typed envelope |
| internal/database/kube_applier_client.go | Expose changefeed APIs; adjust constructors |
| internal/database/informers/types.go | Plumb ChangeFeedClient into informers |
| internal/database/informers/informers.go | Switch desire informers to changefeed listwatch |
| internal/database/informers/informers_test.go | Update informer constructor signature usage |
| internal/database/informers/fleet_informers.go | Add informer ObjectDescription fields |
| internal/database/global_lister.go | Filter out soft-deleted docs in queries |
| internal/database/database.go | Add ChangeFeedClient interface to DB client |
| internal/database/crud_helpers.go | Implement soft-delete replace + TTL |
| internal/database/crud_hcpcluster.go | Filter out soft-deleted active operations |
| internal/api/types_runtime.go | Set ResourceVersion from InstanceVersion |
| internal/api/kubeapplier/types_runtime.go | Set ResourceVersion from InstanceVersion |
| internal/api/fleet/types_runtime.go | Set ResourceVersion from InstanceVersion |
| internal/api/arm/types_runtime.go | Set ResourceVersion from InstanceVersion |
| frontend/pkg/frontend/node_pool.go | Remove delete-time datadump logging |
| frontend/pkg/frontend/external_auth.go | Remove delete-time datadump logging |
| frontend/pkg/frontend/cluster.go | Remove delete-time datadump logging |
| frontend/go.sum | Dependency checksum updates |
| frontend/go.mod | Bump azcosmos dependency |
| fleet/pkg/controllers/base/stamp_watching_controller.go | Provide handler logger in notifier wiring |
| fleet/go.sum | Dependency checksum updates |
| fleet/go.mod | Bump azcosmos dependency |
| docs/cosmos-resource-deletion.md | Document soft-delete approach |
| backend/pkg/informers/types.go | Pass ResourcesDBClient as changefeed client |
| backend/pkg/informers/informers_test.go | Update informer constructors in tests |
| backend/pkg/controllers/validationcontrollers/nodepool_validation_controller.go | Remove cooldown checker plumbing |
| backend/pkg/controllers/validationcontrollers/nodepool_validation_controller_test.go | Update tests for cooldown removal |
| backend/pkg/controllers/validationcontrollers/cluster_validation_controller.go | Remove cooldown checker plumbing |
| backend/pkg/controllers/upgradecontrollers/trigger_node_pool_upgrade_controller.go | Remove cooldown checker plumbing |
| backend/pkg/controllers/upgradecontrollers/trigger_control_plane_upgrade_controller.go | Active create detection via ActiveOperationID |
| backend/pkg/controllers/upgradecontrollers/trigger_control_plane_upgrade_controller_test.go | Update tests for new gating logic |
| backend/pkg/controllers/upgradecontrollers/nodepool_version_controller.go | Queue SPC informer; NeedsWork requires CP versions |
| backend/pkg/controllers/upgradecontrollers/nodepool_version_controller_test.go | Update tests for signature/logic changes |
| backend/pkg/controllers/upgradecontrollers/nodepool_active_version_real_cosmos_test.go | Update tests for cooldown removal |
| backend/pkg/controllers/upgradecontrollers/nodepool_active_version_controller.go | Disable cooldown gating (nil checker) |
| backend/pkg/controllers/upgradecontrollers/nodepool_active_version_controller_test.go | Update tests for cooldown removal |
| backend/pkg/controllers/upgradecontrollers/control_plane_desired_version_controller.go | Active create detection via ActiveOperationID |
| backend/pkg/controllers/upgradecontrollers/control_plane_desired_version_controller_test.go | Update tests for new gating logic |
| backend/pkg/controllers/upgradecontrollers/control_plane_active_version_controller.go | Remove cooldown checker plumbing |
| backend/pkg/controllers/upgradecontrollers/control_plane_active_version_controller_test.go | Update tests for cooldown removal |
| backend/pkg/controllers/statuscontrollers/nodepool_degraded_aggregator.go | Handle NotFound on replace after delete |
| backend/pkg/controllers/statuscontrollers/nodepool_degraded_aggregator_test.go | Update tests for cooldown removal |
| backend/pkg/controllers/statuscontrollers/externalauth_degraded_aggregator.go | Handle NotFound on replace after delete |
| backend/pkg/controllers/statuscontrollers/externalauth_degraded_aggregator_test.go | Update tests for cooldown removal |
| backend/pkg/controllers/statuscontrollers/cluster_degraded_aggregator.go | Handle NotFound on replace after delete |
| backend/pkg/controllers/statuscontrollers/cluster_degraded_aggregator_test.go | Update tests for cooldown removal |
| backend/pkg/controllers/statuscontrollers/aggregator_testhelpers_test.go | Remove unused cooldown test helper |
| backend/pkg/controllers/operationcontrollers/generic_operation.go | Add QueueForInformers (panic stub) |
| backend/pkg/controllers/nodepooldeletion/node_pool_deletion_controller.go | Remove cooldown checker plumbing |
| backend/pkg/controllers/nodepooldeletion/node_pool_deletion_controller_test.go | Update tests for cooldown removal |
| backend/pkg/controllers/nodepooldeletion/node_pool_cluster_service_id_clearer.go | Remove cooldown checker plumbing |
| backend/pkg/controllers/nodepooldeletion/node_pool_cluster_service_id_clearer_test.go | Update tests for cooldown removal |
| backend/pkg/controllers/nodepooldeletion/node_pool_cluster_service_delete_dispatch_controller.go | Remove cooldown checker plumbing |
| backend/pkg/controllers/nodepooldeletion/node_pool_cluster_service_delete_dispatch_controller_test.go | Update tests for cooldown removal |
| backend/pkg/controllers/nodepooldeletion/node_pool_child_resources_cleanup_controller.go | Remove cooldown checker plumbing |
| backend/pkg/controllers/nodepooldeletion/node_pool_child_resources_cleanup_controller_test.go | Update tests for cooldown removal |
| backend/pkg/controllers/nodepoolcreationcontrollers/node_pool_cluster_service_create_controller.go | Remove cooldown checker plumbing |
| backend/pkg/controllers/nodepoolcreationcontrollers/node_pool_cluster_service_create_controller_test.go | Update tests for cooldown removal |
| backend/pkg/controllers/mismatchcontrollers/delete_orphaned_cosmos.go | Add QueueForInformers (panic stub) |
| backend/pkg/controllers/mismatchcontrollers/cluster_service_cluster_matching.go | Suppress “no healthy upstream” failure |
| backend/pkg/controllers/managementclustercontrollers/management_cluster_placement_sync.go | Remove cooldown checker plumbing |
| backend/pkg/controllers/managementclustercontrollers/management_cluster_placement_sync_test.go | Update tests for cooldown removal |
| backend/pkg/controllers/externalauthdeletion/external_auth_deletion_controller.go | Remove cooldown checker plumbing |
| backend/pkg/controllers/externalauthdeletion/external_auth_deletion_controller_test.go | Update tests for cooldown removal |
| backend/pkg/controllers/externalauthdeletion/external_auth_cluster_service_id_clearer.go | Remove cooldown checker plumbing |
| backend/pkg/controllers/externalauthdeletion/external_auth_cluster_service_id_clearer_test.go | Update tests for cooldown removal |
| backend/pkg/controllers/externalauthdeletion/external_auth_cluster_service_delete_dispatch_controller.go | Remove cooldown checker plumbing |
| backend/pkg/controllers/externalauthdeletion/external_auth_cluster_service_delete_dispatch_controller_test.go | Update tests for cooldown removal |
| backend/pkg/controllers/externalauthdeletion/external_auth_child_resources_cleanup_controller.go | Remove cooldown checker plumbing |
| backend/pkg/controllers/externalauthdeletion/external_auth_child_resources_cleanup_controller_test.go | Update tests for cooldown removal |
| backend/pkg/controllers/do_nothing.go | Add QueueForInformers (panic stub) |
| backend/pkg/controllers/datadumpcontrollers/dump_subscription_non_cluster.go | Use time-based cooldown only |
| backend/pkg/controllers/datadumpcontrollers/dump_cluster_recursive.go | Remove cooldown checker plumbing |
| backend/pkg/controllers/datadumpcontrollers/cs_state_dump.go | Simplify cooldown checker usage |
| backend/pkg/controllers/datadumpcontrollers/cs_state_dump_test.go | Update tests for cooldown changes |
| backend/pkg/controllers/datadumpcontrollers/billing_dump.go | Simplify cooldown checker usage |
| backend/pkg/controllers/datadumpcontrollers/billing_dump_test.go | Update tests for cooldown changes |
| backend/pkg/controllers/create_nodepool_scoped_read_desires_controller.go | Remove cooldown checker plumbing |
| backend/pkg/controllers/create_cluster_scoped_read_desires_controller.go | Remove cooldown checker plumbing |
| backend/pkg/controllers/controllerutils/util.go | Controller interface adds QueueForInformers |
| backend/pkg/controllers/controllerutils/nodepool_watching_controller.go | Remove syncer CooldownChecker requirement |
| backend/pkg/controllers/controllerutils/generic_watching_controller.go | Add handler loggers; nil cooldown handling |
| backend/pkg/controllers/controllerutils/external_auth_watching_controller.go | Remove syncer CooldownChecker requirement |
| backend/pkg/controllers/controllerutils/cluster_watching_controller.go | Remove syncer CooldownChecker requirement |
| backend/pkg/controllers/clusterpropertiescontroller/identity_migration.go | Remove cooldown checker plumbing |
| backend/pkg/controllers/clusterpropertiescontroller/identity_migration_test.go | Update tests for cooldown removal |
| backend/pkg/controllers/clusterpropertiescontroller/desired_control_plane_size_sync.go | Remove cooldown checker plumbing |
| backend/pkg/controllers/clusterpropertiescontroller/desired_control_plane_size_sync_test.go | Update tests for cooldown removal |
| backend/pkg/controllers/clusterpropertiescontroller/cluster_properties_test_helpers_test.go | Remove unused cooldown test helper |
| backend/pkg/controllers/clusterpropertiescontroller/cluster_properties_sync.go | Remove cooldown checker plumbing |
| backend/pkg/controllers/clusterpropertiescontroller/cluster_properties_sync_test.go | Update tests for cooldown removal |
| backend/pkg/controllers/clusterpropertiescontroller/cluster_base_domain_prefix_sync.go | Remove cooldown checker plumbing |
| backend/pkg/controllers/clusterpropertiescontroller/cluster_base_domain_prefix_sync_test.go | Update tests for cooldown removal |
| backend/pkg/controllers/clusterdeletion/cluster_deletion_controller.go | Remove cooldown checker plumbing |
| backend/pkg/controllers/clusterdeletion/cluster_deletion_controller_test.go | Update tests for cooldown removal |
| backend/pkg/controllers/clusterdeletion/cluster_cluster_service_id_clearer.go | Remove cooldown checker plumbing |
| backend/pkg/controllers/clusterdeletion/cluster_cluster_service_id_clearer_test.go | Update tests for cooldown removal |
| backend/pkg/controllers/clusterdeletion/cluster_cluster_service_delete_dispatch_controller.go | Remove cooldown checker plumbing |
| backend/pkg/controllers/clusterdeletion/cluster_cluster_service_delete_dispatch_controller_test.go | Update tests for cooldown removal |
| backend/pkg/controllers/clusterdeletion/cluster_child_resources_cleanup_controller.go | Remove cooldown checker plumbing |
| backend/pkg/controllers/clusterdeletion/cluster_child_resources_cleanup_controller_test.go | Update tests for cooldown removal |
| backend/pkg/controllers/billingcontrollers/orphaned_billing_cleanup.go | Add QueueForInformers (panic stub) |
| backend/pkg/app/backend.go | Pass DB client into backend informers |
| backend/go.sum | Dependency checksum updates |
| backend/go.mod | Bump azcosmos dependency |
| admin/server/go.sum | Dependency checksum updates |
| admin/server/go.mod | Bump azcosmos dependency |
| func (c *triggerControlPlaneUpgradeSyncer) shouldTriggerUpgrade(ctx context.Context, cluster *api.HCPOpenShiftCluster) (bool, error) { | ||
| logger := utils.LoggerFromContext(ctx) | ||
| if c.clusterOlderThanGracePeriod(cluster) { | ||
| logger.Info("Cluster is older than grace period, skipping upgrade trigger", "cluster", cluster.Name) |
| 1. Reads the document from Cosmos (`ReadItem`). | ||
| 2. Sets `deletionTimestamp` to the current time. | ||
| 3. Sets `ttl` to 30 (seconds after the write timestamp `_ts`). | ||
| 4. Increments `properties.instanceVersion` so the change feed recognizes the update. |
Comment on lines
+122
to
+126
| // GetChangeFeed reads the in-memory change-feed log. Each | ||
| // successful StoreDocument call records a snapshot of the document; | ||
| // reads return everything past the position encoded in | ||
| // options.Continuation. Deletes are not recorded, which mirrors | ||
| // "latest version" mode in real Cosmos DB. |
Comment on lines
+28
to
+32
| // container. Every successful StoreDocument call (Create or Replace) | ||
| // appends a snapshot of the stored document; reads return everything | ||
| // past the consumer's continuation position. Deletes are intentionally | ||
| // NOT recorded — that mirrors "latest version" change feed mode in | ||
| // production Cosmos DB, which does not surface deletions. |
Comment on lines
+205
to
+211
| func (lv LogValues) AddResourceTypes(values ...azcorearm.ResourceType) LogValues { | ||
| ret := lv | ||
| for _, value := range values { | ||
| ret = ret.AddResourceType(value.String()) | ||
| } | ||
| return ret | ||
| } |
|
@deads2k: The following test failed, say
Full PR test history. Your PR dashboard. DetailsInstructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes-sigs/prow repository. I understand the commands that are listed here. |
Instead of hard-deleting documents with DeleteItem (which the changefeed cannot observe), Delete now sets a DeletionTimestamp and a 30-second TTL on the document via a conditional Replace. This surfaces deletions on the changefeed so controllers receive timely watch.Deleted notifications. Get/GetByID return 404 for soft-deleted documents. All list queries (scoped, global, and ListActiveOperations) filter them out with NOT IS_DEFINED(c.deletionTimestamp). The changefeed watcher checks DeletionTimestamp before shouldDeliverItemFn and emits watch.Deleted for previously-seen resources. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
deads2k
force-pushed
the
cs-193-changefeed-delete
branch
from
1407ce7 to
37f0d66
Compare
|
[APPROVALNOTIFIER] This PR is APPROVED This pull-request has been approved by: deads2k The full list of commands accepted by this bot can be found here. The pull request process is described here DetailsNeeds approval from an approver in each of these files:
Approvers can indicate their approval by writing |
Comment on lines
+407
to
+412
| if objAsTypedDocument.DeletionTimestamp != nil { | ||
| if objPreviouslySeen { | ||
| objDeleted = true | ||
| } else { | ||
| logger.Info("skipping soft-deleted document not previously seen", "content", cosmosObj) | ||
| return nil |
Comment on lines
+401
to
405
| IfMatchEtag: &responseItem.ETag, | ||
| } | ||
| _, err = containerClient.ReplaceItem(ctx, pk, cosmosID, modified, opts) | ||
| if err != nil { | ||
| return utils.TrackError(err) |
Comment on lines
+169
to
+173
| evt := waitForEvent(t, watcher, eventDeadline) | ||
| require.Equal(t, watch.Deleted, evt.Type, "expected Deleted event for soft-deleted cluster") | ||
| gotResourceID, _ := metadataOf(t, evt.Object) | ||
| require.Truef(t, strings.EqualFold(gotResourceID, clusterRID.String()), | ||
| "event resourceID %q != deleted %q", gotResourceID, clusterRID.String()) |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Add deletes to changefeeds