Repository navigation
Expand file tree
/
Copy pathsegment_ranges.py
More file actions
77 lines (62 loc) · 2.95 KB
/
Copy pathsegment_ranges.py
File metadata and controls
77 lines (62 loc) · 2.95 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
#!/usr/bin/env python3
"""Shell-operator hook for Neutron network segment range reconciliation."""
from __future__ import annotations
import sys
from typing import Any
from openstack_sync.hooks.framework import HookConfig
from openstack_sync.hooks.framework import ReconcileResult
from openstack_sync.hooks.framework import SyncPlugin
from openstack_sync.hooks.framework import build_crd_hook_config
from openstack_sync.hooks.framework import hook_enabled
from openstack_sync.hooks.framework import hook_inputs
from openstack_sync.hooks.framework import run_hook
from openstack_sync.hooks.framework import run_sync
from openstack_sync.plugins.common import wait_for_openstack_network
from openstack_sync.plugins.neutron.segment_ranges import prune as prune_module
from openstack_sync.plugins.neutron.segment_ranges import reconcile as reconcile_module
from openstack_sync.plugins.neutron.segment_ranges.config import BINDING_NAME
from openstack_sync.plugins.neutron.segment_ranges.config import ENV_PREFIX
class SegmentRangePlugin(SyncPlugin):
"""Sync NeutronSegmentRange CRs into Neutron network segment ranges."""
noun = "segment range"
def wait_for_api(self, conn: Any) -> None:
wait_for_openstack_network(
conn,
retries=self.config.ready_retries,
delay=self.config.ready_delay,
)
def new_cache(self) -> reconcile_module.RangeCache:
# Keyed by managed range name and shared across every CR in one
# credential group, so the managed-range listing is fetched once and
# reused by each reconcile in the group. Prune does not receive this
# cache (SyncPlugin.prune / PruneRequest carry no cache) and re-lists
# from Neutron itself.
return {}
def reconcile(
self, conn: Any, spec: dict[str, Any], cache: reconcile_module.RangeCache
) -> ReconcileResult:
notes = reconcile_module.sync_segment_range(conn, spec, cache)
return ReconcileResult(notes=notes)
def prune(
self,
conn: Any,
desired_specs: list[dict[str, Any]],
*,
authoritative_empty: bool,
) -> None:
# The framework only calls prune under a cleanup policy that runs it,
# which for this plugin means the chart's PRUNE flag is enabled (see
# SyncPlugin.cleanup_policy: config.prune + a real prune step ->
# FINALIZED_PRUNE). No manual gate on config.prune is needed here.
prune_module.prune_removed_ranges(
conn, desired_specs, authoritative_empty=authoritative_empty
)
def main() -> int:
def run(contexts: list[dict[str, Any]]) -> int:
if not hook_enabled(ENV_PREFIX):
return 0
config = HookConfig.from_env(ENV_PREFIX, binding_name=BINDING_NAME)
return run_sync(SegmentRangePlugin(config), hook_inputs(contexts, config))
return run_hook(lambda: build_crd_hook_config(ENV_PREFIX, BINDING_NAME), run)
if __name__ == "__main__":
sys.exit(main())