Python脚本如何生成EMQX集群配置

本文目录导读：

1. 基础配置生成
2. 自动发现配置生成
3. 完整集群配置生成
4. 配置验证工具
5. 使用示例

我来为您介绍如何使用Python脚本生成EMQX集群配置。

基础配置生成

1 单节点配置模板

import yaml
import json
import os
from typing import List, Dict, Any
class EMQXConfigGenerator:
    def __init__(self, cluster_name: str = "emqx-cluster"):
        self.cluster_name = cluster_name
        self.config = {
            "node": {
                "name": "",
                "cookie": "",
                "data_dir": "data",
                "log_dir": "log"
            },
            "cluster": {
                "name": cluster_name,
                "discovery_strategy": "manual",
                "core_nodes": [],
                "replicant_nodes": []
            }
        }
    def set_node_cookie(self, cookie: str):
        """设置Erlang cookie"""
        self.config["node"]["cookie"] = cookie
    def add_node(self, node_name: str, node_type: str = "core", host: str = "localhost", port: int = 1883):
        """添加节点"""
        node_config = {
            "name": f"{node_name}@{host}",
            "type": node_type,
            "listen_port": port,
            "listener": {
                "tcp": port,
                "ssl": port + 1,
                "ws": port + 2
            }
        }
        if node_type == "core":
            self.config["cluster"]["core_nodes"].append(node_config)
        else:
            self.config["cluster"]["replicant_nodes"].append(node_config)
        return node_config
    def generate_yaml(self, output_path: str = "emqx.yaml"):
        """生成YAML配置"""
        with open(output_path, 'w') as f:
            yaml.dump(self.config, f, default_flow_style=False)
        print(f"配置已生成: {output_path}")
    def generate_json(self, output_path: str = "emqx.json"):
        """生成JSON配置"""
        with open(output_path, 'w') as f:
            json.dump(self.config, f, indent=2)
        print(f"配置已生成: {output_path}")

自动发现配置生成

1 基于ETCD的自动发现

class ETCDDiscoveryConfig:
    def __init__(self):
        self.discovery_config = {
            "cluster": {
                "discovery_strategy": "etcd",
                "etcd": {
                    "server": "http://127.0.0.1:2379",
                    "prefix": "emqx/",
                    "node_ttl": "1m",
                    "ssl": {
                        "enable": False,
                        "certfile": "",
                        "keyfile": ""
                    }
                }
            }
        }
    def set_etcd_servers(self, servers: List[str]):
        """设置ETCD服务器列表"""
        self.discovery_config["cluster"]["etcd"]["server"] = ",".join(servers)
    def enable_ssl(self, cert_file: str, key_file: str):
        """启用SSL"""
        ssl_config = self.discovery_config["cluster"]["etcd"]["ssl"]
        ssl_config["enable"] = True
        ssl_config["certfile"] = cert_file
        ssl_config["keyfile"] = key_file
class DNSDiscoveryConfig:
    def __init__(self):
        self.discovery_config = {
            "cluster": {
                "discovery_strategy": "dns",
                "dns": {
                    "record_type": "a",
                    "name": "emqx-cluster.local",
                    "app": "emqx"
                }
            }
        }
    def set_dns_record(self, record_type: str, name: str, app: str = "emqx"):
        """设置DNS记录"""
        self.discovery_config["cluster"]["dns"]["record_type"] = record_type
        self.discovery_config["cluster"]["dns"]["name"] = name
        self.discovery_config["cluster"]["dns"]["app"] = app

完整集群配置生成

class EMQXClusterConfigGenerator:
    def __init__(self):
        self.generators = []
        self.cluster_config = {
            "emqx_cluster": {
                "version": "5.0",
                "nodes": [],
                "discovery": {},
                "security": {}
            }
        }
    def add_node_generator(self, node_config: dict):
        """添加节点生成器"""
        self.generators.append(node_config)
    def generate_multi_node_config(self, 
                                  node_count: int = 3,
                                  base_name: str = "emqx",
                                  base_port: int = 1883,
                                  discovery: str = "manual"):
        """生成多节点配置"""
        nodes = []
        for i in range(node_count):
            node = {
                "name": f"{base_name}@{base_name}-node-{i+1}",
                "type": "core" if i < 2 else "replicant",
                "listen_port": base_port + i,
                "listener": {
                    "tcp": base_port + i,
                    "ssl": base_port + i + 100,
                    "ws": base_port + i + 200,
                    "dashboard": 18083 + i
                },
                "persistent": True if i < 2 else False
            }
            nodes.append(node)
        self.cluster_config["emqx_cluster"]["nodes"] = nodes
        self.cluster_config["emqx_cluster"]["discovery"] = {
            "strategy": discovery
        }
        return nodes
    def add_security_config(self, 
                           enable_auth: bool = True,
                           enable_acl: bool = True,
                           jwt_secret: str = ""):
        """添加安全配置"""
        self.cluster_config["emqx_cluster"]["security"] = {
            "authentication": {
                "enable": enable_auth,
                "backend": "built-in-database",
                "mechanism": "password-based"
            },
            "authorization": {
                "enable": enable_acl,
                "sources": ["built-in-database"]
            }
        }
        if jwt_secret:
            self.cluster_config["emqx_cluster"]["security"]["authentication"][
                "jwt"] = {
                "secret": jwt_secret,
                "verify_claims": ["sub", "exp"]
            }
    def generate_docker_compose(self, output_path: str = "docker-compose.yml"):
        """生成Docker Compose配置"""
        compose_config = {
            "version": "3.8",
            "services": {}
        }
        for i, node in enumerate(self.cluster_config["emqx_cluster"]["nodes"]):
            service_name = f"emqx-node-{i+1}"
            compose_config["services"][service_name] = {
                "image": "emqx/emqx:5.0",
                "container_name": service_name,
                "environment": [
                    f"EMQX_NODE_NAME={node['name']}",
                    f"EMQX_LISTENER__TCP__DEFAULT={node['listen_port']}",
                    f"EMQX_CLUSTER__DISCOVERY_STRATEGY=manual",
                    f"EMQX_CLUSTER__STATIC__SEEDS=emqx@emqx-node-1,emqx@emqx-node-2"
                ],
                "ports": [
                    f"{node['listen_port']}:{node['listen_port']}",
                    f"{node['listener']['dashboard']}:18083"
                ],
                "volumes": [
                    f"./emqx-{i+1}.conf:/opt/emqx/etc/emqx.conf"
                ]
            }
        with open(output_path, 'w') as f:
            yaml.dump(compose_config, f, default_flow_style=False)
        print(f"Docker Compose配置已生成: {output_path}")
    def generate_kubernetes_config(self, output_path: str = "emqx-cluster.yaml"):
        """生成Kubernetes配置"""
        k8s_config = {
            "apiVersion": "apps/v1",
            "kind": "StatefulSet",
            "metadata": {
                "name": "emqx-cluster",
                "labels": {
                    "app": "emqx"
                }
            },
            "spec": {
                "serviceName": "emqx-headless",
                "replicas": len(self.cluster_config["emqx_cluster"]["nodes"]),
                "selector": {
                    "matchLabels": {
                        "app": "emqx"
                    }
                },
                "template": {
                    "metadata": {
                        "labels": {
                            "app": "emqx"
                        }
                    },
                    "spec": {
                        "containers": [{
                            "name": "emqx",
                            "image": "emqx/emqx:5.0",
                            "ports": [
                                {"containerPort": 1883, "name": "mqtt"},
                                {"containerPort": 8883, "name": "mqtt-ssl"},
                                {"containerPort": 8083, "name": "ws"},
                                {"containerPort": 18083, "name": "dashboard"}
                            ],
                            "env": [
                                {"name": "EMQX_CLUSTER__DISCOVERY_STRATEGY", "value": "dns"},
                                {"name": "EMQX_CLUSTER__DNS__RECORD_TYPE", "value": "a"},
                                {"name": "EMQX_CLUSTER__DNS__NAME", "value": "emqx-headless.default.svc.cluster.local"}
                            ]
                        }]
                    }
                }
            }
        }
        with open(output_path, 'w') as f:
            yaml.dump(k8s_config, f, default_flow_style=False)
        print(f"Kubernetes配置已生成: {output_path}")

配置验证工具

class EMQXConfigValidator:
    @staticmethod
    def validate_cluster_config(config: dict) -> List[str]:
        """验证集群配置"""
        errors = []
        # 验证节点配置
        if "emqx_cluster" not in config:
            errors.append("缺少 emqx_cluster 主节点配置")
            return errors
        nodes = config["emqx_cluster"].get("nodes", [])
        if len(nodes) == 0:
            errors.append("至少需要一个节点")
        # 检查节点名称唯一性
        node_names = [node["name"].split("@")[0] for node in nodes]
        if len(node_names) != len(set(node_names)):
            errors.append("节点名称必须唯一")
        # 检查端口冲突
        ports = [node["listen_port"] for node in nodes]
        if len(ports) != len(set(ports)):
            errors.append("端口不能冲突")
        # 验证发现策略
        discovery = config["emqx_cluster"].get("discovery", {})
        if not discovery.get("strategy"):
            errors.append("必须指定发现策略")
        return errors
    @staticmethod
    def validate_network_config(config: dict) -> List[str]:
        """验证网络配置"""
        errors = []
        for node in config.get("emqx_cluster", {}).get("nodes", []):
            # 验证端口范围
            for port_type in ["tcp", "ssl", "ws", "dashboard"]:
                port = node.get("listener", {}).get(port_type, 0)
                if port < 1024 or port > 65535:
                    errors.append(f"节点 {node['name']} 的 {port_type} 端口无效: {port}")
        return errors

使用示例

def main():
    # 创建配置生成器
    generator = EMQXClusterConfigGenerator()
    # 生成3节点配置
    nodes = generator.generate_multi_node_config(
        node_count=3,
        base_name="emqx-cluster",
        base_port=1883,
        discovery="manual"
    )
    # 添加安全配置
    generator.add_security_config(
        enable_auth=True,
        enable_acl=True,
        jwt_secret="your-jwt-secret"
    )
    # 生成不同格式的配置
    generator.generate_docker_compose()
    generator.generate_kubernetes_config()
    # 验证配置
    validator = EMQXConfigValidator()
    errors = validator.validate_cluster_config(generator.cluster_config)
    if errors:
        print("配置验证失败:")
        for error in errors:
            print(f"  - {error}")
    else:
        print("配置验证成功!")
    # 打印生成的节点信息
    for node in nodes:
        print(f"节点: {node['name']}")
        print(f"  类型: {node['type']}")
        print(f"  端口: {node['listen_port']}")
        print(f"  Dashboard: {node['listener']['dashboard']}")
        print()
if __name__ == "__main__":
    main()

这个Python脚本可以：

1. 自动生成EMQX集群配置（YAML/JSON格式）
2. 支持多种发现策略（manual、etcd、dns）
3. 生成Docker Compose和K8s配置
4. 配置验证确保配置正确
5. 支持安全配置（认证、授权、JWT）

您可以根据实际需求调整配置参数。