如何通过Python脚本实现ELK服务一键启动与停止

你遇到的端口冲突、重启失败问题，核心原因是直接调用subprocess.Popen时没有做进程组管理：ES、Kibana的启动脚本会fork出Java、Node.js运行时子进程，手动kill父进程时，子进程会变成孤儿进程继续占用9200、5601端口，下次启动自然会触发端口占用、ES自动切9201端口、Kibana连不上ES的问题。

实现核心逻辑 #

• 启动服务时将进程归入独立进程组，留存进程ID，停止时直接终止整个进程组，彻底避免孤儿进程残留
• 启动前先校验目标端口是否被占用，避免重复启动
• 加服务健康检查逻辑，确认服务真正可用后再完成启动流程
• 停止服务后校验端口释放状态，确认停止操作生效

可直接复用的代码片段 #

以下代码基于Python标准库实现，适配macOS/Linux环境，Windows环境只需调整进程组创建参数即可：

import os
import time
import signal
import socket
import subprocess
import urllib.request
from typing import Optional

# 配置项，根据自己的实际路径修改
ES_VERSION = "8.13.0"
KIBANA_VERSION = "8.13.0"
BASE_PATH = "/your/local/service/path"  # 替换为存放ES、Kibana的根目录
ES_HOST = "localhost"
ES_PORT = 9200
KIBANA_PORT = 5601

class ELKController:
    def __init__(self):
        self.es_process: Optional[subprocess.Popen] = None
        self.kibana_process: Optional[subprocess.Popen] = None
        self.es_path = f"{BASE_PATH}/elasticsearch/elasticsearch-{ES_VERSION}/bin/elasticsearch"
        self.kibana_path = f"{BASE_PATH}/kibana/kibana-{KIBANA_VERSION}-darwin-x86_64/bin/kibana"

    def _check_port_in_use(self, port: int) -> bool:
        """检查指定端口是否被占用"""
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
            return s.connect_ex(('127.0.0.1', port)) == 0

    def _wait_for_service(self, url: str, timeout: int = 60) -> bool:
        """等待服务启动完成"""
        start = time.time()
        while time.time() - start < timeout:
            try:
                urllib.request.urlopen(url, timeout=2)
                return True
            except Exception:
                time.sleep(2)
        return False

    def start(self) -> bool:
        """启动ES和Kibana服务"""
        # 先检查端口占用
        if self._check_port_in_use(ES_PORT):
            print(f"端口{ES_PORT}已被占用，ES无法启动")
            return False
        if self._check_port_in_use(KIBANA_PORT):
            print(f"端口{KIBANA_PORT}已被占用，Kibana无法启动")
            return False

        # 启动ES，创建独立进程组
        self.es_process = subprocess.Popen(
            self.es_path,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            preexec_fn=os.setsid  # Unix系系统下将进程归入新进程组
        )
        print("正在启动Elasticsearch...")
        if not self._wait_for_service(f"http://{ES_HOST}:{ES_PORT}"):
            print("Elasticsearch启动超时")
            self.stop()
            return False
        print("Elasticsearch启动完成")

        # 启动Kibana
        self.kibana_process = subprocess.Popen(
            self.kibana_path,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            preexec_fn=os.setsid
        )
        print("正在启动Kibana...")
        if not self._wait_for_service(f"http://{ES_HOST}:{KIBANA_PORT}/api/status", timeout=120):
            print("Kibana启动超时")
            self.stop()
            return False
        print("Kibana启动完成，服务全部就绪")
        return True

    def stop(self):
        """停止ES和Kibana服务，终止整个进程组"""
        for proc, name, port in [
            (self.kibana_process, "Kibana", KIBANA_PORT),
            (self.es_process, "Elasticsearch", ES_PORT)
        ]:
            if proc and proc.poll() is None:
                print(f"正在停止{name}...")
                # 给进程组发终止信号
                os.killpg(os.getpgid(proc.pid), signal.SIGTERM)
                # 等待10秒，没退出就强制杀死
                try:
                    proc.wait(timeout=10)
                except subprocess.TimeoutExpired:
                    os.killpg(os.getpgid(proc.pid), signal.SIGKILL)
            # 确认端口释放
            wait_time = 0
            while self._check_port_in_use(port) and wait_time < 15:
                time.sleep(1)
                wait_time +=1
            print(f"{name}已停止")
        self.es_process = None
        self.kibana_process = None

if __name__ == "__main__":
    import sys
    controller = ELKController()
    if len(sys.argv) < 2:
        print("用法: python elk_ctl.py [start|stop|restart]")
        sys.exit(1)
    action = sys.argv[1]
    if action == "start":
        controller.start()
    elif action == "stop":
        controller.stop()
    elif action == "restart":
        controller.stop()
        time.sleep(3)
        controller.start()
    else:
        print(f"不支持的操作: {action}")

更省心的现成方案 #

如果不想自己维护进程管理逻辑，优先选Docker Compose编排：

• 只需要写一份简单的compose配置文件，定义ES、Kibana的版本、端口、数据卷
• Python脚本里只需要调用subprocess.run(["docker", "compose", "up", "-d"])启动，调用subprocess.run(["docker", "compose", "down"])停止
• 天然不存在孤儿进程、端口残留问题，也不用手动处理本地Java、Node.js版本依赖，环境一致性更好，换设备也能直接用

注意：如果你的ES开启了安全认证，健康检查部分加上对应鉴权头即可，其余逻辑不需要改动。

内容的提问来源于stack exchange，提问作者Akshay Jain