Repository navigation
Expand file tree
/
Copy pathexecution.py
More file actions
384 lines (303 loc) · 14.4 KB
/
Copy pathexecution.py
File metadata and controls
384 lines (303 loc) · 14.4 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
"""
Execution of pre-selected handlers, in batches or individually.
These functions are invoked from `kopf._core.reactor.processing`,
where the raw watch-events are interpreted and wrapped into extended *causes*.
The handler execution can also be used in other places, such as in-memory
activities, when there is no underlying Kubernetes object to patch'n'watch.
"""
import asyncio
import contextlib
import dataclasses
import datetime
import enum
from contextvars import ContextVar
from typing import Any, AsyncContextManager, AsyncIterator, Callable, Collection, Iterable, \
Mapping, MutableMapping, NewType, Optional, Sequence, Set, TypeVar
from typing_extensions import Protocol
from kopf._cogs.configs import configuration
from kopf._cogs.helpers import typedefs
from kopf._cogs.structs import ids
from kopf._core.actions import invocation
# The default delay duration for the regular exception in retry-mode.
DEFAULT_RETRY_DELAY = 1 * 60
class PermanentError(Exception):
""" A fatal handler error, the retries are useless. """
class TemporaryError(Exception):
""" A potentially recoverable error, should be retried. """
def __init__(
self,
__msg: Optional[str] = None,
delay: Optional[float] = DEFAULT_RETRY_DELAY,
) -> None:
super().__init__(__msg)
self.delay = delay
class HandlerTimeoutError(PermanentError):
""" An error for the handler's timeout (if set). """
class HandlerRetriesError(PermanentError):
""" An error for the handler's retries exceeded (if set). """
class HandlerChildrenRetry(TemporaryError):
""" An internal pseudo-error to retry for the next sub-handlers attempt. """
class ErrorsMode(enum.Enum):
""" How arbitrary (non-temporary/non-permanent) exceptions are treated. """
IGNORED = enum.auto()
TEMPORARY = enum.auto()
PERMANENT = enum.auto()
# A specialised type to highlight the purpose or origin of the data of type Any,
# to not be mixed with other arbitrary Any values, where it is indeed "any".
Result = NewType('Result', object)
@dataclasses.dataclass(frozen=True)
class Outcome:
"""
An in-memory outcome of one single invocation of one single handler.
Conceptually, an outcome is similar to the async futures, but some cases
are handled specially: e.g., the temporary errors have exceptions,
but the handler should be retried later, unlike with the permanent errors.
Note the difference: `HandlerState` is a persistent state of the handler,
possibly after a few executions, and consisting of simple data types
(for YAML/JSON serialisation) rather than the actual in-memory objects.
"""
final: bool
delay: Optional[float] = None
result: Optional[Result] = None
exception: Optional[Exception] = None
subrefs: Collection[ids.HandlerId] = ()
@dataclasses.dataclass(frozen=True)
class HandlerState:
"""
A persisted state of a single handler, as stored on the resource's status.
Note the difference: `Outcome` 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.
"""
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).
retries: int = 0
success: bool = False
failure: bool = False
@property
def finished(self) -> bool:
return bool(self.success or self.failure)
@property
def sleeping(self) -> bool:
ts = self.delayed
now = datetime.datetime.utcnow()
return not self.finished and ts is not None and ts > now
@property
def awakened(self) -> bool:
return bool(not self.finished and not self.sleeping)
@property
def runtime(self) -> datetime.timedelta:
now = datetime.datetime.utcnow()
return now - (self.started if self.started else now)
class State(Mapping[ids.HandlerId, HandlerState]):
pass
@dataclasses.dataclass
class Cause(invocation.Kwargable):
""" Base non-specific cause as used in the framework's reactor. """
logger: typedefs.Logger
@property
def _kwargs(self) -> Mapping[str, Any]:
# Similar to `dataclasses.asdict()`, but not recursive for other dataclasses.
return {field.name: getattr(self, field.name) for field in dataclasses.fields(self)}
CauseT = TypeVar('CauseT', bound=Cause)
@dataclasses.dataclass(frozen=True)
class Handler:
""" A handler is a function bound with its behavioral constraints. """
id: ids.HandlerId
fn: invocation.Invokable
param: Optional[Any]
errors: Optional[ErrorsMode]
timeout: Optional[float]
retries: Optional[int]
backoff: Optional[float]
# Used in the logs. Overridden in some (but not all) handler types for better log messages.
def __str__(self) -> str:
return f"Handler {self.id!r}"
# Overridden in handlers with fields for causes with field-specific old/new/diff.
def adjust_cause(self, cause: CauseT) -> CauseT:
return cause
class LifeCycleFn(Protocol):
""" A callback type for handlers selection based on the event/cause. """
def __call__(
self,
handlers: Sequence[Handler],
*,
state: State,
**kwargs: Any,
) -> Sequence[Handler]: ...
# The task-local context; propagated down the stack instead of multiple kwargs.
# Used in `@kopf.subhandler` and `kopf.execute()` to add/get the sub-handlers.
sublifecycle_var: ContextVar[Optional[LifeCycleFn]] = ContextVar('sublifecycle_var')
subsettings_var: ContextVar[configuration.OperatorSettings] = ContextVar('subsettings_var')
subrefs_var: ContextVar[Iterable[Set[ids.HandlerId]]] = ContextVar('subrefs_var')
handler_var: ContextVar[Handler] = ContextVar('handler_var')
cause_var: ContextVar[Cause] = ContextVar('cause_var')
ExtraContext = Callable[[], AsyncContextManager[None]]
@contextlib.asynccontextmanager
async def no_extra_context() -> AsyncIterator[None]:
yield
async def execute_handlers_once(
lifecycle: LifeCycleFn,
settings: configuration.OperatorSettings,
handlers: Collection[Handler],
cause: Cause,
state: State,
extra_context: ExtraContext = no_extra_context,
default_errors: ErrorsMode = ErrorsMode.TEMPORARY,
) -> Mapping[ids.HandlerId, Outcome]:
"""
Call the next handler(s) from the chain of the handlers.
Keep the record on the progression of the handlers in the object's state,
and use it on the next invocation to determined which handler(s) to call.
This routine is used both for the global handlers (via global registry),
and for the sub-handlers (via a simple registry of the current handler).
"""
# Filter and select the handlers to be executed right now, on this event reaction cycle.
handlers_todo = [h for h in handlers if state[h.id].awakened]
handlers_plan = lifecycle(handlers_todo, state=state, **cause.kwargs)
# Execute all planned (selected) handlers in one event reaction cycle, even if there are a few.
outcomes: MutableMapping[ids.HandlerId, Outcome] = {}
for handler in handlers_plan:
outcome = await execute_handler_once(
settings=settings,
handler=handler,
state=state[handler.id],
cause=cause,
lifecycle=lifecycle, # just a default for the sub-handlers, not used directly.
extra_context=extra_context,
default_errors=default_errors,
)
outcomes[handler.id] = outcome
return outcomes
async def execute_handler_once(
settings: configuration.OperatorSettings,
handler: Handler,
cause: Cause,
state: HandlerState,
lifecycle: Optional[LifeCycleFn] = None,
extra_context: ExtraContext = no_extra_context,
default_errors: ErrorsMode = ErrorsMode.TEMPORARY,
) -> Outcome:
"""
Execute one and only one handler for one and only one time.
*Execution* means not just *calling* the handler in properly set context
(see `_call_handler`), but also interpreting its result and errors, and
wrapping them into am `Outcome` object -- to be stored in the state.
The *execution* can be long -- depending on how the handler is implemented.
For daemons, it is normal to run for hours and days if needed.
This is different from the regular handlers, which are supposed
to be finished as soon as possible.
This method is not supposed to raise any exceptions from the handlers:
exceptions mean the failure of execution itself.
"""
errors_mode = handler.errors if handler.errors is not None else default_errors
backoff = handler.backoff if handler.backoff is not None else DEFAULT_RETRY_DELAY
logger = cause.logger
# Mutable accumulator for all the sub-handlers of any level deep; populated in `kopf.execute`.
subrefs: Set[ids.HandlerId] = set()
# The exceptions are handled locally and are not re-raised, to keep the operator running.
try:
logger.debug(f"{handler} is invoked.")
if handler.timeout is not None and state.runtime.total_seconds() >= handler.timeout:
raise HandlerTimeoutError(f"{handler} has timed out after {state.runtime}.")
if handler.retries is not None and state.retries >= handler.retries:
raise HandlerRetriesError(f"{handler} has exceeded {state.retries} retries.")
result = await invoke_handler(
handler=handler,
cause=cause,
retry=state.retries,
started=state.started or datetime.datetime.utcnow(), # "or" is for type-checking.
runtime=state.runtime,
settings=settings,
lifecycle=lifecycle, # just a default for the sub-handlers, not used directly.
extra_context=extra_context,
subrefs=subrefs,
)
# The cancellations are an excepted way of stopping the handler. Especially for daemons.
except asyncio.CancelledError:
logger.warning(f"{handler} is cancelled. Will escalate.")
raise
# Unfinished children cause the regular retry, but with less logging and event reporting.
except HandlerChildrenRetry as e:
logger.debug(f"{handler} has unfinished sub-handlers. Will retry soon.")
return Outcome(final=False, exception=e, delay=e.delay, subrefs=subrefs)
# Definitely a temporary error, regardless of the error strictness.
except TemporaryError as e:
logger.error(f"{handler} failed temporarily: {str(e) or repr(e)}")
return Outcome(final=False, exception=e, delay=e.delay, subrefs=subrefs)
# Same as permanent errors below, but with better logging for our internal cases.
except HandlerTimeoutError as e:
logger.error(f"{str(e) or repr(e)}") # already formatted
return Outcome(final=True, exception=e, subrefs=subrefs)
# TODO: report the handling failure somehow (beside logs/events). persistent status?
# Definitely a permanent error, regardless of the error strictness.
except PermanentError as e:
logger.error(f"{handler} failed permanently: {str(e) or repr(e)}")
return Outcome(final=True, exception=e, subrefs=subrefs)
# TODO: report the handling failure somehow (beside logs/events). persistent status?
# Regular errors behave as either temporary or permanent depending on the error strictness.
except Exception as e:
if errors_mode == ErrorsMode.IGNORED:
logger.exception(f"{handler} failed with an exception. Will ignore.")
return Outcome(final=True, subrefs=subrefs)
elif errors_mode == ErrorsMode.TEMPORARY:
logger.exception(f"{handler} failed with an exception. Will retry.")
return Outcome(final=False, exception=e, delay=backoff, subrefs=subrefs)
elif errors_mode == ErrorsMode.PERMANENT:
logger.exception(f"{handler} failed with an exception. Will stop.")
return Outcome(final=True, exception=e, subrefs=subrefs)
# TODO: report the handling failure somehow (beside logs/events). persistent status?
else:
raise RuntimeError(f"Unknown mode for errors: {errors_mode!r}")
# No errors means the handler should be excluded from future runs in this reaction cycle.
else:
logger.info(f"{handler} succeeded.")
return Outcome(final=True, result=result, subrefs=subrefs)
async def invoke_handler(
*,
handler: Handler,
cause: Cause,
retry: int,
started: datetime.datetime,
runtime: datetime.timedelta,
settings: configuration.OperatorSettings,
lifecycle: Optional[LifeCycleFn],
subrefs: Set[ids.HandlerId],
extra_context: ExtraContext,
) -> Optional[Result]:
"""
Invoke one handler only, according to the calling conventions.
Specifically, calculate the handler-specific fields (e.g. field diffs).
Ensure the global context for this asyncio task is set to the handler and
its cause -- for proper population of the sub-handlers via the decorators
(see `@kopf.subhandler`).
"""
# For the field-handlers, the old/new/diff values must match the field, not the whole object.
cause = handler.adjust_cause(cause)
# The context makes it possible and easy to pass the kwargs _through_ the user-space handlers:
# from the framework to the framework's helper functions (e.g. sub-handling, hierarchies, etc).
with invocation.context([
(sublifecycle_var, lifecycle),
(subsettings_var, settings),
(subrefs_var, list(subrefs_var.get([])) + [subrefs]),
(handler_var, handler),
(cause_var, cause),
]):
async with extra_context():
result = await invocation.invoke(
handler.fn,
settings=settings,
kwargsrc=cause,
kwargs=dict(
param=handler.param,
retry=retry,
started=started,
runtime=runtime,
),
)
# Since we know that we invoked the handler, we cast "any" result to a handler result.
return Result(result)