120 lines
4 KiB
Python
Executable file
120 lines
4 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
"""
|
|
Pre-compile SNMP MIB files to JSON for runtime loading in Elixir.
|
|
Uses snmptranslate to extract OID mappings from vendor MIB files.
|
|
"""
|
|
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
from concurrent.futures import ProcessPoolExecutor, as_completed
|
|
from pathlib import Path
|
|
|
|
|
|
def compile_mib(mib_file: Path, vendor_dir: Path) -> tuple[str, int]:
|
|
"""Compile a single MIB file to JSON and return (output_path, symbol_count)."""
|
|
vendor = vendor_dir.name
|
|
mib = mib_file.name
|
|
output_dir = Path("priv/compiled_mibs") / vendor
|
|
output_dir.mkdir(parents=True, exist_ok=True)
|
|
output_file = output_dir / f"{mib.lower()}.json"
|
|
|
|
try:
|
|
# Get symbol list with 5 second timeout
|
|
result = subprocess.run(
|
|
["snmptranslate", f"-M+{vendor_dir}", "-m", mib, "-Ta"],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=5,
|
|
)
|
|
|
|
# Extract symbol names (lines with "OBJECT IDENTIFIER")
|
|
symbols = [
|
|
line.split()[0]
|
|
for line in result.stdout.splitlines()
|
|
if "OBJECT IDENTIFIER" in line
|
|
]
|
|
|
|
# Resolve each symbol to numeric OID
|
|
oid_map = {}
|
|
for symbol in symbols:
|
|
try:
|
|
oid_result = subprocess.run(
|
|
["snmptranslate", f"-M+{vendor_dir}", "-m", mib, "-On", f"{mib}::{symbol}"],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=2,
|
|
)
|
|
oid = oid_result.stdout.strip()
|
|
if oid and oid.startswith("."):
|
|
# Convert .1.3.6.1.4.1.41112 to [1, 3, 6, 1, 4, 1, 41112]
|
|
oid_list = [int(x) for x in oid[1:].split(".")]
|
|
oid_map[symbol] = oid_list
|
|
except (subprocess.TimeoutExpired, ValueError):
|
|
continue
|
|
|
|
# Write JSON if we got any symbols
|
|
if oid_map:
|
|
with open(output_file, "w") as f:
|
|
json.dump(oid_map, f, indent=2, sort_keys=True)
|
|
return f"{vendor}/{mib}", len(oid_map)
|
|
else:
|
|
return f"{vendor}/{mib}", 0
|
|
|
|
except subprocess.TimeoutExpired:
|
|
return f"{vendor}/{mib}", 0
|
|
except Exception as e:
|
|
print(f"Error compiling {vendor}/{mib}: {e}", file=sys.stderr)
|
|
return f"{vendor}/{mib}", 0
|
|
|
|
|
|
def main():
|
|
"""Compile all vendor MIB files in parallel."""
|
|
import multiprocessing
|
|
|
|
mibs_dir = Path("priv/mibs")
|
|
compiled_dir = Path("priv/compiled_mibs")
|
|
|
|
# Clean output directory
|
|
if compiled_dir.exists():
|
|
import shutil
|
|
shutil.rmtree(compiled_dir)
|
|
compiled_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# Find all MIB files
|
|
mib_files = []
|
|
for vendor_dir in sorted(mibs_dir.iterdir()):
|
|
if vendor_dir.is_dir():
|
|
for mib_file in vendor_dir.iterdir():
|
|
if mib_file.is_file():
|
|
mib_files.append((mib_file, vendor_dir))
|
|
|
|
# Use all available CPU cores for maximum parallelism
|
|
max_workers = multiprocessing.cpu_count()
|
|
print(f"Compiling {len(mib_files)} MIB files using {max_workers} parallel workers...")
|
|
|
|
# Compile in parallel
|
|
total_symbols = 0
|
|
successful_files = 0
|
|
|
|
with ProcessPoolExecutor(max_workers=max_workers) as executor:
|
|
futures = {
|
|
executor.submit(compile_mib, mib_file, vendor_dir): (mib_file, vendor_dir)
|
|
for mib_file, vendor_dir in mib_files
|
|
}
|
|
|
|
for future in as_completed(futures):
|
|
mib_path, symbol_count = future.result()
|
|
if symbol_count > 0:
|
|
print(f"✓ {mib_path}: {symbol_count} symbols")
|
|
total_symbols += symbol_count
|
|
successful_files += 1
|
|
|
|
print()
|
|
print(f"✓ MIB compilation complete. {successful_files} files with {total_symbols} total symbols compiled.")
|
|
print(f"✓ Files saved to {compiled_dir}/")
|
|
print("✓ These files are tracked in git and will be loaded at application startup.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|