Блог

Главная - Вопросы по программированию - Почему Nginx усекает потоковый ответ gRPC?

Почему Nginx усекает потоковый ответ gRPC?

#nginx #flask #grpc #gunicorn #gevent

Вопрос:

Я задавал этот вопрос раньше, но решил удалить этот старый вопрос и переформулировать его вместе с минимальным воспроизводимым примером. Проблема в том, что когда я развертываю свой веб-сервер gunicorn на nginx, мои потоковые ответы с моего сервера go через gRPC усекаются. Все подробности можно найти в репозитории. Моя конфигурация nginx для этого сайта выглядит следующим образом:

 server {
    listen 80 default_server;
    server_name example.com;

    location / {
    #include proxy_params;
    proxy_pass http://localhost:5000;
    proxy_buffering off;
    chunked_transfer_encoding off;
    }
}

Код, получающий и анализирующий ответ на переднем конце, выглядит следующим образом:

         <script>
            (async function(){
                const response = await fetch("{{ url_for('realtimedata') }}");
                const reader = response.body.pipeThrough(new TextDecoderStream()).getReader();
                while (true) {
                    const {done, value} = await reader.read();
                    if (done) break;
                    try {
                        console.log('Received', value);
                        const rtd = JSON.parse(value);
                        console.log('Parsed', rtd);
                    } catch(err) {
                        console.log(err);
                    }
                }
            })()
        </script>

Следует отметить, что в отношении данных с сервера go одна служба предоставляет объект данных с 96 полями, а другая служба предоставляет данные с 200 полями. Что делает ответ входящего потока переменной длины (в байтах).

Я хочу использовать gunicorn, потому что у меня может быть несколько слушателей одновременно. Использование gunicorn решило проблему, когда все ответы поступали на веб-сервер, но они распределялись между активными клиентами. Таким образом, каждый клиент получит разный ответ, но не все из них.

ИЗМЕНИТЬ: Я попытался изменить размер объекта ответа на госервере, чтобы он был одинаковым для обеих служб, но усечение все равно произошло. Наличие переменной длины, похоже, не является проблемой. Я также пытался сделать это с помощью uWSGI вместо gunicorn, и проблема сохраняется. Я даже установил uwsgi_buffering off; , и проблема сохраняется.

ОБНОВЛЕНИЕ: Я запустил минимальный воспроизводимый пример с Apache2 вместо Nginx, и у меня та же проблема. Может быть, проблема в чем-то другом.

Ответ №1:

Глядя на ваш код на python, кажется, что перенос данных с бэкенда на интерфейс был бы лучше с помощью websockets. Я переписал ваш бэкэнд, чтобы использовать FastAPI вместо Flask, и изменил конфигурацию nginx.

main.py 

 import asyncio
import dependencies.rpc_pb2 as r
import dependencies.rpc_pb2_grpc as rpc
from fastapi import FastAPI, WebSocket, Request
from fastapi.templating import Jinja2Templates
import grpc
import json
import os

os.environ["GRPC_SSL_CIPHER_SUITES"] = 'HIGH ECDSA'

app = FastAPI()
templates = Jinja2Templates(directory="templates")
server_addr = "localhost"
server_port = 3567

@app.get("/")
def read_root(request: Request):
    return templates.TemplateResponse("index.html", {"request": request})

def parseRtd(rtd):
    rtdDict = {}
    rtdDict["source"] = rtd.source
    rtdDict["is_scanning"] = rtd.is_scanning
    rtdDict["timestamp"] = int(rtd.timestamp)
    rtdDict["data"] = {}
    for key, v in rtd.data.items():
        rtdDict["data"][int(key)] = {"name": v.name, "value": v.value}
    return rtdDict

def get_rtd():
    channel = grpc.insecure_channel(f"{server_addr}:{server_port}")
    stub = rpc.RpcServiceStub(channel)
    for rtd in stub.SubscribeDataStream(r.SubscribeDataRequest()):
        yield parseRtd(rtd)

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    await websocket.send_json({"test": "this is a test"})
    it = get_rtd()
    while True:
        await asyncio.sleep(0.1)
        payload = next(it)
        await websocket.send_json(payload)

index.html 

 <html>
    <head>
        <script src="https://cdnjs.cloudflare.com/ajax/libs/socket.io/4.4.0/socket.io.js" integrity="sha512-nYuHvSAhY5lFZ4ixSViOwsEKFvlxHMU2NHts1ILuJgOS6ptUmAGt/0i5czIgMOahKZ6JN84YFDA mCdky7dD8A==" crossorigin="anonymous" referrerpolicy="no-referrer"></script>
    </head>
    <body>
        <script>
            var ws = new WebSocket("ws://localhost:5000/ws");     
            ws.onopen = function () {
                console.log("websocket was open");
            };
            ws.onclose = () => {
                console.log("Websocket was closed!");
            }
            ws.onerror = (error) =>{
                console.error("Websocket error: "   JSON.stringify(error));
            };
            ws.onmessage = (message) => {
                console.log("MSG: "   message.data );
            };
        </script>
    </body>
</html>

webserver.conf 

 server {
    listen 80 default_server;
    server_name example.com;

    location / {
        include proxy_params;
        proxy_pass http://localhost:5000;
    }

    location /ws {
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header Host $host;
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "Upgrade";
        proxy_pass http://localhost:5000;
    }
}
Метки: