Skip to main content

pod_config_reload

Hot-reload capability for pod datasource configuration.

Provides config-file polling, hub polling, and atomic datasource swapping via a composed DatasourceManager. Instantiated by Pod and used as a composed object (Pod._config_reloader).

Classes​

ConfigReloadMtimeMonitor​

class ConfigReloadMtimeMonitor():

Thread-safe high-water mark for the last-processed config mtime.

Shared between file polling, hub polling, and on-demand reload paths so all have a consistent view of "what has already been processed".

Each reload path advances the value after successfully processing (or deliberately skipping) a change, preventing other paths from redundantly reprocessing it.

Methods​

advance​

def advance(self, value: float) ‑> None:

Advance the high-water mark to value if it exceeds the current value.

get​

def get(self) ‑> float:

Return the last-processed mtime (thread-safe).

ConfigReloadSettings​

class ConfigReloadSettings(    config_path: Path | None = None,    file_poll_interval_seconds: float | None = None,    hub_poll_interval_seconds: float | None = None,    initial_pod_config: PodConfig | None = None,):

Configuration for the hot-reload subsystem, passed as a single object to Pod.

Variables​

  • static config_path : Path | None
  • static file_poll_interval_seconds : float | None
  • static hub_poll_interval_seconds : float | None
  • static initial_pod_config : PodConfig | None

DatasourceConfigReloader​

class DatasourceConfigReloader(    *,    datasource_manager: DatasourceManager,    hub: BitfountHub,    pod_name: str,    config_path: Path | None = None,    file_poll_interval_seconds: float | None = None,    hub_poll_interval_seconds: float | None = None,    initial_pod_config: PodConfig | None = None,    on_datasources_changed: Callable[[set[str], set[str], list[DatasourceConfig]], Awaitable[None]] | None = None,    link_new_datasets_to_cluster: Callable[[list[str]], None] | None = None,):

Composed object providing hot-reload for pod datasource configuration.

Four reload paths are supported, all converging on reload_datasources:

  1. File polling: call start_polling — a background daemon thread watches the config file's mtime and reloads automatically.
  2. On task receipt: call check_and_reload_if_changed — checks both the config file mtime and the hub for changes before processing a task, reloading if either source has changed.
  3. Orchestrator push: call reload_from_pod_config (or reload_from_datasource_configs) — applies config pushed via HTTP.
  4. Hub polling: call start_polling — a background thread polls the hub for config changes using server-side lastUpdated timestamps.

Mtime interplay: Paths 1 & 2 compare the file's mtime against _last_mtime. When the orchestrator pushes a config (path 3) or the hub poller applies a change (path 4), _last_mtime is set to time.time() so that file-based polling won't revert the remote update with stale file contents.

Thread safety: The atomic swap of base_datasources is safe under CPython's GIL. Workers that have already obtained a datasource reference will continue using the old config until their task completes.

Variables​

  • config_path : pathlib.Path | None - The config file path being watched.
  • event_loop : asyncio.events.AbstractEventLoop | None - The pod's event loop, set when polling starts.
  • file_poll_interval_seconds : float | None - The file polling interval in seconds.
  • file_polling_supported : bool - Whether file polling can be started (config_path and interval set).
  • hub_poll_interval_seconds : float | None - The hub polling interval in seconds.
  • hub_polling_supported : bool - Whether hub polling can be started (interval set).

Methods​

check_and_reload_if_changed​

async def check_and_reload_if_changed(self) ‑> bool:

Check all configured sources for config changes and reload if needed.

Intended for on-demand use from an async context (e.g. upon task receipt). Runs hub-based then file-based checks, swallowing errors from each independently so that a failure in one does not prevent the other from executing. Hub runs first to establish the timestamp baseline; the file check then runs and wins if the file is genuinely more recent than the hub's last change.

Returns True if any reload was performed, False otherwise.

check_file_and_reload_if_changed​

async def check_file_and_reload_if_changed(self) ‑> bool:

Check the config file for changes and reload if needed.

check_hub_and_reload_if_changed​

async def check_hub_and_reload_if_changed(self) ‑> bool:

Check the hub for config changes and reload if needed.

reload_datasources​

async def reload_datasources(    self,    new_datasource_configs: list[DatasourceConfig],    *,    origin: "Literal['file', 'hub']" = 'hub',) ‑> None:

Atomically reload datasources from new configuration.

Handles three cases:

  • Removed datasets are dropped from local datasources.
  • New datasets are registered on the hub.
  • Existing datasets are updated (upsert via register_pod).

This method is safe to call from any thread. In-flight workers will continue using their existing datasource references until completion.

Arguments

  • new_datasource_configs: The new datasource configurations to apply.
  • origin: Source of the reload request. File reloads may link new datasets to the pod compute cluster; hub reloads do not.

reload_from_datasource_configs​

async def reload_from_datasource_configs(    self, datasource_configs: list[DatasourceConfig],) ‑> None:

Reload datasources directly from a list of datasource configs.

Skips pod-level validation — the caller takes responsibility for ensuring pod-level settings are compatible. Intended for use cases where only datasource configs are available (no full PodConfig).

Arguments

  • datasource_configs: The new datasource configurations to apply.

reload_from_pod_config​

async def reload_from_pod_config(self, pod_config: PodConfig) ‑> None:

Validate pod-level fields and reload datasources from a full PodConfig.

Intended for use when the orchestrator pushes a new config (not file-backed). Validates that pod-level fields are unchanged, then applies datasource changes via reload_datasources.

Arguments

  • pod_config: The new full pod configuration.

Raises

  • PodConfigReloadError: If pod-level configuration has changed (caller should trigger a pod restart).
  • ValueError: If the config contains no datasources.

Note Changes to config_reload_file_poll_interval_seconds are silently ignored (it is excluded from pod-level comparison). The polling interval cannot be changed at runtime without a pod restart.

start_polling​

def start_polling(self, mailbox: _PodMailbox | None = None) ‑> None:

Start background polling for config changes (paths 1 & 4).

Starts file polling if file_polling_supported, hub polling if hub_polling_supported. Raises if neither is available.

Arguments

  • mailbox: The pod's mailbox for re-registering dataset names after reload.

stop_polling​

def stop_polling(self) ‑> None:

Stop all polling (file and hub).

update_pod_name​

def update_pod_name(self, pod_name: str) ‑> None:

Update the cached pod name used when reconnecting to the message service.

The reloader caches the pod name at construction; identifier-length enforcement can shorten it afterwards. A reload reconnect calls connect_pod directly (bypassing the mailbox boundary guard), so the cached name must be kept in sync with the enforced one or the reconnect is rejected as over-length.

PodConfigReloadError​

class PodConfigReloadError(*args, **kwargs):

Raised when a config reload cannot proceed or is rejected.

Ancestors​