本文目录导读：
我来详细说明如何用Python脚本生成SaltStack配置。
基础配置模板
#!/usr/bin/env python3
"""
生成SaltStack配置文件的Python脚本
"""
import os
import json
import yaml
from typing import Dict, List, Any
class SaltConfigGenerator:
def __init__(self, output_dir: str = "./salt_config"):
self.output_dir = output_dir
self.config = {}
def generate_master_config(self, master_ip: str = "127.0.0.1") -> Dict:
"""生成Salt Master配置"""
config = {
"interface": master_ip,
"publish_port": 4505,
"ret_port": 4506,
"auto_accept": True,
"file_roots": {
"base": ["/srv/salt"]
},
"pillar_roots": {
"base": ["/srv/pillar"]
},
"log_level": "info",
"log_file": "/var/log/salt/master",
"worker_threads": 5,
"max_open_files": 100000,
"timeout": 5,
"loop_interval": 60,
"output": "nested",
"color": True,
"sock_dir": "/var/run/salt/master",
"pidfile": "/var/run/salt-master.pid",
"root_dir": "/",
"conf_file": "/etc/salt/master",
"cachedir": "/var/cache/salt/master",
"keep_jobs": 24,
"gather_job_timeout": 10,
"syndic_wait": 5,
"preserve_minion_cache": False,
"win_repo_cachefile": "salt://win/repo/winrepo.p",
"ext_pillar": [],
"ext_job_cache": "",
"master_stats": False,
"master_job_cache": "local_cache",
"enforce_mine_cache": False,
"mine_enabled": True,
"mine_return_job": False,
"mine_interval": 60,
"cli_summary": False,
"max_minions": 0,
"consecutive_errors": 0,
"rotate_aes_key": True,
"req_server_niceness": None,
"pub_server_niceness": None,
"preserve_minion_cache": False
}
return config
def generate_minion_config(self,
minion_id: str,
master_ip: str,
environment: str = "production") -> Dict:
"""生成Salt Minion配置"""
config = {
"master": master_ip,
"master_port": 4506,
"id": minion_id,
"environment": environment,
"log_level": "info",
"log_file": f"/var/log/salt/minion-{minion_id}",
"log_level_logfile": "info",
"log_datefmt": "%H:%M:%S",
"log_fmt_console": "[%(levelname)-8s] %(message)s",
"log_fmt_logfile": "%(asctime)s,%(msecs)03d [%(name)-17s:%(lineno)-4d][%(levelname)-8s][%(process)d] %(message)s",
"log_granular_levels": {},
"file_client": "remote",
"use_master_when_local": False,
"state_verbose": True,
"state_output": "full",
"state_aggregate": False,
"acceptance_wait_time": 10,
"acceptance_wait_time_max": 0,
"rejected_retry": False,
"random_reauth_delay": 60,
"reauth_delay": 60,
"loop_interval": 1,
"verify_env": True,
"cache_jobs": False,
"grains": {
"roles": [],
"environment": environment,
"deployment": "standard"
},
"pillar_opts": True,
"minion_sign_messages": False,
"zmq_monitor": False,
"master_type": "str",
"master_uri": "",
"local": False,
"ipc_mode": "ipc",
"tcp_pub_port": 4510,
"tcp_pull_port": 4511,
"tcp_master_pub_port": 4512,
"tcp_master_pull_port": 4513
}
return config
def generate_pillar_config(self, pillar_data: Dict) -> Dict:
"""生成Pillar配置文件"""
return {
"pillar": {
"base": pillar_data
}
}
def save_yaml_config(self, config: Dict, filename: str):
"""保存为YAML格式"""
filepath = os.path.join(self.output_dir, filename)
os.makedirs(os.path.dirname(filepath), exist_ok=True)
with open(filepath, 'w') as f:
yaml.dump(config, f, default_flow_style=False, indent=2)
print(f"配置已保存: {filepath}")
def save_json_config(self, config: Dict, filename: str):
"""保存为JSON格式"""
filepath = os.path.join(self.output_dir, filename)
os.makedirs(os.path.dirname(filepath), exist_ok=True)
with open(filepath, 'w') as f:
json.dump(config, f, indent=2)
print(f"配置已保存: {filepath}")
def generate_top_file(self, environments: List[str]):
"""生成top.sls文件"""
top_config = {
"base": {
"*": ["core", "security"],
"web*": ["nginx", "php", "node"],
"db*": ["mysql", "postgresql"],
"cache*": ["redis", "memcached"]
}
}
if environments:
for env in environments:
if env != "base":
top_config[env] = {"'*'": ["default"]}
return top_config
def generate_state_file(self, state_name: str, state_data: Dict):
"""生成状态文件"""
return {state_name: state_data}
def generate_schedule_config(self, jobs: List[Dict]) -> Dict:
"""生成计划任务配置"""
schedule = {}
for job in jobs:
schedule[job['name']] = {
"function": job.get('function', 'state.apply'),
"args": job.get('args', []),
"kwargs": job.get('kwargs', {}),
"seconds": job.get('interval', 3600),
"splay": job.get('splay', 10),
"run_on_start": job.get('run_on_start', False),
"returner": job.get('returner', '')
}
return {"schedule": schedule}
高级配置生成器
#!/usr/bin/env python3
"""
高级配置生成器，支持模板化和多环境
"""
import os
import sys
import jinja2
import yaml
from typing import Dict, List, Optional, Any
from dataclasses import dataclass, asdict
from pathlib import Path
@dataclass
class SaltConfigTemplate:
"""配置模板数据类"""
master_ip: str = "127.0.0.1"
minion_id: str = "minion01"
environment: str = "production"
roles: List[str] = None
log_level: str = "info"
def __post_init__(self):
if self.roles is None:
self.roles = ["base"]
class AdvancedConfigGenerator:
def __init__(self, template_dir: str = "templates"):
self.template_dir = template_dir
self.jinja_env = jinja2.Environment(
loader=jinja2.FileSystemLoader(template_dir),
keep_trailing_newline=True
)
def render_template(self, template_name: str, context: Dict) -> str:
"""渲染Jinja2模板"""
template = self.jinja_env.get_template(template_name)
return template.render(context)
def generate_master_config_template(self) -> str:
"""生成Master配置模板"""
return """# Salt Master Configuration
# Generated by Python Config Generator
# Date: {{ generation_date }}
interface: {{ master_ip }}
publish_port: {{ publish_port }}
ret_port: {{ ret_port }}
# File System
file_roots:
base:
- /srv/salt
{{ environment }}:
- /srv/salt/{{ environment }}
pillar_roots:
base:
- /srv/pillar
{{ environment }}:
- /srv/pillar/{{ environment }}
# Logging
log_level: {{ log_level }}
log_file: /var/log/salt/master
# Performance
worker_threads: {{ worker_threads }}
max_open_files: {{ max_open_files }}
# Security
auto_accept: {{ auto_accept }}
{% if auto_sign_master_sign_pubkey %}
autosign_file: /etc/salt/autosign.conf
{% endif %}
# External Pillar
{% if external_pillars %}
ext_pillar:
{% for pillar in external_pillars %}
- {{ pillar.type }}: {{ pillar.config }}
{% endfor %}
{% endif %}
# Reactor
{% if reactor_config %}
reactor:
{% for event, actions in reactor_config.items() %}
- '{{ event }}':
{% for action in actions %}
- /srv/reactor/{{ action }}
{% endfor %}
{% endfor %}
{% endif %}
"""
def generate_minion_config_template(self) -> str:
"""生成Minion配置模板"""
return """# Salt Minion Configuration
# Generated by Python Config Generator
# Minion ID: {{ minion_id }}
master: {{ master_ip }}
master_port: {{ master_port }}
id: {{ minion_id }}
environment: {{ environment }}
# Logging
log_level: {{ log_level }}
log_file: /var/log/salt/minion-{{ minion_id }}
# Grains
grains:
roles:
{% for role in roles %}
- {{ role }}
{% endfor %}
environment: {{ environment }}
datacenter: {{ datacenter }}
deployment: {{ deployment }}
# Features
features:
use_master_when_local: {{ use_master_when_local }}
cache_jobs: {{ cache_jobs }}
# Startup
startup_states: {{ startup_states }}
schedule:
{% for job in scheduled_jobs %}
{{ job.name }}:
function: {{ job.function }}
seconds: {{ job.interval }}
splay: {{ job.splay }}
run_on_start: {{ job.run_on_start }}
{% endfor %}
"""
def generate_complete_deployment(self,
template_data: SaltConfigTemplate,
output_dir: str = "./generated_config"):
"""生成完整的部署配置"""
os.makedirs(output_dir, exist_ok=True)
os.makedirs(f"{output_dir}/master", exist_ok=True)
os.makedirs(f"{output_dir}/minions", exist_ok=True)
os.makedirs(f"{output_dir}/states", exist_ok=True)
os.makedirs(f"{output_dir}/pillar", exist_ok=True)
context = {
"generation_date": __import__('datetime').datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
"master_ip": template_data.master_ip,
"minion_id": template_data.minion_id,
"environment": template_data.environment,
"roles": template_data.roles,
"log_level": template_data.log_level,
"publish_port": 4505,
"ret_port": 4506,
"worker_threads": 5,
"max_open_files": 100000,
"auto_accept": True,
"auto_sign_master_sign_pubkey": False,
"external_pillars": [],
"reactor_config": {
"salt/minion/*/start": ["minion_start.sls"],
"salt/beacon/*/*/": ["beacon_alert.sls"]
},
"master_port": 4506,
"datacenter": "dc1",
"deployment": "standard",
"use_master_when_local": False,
"startup_states": "",
"cache_jobs": True,
"scheduled_jobs": [
{
"name": "highstate_check",
"function": "state.highstate",
"interval": 3600,
"splay": 300,
"run_on_start": True
},
{
"name": "package_update",
"function": "pkg.upgrade",
"interval": 86400,
"splay": 600,
"run_on_start": False
}
]
}
# 生成Master配置
master_config = self.jinja_env.from_string(
self.generate_master_config_template()
).render(context)
with open(f"{output_dir}/master/master.conf", 'w') as f:
f.write(master_config)
# 生成Minion配置
minion_config = self.jinja_env.from_string(
self.generate_minion_config_template()
).render(context)
with open(f"{output_dir}/minions/{context['minion_id']}.conf", 'w') as f:
f.write(minion_config)
return output_dir
# 使用示例
if __name__ == "__main__":
from datetime import datetime
# 基本配置生成
generator = SaltConfigGenerator()
# 生成Master配置
master_config = generator.generate_master_config("192.168.1.100")
generator.save_yaml_config(master_config, "master/master.conf")
# 生成Minion配置
minion_config = generator.generate_minion_config(
minion_id="web-server-01",
master_ip="192.168.1.100",
environment="production"
)
generator.save_yaml_config(minion_config, "minions/web-server-01.conf")
# 高级配置生成
advanced_gen = AdvancedConfigGenerator()
config = SaltConfigTemplate(
master_ip="192.168.1.100",
minion_id="web-server-01",
environment="production",
roles=["web", "app", "monitoring"],
log_level="info"
)
output_dir = advanced_gen.generate_complete_deployment(config)
print(f"配置已生成到: {output_dir}")
批量配置生成器
#!/usr/bin/env python3
"""
批量生成多台服务器的Salt配置
"""
import csv
import json
from typing import List, Dict
import concurrent.futures
class BatchConfigGenerator:
"""批量配置生成器"""
def __init__(self, master_ip: str = "192.168.1.100"):
self.master_ip = master_ip
self.generator = SaltConfigGenerator()
def generate_from_csv(self, csv_file: str):
"""从CSV文件批量生成配置"""
with open(csv_file, 'r') as f:
reader = csv.DictReader(f)
for row in reader:
self.generate_single_config(row)
def generate_from_json(self, json_file: str):
"""从JSON文件批量生成配置"""
with open(json_file, 'r') as f:
data = json.load(f)
for item in data:
self.generate_single_config(item)
def generate_single_config(self, config_data: Dict):
"""生成单个配置"""
minion_config = self.generator.generate_minion_config(
minion_id=config_data.get('hostname', 'minion'),
master_ip=self.master_ip,
environment=config_data.get('environment', 'production')
)
# 添加自定义grains
if 'grains' in config_data:
minion_config['grains'].update(config_data['grains'])
# 生成文件名
filename = f"{config_data['hostname']}.conf"
self.generator.save_yaml_config(minion_config, f"minions/{filename}")
def generate_parallel(self, configs: List[Dict]):
"""并行生成配置"""
with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
futures = [
executor.submit(self.generate_single_config, config)
for config in configs
]
concurrent.futures.wait(futures)
# 使用示例
if __name__ == "__main__":
# 批量生成示例
batch_gen = BatchConfigGenerator(master_ip="192.168.1.100")
# 从JSON批量生成
servers = [
{
"hostname": "web-01",
"environment": "production",
"grains": {"roles": ["web"], "region": "us-east-1"}
},
{
"hostname": "db-01",
"environment": "production",
"grains": {"roles": ["database"], "region": "us-east-1"}
},
{
"hostname": "cache-01",
"environment": "staging",
"grains": {"roles": ["cache"], "region": "us-west-2"}
}
]
batch_gen.generate_parallel(servers)
print("批量配置生成完成！")
配置验证器
#!/usr/bin/env python3
"""
Salt配置验证器
"""
class ConfigValidator:
"""配置验证器"""
@staticmethod
def validate_master_config(config: Dict) -> List[str]:
"""验证Master配置"""
errors = []
required_fields = ['interface', 'publish_port', 'ret_port']
for field in required_fields:
if field not in config:
errors.append(f"缺少必要字段: {field}")
if 'publish_port' in config:
if not (1024 <= config['publish_port'] <= 65535):
errors.append("publish_port必须在1024-65535之间")
if config.get('interface') == '0.0.0.0':
errors.append("警告: interface设置为0.0.0.0可能不安全")
return errors
@staticmethod
def validate_minion_config(config: Dict) -> List[str]:
"""验证Minion配置"""
errors = []
if 'master' not in config and 'master' not in str(config):
errors.append("必须指定master地址")
if 'id' not in config:
errors.append("必须指定minion ID")
# 验证日志级别
valid_log_levels = ['debug', 'info', 'warning', 'error', 'critical']
if 'log_level' in config:
if config['log_level'] not in valid_log_levels:
errors.append(f"无效的日志级别: {config['log_level']}")
return errors
# 使用示例
validator = ConfigValidator()
errors = validator.validate_minion_config({'id': 'test-minion'})
if errors:
print("验证错误:", errors)
else:
print("配置验证通过")
这些脚本可以根据你的具体需求进行调整和扩展,主要功能包括：
- 基础配置生成 - 生成Master和Minion的基础配置
- 模板化配置 - 使用Jinja2模板生成动态配置
- 批量生成 - 从CSV/JSON批量生成多台服务器配置
- 配置验证 - 验证生成的配置是否正确
- 多环境支持 - 支持不同环境的配置生成
使用时只需修改配置参数并运行相应的脚本即可生成完整的SaltStack配置文件。
