R1 475 deeploy api fixes (#290)

* fix: update the update pipeline logic to remove and re-create the pipeline

* fix: add comments

* chore: inc version

Author: toderian

extensions/business/deeploy/deeploy_manager_api.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

extensions/business/deeploy/deeploy_mixin.py

@@ -110,7 +110,7 @@ def __check_is_oracle(self, inputs):
       raise ValueError("Sender {} is not an oracle".format(sender))
     return True
 
-  def __create_pipeline_on_nodes(self, nodes, inputs, app_id, app_alias, app_type, sender, job_app_type=None):
+  def __create_pipeline_on_nodes(self, nodes, inputs, app_id, app_alias, app_type, sender, job_app_type=None, dct_deeploy_specs=None):
     """
     Create new pipelines on each node and set CSTORE `response_key` for the "callback" action
     """
@@ -124,18 +124,34 @@ def __create_pipeline_on_nodes(self, nodes, inputs, app_id, app_alias, app_type,
     response_keys = self.defaultdict(list)
 
     ts = self.time()
-    dct_deeploy_specs = {
-      DEEPLOY_KEYS.JOB_ID: job_id,
-      DEEPLOY_KEYS.PROJECT_ID: project_id,
-      DEEPLOY_KEYS.PROJECT_NAME: project_name,
-      DEEPLOY_KEYS.NR_TARGET_NODES: len(nodes),
-      DEEPLOY_KEYS.CURRENT_TARGET_NODES: nodes,
-      DEEPLOY_KEYS.JOB_TAGS: job_tags,
-      DEEPLOY_KEYS.DATE_CREATED: ts,
-      DEEPLOY_KEYS.DATE_UPDATED: ts,
-      DEEPLOY_KEYS.SPARE_NODES: spare_nodes,
-      DEEPLOY_KEYS.ALLOW_REPLICATION_IN_THE_WILD: allow_replication_in_the_wild,
-    }
+    if dct_deeploy_specs:
+      dct_deeploy_specs = self.deepcopy(dct_deeploy_specs)
+      dct_deeploy_specs[DEEPLOY_KEYS.DATE_UPDATED] = ts
+      if DEEPLOY_KEYS.DATE_CREATED not in dct_deeploy_specs:
+        dct_deeploy_specs[DEEPLOY_KEYS.DATE_CREATED] = ts
+    else:
+      dct_deeploy_specs = {
+        DEEPLOY_KEYS.NR_TARGET_NODES: len(nodes),
+        DEEPLOY_KEYS.CURRENT_TARGET_NODES: nodes,
+        DEEPLOY_KEYS.JOB_TAGS: job_tags,
+        DEEPLOY_KEYS.DATE_CREATED: ts,
+        DEEPLOY_KEYS.DATE_UPDATED: ts,
+        DEEPLOY_KEYS.SPARE_NODES: spare_nodes,
+        DEEPLOY_KEYS.ALLOW_REPLICATION_IN_THE_WILD: allow_replication_in_the_wild,
+      }
+
+    if job_id is not None or DEEPLOY_KEYS.JOB_ID not in dct_deeploy_specs:
+      dct_deeploy_specs[DEEPLOY_KEYS.JOB_ID] = job_id
+    if project_id is not None or DEEPLOY_KEYS.PROJECT_ID not in dct_deeploy_specs:
+      dct_deeploy_specs[DEEPLOY_KEYS.PROJECT_ID] = project_id
+    if project_name is not None or DEEPLOY_KEYS.PROJECT_NAME not in dct_deeploy_specs:
+      dct_deeploy_specs[DEEPLOY_KEYS.PROJECT_NAME] = project_name
+    dct_deeploy_specs[DEEPLOY_KEYS.NR_TARGET_NODES] = len(nodes)
+    dct_deeploy_specs[DEEPLOY_KEYS.CURRENT_TARGET_NODES] = nodes
+    dct_deeploy_specs[DEEPLOY_KEYS.JOB_TAGS] = job_tags
+    dct_deeploy_specs[DEEPLOY_KEYS.SPARE_NODES] = spare_nodes
+    dct_deeploy_specs[DEEPLOY_KEYS.ALLOW_REPLICATION_IN_THE_WILD] = allow_replication_in_the_wild
+
     detected_job_app_type = job_app_type or self.deeploy_detect_job_app_type(plugins)
     if detected_job_app_type in JOB_APP_TYPES_ALL:
       dct_deeploy_specs[DEEPLOY_KEYS.JOB_APP_TYPE] = detected_job_app_type
@@ -419,6 +435,46 @@ def _prepare_updated_deeploy_specs(self, owner, app_id, job_id, discovered_plugi
     refreshed_specs[DEEPLOY_KEYS.DATE_UPDATED] = self.time()
     return refreshed_specs
 
+  def _gather_running_pipeline_context(self, owner, app_id=None, job_id=None):
+    """
+    Collect information about currently running pipeline instances for a job/app.
+
+    Ensures follow-up operations keep parity with the active deployment state.
+
+    Returns
+    -------
+    dict
+      {
+        'discovered_instances': list,
+        'nodes': list[str],
+        'deeploy_specs': dict | None,
+      }
+    """
+    discovered_instances = self._discover_plugin_instances(app_id=app_id, job_id=job_id, owner=owner)
+    nodes = []
+    for instance in discovered_instances:
+      node_addr = instance.get(DEEPLOY_PLUGIN_DATA.NODE)
+      if node_addr and node_addr not in nodes:
+        nodes.append(node_addr)
+
+    if not nodes:
+      msg = f"{DEEPLOY_ERRORS.NODES3}: No running workers found for provided "
+      msg += f"{f'app_id {app_id}' if app_id else f'job_id {job_id}'} and owner '{owner}'."
+      raise ValueError(msg)
+
+    deeploy_specs = self._prepare_updated_deeploy_specs(
+      owner=owner,
+      app_id=app_id,
+      job_id=job_id,
+      discovered_plugin_instances=discovered_instances,
+    )
+
+    return {
+      "discovered_instances": discovered_instances,
+      "nodes": nodes,
+      "deeploy_specs": deeploy_specs,
+    }
+
   def __prepare_plugins_for_update(self, inputs, discovered_plugin_instances):
     """
     Prepare plugins for update using discovered instances instead of creating new ones
@@ -1333,7 +1389,7 @@ def deeploy_prepare_plugins(self, inputs):
     plugins = [plugin]
     return plugins
 
-  def check_and_deploy_pipelines(self, sender, inputs, app_id, app_alias, app_type, update_nodes, new_nodes, discovered_plugin_instances=[], dct_deeploy_specs=None, job_app_type=None):
+  def check_and_deploy_pipelines(self, sender, inputs, app_id, app_alias, app_type, update_nodes, new_nodes, discovered_plugin_instances=[], dct_deeploy_specs=None, job_app_type=None, dct_deeploy_specs_create=None):
     """
     Validate the inputs and deploy the pipeline on the target nodes.
     """
@@ -1349,7 +1405,7 @@ def check_and_deploy_pipelines(self, sender, inputs, app_id, app_alias, app_type
       update_response_keys = self.__update_pipeline_on_nodes(update_nodes, inputs, app_id, app_alias, app_type, sender, discovered_plugin_instances, dct_deeploy_specs, job_app_type=job_app_type)
       response_keys.update(update_response_keys)
     if len(new_nodes) > 0:
-      new_response_keys = self.__create_pipeline_on_nodes(new_nodes, inputs, app_id, app_alias, app_type, sender, job_app_type=job_app_type)
+      new_response_keys = self.__create_pipeline_on_nodes(new_nodes, inputs, app_id, app_alias, app_type, sender, job_app_type=job_app_type, dct_deeploy_specs=dct_deeploy_specs_create)
       response_keys.update(new_response_keys)
 
     # Phase 3: Wait until all the responses are received via CSTORE and compose status response
@@ -1873,8 +1929,9 @@ def check_running_pipelines_and_add_to_r1fs(self):
 
     return netmon_job_ids
 
-  def delete_pipeline_from_nodes(self, app_id=None, job_id=None, owner=None, allow_missing=False):
-    discovered_instances = self._discover_plugin_instances(app_id=app_id, job_id=job_id, owner=owner)
+  def delete_pipeline_from_nodes(self, app_id=None, job_id=None, owner=None, allow_missing=False, discovered_instances=None):
+    if discovered_instances is None:
+      discovered_instances = self._discover_plugin_instances(app_id=app_id, job_id=job_id, owner=owner)
 
     if len(discovered_instances) == 0:
       if allow_missing:

ver.py

@@ -1 +1 @@
-__VER__ = '2.9.813'
+__VER__ = '2.9.814'