From 88b2e71a774a471cbb9b1394ec78e79d229232b5 Mon Sep 17 00:00:00 2001 From: master <727475183@qq.com> Date: Fri, 8 May 2026 21:00:14 +0800 Subject: [PATCH] fix(event): add 401 handling to prevent infinite retry loop on auth failure - UploadError struct to distinguish fatal auth errors from network errors - Clear buffer and throttle on 401/403 to save bandwidth - Prevent dead-loop retry when token is invalid or expired --- .dockerignore | 11 ++++++++++ Dockerfile | 38 +++++++++++++++++++++++++++++++++ data/config/edge.yaml | 16 ++++++++++++++ docker-compose.yml | 34 ++++++++++++++++++++++++++++++ entrypoint.sh | 41 ++++++++++++++++++++++++++++++++++++ internal/event/uploader.go | 43 ++++++++++++++++++++++++++++++-------- 6 files changed, 174 insertions(+), 9 deletions(-) create mode 100644 .dockerignore create mode 100644 Dockerfile create mode 100644 data/config/edge.yaml create mode 100644 docker-compose.yml create mode 100644 entrypoint.sh diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..167e1de --- /dev/null +++ b/.dockerignore @@ -0,0 +1,11 @@ +.git/ +.gitignore +build/ +data/ +*.om +*.onnx +*.tar.gz +logs/ +venv/ +__pycache__/ +*.pyc diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..d949a8e --- /dev/null +++ b/Dockerfile @@ -0,0 +1,38 @@ +# 使用华为云镜像加速的 Python 3.9 Slim +FROM swr.cn-north-4.myhuaweicloud.com/ddn-k8s/docker.io/library/python:3.9-slim + +ENV DEBIAN_FRONTEND=noninteractive +# 安装 ffmpeg, curl 及 OpenCV 基础依赖 (使用华为云 apt 源) +RUN sed -i 's/deb.debian.org/repo.huaweicloud.com/g' /etc/apt/sources.list && \ + sed -i 's/security.debian.org/repo.huaweicloud.com/g' /etc/apt/sources.list && \ + apt-get update && apt-get install -y --no-install-recommends \ + ffmpeg curl libgl1 libglib2.0-0 \ + && rm -rf /var/lib/apt/lists/* + +WORKDIR /opt/tianyan + +# 1. 复制已编译的 Go 二进制 (由宿主机构建) +COPY build/edge-agent /opt/tianyan/edge-agent + +# 2. 安装 Python 依赖 (使用华为云 pip 源) +COPY python/requirements.txt . +RUN pip3 install --no-cache-dir -r requirements.txt -i https://repo.huaweicloud.com/repository/pypi/simple \ + && rm -rf /root/.cache/pip + +# 3. 复制业务代码与脚本 +COPY python/ /opt/tianyan/python/ +COPY scripts/ /opt/tianyan/scripts/ + +# 4. 创建运行时目录 +RUN mkdir -p /opt/tianyan/config /opt/tianyan/model /tmp + +# 环境变量 (CANN 路径将由 docker-compose 挂载覆盖) +ENV PYTHONPATH=/usr/local/Ascend/ascend-toolkit/latest/python/site-packages +ENV LD_LIBRARY_PATH=/usr/local/Ascend/ascend-toolkit/latest/lib64:/usr/local/Ascend/driver/lib64 +ENV PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin + +# 启动脚本 +COPY entrypoint.sh /opt/tianyan/entrypoint.sh +RUN chmod +x /opt/tianyan/entrypoint.sh + +ENTRYPOINT ["/opt/tianyan/entrypoint.sh"] diff --git a/data/config/edge.yaml b/data/config/edge.yaml new file mode 100644 index 0000000..bbe2000 --- /dev/null +++ b/data/config/edge.yaml @@ -0,0 +1,16 @@ +device_uuid: 8541db9f77826e39605ef2c032f8fb93 +edge_id: edge-demo-001 +cloud_url: http://101.36.73.102:8004 +mqtt_broker: tcp://101.36.73.102:1883 +mqtt_user: "" +mqtt_pass: "" +edge_token: "be079dc5ec8d7c3796d052f3c10143cd" +rtsp_urls: + - "http://101.36.73.102:8080/rtp/34020000002000000003_34020000001310000001.live.flv?secret=be079dc5ec8d7c3796d052f3c10143cd" +infer_socket: /tmp/edge-infer.sock +infer_fps: 2 +infer_workers: 3 +conf_threshold: 0.2 +dedup_window_sec: 30 +ota_url: http://101.36.73.102:8087 +version: 1.0.0 diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..fd1f015 --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,34 @@ +version: '3.8' + +services: + tianyan-edge: + build: + context: . + dockerfile: Dockerfile + image: tianyan-edge:latest + container_name: tianyan-edge + restart: unless-stopped + network_mode: "host" + pid: "host" + + # 显式映射 NPU 设备 + devices: + - /dev/davinci0 + - /dev/davinci_manager + - /dev/devmm_svm + - /dev/hisi_hdc + + volumes: + # 🔑 核心：挂载宿主机 CANN 驱动与库 + - /usr/local/Ascend:/usr/local/Ascend:ro + + # 挂载配置与模型 + - ./data/config:/opt/tianyan/config + - ./data/model:/opt/tianyan/model + + # 共享 Unix Socket 与日志 + - /tmp:/tmp + - ./data/logs:/var/log/tianyan + + environment: + - TZ=Asia/Shanghai diff --git a/entrypoint.sh b/entrypoint.sh new file mode 100644 index 0000000..3c28811 --- /dev/null +++ b/entrypoint.sh @@ -0,0 +1,41 @@ +#!/bin/bash +set -e + +echo "🚀 Starting Tianyan Edge Container..." + +# 1. 启动 NPU 推理服务 (后台) +echo "[1/2] Starting infer_server.py..." +cd /opt/tianyan +python3 python/infer_server.py > /var/log/tianyan/infer.log 2>&1 & +INFER_PID=$! + +# 2. 等待 Unix Socket 创建 (最多 20s) +echo "⏳ Waiting for inference socket at $INFER_SOCKET..." +for i in $(seq 1 20); do + if [ -S "${INFER_SOCKET:-/tmp/edge-infer.sock}" ]; then + echo "✅ Socket ready. Starting edge-agent..." + break + fi + sleep 1 +done + +if [ ! -S "${INFER_SOCKET:-/tmp/edge-infer.sock}" ]; then + echo "❌ ERROR: Inference socket not created. Check NPU/CANN status." + cat /var/log/tianyan/infer.log + kill $INFER_PID 2>/dev/null + exit 1 +fi + +# 3. 启动 Go 边缘代理 +/opt/tianyan/edge-agent -config /opt/tianyan/config/edge.yaml > /var/log/tianyan/agent.log 2>&1 & +AGENT_PID=$! + +# 4. 信号捕获与优雅退出 +trap "echo 'Shutting down...'; kill $INFER_PID $AGENT_PID 2>/dev/null; wait; exit 0" SIGINT SIGTERM + +# 阻塞主进程 +wait -n +EXIT_CODE=$? +echo "Process exited with code $EXIT_CODE" +kill $INFER_PID $AGENT_PID 2>/dev/null +exit $EXIT_CODE diff --git a/internal/event/uploader.go b/internal/event/uploader.go index 70f3872..a6447a4 100644 --- a/internal/event/uploader.go +++ b/internal/event/uploader.go @@ -33,8 +33,15 @@ func (u *Uploader) Run(ctx context.Context) { case <-ctx.Done(): return case ev := <-u.events: - if !u.upload(client, ev) { - u.buffer(ev) + if err := u.upload(client, ev); err != nil { + if ue, ok := err.(*UploadError); ok && ue.Fatal { + log.Printf("uploader: FATAL: %s. Clearing buffer and stopping.", ue.Msg) + u.buf = nil // Drop all pending events + // Give some time before returning or loop with delay + time.Sleep(1 * time.Minute) + } else { + u.buffer(ev) + } } case <-ticker.C: u.flushBuffer(client) @@ -53,14 +60,26 @@ func (u *Uploader) buffer(ev SuspectedEvent) { func (u *Uploader) flushBuffer(client *http.Client) { remaining := u.buf[:0] for _, ev := range u.buf { - if !u.upload(client, ev) { + if err := u.upload(client, ev); err != nil { + if ue, ok := err.(*UploadError); ok && ue.Fatal { + log.Printf("uploader: FATAL flush: %s. Dropping remaining buffer.", ue.Msg) + u.buf = nil + return + } remaining = append(remaining, ev) } } u.buf = remaining } -func (u *Uploader) upload(client *http.Client, ev SuspectedEvent) bool { +type UploadError struct { + Fatal bool + Msg string +} + +func (e *UploadError) Error() string { return e.Msg } + +func (u *Uploader) upload(client *http.Client, ev SuspectedEvent) error { url := fmt.Sprintf("%s/api/v1/edge/events/suspected", u.cfg.CloudURL) body, _ := json.Marshal(ev) req, _ := http.NewRequest(http.MethodPost, url, bytes.NewReader(body)) @@ -69,12 +88,18 @@ func (u *Uploader) upload(client *http.Client, ev SuspectedEvent) bool { resp, err := client.Do(req) if err != nil { log.Printf("uploader: upload error: %v", err) - return false + return nil // Network error, retryable } defer resp.Body.Close() - ok := resp.StatusCode == 200 || resp.StatusCode == 201 - if !ok { - log.Printf("uploader: upload failed status=%d", resp.StatusCode) + + if resp.StatusCode == 401 || resp.StatusCode == 403 { + return &UploadError{Fatal: true, Msg: fmt.Sprintf("unauthorized (status %d), check token", resp.StatusCode)} } - return ok + + if resp.StatusCode == 200 || resp.StatusCode == 201 { + return nil + } + + log.Printf("uploader: upload failed status=%d", resp.StatusCode) + return nil // Server error, retryable }