fix: cache file race condition (#2872)

Author: xiaoj655

apiserver/paasng/paasng/platform/agent_sandbox/management/commands/cleanup_expired_agent_sandboxes.py

@@ -17,7 +17,6 @@
 
 """清理过期的沙箱实例"""
 
-import threading
 import uuid
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from dataclasses import dataclass
@@ -26,6 +25,8 @@
 from django.db import close_old_connections
 from django.utils import timezone
 
+from paas_wl.infras.resources.base.base import get_client_by_cluster_name
+from paas_wl.infras.resources.base.kube_client import CoreDynamicClient
 from paasng.platform.agent_sandbox.constants import SandboxStatus
 from paasng.platform.agent_sandbox.exceptions import SandboxError
 from paasng.platform.agent_sandbox.models import Sandbox
@@ -50,6 +51,24 @@ class _DeleteResult:
     error: SandboxError | None = None
 
 
+def _stdout_log_line(message: str) -> str:
+    ts = timezone.localtime(timezone.now()).strftime("%Y-%m-%d %H:%M:%S")
+    return f"[{ts}] {message}"
+
+
+def _warmup_kube_discovery_cache(cluster_names: set[str]) -> None:
+    """
+    预热（初始化）kubernetes dynamic client 的 discovery 磁盘缓存。
+    避免并发初始化缓存时同时读写该文件。
+    Ref: https://github.com/kubernetes-client/python/issues/2037
+    """
+    for target in sorted(cluster_names):
+        if not target:
+            continue
+        client = get_client_by_cluster_name(target)
+        CoreDynamicClient(client)
+
+
 def _delete_expired_sandbox(sandbox_uuid: uuid.UUID) -> _DeleteResult:
     close_old_connections()
     sandbox = Sandbox.objects.select_related("application").get(uuid=sandbox_uuid)
@@ -99,31 +118,38 @@ def handle(self, dry_run, concurrency, *args, **options):
             return
 
         sandbox_uuids = list(sandboxes.values_list("uuid", flat=True))
+        targets = set(sandboxes.values_list("target", flat=True).distinct())
         deleted_count = 0
         failed_count = 0
-        output_lock = threading.Lock()
         workers = min(concurrency, len(sandbox_uuids))
 
+        if targets:
+            self.stdout.write(f"预热 {len(targets)} 个集群的 K8s discovery 缓存: {', '.join(sorted(targets))}")
+            _warmup_kube_discovery_cache(targets)
+
         self.stdout.write(f"并发数: {workers}")
 
         with ThreadPoolExecutor(max_workers=workers) as executor:
             futures = [executor.submit(_delete_expired_sandbox, sbx_uuid) for sbx_uuid in sandbox_uuids]
             for future in as_completed(futures):
                 result = future.result()
-                with output_lock:
-                    if result.success:
-                        deleted_count += 1
-                        self.stdout.write(
+                if result.success:
+                    deleted_count += 1
+                    self.stdout.write(
+                        _stdout_log_line(
                             f"已删除 {deleted_count}/{total} uuid={result.sandbox.uuid} "
                             f"name={result.sandbox.name} app_code={result.sandbox.application.code}"
                         )
-                    else:
-                        failed_count += 1
-                        self.stdout.write(
-                            self.style.ERROR(
+                    )
+                else:
+                    failed_count += 1
+                    self.stdout.write(
+                        self.style.ERROR(
+                            _stdout_log_line(
                                 f"删除失败 uuid={result.sandbox.uuid} name={result.sandbox.name} "
                                 f"app_code={result.sandbox.application.code}: {result.error}"
                             )
                         )
+                    )
 
         self.stdout.write(f"\n清理完成: 删除 {deleted_count} 个, 失败 {failed_count} 个")