Set up a repository for custom Zabbix templates and add the first template for Dell PowerVault ME storage systems. - Add conventions, version policy, and release documentation. - Add validation, index, packaging, export, and import-test scripts. - Add a disposable Zabbix test stack with version-aware imports. - Add PowerVault ME SNMP exports for Zabbix 7.0 and 8.0. - Cover system information, unit health, sensors, ports, and firmware revisions.
355 lines
10 KiB
Python
Executable File
355 lines
10 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Validate template package structure and manifest.yaml."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import re
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
TEMPLATES_DIR = ROOT / "templates"
|
|
|
|
CATEGORIES = {
|
|
"app",
|
|
"db",
|
|
"network",
|
|
"os",
|
|
"hardware",
|
|
"cloud",
|
|
"container",
|
|
"web",
|
|
"virtualization",
|
|
"power",
|
|
"printer",
|
|
"san",
|
|
"custom",
|
|
}
|
|
METHODS = {
|
|
"agent",
|
|
"agent2",
|
|
"snmp",
|
|
"http",
|
|
"ipmi",
|
|
"jmx",
|
|
"ssh",
|
|
"telnet",
|
|
"odbc",
|
|
"external",
|
|
"script",
|
|
"calculated",
|
|
"dependent",
|
|
"other",
|
|
}
|
|
SEMVER_RE = re.compile(r"^\d+\.\d+\.\d+$")
|
|
ZABBIX_VERSION_RE = re.compile(r"^\d+\.\d+$")
|
|
ID_RE = re.compile(r"^[a-z0-9_]+(?:/[a-z0-9_]+)+$")
|
|
REQUIRED_FIELDS = (
|
|
"id",
|
|
"name",
|
|
"description",
|
|
"version",
|
|
"category",
|
|
"zabbix",
|
|
"maintainer",
|
|
)
|
|
|
|
|
|
class Result:
|
|
def __init__(self) -> None:
|
|
self.errors: list[str] = []
|
|
self.warnings: list[str] = []
|
|
|
|
def error(self, path: Path | str, message: str) -> None:
|
|
self.errors.append(f"{display(path)}: {message}")
|
|
|
|
def warning(self, path: Path | str, message: str) -> None:
|
|
self.warnings.append(f"{display(path)}: {message}")
|
|
|
|
|
|
def display(path: Path | str) -> str:
|
|
path = Path(path)
|
|
try:
|
|
return path.relative_to(ROOT).as_posix()
|
|
except ValueError:
|
|
return str(path)
|
|
|
|
|
|
def load_yaml(path: Path, result: Result):
|
|
try:
|
|
with path.open(encoding="utf-8") as fh:
|
|
return yaml.safe_load(fh)
|
|
except yaml.YAMLError as exc:
|
|
result.error(path, f"Could not parse YAML: {exc}")
|
|
return None
|
|
|
|
|
|
def version_key(value: str) -> tuple[int, int]:
|
|
major, minor = value.split(".")
|
|
return int(major), int(minor)
|
|
|
|
|
|
def validate_exports(
|
|
package_dir: Path,
|
|
min_version: str,
|
|
primary_name: str,
|
|
result: Result,
|
|
) -> None:
|
|
version_dirs = sorted(
|
|
(
|
|
path
|
|
for path in package_dir.iterdir()
|
|
if path.is_dir() and ZABBIX_VERSION_RE.match(path.name)
|
|
),
|
|
key=lambda path: version_key(path.name),
|
|
)
|
|
|
|
if not version_dirs:
|
|
result.error(package_dir, "No version directory found (for example, 7.0/)")
|
|
return
|
|
|
|
if not any(path.name == min_version for path in version_dirs):
|
|
result.error(
|
|
package_dir,
|
|
f"Missing `{min_version}/` directory required by `zabbix.min`",
|
|
)
|
|
|
|
for version_dir in version_dirs:
|
|
if version_key(version_dir.name) < version_key(min_version):
|
|
result.warning(
|
|
version_dir,
|
|
f"`{version_dir.name}/` is older than `zabbix.min`",
|
|
)
|
|
|
|
yaml_files = sorted(version_dir.glob("*.yaml")) + sorted(
|
|
version_dir.glob("*.yml")
|
|
)
|
|
if not yaml_files:
|
|
result.warning(
|
|
version_dir,
|
|
"No exported YAML found",
|
|
)
|
|
continue
|
|
|
|
names: set[str] = set()
|
|
for yaml_path in yaml_files:
|
|
data = load_yaml(yaml_path, result)
|
|
if not isinstance(data, dict):
|
|
continue
|
|
|
|
export = data.get("zabbix_export")
|
|
if not isinstance(export, dict):
|
|
result.error(yaml_path, "Missing `zabbix_export`")
|
|
continue
|
|
|
|
export_version = export.get("version")
|
|
if str(export_version) != version_dir.name:
|
|
result.error(
|
|
yaml_path,
|
|
f"`zabbix_export.version` is `{export_version}` but the "
|
|
f"directory is `{version_dir.name}`",
|
|
)
|
|
|
|
if export.get("hosts"):
|
|
result.warning(
|
|
yaml_path,
|
|
"Hosts are included; export templates only",
|
|
)
|
|
|
|
templates = export.get("templates")
|
|
if not isinstance(templates, list) or not templates:
|
|
result.error(yaml_path, "Missing `zabbix_export.templates`")
|
|
continue
|
|
|
|
for template in templates:
|
|
if not isinstance(template, dict):
|
|
continue
|
|
for key in ("template", "name"):
|
|
value = template.get(key)
|
|
if value:
|
|
names.add(str(value))
|
|
|
|
if primary_name and primary_name not in names:
|
|
result.warning(
|
|
version_dir,
|
|
f"Template name `{primary_name}` from manifest.yaml was not "
|
|
"found in the export",
|
|
)
|
|
|
|
|
|
def validate_manifest(
|
|
path: Path,
|
|
templates_root: Path,
|
|
result: Result,
|
|
) -> tuple[str, str] | None:
|
|
data = load_yaml(path, result)
|
|
if not isinstance(data, dict):
|
|
result.error(path, "manifest.yaml must contain a mapping")
|
|
return None
|
|
|
|
rel = path.parent.relative_to(templates_root).as_posix()
|
|
|
|
for field in REQUIRED_FIELDS:
|
|
value = data.get(field)
|
|
if value is None or value == "":
|
|
result.error(path, f"Missing required field `{field}`")
|
|
|
|
template_id = str(data.get("id") or "")
|
|
if template_id:
|
|
if not ID_RE.match(template_id):
|
|
result.error(
|
|
path,
|
|
"`id` must use lowercase letters, digits, and `_`, "
|
|
"separated by `/`",
|
|
)
|
|
if template_id != rel:
|
|
result.error(
|
|
path,
|
|
f"`id` must match the directory path `{rel}`",
|
|
)
|
|
|
|
category = str(data.get("category") or "")
|
|
if category:
|
|
if category not in CATEGORIES:
|
|
result.error(path, f"Unknown category `{category}`")
|
|
if template_id and template_id.split("/")[0] != category:
|
|
result.error(path, "`category` must match the first segment of `id`")
|
|
|
|
version = data.get("version")
|
|
if version is not None and not SEMVER_RE.match(str(version)):
|
|
result.error(path, "`version` must use SemVer (for example, 1.0.0)")
|
|
|
|
method = data.get("method")
|
|
if method and str(method) not in METHODS:
|
|
result.error(path, f"Unknown `method`: {method}")
|
|
|
|
for filename in ("README.md", "CHANGELOG.md"):
|
|
if not (path.parent / filename).is_file():
|
|
result.error(path, f"{filename} is missing")
|
|
|
|
zabbix = data.get("zabbix")
|
|
min_version = None
|
|
if not isinstance(zabbix, dict):
|
|
result.error(path, "`zabbix` must be a mapping")
|
|
else:
|
|
min_version = zabbix.get("min")
|
|
if not min_version or not ZABBIX_VERSION_RE.match(str(min_version)):
|
|
result.error(path, "`zabbix.min` must use the `7.0` format")
|
|
min_version = None
|
|
else:
|
|
min_version = str(min_version)
|
|
|
|
max_version = zabbix.get("max")
|
|
if max_version is not None:
|
|
if not ZABBIX_VERSION_RE.match(str(max_version)):
|
|
result.error(
|
|
path,
|
|
"`zabbix.max` must use the `8.0` format or be null",
|
|
)
|
|
elif min_version and version_key(str(max_version)) < version_key(
|
|
min_version
|
|
):
|
|
result.error(path, "`zabbix.max` is lower than `zabbix.min`")
|
|
|
|
tested = zabbix.get("tested") or []
|
|
if not isinstance(tested, list):
|
|
result.error(path, "`zabbix.tested` must be a list")
|
|
else:
|
|
for item in tested:
|
|
if not ZABBIX_VERSION_RE.match(str(item)):
|
|
result.error(
|
|
path,
|
|
f"`zabbix.tested` entry `{item}` must use the `7.0` format",
|
|
)
|
|
|
|
depends_on = data.get("depends_on") or []
|
|
if not isinstance(depends_on, list):
|
|
result.error(path, "`depends_on` must be a list")
|
|
|
|
if min_version:
|
|
validate_exports(
|
|
path.parent,
|
|
min_version,
|
|
str(data.get("name") or ""),
|
|
result,
|
|
)
|
|
|
|
return template_id, str(data.get("name") or "")
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(
|
|
description="Validate template package structure and manifest.yaml."
|
|
)
|
|
parser.add_argument(
|
|
"path",
|
|
nargs="?",
|
|
default=str(TEMPLATES_DIR),
|
|
help="Directory to validate (default: templates/)",
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
root = Path(args.path).resolve()
|
|
if not root.exists():
|
|
print(f"ERROR Target directory does not exist: {root}", file=sys.stderr)
|
|
return 1
|
|
|
|
manifests = sorted(root.rglob("manifest.yaml"))
|
|
result = Result()
|
|
seen_ids: dict[str, Path] = {}
|
|
seen_names: dict[str, Path] = {}
|
|
|
|
for manifest in manifests:
|
|
info = validate_manifest(manifest, root, result)
|
|
if not info:
|
|
continue
|
|
template_id, name = info
|
|
if template_id:
|
|
if template_id in seen_ids:
|
|
result.error(
|
|
manifest,
|
|
f"Duplicate `id`: {template_id} "
|
|
f"({display(seen_ids[template_id])})",
|
|
)
|
|
else:
|
|
seen_ids[template_id] = manifest
|
|
if name:
|
|
if name in seen_names:
|
|
result.error(
|
|
manifest,
|
|
f"Duplicate template name: {name} "
|
|
f"({display(seen_names[name])})",
|
|
)
|
|
else:
|
|
seen_names[name] = manifest
|
|
|
|
for warning in result.warnings:
|
|
print(f"WARN {warning}")
|
|
for error in result.errors:
|
|
print(f"ERROR {error}")
|
|
|
|
if not manifests:
|
|
print("No template packages found.")
|
|
return 0
|
|
|
|
print()
|
|
if result.errors:
|
|
print(
|
|
f"Validation failed: {len(manifests)} package(s), "
|
|
f"{len(result.errors)} error(s), {len(result.warnings)} warning(s)"
|
|
)
|
|
return 1
|
|
|
|
print(
|
|
f"Validation passed: {len(manifests)} package(s), "
|
|
f"{len(result.warnings)} warning(s)"
|
|
)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|