kopf
kopf copied to clipboard
Any way to hook into errors in handlers?
Keywords
handlers subhandlers error_handling errors error_hook
Problem
In handlers, i.e.:
@kopf.on.update("my_api_group", "v1", "my_resource", timeout=600)
async def my_on_update(
namespace: str,
name: str,
patch: kopf.Patch,
status: kopf.Status,
diff: kopf.Diff,
**_kwargs,
):
...
is there a way to know when the timeout of the handler actually happens (or any other PermanentError), to, i.e., get notified about it? We would like to know when a handler fails permanently so we can take remedial actions. There doesn't seem to be a way to do that...
i.e. thinking of some sort of:
def my_error_handler(exc):
...
@kopf.on.update("my_api_group", "v1", "my_resource", timeout=600, on_error=my_error_handler)
async def my_on_update(
namespace: str,
name: str,
patch: kopf.Patch,
status: kopf.Status,
diff: kopf.Diff,
**_kwargs,
):
...
Thanks!
Hello. No, there is no such thing as handlers for handlers.
But you can replace the "out-of-the-box" timeout= with your own implementation:
- https://kopf.readthedocs.io/en/stable/kwargs/#retrying-and-timing
import kopf, datetime, contextlib
@kopf.on.update("my_api_group", "v1", "my_resource", )
async def my_on_update(
runtime: datetime.timedelta,
retry: int,
**_,
):
if runtime >= datetime.timedelta(minutes=5) or retry >= 10:
with contextlib.suppress(Exception): # to prevent retries in case of errors in the error handler
my_error_handler(...)
raise kopf.PermanentError("Timed out or out of retries.")
...
You can make this into your own decorator if you wish.