From 6e946da88cc3eb5aa797b4295a1573bae59f3149 Mon Sep 17 00:00:00 2001 From: Vitalii <87299468+vitalii-t12@users.noreply.github.com> Date: Thu, 11 Sep 2025 18:30:35 +0300 Subject: [PATCH 1/4] C.A.R. refactor (#234) * refactor: enhance container management with Docker integration - Replace subprocess calls with Docker SDK for Python in container management functions. - Add health check and image update mechanisms for better container monitoring. - Improve logging and error handling during container operations. - Update configuration options for endpoint polling and image management. - Refactor utility methods to streamline container operations and enhance readability. * fix: add possibility to disable health endpoint check * fix: volumes mounting * fix: volumes set up * fix: env setup * fix: inverted ports mapping * fix: create a method for setting up env and ports * fix: rename methods * fix: remove dead code * fix: volumes for ADCAR * fix: docker registry login * fix: env set up * fix: use vcs_data in worker app runner config * fix: remove validation * chore: increment version * fix: add current time check * fix: save logs on restart --- .../admin_container_app_runner.py | 11 +- .../container_apps/container_app_runner.py | 575 ++++++++++++------ .../container_apps/container_utils.py | 416 +++---------- .../container_apps/worker_app_runner.py | 196 ++---- ver.py | 2 +- 5 files changed, 543 insertions(+), 657 deletions(-) diff --git a/extensions/business/container_apps/admin_container_app_runner.py b/extensions/business/container_apps/admin_container_app_runner.py index 3d9cba92..d3939137 100644 --- a/extensions/business/container_apps/admin_container_app_runner.py +++ b/extensions/business/container_apps/admin_container_app_runner.py @@ -44,20 +44,21 @@ def on_init(self): if not self.volumes: self.volumes = {} - if self.cfg_mount_edge_node_data_volume == True: - self.volumes[EDGE_NODE_DATA_PATH] = EDGE_NODE_DATA_MOUNT_POINT return - def _setup_volumes(self): + def _configure_volumes(self): """ Processes the volumes specified in the configuration. """ + default_volume_rights = "rw" + + if self.cfg_mount_edge_node_data_volume == True: + self.volumes[EDGE_NODE_DATA_PATH] = {"bind": EDGE_NODE_DATA_MOUNT_POINT, "mode": default_volume_rights} if hasattr(self, 'cfg_volumes') and self.cfg_volumes and len(self.cfg_volumes) > 0: for host_path, container_path in self.cfg_volumes.items(): original_path = str(host_path) - self.volumes[original_path] = container_path - + self.volumes[original_path] = {"bind": container_path, "mode": default_volume_rights} # endfor each host path # endif volumes return diff --git a/extensions/business/container_apps/container_app_runner.py b/extensions/business/container_apps/container_app_runner.py index 3e88af85..502e4e8f 100644 --- a/extensions/business/container_apps/container_app_runner.py +++ b/extensions/business/container_apps/container_app_runner.py @@ -21,12 +21,13 @@ """ -import shutil -import socket -import subprocess +import docker +import requests +import threading import time +import socket -from naeural_core.business.base.web_app.base_web_app_plugin import BaseWebAppPlugin as BasePlugin +from naeural_core.business.base.web_app.base_tunnel_engine_plugin import BaseTunnelEnginePlugin as BasePlugin from .container_utils import _ContainerUtilsMixin # provides container management support currently empty it is embedded in the plugin @@ -59,7 +60,7 @@ "DEBUG_WEB_APP": False, # If True, will run the web app in debug mode "CAR_VERBOSE": 1, - # Container-specific config options + # Container-specific config options "IMAGE": None, # Required container image, e.g. "my_repo/my_app:latest" "CR_DATA": { # dict of container registry data "SERVER": 'docker.io', # Optional container registry URL @@ -73,25 +74,29 @@ "cpu": 1, # e.g. "0.5" for half a CPU, or "1.0" for one CPU core "gpu": 0, "memory": "512m", # e.g. "512m" for 512MB, - "ports": [] # dict of container_port: host_port mappings (e.g. {8080: 8081}) or list of container ports (e.g. [8080, 9000]) + "ports": [] # dict of host_port: container_port mappings (e.g. {8080: 8081}) or list of container ports (e.g. [8080, 9000]) }, "RESTART_POLICY": "always", # "always" will restart the container if it stops "IMAGE_PULL_POLICY": "always", # "always" will always pull the image "AUTOUPDATE" : True, # If True, will check for image updates and pull them if available - "AUTOUPDATE_INTERVAL": 100, + "AUTOUPDATE_INTERVAL": 100, "VOLUMES": {}, # dict mapping host paths to container paths, e.g. {"/host/path": "/container/path"} - + + # Application endpoint polling + "ENDPOINT_POLL_INTERVAL": 0, # seconds between endpoint health checks + "ENDPOINT_URL": None, # endpoint to poll for health checks + #### Logging "SHOW_LOG_EACH" : 60, # seconds to show logs - "SHOW_LOG_LAST_LINES" : 5, # last lines to show + "SHOW_LOG_LAST_LINES" : 5, # last lines to show "MAX_LOG_LINES" : 10_000, # max lines to keep in memory - + # end of container-specific config options - + 'VALIDATION_RULES': { **BasePlugin.CONFIG['VALIDATION_RULES'], - }, + }, } @@ -133,21 +138,40 @@ def Pd(self, s, *args, score=-1, **kwargs): def __reset_vars(self): + self.container = None self.container_id = None self.container_name = self.cfg_instance_id + "_" + self.uuid(4) - self.container_proc = None - self.container_run_command_key = "" + self.docker_client = docker.from_env() self.container_logs = self.deque(maxlen=self.cfg_max_log_lines) # Handle port allocation for main port and additional ports - self.extra_ports_mapping = {} # Dictionary to store container_port -> host_port mappings + self.extra_ports_mapping = {} # Dictionary to store host_port -> container_port mappings + self.inverted_ports_mapping = {} # inverted mapping for docker-py container_port -> host_port self.volumes = {} + self.env = {} self.dynamic_env = {} self._is_manually_stopped = False # Flag to indicate if the container was manually stopped + # Initialize tunnel process + self.tunnel_process = None + + # Log streaming + self.log_thread = None + self._stop_event = threading.Event() + + # Container start time tracking + self.container_start_time = None + + # Periodic intervals + self._last_endpoint_check = 0 + self._last_image_check = 0 + + # Image update tracking + self.current_image_hash = None + return def on_init(self): @@ -155,41 +179,44 @@ def on_init(self): Lifecycle hook called once the plugin is initialized. Authenticates with the container registry (if config is provided). Determines whether Docker or Podman is available, sets up port (if needed), - and prepares for container run. + and prepares for container run. """ - - self.__last_autoupdate_check = 0 - + self.__reset_vars() super(ContainerAppRunnerPlugin, self).on_init() self.container_start_time = self.time() - self._detect_cli_tool() # detect if we have docker or podman + # Login to container registry if credentials are provided + if not self._login_to_registry(): + raise RuntimeError("Failed to login to container registry. Cannot proceed without authentication.") - self._setup_dynamic_env() # setup dynamic env vars for the container + self.reset_tunnel_engine() + + self._configure_dynamic_env() # setup dynamic env vars for the container self._setup_resource_limits_and_ports() # setup container resource limits (CPU, GPU, memory, ports) - self._setup_volumes() # setup container volumes + self._configure_volumes() # setup container volumes + + self._setup_env_and_ports() return - def on_command(self, data, **kwargs): """ Called when a INSTANCE_COMMAND is received by the plugin instance. - + The command is sent via `cmdapi_send_instance_command` from a commanding node (Deeploy plugin) as in below simplified example: - + ```python pipeline = "some_app_pipeline" signature = "CONTAINER_APP_RUNNER" instance_id = "CONTAINER_APP_1e8dac" node_address = "0xai_1asdfG11sammamssdjjaggxffaffaheASSsa" - + instance_command = "RESTART" - + plugin.cmdapi_send_instance_command( pipeline=pipeline, signature=signature, @@ -198,9 +225,9 @@ def on_command(self, data, **kwargs): node_address=node_address, ) ``` - + while the `on_command` method should look like this: - + ```python def on_command(self, data, **kwargs): if data == "RESTART": @@ -213,7 +240,7 @@ def on_command(self, data, **kwargs): self.P(f"Unknown command: {data}") return ``` - + """ self.P(f"Received a command: {data}") self.P(f"Command kwargs: {kwargs}") @@ -221,8 +248,7 @@ def on_command(self, data, **kwargs): if data == "RESTART": self.P("Restarting container...") self._is_manually_stopped = False - self._stop_container_and_save_logs_to_disk() - self._container_maybe_reload(force_restart=True) + self._restart_container() return elif data == "STOP": @@ -233,7 +259,7 @@ def on_command(self, data, **kwargs): else: self.P(f"Unknown plugin command: {data}") return - + def on_post_container_start(self): """ Lifecycle hook called after the container is started. @@ -246,132 +272,175 @@ def on_post_container_start(self): return - def get_setup_commands(self): + + + def start_tunnel_engine(self): """ - TODO: fix the attack vector here, we should not allow arbitrary commands to be run - + Start the tunnel engine using the base tunnel engine functionality. + """ + if self.cfg_tunnel_engine_enabled: + engine_name = "Cloudflare" if self.use_cloudflare() else "ngrok" + self.P(f"Starting {engine_name} tunnel...", color='b') + self.tunnel_process = self.run_tunnel_engine() + if self.tunnel_process: + self.P(f"{engine_name} tunnel started successfully", color='g') + else: + self.P(f"Failed to start {engine_name} tunnel", color='r') + return + + def stop_tunnel_engine(self): + """ + Stop the tunnel engine. """ - cfg_setup_commands = self.cfg_setup_commands - setup_commands = [] - if isinstance(cfg_setup_commands, str): - setup_commands.append(cfg_setup_commands) - elif isinstance(cfg_setup_commands, list): - setup_commands.extend(cfg_setup_commands) - container_login_command = self._get_container_login_command() + if self.tunnel_process: + engine_name = "Cloudflare" if self.use_cloudflare() else "ngrok" + self.P(f"Stopping {engine_name} tunnel...", color='b') + self.stop_tunnel_command(self.tunnel_process) + self.tunnel_process = None + self.P(f"{engine_name} tunnel stopped", color='g') + return - if container_login_command: - setup_commands.append(container_login_command) + def start_container(self): + """Start the Docker container.""" + self.P(f"Launching container with image '{self.cfg_image}'...") - if self.cfg_image_pull_policy == "always": - setup_commands.append(self._get_container_pull_image_command()) + self.P(f"Container data:") + self.P(f" Image: {self.cfg_image}") + self.P(f" Ports: {self.json_dumps(self.inverted_ports_mapping) if self.inverted_ports_mapping else 'None'}") + self.P(f" Env: {self.json_dumps(self.env) if self.env else 'None'}") + self.P(f" Volumes: {self.json_dumps(self.volumes) if self.volumes else 'None'}") + self.P(f" Resources: {self.json_dumps(self.cfg_container_resources) if self.cfg_container_resources else 'None'}") + self.P(f" Restart policy: {self.cfg_restart_policy}") + self.P(f" Pull policy: {self.cfg_image_pull_policy}") try: - setup_commands = super(ContainerAppRunnerPlugin, self).get_setup_commands() + setup_commands + self.container = self.docker_client.containers.run( + self.cfg_image, + detach=True, + ports=self.inverted_ports_mapping, + environment=self.env, + volumes=self.volumes, + # restart_policy={"Name": self.cfg_restart_policy} if self.cfg_restart_policy != "no" else None, + name=self.container_name, + ) except Exception as e: - pass + self.P(f"Could not start container: {e}", color='r') + self.container = None + return None + + self.container_id = self.container.short_id + self.P(f"Container started (ID: {self.container.short_id})", color='g') + return self.container + + def stop_container(self): + """Stop and remove the Docker container if it is running.""" + if not self.container: + self.P("No container to stop", color='y') + return - return setup_commands + try: + # Stop the container (gracefully) + self.P(f"Stopping container {self.container.short_id}...", color='b') + self.container.stop(timeout=5) + self.P(f"Container {self.container.short_id} stopped successfully", color='g') + except Exception as e: + self.P(f"Error stopping container: {e}", color='r') + # end try + try: + self.P(f"Removing container {self.container.short_id}...", color='b') + self.container.remove() + self.P(f"Container {self.container.short_id} removed successfully", color='g') + except Exception as e: + self.P(f"Error removing container: {e}", color='r') + finally: + self.container = None + self.container_id = None + # end try + return - def get_start_commands(self): - """ - TODO: fix the attack vector here, we should not allow arbitrary commands to be run - - """ - cfg_start_commands = self.cfg_start_commands - start_commands = [] - if isinstance(cfg_start_commands, str): - start_commands.append(cfg_start_commands) - elif isinstance(cfg_start_commands, list): - start_commands.extend(cfg_start_commands) - start_commands.append(self._get_container_run_command()) - self.container_run_command_key = f"start_{len(start_commands) - 1}" + def _stream_logs(self, log_stream): + """Consume a log iterator from container logs and print its output.""" + if not log_stream: + self.P("No log stream provided", color='y') + return try: - start_commands = start_commands + super(ContainerAppRunnerPlugin, self).get_start_commands() + for log_bytes in log_stream: + if log_bytes is None: + break + try: + log_str = log_bytes.decode("utf-8", errors="replace") + except Exception as e: + self.P(f"Warning: Could not decode log bytes: {e}", color='y') + log_str = str(log_bytes) + + self.P(f"[CONTAINER] {log_str}", color='d', end='') + self.container_logs.append(log_str) + + if self._stop_event.is_set(): + self.P("Log streaming stopped by stop event", color='y') + break except Exception as e: - pass - return start_commands + self.P(f"Exception while streaming logs: {e}", color='r') + # end try + return - def on_log_handler(self, log, key): - if key == self.container_run_command_key: - self.container_logs.append(log) + def _check_health_endpoint(self, current_time=None): + if not self.container or not self.cfg_endpoint_url or self.cfg_endpoint_poll_interval <= 0: + return - def _detect_cli_tool(self): - """ - Detects whether Docker or Podman is available on the system. - """ - if shutil.which("docker"): - self.cli_tool = "docker" - elif shutil.which("podman"): - self.cli_tool = "podman" - else: - raise RuntimeError("No container runtime (Docker/Podman) found on this system.") - #endif + if current_time - self._last_endpoint_check >= self.cfg_endpoint_poll_interval: + self._last_endpoint_check = current_time + self._poll_endpoint() + # end if time elapsed return + def _poll_endpoint(self): + """Poll the container's health endpoint and log the response.""" + if not self.port: + self.P("No port allocated, cannot poll endpoint", color='r') + return + + if not self.cfg_endpoint_url: + self.P("No endpoint URL configured, skipping health check", color='y') + return + + url = f"http://localhost:{self.port}{self.cfg_endpoint_url}" + + try: + resp = requests.get(url, timeout=5) + status = resp.status_code + + if status == 200: + self.P(f"Health check: {url} -> {status} OK", color='g') + else: + self.P(f"Health check: {url} -> {status} Error", color='r') + except requests.RequestException as e: + self.P(f"Health check failed: {url} - {e}", color='r') + except Exception as e: + self.P(f"Unexpected error during health check: {e}", color='r') + # end try + return + + def _check_container_status(self): + try: + if self.container: + # Refresh container status + self.container.reload() + if self.container.status != "running": + self.P(f"Container stopped unexpectedly (exit code {self.container.attrs.get('State', {}).get('ExitCode')})", color='r') + return False + # end if container not running + # end if self.container + return True + except Exception as e: + self.P(f"Could not check container status: {e}", color='r') + self.container = None + # end try + return False + - # TODO: move to base class - def _allocate_port(self, required_port=0, allow_dynamic=False, sleep_time=5): - """ - Allocates an available port on the host system for container port mapping. - - This method finds an available port on the host system that can be used for container port mapping. - If required_port is 0 (default), the OS will automatically select any available port. - If required_port is specified, the method will attempt to bind to that specific port. - - The method uses a socket-based approach to port allocation: - 1. Creates a new TCP socket - 2. Sets SO_REUSEADDR option to allow immediate reuse of the port - 3. Binds to the specified port (or any available port if 0) - 4. Retrieves the actual port number that was bound - 5. Closes the socket to release it for actual use - - Args: - required_port (int, optional): The specific port number to allocate. - If 0 (default), the OS will select any available port. - - Returns: - int: The allocated port number. This will be the same as required_port if specified - and available, or a randomly assigned port if required_port is 0. - - Note: - The socket is closed immediately after port allocation to allow the port to be used - by the container. This is a common technique for port allocation in container runtimes. - """ - port = None - if required_port != 0: - self.P(f"Trying to allocate requested port {required_port} ...") - done = False - while not done: - try: - sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) - sock.bind(("", required_port)) - port = sock.getsockname()[1] - sock.close() - done = True - except Exception as e: - port = None - if allow_dynamic: - self.P(f"Failed to allocate requested port {required_port}: {e}", color='r') - done = True # if allow_dynamic is True, we stop trying to bind to the required port - required_port = 0 # reset to allow dynamic port allocation - else: - self.P(f"Port {required_port} is not available. Retrying in {sleep_time} seconds...", color='r') - self.sleep(sleep_time) # wait before retrying - # endtry - # endwhile done - #endif required_port != 0 - - if required_port == 0 and port is None: - sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) - sock.bind(("", 0)) - port = sock.getsockname()[1] - sock.close() - #endif - return port def _stop_container_and_save_logs_to_disk(self): @@ -380,24 +449,23 @@ def _stop_container_and_save_logs_to_disk(self): Then logs are saved to disk. """ self.P(f"Stopping container app '{self.container_id}' ...") + + # Stop log streaming + self._stop_event.set() + if self.log_thread: + self.log_thread.join(timeout=5) + # Stop the container if it's running + self.stop_container() - self._container_kill(self.container_id) - self._maybe_close_setup_commands() - self._maybe_close_start_commands() - self._maybe_read_and_stop_all_log_readers() # Stop tunnel engine if needed - self.P("Stopping tunnel engine tunnel ...") - self.maybe_stop_tunnel_engine() - self.P("Tunnel engine stopped.") + self.stop_tunnel_engine() # Save logs to disk - # We'll store them in a single structure: a list of lines from dct_logs or so - # We can do: logs, err_logs = self._get_delta_logs() or a custom approach try: # using parent class method to save logs self.diskapi_save_pickle_to_output( - obj=self.container_logs, filename="container_logs.pkl" + obj=list(self.container_logs), filename="container_logs.pkl" ) self.P("Container logs saved to disk.") except Exception as exc: @@ -412,51 +480,186 @@ def on_close(self): Stops tunnel if started. """ self._stop_container_and_save_logs_to_disk() - self._container_kill(self.container_id) + super(ContainerAppRunnerPlugin, self).on_close() - def _maybe_autoupdate_container(self): - if self.cfg_autoupdate and self.container_id is not None: - # Check if the image exists and pull it if needed - last_checked = self.time() - self.__last_autoupdate_check - needs_update = last_checked > self.cfg_autoupdate_interval - if needs_update: - if self._container_exists(self.container_id): - self.__last_autoupdate_check = self.time() - self.Pd("Checking for container image updates ...", score=30) - try: - # TODO: use get container has instead of pulling the image - pulled = self._container_pull_image() - if pulled: - # If the image was pulled, we can restart the container - self.P("Stopping container to use the new image ...") - self._stop_container_and_save_logs_to_disk() - self._restart_container() - else: - self.Pd("No updates found for the container image.", score=30) - except Exception as e: - self.P(f"Failed to pull image {self.cfg_image}: {e}", color='r') - else: - self.P("Container does not exist, skipping image update check.") - #endif container exists - # endif needs_update - #endif autoupdate enabled + def _get_latest_image_hash(self): + """ + Get the latest identifier for the configured Docker image tag. + + This method tries to resolve the remote content digest for ``self.cfg_image`` by + asking the Docker daemon to perform a metadata-only pull (if the image is + already up to date, no layers are re-downloaded). It returns the repo digest + (e.g., ``sha256:...``) when available; if not available, it falls back to the + local image ID. + + Returns + ------- + str or None + A digest like ``sha256:`` (preferred) or the local image ID. Returns + ``None`` if neither can be obtained. + + Notes + ----- + - Works for public and private registries as long as the Docker daemon has + credentials configured. + - This call contacts the registry; tune ``poll_interval`` appropriately. + """ + if not self.cfg_image: + self.P("No Docker image configured", color='r') + return None + + # Ensure we're logged in to the registry before pulling + if not self._login_to_registry(): + raise RuntimeError("Failed to login to container registry. Cannot proceed without authentication.") + + try: + self.P(f"Image check: pulling '{self.cfg_image}' for metadata...", color='b') + img = self.docker_client.images.pull(self.cfg_image) + # docker-py may return Image or list[Image] + if isinstance(img, list) and img: + img = img[-1] + # Ensure attributes loaded + try: + img.reload() + except Exception as e: + self.P(f"Warning: Could not reload image attributes: {e}", color='y') + # end try + + attrs = getattr(img, "attrs", {}) or {} + repo_digests = attrs.get("RepoDigests") or [] + if repo_digests: + # 'repo@sha256:...' + digest = repo_digests[0].split("@")[-1] + return digest + # Fallback to image id (sha256:...) + return getattr(img, "id", None) + + except Exception as e: + self.P(f"Image pull failed: {e}", color='r') + # Fallback: check local image only + try: + self.P(f"Checking local image: {self.cfg_image}", color='b') + img = self.docker_client.images.get(self.cfg_image) + try: + img.reload() + except Exception as e: + self.P(f"Warning: Could not reload local image attributes: {e}", color='y') + # end try reload + attrs = getattr(img, "attrs", {}) or {} + repo_digests = attrs.get("RepoDigests") or [] + if repo_digests: + digest = repo_digests[0].split("@")[-1] + return digest + return getattr(img, "id", None) + + except Exception as e2: + self.P(f"Could not get local image: {e2}", color='r') + # end try check for local image + # end try + return None + + def _check_image_updates(self, current_time=None): + """Check for a new version of the Docker image and restart container if found.""" + if not self.cfg_autoupdate: + return + + if current_time - self._last_image_check >= self.cfg_autoupdate_interval: + self._last_image_check = current_time + latest_image_hash = self._get_latest_image_hash() + if latest_image_hash and self.current_image_hash and latest_image_hash != self.current_image_hash: + self.P(f"New image version detected ({latest_image_hash} != {self.current_image_hash}). Restarting container...", color='y') + # Update current_image_hash to the new one + self.current_image_hash = latest_image_hash + # Restart container from scratch + self._restart_container() + elif latest_image_hash: + self.P(f"Current image hash: {self.current_image_hash} vs latest: {latest_image_hash}") + # end if new image hash + # end if time elapsed + return + + def _restart_container(self): + """Restart the container from scratch.""" + self.P("Restarting container from scratch...", color='b') + self._stop_container_and_save_logs_to_disk() + # Start a new container + self._stop_event.clear() # reset stop flag for new log thread + self.container = self.start_container() + self.start_tunnel_engine() + self.container_start_time = self.time() + + # Start log streaming + if self.container: + self.log_thread = threading.Thread( + target=self._stream_logs, + args=(self.container.logs(stream=True, follow=True),), + daemon=True, + ) + self.log_thread.start() return + def _handle_initial_launch(self): + """Handle the initial container launch.""" + try: + self.P("Initial container launch...", color='b') + # Initialize current image hash for update tracking + self.current_image_hash = self._get_latest_image_hash() + self.container = self.start_container() + self.container_start_time = self.time() + + # Start log streaming + if self.container: + self.log_thread = threading.Thread( + target=self._stream_logs, + args=(self.container.logs(stream=True, follow=True),), + daemon=True, + ) + self.log_thread.start() + + self.P("Container launched successfully", color='g') + self.P(self.container) + if self.current_image_hash: + self.P(f"Current image hash: {self.current_image_hash}", color='d') + except Exception as e: + self.P(f"Could not start container: {e}", color='r') + # end try + return + + def _perform_periodic_monitoring(self): + """Perform periodic monitoring tasks.""" + current_time = self.time() + self._check_health_endpoint(current_time) + self._check_image_updates(current_time) + return def process(self): """ This is the main process loop for the plugin that gets called each PROCESS_DELAY seconds and it performs the following: - - 1. self._container_maybe_reload() - check if the container is still running and perform the policy - specified in the restart policy. - 2. self._container_retrieve_and_maybe_show_logs() - check if the logs should be show as well as complete the logs - + + 1. Initialize and start tunnel engine if needed + 2. Check if container is running and restart if needed + 3. Perform periodic monitoring (health checks, etc.) + 4. Tunnel engine ping and maintenance + """ - self._maybe_set_container_id_and_show_app_info() - self._maybe_autoupdate_container() - self._container_maybe_reload() + self.maybe_init_tunnel_engine() + + if not self.container: + self._handle_initial_launch() + + self.maybe_start_tunnel_engine() + + # Start tunnel engine if not already running + if self.cfg_tunnel_engine_enabled and not self.tunnel_process: + self.start_tunnel_engine() + + if not self._check_container_status(): + return + + self._perform_periodic_monitoring() + self.maybe_tunnel_engine_ping() return \ No newline at end of file diff --git a/extensions/business/container_apps/container_utils.py b/extensions/business/container_apps/container_utils.py index 1683df9e..3896ca22 100644 --- a/extensions/business/container_apps/container_utils.py +++ b/extensions/business/container_apps/container_utils.py @@ -27,68 +27,42 @@ def _get_cr_data(self): cr_username = cr_data.get('USERNAME') cr_password = cr_data.get('PASSWORD') return cr_server, cr_username, cr_password - - def _get_container_login_command(self): - # Login to container registry if provided - cr_server, cr_username, cr_password = self._get_cr_data() - - if cr_server and cr_username and cr_password: - login_cmd = [ - self.cli_tool, "login", - str(cr_server), - "-u", str(cr_username), - "-p", str(cr_password), - ] - return " ".join(login_cmd) - - return None - - def _get_container_pull_image_command(self): - """ - Pull the container image (Docker/Podman). + def _login_to_registry(self): """ - full_ref = str(self.cfg_image) - cmd = [self.cli_tool, "pull", full_ref] + Login to a private container registry using credentials from _get_cr_data. - cr_server, _, _ = self._get_cr_data() - - if cr_server and not str(self.cfg_image).startswith(cr_server): - # If image doesn't have the registry prefix, prepend it - full_ref = f"{cr_server.rstrip('/')}/{self.cfg_image}" - cmd = [self.cli_tool, "pull", full_ref] + Returns: + bool: True if login successful, False otherwise + """ + cr_server, cr_username, cr_password = self._get_cr_data() + self.P(f"Container registry data: SERVER={cr_server}, USERNAME={cr_username}, PASSWORD={'***' if cr_password else None}") + # Skip login if no credentials provided + if not cr_username or not cr_password or not cr_server: + self.P("No registry credentials provided, skipping login", color='y') + return True - return " ".join(cmd) + self.P(f"Logging into container registry: {cr_server}", color='b') - def _container_pull_image(self): - """ - Pull the container image (Docker/Podman). - """ - pulled = False - full_ref = str(self.cfg_image) - cmd_str = self._get_container_pull_image_command() - cmd = cmd_str.split() try: - result = subprocess.check_output(cmd) - # now check if the image was pulled or if it was already present - if "Image is up to date" in result.decode("utf-8", errors="ignore"): - self.Pd(f"Image {full_ref} is already up to date.", score=30) - else: - self.Pd(f"Image {full_ref} pulled successfully.") - pulled = True - except Exception as exc: - raise RuntimeError(f"Error pulling image: {exc}") - # end if result - self.Pd(f"Image {full_ref} pulled successfully: {result.decode('utf-8', errors='ignore')}", score=30) - return pulled - - + result = self.docker_client.login( + username=cr_username, + password=cr_password, + registry=cr_server + ) + self.P(f"Successfully logged into registry {cr_server}", color='g') + return True + except Exception as e: + self.P(f"Docker client login failed: {e}", color='y') + + return False + def _get_default_env_vars(self): """ Get the default environment variables for the container. - + WARNING: This is a critical method that should be thoroughly reviewed for attack vectors. - + Returns: dict: Default environment variables. """ @@ -109,180 +83,6 @@ def _get_default_env_vars(self): return dct_env - - def _get_container_run_command(self): - """ - Launch the container in detached mode, returning its ID. - """ - - cmd = [ - self.cli_tool, "run", "--rm", "--name", str(self.container_name), - ] - - # Resource limits - if self._cpu_limit: - cmd += ["--cpus", str(self._cpu_limit)] - - if self._mem_limit: - cmd += ["--memory", str(self._mem_limit)] - - # Port mappings if we have any - if hasattr(self, 'extra_ports_mapping') and self.extra_ports_mapping: - for host_port, container_port in self.extra_ports_mapping.items(): - if host_port == self.port: - continue - cmd += ["-p", f"{host_port}:{container_port}"] - - if self.port and self.cfg_port: - cmd += ["-p", f"{self.port}:{self.cfg_port}"] - - # Env vars - for key, val in self.cfg_env.items(): - cmd += ["-e", f"{key}={val}"] - - for key, val in self.dynamic_env.items(): - cmd += ["-e", f"{key}={val}"] - - # now add the default env vars - for key, val in self._get_default_env_vars().items(): - cmd += ["-e", f"{key}={val}"] - - # Volume mounts - if len(self.volumes) > 0: - for volume_label, container_path in self.volumes.items(): - # Create a named volume with the prefixed sanitized name - volume_spec = f"{volume_label}:{container_path}" - cmd += ["-v", volume_spec] - #endfor - self.P("Note: These named volumes will persist until manually removed with 'docker volume rm'") - - # Possibly prefix the registry to the image reference - image_ref = str(self.cfg_image) - cr_server, _, _ = self._get_cr_data() - - if cr_server and not image_ref.startswith(str(cr_server)): - image_ref = f"{cr_server.rstrip('/')}/{image_ref}" - - cmd.append(image_ref) - - str_cmd = " ".join(cmd) - - return str_cmd - - - def _container_exists(self, cid): - """ - Check if container with ID cid is still running. - """ - result = False - if cid is not None: - ps_cmd = [self.cli_tool, "ps", "-q", "-f", f"id={cid}"] - try: - ps_res = subprocess.run(ps_cmd, capture_output=True) - if ps_res.returncode == 0: - output = ps_res.stdout.decode("utf-8", errors="ignore").strip() - result = len(output) > 0 and output in cid - except Exception as e: - self.P(f"Error checking container existence: {e}", color='r') - return result - - - def _container_is_running(self, cid): - """ - Check if the container is still running similar to _container_exists. - """ - cmd = [self.cli_tool, "inspect", "-f", "{{.State.Running}}", cid] - try: - res = subprocess.run(cmd, capture_output=True, check=True) - is_running = res.stdout.decode("utf-8").strip() == "true" - except Exception as e: - self.P(f"Container status check: {e}", color='r') - is_running = False - return is_running - - def _container_kill(self, cid): - """ - Force kill a container by ID (if it exists). - """ - if not self._container_exists(cid): - self.P(f"Container {cid} does not exist. Cannot kill.") - return - # Use the CLI tool to kill the container - kill_cmd = [self.cli_tool, "rm", "-f", cid] - self.P(f"Stopping container {cid} ...") - res = subprocess.run(kill_cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE) - if res.returncode != 0: - err = res.stderr.decode("utf-8", errors="ignore") - self.P(f"Error stopping container {cid}: {err}", color='r') - else: - self.P(f"Container {cid} stopped successfully.") - return - - def _get_container_id(self): - cmd = [self.cli_tool, "ps", "-q", "-f", f"name={self.container_name}"] - try: - res = subprocess.run(cmd, capture_output=True, check=True) - container_id = res.stdout.decode("utf-8").strip() - if container_id: - self.container_id = container_id - self.P(f"Container ID: {self.container_id}") - return container_id - else: - self.Pd("No container found with the specified name.", color='r', score=29) - except subprocess.CalledProcessError as e: - self.P(f"Error getting container ID: {e}", color='r') - return None - - def _container_maybe_reload(self, force_restart=False): - """ - Check if the container is still running and perform the policy specified in the restart policy. - """ - if self.container_id is None: - self.Pd("Container ID is not set. Cannot check container status.") - return - - if self._is_manually_stopped == True: - self.Pd("Container is manually stopped. No action taken.") - return - - is_running = self._container_is_running(self.container_id) - - if force_restart: - self.P(f"Force restarting container {self.container_id} ...") - self._restart_container() - return - - if not is_running: - self.P(f"Container {self.container_id} has stopped.") - # Handle restart policy - if self.cfg_restart_policy == "always": - self.P(f"Restarting container {self.container_id} ...") - self._restart_container() - else: - self.P(f"Container {self.container_id} has stopped. No action taken.") - return - - def _restart_container(self): - self._container_kill(self.container_id) - self._reload_server() - self.container_id = None - self.container_start_time = self.time() # Reset the start time after restart - return - - def _maybe_set_container_id_and_show_app_info(self): - if self.container_id is None: - # this is the first time we are starting the container, so we need to get its ID - container_id = self._get_container_id() - if container_id: - self.container_id = container_id - self.P(f"Container ID set to: {self.container_id}") - self.on_post_container_start() # Call the lifecycle hoo - self._maybe_send_plugin_start_confirmation() - self._show_container_app_info() - #endif - #endif - return - def _maybe_send_plugin_start_confirmation(self): """ Sets up confirmation data about plugin start in CHAINSTORE. @@ -304,18 +104,18 @@ def _maybe_send_plugin_start_confirmation(self): self.chainstore_set(response_key, to_save) self.sleep(0.100) # wait 100 ms return - + def _setup_dynamic_env_var_host_ip(self): """ Definition for `host_ip` dynamic env var type. """ return self.log.get_localhost_ip() - + def _setup_dynamic_env_var_some_other_calc_type(self): """ Example definition for `some_other_calc_type` dynamic env var type. """ return "some_other_value" - def _setup_dynamic_env(self): + def _configure_dynamic_env(self): """ Set up dynamic environment variables based on the configuration. @@ -354,60 +154,6 @@ def _setup_dynamic_env(self): self.P(f"Dynamic env var {variable_name} = {variable_value}") #endfor each variable - def _show_container_app_info(self): - """ - Displays the current resource limits for the container. - This is a placeholder method and can be expanded as needed. - """ - cr_server, cr_username, cr_password = self._get_cr_data() - - msg = "Container info:\n" - msg += f" Container ID: {self.container_id}\n" - msg += f" Start Time: {self.time_to_str(self.container_start_time)}\n" - msg += f" Resource CPU: {self._cpu_limit} cores\n" - msg += f" Resource GPU: {self._gpu_limit}\n" - msg += f" Resource Mem: {self._mem_limit}\n" - msg += f" Target Image: {self.cfg_image}\n" - msg += f" CR: {cr_server}\n" - msg += f" CR User: {cr_username}\n" - msg += f" CR Pass: {'*' * len(cr_password) if cr_password else 'None'}\n" - msg += f" Env Vars: {self.cfg_env}\n" - msg += f" Cont. Port: {self.cfg_port}\n" - msg += f" Restart: {self.cfg_restart_policy}\n" - msg += f" Image Pull: {self.cfg_image_pull_policy}\n" - if self.volumes and len(self.volumes) > 0: - msg += " Volumes:\n" - for host_path, container_path in self.volumes.items(): - msg += f" Host {host_path} → Container {container_path}\n" - if self.extra_ports_mapping: - msg += " Extra Ports Mapping:\n" - for host_port, container_port in self.extra_ports_mapping.items(): - msg += f" Host {host_port} → Container {container_port}\n" - msg += f" Ngrok Host Port: {self.port}\n" - msg += f" CLI Tool: {self.cli_tool}\n" - self.P(msg) - return - - - def _run_command_in_container(self, command): - """ - Run a command inside the container. - - Args: - command (str): The command to run inside the container. - """ - if not self.container_id: - self.P("Container ID is not set. Cannot run command.") - return - - cmd = [self.cli_tool, "exec", "-i", self.container_id] + command.split() - try: - result = subprocess.run(cmd, capture_output=True, text=True, check=True) - self.P(f"Command output: {result.stdout}") - except subprocess.CalledProcessError as e: - self.P(f"Error running command in container: {e.stderr}", color='r') - - return ## END CONTAINER MIXIN ### ### NEW CONTAINER MIXIN METHODS ### @@ -539,10 +285,11 @@ def _setup_resource_limits_and_ports(self): self.port = self._allocate_port(allow_dynamic=True) # Allocate a port for the container if needed return - def _setup_volumes(self): + def _configure_volumes(self): """ Processes the volumes specified in the configuration. """ + default_volume_rights = "rw" if hasattr(self, 'cfg_volumes') and self.cfg_volumes and len(self.cfg_volumes) > 0: for host_path, container_path in self.cfg_volumes.items(): original_path = str(host_path) @@ -553,7 +300,7 @@ def _setup_volumes(self): self.P(f" Converted '{original_path}' → named volume '{prefixed_name}'") full_host_path = self.os_path.join(CONTAINER_VOLUMES_PATH, prefixed_name) - self.volumes[full_host_path] = container_path + self.volumes[full_host_path] = {"bind": container_path, "mode": default_volume_rights} # endfor each host path # endif volumes @@ -563,99 +310,115 @@ def _setup_volumes(self): ### END NEW CONTAINER MIXIN METHODS ### ### COMMON CONTAINER UTILITY METHODS ### - + def _setup_env_and_ports(self): + # Environment variables + # allow cfg_env to override default env vars + self.env = self._get_default_env_vars() + self.env.update(self.dynamic_env) + if self.cfg_env: + self.env.update(self.cfg_env) + if self.dynamic_env: + self.env.update(self.dynamic_env) + # endif dynamic env + + # Ports mapping + ports_mapping = self.extra_ports_mapping.copy() if self.extra_ports_mapping else {} + if self.cfg_port and self.port: + ports_mapping[self.port] = self.cfg_port + # end if main port + self.inverted_ports_mapping = {f"{v}/tcp": str(k) for k, v in ports_mapping.items()} + + return + def _validate_container_config(self): """Validate container configuration before starting.""" if not self.cfg_image: raise ValueError("IMAGE is required") - + if not isinstance(self.cfg_image, str): raise ValueError("IMAGE must be a string") - + # Validate container resources if provided if hasattr(self, 'cfg_container_resources') and self.cfg_container_resources: if not isinstance(self.cfg_container_resources, dict): raise ValueError("CONTAINER_RESOURCES must be a dictionary") - + # Validate environment variables if provided if hasattr(self, 'cfg_env') and self.cfg_env: if not isinstance(self.cfg_env, dict): raise ValueError("ENV must be a dictionary") - + return True - def _get_container_health_status(self): - """Get container health status.""" - if not hasattr(self, 'container_id') or not self.container_id: + def _get_container_health_status(self, container=None): + """Get container health status using Docker client.""" + if container is None: + container = getattr(self, 'container', None) + + if container is None: return "not_started" - + try: - is_running = self._container_is_running(self.container_id) - return "running" if is_running else "stopped" + container.reload() + return container.status except Exception as e: self.P(f"Error checking container health: {e}", color='r') return "error" - def _cleanup_container_resources(self): - """Clean up container resources on shutdown.""" - if hasattr(self, 'container_id') and self.container_id: - self.P(f"Cleaning up container resources for {self.container_id}", color='b') - self._container_kill(self.container_id) - self.container_id = None - self.P("Container resources cleaned up", color='g') def _validate_git_config(self): """Validate Git configuration for repository access.""" if not hasattr(self, 'cfg_git_repo_owner') or not hasattr(self, 'cfg_git_repo_name'): return False - + if not self.cfg_git_repo_owner or not self.cfg_git_repo_name: self.P("Git repository owner or name not configured", color='y') return False - + # Check if we have credentials for private repos if hasattr(self, 'cfg_git_token') and not self.cfg_git_token: self.P("Warning: No Git token provided, repository must be public", color='y') - + return True def _validate_endpoint_config(self): """Validate endpoint configuration for health checks.""" if not hasattr(self, 'cfg_endpoint_url') or not self.cfg_endpoint_url: return False - + # Basic URL validation if not isinstance(self.cfg_endpoint_url, str): self.P("Endpoint URL must be a string", color='r') return False - + if not self.cfg_endpoint_url.startswith('/'): self.P("Endpoint URL must start with '/'", color='r') return False - + if '..' in self.cfg_endpoint_url: self.P("Endpoint URL contains invalid path traversal", color='r') return False - + return True def _get_container_info(self): """Get comprehensive container information.""" + container = getattr(self, 'container', None) info = { - 'container_id': getattr(self, 'container_id', None), + 'container_id': container.short_id if container else None, 'container_name': getattr(self, 'container_name', None), 'image': getattr(self, 'cfg_image', None), - 'status': self._get_container_health_status(), + 'status': self._get_container_health_status(container), 'port': getattr(self, 'port', None), 'start_time': getattr(self, 'container_start_time', None), } - + if hasattr(self, 'extra_ports_mapping') and self.extra_ports_mapping: info['extra_ports'] = self.extra_ports_mapping - + if hasattr(self, 'volumes') and self.volumes: info['volumes'] = self.volumes - + return info def _log_container_info(self): @@ -670,27 +433,30 @@ def _validate_port_allocation(self, port): """Validate that a port is properly allocated.""" if not port: return False - + if not isinstance(port, int): return False - + if port < 1 or port > 65535: return False - + return True def _safe_get_container_stats(self): """Safely get container statistics without raising exceptions.""" - if not hasattr(self, 'container_id') or not self.container_id: + container = getattr(self, 'container', None) + if not container: return None - + try: - # This would need to be implemented based on the container runtime - # For now, return basic info + container.reload() return { - 'id': self.container_id, - 'status': self._get_container_health_status(), - 'running': self._container_is_running(self.container_id) if self.container_id else False + 'id': container.short_id, + 'status': container.status, + 'running': container.status == "running", + 'image': container.image.tags[0] if container.image.tags else str(container.image.id), + 'created': container.attrs.get('Created', 'Unknown'), + 'ports': container.attrs.get('NetworkSettings', {}).get('Ports', {}) } except Exception as e: self.P(f"Error getting container stats: {e}", color='r') diff --git a/extensions/business/container_apps/worker_app_runner.py b/extensions/business/container_apps/worker_app_runner.py index 80df7d65..81c4c1bd 100644 --- a/extensions/business/container_apps/worker_app_runner.py +++ b/extensions/business/container_apps/worker_app_runner.py @@ -63,20 +63,20 @@ "PASSWORD": None, # Optional registry password or token }, + "VCS_DATA": { + "PROVIDER": "github", # currently only "github" is supported + "USERNAME": None, # GitHub username for cloning (if private repo) + "TOKEN": None, # GitHub personal access token for cloning (if private repo) + "REPO_OWNER": None, # GitHub repository owner (user or org) + "REPO_NAME": None, # GitHub repository name + "BRANCH": "main", # branch to monitor for updates + "POLL_INTERVAL": 60, # seconds between Git commit checks + }, + # Environment variables for the container "ENV": {}, "DYNAMIC_ENV": {}, - # Git config - "GIT_USERNAME": None, # GitHub username for cloning (if private repo) - "GIT_TOKEN": None, # GitHub personal access token for cloning (if private repo) - "GIT_REPO_OWNER": None, # GitHub repository owner (user or org) - "GIT_REPO_NAME": None, # GitHub repository name - - # Git monitoring configuration - "GIT_BRANCH": "main", # branch to monitor for updates - "GIT_POLL_INTERVAL": 60, # seconds between Git commit checks - # Docker image monitoring "IMAGE_POLL_INTERVAL": 300, # seconds between Docker image checks @@ -87,7 +87,7 @@ # Application endpoint polling "ENDPOINT_POLL_INTERVAL": 30, # seconds between endpoint health checks - "ENDPOINT_URL": "/edgenode", # endpoint to poll for health checks + "ENDPOINT_URL": None, # endpoint to poll for health checks, for example "/edgenode" or "/health" "PORT": None, # internal container port if it's a web app (int) # Container resource limits @@ -100,122 +100,8 @@ }, # Chainstore response configuration - "CHAINSTORE_RESPONSE_KEY": '', # Optional key to send confirmation data to chainstore - - 'VALIDATION_RULES': { - **BasePlugin.CONFIG['VALIDATION_RULES'], - - 'IMAGE': { - 'TYPE': 'str', - 'DESCRIPTION': 'Docker image to use for the container', - 'REQUIRED': True, - }, - - 'BUILD_AND_RUN_COMMANDS': { - 'TYPE': 'list', - 'DESCRIPTION': 'Commands to run in container for building and starting the app', - 'MIN_LEN': 1, - }, - - 'GIT_REPO_OWNER': { - 'TYPE': 'str', - 'DESCRIPTION': 'GitHub repository owner (user or org)', - }, - - 'GIT_REPO_NAME': { - 'TYPE': 'str', - 'DESCRIPTION': 'GitHub repository name', - }, - - 'GIT_USERNAME': { - 'TYPE': 'str', - 'DESCRIPTION': 'GitHub username for cloning (if private repo)', - }, - - 'GIT_TOKEN': { - 'TYPE': 'str', - 'DESCRIPTION': 'GitHub personal access token for cloning (if private repo)', - 'DEFAULT': '', - }, - - 'GIT_BRANCH': { - 'TYPE': 'str', - 'DESCRIPTION': 'Branch to monitor for updates', - 'DEFAULT': 'main', - }, - - 'GIT_POLL_INTERVAL': { - 'TYPE': 'int', - 'MIN_VAL': 10, - 'MAX_VAL': 3600, - 'DESCRIPTION': 'Seconds between Git commit checks', - 'DEFAULT': 90, - }, - - 'IMAGE_POLL_INTERVAL': { - 'TYPE': 'int', - 'MIN_VAL': 60, - 'MAX_VAL': 3600, - 'DESCRIPTION': 'Seconds between Docker image checks', - 'DEFAULT': 600, - }, - - 'ENDPOINT_POLL_INTERVAL': { - 'TYPE': 'int', - 'MIN_VAL': 5, - 'MAX_VAL': 300, - 'DESCRIPTION': 'Seconds between endpoint health checks', - }, - - 'ENDPOINT_URL': { - 'TYPE': 'str', - 'DESCRIPTION': 'Endpoint to poll for health checks', - 'DEFAULT': None, - }, - - 'PORT': { - 'TYPE': 'int', - 'MIN_VAL': 1, - 'MAX_VAL': 65535, - 'DESCRIPTION': 'Internal container port if it\'s a web app', - }, - - 'RESTART_POLICY': { - 'TYPE': 'str', - 'DESCRIPTION': 'Container restart policy', - 'ALLOWED_VALUES': ['always', 'on-failure', 'unless-stopped', 'no'], - 'DEFAULT': 'always', - }, - - 'IMAGE_PULL_POLICY': { - 'TYPE': 'str', - 'DESCRIPTION': 'Docker image pull policy', - 'ALLOWED_VALUES': ['always', 'if-not-present', 'never'], - 'DEFAULT': 'always', - }, - - 'CONTAINER_RESOURCES': { - 'TYPE': 'dict', - 'DESCRIPTION': 'Container resource limits (CPU, GPU, memory, ports)', - }, - - 'CR_DATA': { - 'TYPE': 'dict', - 'DESCRIPTION': 'Container registry data (server, username, password)', - }, - - 'ENV': { - 'TYPE': 'dict', - 'DESCRIPTION': 'Environment variables for the container', - 'DEFAULT': {}, - }, - - 'DYNAMIC_ENV': { - 'TYPE': 'dict', - 'DESCRIPTION': 'Dynamic environment variables for the container', - 'DEFAULT': {}, - } - }, + "CHAINSTORE_RESPONSE_KEY": None, # Optional key to send confirmation data to chainstore + } @@ -239,12 +125,10 @@ def on_init(self): self._set_default_branch() self._setup_resource_limits_and_ports() # setup container resource limits (CPU, GPU, memory, ports) - self._setup_dynamic_env() # setup dynamic env vars for the container + self._configure_dynamic_env() # setup dynamic env vars for the container - self.repo_url = f"https://{self.cfg_git_username}:{self.cfg_git_token}@github.com/{self.cfg_git_repo_owner}/{self.cfg_git_repo_name}.git" - - # Initialize tunnel process - self.tunnel_process = None + vcs_data = self.cfg_vcs_data or {} + self.repo_url = f"https://{vcs_data.get('USERNAME')}:{vcs_data.get('TOKEN')}@github.com/{vcs_data.get('REPO_OWNER')}/{vcs_data.get('REPO_NAME')}.git" self.P(f"WorkerAppRunnerPlugin initialized (version {__VER__})", color='g') return @@ -263,6 +147,9 @@ def __reset_vars(self): self._last_image_check = 0 self._last_endpoint_check = 0 + # Initialize tunnel process + self.tunnel_process = None + # Determine default branch via GitHub API (so we know which branch to monitor) self.branch = None @@ -309,18 +196,22 @@ def stop_tunnel_engine(self): def _set_default_branch(self): """Determine the default branch of the repository via GitHub API.""" - if self.cfg_git_repo_owner and self.cfg_git_repo_name: + vcs_data = self.cfg_vcs_data or {} + repo_owner = vcs_data.get('REPO_OWNER') + repo_name = vcs_data.get('REPO_NAME') + + if repo_owner and repo_name: try: resp = self._get_latest_commit(return_data=True) if resp is not None: _, data = resp self.P(f"Repository info:\n {json.dumps(data, indent=2)}", color='b') self.branch = data.get("default_branch", None) - self.P(f"Default branch for {self.cfg_git_repo_owner}/{self.cfg_git_repo_name} is '{self.branch}'", color='y') + self.P(f"Default branch for {repo_owner}/{repo_name} is '{self.branch}'", color='y') except Exception as e: self.P(f"[WARN] Could not determine default branch: {e}") if not self.branch: - self.branch = "main" # Fallback to 'main' if not determined + self.branch = vcs_data.get('BRANCH', "main") # Use VCS_DATA branch or fallback to 'main' return @@ -468,6 +359,9 @@ def _check_health_endpoint(self, current_time=None): if not self.container or not self.cfg_endpoint_url: return + if not current_time: + current_time = self.time() + if current_time - self._last_endpoint_check >= self.cfg_endpoint_poll_interval: self._last_endpoint_check = current_time self._poll_endpoint() @@ -504,15 +398,20 @@ def _poll_endpoint(self): def _get_latest_commit(self, return_data=False): """Fetch the latest commit SHA of the repository's monitored branch via GitHub API.""" - if not self.cfg_git_repo_owner or not self.cfg_git_repo_name: + vcs_data = self.cfg_vcs_data or {} + repo_owner = vcs_data.get('REPO_OWNER') + repo_name = vcs_data.get('REPO_NAME') + token = vcs_data.get('TOKEN') + + if not repo_owner or not repo_name: self.P("Git repository owner or name not configured", color='y') return None if self.branch is None: - api_url = f"https://api.github.com/repos/{self.cfg_git_repo_owner}/{self.cfg_git_repo_name}" + api_url = f"https://api.github.com/repos/{repo_owner}/{repo_name}" else: - api_url = f"https://api.github.com/repos/{self.cfg_git_repo_owner}/{self.cfg_git_repo_name}/branches/{self.branch}" - headers = {"Authorization": f"token {self.cfg_git_token}"} if self.cfg_git_token else {} + api_url = f"https://api.github.com/repos/{repo_owner}/{repo_name}/branches/{self.branch}" + headers = {"Authorization": f"token {token}"} if token else {} try: self.P(f"Commit check: {api_url}", color='b') @@ -540,7 +439,14 @@ def _get_latest_commit(self, return_data=False): def _check_git_updates(self, current_time=None): """Check for a new commit in the monitored branch and restart container if found.""" - if current_time - self._last_git_check < self.cfg_git_poll_interval: + if not current_time: + current_time = self.time() + + vcs_data = self.cfg_vcs_data or {} + poll_interval = vcs_data.get('POLL_INTERVAL', 60) + + if current_time - self._last_git_check >= poll_interval: + self._last_git_check = current_time latest_commit = self._get_latest_commit() if latest_commit and self.current_commit and latest_commit != self.current_commit: self.P(f"New commit detected ({latest_commit[:7]} != {self.current_commit[:7]}). Restarting container...", color='y') @@ -585,14 +491,20 @@ def _get_latest_image_hash(self): img = self.docker_client.images.pull(self.cfg_image) # docker-py may return Image or list[Image] if isinstance(img, list) and img: + self.P("Multiple images returned, using the last one", color='y') + self.P(self.json_dumps([i.id for i in img]), color='y') img = img[-1] # Ensure attributes loaded + self.P(f"Image pulled: {getattr(img, 'id', 'unknown id')}", color='g') try: img.reload() except Exception as e: self.P(f"Warning: Could not reload image attributes: {e}", color='y') # end try + self.P("Image loaded") + + attrs = getattr(img, "attrs", {}) or {} repo_digests = attrs.get("RepoDigests") or [] if repo_digests: @@ -628,6 +540,8 @@ def _get_latest_image_hash(self): def _check_image_updates(self, current_time=None): """Check for a new version of the Docker image and restart container if found.""" + if not current_time: + current_time = self.time() if current_time - self._last_image_check >= self.cfg_image_poll_interval: self._last_image_check = current_time latest_image_hash = self._get_latest_image_hash() @@ -711,6 +625,8 @@ def _perform_periodic_monitoring(self): self._check_health_endpoint(current_time) self._check_git_updates(current_time) + + self._check_image_updates(current_time) return diff --git a/ver.py b/ver.py index 2ecc3786..ec383a4a 100644 --- a/ver.py +++ b/ver.py @@ -1,2 +1,2 @@ -__VER__ = '2.9.650' +__VER__ = '2.9.651' From 2104d43b2d46f1874ee42b480e4f47b21e487a80 Mon Sep 17 00:00:00 2001 From: Cristi Bleotiu Date: Fri, 12 Sep 2025 14:11:05 +0300 Subject: [PATCH 2/4] feat: moved presence threshold for ora sync to network constants + optimized processing of received agreements in ora sync --- .../sync_mixins/ora_sync_constants.py | 9 +- .../sync_mixins/ora_sync_states_mixin.py | 98 ++++++++++++------- .../sync_mixins/ora_sync_utils_mixin.py | 31 +++++- ver.py | 2 +- 4 files changed, 93 insertions(+), 47 deletions(-) diff --git a/extensions/business/oracle_sync/sync_mixins/ora_sync_constants.py b/extensions/business/oracle_sync/sync_mixins/ora_sync_constants.py index f51b106a..14d0d0da 100644 --- a/extensions/business/oracle_sync/sync_mixins/ora_sync_constants.py +++ b/extensions/business/oracle_sync/sync_mixins/ora_sync_constants.py @@ -1,4 +1,8 @@ -from naeural_core.constants import SUPERVISOR_MIN_AVAIL_PRC, EPOCH_MAX_VALUE, ORACLE_SYNC_USE_R1FS +from naeural_core.constants import ( + SUPERVISOR_MIN_AVAIL_PRC, EPOCH_MAX_VALUE, ORACLE_SYNC_USE_R1FS, + ORACLE_SYNC_BLOCKCHAIN_PRESENCE_MIN_THRESHOLD, + ORACLE_SYNC_ONLINE_PRESENCE_MIN_THRESHOLD +) MAX_RECEIVED_MESSAGES_SIZE = 1000 DEBUG_MODE = False @@ -21,9 +25,6 @@ ORACLE_SYNC_ACCEPTED_REPORTS_THRESHOLD = 0 ORACLE_SYNC_ACCEPTED_MEDIAN_ERROR_MARGIN = EPOCH_MAX_VALUE - POTENTIALLY_FULL_AVAILABILITY_THRESHOLD -ORACLE_SYNC_BLOCKCHAIN_PRESENCE_MIN_THRESHOLD = 0.3 -ORACLE_SYNC_ONLINE_PRESENCE_MIN_THRESHOLD = 0.4 - ORACLE_SYNC_IGNORE_REQUESTS_SECONDS = 3 * 60 # 3 minutes before the epoch end requests will be ignored class OracleSyncCt: diff --git a/extensions/business/oracle_sync/sync_mixins/ora_sync_states_mixin.py b/extensions/business/oracle_sync/sync_mixins/ora_sync_states_mixin.py index e27d22f4..4892c7ef 100644 --- a/extensions/business/oracle_sync/sync_mixins/ora_sync_states_mixin.py +++ b/extensions/business/oracle_sync/sync_mixins/ora_sync_states_mixin.py @@ -1217,10 +1217,63 @@ def handle_received_agreed_median_table(self, dct_message: dict): if not self._check_received_epoch__agreed_median_table_ok(sender, oracle_data): return success + first_needed_epoch = self._last_epoch_synced + 1 + last_needed_epoch = self._current_epoch - 1 + epochs_range = range(first_needed_epoch, last_needed_epoch + 1) + + # In case the message contains more than the needed epoch keys + # only the needed keys will be downloaded from R1FS if needed. + needed_epoch_keys = [ + str(x) + for x in epochs_range + ] + + # sort epoch_keys in ascending order + epoch_keys = oracle_data[OracleSyncCt.EPOCH_KEYS] + received_epochs = sorted(list(set(epoch_keys))) + # filter received epochs to keep only the needed ones + received_epochs = [ + epoch for epoch in received_epochs + if epoch in epochs_range + ] + + stage = oracle_data[OracleSyncCt.STAGE] + log_str = self.log_received_message( + sender=sender, + data=self.dct_agreed_availability_signatures, + stage=stage, + return_str=True + ) + log_str += f", {received_epochs = }\n" + log_str += f"Keeping only tables for epochs [{first_needed_epoch}, {last_needed_epoch}]" + self.P(log_str) + + cnt_expected_epochs = last_needed_epoch - first_needed_epoch + 1 + enough_data_received = len(received_epochs) == cnt_expected_epochs + limits_received = first_needed_epoch in received_epochs and last_needed_epoch in received_epochs + + if not enough_data_received or not limits_received: + # Expected epochs in range [last_epoch_synced + 1, current_epoch - 1] + # received epochs don t contain the full range + if self.cfg_debug_sync: + min_epoch = min(received_epochs) if len(received_epochs) > 0 else None + max_epoch = max(received_epochs) if len(received_epochs) > 0 else None + cnt_epochs = len(received_epochs) + msg = f'Expected epochs {cnt_epochs} in range [{first_needed_epoch}, {last_needed_epoch}] ' + msg += f'and received only {len(received_epochs)} epochs (min: {min_epoch}, max: {max_epoch}). ' + msg += f'Ignoring...' + self.P(msg, color='r') + return success + # endif received epochs not containing the full requested interval + # Here, both the agreed median table and the agreement signatures should have the same keys, # but as string instead of int. We also know that in epoch_keys we have # the keys in int format. Thus, we need to convert the keys of the received tables dct_epoch_agreed_median_table = oracle_data[OracleSyncCt.EPOCH__AGREED_MEDIAN_TABLE] + dct_epoch_agreed_median_table = { + k: v + for k, v in dct_epoch_agreed_median_table.items() if k in needed_epoch_keys + } agreement_success, dct_epoch_agreed_median_table, dct_epoch_agreement_cid = self.r1fs_get_data_from_nested_message( nested_message_dict=dct_epoch_agreed_median_table, return_cids=True, @@ -1231,9 +1284,14 @@ def handle_received_agreed_median_table(self, dct_message: dict): return success # endif not agreement_success dct_epoch_agreement_signatures = oracle_data[OracleSyncCt.EPOCH__AGREEMENT_SIGNATURES] + dct_epoch_agreement_signatures = { + k: v + for k, v in dct_epoch_agreement_signatures.items() if k in needed_epoch_keys + } signatures_success, dct_epoch_agreement_signatures, dct_epoch_signatures_cid = self.r1fs_get_data_from_nested_message( nested_message_dict=dct_epoch_agreement_signatures, return_cids=True, + process_only_keys=needed_epoch_keys ) if not signatures_success: if self.cfg_debug_sync: @@ -1241,7 +1299,6 @@ def handle_received_agreed_median_table(self, dct_message: dict): return success # endif not signatures_success dct_epoch_is_valid = oracle_data[OracleSyncCt.EPOCH__IS_VALID] - epoch_keys = oracle_data[OracleSyncCt.EPOCH_KEYS] id_to_node_address = oracle_data.get(OracleSyncCt.ID_TO_NODE_ADDRESS, {}) # Unsqueeze the epoch dictionaries if they are squeezed. [dct_epoch_agreed_median_table, dct_epoch_agreement_signatures] = self._maybe_unsqueeze_epoch_dictionaries( @@ -1249,9 +1306,6 @@ def handle_received_agreed_median_table(self, dct_message: dict): id_to_keys=id_to_node_address, ) - # sort epoch_keys in ascending order - received_epochs = sorted(epoch_keys) - # convert to dict with int keys dct_epoch_agreed_median_table = { # In case the agreement table is sent through R1FS, the keys will already be in int format. @@ -1282,7 +1336,6 @@ def handle_received_agreed_median_table(self, dct_message: dict): self.P(msg) # endif debug_sync_full - message_invalid = False for epoch, agreed_median_table in dct_epoch_agreed_median_table.items(): # At this point we did not need to convert the keys of the dictionaries yet, # because in valid messages both the agreement table and the agreement signatures @@ -1306,40 +1359,11 @@ def handle_received_agreed_median_table(self, dct_message: dict): debug=False ): # if one signature for the received table is invalid, ignore the entire message - message_invalid = True - break + if self.cfg_debug_sync: + self.P(f"Received invalid availability table from {sender = }. Ignoring", color='r') + return success # end for epoch agreed table - if message_invalid: - if self.cfg_debug_sync: - self.P(f"Received invalid availability table from {sender = }. Ignoring", color='r') - return success - # endif - - if self._last_epoch_synced + 1 not in received_epochs or self._current_epoch - 1 not in received_epochs: - # Expected epochs in range [last_epoch_synced + 1, current_epoch - 1] - # received epochs don t contain the full range - if self.cfg_debug_sync: - min_epoch = min(received_epochs) if len(received_epochs) > 0 else None - max_epoch = max(received_epochs) if len(received_epochs) > 0 else None - msg = (f'Expected epochs in range [{self._last_epoch_synced + 1}, {self._current_epoch - 1}] ' - f'and received only {len(received_epochs)} epochs (min: {min_epoch}, max: {max_epoch}). ' - f'Ignoring...') - self.P(msg, color='r') - return success - # endif received epochs not containing the full requested interval - - stage = oracle_data[OracleSyncCt.STAGE] - log_str = self.log_received_message( - sender=sender, - data=self.dct_agreed_availability_signatures, - stage=stage, - return_str=True - ) - log_str += f", {received_epochs = }\n" - log_str += f"Keeping only tables for epochs [{self._last_epoch_synced + 1}, {self._current_epoch - 1}]" - self.P(log_str) - epochs_range = range(self._last_epoch_synced + 1, self._current_epoch) self.dct_agreed_availability_table[sender] = { # No need for get here, since in S0 we send a continuous range of epochs. i: dct_epoch_agreed_median_table[i] diff --git a/extensions/business/oracle_sync/sync_mixins/ora_sync_utils_mixin.py b/extensions/business/oracle_sync/sync_mixins/ora_sync_utils_mixin.py index ccd2ff5c..246e53fc 100644 --- a/extensions/business/oracle_sync/sync_mixins/ora_sync_utils_mixin.py +++ b/extensions/business/oracle_sync/sync_mixins/ora_sync_utils_mixin.py @@ -151,6 +151,7 @@ def r1fs_get_data_from_nested_message( self, nested_message_dict: dict, ignore_keys: list = None, + process_only_keys: list = None, return_cids: bool = False, debug=True ): @@ -168,6 +169,10 @@ def r1fs_get_data_from_nested_message( A list of keys to ignore when extracting data from the message. This can be used to skip certain keys that are not relevant for the extraction. By default, None, which means no keys will be ignored. + process_only_keys : list, optional + If provided, only the keys in this list will be processed. + If empty list, no keys will be processed. + By default, None, which means all keys will be processed. debug : bool, optional Whether to print debug messages, by default True @@ -192,20 +197,36 @@ def r1fs_get_data_from_nested_message( self.P(f"`ignore_keys` is {type(ignore_keys)} != list. Using empty list instead.", color='r') ignore_keys = [] # endif ignore_keys is not list + + if isinstance(process_only_keys, str): + process_only_keys = [process_only_keys] + # endif process_only_keys is str + if not isinstance(process_only_keys, list): + if debug and process_only_keys is not None: + self.P(f"`process_only_keys` is {type(process_only_keys)} != list. Using None instead.", color='r') + process_only_keys = None + else: + if debug: + self.P(f"Processing only keys in `process_only_keys`: {process_only_keys}.") + # endif process_only_keys is not list + updated_values = {} cids = {} for key, msg_data in nested_message_dict.items(): - # 1. Check if the key is in the ignore keys. + # 1. Check if process_only_keys is provided and if the current key is in it. + if process_only_keys is not None and key not in process_only_keys: + continue + # 2. Check if the key is in the ignore keys. if key in ignore_keys: continue - # 2. Check if the data is a CID or data. + # 3. Check if the data is a CID or data. if isinstance(msg_data, str): if debug: self.P(f"Attempting to get data from R1FS using CID {msg_data}.") - # 3. Attempt to get the data from R1FS. + # 4. Attempt to get the data from R1FS. res = self.r1fs_get_pickle(cid=msg_data, debug=debug) if res is not None and debug: - # 4. If the retrieval was successful, store the result. + # 5. If the retrieval was successful, store the result. updated_values[key] = res cids[key] = msg_data self.P(f"Successfully retrieved data from R1FS using CID {msg_data}.") @@ -214,7 +235,7 @@ def r1fs_get_data_from_nested_message( break # endif # endfor key, data - # 5. Update the nested message dictionary with the retrieved values. + # 6. Update the nested message dictionary with the retrieved values. nested_message_dict.update(updated_values) return (success, nested_message_dict, cids) if return_cids else (success, nested_message_dict) diff --git a/ver.py b/ver.py index ec383a4a..13f63ba3 100644 --- a/ver.py +++ b/ver.py @@ -1,2 +1,2 @@ -__VER__ = '2.9.651' +__VER__ = '2.9.652' From 1f8ec379212583871d7c2ba5abed63fb00360554 Mon Sep 17 00:00:00 2001 From: Cristi Bleotiu Date: Mon, 15 Sep 2025 18:47:42 +0300 Subject: [PATCH 3/4] fix: ora sync logging --- .../business/oracle_sync/sync_mixins/ora_sync_states_mixin.py | 2 +- ver.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/extensions/business/oracle_sync/sync_mixins/ora_sync_states_mixin.py b/extensions/business/oracle_sync/sync_mixins/ora_sync_states_mixin.py index 4892c7ef..2dd9ec1a 100644 --- a/extensions/business/oracle_sync/sync_mixins/ora_sync_states_mixin.py +++ b/extensions/business/oracle_sync/sync_mixins/ora_sync_states_mixin.py @@ -1360,7 +1360,7 @@ def handle_received_agreed_median_table(self, dct_message: dict): ): # if one signature for the received table is invalid, ignore the entire message if self.cfg_debug_sync: - self.P(f"Received invalid availability table from {sender = }. Ignoring", color='r') + self.P(f"Received invalid availability table from {sender = }[{epoch=} invalid]. Ignoring", color='r') return success # end for epoch agreed table diff --git a/ver.py b/ver.py index 13f63ba3..7227afc1 100644 --- a/ver.py +++ b/ver.py @@ -1,2 +1,2 @@ -__VER__ = '2.9.652' +__VER__ = '2.9.653' From e470ff8919ef153b7c237fe6f19af3f5a89b9469 Mon Sep 17 00:00:00 2001 From: Vitalii <87299468+vitalii-t12@users.noreply.github.com> Date: Tue, 16 Sep 2025 13:19:19 +0300 Subject: [PATCH 4/4] Fix (HOT) CAR send chainstore response (#244) * fix: send plugin start confirmation for CAR * fix: file path for base64 file read * chore: increment version * fix: add async job deeploy description --- .../container_apps/container_app_runner.py | 3 +++ .../business/container_apps/container_utils.py | 15 --------------- extensions/business/deeploy/deeploy_mixin.py | 15 +++++++++++++++ extensions/business/r1fs/r1fs_manager_api.py | 1 + ver.py | 2 +- 5 files changed, 20 insertions(+), 16 deletions(-) diff --git a/extensions/business/container_apps/container_app_runner.py b/extensions/business/container_apps/container_app_runner.py index 502e4e8f..469dffe0 100644 --- a/extensions/business/container_apps/container_app_runner.py +++ b/extensions/business/container_apps/container_app_runner.py @@ -330,6 +330,9 @@ def start_container(self): self.container_id = self.container.short_id self.P(f"Container started (ID: {self.container.short_id})", color='g') + + self._maybe_send_plugin_start_confirmation() + return self.container def stop_container(self): diff --git a/extensions/business/container_apps/container_utils.py b/extensions/business/container_apps/container_utils.py index 3896ca22..188eb808 100644 --- a/extensions/business/container_apps/container_utils.py +++ b/extensions/business/container_apps/container_utils.py @@ -366,21 +366,6 @@ def _get_container_health_status(self, container=None): return "error" - def _validate_git_config(self): - """Validate Git configuration for repository access.""" - if not hasattr(self, 'cfg_git_repo_owner') or not hasattr(self, 'cfg_git_repo_name'): - return False - - if not self.cfg_git_repo_owner or not self.cfg_git_repo_name: - self.P("Git repository owner or name not configured", color='y') - return False - - # Check if we have credentials for private repos - if hasattr(self, 'cfg_git_token') and not self.cfg_git_token: - self.P("Warning: No Git token provided, repository must be public", color='y') - - return True - def _validate_endpoint_config(self): """Validate endpoint configuration for health checks.""" if not hasattr(self, 'cfg_endpoint_url') or not self.cfg_endpoint_url: diff --git a/extensions/business/deeploy/deeploy_mixin.py b/extensions/business/deeploy/deeploy_mixin.py index 2c674366..617e7e9e 100644 --- a/extensions/business/deeploy/deeploy_mixin.py +++ b/extensions/business/deeploy/deeploy_mixin.py @@ -181,6 +181,21 @@ def __get_pipeline_responses(self, response_keys, timeout_seconds=300): tuple: (dct_status, str_status) where: dct_status: Dictionary of response statuses str_status: Overall status ('success', 'timeout', or 'pending') + + Async job deeploy: + + 1. Oracle A receives launch and sends command then responds with command-hash "X" + setting `async_status=true` so that the API does NOT return immediately. + Default async_status=true! + X must be unique and stored in the pipeline definition (maybe the pipeline name or job_id) + 2. UI checks X via Oracle B + 3. Oracle B checks pipeline status (already received via net-config) via netmon AND looks at + chainstore-response var from plugin instance and will respond False (not ready yet) + 4. UI again checks X via Oracle C + 5. Oracle C checks pipeline status via netmon AND looks at chainstore-response and sees + correct status (job status updated from CAR/WAR to CStore) + 6. UI shows success + """ dct_status = {} str_status = DEEPLOY_STATUS.PENDING diff --git a/extensions/business/r1fs/r1fs_manager_api.py b/extensions/business/r1fs/r1fs_manager_api.py index 2eb21e4d..e38766c2 100644 --- a/extensions/business/r1fs/r1fs_manager_api.py +++ b/extensions/business/r1fs/r1fs_manager_api.py @@ -221,6 +221,7 @@ def get_file_base64(self, cid: str, secret: str = None): # first parameter must self.P(f"Trying to download file -> {cid}") file = self.r1fs.get_file(cid=cid, secret=secret) + file = file.replace("/edge_node", ".") if file else file filename = file.split('/')[-1] if file else None self.P(f"File retrieved: {file}") file_base64 = self.diskapi_load_r1fs_file(file, verbose=True, to_base64=True) diff --git a/ver.py b/ver.py index 7227afc1..40cf7004 100644 --- a/ver.py +++ b/ver.py @@ -1,2 +1,2 @@ -__VER__ = '2.9.653' +__VER__ = '2.9.660'