Python脚本如何生成SaltStack配置

本文目录导读：

1. 基础配置模板
2. 高级配置生成器
3. 批量配置生成器
4. 配置验证器

我来详细说明如何用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("配置验证通过")

这些脚本可以根据你的具体需求进行调整和扩展,主要功能包括：

1. 基础配置生成 - 生成Master和Minion的基础配置
2. 模板化配置 - 使用Jinja2模板生成动态配置
3. 批量生成 - 从CSV/JSON批量生成多台服务器配置
4. 配置验证 - 验证生成的配置是否正确
5. 多环境支持 - 支持不同环境的配置生成

使用时只需修改配置参数并运行相应的脚本即可生成完整的SaltStack配置文件。