Repository: nolar/kopf

Issue:
Endless touch-dummy Annotation Patching On Pod Causes High API Load
## Long story short

While looking into #685 I noticed weird patching behavior from kopf regarding the `touch-dummy` annotation. As soon as a pod reaches the completed state kopf constantly patches the `touch-dummy` annotation on the pod creating significant load on the API server.

## Description

The operator has a handler watching for completed pods. It is decorated like so:
```python
    @kopf.on.resume("", "v1", "pods", labels=cfg.test_pod_selector_label_dict)
    @kopf.on.field(
        "",
        "v1",
        "pods",
        id="await_pod_term",
        field="status.phase",
        labels=cfg.test_pod_selector_label_dict,
    )
```

There also is a 10sec interval timer handling unschedulable pods.
```python
    @kopf.on.timer(
        "",
        "v1",
        "pods",
        interval=cfg.watch_for_unschedulable_pod.interval,
        sharp=cfg.watch_for_unschedulable_pod.sharp,
        labels=cfg.test_pod_selector_label_dict,
        when=lambda namespace, name, status, **_: pods.get_pod_condition_or_none_from( 
            status, pods.PodCondTypeEnum.PodScheduled, pods.PodCondStatusEnum.false
        )
        is not None,
    )
```

<details>
<summary>
I have the following pod sitting in the completed state for now ~23h. (Showing only metadata as that should be the relevant part I guess)
</summary>

```yaml
apiVersion: v1
kind: Pod
metadata:
  annotations:
    kopf.zalando.org/await_pod_term.status.phase: '{"started":"2021-02-15T22:02:03.350703","stopped":"2021-02-15T22:02:03.353473","purpose":"update","retries"
:1,"success":true,"failure":false}'
    kopf.zalando.org/last-handled-configuration: |
      {"spec":{"volumes":[{"name":"test-config","configMap":{"name":"ts-job-1179-imuvqvkxczsjxvtpmau9pw","defaultMode":420}},{"name":"rasbuild","persistentVolumeClaim":{"claimName":"job-sesam-test-1179-rasbuild"}},{"name":"default-token-xlckc","secret":{"secretName":"default-token-xlckc","defaultMode":420}}],"conta
iners":[{"name":"ts-runner","image":"[redacted]","command":["/bin/sh"],"args":["-c","python [redacted]"],"env":[{"name":"TS_RABBITMQ_HOST","value":"[redacted]"},{"name":"TS_RABBITMQ_USER","value":"[redacted]"},{"name":"TS_RABBITMQ_PASSWORD","value":"[redacted]"}],"resources":{"limits":{"cpu":"2","memory":"2Gi"},"requests":{"cpu":"100m","memory":"1Gi"}},"volumeMounts":[{"name":"test-config","mountPath":"[redacted]"},{"name":"rasbuild","mountPath":"[redacted]"},{"name":"default-token-xlckc","readOnly":true,"mountPath":"/var/run/secrets/kubernetes.io/serviceaccount"}],"lifecycle":{"preStop":{"exec":{"command":["echo","'bye'"]}}},"terminationMessagePath":"/dev/termination-log","terminationMessagePolicy":"File","imagePullPolicy":"Always","securityContext":{"allowPrivilegeEscalation":false}},{"name":"ts-agent","image":"[redacted]","ports":[{"name":"agent-endpoint","containerPort":2233,"protocol":"TCP"}],"resources":
{"requests":{"cpu":"500m","memory":"128Mi"}},"volumeMounts":[{"name":"default-token-xlckc","readOnly":true,"mountPath":"/var/run/secrets/kubernetes.io/service
account"}],"lifecycle":{"preStop":{"exec":{"command":["echo","'bye'"]}}},"terminationMessagePath":"/dev/termination-log","terminationMessagePolicy":"File","imagePullPolicy":"Always","securityContext":{"allowPrivilegeEscalation":false}}],"restartPolicy":"Never","terminationGracePeriodSeconds":30,"dnsPolicy":"Cluster
First","serviceAccountName":"default","serviceAccount":"default","nodeName":"[redacted]","securityContext":{},"affinity":{"nodeAffinity":{"requiredDuringSchedulingIgnoredDuringExecution":{"nodeSelectorTerms":[{"matchExpressions":[{"key":"feature.node.kubernetes.io/cpu-cpuid.VMX","operator":"In","values":["true"]}]}]}}},"schedulerName":"default-scheduler","tolerations":[{"key":"node.kubernetes.io/not-ready","operator":"Exists","effect":"NoExecute","tolerationSeconds":300}
,"key":"node.kubernetes.io/unreachable","operator":"Exists","effect":"NoExecute","tolerationSeconds":300}],"priorityClassName":"ts-default-priority-intel","priority":1,"enableServiceLinks":false},"metadata":{"labels":{"app":"ts-runner","ts.job.name":"sesam-test","ts.job.run-id":"1179"}},"status":{"phase":"Running"}}
    kopf.zalando.org/mark_containters_ready: '{"started":"2021-02-15T22:02:03.350719","purpose":"update","retries":0,"success":false,"failure":false}'
    kopf.zalando.org/register_pod_handling.__locals__.watch_-P7tEVg: '{"started":"2021-02-16T10:03:13.216603","stopped":"2021-02-16T10:03:13.218999","purpose":"update","retries":1,"success":true,"failure":false}'
    kopf.zalando.org/register_pod_handling.__locals__.watch_for_completed_pod: '{"started":"2021-02-16T10:03:13.216603","stopped":"2021-02-16T10:03:13.218999","purpose":"update","retries":1,"success":true,"failure":false}'
    kopf.zalando.org/touch-dummy: 2021-02-16T21:23:57.630443
  creationTimestamp: "2021-02-15T22:01:34Z"
  finalizers:
  - kopf.zalando.org/KopfFinalizerMarker
  labels:
    app: ts-runner
    ts.job.name: sesam-test
    ts.job.run-id: "1179"
  name: ts-job-runner-1179-jcxbd3kbmruutrshxpky5k
  namespace: kopf-prototype-target
  ownerReferences:
  - apiVersion: [redacted]/v1
    blockOwnerDeletion: true
    controller: true
    kind: TsJob
    name: ts-job-1179-evuncgeuwcqrndyiqdsmk6
    uid: 007cc7a9-a1d7-4013-9579-e12edfe8988b
  resourceVersion: "5215406"
  selfLink: /api/v1/namespaces/kopf-prototype-target/pods/ts-job-runner-1179-jcxbd3kbmruutrshxpky5k
  uid: fba8f54e-2a75-4075-9420-734f92599477
...
[redacted]
...
```
</details>

As soon as kopf starts the `touch-dummy` updates start on the completed pod multiple times a second.
```
kubectl -n kopf-prototype-target get pod --watch --template='{{.metadata.name}}: {{index .metadata.annotations "kopf.zalando.org/touch-dummy"}}{{"\n"}}'
ts-job-runner-1179-jcxbd3kbmruutrshxpky5k: 2021-02-16T17:23:18.714713
ts-job-runner-1179-jcxbd3kbmruutrshxpky5k: 2021-02-16T17:23:18.886211
ts-job-runner-1179-jcxbd3kbmruutrshxpky5k: 2021-02-16T17:23:19.068260
ts-job-runner-1179-jcxbd3kbmruutrshxpky5k: 2021-02-16T17:23:19.251591
ts-job-runner-1179-jcxbd3kbmruutrshxpky5k: 2021-02-16T17:23:19.425195
ts-job-runner-1179-jcxbd3kbmruutrshxpky5k: 2021-02-16T17:23:19.596808
ts-job-runner-1179-jcxbd3kbmruutrshxpky5k: 2021-02-16T17:23:19.764809
ts-job-runner-1179-jcxbd3kbmruutrshxpky5k: 2021-02-16T17:23:19.933829
ts-job-runner-1179-jcxbd3kbmruutrshxpky5k: 2021-02-16T17:23:20.124813
ts-job-runner-1179-jcxbd3kbmruutrshxpky5k: 2021-02-16T17:23:20.307736
ts-job-runner-1179-jcxbd3kbmruutrshxpky5k: 2021-02-16T17:23:20.485288
ts-job-runner-1179-jcxbd3kbmruutrshxpky5k: 2021-02-16T17:23:20.665172
ts-job-runner-1179-jcxbd3kbmruutrshxpky5k: 2021-02-16T17:23:20.846178
ts-job-runner-1179-jcxbd3kbmruutrshxpky5k: 2021-02-16T17:23:21.022081
ts-job-runner-1179-jcxbd3kbmruutrshxpky5k: 2021-02-16T17:23:21.196700
ts-job-runner-1179-jcxbd3kbmruutrshxpky5k: 2021-02-16T17:23:21.370981
ts-job-runner-1179-jcxbd3kbmruutrshxpky5k: 2021-02-16T17:23:21.550282
ts-job-runner-1179-jcxbd3kbmruutrshxpky5k: 2021-02-16T17:23:21.718212
ts-job-runner-1179-jcxbd3kbmruutrshxpky5k: 2021-02-16T17:23:21.889359
...
```

Do you have any idea why that is happening? When debugging this it looked like every `MODIFIED` event on the pod triggers the patching. The patching however creates a new `MODIFIED` event, triggering another patch > a new event > rinse > repeat.


## Environment


* Kopf version: 1.29.2
* Kubernetes version: 1.17
* Python version: 3.9.1rc1
* OS/platform: Linux

<details><summary>Python packages installed</summary>
<!-- use `pip freeze --all` -->

```
aioconsole==0.3.1
aioelasticsearch==0.7.0
aiohttp==3.7.3
aiojobs==0.3.0
aiomonitor==0.4.5
alembic==1.4.3
amqp==5.0.2
appdirs==1.4.4
APScheduler==3.6.3
arrow==0.17.0
async-timeout==3.0.1
asynctest==0.13.0
asynqp==0.6
attrs==20.3.0
billiard==3.6.3.0
black==20.8b1
bleach==3.2.1
blinker==1.4
cachetools==4.1.1
celery==5.0.2
certifi==2020.11.8
cffi==1.14.4
chardet==3.0.4
click==5.1
click-didyoumean==0.0.3
click-repl==0.1.6
colorama==0.4.4
croniter==0.3.36
cryptography==3.2.1
distro==1.5.0
docker==4.4.0
docutils==0.16
elasticsearch==7.10.0
elasticsearch-dsl==7.3.0
fcache==0.4.7
flake8==3.8.4
flake8-codeclimate==0.3.0
flake8-comprehensions==3.3.0
Flask==1.1.2
freezegun==1.0.0
future==0.18.2
gitdb==4.0.5
GitPython==3.1.11
google-auth==1.23.0
httplib2==0.18.1
humanize==3.1.0
idna==2.10
iniconfig==1.1.1
iso8601==0.1.13
isort==5.6.4
itsdangerous==1.1.0
janus==0.6.1
jeepney==0.6.0
Jinja2==2.11.2
json-tricks==3.15.5
keyring==21.5.0
kombu==5.0.2
kopf==1.29.2
kubernetes==12.0.1
ldap3==2.8.1
lxml==4.6.2
lz4==3.1.1
Mako==1.1.3
MarkupSafe==1.1.1
matrix-client==0.3.2
mccabe==0.6.1
mmh3==2.5.1
multidict==5.1.0
mypy-extensions==0.4.3
natsort==7.1.0
oauth2client==4.1.3
oauthlib==3.1.0
opentelemetry-api==0.16b1
opentelemetry-exporter-jaeger==0.16b1
opentelemetry-instrumentation==0.16b1
opentelemetry-instrumentation-aiohttp-client==0.16b1
opentelemetry-instrumentation-sqlalchemy==0.16b1
opentelemetry-sdk==0.16b1
packaging==20.7
pathspec==0.8.1
pip==21.0.1
pkg-resources==0.0.0
pkginfo==1.6.1
pluggy==0.13.1
prometheus-client==0.9.0
prompt-toolkit==3.0.8
psutil==5.7.3
psycopg2-binary==2.8.6
py==1.9.0
pyasn1==0.4.8
pyasn1-modules==0.2.8
pycodestyle==2.6.0
pycparser==2.20
pydantic==1.7.3
pydevd-pycharm==203.5981.165
pyflakes==2.2.0
Pygments==2.7.2
pykube-ng==20.10.0
pyOpenSSL==20.0.0
pyparsing==2.4.7
pystack-debugger==0.9.0
pytest==6.1.2
pytest-aiohttp==0.3.0
pytest-asyncio==0.14.0
pytest-freezegun==0.4.2
pytest-mock==3.3.1
python-dateutil==2.8.1
python-editor==1.0.4
python-json-logger==2.0.1
pytz==2020.4
PyYAML==5.3.1
readme-renderer==28.0
regex==2020.11.13
requests==2.25.0
requests-oauthlib==1.3.0
requests-toolbelt==0.9.1
rfc3986==1.4.0
rsa==4.6
sd-notify==0.1.0
SecretStorage==3.3.0
sentry-sdk==0.19.4
setproctitle==1.2
setuptools==44.0.0
setuptools-scm==4.1.2
shortuuid==1.0.1
singleton-decorator==1.0.0
six==1.15.0
smmap==3.0.4
SQLAlchemy==1.3.20
tabulate==0.8.7
termcolor==1.1.0
terminaltables==3.1.0
TestLink-API-Python-client==0.8.1
thrift==0.13.0
toml==0.10.2
tqdm==4.54.0
twine==3.2.0
typed-ast==1.4.1
typing-extensions==3.7.4.3
tzlocal==2.1
UnleashClient==3.5.0
urllib3==1.26.2
uvloop==0.14.0
vine==5.0.0
voluptuous==0.12.0
wcwidth==0.2.5
webencodings==0.5.1
websocket-client==0.57.0
Werkzeug==1.0.1
wheel==0.36.0
wrapt==1.12.1
yarl==1.6.3
zstandard==0.14.0
zulip==0.7.1
```
</details>



Discussion:

Patch:
diff --git a/kopf/storage/states.py b/kopf/storage/states.py
index f654c00..aa21170 100644
--- a/kopf/storage/states.py
+++ b/kopf/storage/states.py
@@ -49,7 +49,13 @@ class HandlerState:
 
     Note the difference: `HandlerOutcome` is for in-memory results of handlers,
     which is then additionally converted before being storing as a state.
+
+    Active handler states are those used in .done/.delays for the current
+    handling cycle & the current cause. Passive handler states are those
+    carried over for logging of counts/extras, and for final state purging,
+    but not participating in the current handling cycle.
     """
+    active: Optional[bool] = None  # is it used in done/delays [T]? or only in counters/purges [F]?
     started: Optional[datetime.datetime] = None  # None means this information was lost.
     stopped: Optional[datetime.datetime] = None  # None means it is still running (e.g. delayed).
     delayed: Optional[datetime.datetime] = None  # None means it is finished (succeeded/failed).
@@ -64,6 +70,7 @@ class HandlerState:
     @classmethod
     def from_scratch(cls, *, purpose: Optional[handlers_.Reason] = None) -> "HandlerState":
         return cls(
+            active=True,
             started=datetime.datetime.utcnow(),
             purpose=purpose,
         )
@@ -71,6 +78,7 @@ class HandlerState:
     @classmethod
     def from_storage(cls, __d: progress.ProgressRecord) -> "HandlerState":
         return cls(
+            active=False,
             started=_datetime_fromisoformat(__d.get('started')) or datetime.datetime.utcnow(),
             stopped=_datetime_fromisoformat(__d.get('stopped')),
             delayed=_datetime_fromisoformat(__d.get('delayed')),
@@ -100,6 +108,9 @@ class HandlerState:
         # Nones are not stored by Kubernetes, so we filter them out for comparison.
         return {key: val for key, val in self.for_storage().items() if val is not None}
 
+    def as_active(self) -> "HandlerState":
+        return dataclasses.replace(self, active=True)
+
     def with_purpose(
             self,
             purpose: Optional[handlers_.Reason],
@@ -113,6 +124,7 @@ class HandlerState:
         now = datetime.datetime.utcnow()
         cls = type(self)
         return cls(
+            active=self.active,
             purpose=self.purpose,
             started=self.started if self.started else now,
             stopped=self.stopped if self.stopped else now if outcome.final else None,
@@ -213,6 +225,8 @@ class State(Mapping[ids.HandlerId, HandlerState]):
         for handler in handlers:
             if handler.id not in handler_states:
                 handler_states[handler.id] = HandlerState.from_scratch(purpose=self.purpose)
+            else:
+                handler_states[handler.id] = handler_states[handler.id].as_active()
         cls = type(self)
         return cls(handler_states, purpose=self.purpose)
 
@@ -286,8 +300,7 @@ class State(Mapping[ids.HandlerId, HandlerState]):
         # In particular, no handlers means that it is "done" even before doing.
         return all(
             handler_state.finished for handler_state in self._states.values()
-            if self.purpose is None or handler_state.purpose is None
-               or handler_state.purpose == self.purpose
+            if handler_state.active
         )
 
     @property
@@ -337,9 +350,7 @@ class State(Mapping[ids.HandlerId, HandlerState]):
         return [
             max(0, (handler_state.delayed - now).total_seconds()) if handler_state.delayed else 0
             for handler_state in self._states.values()
-            if not handler_state.finished
-            if self.purpose is None or handler_state.purpose is None
-               or handler_state.purpose == self.purpose
+            if handler_state.active and not handler_state.finished
         ]
 
 


Test patch:
diff --git a/tests/persistence/test_states.py b/tests/persistence/test_states.py
index 91d8524..68e7c13 100644
--- a/tests/persistence/test_states.py
+++ b/tests/persistence/test_states.py
@@ -83,6 +83,81 @@ def test_created_empty_from_filled_storage_without_handlers(storage, handler, bo
     assert state.extras == {}
 
 
+#
+# Active/passive states.
+#
+
+
+def test_created_from_storage_as_passive(storage, handler):
+    body = {'status': {'kopf': {'progress': {'some-id': {}}}}}
+    state = State.from_storage(body=Body(body), handlers=[handler], storage=storage)
+    assert len(state) == 1
+    assert state['some-id'].active is False
+
+
+def test_created_from_handlers_as_active(storage, handler):
+    state = State.from_scratch()
+    state = state.with_handlers([handler])
+    assert len(state) == 1
+    assert state['some-id'].active is True
+
+
+def test_switched_from_passive_to_active(storage, handler):
+    body = {'status': {'kopf': {'progress': {'some-id': {'purpose': None}}}}}
+    state = State.from_storage(body=Body(body), handlers=[handler], storage=storage)
+    state = state.with_handlers([handler])
+    assert len(state) == 1
+    assert state['some-id'].active is True
+
+
+def test_passed_through_with_outcomes_when_passive(storage, handler):
+    body = {'status': {'kopf': {'progress': {'some-id': {'purpose': None}}}}}
+    state = State.from_storage(body=Body(body), handlers=[handler], storage=storage)
+    state = state.with_outcomes({'some-id': HandlerOutcome(final=True)})
+    assert len(state) == 1
+    assert state['some-id'].active is False
+
+
+def test_passed_through_with_outcomes_when_active(storage, handler):
+    state = State.from_scratch()
+    state = state.with_handlers([handler])
+    state = state.with_outcomes({'some-id': HandlerOutcome(final=True)})
+    assert len(state) == 1
+    assert state['some-id'].active is True
+
+
+def test_passive_states_are_not_used_in_done_calculation(storage, handler):
+    body = {'status': {'kopf': {'progress': {'some-id': {}}}}}
+    state = State.from_storage(body=Body(body), handlers=[handler], storage=storage)
+    assert len(state) == 1
+    assert state.done is True  # because the unfinished handler state is ignored
+
+
+def test_active_states_are_used_in_done_calculation(storage, handler):
+    body = {'status': {'kopf': {'progress': {'some-id': {}}}}}
+    state = State.from_storage(body=Body(body), handlers=[handler], storage=storage)
+    state = state.with_handlers([handler])
+    assert len(state) == 1
+    assert state.done is False
+
+
+@freezegun.freeze_time(TS0)
+def test_passive_states_are_not_used_in_delays_calculation(storage, handler):
+    body = {'status': {'kopf': {'progress': {'some-id': {'delayed': TS1_ISO}}}}}
+    state = State.from_storage(body=Body(body), handlers=[handler], storage=storage)
+    assert len(state) == 1
+    assert state.delays == []
+
+
+@freezegun.freeze_time(TS0)
+def test_active_states_are_used_in_delays_calculation(storage, handler):
+    body = {'status': {'kopf': {'progress': {'some-id': {'delayed': TS1_ISO}}}}}
+    state = State.from_storage(body=Body(body), handlers=[handler], storage=storage)
+    state = state.with_handlers([handler])
+    assert len(state) == 1
+    assert state.delays == [1.0]
+
+
 #
 # Purpose propagation and re-purposing of the states (overall and per-handler):
 #
@@ -205,7 +280,7 @@ def test_with_handlers_relevant_to_the_purpose(
         storage, handler, body, expected_counts, expected_done, expected_delays, reason):
     body['status']['kopf']['progress']['some-id']['purpose'] = reason.value
     state = State.from_storage(body=Body(body), handlers=[handler], storage=storage)
-    state = state.with_purpose(reason)
+    state = state.with_purpose(reason).with_handlers([handler])
     assert len(state) == 1
     assert state.extras == {}
     assert state.counts == expected_counts
@@ -225,7 +300,7 @@ def test_with_handlers_relevant_to_the_purpose_and_delayed(
     body['status']['kopf']['progress']['some-id']['delayed'] = TS1_ISO
     body['status']['kopf']['progress']['some-id']['purpose'] = reason.value
     state = State.from_storage(body=Body(body), handlers=[handler], storage=storage)
-    state = state.with_purpose(reason)
+    state = state.with_purpose(reason).with_handlers([handler])
     assert len(state) == 1
     assert state.extras == {}
     assert state.counts == expected_counts