Merge pull request #291 from Ratio1/develop

Develop

Author: cristibleotiu

extensions/business/container_apps/container_app_runner.py

@@ -717,3 +717,140 @@ def _validate_docker_image_format(self, image_name):
     return True
 
   ### END COMMON CONTAINER UTILITY METHODS ###
+
+  ### EXTRA TUNNELS METHODS ###
+
+  def _normalize_extra_tunnel_config(self, container_port, config):
+    """
+    Normalize extra tunnel config to standard format.
+
+    Currently supports simple string format: "token"
+    Named "tunnel config" for future extensibility (e.g., dict with engine, enabled, etc.)
+
+    Args:
+      container_port: Container port (int)
+      config: Token string (future: could be dict with {"token": str, "engine": str, "enabled": bool})
+
+    Returns:
+      str: Normalized token string (future: could return dict)
+
+    Note:
+      This method is designed to be extended in the future to support more complex
+      tunnel configurations beyond simple token strings, such as:
+      - Different tunnel engines (cloudflare, ngrok, custom)
+      - Per-tunnel enabled/disabled flags
+      - Custom tunnel parameters
+      For now, it only handles string tokens for simplicity.
+    """
+    if isinstance(config, str):
+      return config.strip()
+    else:
+      raise ValueError(
+        f"EXTRA_TUNNELS[{container_port}] must be a string token, got {type(config)}"
+      )
+
+  def _allocate_extra_tunnel_ports(self, container_ports):
+    """
+    Allocate host ports for container ports defined only in EXTRA_TUNNELS.
+
+    This handles the case where CONTAINER_RESOURCES["ports"] is empty but
+    EXTRA_TUNNELS defines ports that need to be exposed.
+
+    Args:
+      container_ports: List of container ports to allocate
+    """
+    self.P(f"Allocating host ports for {len(container_ports)} EXTRA_TUNNELS ports...", color='b')
+
+    for container_port in container_ports:
+      # Check if already allocated (shouldn't be, but safety check)
+      if container_port in self.extra_ports_mapping.values():
+        self.Pd(f"Port {container_port} already allocated, skipping")
+        continue
+
+      # Allocate dynamic host port
+      host_port = self._allocate_port(allow_dynamic=True)
+      self.extra_ports_mapping[host_port] = container_port
+
+      self.P(f"  Allocated host port {host_port} -> container port {container_port}", color='g')
+
+      # Special handling if this is the main PORT
+      if self.cfg_port == container_port and not self.port:
+        self.port = host_port
+        self.P(f"  Main PORT {container_port} mapped to host port {host_port}", color='g')
+
+    # Rebuild inverted_ports_mapping for Docker
+    self.inverted_ports_mapping = {
+      f"{container_port}/tcp": str(host_port)
+      for host_port, container_port in self.extra_ports_mapping.items()
+    }
+
+    self.P(f"Port allocation complete. Total ports: {len(self.extra_ports_mapping)}", color='g')
+    return
+
+  def _validate_extra_tunnels_config(self):
+    """
+    Validate EXTRA_TUNNELS configuration.
+
+    Key behaviors:
+    1. If TUNNEL_ENGINE_ENABLED=False, EXTRA_TUNNELS are IGNORED
+    2. Container ports can be defined only in EXTRA_TUNNELS (not in CONTAINER_RESOURCES)
+    3. Ports from EXTRA_TUNNELS will be allocated dynamically if needed
+    4. Dict keys can be strings or integers
+
+    Returns:
+      bool: True if valid
+    """
+    # Master switch check
+    if not self.cfg_tunnel_engine_enabled:
+      if self.cfg_extra_tunnels:
+        self.P(
+          f"TUNNEL_ENGINE_ENABLED=False: Ignoring {len(self.cfg_extra_tunnels)} EXTRA_TUNNELS",
+          color='y'
+        )
+      return True
+
+    if not self.cfg_extra_tunnels:
+      self.Pd("No EXTRA_TUNNELS configured")
+      return True
+
+    if not isinstance(self.cfg_extra_tunnels, dict):
+      raise ValueError("EXTRA_TUNNELS must be a dictionary {container_port: token}")
+
+    # Track which ports need to be allocated
+    ports_to_allocate = []
+
+    for port_key, tunnel_config in self.cfg_extra_tunnels.items():
+      # Convert port key to integer (handle both string and int keys)
+      try:
+        container_port = int(port_key)
+      except (ValueError, TypeError):
+        raise ValueError(f"EXTRA_TUNNELS key must be integer port, got: {port_key}")
+
+      # Check if port is already allocated
+      is_already_mapped = container_port in self.extra_ports_mapping.values()
+
+      if not is_already_mapped:
+        # Port not in CONTAINER_RESOURCES["ports"], will need to allocate
+        self.Pd(
+          f"EXTRA_TUNNELS port {container_port} not in CONTAINER_RESOURCES['ports'], "
+          f"will allocate dynamically"
+        )
+        ports_to_allocate.append(container_port)
+
+      # Normalize and validate tunnel config
+      try:
+        normalized = self._normalize_extra_tunnel_config(container_port, tunnel_config)
+        if not normalized:
+          raise ValueError(f"EXTRA_TUNNELS[{container_port}] token is empty")
+        self.extra_tunnel_configs[container_port] = normalized
+      except Exception as e:
+        raise ValueError(f"EXTRA_TUNNELS[{container_port}] validation failed: {e}")
+
+    # Allocate ports for EXTRA_TUNNELS not in CONTAINER_RESOURCES
+    if ports_to_allocate:
+      self._allocate_extra_tunnel_ports(ports_to_allocate)
+
+    self.P(f"EXTRA_TUNNELS validated: {len(self.extra_tunnel_configs)} tunnel(s) configured", color='g')
+    return True
+
+  ### END EXTRA TUNNELS METHODS ###

extensions/business/container_apps/container_utils.py

@@ -197,7 +197,7 @@ class DEEPLOY_PLUGIN_DATA:
 
 CONTAINER_APP_RUNNER_SIGNATURE = 'CONTAINER_APP_RUNNER'
 WORKER_APP_RUNNER_SIGNATURE = 'WORKER_APP_RUNNER'
-
+CONTAINERIZED_APPS_SIGNATURES = [CONTAINER_APP_RUNNER_SIGNATURE, WORKER_APP_RUNNER_SIGNATURE]
 
 class JOB_APP_TYPES:
   GENERIC = "generic"

extensions/business/deeploy/deeploy_const.py

@@ -208,58 +208,90 @@ def _process_pipeline_request(
 
       # Get nodes based on operation type
       discovered_plugin_instances = []
+      deployment_nodes = []
+      confirmation_nodes = []
+      deeploy_specs_for_update = None
       if is_create:
-        nodes = self._check_nodes_availability(inputs)
+        deployment_nodes = self._check_nodes_availability(inputs)
+        confirmation_nodes = list(deployment_nodes)
       else:
-        discovered_plugin_instances = self._discover_plugin_instances(app_id=app_id, job_id=job_id, owner=sender)
-
+        # Discover the live deployment so we can validate node affinity and reuse existing specs.
+        pipeline_context = self._gather_running_pipeline_context(
+          owner=sender,
+          app_id=app_id,
+          job_id=job_id,
+        )
+        discovered_plugin_instances = pipeline_context["discovered_instances"]
+        current_nodes = pipeline_context["nodes"]
+        deeploy_specs_for_update = pipeline_context["deeploy_specs"]
         self.P(f"Discovered plugin instances: {self.json_dumps(discovered_plugin_instances)}")
-        deeploy_specs_for_update = None
-        if job_app_type in (JOB_APP_TYPES.NATIVE, JOB_APP_TYPES.GENERIC, JOB_APP_TYPES.SERVICE):
-          discovered_plugin_instances = self._ensure_plugin_instance_ids(
-            inputs=inputs,
-            discovered_plugin_instances=discovered_plugin_instances,
-            owner=sender,
-            app_id=app_id,
-            job_id=job_id,
-          )
-          deeploy_specs_for_update = self._prepare_updated_deeploy_specs(
-            owner=sender,
-            app_id=app_id,
-            job_id=job_id,
-            discovered_plugin_instances=discovered_plugin_instances,
+
+        requested_nodes = inputs.get(DEEPLOY_KEYS.TARGET_NODES, None) or []
+        normalized_requested_nodes = [
+          self._check_and_maybe_convert_address(node) for node in requested_nodes
+        ] if requested_nodes else []
+
+        if normalized_requested_nodes:
+          # Reject updates that request a different node set than the one currently running.
+          if set(normalized_requested_nodes) != set(current_nodes):
+            msg = (
+              f"{DEEPLOY_ERRORS.NODES2}: Update request must target existing nodes {current_nodes}. "
+              f"Received {normalized_requested_nodes}."
+            )
+            raise ValueError(msg)
+
+        requested_nodes_count = inputs.get(DEEPLOY_KEYS.TARGET_NODES_COUNT, 0)
+        if requested_nodes_count and requested_nodes_count != len(current_nodes):
+          msg = (
+            f"{DEEPLOY_ERRORS.NODES2}: Update request must keep the original number of nodes "
+            f"({len(current_nodes)}). Received {requested_nodes_count}."
           )
-        nodes = [instance[DEEPLOY_PLUGIN_DATA.NODE] for instance in discovered_plugin_instances]
+          raise ValueError(msg)
 
-      if is_create:
-        dct_status, str_status = self.check_and_deploy_pipelines(
-          sender=sender,
-          inputs=inputs,
-          app_id=app_id,
-          app_alias=app_alias,
-          app_type=app_type,
-          new_nodes=nodes,
-          update_nodes=[],
-          discovered_plugin_instances=discovered_plugin_instances,
-          job_app_type=job_app_type,
-        )
-      else:
-        dct_status, str_status = self.check_and_deploy_pipelines(
-          sender=sender,
-          inputs=inputs,
+        inputs[DEEPLOY_KEYS.TARGET_NODES] = current_nodes
+        inputs.target_nodes = current_nodes
+        inputs[DEEPLOY_KEYS.TARGET_NODES_COUNT] = len(current_nodes)
+        inputs.target_nodes_count = len(current_nodes)
+
+        # TODO: Assess whether removing the running pipeline before redeploying is safe when the new launch fails.
+        self.delete_pipeline_from_nodes(
           app_id=app_id,
-          app_alias=app_alias,
-          app_type=app_type,
-          new_nodes=[],
-          update_nodes=nodes,
-          discovered_plugin_instances=discovered_plugin_instances,
-          dct_deeploy_specs=deeploy_specs_for_update,
-          job_app_type=job_app_type,
+          job_id=job_id,
+          owner=sender,
+          discovered_instances=discovered_plugin_instances,
         )
+
+        deployment_nodes = self._check_nodes_availability(inputs)
+        if set(deployment_nodes) != set(current_nodes):
+          msg = (
+            f"{DEEPLOY_ERRORS.NODES2}: Failed to validate that update runs on existing nodes. "
+            f"Expected {current_nodes}, validated {deployment_nodes}."
+          )
+          raise ValueError(msg)
+        confirmation_nodes = list(deployment_nodes)
+        discovered_plugin_instances = []
+
+      inputs[DEEPLOY_KEYS.TARGET_NODES] = deployment_nodes
+      inputs.target_nodes = deployment_nodes
+      inputs[DEEPLOY_KEYS.TARGET_NODES_COUNT] = len(deployment_nodes)
+      inputs.target_nodes_count = len(deployment_nodes)
+
+      dct_status, str_status = self.check_and_deploy_pipelines(
+        sender=sender,
+        inputs=inputs,
+        app_id=app_id,
+        app_alias=app_alias,
+        app_type=app_type,
+        new_nodes=deployment_nodes,
+        update_nodes=[],
+        discovered_plugin_instances=discovered_plugin_instances,
+        dct_deeploy_specs_create=deeploy_specs_for_update,
+        job_app_type=job_app_type,
+      )
 
-      if str_status in [DEEPLOY_STATUS.SUCCESS, DEEPLOY_STATUS.COMMAND_DELIVERED]:
-        if (dct_status is not None and is_confirmable_job and len(nodes) == len(dct_status)) or not is_confirmable_job:
-          eth_nodes = [self.bc.node_addr_to_eth_addr(node) for node in nodes]
+      if is_create and str_status in [DEEPLOY_STATUS.SUCCESS, DEEPLOY_STATUS.COMMAND_DELIVERED]:
+        if (dct_status is not None and is_confirmable_job and len(confirmation_nodes) == len(dct_status)) or not is_confirmable_job:
+          eth_nodes = [self.bc.node_addr_to_eth_addr(node) for node in confirmation_nodes]
           eth_nodes = sorted(eth_nodes)
           self.bc.submit_node_update(
             job_id=job_id,
@@ -483,6 +515,7 @@ def update_pipeline(
 
     Notes
     -----
+    - Existing pipelines are stopped and redeployed in place; requests must reference the active node set.
     - Updates are applied to existing plugin instances on the same nodes
     - For multi-plugin pipelines, all plugins are updated with new configurations
     - Resource validation applies the same as create operations