SageMaker Batch Transform访问nginx失败问题排查与解决

基于Docker在SageMaker部署机器学习模型时，训练作业可正常运行，但使用Batch Transform执行模型推理时出现nginx相关报错，错误日志如下：

涉及的配置与代码文件如下：

1. nginx.conf配置 #

worker_processes 1;
daemon off; # Prevent forking


pid /tmp/nginx.pid;
error_log /var/log/nginx/error.log;

events {
  # defaults
}

http {
  include /etc/nginx/mime.types;
  default_type application/octet-stream;
  access_log /var/log/nginx/access.log combined;
  
  upstream gunicorn {
    server unix:/tmp/gunicorn.sock;
  }

  server {
    listen 8080 deferred;
    client_max_body_size 5m;

    keepalive_timeout 5;
    proxy_read_timeout 1200s;

    location ~ ^/(ping|invocations) {
      proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
      proxy_set_header Host $http_host;
      proxy_redirect off;
      proxy_pass http://gunicorn;
    }

    location / {
      return 404 "{}";
    }
  }
}

2. serve启动脚本 #

#!/usr/bin/env python

# This file implements the scoring service shell. You don't necessarily need to modify it for various
# algorithms. It starts nginx and gunicorn with the correct configurations and then simply waits until
# gunicorn exits.
#
# The flask server is specified to be the app object in wsgi.py
#
# We set the following parameters:
#
# Parameter                Environment Variable              Default Value
# ---------                --------------------              -------------
# number of workers        MODEL_SERVER_WORKERS              the number of CPU cores
# timeout                  MODEL_SERVER_TIMEOUT              60 seconds

import multiprocessing
import os
import signal
import subprocess
import sys

cpu_count = multiprocessing.cpu_count()

model_server_timeout = os.environ.get('MODEL_SERVER_TIMEOUT', 60)
model_server_workers = int(os.environ.get('MODEL_SERVER_WORKERS', cpu_count))

def sigterm_handler(nginx_pid, gunicorn_pid):
    try:
        os.kill(nginx_pid, signal.SIGQUIT)
    except OSError:
        pass
    try:
        os.kill(gunicorn_pid, signal.SIGTERM)
    except OSError:
        pass

    sys.exit(0)

def start_server():
    print('Starting the inference server with {} workers.'.format(model_server_workers))


    # link the log streams to stdout/err so they will be logged to the container logs
    subprocess.check_call(['ln', '-sf', '/dev/stdout', '/var/log/nginx/access.log'])
    subprocess.check_call(['ln', '-sf', '/dev/stderr', '/var/log/nginx/error.log'])

    nginx = subprocess.Popen(['nginx', '-c', '/opt/program/nginx.conf'])
    gunicorn = subprocess.Popen(['gunicorn',
                                 '--timeout', str(model_server_timeout),
                                 '-k', 'sync',
                                 '-b', 'unix:/tmp/gunicorn.sock',
                                 '-w', str(model_server_workers),
                                 'wsgi:app'])

    signal.signal(signal.SIGTERM, lambda a, b: sigterm_handler(nginx.pid, gunicorn.pid))

    # If either subprocess exits, so do we.
    pids = set([nginx.pid, gunicorn.pid])
    while True:
        pid, _ = os.wait()
        if pid in pids:
            break

    sigterm_handler(nginx.pid, gunicorn.pid)
    print('Inference server exiting')

# The main routine just invokes the start function.

if __name__ == '__main__':
    start_server()

3. predictor.py Flask推理处理逻辑 #

# This is the file that implements a flask server to do inferences. It's the file that you will modify to
# implement the scoring for your own algorithm.

from __future__ import print_function

import io
import json
import os
import pickle
import signal
import sys
import traceback

import flask
import pandas as pd

prefix = "/opt/ml/"
model_path = os.path.join(prefix, "model")

# A singleton for holding the model. This simply loads the model and holds it.
# It has a predict function that does a prediction based on the model and the input data.


class ScoringService(object):
    model = None  # Where we keep the model when it's loaded

    @classmethod
    def get_model(cls):
        """Get the model object for this instance, loading it if it's not already loaded."""
        if cls.model == None:
            with open(os.path.join(model_path, "_logreg_cap.pkl"), "rb") as inp:
                cls.model = pickle.load(inp)
        return cls.model

    @classmethod
    def predict(cls, input):
        """For the input, do the predictions and return them.

        Args:
            input (a pandas dataframe): The data on which to do the predictions. There will be
                one prediction per row in the dataframe"""
        clf = cls.get_model()
        return clf.predict(input)


# The flask app for serving predictions
app = flask.Flask(__name__)


@app.route("/ping", methods=["GET"])
def ping():
    """Determine if the container is working and healthy. In this sample container, we declare
    it healthy if we can load the model successfully."""
    health = ScoringService.get_model() is not None  # You can insert a health check here

    status = 200 if health else 404
    return flask.Response(response="\n", status=status, mimetype="application/json")


@app.route("/invocations", methods=["POST"])
def transformation():
    """Do an inference on a single batch of data. In this sample server, we take data as CSV, convert
    it to a pandas data frame for internal use and then convert the predictions back to CSV (which really
    just means one prediction per line, since there's a single column.
    """
    data = None

    # Convert from CSV to pandas
    if flask.request.content_type == "text/csv":
        data = flask.request.data.decode("utf-8")
        s = io.StringIO(data)
        data = pd.read_csv(s, header=None)
    else:
        return flask.Response(
            response="This predictor only supports CSV data", status=415, mimetype="text/plain"
        )

    print("Invoked with {} records".format(data.shape[0]))

    # Do the prediction
    predictions = ScoringService.predict(data)

    # Convert from numpy back to CSV
    out = io.StringIO()
    pd.DataFrame({"results": predictions}).to_csv(out, header=False, index=False)
    result = out.getvalue()

    return flask.Response(response=result, status=200, mimetype="text/csv")

该nginx报错本质是Nginx无法正常将请求转发给后端Gunicorn服务，或请求在Nginx层被拦截，按以下步骤修复即可：

1. 修正nginx.conf配置，解决权限与请求大小限制问题
   在配置文件最顶部添加user root;，指定Nginx以root用户运行，和Gunicorn、serve脚本的运行用户保持一致，避免Unix Socket文件权限不足导致的502错误；同时将client_max_body_size 5m;调整为client_max_body_size 100m;，避免单批推理数据超过大小限制被Nginx拦截。修改后的nginx.conf核心片段如下：

   user root; # 新增该行
   worker_processes 1;
   daemon off; # Prevent forking
   # ... 其余原有配置保持不变
   server {
       listen 8080 deferred;
       client_max_body_size 100m; # 修改该行，调大请求体上限
       # ... 其余location配置保持不变
   }
   

2. 修正serve启动脚本，添加Gunicorn Socket文件权限参数
   在Gunicorn启动命令中添加--umask 007参数，保证生成的/tmp/gunicorn.sock文件有足够权限让Nginx进程访问，修改后的Gunicorn启动片段如下：

   gunicorn = subprocess.Popen(['gunicorn',
                                '--timeout', str(model_server_timeout),
                                '-k', 'sync',
                                '-b', 'unix:/tmp/gunicorn.sock',
                                '--umask', '007', # 新增该行
                                '-w', str(model_server_workers),
                                'wsgi:app'])
   

3. 修正predictor.py的Content-Type校验逻辑，兼容SageMaker Batch Transform的请求头
   SageMaker Batch Transform传递CSV数据时，部分版本会使用application/csv作为Content-Type，原有逻辑仅识别text/csv会直接返回415错误，触发Nginx上报上游错误。将Content-Type判断逻辑修改为兼容两种格式：

   # 原判断逻辑：if flask.request.content_type == "text/csv":
   # 替换为：
   if flask.request.content_type in ["text/csv", "application/csv"]:
   

4. 校验Batch Transform作业配置
   创建批量转换作业时，手动指定输入ContentType为text/csv，输出Accept为text/csv，同时将MaxPayloadInMB参数设置为100，和Nginx的请求大小限制保持一致，避免平台侧拆分的请求批次超过服务承载上限。

修复完成后重新构建Docker镜像并推送至ECR，本地可先通过以下命令验证服务可用性，确认正常后再提交Batch Transform作业即可：

• 健康检查：curl http://localhost:8080/ping，预期返回200状态码
• 推理测试：curl -X POST -H "Content-Type: text/csv" --data-binary @你的测试csv文件路径 http://localhost:8080/invocations，预期正常返回推理结果

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