#!/usr/bin/env python3 import copy import hashlib import importlib.util import inspect import json import os import tempfile from pathlib import Path class RegistryError(Exception): pass class Plugin: def __init__(self, plugin_root, name): self.name = name self.path = Path(plugin_root) / name self.manifest_path = self.path / "manifest.json" if not self.manifest_path.is_file(): raise RegistryError( f"plugin manifest not found: {self.manifest_path}" ) with self.manifest_path.open("r", encoding="utf-8") as f: self.manifest = json.load(f) module_name = self.manifest.get("module", "plugin.py") module_path = self.path / module_name if not module_path.is_file(): raise RegistryError( f"plugin module not found: {module_path}" ) spec = importlib.util.spec_from_file_location( f"managed_containerd_plugin_{name}", module_path, ) if spec is None or spec.loader is None: raise RegistryError( f"unable to load plugin: {name}" ) self.module = importlib.util.module_from_spec(spec) spec.loader.exec_module(self.module) @property def actions(self): return list(self.manifest.get("actions", [])) @property def required_config(self): return list( self.manifest.get("required_config", []) ) @property def optional_config(self): return list( self.manifest.get("optional_config", []) ) @property def config_keys(self): return list(dict.fromkeys( self.required_config + self.optional_config )) def validate_config(self, config): missing = [ key for key in self.required_config if not config.get(key) ] if missing: raise RegistryError( "missing required configuration: " + ", ".join(missing) ) def execute( self, action, config, runtime_state, request_parameters=None, ): if action not in self.actions: raise RegistryError( f"plugin {self.name} does not support action {action}" ) handler_name = action.replace(".", "_") handler = getattr(self.module, handler_name, None) if handler is None: raise RegistryError( f"plugin {self.name} advertises {action} " "but does not implement it" ) # Backward-compatible generic state injection. Existing one-argument # plugins continue to receive only config; state-aware plugins can # declare a second positional parameter named runtime_state. handler_parameters = list( inspect.signature(handler).parameters.values() ) if len(handler_parameters) >= 3: return handler( config, runtime_state, request_parameters or {}, ) if len(handler_parameters) >= 2: return handler(config, runtime_state) return handler(config) class Resource: def __init__( self, resource_id, plugin, config, state_path=None, ): self.resource_id = resource_id self.plugin = plugin self.config = config self.state_path = ( Path(state_path) if state_path is not None else None ) # Generic daemon-lifetime state storage. The daemon does not interpret # plugin keys or values; applications may use this for transition # history, grace windows, caches, or other resource-local state. self.runtime_state = {} self.plugin.validate_config(config) self.load_state() def load_state(self): if self.state_path is None: return if not self.state_path.exists(): return try: with self.state_path.open( "r", encoding="utf-8", ) as stream: state = json.load(stream) except (OSError, json.JSONDecodeError) as exc: raise RegistryError( f"unable to load resource state: {self.resource_id}" ) from exc if not isinstance(state, dict): raise RegistryError( f"resource state is not an object: {self.resource_id}" ) self.runtime_state = state # Monotonic timestamps cannot survive a daemon restart. postgresql_state = self.runtime_state.get("postgresql") if isinstance(postgresql_state, dict): postgresql_state.pop( "reconnect_started_at", None, ) postgresql_state.pop( "state_changed_at", None, ) def persistent_state(self): state = copy.deepcopy(self.runtime_state) postgresql_state = state.get("postgresql") if isinstance(postgresql_state, dict): postgresql_state.pop( "reconnect_started_at", None, ) postgresql_state.pop( "state_changed_at", None, ) return state def save_state(self): if self.state_path is None: return self.state_path.parent.mkdir( parents=True, exist_ok=True, ) temporary_path = None try: with tempfile.NamedTemporaryFile( mode="w", encoding="utf-8", dir=self.state_path.parent, prefix=f".{self.state_path.name}.", delete=False, ) as stream: temporary_path = Path(stream.name) json.dump( self.persistent_state(), stream, separators=(",", ":"), sort_keys=True, ) stream.write("\n") os.chmod(temporary_path, 0o600) os.replace(temporary_path, self.state_path) except OSError as exc: if temporary_path is not None: temporary_path.unlink(missing_ok=True) raise RegistryError( f"unable to save resource state: {self.resource_id}" ) from exc def describe(self): return { "resource": self.resource_id, "plugin": self.plugin.name, "plugin_version": self.plugin.manifest.get( "version" ), "actions": self.plugin.actions, "configured": True, } class Registry: def __init__(self): self.plugin_root = os.environ.get( "MANAGED_CONTAINERD_PLUGIN_ROOT", "/opt/managed-containerd/plugins", ) self.resources = {} self.state_dir = os.environ.get( "MANAGED_CONTAINERD_STATE_DIR" ) self._load_environment_resource() def _load_environment_resource(self): resource_id = os.environ.get("RESOURCE_ID") plugin_name = os.environ.get("PLUGIN") if not resource_id: raise RegistryError( "RESOURCE_ID is not configured" ) if not plugin_name: raise RegistryError( "PLUGIN is not configured" ) plugin = Plugin( self.plugin_root, plugin_name, ) config = { key: os.environ.get(key) for key in plugin.config_keys } state_path = None if self.state_dir: state_name = hashlib.sha256( resource_id.encode("utf-8") ).hexdigest() + ".json" state_path = ( Path(self.state_dir) / state_name ) self.resources[resource_id] = Resource( resource_id, plugin, config, state_path, ) def list_resources(self): return { resource_id: resource.describe() for resource_id, resource in self.resources.items() } def execute( self, resource_id, action, request_parameters=None, ): resource = self.resources.get(resource_id) if resource is None: raise RegistryError( f"unknown resource: {resource_id}" ) result = resource.plugin.execute( action, resource.config, resource.runtime_state, request_parameters, ) resource.save_state() return { "ok": True, "resource": resource_id, "plugin": resource.plugin.name, "action": action, "result": result, }