|
14 | 14 | ScanContext, |
15 | 15 | SEALOS_CPU_REQUEST_BY_LIMIT, |
16 | 16 | SEALOS_MEMORY_REQUEST_BY_LIMIT, |
| 17 | + TEMPLATE_DEPLOY_KEY, |
17 | 18 | Violation, |
18 | 19 | ) |
19 | 20 | from check_consistency_parser import find_line |
@@ -114,6 +115,62 @@ def check_pvc_storage_limit(context: ScanContext) -> List[Violation]: |
114 | 115 | return violations |
115 | 116 |
|
116 | 117 |
|
| 118 | +def check_statefulset_template_deploy_labels(context: ScanContext) -> List[Violation]: |
| 119 | + violations: List[Violation] = [] |
| 120 | + expected_value = "${{ defaults.app_name }}" |
| 121 | + |
| 122 | + for doc in context.yaml_documents: |
| 123 | + if doc.skip_checks or not isinstance(doc.data, dict): |
| 124 | + continue |
| 125 | + if doc.data.get("kind") != "StatefulSet": |
| 126 | + continue |
| 127 | + |
| 128 | + metadata = doc.data.get("metadata") |
| 129 | + labels = metadata.get("labels") if isinstance(metadata, dict) else None |
| 130 | + if not isinstance(labels, dict) or labels.get(TEMPLATE_DEPLOY_KEY) != expected_value: |
| 131 | + add_doc_violation( |
| 132 | + violations, |
| 133 | + rule_id="R041", |
| 134 | + doc=doc, |
| 135 | + pattern=rf"^\s*{re.escape(TEMPLATE_DEPLOY_KEY)}\s*:", |
| 136 | + default_pattern=r"^\s*labels\s*:", |
| 137 | + message=( |
| 138 | + f"StatefulSet metadata.labels.{TEMPLATE_DEPLOY_KEY} " |
| 139 | + f"must be {expected_value} for Template instance tracking" |
| 140 | + ), |
| 141 | + ) |
| 142 | + |
| 143 | + spec = doc.data.get("spec") |
| 144 | + volume_claim_templates = spec.get("volumeClaimTemplates") if isinstance(spec, dict) else None |
| 145 | + if not isinstance(volume_claim_templates, list): |
| 146 | + continue |
| 147 | + |
| 148 | + for volume_claim_template in volume_claim_templates: |
| 149 | + if not isinstance(volume_claim_template, dict): |
| 150 | + continue |
| 151 | + vct_metadata = volume_claim_template.get("metadata") |
| 152 | + vct_labels = vct_metadata.get("labels") if isinstance(vct_metadata, dict) else None |
| 153 | + if isinstance(vct_labels, dict) and vct_labels.get(TEMPLATE_DEPLOY_KEY) == expected_value: |
| 154 | + continue |
| 155 | + |
| 156 | + name = "<unknown>" |
| 157 | + if isinstance(vct_metadata, dict) and isinstance(vct_metadata.get("name"), str): |
| 158 | + name = vct_metadata["name"] |
| 159 | + add_doc_violation( |
| 160 | + violations, |
| 161 | + rule_id="R041", |
| 162 | + doc=doc, |
| 163 | + pattern=rf"^\s*{re.escape(TEMPLATE_DEPLOY_KEY)}\s*:", |
| 164 | + default_pattern=rf"^\s*name\s*:\s*{re.escape(name)}\s*$", |
| 165 | + message=( |
| 166 | + f"StatefulSet volumeClaimTemplates[{name}] metadata.labels.{TEMPLATE_DEPLOY_KEY} " |
| 167 | + f"must be {expected_value} so Template can track and clean PVCs" |
| 168 | + ), |
| 169 | + ) |
| 170 | + |
| 171 | + return violations |
| 172 | + |
| 173 | + |
117 | 174 | def _display_allowed(values: Dict[str, str]) -> str: |
118 | 175 | return "/".join(values.keys()) |
119 | 176 |
|
@@ -407,6 +464,7 @@ def check_database_cluster_visibility_labels(context: ScanContext) -> List[Viola |
407 | 464 | "R005": Rule("R005", check_no_emptydir), |
408 | 465 | "R006": Rule("R006", check_image_pull_policy), |
409 | 466 | "R011": Rule("R011", check_pvc_storage_limit), |
| 467 | + "R041": Rule("R041", check_statefulset_template_deploy_labels), |
410 | 468 | "R019": Rule("R019", check_database_cluster_component_resources), |
411 | 469 | "R040": Rule("R040", check_database_cluster_visibility_labels), |
412 | 470 | "R038": Rule("R038", check_managed_workload_resource_ladder), |
|
0 commit comments