diff --git a/extensions/business/deeploy/deeploy_manager_api.py b/extensions/business/deeploy/deeploy_manager_api.py index 0ee61d87..283747e9 100644 --- a/extensions/business/deeploy/deeploy_manager_api.py +++ b/extensions/business/deeploy/deeploy_manager_api.py @@ -269,6 +269,85 @@ def _redact_failed_request_payload(self, payload): """ return self._redact_per_node_config_for_log(payload) + def _gather_persisted_pipeline_update_context(self, owner, app_id, job_id, project_id=None): + pipeline_cid = self._get_pipeline_from_cstore(job_id) if job_id is not None else None + if not pipeline_cid: + raise ValueError( + f"{DEEPLOY_ERRORS.NODES3}: No running workers or persisted pipeline found for " + f"{f'app_id {app_id}' if app_id else f'job_id {job_id}'} and owner '{owner}'." + ) + + pipeline = self.get_pipeline_from_r1fs( + pipeline_cid, + timeout=30, + pin=False, + raise_on_error=False, + show_logs=False, + ) + if not isinstance(pipeline, dict): + raise ValueError(f"{DEEPLOY_ERRORS.NODES3}: Persisted pipeline '{pipeline_cid}' is unavailable.") + + pipeline_owner = pipeline.get(NetMonCt.OWNER.upper()) + if str(pipeline_owner).lower() != str(owner).lower(): + raise ValueError(f"{DEEPLOY_ERRORS.REQUEST3}. Persisted pipeline owner does not match request owner.") + + pipeline_app_id = pipeline.get("NAME") + if str(pipeline_app_id).lower() != str(app_id).lower(): + raise ValueError(f"{DEEPLOY_ERRORS.REQUEST3}. Persisted pipeline app_id does not match request app_id.") + + deeploy_specs = pipeline.get(NetMonCt.DEEPLOY_SPECS.upper()) or {} + if not isinstance(deeploy_specs, dict): + deeploy_specs = {} + + if str(deeploy_specs.get(DEEPLOY_KEYS.JOB_ID)).lower() != str(job_id).lower(): + raise ValueError(f"{DEEPLOY_ERRORS.REQUEST3}. Persisted pipeline job_id does not match request job_id.") + if project_id is not None and str(deeploy_specs.get(DEEPLOY_KEYS.PROJECT_ID)).lower() != str(project_id).lower(): + raise ValueError(f"{DEEPLOY_ERRORS.REQUEST3}. Persisted pipeline project_id does not match request project_id.") + + nodes = deeploy_specs.get(DEEPLOY_KEYS.CURRENT_TARGET_NODES) or [] + if not isinstance(nodes, list): + nodes = [] + + discovered_instances = [] + plugins = pipeline.get(NetMonCt.PLUGINS.upper()) or [] + if not isinstance(plugins, list): + plugins = [] + + for plugin in plugins: + if not isinstance(plugin, dict): + continue + signature = plugin.get(ct.CONFIG_PLUGIN.K_SIGNATURE) or plugin.get("signature") + if not signature: + continue + instances = plugin.get(ct.CONFIG_PLUGIN.K_INSTANCES, []) + if not isinstance(instances, list): + continue + for instance in instances: + if not isinstance(instance, dict): + continue + instance_id = instance.get(ct.CONFIG_INSTANCE.K_INSTANCE_ID) or instance.get(DEEPLOY_KEYS.PLUGIN_INSTANCE_ID) + discovered_instances.append({ + DEEPLOY_PLUGIN_DATA.APP_ID: pipeline_app_id, + DEEPLOY_PLUGIN_DATA.INSTANCE_ID: instance_id, + DEEPLOY_PLUGIN_DATA.PLUGIN_SIGNATURE: signature, + DEEPLOY_PLUGIN_DATA.PLUGIN_INSTANCE: { + "instance_conf": self.deepcopy(instance), + }, + DEEPLOY_PLUGIN_DATA.NODE: nodes[0] if nodes else None, + DEEPLOY_PLUGIN_DATA.CHAINSTORE_RESPONSE_KEY: instance.get(ct.BIZ_PLUGIN_DATA.CHAINSTORE_RESPONSE_KEY), + }) + + if not discovered_instances: + raise ValueError(f"{DEEPLOY_ERRORS.NODES3}: Persisted pipeline '{pipeline_cid}' has no plugin instances.") + + return { + "discovered_instances": discovered_instances, + "nodes": nodes, + "discovered_nodes": [], + "deeploy_specs": deeploy_specs, + "from_persisted_pipeline": True, + } + def __handle_error(self, exc, request, extra_error_code=DEEPLOY_ERRORS.GENERIC): """ Handle the error and return a response. @@ -658,6 +737,7 @@ def _process_pipeline_request( prepared_create_deploy_plan = None skip_create_response_key_reset = False previous_pipeline_cid = None + update_context_from_persisted_pipeline = False if is_create: is_valid = self.deeploy_check_payment_and_job_owner(inputs, auth_result[DEEPLOY_KEYS.ESCROW_OWNER], is_create=is_create, debug=self.cfg_deeploy_verbose > 1) if not is_valid: @@ -677,11 +757,22 @@ def _process_pipeline_request( nodes_changed = True else: # Discover the live deployment so we can validate node affinity and reuse existing specs. - pipeline_context = self._gather_running_pipeline_context( - owner=auth_result[DEEPLOY_KEYS.ESCROW_OWNER], - app_id=app_id, - job_id=job_id, - ) + try: + pipeline_context = self._gather_running_pipeline_context( + owner=auth_result[DEEPLOY_KEYS.ESCROW_OWNER], + app_id=app_id, + job_id=job_id, + ) + except ValueError as exc: + if DEEPLOY_ERRORS.NODES3 not in str(exc): + raise + pipeline_context = self._gather_persisted_pipeline_update_context( + owner=auth_result[DEEPLOY_KEYS.ESCROW_OWNER], + app_id=app_id, + job_id=job_id, + project_id=inputs.get(DEEPLOY_KEYS.PROJECT_ID, None), + ) + update_context_from_persisted_pipeline = True discovered_plugin_instances = pipeline_context["discovered_instances"] current_nodes = pipeline_context["nodes"] deeploy_specs_for_update = pipeline_context["deeploy_specs"] @@ -850,12 +941,20 @@ def _process_pipeline_request( skip_create_response_key_reset = True # All validations and response-key resets passed; remove the running job and redeploy. - self.delete_pipeline_from_nodes( - app_id=app_id, - job_id=job_id, - owner=auth_result[DEEPLOY_KEYS.ESCROW_OWNER], - discovered_instances=discovered_plugin_instances, - ) + if update_context_from_persisted_pipeline: + # TODO: stop stale offline old-node pipelines through ChainDist reconciliation when they return. + self.Pd( + f"Skipping live pipeline stop for offline update fallback on job_id={job_id}; " + "new pipeline will be deployed to validated target nodes.", + color='y', + ) + else: + self.delete_pipeline_from_nodes( + app_id=app_id, + job_id=job_id, + owner=auth_result[DEEPLOY_KEYS.ESCROW_OWNER], + discovered_instances=discovered_plugin_instances, + ) deployment_nodes = list(validated_nodes) confirmation_nodes = list(validated_nodes) diff --git a/extensions/business/deeploy/tests/test_update_requests.py b/extensions/business/deeploy/tests/test_update_requests.py index 100ef05a..36b7861a 100644 --- a/extensions/business/deeploy/tests/test_update_requests.py +++ b/extensions/business/deeploy/tests/test_update_requests.py @@ -32,6 +32,7 @@ def decorator(fn): from extensions.business.deeploy.deeploy_const import ( DEEPLOY_DYNAMIC_ENV_KEYS, DEEPLOY_DYNAMIC_ENV_TYPES, + DEEPLOY_ERRORS, DEEPLOY_KEYS, DEEPLOY_PLUGIN_DATA, DEEPLOY_STATUS, @@ -80,7 +81,11 @@ def _make_process_update_plugin(self, discovered_instances, nodes=None, deeploy_ plugin._ensure_plugin_instance_ids = lambda *args, **kwargs: None plugin._check_nodes_availability = lambda inputs: nodes or ["node-1"] - called = {"delete": 0, "deploy": 0, "deploy_kwargs": None, "queued": 0} + called = {"delete": 0, "deploy": 0, "deploy_kwargs": None, "queued": 0, "bc_update": 0} + plugin.bc = types.SimpleNamespace( + node_addr_to_eth_addr=lambda node: node, + submit_node_update=lambda **kwargs: called.__setitem__("bc_update", called["bc_update"] + 1), + ) plugin.delete_pipeline_from_nodes = lambda **kwargs: called.__setitem__("delete", called["delete"] + 1) def check_and_deploy_pipelines(**kwargs): @@ -985,6 +990,80 @@ def check_nodes_availability(inputs): self.assertEqual(called["deploy"], 1) self.assertEqual([context for context, _, _ in validation_calls], ["payment", "nodes"]) + def test_process_update_uses_persisted_pipeline_when_all_old_nodes_are_offline(self): + plugin, called = self._make_process_update_plugin( + discovered_instances=[], + nodes=["old-node-1", "old-node-2"], + deeploy_specs={ + DEEPLOY_KEYS.JOB_ID: 11, + DEEPLOY_KEYS.PROJECT_ID: "0xProject", + DEEPLOY_KEYS.CURRENT_TARGET_NODES: ["old-node-1", "old-node-2"], + DEEPLOY_KEYS.JOB_APP_TYPE: "generic", + }, + ) + plugin._gather_running_pipeline_context = lambda **kwargs: (_ for _ in ()).throw( + ValueError(f"{DEEPLOY_ERRORS.NODES3}: No running workers found") + ) + plugin._get_pipeline_from_cstore = lambda job_id: "cid-old-pipeline" + plugin.get_pipeline_from_r1fs = lambda *args, **kwargs: { + "NAME": "app-123", + "OWNER": "0xOwner", + "DEEPLOY_SPECS": { + DEEPLOY_KEYS.JOB_ID: 11, + DEEPLOY_KEYS.PROJECT_ID: "0xProject", + DEEPLOY_KEYS.CURRENT_TARGET_NODES: ["old-node-1", "old-node-2"], + DEEPLOY_KEYS.JOB_APP_TYPE: "generic", + }, + "PLUGINS": [ + { + plugin.ct.CONFIG_PLUGIN.K_SIGNATURE: "CONTAINER_APP_RUNNER", + plugin.ct.CONFIG_PLUGIN.K_INSTANCES: [ + { + plugin.ct.CONFIG_INSTANCE.K_INSTANCE_ID: "current-instance", + DEEPLOY_KEYS.PLUGIN_NAME: "worker", + "IMAGE": "repo/app:1.0", + "CONTAINER_RESOURCES": {"cpu": 1, "memory": "256m"}, + }, + ], + }, + ], + } + plugin._check_nodes_availability = lambda inputs: ["new-node-1"] + + response = plugin._process_pipeline_request( + { + DEEPLOY_KEYS.APP_ID: "app-123", + DEEPLOY_KEYS.APP_ALIAS: "app", + DEEPLOY_KEYS.JOB_ID: 11, + DEEPLOY_KEYS.PROJECT_ID: "0xProject", + DEEPLOY_KEYS.JOB_APP_TYPE: "generic", + DEEPLOY_KEYS.PIPELINE_INPUT_TYPE: "void", + DEEPLOY_KEYS.CHAINSTORE_RESPONSE: False, + DEEPLOY_KEYS.TARGET_NODES: ["new-node-1"], + DEEPLOY_KEYS.TARGET_NODES_COUNT: 1, + DEEPLOY_KEYS.PLUGINS: [ + { + DEEPLOY_KEYS.PLUGIN_SIGNATURE: "CONTAINER_APP_RUNNER", + DEEPLOY_KEYS.PLUGIN_INSTANCE_ID: "current-instance", + DEEPLOY_KEYS.PLUGIN_NAME: "worker", + "IMAGE": "repo/app:2.0", + "CONTAINER_RESOURCES": {"cpu": 1, "memory": "256m"}, + }, + ], + }, + is_create=False, + async_mode=True, + ) + + self.assertEqual(response[DEEPLOY_KEYS.STATUS], DEEPLOY_STATUS.COMMAND_DELIVERED) + self.assertEqual(called["delete"], 0) + self.assertEqual(called["deploy"], 1) + self.assertEqual(called["queued"], 1) + self.assertEqual(called["bc_update"], 1) + self.assertEqual(called["deploy_kwargs"]["new_nodes"], ["new-node-1"]) + redeploy_plugins = called["deploy_kwargs"]["inputs"][DEEPLOY_KEYS.PLUGINS] + self.assertEqual(redeploy_plugins[0]["IMAGE"], "repo/app:2.0") + def test_process_update_rejects_job_app_type_change_before_payment_or_delete(self): plugin, called = self._make_process_update_plugin( discovered_instances=[ diff --git a/ver.py b/ver.py index aa20a35b..18ac0b5b 100644 --- a/ver.py +++ b/ver.py @@ -1 +1 @@ -__VER__ = '2.10.380' +__VER__ = '2.10.390'