11903476210

Committed 19 Nov 2024 12:07AM UTC coverage: 89.917% (+0.4%) from 89.481%

Build # 11903476210

Build Type

Pull #1137

github

Committed by

web-flow

Commit Message

Merge 6260e59a7 into 6b8ce27a8

Pull Request Pull Request #1137: Upgrade the CI tooling (Ubuntu to 24.04, MyPy to 1.13.0, etc)

Run Details

2122 of 2428 branches covered (87.4%)

Branch coverage included in aggregate %.

4 of 4 new or added lines in 1 file covered. (100.0%)

2 existing lines in 2 files now uncovered.

5494 of 6042 relevant lines covered (90.93%)

11.72 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

89.6

/kopf/_core/reactor/orchestration.py

"""
Orchestrating the tasks for served resources & namespaces.

The resources & namespaces are observed in :mod:`.observation`, where they
are stored in the "insights" -- a description of the current cluster setup.
They are used as the input for the orchestration.

For every combination of every actual resource & every actual namespace,
there is a watcher task and a few optional peering tasks/toggles.
The tasks are started when new values are added to any of these dimension,
or stopped when some existing values are removed.

There are several kinds of tasks:

* Regular watchers (watch-streams) -- the main one.
* Peering watchers (watch-streams).
* Peering keep-alives (pingers).

The peering tasks are started only when the peering is enabled at all.
For peering, the resource is not used, only the namespace is of importance.

Some special watchers for the meta-level resources -- i.e. for dimensions --
are started and stopped separately, not as part of the the orchestration.
"""
import asyncio
import dataclasses
import functools
import itertools
import logging
from typing import Any, Collection, Container, Dict, Iterable, \
                   MutableMapping, NamedTuple, Optional, Protocol

from kopf._cogs.aiokits import aiotasks, aiotoggles
from kopf._cogs.configs import configuration
from kopf._cogs.structs import bodies, references
from kopf._core.engines import peering
from kopf._core.reactor import queueing

logger = logging.getLogger(__name__)


class EnsembleKey(NamedTuple):
    resource: references.Resource
    namespace: references.Namespace


# Differs from queueing.WatchStreamProcessor by the resource=… kwarg.
class ResourceWatchStreamProcessor(Protocol):
    async def __call__(
            self,
            *,
            resource: references.Resource,
            raw_event: bodies.RawEvent,
            stream_pressure: Optional[asyncio.Event] = None,  # None for tests
            resource_indexed: Optional[aiotoggles.Toggle] = None,  # None for tests & observation
            operator_indexed: Optional[aiotoggles.ToggleSet] = None,  # None for tests & observation
    ) -> None: ...


@dataclasses.dataclass
class Ensemble:

    # Global synchronisation point on the cache pre-populating stage and overall cache readiness.
    # Note: there is no need for ToggleSet; it is checked by emptiness of items inside.
    #       ToggleSet is used because it is the closest equivalent of such a primitive.
    operator_indexed: aiotoggles.ToggleSet

    # Multidimentional pausing: for every namespace, and a few for the whole cluster (for CRDs).
    operator_paused: aiotoggles.ToggleSet
    peering_missing: aiotoggles.Toggle
    conflicts_found: Dict[EnsembleKey, aiotoggles.Toggle] = dataclasses.field(default_factory=dict)

    # Multidimensional tasks -- one for every combination of relevant dimensions.
    watcher_tasks: Dict[EnsembleKey, aiotasks.Task] = dataclasses.field(default_factory=dict)
    peering_tasks: Dict[EnsembleKey, aiotasks.Task] = dataclasses.field(default_factory=dict)
    pinging_tasks: Dict[EnsembleKey, aiotasks.Task] = dataclasses.field(default_factory=dict)

    def get_keys(self) -> Collection[EnsembleKey]:
        return (frozenset(self.watcher_tasks) |
                frozenset(self.peering_tasks) |
                frozenset(self.pinging_tasks) |
                frozenset(self.conflicts_found))

    def get_tasks(self, keys: Container[EnsembleKey]) -> Collection[aiotasks.Task]:
        return {task
                for tasks in [self.watcher_tasks, self.peering_tasks, self.pinging_tasks]
                for key, task in tasks.items() if key in keys}

    def get_flags(self, keys: Container[EnsembleKey]) -> Collection[aiotoggles.Toggle]:
        return {toggle for key, toggle in self.conflicts_found.items() if key in keys}

    def del_keys(self, keys: Container[EnsembleKey]) -> None:
        d: MutableMapping[EnsembleKey, Any]
        for d in [self.watcher_tasks, self.peering_tasks, self.pinging_tasks]:
            for key in set(d):
                if key in keys:
                    del d[key]
        for d in [self.conflicts_found]:  # separated for easier type inferrence
            for key in set(d):
                if key in keys:
                    del d[key]


async def ochestrator(
        *,
        processor: ResourceWatchStreamProcessor,
        settings: configuration.OperatorSettings,
        identity: peering.Identity,
        insights: references.Insights,
        operator_paused: aiotoggles.ToggleSet,
) -> None:
    peering_missing = await operator_paused.make_toggle(name='peering CRD is missing')
    ensemble = Ensemble(
        peering_missing=peering_missing,
        operator_paused=operator_paused,
        operator_indexed=aiotoggles.ToggleSet(all),
    )
    try:
        async with insights.revised:
            while True:
                await insights.revised.wait()
                await adjust_tasks(
                    processor=processor,
                    insights=insights,
                    settings=settings,
                    identity=identity,
                    ensemble=ensemble,
                )
    except asyncio.CancelledError:
        tasks = ensemble.get_tasks(ensemble.get_keys())
        await aiotasks.stop(tasks, title="streaming", logger=logger, interval=10)
        raise


# Directly corresponds to one iteration of an orchestrator, but it is extracted for testability:
# for a simulation of the insights (inputs) and an assertion of the tasks & toggles (outputs).
async def adjust_tasks(
        *,
        processor: ResourceWatchStreamProcessor,
        insights: references.Insights,
        settings: configuration.OperatorSettings,
        identity: peering.Identity,
        ensemble: Ensemble,
) -> None:
    peering_selectors = peering.guess_selectors(settings=settings)
    peering_resources = {insights.backbone[s] for s in peering_selectors if s in insights.backbone}

    # Pause or resume all streams if the peering CRDs are absent but required.
    # Ignore the CRD absence in auto-detection mode: pause only when (and if) the CRDs are added.
    await ensemble.peering_missing.turn_to(settings.peering.mandatory and not peering_resources)

    # Stop & start the tasks to match the task matrix with the cluster insights.
    # As a rule of thumb, stop the tasks first, start later -- not vice versa!
    await terminate_redundancies(ensemble=ensemble,
                                 remaining_resources=insights.watched_resources | peering_resources,
                                 remaining_namespaces=insights.namespaces | {None})
    await spawn_missing_peerings(ensemble=ensemble,
                                 settings=settings,
                                 identity=identity,
                                 resources=peering_resources,
                                 namespaces=insights.namespaces)
    await spawn_missing_watchers(ensemble=ensemble,
                                 settings=settings,
                                 processor=processor,
                                 indexed_resources=insights.indexed_resources,
                                 watched_resources=insights.watched_resources,
                                 watched_namespaces=insights.namespaces)


async def terminate_redundancies(
        *,
        remaining_resources: Collection[references.Resource],
        remaining_namespaces: Collection[references.Namespace],
        ensemble: Ensemble,
) -> None:
    # Do not distinguish the keys: even for the case when the peering CRD is served by the operator,
    # for the peering CRD or namespace deletion, both tasks are stopped together, never apart.
    redundant_keys = {key for key in ensemble.get_keys()
                      if key.namespace not in remaining_namespaces
                      or key.resource not in remaining_resources}
    redundant_tasks = ensemble.get_tasks(redundant_keys)
    redundant_flags = ensemble.get_flags(redundant_keys)
    await aiotasks.stop(redundant_tasks, title="streaming", logger=logger, interval=10, quiet=True)
    await ensemble.operator_paused.drop_toggles(redundant_flags)
    ensemble.del_keys(redundant_keys)


async def spawn_missing_peerings(
        *,
        settings: configuration.OperatorSettings,
        identity: peering.Identity,
        resources: Collection[references.Resource],
        namespaces: Collection[references.Namespace],
        ensemble: Ensemble,
) -> None:
    for resource, namespace in itertools.product(resources, namespaces):
        namespace = namespace if resource.namespaced else None
        dkey = EnsembleKey(resource=resource, namespace=namespace)
        if dkey not in ensemble.peering_tasks:
            what = f"{settings.peering.name}@{namespace}"
            is_preactivated = settings.peering.mandatory
            conflicts_found = await ensemble.operator_paused.make_toggle(is_preactivated, name=what)
            ensemble.conflicts_found[dkey] = conflicts_found
            ensemble.pinging_tasks[dkey] = aiotasks.create_guarded_task(
                name=f"peering keep-alive for {what}", logger=logger, cancellable=True,
                coro=peering.keepalive(
                    namespace=namespace,
                    resource=resource,
                    settings=settings,
                    identity=identity))
            ensemble.peering_tasks[dkey] = aiotasks.create_guarded_task(
                name=f"peering observer for {what}", logger=logger, cancellable=True,
                coro=queueing.watcher(
                    settings=settings,
                    resource=resource,
                    namespace=namespace,
                    processor=functools.partial(peering.process_peering_event,
                                                conflicts_found=conflicts_found,
                                                namespace=namespace,
                                                resource=resource,
                                                settings=settings,
                                                identity=identity)))

    # Ensure that all guarded tasks got control for a moment to enter the guard.
    await asyncio.sleep(0)


async def spawn_missing_watchers(
        *,
        processor: ResourceWatchStreamProcessor,
        settings: configuration.OperatorSettings,
        indexed_resources: Container[references.Resource],  # only "if in", never "for in"!
        watched_resources: Iterable[references.Resource],
        watched_namespaces: Iterable[references.Namespace],
        ensemble: Ensemble,
) -> None:

    # Block the operator globally until specialised per-resource-kind blockers are created.
    # NB: Must be created before the point of parallelisation!
    operator_blocked = await ensemble.operator_indexed.make_toggle(name="orchestration blocker")

    # Spawn watchers and create the specialised per-resource-kind blockers.
    for resource, namespace in itertools.product(watched_resources, watched_namespaces):
        namespace = namespace if resource.namespaced else None
        dkey = EnsembleKey(resource=resource, namespace=namespace)
        if dkey not in ensemble.watcher_tasks:
            what = f"{resource}@{namespace}"
            resource_indexed: Optional[aiotoggles.Toggle] = None
            if resource in indexed_resources:
                resource_indexed = await ensemble.operator_indexed.make_toggle(name=what)
            ensemble.watcher_tasks[dkey] = aiotasks.create_guarded_task(
                name=f"watcher for {what}", logger=logger, cancellable=True,
                coro=queueing.watcher(
                    operator_paused=ensemble.operator_paused,
                    operator_indexed=ensemble.operator_indexed,
                    resource_indexed=resource_indexed,
                    settings=settings,
                    resource=resource,
                    namespace=namespace,
                    processor=functools.partial(processor, resource=resource)))

    # Unblock globally, let the specialised per-resource-kind blockers hold the readiness.
    await ensemble.operator_indexed.drop_toggle(operator_blocked)

    # Ensure that all guarded tasks got control for a moment to enter the guard.
    await asyncio.sleep(0)

1	"""
2	Orchestrating the tasks for served resources & namespaces.
3
4	The resources & namespaces are observed in :mod:`.observation`, where they
5	are stored in the "insights" -- a description of the current cluster setup.
6	They are used as the input for the orchestration.
7
8	For every combination of every actual resource & every actual namespace,
9	there is a watcher task and a few optional peering tasks/toggles.
10	The tasks are started when new values are added to any of these dimension,
11	or stopped when some existing values are removed.
12
13	There are several kinds of tasks:
14
15	* Regular watchers (watch-streams) -- the main one.
16	* Peering watchers (watch-streams).
17	* Peering keep-alives (pingers).
18
19	The peering tasks are started only when the peering is enabled at all.
20	For peering, the resource is not used, only the namespace is of importance.
21
22	Some special watchers for the meta-level resources -- i.e. for dimensions --
23	are started and stopped separately, not as part of the the orchestration.
24	"""
25	import asyncio	13✔
26	import dataclasses	13✔
27	import functools	13✔
28	import itertools	13✔
29	import logging	13✔
30	from typing import Any, Collection, Container, Dict, Iterable, \	13✔
31	MutableMapping, NamedTuple, Optional, Protocol
32
33	from kopf._cogs.aiokits import aiotasks, aiotoggles	13✔
34	from kopf._cogs.configs import configuration	13✔
35	from kopf._cogs.structs import bodies, references	13✔
36	from kopf._core.engines import peering	13✔
37	from kopf._core.reactor import queueing	13✔
38
39	logger = logging.getLogger(__name__)	13✔
40
41
42	class EnsembleKey(NamedTuple):	13✔
43	resource: references.Resource	13✔
44	namespace: references.Namespace	13✔
45
46
47	# Differs from queueing.WatchStreamProcessor by the resource=… kwarg.
48	class ResourceWatchStreamProcessor(Protocol):	13✔
49	async def __call__(	13!
50	self,
51	*,
52	resource: references.Resource,
53	raw_event: bodies.RawEvent,
54	stream_pressure: Optional[asyncio.Event] = None, # None for tests
55	resource_indexed: Optional[aiotoggles.Toggle] = None, # None for tests & observation
56	operator_indexed: Optional[aiotoggles.ToggleSet] = None, # None for tests & observation
57	) -> None: ...
58
59
60	@dataclasses.dataclass	13✔
61	class Ensemble:	13✔
62
63	# Global synchronisation point on the cache pre-populating stage and overall cache readiness.
64	# Note: there is no need for ToggleSet; it is checked by emptiness of items inside.
65	# ToggleSet is used because it is the closest equivalent of such a primitive.
66	operator_indexed: aiotoggles.ToggleSet	13✔
67
68	# Multidimentional pausing: for every namespace, and a few for the whole cluster (for CRDs).
69	operator_paused: aiotoggles.ToggleSet	13✔
70	peering_missing: aiotoggles.Toggle	13✔
71	conflicts_found: Dict[EnsembleKey, aiotoggles.Toggle] = dataclasses.field(default_factory=dict)	13✔
72
73	# Multidimensional tasks -- one for every combination of relevant dimensions.
74	watcher_tasks: Dict[EnsembleKey, aiotasks.Task] = dataclasses.field(default_factory=dict)	13✔
75	peering_tasks: Dict[EnsembleKey, aiotasks.Task] = dataclasses.field(default_factory=dict)	13✔
76	pinging_tasks: Dict[EnsembleKey, aiotasks.Task] = dataclasses.field(default_factory=dict)	13✔
77
78	def get_keys(self) -> Collection[EnsembleKey]:	13✔
79	return (frozenset(self.watcher_tasks) \|	13✔
80	frozenset(self.peering_tasks) \|
81	frozenset(self.pinging_tasks) \|
82	frozenset(self.conflicts_found))
83
84	def get_tasks(self, keys: Container[EnsembleKey]) -> Collection[aiotasks.Task]:	13✔
85	return {task	13✔
86	for tasks in [self.watcher_tasks, self.peering_tasks, self.pinging_tasks]
87	for key, task in tasks.items() if key in keys}
88
89	def get_flags(self, keys: Container[EnsembleKey]) -> Collection[aiotoggles.Toggle]:	13✔
90	return {toggle for key, toggle in self.conflicts_found.items() if key in keys}	13✔
91
92	def del_keys(self, keys: Container[EnsembleKey]) -> None:	13✔
93	d: MutableMapping[EnsembleKey, Any]
94	for d in [self.watcher_tasks, self.peering_tasks, self.pinging_tasks]:	13✔
95	for key in set(d):	13✔
96	if key in keys:	13✔
97	del d[key]	13✔
98	for d in [self.conflicts_found]: # separated for easier type inferrence	13✔
99	for key in set(d):	13✔
100	if key in keys:	13✔
101	del d[key]	13✔
102
103
104	async def ochestrator(	13✔
105	*,
106	processor: ResourceWatchStreamProcessor,
107	settings: configuration.OperatorSettings,
108	identity: peering.Identity,
109	insights: references.Insights,
110	operator_paused: aiotoggles.ToggleSet,
111	) -> None:
112	peering_missing = await operator_paused.make_toggle(name='peering CRD is missing')	×
113	ensemble = Ensemble(	×
114	peering_missing=peering_missing,
115	operator_paused=operator_paused,
116	operator_indexed=aiotoggles.ToggleSet(all),
117	)
118	try:	×
119	async with insights.revised:	×
UNCOV 120	while True:
121	await insights.revised.wait()	×
122	await adjust_tasks(	×
123	processor=processor,
124	insights=insights,
125	settings=settings,
126	identity=identity,
127	ensemble=ensemble,
128	)
129	except asyncio.CancelledError:	×
130	tasks = ensemble.get_tasks(ensemble.get_keys())	×
131	await aiotasks.stop(tasks, title="streaming", logger=logger, interval=10)	×
132	raise	×
133
134
135	# Directly corresponds to one iteration of an orchestrator, but it is extracted for testability:
136	# for a simulation of the insights (inputs) and an assertion of the tasks & toggles (outputs).
137	async def adjust_tasks(	13✔
138	*,
139	processor: ResourceWatchStreamProcessor,
140	insights: references.Insights,
141	settings: configuration.OperatorSettings,
142	identity: peering.Identity,
143	ensemble: Ensemble,
144	) -> None:
145	peering_selectors = peering.guess_selectors(settings=settings)	13✔
146	peering_resources = {insights.backbone[s] for s in peering_selectors if s in insights.backbone}	13✔
147
148	# Pause or resume all streams if the peering CRDs are absent but required.
149	# Ignore the CRD absence in auto-detection mode: pause only when (and if) the CRDs are added.
150	await ensemble.peering_missing.turn_to(settings.peering.mandatory and not peering_resources)	13✔
151
152	# Stop & start the tasks to match the task matrix with the cluster insights.
153	# As a rule of thumb, stop the tasks first, start later -- not vice versa!
154	await terminate_redundancies(ensemble=ensemble,	13✔
155	remaining_resources=insights.watched_resources \| peering_resources,
156	remaining_namespaces=insights.namespaces \| {None})
157	await spawn_missing_peerings(ensemble=ensemble,	13✔
158	settings=settings,
159	identity=identity,
160	resources=peering_resources,
161	namespaces=insights.namespaces)
162	await spawn_missing_watchers(ensemble=ensemble,	13✔
163	settings=settings,
164	processor=processor,
165	indexed_resources=insights.indexed_resources,
166	watched_resources=insights.watched_resources,
167	watched_namespaces=insights.namespaces)
168
169
170	async def terminate_redundancies(	13✔
171	*,
172	remaining_resources: Collection[references.Resource],
173	remaining_namespaces: Collection[references.Namespace],
174	ensemble: Ensemble,
175	) -> None:
176	# Do not distinguish the keys: even for the case when the peering CRD is served by the operator,
177	# for the peering CRD or namespace deletion, both tasks are stopped together, never apart.
178	redundant_keys = {key for key in ensemble.get_keys()	13✔
179	if key.namespace not in remaining_namespaces
180	or key.resource not in remaining_resources}
181	redundant_tasks = ensemble.get_tasks(redundant_keys)	13✔
182	redundant_flags = ensemble.get_flags(redundant_keys)	13✔
183	await aiotasks.stop(redundant_tasks, title="streaming", logger=logger, interval=10, quiet=True)	13✔
184	await ensemble.operator_paused.drop_toggles(redundant_flags)	13✔
185	ensemble.del_keys(redundant_keys)	13✔
186
187
188	async def spawn_missing_peerings(	13✔
189	*,
190	settings: configuration.OperatorSettings,
191	identity: peering.Identity,
192	resources: Collection[references.Resource],
193	namespaces: Collection[references.Namespace],
194	ensemble: Ensemble,
195	) -> None:
196	for resource, namespace in itertools.product(resources, namespaces):	13✔
197	namespace = namespace if resource.namespaced else None	13✔
198	dkey = EnsembleKey(resource=resource, namespace=namespace)	13✔
199	if dkey not in ensemble.peering_tasks:	13✔
200	what = f"{settings.peering.name}@{namespace}"	13✔
201	is_preactivated = settings.peering.mandatory	13✔
202	conflicts_found = await ensemble.operator_paused.make_toggle(is_preactivated, name=what)	13✔
203	ensemble.conflicts_found[dkey] = conflicts_found	13✔
204	ensemble.pinging_tasks[dkey] = aiotasks.create_guarded_task(	13✔
205	name=f"peering keep-alive for {what}", logger=logger, cancellable=True,
206	coro=peering.keepalive(
207	namespace=namespace,
208	resource=resource,
209	settings=settings,
210	identity=identity))
211	ensemble.peering_tasks[dkey] = aiotasks.create_guarded_task(	13✔
212	name=f"peering observer for {what}", logger=logger, cancellable=True,
213	coro=queueing.watcher(
214	settings=settings,
215	resource=resource,
216	namespace=namespace,
217	processor=functools.partial(peering.process_peering_event,
218	conflicts_found=conflicts_found,
219	namespace=namespace,
220	resource=resource,
221	settings=settings,
222	identity=identity)))
223
224	# Ensure that all guarded tasks got control for a moment to enter the guard.
225	await asyncio.sleep(0)	13✔
226
227
228	async def spawn_missing_watchers(	13✔
229	*,
230	processor: ResourceWatchStreamProcessor,
231	settings: configuration.OperatorSettings,
232	indexed_resources: Container[references.Resource], # only "if in", never "for in"!
233	watched_resources: Iterable[references.Resource],
234	watched_namespaces: Iterable[references.Namespace],
235	ensemble: Ensemble,
236	) -> None:
237
238	# Block the operator globally until specialised per-resource-kind blockers are created.
239	# NB: Must be created before the point of parallelisation!
240	operator_blocked = await ensemble.operator_indexed.make_toggle(name="orchestration blocker")	13✔
241
242	# Spawn watchers and create the specialised per-resource-kind blockers.
243	for resource, namespace in itertools.product(watched_resources, watched_namespaces):	13✔
244	namespace = namespace if resource.namespaced else None	13✔
245	dkey = EnsembleKey(resource=resource, namespace=namespace)	13✔
246	if dkey not in ensemble.watcher_tasks:	13✔
247	what = f"{resource}@{namespace}"	13✔
248	resource_indexed: Optional[aiotoggles.Toggle] = None	13✔
249	if resource in indexed_resources:	13!
250	resource_indexed = await ensemble.operator_indexed.make_toggle(name=what)	×
251	ensemble.watcher_tasks[dkey] = aiotasks.create_guarded_task(	13✔
252	name=f"watcher for {what}", logger=logger, cancellable=True,
253	coro=queueing.watcher(
254	operator_paused=ensemble.operator_paused,
255	operator_indexed=ensemble.operator_indexed,
256	resource_indexed=resource_indexed,
257	settings=settings,
258	resource=resource,
259	namespace=namespace,
260	processor=functools.partial(processor, resource=resource)))
261
262	# Unblock globally, let the specialised per-resource-kind blockers hold the readiness.
263	await ensemble.operator_indexed.drop_toggle(operator_blocked)	13✔
264
265	# Ensure that all guarded tasks got control for a moment to enter the guard.
266	await asyncio.sleep(0)	13✔

nolar / kopf / 11903476210

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous