Python脚本如何生成Consul配置

wen 实用脚本 2026-07-06 6

本文目录导读：

生成HCL格式配置（推荐用于配置文件）
生成HCL格式（Consul原生格式）
使用YAML生成配置（适用于更复杂的场景）
动态服务注册（通过API）
使用模板引擎（适用于动态配置）
最佳实践建议

我来介绍几种用Python生成Consul配置的方法：

生成HCL格式配置（推荐用于配置文件）

基本配置生成

import json
import os
def generate_consul_config(service_name, port, tags=None, health_check=None):
    """生成Consul服务注册配置"""
    config = {
        "service": {
            "name": service_name,
            "port": port,
            "tags": tags or ["python"],
            "check": health_check or {
                "http": f"http://localhost:{port}/health",
                "interval": "10s",
                "timeout": "5s"
            }
        }
    }
    # 保存为JSON格式（Consul也支持JSON）
    filename = f"{service_name}.json"
    with open(f"/etc/consul.d/{filename}", 'w') as f:
        json.dump(config, f, indent=2)
    return filename
# 使用示例
generate_consul_config(
    service_name="web-api",
    port=8080,
    tags=["api", "production"],
    health_check={
        "http": "http://localhost:8080/api/health",
        "interval": "15s",
        "deregister_critical_service_after": "30s"
    }
)

多服务批量生成

def batch_generate_consul_configs(services):
    """批量生成多个服务配置"""
    configs = {}
    for service in services:
        config = {
            "service": {
                "name": service["name"],
                "port": service["port"],
                "tags": service.get("tags", []),
                "check": {
                    "http": f"http://localhost:{service['port']}/{service.get('health_path', 'health')}",
                    "interval": service.get("interval", "10s"),
                    "timeout": service.get("timeout", "5s")
                }
            }
        }
        # 如果有额外配置
        if "meta" in service:
            config["service"]["meta"] = service["meta"]
        filename = f"{service['name']}.json"
        configs[filename] = config
    # 保存到指定目录
    output_dir = "/etc/consul.d/"
    os.makedirs(output_dir, exist_ok=True)
    for filename, config in configs.items():
        filepath = os.path.join(output_dir, filename)
        with open(filepath, 'w') as f:
            json.dump(config, f, indent=2)
        print(f"Generated: {filepath}")
    return configs
# 使用示例
services = [
    {
        "name": "api-gateway",
        "port": 3000,
        "tags": ["gateway", "production"],
        "health_path": "status"
    },
    {
        "name": "user-service",
        "port": 5000,
        "tags": ["microservice", "v1"],
        "meta": {"version": "1.0.0"}
    }
]
batch_generate_consul_configs(services)

生成HCL格式（Consul原生格式）

def generate_hcl_config(agent_config):
    """生成HCL格式的Consul代理配置"""
    hcl_parts = []
    # 基本配置
    hcl_parts.append(f"""
data_dir = "{agent_config.get('data_dir', '/opt/consul')}"
node_name = "{agent_config['node_name']}"
datacenter = "{agent_config.get('datacenter', 'dc1')}"
""")
    # 服务器配置
    if agent_config.get('server'):
        hcl_parts.append(f"""
server = true
bootstrap_expect = {agent_config['bootstrap_expect']}
""")
    # 客户端地址
    client_addr = agent_config.get('client_addr', '0.0.0.0')
    hcl_parts.append(f"""
client_addr = "{client_addr}"
bind_addr = "{agent_config.get('bind_addr', '0.0.0.0')}"
""")
    # 日志配置
    hcl_parts.append(f"""
log_level = "{agent_config.get('log_level', 'INFO')}"
log_file = "{agent_config.get('log_file', '/var/log/consul/')}"
""")
    # UI配置
    if agent_config.get('ui', True):
        hcl_parts.append("""
ui_config {
  enabled = true
}
""")
    # 加密配置
    if 'encrypt_key' in agent_config:
        hcl_parts.append(f"""
encrypt = "{agent_config['encrypt_key']}"
""")
    # 连接配置
    if 'ports' in agent_config:
        ports = agent_config['ports']
        hcl_parts.append(f"""
ports {{
  http = {ports.get('http', 8500)}
  https = {ports.get('https', -1)}
  dns = {ports.get('dns', 8600)}
  grpc = {ports.get('grpc', 8502)}
  serf_lan = {ports.get('serf_lan', 8301)}
  serf_wan = {ports.get('serf_wan', 8302)}
  server = {ports.get('server', 8300)}
}}
""")
    # Retry join配置
    if 'retry_join' in agent_config:
        joins = '", "'.join(agent_config['retry_join'])
        hcl_parts.append(f"""
retry_join = ["{joins}"]
""")
    return ''.join(hcl_parts)
# 使用示例
config = {
    'node_name': 'node1',
    'datacenter': 'dc1',
    'server': True,
    'bootstrap_expect': 3,
    'client_addr': '0.0.0.0',
    'bind_addr': '0.0.0.0',
    'log_level': 'INFO',
    'data_dir': '/opt/consul',
    'ui': True,
    'retry_join': ['192.168.1.1', '192.168.1.2', '192.168.1.3']
}
hcl_content = generate_hcl_config(config)
with open('/etc/consul.d/agent.hcl', 'w') as f:
    f.write(hcl_content)

使用YAML生成配置（适用于更复杂的场景）

import yaml
class ConsulConfigGenerator:
    """Consul配置生成器"""
    def __init__(self):
        self.services = []
        self.agent_config = {}
    def add_service(self, name, port, tags=None, health_check=None):
        """添加服务配置"""
        service = {
            "id": f"{name}-{port}",
            "name": name,
            "port": port,
            "tags": tags or [],
            "check": health_check or {
                "http": f"http://localhost:{port}/health",
                "interval": "10s"
            }
        }
        self.services.append(service)
        return self
    def set_agent_config(self, **kwargs):
        """设置代理配置"""
        self.agent_config.update(kwargs)
        return self
    def to_yaml(self, filepath):
        """导出为YAML格式"""
        config = {
            "services": self.services,
            **self.agent_config
        }
        with open(filepath, 'w') as f:
            yaml.dump(config, f, default_flow_style=False)
        return filepath
    def to_json(self, filepath):
        """导出为JSON格式"""
        config = {
            "services": self.services,
            **self.agent_config
        }
        with open(filepath, 'w') as f:
            json.dump(config, f, indent=2)
        return filepath
# 使用示例
generator = ConsulConfigGenerator()
generator\
    .add_service("web", 8080, tags=["production"])\
    .add_service("api", 5000, tags=["v1", "internal"])\
    .set_agent_config(
        data_dir="/opt/consul",
        log_level="INFO",
        enable_script_checks=False
    )
# 生成配置文件
generator.to_json("/etc/consul.d/services.json")
generator.to_yaml("/etc/consul.d/services.yaml")

动态服务注册（通过API）

import requests
import time
class ConsulServiceRegistrar:
    """通过Consul API注册服务"""
    def __init__(self, consul_host='localhost', consul_port=8500):
        self.base_url = f"http://{consul_host}:{consul_port}/v1"
    def register_service(self, service_name, port, tags=None, health_path='/health'):
        """注册服务"""
        registration = {
            "ID": f"{service_name}-{port}",
            "Name": service_name,
            "Port": port,
            "Tags": tags or [],
            "Check": {
                "HTTP": f"http://localhost:{port}{health_path}",
                "Interval": "10s",
                "Timeout": "5s",
                "DeregisterCriticalServiceAfter": "30s"
            }
        }
        response = requests.put(
            f"{self.base_url}/agent/service/register",
            json=registration
        )
        if response.status_code == 200:
            print(f"Service {service_name} registered successfully")
            return True
        else:
            print(f"Failed to register service: {response.text}")
            return False
    def deregister_service(self, service_id):
        """注销服务"""
        response = requests.put(
            f"{self.base_url}/agent/service/deregister/{service_id}"
        )
        return response.status_code == 200
    def list_services(self):
        """列出所有服务"""
        response = requests.get(f"{self.base_url}/agent/services")
        return response.json()
# 使用示例
registrar = ConsulServiceRegistrar()
registrar.register_service(
    "my-service", 
    8080, 
    tags=["python", "api"],
    health_path="/api/health"
)

使用模板引擎（适用于动态配置）

from jinja2 import Template
import json
# 定义模板
HCL_TEMPLATE = """
# Consul配置 - 自动生成
data_dir = "{{ data_dir }}"
node_name = "{{ node_name }}"
datacenter = "{{ datacenter }}"
{% if is_server %}
server = true
bootstrap_expect = {{ bootstrap_expect }}
{% endif %}
{% if retry_join %}
retry_join = [{% for node in retry_join %}"{{ node }}"{% if not loop.last %}, {% endif %}{% endfor %}]
{% endif %}
ports {
  http = {{ ports.http | default(8500) }}
  dns  = {{ ports.dns | default(8600) }}
}
{% if services %}
# 服务配置
{% for service in services %}
service {
  name = "{{ service.name }}"
  port = {{ service.port }}
  tags = [{% for tag in service.tags %}"{{ tag }}"{% if not loop.last %}, {% endif %}{% endfor %}]
  check {
    http     = "{{ service.health_check.url }}"
    interval = "{{ service.health_check.interval }}"
    timeout  = "{{ service.health_check.timeout }}"
  }
}
{% endfor %}
{% endif %}
"""
def generate_config_from_template(config_data):
    """使用模板生成配置"""
    template = Template(HCL_TEMPLATE)
    return template.render(config_data)
# 使用示例
config_data = {
    "data_dir": "/opt/consul",
    "node_name": "production-1",
    "datacenter": "dc1",
    "is_server": True,
    "bootstrap_expect": 3,
    "retry_join": ["10.0.0.1", "10.0.0.2"],
    "ports": {
        "http": 8500,
        "dns": 8600
    },
    "services": [
        {
            "name": "web",
            "port": 8080,
            "tags": ["frontend", "production"],
            "health_check": {
                "url": "http://localhost:8080/health",
                "interval": "10s",
                "timeout": "5s"
            }
        }
    ]
}
hcl_output = generate_config_from_template(config_data)
with open("consul_config.hcl", "w") as f:
    f.write(hcl_output)

最佳实践建议

验证配置：生成后使用consul validate命令检查
版本控制：将配置生成脚本纳入版本管理
安全性：避免在配置中硬编码敏感信息
环境差异化：支持不同环境（dev/staging/prod）的配置
错误处理：添加适当的异常处理

这些方法可以根据你的具体需求选择使用，对于静态配置推荐使用JSON/YAML格式，对于动态场景推荐使用Consul API。