线程化Flask应用在Jenkins任务中报OSError:端口已占用错误

本地环境中以线程方式启动Flask应用测试完全正常，但在Jenkins流水线运行测试时，出现端口已占用错误。

环境信息 #

Jenkins流水线执行的测试命令 #

python3.8 -m pytest -n 1 --reruns 1 -v -m Smoke --tb=short --instafail --durations=0

pytest钩子中启动Flask线程的代码 #

flask_thread = threading.Thread(target=run_flask_app)
flask_thread.daemon = True
flask_thread.start()
wait_for_trigger()

Flask应用代码 #

from flask import Flask, request
from threading import Event
from werkzeug.serving import run_simple
import logging

app = Flask(__name__)
test_ready_event = Event()
logging.basicConfig(filename='flask_thread.log', level=logging.DEBUG,
                    format='%(asctime)s - %(levelname)s - %(message)s')


@app.route('/trigger', methods=['GET'])
def trigger_test_continue():
    test_ready_event.set()
    print("Trigger received. Signalling pytest to continue.")
    logging.info("Trigger received. Signalling pytest to continue.")
    return "Continuing tests.", 200


def run_flask_app():
    logging.info("Starting Flask app.")
    run_simple('0.0.0.0', 5055, app, use_reloader=False, use_debugger=True)


def stop_flask_app():
    func = request.environ.get('werkzeug.server.shutdown')
    if func is None:
        raise RuntimeError('Not running with the Werkzeug Server')
    func()
    logging.info("Flask app stopped.")


def wait_for_trigger():
    print("Waiting for the trigger to continue tests...")
    logging.info("Waiting for the trigger to continue tests...")
    test_ready_event.wait()
    print("Trigger received. Continuing with tests.")
    logging.info("Trigger received. Continuing with tests.")

报错信息 #

Exception in thread Thread-1:
Traceback (most recent call last):
  File "/usr/lib/python3.8/threading.py", line 932, in _bootstrap_inner
    self.run()
  File "/usr/lib/python3.8/threading.py", line 870, in run
    self._target(*self._args, **self._kwargs)
  File "/var/lib/path/execution_stopper.py", line 22, in run_flask_app
    run_simple('0.0.0.0', 5055, app, use_reloader=False, use_debugger=True)
  File "/usr/local/lib/python3.8/dist-packages/werkzeug/serving.py", line 1052, in run_simple
    inner()
  File "/usr/local/lib/python3.8/dist-packages/werkzeug/serving.py", line 996, in inner
    srv = make_server(
  File "/usr/local/lib/python3.8/dist-packages/werkzeug/serving.py", line 862, in make_server
    return BaseWSGIServer(
  File "/usr/local/lib/python3.8/dist-packages/werkzeug/serving.py", line 740, in __init__
    HTTPServer.__init__(self, server_address, handler)
  File "/usr/lib/python3.8/socketserver.py", line 452, in __init__
    self.server_bind()
  File "/usr/lib/python3.8/http/server.py", line 138, in server_bind
    socketserver.TCPServer.server_bind(self)
  File "/usr/lib/python3.8/socketserver.py", line 466, in server_bind
    self.socket.bind(self.server_address)
OSError: [Errno 98] Address already in use

Jenkins机器端口检查结果 #

执行命令：

sudo netstat -tulpn | grep :5005

返回结果：

tcp        0      0 0.0.0.0:5005            0.0.0.0:*               LISTEN      1636790/python3.8

解决思路与方案 #

1. 清理残留进程 #

从端口检查结果来看，之前的Python进程未正常终止，占用了目标端口。在Jenkins流水线的测试步骤前添加清理命令：

# 杀掉占用5005端口的Python进程
sudo kill -9 $(sudo lsof -t -i:5005) || true

2. 确保Flask进程在测试结束后自动终止 #

当前stop_flask_app方法依赖请求环境，无法主动触发终止。修改代码，通过全局事件控制Flask服务启停：

from threading import Event

# 添加全局停止事件
app_stop_event = Event()

def run_flask_app():
    logging.info("Starting Flask app.")
    from werkzeug.serving import make_server
    # 创建服务器实例
    server = make_server('0.0.0.0', 5055, app, use_reloader=False, use_debugger=True)
    # 启动服务线程
    server_thread = threading.Thread(target=server.serve_forever)
    server_thread.start()
    # 等待停止信号
    app_stop_event.wait()
    server.shutdown()
    logging.info("Flask app stopped.")

# 在pytest会话结束时触发停止事件
def pytest_sessionfinish(session, exitstatus):
    app_stop_event.set()

3. 使用随机端口避免冲突 #

如果Jenkins存在多任务并发执行的情况，可改为使用随机端口启动Flask：

def run_flask_app():
    logging.info("Starting Flask app.")
    from werkzeug.serving import make_server
    # 传入0让系统分配随机端口
    server = make_server('0.0.0.0', 0, app, use_reloader=False, use_debugger=True)
    # 获取实际分配的端口，供后续测试调用
    port = server.server_address[1]
    logging.info(f"Flask app running on port {port}")
    server_thread = threading.Thread(target=server.serve_forever)
    server_thread.start()
    app_stop_event.wait()
    server.shutdown()

4. 限制流水线并发执行 #

若多个流水线实例同时运行导致端口冲突，可在Jenkins任务配置中开启“不允许并发构建”，确保同一时间只有一个测试任务执行。

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