feat: export retained snapshot image references for cleanup

console/services/cleanup_inventory.py

@@ -148,3 +148,56 @@ def deployment_resource(row, region):
                                  row.get("app_upgrade_record__old_version") or "—",
                                  row.get("app_upgrade_record__version") or "—")}]
     return result
+
+
+def snapshot_reference_resource(row, region):
+    """Export retained snapshot image references, never backup config or secrets.
+
+    A missing build image, plugin relation or custom Kubernetes override prevents
+    this source from claiming complete coverage; known images still protect data.
+    """
+    result = _base("snapshot-reference:{}:{}".format(region, row["ID"]), region, "templates",
+                   "application_snapshot", "保留快照 / {}".format(row["ID"]), "")
+    result["source"] = "platform_snapshot_references"
+    images = set()
+    complete = True
+    try:
+        snapshot = json.loads(row.get("snapshot") or "")
+        if not isinstance(snapshot, dict) or not isinstance(snapshot.get("components"), list):
+            raise ValueError()
+        group = snapshot.get("component_group") or {}
+        if not isinstance(group, dict):
+            raise ValueError()
+        name = _display_name(group.get("group_name"), "保留快照")
+        version = _display_name(group.get("group_version"), str(row["ID"]))
+        result["name"] = "{} / {}".format(name, version)
+        for component in snapshot["components"]:
+            if not isinstance(component, dict) or not isinstance(component.get("service_base"), dict):
+                raise ValueError()
+            base = component["service_base"]
+            if not isinstance(base.get("service_id"), str) or not base["service_id"]:
+                raise ValueError()
+            source = component.get("service_source") or {}
+            if not isinstance(source, dict):
+                raise ValueError()
+            found = False
+            for value in (base.get("image"), source.get("image")):
+                if value in (None, ""):
+                    continue
+                image = _image(value)
+                if image:
+                    images.add(image)
+                    found = True
+                else:
+                    complete = False
+            if not found and base.get("service_source") != "third_party":
+                complete = False
+            if component.get("service_plugin_relation") or component.get("component_k8s_attributes"):
+                complete = False
+    except (ValueError, TypeError):
+        complete = False
+    result["images"] = sorted(images)
+    result["observed"] = complete
+    result["protection"] = "referenced" if complete else "reference_unknown"
+    result["usageStatus"] = "referenced" if images else "unknown"
+    return result

console/tests/cleanup_snapshot_references_test.py

@@ -0,0 +1,38 @@
+import json
+import unittest
+from console.services.cleanup_inventory import snapshot_reference_resource
+
+
+class SnapshotReferenceTests(unittest.TestCase):
+    def test_projects_images_without_snapshot_configuration(self):
+        row = {'ID': 7, 'snapshot': json.dumps({
+            'component_group': {'group_name': '订单系统', 'group_version': 'v1'},
+            'components': [{'service_base': {'service_id': 's', 'image': 'goodrain.me/app:v1'},
+                            'service_source': {'image': 'goodrain.me/app:v0'},
+                            'service_env_vars': [{'value': 'do-not-export'}],
+                            'service_auths': [{'value': 'do-not-export'}]}]})}
+        result = snapshot_reference_resource(row, 'r')
+        self.assertEqual(result['images'], ['goodrain.me/app:v0', 'goodrain.me/app:v1'])
+        self.assertEqual(result['name'], '订单系统 / v1')
+        self.assertTrue(result['observed'])
+        self.assertEqual(result['protection'], 'referenced')
+        self.assertNotIn('do-not-export', json.dumps(result))
+        self.assertNotIn('retirement', result)
+
+    def test_unknown_code_build_or_sidecar_references_are_incomplete(self):
+        for component in [
+                {'service_base': {'service_id': 's', 'image': ''}},
+                {'service_base': {'service_id': 's', 'image': 'goodrain.me/app:v1'},
+                 'service_plugin_relation': [{'plugin_id': 'p'}]},
+                {'service_base': {'service_id': 's', 'image': 'https://do-not-export@example.invalid'}},
+                {'service_base': {'service_id': 's', 'image': 'goodrain.me/app:v1'},
+                 'component_k8s_attributes': [{'name': 'custom'}]}]:
+            result = snapshot_reference_resource({'ID': 7, 'snapshot': json.dumps({'components': [component]})}, 'r')
+            self.assertFalse(result['observed'])
+            self.assertEqual(result['protection'], 'reference_unknown')
+            self.assertNotIn('do-not-export', json.dumps(result))
+
+    def test_malformed_snapshot_does_not_become_empty_complete_inventory(self):
+        for raw in ['invalid', 'null', '{}', '{"components":[null]}']:
+            result = snapshot_reference_resource({'ID': 7, 'snapshot': raw}, 'r')
+            self.assertFalse(result['observed'])

console/views/cleanup_inventory.py

@@ -11,10 +11,11 @@
 from console.services.cleanup_retirement import template_retirement_targets, snapshot_protected_components, RetirementConflict
 from console.services.cleanup_gateway import CleanupGatewayUnavailable
 from console.services.cleanup_installation import resolve_gateway_key
-from console.models.main import AppVersionTemplateRelation, RainbondCenterApp, RainbondCenterAppVersion, ServiceUpgradeRecord
+from console.models.main import (AppVersionTemplateRelation, RainbondCenterApp, RainbondCenterAppVersion,
+                                 ServiceUpgradeRecord, AppUpgradeSnapshot)
 from console.repositories.region_repo import region_repo
 from console.services.cleanup_inventory import (template_resource, verify_source_request, version_resources,
-                                                deployment_resource, failed_scope_label)
+                                                deployment_resource, failed_scope_label, snapshot_reference_resource)
 from www.apiclient.regionapi import RegionInvokeApi
 from www.models.main import ServiceGroup, ServiceGroupRelation, Tenants, TenantServiceInfo
 
@@ -51,7 +52,7 @@ def get(self, request: Request, enterprise_id: str, region_name: str) -> Respons
         try:
             cursor = int(request.query_params.get("cursor", "0"))
             upper = int(request.query_params.get("upper", "0"))
-            if cursor < 0 or upper < 0 or kind not in ("templates", "versions", "deployments"):
+            if cursor < 0 or upper < 0 or kind not in ("templates", "versions", "deployments", "snapshots"):
                 raise ValueError()
         except ValueError:
             return Response({"errorCode": "INVALID_REQUEST"}, status=400)
@@ -68,6 +69,16 @@ def get(self, request: Request, enterprise_id: str, region_name: str) -> Respons
             template_fields = ("ID", "app_id", "version", "share_team", "app_template", "enterprise_id", "region_name",
                                "source", "is_complete", "cleanup_activation_revision")
             page = list(template_query.filter(ID__gt=cursor, ID__lte=upper).order_by("ID").values(*template_fields)[:26])
+        elif kind == "snapshots":
+            # Include the enterprise's retained snapshots across regions. A
+            # cross-region image match protects content; it never grants delete.
+            snapshot_query = AppUpgradeSnapshot.objects.filter(tenant_id__in=teams.values_list("tenant_id", flat=True))
+            if upper == 0:
+                upper = snapshot_query.order_by("-ID").values_list("ID", flat=True).first() or 0
+            if cursor > upper:
+                return Response({"errorCode": "INVALID_REQUEST"}, status=400)
+            page = [dict(row) for row in snapshot_query.filter(
+                ID__gt=cursor, ID__lte=upper).order_by("ID").values("ID", "snapshot")[:26]]
         elif kind == "deployments":
             deployment_query = ServiceUpgradeRecord.objects.filter(
                 app_upgrade_record__tenant_id__in=teams.values_list("tenant_id", flat=True),
@@ -101,7 +112,7 @@ def get(self, request: Request, enterprise_id: str, region_name: str) -> Respons
                 app_id__in=[row["app_id"] for row in page]).values_list("app_id", "app_name"))
             for row in page:
                 row["app_name"] = template_names.get(row["app_id"], "")
-        else:
+        elif kind != "snapshots":
             team_ids = {team["tenant_id"]: team["tenant_alias"] or team["tenant_name"] for team in team_rows}
             relations = list(ServiceGroupRelation.objects.filter(
                 tenant_id__in=team_ids, region_name=region_name,
@@ -132,6 +143,9 @@ def get(self, request: Request, enterprise_id: str, region_name: str) -> Respons
                 row["retirement"] = targets.get(row["ID"])
             resources = [template_resource(row, region_name, row["app_id"] in hidden) for row in page]
             failures.extend(failed_scope_label(row) for row in resources if not row["observed"])
+        elif kind == "snapshots":
+            resources = [snapshot_reference_resource(row, region_name) for row in page]
+            failures.extend(failed_scope_label(row) for row in resources if not row["observed"])
         elif kind == "deployments":
             resources = [deployment_resource(row, region_name) for row in page]
         else:

scripts/check_cleanup_retirement.py

@@ -98,6 +98,28 @@ def test_invalid_snapshot_preserves_inventory_but_blocks_retirement(self):
             with self.assertRaises(RetirementConflict):
                 retire_template('e', 'r', self.expected, self.key)
 
+        def test_snapshot_reference_page_is_scoped_and_sanitized(self):
+            import json
+            from types import SimpleNamespace
+            from unittest.mock import patch
+            from console.views.cleanup_inventory import CleanupInventoryView
+            Tenants.objects.create(tenant_id='other', tenant_name='other', namespace='other', enterprise_id='other')
+            for team, image in [('team', 'goodrain.me/owned:v1'), ('other', 'goodrain.me/foreign:v1')]:
+                AppUpgradeSnapshot.objects.create(tenant_id=team, snapshot_id=team, snapshot=json.dumps({
+                    'components': [{'service_base': {'service_id': team, 'image': image},
+                                    'service_auths': [{'value': 'do-not-export'}]}]}))
+            request = SimpleNamespace(method='GET', headers={}, query_params={'kind': 'snapshots'},
+                                      get_full_path=lambda: '/inventory')
+            with patch('console.views.cleanup_inventory.resolve_gateway_key', return_value=self.key), \
+                    patch('console.views.cleanup_inventory.verify_source_request', return_value=True), \
+                    patch('console.views.cleanup_inventory.region_repo.get_enterprise_region_by_region_name', return_value=True):
+                response = CleanupInventoryView().get(request, 'e', 'r')
+            self.assertEqual(response.status_code, 200)
+            self.assertEqual(len(response.data['resources']), 1)
+            self.assertEqual(response.data['resources'][0]['images'], ['goodrain.me/owned:v1'])
+            self.assertNotIn('do-not-export', json.dumps(response.data))
+            self.assertFalse(response.data['referencesComplete'])
+
         def test_foreign_region_is_protected(self):
             with self.assertRaises(RetirementConflict):
                 retire_template('e', 'other', self.expected, self.key)