| name | YAML验证器 |
| description | 当验证YAML文件、检查Kubernetes清单、调试Docker Compose或验证配置文件时,在部署前验证YAML。 |
| license | MIT |
YAML验证器技能
概述
YAML是DevOps中的核心技术 - Kubernetes、Docker Compose、GitHub Actions都使用YAML。无效的YAML会以神秘的错误消息静默破坏部署。
核心原则: 无效的YAML = 部署失败。在部署前验证。
何时使用
始终:
- 在部署Kubernetes清单之前
- 验证Docker Compose文件
- 检查持续集成/持续部署工作流文件
- 在提交配置文件之前
- 验证应用程序配置
- 检查基础设施即代码文件
触发短语:
- "验证这个YAML文件"
- "检查Kubernetes配置"
- "这个Docker Compose文件有问题吗?"
- "YAML语法错误"
- "部署失败,检查配置"
- "配置文件验证"
YAML验证功能
语法检查
- 缩进验证
- 引号检查
- 特殊字符处理
- 注释格式验证
- 文档分隔符检查
结构验证
- 必需字段检查
- 数据类型验证
- 嵌套结构验证
- 引用完整性检查
- 模式匹配验证
语义分析
- 业务逻辑验证
- 依赖关系检查
- 配置一致性验证
- 最佳实践检查
- 安全性审查
常见YAML错误
缩进错误
问题:
YAML使用空格缩进,不能使用制表符
错误示例:
services:
web:
image: nginx:latest ← 使用了制表符
database:
image: postgres:13
解决方案:
- 使用2个或4个空格进行缩进
- 统一缩进风格
- 配置编辑器显示空白字符
引号问题
问题:
特殊字符需要引号包围
错误示例:
password: my#password ← #会被当作注释
url: https://example.com/api/v1/resource?id=123&key=value
解决方案:
password: "my#password"
url: "https://example.com/api/v1/resource?id=123&key=value"
数据类型混淆
问题:
YAML会自动推断数据类型
错误示例:
port: "8080" ← 字符串而不是数字
enabled: "true" ← 字符串而不是布尔值
timeout: 30s ← 无效的时间格式
解决方案:
port: 8080
enabled: true
timeout: 30
代码实现示例
YAML验证器
import yaml
import json
import re
from typing import Dict, List, Any, Optional, Union
from dataclasses import dataclass
from enum import Enum
from pathlib import Path
import jsonschema
class ValidationLevel(Enum):
ERROR = "error"
WARNING = "warning"
INFO = "info"
@dataclass
class ValidationResult:
"""验证结果"""
level: ValidationLevel
message: str
line: Optional[int] = None
column: Optional[int] = None
suggestion: Optional[str] = None
@dataclass
class YAMLDocument:
"""YAML文档"""
content: Dict[str, Any]
line_map: Dict[int, str]
metadata: Dict[str, Any]
class YAMLValidator:
"""YAML验证器"""
():
.validation_rules = ._initialize_validation_rules()
.kubernetes_schemas = ._load_kubernetes_schemas()
.docker_compose_schema = ._load_docker_compose_schema()
() -> [ValidationResult]:
:
(file_path, , encoding=) f:
content = f.read()
lines = content.split()
Exception e:
[ValidationResult(
level=ValidationLevel.ERROR,
message=
)]
.validate_yaml_content(content, lines)
() -> [ValidationResult]:
results = []
syntax_results = ._validate_syntax(content, lines)
results.extend(syntax_results)
(r.level == ValidationLevel.ERROR r syntax_results):
results
:
yaml_data = yaml.safe_load(content)
structure_results = ._validate_structure(yaml_data, lines)
results.extend(structure_results)
semantic_results = ._validate_semantics(yaml_data, lines)
results.extend(semantic_results)
framework_results = ._validate_framework_specific(yaml_data, lines)
results.extend(framework_results)
yaml.YAMLError e:
results.append(ValidationResult(
level=ValidationLevel.ERROR,
message=,
line=(e, , ).line + (e, ) ,
column=(e, , ).column + (e, )
))
results
() -> [ValidationResult]:
results = []
lines:
lines = content.split()
line_num, line (lines, ):
line:
results.append(ValidationResult(
level=ValidationLevel.ERROR,
message=,
line=line_num,
column=line.find() + ,
suggestion=
))
line.strip():
leading_spaces = (line) - (line.lstrip())
leading_spaces % != :
results.append(ValidationResult(
level=ValidationLevel.WARNING,
message=,
line=line_num,
column=,
suggestion=
))
line line.strip().startswith():
parts = line.split(, )
(parts) == :
value = parts[].strip()
value (value.startswith() value.startswith()):
re.search(, value):
results.append(ValidationResult(
level=ValidationLevel.WARNING,
message=,
line=line_num,
column=(parts[]) + ,
suggestion=
))
results
() -> [ValidationResult]:
results = []
(yaml_data, ):
required_fields = .validation_rules.get(, [])
field required_fields:
field yaml_data:
results.append(ValidationResult(
level=ValidationLevel.ERROR,
message=,
suggestion=
))
type_rules = .validation_rules.get(, {})
._check_data_types(yaml_data, type_rules, , results)
structure_rules = .validation_rules.get(, {})
._check_structure(yaml_data, structure_rules, , results)
results
():
(data, ):
key, value data.items():
current_path = path key
key type_rules:
expected_type = type_rules[key]
._is_valid_type(value, expected_type):
results.append(ValidationResult(
level=ValidationLevel.ERROR,
message=,
suggestion=
))
(value, (, )):
._check_data_types(value, type_rules, current_path, results)
(data, ):
i, item (data):
current_path =
(item, (, )):
._check_data_types(item, type_rules, current_path, results)
() -> :
type_mapping = {
: ,
: ,
: (, ),
: ,
: ,
:
}
expected_python_type = type_mapping.get(expected_type)
expected_python_type:
(value, expected_python_type)
():
(data, ):
key, value data.items():
current_path = path key
key structure_rules:
rule = structure_rules[key]
(rule, ):
rule:
field rule[]:
field value:
results.append(ValidationResult(
level=ValidationLevel.ERROR,
message=,
suggestion=
))
rule (value, ):
(value) > rule[]:
results.append(ValidationResult(
level=ValidationLevel.WARNING,
message=,
suggestion=
))
(value, (, )):
._check_structure(value, structure_rules, current_path, results)
() -> [ValidationResult]:
results = []
semantic_rules = .validation_rules.get(, {})
._check_port_ranges(yaml_data, results)
._check_resource_limits(yaml_data, results)
._check_environment_variables(yaml_data, results)
._check_secrets(yaml_data, results)
results
():
(data, ):
key, value data.items():
current_path = path key
key.lower() (value, ):
( <= value <= ):
results.append(ValidationResult(
level=ValidationLevel.ERROR,
message=,
suggestion=
))
._check_port_ranges(value, results, current_path)
(data, ):
i, item (data):
current_path =
._check_port_ranges(item, results, current_path)
():
(data, ):
key, value data.items():
current_path = path key
key [, ] (value, ):
re.(, value):
results.append(ValidationResult(
level=ValidationLevel.WARNING,
message=,
suggestion=
))
key [] (value, ):
re.(, value):
results.append(ValidationResult(
level=ValidationLevel.WARNING,
message=,
suggestion=
))
._check_resource_limits(value, results, current_path)
(data, ):
i, item (data):
current_path =
._check_resource_limits(item, results, current_path)
():
(data, ):
key, value data.items():
current_path = path key
key.lower() [, , ] (value, ):
env_key, env_value value.items():
(env_value, ) env_value.startswith() env_value.endswith():
var_name = env_value[:-]
re.(, var_name):
results.append(ValidationResult(
level=ValidationLevel.WARNING,
message=,
suggestion=
))
._check_environment_variables(value, results, current_path)
(data, ):
i, item (data):
current_path =
._check_environment_variables(item, results, current_path)
():
sensitive_keys = [, , , , , ]
(data, ):
key, value data.items():
current_path = path key
(sensitive key.lower() sensitive sensitive_keys):
(value, ) (value) > :
results.append(ValidationResult(
level=ValidationLevel.WARNING,
message=,
suggestion=
))
._check_secrets(value, results, current_path)
(data, ):
i, item (data):
current_path =
._check_secrets(item, results, current_path)
() -> [ValidationResult]:
results = []
framework_type = ._detect_framework_type(yaml_data)
framework_type == :
results.extend(._validate_kubernetes(yaml_data))
framework_type == :
results.extend(._validate_docker_compose(yaml_data))
framework_type == :
results.extend(._validate_github_actions(yaml_data))
results
() -> :
(yaml_data, ):
keys = yaml_data.keys()
keys keys:
keys keys yaml_data.get(, {}):
keys keys:
() -> [ValidationResult]:
results = []
required_fields = [, , ]
field required_fields:
field yaml_data:
results.append(ValidationResult(
level=ValidationLevel.ERROR,
message=,
suggestion=
))
yaml_data (yaml_data[], ):
spec = yaml_data[]
spec spec.get(, {}):
spec spec[]:
template_labels = spec[][].get(, {})
selector_labels = spec[][]
key, value selector_labels.items():
template_labels.get(key) != value:
results.append(ValidationResult(
level=ValidationLevel.ERROR,
message=,
suggestion=
))
results
() -> [ValidationResult]:
results = []
yaml_data:
version = yaml_data[]
(version, ) version.startswith() version.startswith():
results.append(ValidationResult(
level=ValidationLevel.WARNING,
message=,
suggestion=
))
yaml_data (yaml_data[], ):
service_name, service_config yaml_data[].items():
(service_config, ):
service_config:
results.append(ValidationResult(
level=ValidationLevel.ERROR,
message=,
suggestion=
))
service_config (service_config[], ):
port service_config[]:
(port, ) port:
host_port, container_port = port.split(, )
:
host_port_num = (host_port)
( <= host_port_num <= ):
results.append(ValidationResult(
level=ValidationLevel.ERROR,
message=,
suggestion=
))
ValueError:
results.append(ValidationResult(
level=ValidationLevel.ERROR,
message=,
suggestion=
))
results
() -> [ValidationResult]:
results = []
required_fields = [, ]
field required_fields:
field yaml_data:
results.append(ValidationResult(
level=ValidationLevel.ERROR,
message=,
suggestion=
))
yaml_data (yaml_data[], ):
job_name, job_config yaml_data[].items():
(job_config, ):
job_config:
results.append(ValidationResult(
level=ValidationLevel.ERROR,
message=,
suggestion=
))
job_config:
results.append(ValidationResult(
level=ValidationLevel.ERROR,
message=,
suggestion=
))
(job_config[], ):
i, step (job_config[]):
(step, ) step step:
results.append(ValidationResult(
level=ValidationLevel.ERROR,
message=,
suggestion=
))
results
() -> [, ]:
{
: [],
: {},
: {},
: {}
}
() -> [, ]:
{}
() -> [, ]:
{}
() -> :
results:
report = []
by_level = {
ValidationLevel.ERROR: [],
ValidationLevel.WARNING: [],
ValidationLevel.INFO: []
}
result results:
by_level[result.level].append(result)
level [ValidationLevel.ERROR, ValidationLevel.WARNING, ValidationLevel.INFO]:
by_level[level]:
level_name = level.value.upper()
report.append()
result by_level[level]:
location =
result.line:
location =
result.column:
location +=
location +=
report.append()
result.suggestion:
report.append()
report.append()
error_count = (by_level[ValidationLevel.ERROR])
warning_count = (by_level[ValidationLevel.WARNING])
info_count = (by_level[ValidationLevel.INFO])
summary =
error_count > :
summary +=
warning_count > :
summary +=
:
summary +=
report.append(summary)
.join(report)
():
validator = YAMLValidator()
yaml_content =
results = validator.validate_yaml_content(yaml_content)
report = validator.generate_validation_report(results)
(report)
__name__ == :
main()
YAML格式化工具
import yaml
from typing import Dict, Any, Optional
from pathlib import Path
class YAMLFormatter:
"""YAML格式化工具"""
def __init__(self):
self.indent_size = 2
self.sort_keys = False
self.default_flow_style = False
def format_yaml_file(self, file_path: str, output_path: Optional[str] = None) -> bool:
"""格式化YAML文件"""
try:
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
data = yaml.safe_load(content)
formatted_content = yaml.dump(
data,
indent=self.indent_size,
sort_keys=self.sort_keys,
default_flow_style=self.default_flow_style,
allow_unicode=True,
encoding=None
)
output_file = output_path or file_path
(output_file, , encoding=) f:
f.write(formatted_content)
Exception e:
()
() -> :
:
(yaml_file, , encoding=) f:
data = yaml.safe_load(f)
(json_file, , encoding=) f:
json.dump(data, f, indent=, ensure_ascii=)
Exception e:
()
() -> :
:
(json_file, , encoding=) f:
data = json.load(f)
(yaml_file, , encoding=) f:
yaml.dump(data, f, indent=.indent_size, allow_unicode=)
Exception e:
()
():
formatter = YAMLFormatter()
success = formatter.format_yaml_file()
success:
()
formatter.yaml_to_json(, )
formatter.json_to_yaml(, )
__name__ == :
main()
YAML最佳实践
文件组织
- 逻辑分组: 相关配置放在一起
- 注释说明: 添加必要的注释
- 命名规范: 使用一致的命名风格
- 模块化: 将大文件拆分为小文件
安全考虑
- 敏感信息: 使用环境变量或密钥管理
- 权限控制: 限制文件访问权限
- 加密存储: 对敏感配置进行加密
- 审计日志: 记录配置变更
版本控制
- Git管理: 使用Git跟踪配置变更
- 分支策略: 合理使用分支管理
- 代码审查: 配置变更需要审查
- 自动化测试: 自动验证配置正确性
相关技能
- kubernetes-analyzer - Kubernetes配置分析
- docker-analyzer - Docker配置分析
- config-validator - 配置验证
- infrastructure-as-code - 基础设施即代码