feat: svc-redis 僵尸实例问题与分配逻辑优化(part3) (#2756)

Author: xiaoj655

apiserver/paasng/paasng/accessories/servicehub/remote/client.py

@@ -18,6 +18,7 @@
 """Client for remote services"""
 
 import logging
+import time
 from contextlib import contextmanager
 from dataclasses import MISSING, dataclass
 from typing import Dict, List
@@ -41,6 +42,8 @@ class RemoteSvcConfig:
     provision_params_tmpl: Dict
     jwt_auth_conf: Dict
     prefer_async_delete: bool = True
+    # 是否使用幂等创建实例接口
+    prefer_idempotent_provision: bool = False
     is_ready: bool = True
 
     @classmethod
@@ -86,6 +89,7 @@ def __post_init__(self):
         self.retrieve_instance_by_name_url = urljoin(self.endpoint_url, "services/{service_id}/instances/?name={name}")
         self.update_inst_config_url = urljoin(self.endpoint_url, "instances/{instance_id}/config/")
         self.create_instance_url = urljoin(self.endpoint_url, "services/{service_id}/instances/{instance_id}/")
+        self.idem_prov = urljoin(self.endpoint_url, "services/{service_id}/instances/idem_prov/")
         self.delete_instance_url = urljoin(self.endpoint_url, "instances/{instance_id}/")
         self.async_delete_instance_url = urljoin(self.endpoint_url, "instances/{instance_id}/async_delete")
         # 增强服务绑定
@@ -132,6 +136,25 @@ def validate_resp(resp: requests.Response):
                 response_text=resp.text,
             )
 
+    def _post_until_ready(self, url: str, payload: Dict) -> requests.Response:
+        """Retry POST requests until the async endpoint returns 200/201 or timeout."""
+        deadline = time.monotonic() + self.REQUEST_CREATE_TIMEOUT
+
+        poll_interval = 0.5
+        while time.monotonic() < deadline:
+            resp = requests.post(url, json=payload, auth=self.auth, timeout=self.REQUEST_CREATE_TIMEOUT)
+            self.validate_resp(resp)
+            # 200，201 表示资源已就绪，可以返回了
+            # 202，表示请求已接受，但资源未就绪，需要继续轮询
+            if resp.status_code in {200, 201}:
+                return resp
+            time.sleep(poll_interval)
+            poll_interval = min(poll_interval * 2, 4)
+
+        raise RemoteClientError(
+            f"POST to {desensitize_url(url)} not ready after retrying for {self.REQUEST_CREATE_TIMEOUT} seconds"
+        )
+
     def get_meta_info(self) -> Dict:
         """Get service's meta info
 
@@ -214,6 +237,19 @@ def provision_instance(self, service_id: str, plan_id: str, instance_id: str, pa
             self.validate_resp(resp)
             return resp.json()
 
+    def idempotent_provision_instance(self, service_id: str, plan_id: str, params: Dict) -> Dict:
+        """Idempotently provision a new instance,
+        `params['engine_app_name']` treated as the idempotency key, so it's required
+
+        :raises: RemoteClientError
+        :return: <instance dict>
+        """
+        url = self.config.idem_prov.format(service_id=service_id)
+        payload = {"plan_id": plan_id, "params": params}
+        with wrap_request_exc(self):
+            resp = self._post_until_ready(url, payload)
+            return resp.json()
+
     def retrieve_instance(self, instance_id: str) -> Dict:
         """Retrieve a provisioned instance info
 
@@ -291,8 +327,7 @@ def update_instance_config(self, instance_id: str, config: Dict):
     def create_client_side_instance(self, service_id: str, instance_id: str, params: Dict):
         url = self.config.create_client_side_instance_url.format(service_id=service_id, instance_id=instance_id)
         with wrap_request_exc(self):
-            resp = requests.post(url, json=params, auth=self.auth, timeout=self.REQUEST_CREATE_TIMEOUT)
-            self.validate_resp(resp)
+            resp = self._post_until_ready(url, params)
             return resp.json()
 
     def destroy_client_side_instance(self, instance_id: str):

apiserver/paasng/paasng/accessories/servicehub/remote/manager.py

@@ -110,6 +110,7 @@ def semantic_version_gte(self, version: str) -> bool:
 DEFAULT_META_INFO = MetaInfo(version=None)
 VERSION_WITH_INST_CONFIG = "0.1.0"
 VERSION_WITH_REST_UPSERT = "0.2.0"
+VERSION_WITH_IDEMPOTENT_PROVISION = "2.0.4"
 
 
 @dataclass
@@ -179,6 +180,10 @@ def supports_rest_upsert(self) -> bool:
         """Check if current service supports Feature: RestFul upsert Service/Plan"""
         return self.meta_info.semantic_version_gte(VERSION_WITH_REST_UPSERT)
 
+    def supports_idempotent_provision(self) -> bool:
+        """Check if current service supports idempotent provision, which means provisioning an already provisioned instance will not cause error"""
+        return self.meta_info.semantic_version_gte(VERSION_WITH_IDEMPOTENT_PROVISION)
+
 
 @dataclass
 class EnvClusterInfo:
@@ -237,12 +242,18 @@ def provision(self):
             logger.warning(f"remote service {self.get_service().name} is not ready, skip")
             return
 
-        instance_id = str(uuid.uuid4())
         try:
             params = self.render_params(self.remote_config.provision_params_tmpl)
-            self.remote_client.provision_instance(
-                str(self.db_obj.service_id), str(self.db_obj.plan_id), instance_id, params=params
-            )
+            if self.get_service().supports_idempotent_provision():
+                resp = self.remote_client.idempotent_provision_instance(
+                    str(self.db_obj.service_id), str(self.db_obj.plan_id), params=params
+                )
+                instance_id = resp["uuid"]
+            else:
+                instance_id = str(uuid.uuid4())
+                self.remote_client.provision_instance(
+                    str(self.db_obj.service_id), str(self.db_obj.plan_id), instance_id, params=params
+                )
         except Exception as e:
             logger.exception(f"Error provisioning new instance for {self.db_application.name}")
             raise exceptions.ProvisionInstanceError(

apiserver/paasng/tests/paasng/accessories/servicehub/remote/test_manager.py

@@ -150,6 +150,48 @@ def test_provision(
                     assert bool(all(mocked_provision.call_args[0]))
                     assert mocked_provision.call_args[1]["params"]["username"] == rel.db_engine_app.name
 
+    @mock.patch("paas_wl.workloads.networking.egress.shim.get_cluster_egress_info")
+    @mock.patch("paasng.accessories.servicehub.remote.client.RemoteServiceClient.provision_instance")
+    @mock.patch("paasng.accessories.servicehub.remote.client.RemoteServiceClient.idempotent_provision_instance")
+    def test_provision_with_idempotent_api(
+        self,
+        mocked_idempotent_provision,
+        mocked_provision,
+        get_cluster_egress_info,
+        config,
+        store,
+        bk_module,
+        bk_service,
+        bk_plan_1,
+    ):
+        """Test service instance provision through idempotent API"""
+        get_cluster_egress_info.return_value = {"egress_ips": ["1.1.1.1"], "digest_version": "foo"}
+        mocked_idempotent_provision.return_value = {"uuid": str(uuid.uuid4())}
+        config.prefer_idempotent_provision = True
+
+        plans = [bk_plan_1]
+        mgr = RemoteServiceMgr(store=store)
+        bk_service.plans = plans
+        bk_service.meta_info = MetaInfo(version="2.0.4")
+
+        SvcBindingPolicyManager(bk_service, DEFAULT_TENANT_ID).set_uniform(plans=[plans[0].uuid])
+        mgr.bind_service(bk_service, bk_module)
+
+        with mock.patch.object(mgr, "get") as get_service:
+            get_service.return_value = bk_service
+            env = bk_module.get_envs("stag")
+            rel = next(mgr.list_unprovisioned_rels(env.engine_app))
+            rel.provision()
+
+            assert rel.is_provisioned() is True
+            assert str(rel.db_obj.service_instance_id) == mocked_idempotent_provision.return_value["uuid"]
+            mocked_idempotent_provision.assert_called_once_with(
+                str(rel.db_obj.service_id),
+                str(rel.db_obj.plan_id),
+                params={"username": rel.db_engine_app.name},
+            )
+            mocked_provision.assert_not_called()
+
     @mock.patch("paasng.accessories.servicehub.remote.manager.EnvClusterInfo.get_egress_info")
     def test_render_params(
         self,