Repository: nolar/kopf

Issue:
Namespace deletion is stuck when using namespace selector startup mode
### Long story short

When running a kopf operator with namespace selector AND deleting this namespace that contains a resource with a finalizer: kopf may not call the delete handler method.

Deleting the namespace stops the watch-stream for this namespace before the resources contained in that namespace get the chance have its delete method to be called.

Therefore, the finalizer is never removed and the namespace deletion is stuck waiting all resources to be deleted (but they won't since the watch-stream is stopped).

We manage to find a workaround using the clusterwide setting.

Earlier, we say "may not call"; in some cases the delete handler is called but it is quickly canceled by a CancelledError coming from kopf. We suppose it may be relatee to the event queue size. In the simple example provided, we never manage to get the delete handler to be called.

### Kopf version

1.36.2

### Kubernetes version

1.27.4

### Python version

3.11.1

### Code

```python
import kopf
import asyncio
import threading
import logging
from kopf._core.actions import loggers


@kopf.on.create('kopfexamples')
def create_fn(**kwargs):
    print('CREATED!!!')


@kopf.on.delete('kopfexamples')
async def delete_fn(**kwargs):
    await asyncio.sleep(2)
    print(f'DELETED!!!')


@kopf.on.startup()  # type: ignore
def configure(settings: kopf.OperatorSettings, **_) -> None:
    settings.posting.level = logging.DEBUG
    loggers.configure(
        debug=True,
        verbose=False,
        quiet=False,
        log_format=loggers.LogFormat.PLAIN,  # loggers.LogFormat.JSON,
        log_refkey="k8s_obj",
        log_prefix=False,
    )

def kopf_thread():
    asyncio.run(kopf.operator(namespaces=["example"]))


def main():
    logging.getLogger("kopf").setLevel(logging.DEBUG)
    thread = threading.Thread(target=kopf_thread)
    thread.start()
    thread.join()


if __name__ == "__main__":
    main()
```


### Logs

```none
Adding the finalizer, thus preventing the actual deletion.
Patching with: {'metadata': {'finalizers': ['kopf.zalando.org/KopfFinalizerMarker']}}
Creation is in progress: {'apiVersion': 'kopf.dev/v1', 'kind': 'KopfExample', 'metadata': {'annotations': {'kubectl.kubernetes.io/last-applied-configuration': '{"apiVersion":"kopf.dev/v1","kind":"KopfExample","metadata":{"annotations":{"someannotation":"somevalue"},"labels":{"somelabel":"somevalue"},"name":"kopf-example-1","namespace":"example"},"spec":{"duration":"1m","field":"value","items":["item1","item2"]}}\n', 'someannotation': 'somevalue'}, 'creationTimestamp': '2024-01-10T15:07:37Z', 'finalizers': ['kopf.zalando.org/KopfFinalizerMarker'], 'generation': 1, 'labels': {'somelabel': 'somevalue'}, 'managedFields': [{'apiVersion': 'kopf.dev/v1', 'fieldsType': 'FieldsV1', 'fieldsV1': {'f:metadata': {'f:finalizers': {'.': {}, 'v:"kopf.zalando.org/KopfFinalizerMarker"': {}}}}, 'manager': 'kopf', 'operation': 'Update', 'time': '2024-01-10T15:07:37Z'}, {'apiVersion': 'kopf.dev/v1', 'fieldsType': 'FieldsV1', 'fieldsV1': {'f:metadata': {'f:annotations': {'.': {}, 'f:kubectl.kubernetes.io/last-applied-configuration': {}, 'f:someannotation': {}}, 'f:labels': {'.': {}, 'f:somelabel': {}}}, 'f:spec': {'.': {}, 'f:duration': {}, 'f:field': {}, 'f:items': {}}}, 'manager': 'kubectl-client-side-apply', 'operation': 'Update', 'time': '2024-01-10T15:07:37Z'}], 'name': 'kopf-example-1', 'namespace': 'example', 'resourceVersion': '334932', 'uid': '829ca9ed-c7a5-4426-baab-7b3762f9b7fb'}, 'spec': {'duration': '1m', 'field': 'value', 'items': ['item1', 'item2']}}
Handler 'create_fn' is invoked.
Handler 'create_fn' succeeded.
Creation is processed: 1 succeeded; 0 failed.
Patching with: {'metadata': {'annotations': {'kopf.zalando.org/last-handled-configuration': '{"spec":{"duration":"1m","field":"value","items":["item1","item2"]},"metadata":{"labels":{"somelabel":"somevalue"},"annotations":{"someannotation":"somevalue"}}}\n'}}}
CREATED!!!
Something has changed, but we are not interested (the essence is the same).
Handling cycle is finished, waiting for new changes.
Stopping the watch-stream for kopfexamples.v1.kopf.dev in 'example'.
```


### Additional information

After some code digging trying to understand what's going on, we think the problem is here and the namespace is removed from the watching list without any check:
https://github.com/nolar/kopf/blob/91e8fd6fca22685dd487dc58c37c13409709fc86/kopf/_core/reactor/observation.py#L202

Discussion hints:

Solution patch:
diff --git a/kopf/_core/reactor/observation.py b/kopf/_core/reactor/observation.py
index cf30262b..a5cf5b9b 100644
--- a/kopf/_core/reactor/observation.py
+++ b/kopf/_core/reactor/observation.py
@@ -201,7 +201,11 @@ def revise_namespaces(
         namespace = references.NamespaceName(raw_event['object']['metadata']['name'])
         matched = any(references.match_namespace(namespace, pattern) for pattern in namespaces)
         deleted = is_deleted(raw_event)
-        if deleted:
+        blockers = get_blockers(raw_event)
+        if deleted and blockers:
+            for reason, message in blockers:
+                logger.debug(f"Namespace {namespace!r} termination pending: {reason}: {message}")
+        elif deleted:
             insights.namespaces.discard(namespace)
         elif matched:
             insights.namespaces.add(namespace)
@@ -334,6 +338,14 @@ def _disable_unsuitable_resources(
 
 
 def is_deleted(raw_event: bodies.RawEvent) -> bool:
+    # Simply marking for deletion is not enough, it must prove it can be deleted first.
+    has_conditions = bool(raw_event['object'].get('status', {}).get('conditions'))
     marked_as_deleted = bool(raw_event['object'].get('metadata', {}).get('deletionTimestamp'))
-    really_is_deleted = raw_event['type'] == 'DELETED'
-    return marked_as_deleted or really_is_deleted
+    really_is_deleted = raw_event['type'] == 'DELETED'  # does not arrive sometimes
+    return (marked_as_deleted and has_conditions) or really_is_deleted
+
+
+def get_blockers(raw_event: bodies.RawEvent) -> list[tuple[str | None, str | None]]:
+    conditions = raw_event['object'].get('status', {}).get('conditions', [])
+    conditions = [cond for cond in conditions if cond['status'] == 'True']
+    return [(cond['reason'], cond['message']) for cond in conditions]


Test patch:
diff --git a/tests/observation/test_revision_of_namespaces.py b/tests/observation/test_revision_of_namespaces.py
index df754797..5b41c220 100644
--- a/tests/observation/test_revision_of_namespaces.py
+++ b/tests/observation/test_revision_of_namespaces.py
@@ -19,13 +19,39 @@ def test_bodies_for_additional_population(registry):
     assert insights.namespaces == {'ns1', 'ns2'}
 
 
-def test_bodies_for_deletion_via_timestamp(registry):
+def test_bodies_for_deletion_via_timestamp_without_conditions(registry):
     b1 = RawBody(metadata={'name': 'ns1'})
     b2 = RawBody(metadata={'name': 'ns1', 'deletionTimestamp': '...'})
     insights = Insights()
     revise_namespaces(insights=insights, namespaces=['ns*'], raw_bodies=[b1])
     revise_namespaces(insights=insights, namespaces=['ns*'], raw_bodies=[b2])
+    assert insights.namespaces == {'ns1'}
+
+
+def test_bodies_for_deletion_via_timestamp_with_true_conditions(registry, assert_logs):
+    b1 = RawBody(metadata={'name': 'ns1'})
+    b2 = RawBody(metadata={'name': 'ns1', 'deletionTimestamp': '...'},
+                 status={'conditions': [{'type': 'Whatever',
+                                         'status': 'True',
+                                         'reason': 'SomeReason',
+                                         'message': 'Some message'}]})
+    insights = Insights()
+    revise_namespaces(insights=insights, namespaces=['ns*'], raw_bodies=[b1])
+    revise_namespaces(insights=insights, namespaces=['ns*'], raw_bodies=[b2])
+    assert insights.namespaces == {'ns1'}
+    assert_logs(["Namespace 'ns1' termination pending: SomeReason: Some message"])
+
+
+def test_bodies_for_deletion_via_timestamp_with_false_conditions(registry, assert_logs):
+    b1 = RawBody(metadata={'name': 'ns1'})
+    b2 = RawBody(metadata={'name': 'ns1', 'deletionTimestamp': '...'},
+                 status={'conditions': [{'type': 'Whatever',
+                                         'status': 'False'}]})
+    insights = Insights()
+    revise_namespaces(insights=insights, namespaces=['ns*'], raw_bodies=[b1])
+    revise_namespaces(insights=insights, namespaces=['ns*'], raw_bodies=[b2])
     assert not insights.namespaces
+    assert_logs(prohibited=["Namespace '.*' termination pending:"])
 
 
 def test_bodies_ignored_for_mismatching(registry):
@@ -51,13 +77,39 @@ def test_events_for_additional_population(registry):
     assert insights.namespaces == {'ns1', 'ns2'}
 
 
-def test_events_for_deletion_via_timestamp(registry):
+def test_events_for_deletion_via_timestamp_without_conditions(registry):
     e1 = RawEvent(type=None, object=RawBody(metadata={'name': 'ns1'}))
     e2 = RawEvent(type=None, object=RawBody(metadata={'name': 'ns1', 'deletionTimestamp': '...'}))
     insights = Insights()
     revise_namespaces(insights=insights, namespaces=['ns*'], raw_events=[e1])
     revise_namespaces(insights=insights, namespaces=['ns*'], raw_events=[e2])
+    assert insights.namespaces == {'ns1'}
+
+
+def test_events_for_deletion_via_timestamp_with_true_conditions(registry, assert_logs):
+    e1 = RawEvent(type=None, object=RawBody(metadata={'name': 'ns1'}))
+    e2 = RawEvent(type=None, object=RawBody(metadata={'name': 'ns1', 'deletionTimestamp': '...'},
+                                            status={'conditions': [{'type': 'Whatever',
+                                                                    'status': 'True',
+                                                                    'reason': 'SomeReason',
+                                                                    'message': 'Some message'}]}))
+    insights = Insights()
+    revise_namespaces(insights=insights, namespaces=['ns*'], raw_events=[e1])
+    revise_namespaces(insights=insights, namespaces=['ns*'], raw_events=[e2])
+    assert insights.namespaces == {'ns1'}
+    assert_logs(["Namespace 'ns1' termination pending: SomeReason: Some message"])
+
+
+def test_events_for_deletion_via_timestamp_with_false_conditions(registry, assert_logs):
+    e1 = RawEvent(type=None, object=RawBody(metadata={'name': 'ns1'}))
+    e2 = RawEvent(type=None, object=RawBody(metadata={'name': 'ns1', 'deletionTimestamp': '...'},
+                                            status={'conditions': [{'type': 'Whatever',
+                                                                    'status': 'False'}]}))
+    insights = Insights()
+    revise_namespaces(insights=insights, namespaces=['ns*'], raw_events=[e1])
+    revise_namespaces(insights=insights, namespaces=['ns*'], raw_events=[e2])
     assert not insights.namespaces
+    assert_logs(prohibited=["Namespace '.*' termination pending:"])
 
 
 def test_events_for_deletion_via_event_type(registry):