From 645e0b2e18ca60d6ca162470125822b90c0c6ea0 Mon Sep 17 00:00:00 2001 From: scttbnsn <80784472+scttbnsn@users.noreply.github.com> Date: Tue, 28 Jul 2026 15:05:57 -0400 Subject: [PATCH 1/2] =?UTF-8?q?=E2=9C=A8=20feat(agent):=20graduate=20Portw?= =?UTF-8?q?ing=20edge=20streaming?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../workflows/quality-portwing-fleet-soak.yml | 176 +++++ CHANGELOG.md | 9 + README.de.md | 2 +- README.es.md | 2 +- README.fr.md | 2 +- README.md | 2 +- README.pl.md | 2 +- README.pt-BR.md | 2 +- README.zh-CN.md | 2 +- app/agent/EdgeAgentAdapter.test.ts | 425 ++++++++++++ app/agent/EdgeAgentAdapter.ts | 163 +++++ app/api/api.ts | 4 +- app/api/container/log-stream.test.ts | 441 +++++++++++++ app/api/container/log-stream.ts | 166 +++++ app/api/index.test.ts | 6 +- app/api/index.ts | 6 +- app/api/openapi/index.ts | 3 +- app/api/openapi/paths/portwing.test.ts | 18 +- app/api/openapi/paths/portwing.ts | 10 +- app/configuration/index.test.ts | 8 +- app/configuration/index.ts | 6 +- content/docs/current/api/index.mdx | 6 +- content/docs/current/api/portwing.mdx | 42 +- .../current/configuration/agents/index.mdx | 17 +- scripts/portwing-fleet-soak.mjs | 614 ++++++++++++++++++ 25 files changed, 2085 insertions(+), 49 deletions(-) create mode 100644 .github/workflows/quality-portwing-fleet-soak.yml create mode 100644 scripts/portwing-fleet-soak.mjs diff --git a/.github/workflows/quality-portwing-fleet-soak.yml b/.github/workflows/quality-portwing-fleet-soak.yml new file mode 100644 index 000000000..4b53ded20 --- /dev/null +++ b/.github/workflows/quality-portwing-fleet-soak.yml @@ -0,0 +1,176 @@ +name: "⏱️ Quality: Portwing Fleet Soak" +run-name: >- + ${{ + github.event_name == 'schedule' && '⏱️ Portwing Fleet Soak — Weekly' || + github.event_name == 'pull_request' && format('⏱️ Portwing Fleet Soak — PR #{0}', github.event.pull_request.number) || + format('⏱️ Portwing Fleet Soak — Manual by {0}', github.actor) + }} + +on: + workflow_dispatch: + inputs: + duration: + description: "Soak duration (for example 5m or 1h)" + required: false + default: "5m" + agents: + description: "Number of real Portwing processes" + required: false + default: "8" + exec_per_agent: + description: "Concurrent exec sessions per agent" + required: false + default: "3" + pull_request: + branches: [main, 'dev/**'] + paths: + - '.github/workflows/quality-portwing-fleet-soak.yml' + - 'app/agent/**' + - 'app/api/portwing-ws.ts' + - 'app/api/container/log-stream.ts' + - 'app/configuration/**' + - 'scripts/portwing-fleet-soak.mjs' + schedule: + - cron: '45 7 * * 0' + +permissions: + contents: read + +concurrency: + group: portwing-fleet-soak-${{ github.head_ref || github.ref_name }} + cancel-in-progress: true + +jobs: + fleet-soak: + name: "⏱️ Cross-repo fleet, exec, reconnect, backpressure" + runs-on: ubuntu-latest + timeout-minutes: 45 + + steps: + - name: Harden Runner + uses: step-security/harden-runner@bf7454d06d71f1098171f2acdf0cd4708d7b5920 # v2.20.0 + with: + egress-policy: audit + + - name: Checkout Drydock + uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3 + with: + persist-credentials: false + path: drydock + + - name: Checkout Portwing + uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3 + with: + repository: CodesWhat/portwing + ref: main + persist-credentials: false + path: portwing + + - name: Setup Node.js + uses: actions/setup-node@48b55a011bda9f5d6aeb4c2d9c7362e8dae4041e # v6.4.0 + with: + node-version: 24 + package-manager-cache: false + + - name: Setup Go + uses: actions/setup-go@4a3601121dd01d1626a1e23e37211e3254c1c06c # v6.4.0 + with: + go-version-file: portwing/go.mod + + - name: Install and build Drydock + working-directory: drydock/app + run: | + set -euo pipefail + npm ci + npm run build + + - name: Build Portwing and mock Docker + working-directory: portwing + run: | + set -euo pipefail + go build -o "${RUNNER_TEMP}/portwing" ./cmd/portwing + go build -o "${RUNNER_TEMP}/mockdocker" ./benchmarks/cmd/mockdocker + + - name: Resolve soak parameters + id: parameters + env: + INPUT_DURATION: ${{ github.event.inputs.duration }} + INPUT_AGENTS: ${{ github.event.inputs.agents }} + INPUT_EXEC_PER_AGENT: ${{ github.event.inputs.exec_per_agent }} + run: | + set -euo pipefail + if [ "${GITHUB_EVENT_NAME}" = "schedule" ]; then + duration="30m" + agents="24" + exec_per_agent="4" + rss_threshold="268435456" + heap_threshold="134217728" + elif [ "${GITHUB_EVENT_NAME}" = "pull_request" ]; then + duration="45s" + agents="8" + exec_per_agent="3" + rss_threshold="134217728" + heap_threshold="67108864" + else + duration="${INPUT_DURATION:-5m}" + agents="${INPUT_AGENTS:-8}" + exec_per_agent="${INPUT_EXEC_PER_AGENT:-3}" + rss_threshold="268435456" + heap_threshold="134217728" + fi + { + echo "duration=${duration}" + echo "agents=${agents}" + echo "exec_per_agent=${exec_per_agent}" + echo "rss_threshold=${rss_threshold}" + echo "heap_threshold=${heap_threshold}" + } >> "${GITHUB_OUTPUT}" + + - name: Run real-process fleet soak + env: + SOAK_DURATION: ${{ steps.parameters.outputs.duration }} + SOAK_AGENTS: ${{ steps.parameters.outputs.agents }} + SOAK_EXEC_PER_AGENT: ${{ steps.parameters.outputs.exec_per_agent }} + SOAK_RSS_THRESHOLD: ${{ steps.parameters.outputs.rss_threshold }} + SOAK_HEAP_THRESHOLD: ${{ steps.parameters.outputs.heap_threshold }} + run: | + set -euo pipefail + node --expose-gc drydock/scripts/portwing-fleet-soak.mjs \ + --portwing-bin "${RUNNER_TEMP}/portwing" \ + --mockdocker-bin "${RUNNER_TEMP}/mockdocker" \ + --portwing-repo portwing \ + --duration "${SOAK_DURATION}" \ + --agents "${SOAK_AGENTS}" \ + --exec-per-agent "${SOAK_EXEC_PER_AGENT}" \ + --rss-growth-threshold-bytes "${SOAK_RSS_THRESHOLD}" \ + --heap-growth-threshold-bytes "${SOAK_HEAP_THRESHOLD}" \ + --output artifacts/portwing-fleet-soak.json + + - name: Upload fleet-soak evidence + if: always() + uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 + with: + name: portwing-fleet-soak-${{ github.run_id }}-${{ github.run_attempt }} + path: artifacts/portwing-fleet-soak.json + if-no-files-found: error + retention-days: 90 + + - name: Summarize + if: always() + run: | + set -euo pipefail + if [ ! -f artifacts/portwing-fleet-soak.json ]; then + echo "### Portwing fleet soak" >> "${GITHUB_STEP_SUMMARY}" + echo "No evidence file was produced." >> "${GITHUB_STEP_SUMMARY}" + exit 0 + fi + { + echo "### Portwing fleet soak" + echo "- Result: $(jq -r 'if .passed then "PASS" else "FAIL" end' artifacts/portwing-fleet-soak.json)" + echo "- Agents: $(jq -r '.configuration.agents' artifacts/portwing-fleet-soak.json)" + echo "- Duration: $(jq -r '.configuration.duration' artifacts/portwing-fleet-soak.json)" + echo "- Exec sessions: $(jq -r '.phases.execSessionsStarted' artifacts/portwing-fleet-soak.json)" + echo "- Reconnects observed: $(jq -r '.phases.reconnectsObserved' artifacts/portwing-fleet-soak.json)" + echo "- Agent RSS growth: $(jq -r '.resources.agentRss.growth' artifacts/portwing-fleet-soak.json) bytes" + echo "- Controller heap growth: $(jq -r '.resources.controllerHeap.growth' artifacts/portwing-fleet-soak.json) bytes" + } >> "${GITHUB_STEP_SUMMARY}" diff --git a/CHANGELOG.md b/CHANGELOG.md index 8e151b987..ee2fd7963 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- **Continuous Portwing edge log streams.** The authenticated container-log WebSocket now bridges correlated `dd:container_log_chunk`, `dd:container_log_end`, and `dd:container_log_error` frames, cancels the agent stream when the viewer closes, preserves stdout/stderr and timestamp decoding, and caps each downstream viewer at 1 MiB of buffered data. Older Portwing agents degrade to their one-shot response without breaking the viewer. +- **Real Portwing fleet-soak workflow.** Pull requests and the weekly quality schedule run actual signed Portwing processes against Drydock's production edge gateway through concurrent exec, sustained logs, forced controller backpressure, and reconnect storms. Machine-readable evidence records RSS/heap budgets and is retained for 90 days. + +### Changed + +- **The `portwing/1.0` edge endpoint is enabled by default.** `DD_EXPERIMENTAL_PORTWING=false` remains as an emergency disable for new edge connections. OpenAPI, operator docs, and translated feature summaries now describe the stable/default-on behavior. + ## [1.6.0-rc.8] — 2026-07-28 ### Fixed diff --git a/README.de.md b/README.de.md index 904539927..7da3436d0 100644 --- a/README.de.md +++ b/README.de.md @@ -251,7 +251,7 @@ Die meisten Tools erzwingen einen Kompromiss. Die Auto-Updater (Watchtower, Ouro | 🪝 | **Lebenszyklus-Hooks** | Shell-Befehle vor und nach dem Update über Container-Labels, mit Zeitüberschreitungen pro Hook und Steuerung des Abbruchs bei Fehler. | | 🗂️ | **Docker Compose-Updates** | Ziehen Sie Compose-Dienste über die Docker Engine-API mit YAML-erhaltendem Image-Patching ab und erstellen Sie sie neu. | | 🎛️ | **Richtlinie pro Container** | Regex-Tag-Regeln und Trigger-Routing verwenden `dd.*`-Labels; Reifegrenzen, Skip/Snooze/Pin und Wartungsfenster werden über die Benutzeroberfläche/API oder die Watcher-Konfiguration gespeichert. | -| 🛰️ | **Verteilte Agenten** | Überwachen Sie Remote-Docker-Hosts über SSE. Edge-Agents hinter NAT wählen sich über WebSocket mit Ed25519-Schlüsselauthentifizierung aus, kein eingehender Port erforderlich (`DD_EXPERIMENTAL_PORTWING=true`). | +| 🛰️ | **Verteilte Agenten** | Überwachen Sie Remote-Docker-Hosts über SSE. Portwing-Edge-Agents hinter NAT wählen sich über WebSocket mit Ed25519-Schlüsselauthentifizierung und kontinuierlichen Live-Protokollen ein; kein eingehender Port ist erforderlich. Der Endpunkt ist standardmäßig aktiviert; `DD_EXPERIMENTAL_PORTWING=false` ist die Notabschaltung. | | 🖥️ | **Web-Dashboard** | Vue 3-Benutzeroberfläche mit einem anpassbaren Widget-Raster ohne Abhängigkeiten, reaktionsfähigen Tabellen-/Kartenansichten, Live-SSE-Updates, Steuerelementen für Benachrichtigungsglocken sowie Details, Protokollen und Statistiken pro Container. | | 🔗 | **REST-API und Webhooks** | Token-authentifizierte Endpunkte für CI/CD-Überwachungs- und Update-Trigger sowie signierte Registrierungs-Webhook-Aufnahme für Push-Ereignisse. | | 🔐 | **OIDC-Authentifizierung** | Sichern Sie das Dashboard mit OpenID Connect (Authelia, Auth0, Authentik). Alle Authentifizierungsflüsse werden standardmäßig nicht geschlossen. | diff --git a/README.es.md b/README.es.md index cbe4b15f3..14dbab4ce 100644 --- a/README.es.md +++ b/README.es.md @@ -251,7 +251,7 @@ La mayoría de las herramientas obligan a hacer concesiones. Los actualizadores | 🪝 | **Ganchos de ciclo de vida** | Comandos de shell previos y posteriores a la actualización a través de etiquetas de contenedor, con tiempos de espera por gancho y control de cancelación en caso de falla. | | 🗂️ | **Actualizaciones Docker Compose** | Extraiga y vuelva a crear servicios de Compose a través de la API Docker Engine con parches de imágenes que preservan YAML. | | 🎛️ | **Política por contenedor** | Las reglas de etiquetas Regex y el enrutamiento de activación utilizan etiquetas `dd.*`; Las puertas de madurez, saltar/posponer/fijar y las ventanas de mantenimiento se almacenan a través de UI/API o la configuración del observador. | -| 🛰️ | **Agentes distribuidos** | Supervise hosts Docker remotos a través de SSE. Los agentes perimetrales detrás de NAT marcan a través de WebSocket con autenticación de clave Ed25519, no se requiere puerto de entrada (`DD_EXPERIMENTAL_PORTWING=true`). | +| 🛰️ | **Agentes distribuidos** | Supervise hosts Docker remotos a través de SSE. Los agentes Portwing detrás de NAT se conectan por WebSocket con autenticación Ed25519 y registros continuos en vivo, sin necesidad de un puerto de entrada. El endpoint está habilitado de forma predeterminada; `DD_EXPERIMENTAL_PORTWING=false` es la desactivación de emergencia. | | 🖥️ | **Panel web** | Interfaz de usuario de Vue 3 con una cuadrícula de widgets personalizable sin dependencia, vistas de tablas/tarjetas responsivas, actualizaciones SSE en vivo, controles de campana de notificación y detalles, registros y estadísticas por contenedor. | | 🔗 | **API REST y webhooks** | Puntos finales autenticados por token para activación de actualizaciones y vigilancia de CI/CD, además de ingesta de webhooks de registro firmados para eventos push. | | 🔐 | **Autenticación OIDC** | Asegure el tablero con OpenID Connect (Authelia, Auth0, Authentik). Todos los flujos de autenticación fallan al cerrarse de forma predeterminada. | diff --git a/README.fr.md b/README.fr.md index 173a132b8..bcd5221c7 100644 --- a/README.fr.md +++ b/README.fr.md @@ -251,7 +251,7 @@ La plupart des outils imposent un compromis. Les mises à jour automatiques (Wat | 🪝 | **Crochets de cycle de vie** | Commandes shell avant et après la mise à jour via des étiquettes de conteneur, avec délais d'attente par hook et contrôle d'abandon en cas d'échec. | | 🗂️ | **Mises à jour Docker Compose** | Extrayez et recréez les services Compose via l'API Docker Engine avec des correctifs d'image préservant YAML. | | 🎛️ | **Politique par conteneur** | Les règles de balise Regex et le routage des déclencheurs utilisent les étiquettes `dd.*` ; les portes de maturité, les sauts/répétitions/épingles et les fenêtres de maintenance sont stockés via l'interface utilisateur/API ou la configuration de l'observateur. | -| 🛰️ | **Agents distribués** | Surveillez les hôtes Docker distants via SSE. Les agents Edge derrière NAT appellent via WebSocket avec l'authentification par clé Ed25519, aucun port entrant requis (`DD_EXPERIMENTAL_PORTWING=true`). | +| 🛰️ | **Agents distribués** | Surveillez les hôtes Docker distants via SSE. Les agents Portwing derrière un NAT se connectent via WebSocket avec authentification Ed25519 et journaux continus en direct, sans port entrant. Le point de terminaison est activé par défaut ; `DD_EXPERIMENTAL_PORTWING=false` est la désactivation d'urgence. | | 🖥️ | **Tableau de bord Web** | Interface utilisateur Vue 3 avec une grille de widgets personnalisable sans dépendance, des vues de table/carte réactives, des mises à jour SSE en direct, des contrôles de cloche de notification et des détails, journaux et statistiques par conteneur. | | 🔗 | **API REST et webhooks** | Points de terminaison authentifiés par jeton pour les déclencheurs de surveillance et de mise à jour CI/CD, ainsi que l'ingestion de webhooks de registre signés pour les événements push. | | 🔐 | **Authentification OIDC** | Sécurisez le tableau de bord avec OpenID Connect (Authelia, Auth0, Authentik). Tous les flux d'authentification échouent à la fermeture par défaut. | diff --git a/README.md b/README.md index f7cc5bd29..2f8cbc86b 100644 --- a/README.md +++ b/README.md @@ -257,7 +257,7 @@ Most tools force a tradeoff. The auto-updaters (Watchtower, Ouroboros) pull and | 🪝 | **Lifecycle Hooks** | Pre and post-update shell commands via container labels, with per-hook timeouts and abort-on-failure control. | | 🗂️ | **Docker Compose Updates** | Pull and recreate Compose services through the Docker Engine API with YAML-preserving image patching. | | 🎛️ | **Per-Container Policy** | Regex tag rules and trigger routing use `dd.*` labels; maturity gates, skip/snooze/pin, and maintenance windows are stored via UI/API or watcher configuration. | -| 🛰️ | **Distributed Agents** | Monitor remote Docker hosts over SSE. Edge agents behind NAT dial out over WebSocket with Ed25519 key auth, no inbound port required (`DD_EXPERIMENTAL_PORTWING=true`). | +| 🛰️ | **Distributed Agents** | Monitor remote Docker hosts over SSE. Portwing edge agents behind NAT dial out over WebSocket with Ed25519 key auth and continuous live logs, with no inbound port required. The endpoint is enabled by default; `DD_EXPERIMENTAL_PORTWING=false` is an emergency disable. | | 🖥️ | **Web Dashboard** | Vue 3 UI with a zero-dependency customizable widget grid, responsive table/card views, live SSE updates, notification-bell controls, and per-container detail, logs, and stats. | | 🔗 | **REST API & Webhooks** | Token-authenticated endpoints for CI/CD watch and update triggers, plus signed registry webhook ingestion for push events. | | 🔐 | **OIDC Authentication** | Secure the dashboard with OpenID Connect (Authelia, Auth0, Authentik). All auth flows fail closed by default. | diff --git a/README.pl.md b/README.pl.md index 0b693e3e2..8485077b4 100644 --- a/README.pl.md +++ b/README.pl.md @@ -251,7 +251,7 @@ Większość narzędzi wymusza kompromis. Automatyczne aktualizacje (Watchtower, | 🪝 | **Haki cyklu życia** | Polecenia powłoki przed i po aktualizacji za pośrednictwem etykiet kontenerów, z limitami czasu dla poszczególnych haków i kontrolą przerwania w przypadku awarii. | | 🗂️ | **Aktualizacje Docker Compose** | Pobieraj i odtwarzaj usługi Compose za pośrednictwem interfejsu API Docker Engine z łataniem obrazów zachowującym YAML. | | 🎛️ | **Zasady dotyczące kontenera** | Reguły tagów Regex i routing wyzwalaczy korzystają z etykiet `dd.*`; bramki dojrzałości, pomijanie/odkładanie/przypinanie i okna konserwacji są przechowywane za pośrednictwem interfejsu użytkownika/API lub konfiguracji obserwatora. | -| 🛰️ | **Agenci rozproszoni** | Monitoruj zdalne hosty Dockera za pośrednictwem SSE. Agenci brzegowi za NAT wybierają numer przez WebSocket z uwierzytelnianiem za pomocą klucza Ed25519, nie jest wymagany port wejściowy (`DD_EXPERIMENTAL_PORTWING=true`). | +| 🛰️ | **Agenci rozproszoni** | Monitoruj zdalne hosty Dockera za pośrednictwem SSE. Agenci Portwing za NAT łączą się przez WebSocket z uwierzytelnianiem Ed25519 i ciągłymi logami na żywo, bez portu wejściowego. Punkt końcowy jest domyślnie włączony; `DD_EXPERIMENTAL_PORTWING=false` służy do awaryjnego wyłączenia. | | 🖥️ | **Panel sieciowy** | Interfejs użytkownika Vue 3 z konfigurowalną siatką widżetów o zerowej zależności, responsywnymi widokami tabel/kart, aktualizacjami SSE na żywo, sterowaniem dzwonkiem powiadomień oraz szczegółami, dziennikami i statystykami dotyczącymi poszczególnych kontenerów. | | 🔗 | **REST API i webhooki** | Punkty końcowe uwierzytelniane tokenem dla wyzwalaczy monitorowania i aktualizacji CI/CD oraz pozyskiwania podpisanego elementu webhook rejestru dla zdarzeń push. | | 🔐 | **Uwierzytelnianie OIDC** | Zabezpiecz deskę rozdzielczą za pomocą OpenID Connect (Authelia, Auth0, Authentik). Domyślnie wszystkie przepływy uwierzytelniania nie są zamykane. | diff --git a/README.pt-BR.md b/README.pt-BR.md index e9a0c7e5f..56b37ee7a 100644 --- a/README.pt-BR.md +++ b/README.pt-BR.md @@ -251,7 +251,7 @@ A maioria das ferramentas força uma compensação. Os atualizadores automático | 🪝 | **Ganchos de ciclo de vida** | Comandos shell pré e pós-atualização por meio de rótulos de contêiner, com tempos limite por gancho e controle de aborto em caso de falha. | | 🗂️ | **Atualizações Docker Compose** | Extraia e recrie serviços do Compose por meio da API Docker Engine com patch de imagem com preservação de YAML. | | 🎛️ | **Política por contêiner** | As regras de tag Regex e o roteamento de gatilho usam rótulos `dd.*`; portas de maturidade, pular/adiar/fixar e janelas de manutenção são armazenadas via UI/API ou configuração do inspetor. | -| 🛰️ | **Agentes Distribuídos** | Monitore hosts Docker remotos por SSE. Agentes de borda por trás da discagem NAT por WebSocket com autenticação de chave Ed25519, sem necessidade de porta de entrada (`DD_EXPERIMENTAL_PORTWING=true`). | +| 🛰️ | **Agentes Distribuídos** | Monitore hosts Docker remotos por SSE. Agentes Portwing atrás de NAT conectam-se por WebSocket com autenticação Ed25519 e logs contínuos ao vivo, sem porta de entrada. O endpoint é habilitado por padrão; `DD_EXPERIMENTAL_PORTWING=false` é a desativação de emergência. | | 🖥️ | **Painel Web** | UI Vue 3 com uma grade de widget personalizável de dependência zero, visualizações responsivas de tabela/cartão, atualizações SSE ao vivo, controles de sino de notificação e detalhes, registros e estatísticas por contêiner. | | 🔗 | **API REST e webhooks** | Endpoints autenticados por token para monitoramento de CI/CD e gatilhos de atualização, além de ingestão de webhook de registro assinado para eventos push. | | 🔐 | **Autenticação OIDC** | Proteja o painel com OpenID Connect (Authelia, Auth0, Authentik). Todos os fluxos de autenticação falham quando fechados por padrão. | diff --git a/README.zh-CN.md b/README.zh-CN.md index 201db59e7..14bfa3073 100644 --- a/README.zh-CN.md +++ b/README.zh-CN.md @@ -251,7 +251,7 @@ docker run -d \ | 🪝 | **生命周期挂钩** |通过容器标签执行更新前和更新后的 shell 命令，并具有每个钩子超时和失败时中止控制。 | | 🗂️ | **Docker Compose 更新** |通过 Docker Engine API 以及保留 YAML 的映像修补来拉取并重新创建 Compose 服务。 | | 🎛️ | **每个容器的政策** |正则表达式标签规则和触发路由使用`dd.*`标签；成熟度门、跳过/暂停/固定和维护窗口通过 UI/API 或观察者配置存储。 | -| 🛰️ | **分布式代理** |通过 SSE 监控远程 Docker 主机。 NAT 后面的边缘代理使用 Ed25519 密钥身份验证通过 WebSocket 拨出，无需入站端口 (`DD_EXPERIMENTAL_PORTWING=true`)。 | +| 🛰️ | **分布式代理** |通过 SSE 监控远程 Docker 主机。NAT 后面的 Portwing 边缘代理使用 Ed25519 密钥身份验证通过 WebSocket 主动连接，并支持连续实时日志，无需入站端口。该端点默认启用；`DD_EXPERIMENTAL_PORTWING=false` 仅用于紧急禁用。 | | 🖥️ | **网络仪表板** | Vue 3 UI 具有零依赖可定制小部件网格、响应式表格/卡片视图、实时 SSE 更新、通知铃控件以及每个容器的详细信息、日志和统计信息。 | | 🔗 | **REST API 和 Webhook** |用于 CI/CD 监视和更新触发器的令牌身份验证端点，以及用于推送事件的签名注册表 Webhook 摄取。 | | 🔐 | **OIDC 身份验证** |使用 OpenID Connect（Authelia、Auth0、Authentik）保护仪表板。默认情况下，所有身份验证流程都会失败关闭。 | diff --git a/app/agent/EdgeAgentAdapter.test.ts b/app/agent/EdgeAgentAdapter.test.ts index 99018cd96..f78e3ab6b 100644 --- a/app/agent/EdgeAgentAdapter.test.ts +++ b/app/agent/EdgeAgentAdapter.test.ts @@ -811,6 +811,431 @@ describe('EdgeAgentAdapter — requestContainerLogs', () => { }); }); +describe('EdgeAgentAdapter — continuous container log streams', () => { + beforeEach(() => { + vi.clearAllMocks(); + }); + + test('opens the dd streaming variant and delivers correlated chunks through completion', async () => { + const { adapter, ws } = createAdapter(); + adapter.activate(); + const chunks: Array<{ stream: string; logs: string }> = []; + const ended = vi.fn(); + const failed = vi.fn(); + const streamMethod = ( + adapter as unknown as { + streamContainerLogs?: ( + containerId: string, + options: Record, + handlers: { + onChunk: (chunk: { stream: string; logs: string }) => void; + onEnd: (reason?: string) => void; + onError: (error: Error) => void; + }, + ) => { cancel: () => void }; + } + ).streamContainerLogs; + + expect(typeof streamMethod).toBe('function'); + if (!streamMethod) return; + + streamMethod.call( + adapter, + 'container-live', + { tail: 25, since: 42, follow: true, timestamps: true }, + { + onChunk: (chunk) => chunks.push(chunk), + onEnd: ended, + onError: failed, + }, + ); + + const request = JSON.parse(ws.sentMessages.at(-1) ?? '{}') as { + type: string; + data: { requestId: string; stream: boolean; follow: boolean; containerId: string }; + }; + expect(request.type).toBe('dd:container_log_request'); + expect(request.data).toMatchObject({ + containerId: 'container-live', + stream: true, + follow: true, + }); + + sendFrame(ws, 'dd:container_log_chunk', { + requestId: request.data.requestId, + containerId: 'container-live', + stream: 'stderr', + logs: '2026-07-28T18:00:00Z warning\n', + }); + sendFrame(ws, 'dd:container_log_end', { + requestId: request.data.requestId, + containerId: 'container-live', + reason: 'eof', + }); + await new Promise((resolve) => setTimeout(resolve, 0)); + + expect(chunks).toEqual([ + { + stream: 'stderr', + logs: '2026-07-28T18:00:00Z warning\n', + }, + ]); + expect(ended).toHaveBeenCalledWith('eof'); + expect(failed).not.toHaveBeenCalled(); + }); + + test('cancels one live dd log stream without disturbing another request', () => { + const { adapter, ws } = createAdapter(); + adapter.activate(); + const streamMethod = ( + adapter as unknown as { + streamContainerLogs?: ( + containerId: string, + options: Record, + handlers: { onChunk: () => void; onEnd: () => void; onError: () => void }, + ) => { cancel: () => void }; + } + ).streamContainerLogs; + + expect(typeof streamMethod).toBe('function'); + if (!streamMethod) return; + + const handle = streamMethod.call( + adapter, + 'container-cancel', + {}, + { onChunk: vi.fn(), onEnd: vi.fn(), onError: vi.fn() }, + ); + const request = JSON.parse(ws.sentMessages.at(-1) ?? '{}') as { + data: { requestId: string }; + }; + + handle.cancel(); + + expect(JSON.parse(ws.sentMessages.at(-1) ?? '{}')).toEqual({ + type: 'dd:container_log_cancel', + data: { + requestId: request.data.requestId, + containerId: 'container-cancel', + }, + }); + }); + + test('treats a legacy one-shot response as a complete stream fallback', async () => { + const { adapter, ws } = createAdapter(); + adapter.activate(); + const chunks: string[] = []; + const ended = vi.fn(); + const streamMethod = ( + adapter as unknown as { + streamContainerLogs?: ( + containerId: string, + options: Record, + handlers: { + onChunk: (chunk: { logs: string }) => void; + onEnd: (reason?: string) => void; + onError: (error: Error) => void; + }, + ) => { cancel: () => void }; + } + ).streamContainerLogs; + + expect(typeof streamMethod).toBe('function'); + if (!streamMethod) return; + + streamMethod.call( + adapter, + 'legacy-container', + {}, + { + onChunk: (chunk) => chunks.push(chunk.logs), + onEnd: ended, + onError: vi.fn(), + }, + ); + const request = JSON.parse(ws.sentMessages.at(-1) ?? '{}') as { + data: { requestId: string }; + }; + sendFrame(ws, 'dd:container_log_response', { + requestId: request.data.requestId, + containerId: 'legacy-container', + logs: 'legacy buffered logs\n', + }); + await new Promise((resolve) => setTimeout(resolve, 0)); + + expect(chunks).toEqual(['legacy buffered logs\n']); + expect(ended).toHaveBeenCalledWith('legacy-response'); + }); + + test('ends an empty legacy fallback without emitting an empty chunk', async () => { + const { adapter, ws } = createAdapter(); + adapter.activate(); + const chunked = vi.fn(); + const ended = vi.fn(); + adapter.streamContainerLogs( + 'legacy-empty', + {}, + { onChunk: chunked, onEnd: ended, onError: vi.fn() }, + ); + const request = JSON.parse(ws.sentMessages.at(-1) ?? '{}') as { + data: { requestId: string }; + }; + + sendFrame(ws, 'dd:container_log_response', { + requestId: request.data.requestId, + containerId: 'legacy-empty', + logs: '', + }); + await new Promise((resolve) => setTimeout(resolve, 0)); + + expect(chunked).not.toHaveBeenCalled(); + expect(ended).toHaveBeenCalledWith('legacy-response'); + }); + + test('delivers a correlated terminal error and ignores later chunks', async () => { + const { adapter, ws } = createAdapter(); + adapter.activate(); + const chunked = vi.fn(); + const failed = vi.fn(); + const handle = adapter.streamContainerLogs( + 'broken-container', + {}, + { onChunk: chunked, onEnd: vi.fn(), onError: failed }, + ); + const request = JSON.parse(ws.sentMessages.at(-1) ?? '{}') as { + data: { requestId: string }; + }; + + sendFrame(ws, 'dd:container_log_error', { + requestId: request.data.requestId, + containerId: 'broken-container', + error: 'docker stream failed', + }); + sendFrame(ws, 'dd:container_log_chunk', { + requestId: request.data.requestId, + containerId: 'broken-container', + stream: 'stdout', + logs: 'late data', + }); + await new Promise((resolve) => setTimeout(resolve, 0)); + + expect(failed).toHaveBeenCalledWith(new Error('docker stream failed')); + expect(chunked).not.toHaveBeenCalled(); + handle.cancel(); + }); + + test('fails every live log stream when the agent connection closes', async () => { + const { adapter } = createAdapter(); + adapter.activate(); + const firstFailed = vi.fn(); + const secondFailed = vi.fn(); + adapter.streamContainerLogs( + 'first', + {}, + { + onChunk: vi.fn(), + onEnd: vi.fn(), + onError: firstFailed, + }, + ); + adapter.streamContainerLogs( + 'second', + {}, + { + onChunk: vi.fn(), + onEnd: vi.fn(), + onError: secondFailed, + }, + ); + + await adapter.onDisconnect(); + + expect(firstFailed).toHaveBeenCalledWith(new Error('connection closed')); + expect(secondFailed).toHaveBeenCalledWith(new Error('connection closed')); + }); + + test('ignores malformed, uncorrelated, and duplicate terminal stream frames', async () => { + const { adapter, ws } = createAdapter(); + adapter.activate(); + const chunked = vi.fn(); + const ended = vi.fn(); + const failed = vi.fn(); + adapter.streamContainerLogs( + 'strict-container', + {}, + { onChunk: chunked, onEnd: ended, onError: failed }, + ); + const request = JSON.parse(ws.sentMessages.at(-1) ?? '{}') as { + data: { requestId: string }; + }; + + for (const data of [ + {}, + { requestId: request.data.requestId }, + { containerId: 'strict-container' }, + { requestId: request.data.requestId, containerId: 'strict-container' }, + { requestId: 'unknown', containerId: 'strict-container', logs: 'ignored' }, + { requestId: request.data.requestId, containerId: 'other', logs: 'ignored' }, + ]) { + sendFrame(ws, 'dd:container_log_chunk', data); + } + sendFrame(ws, 'dd:container_log_chunk', { + requestId: request.data.requestId, + containerId: 'strict-container', + logs: '', + }); + sendFrame(ws, 'dd:container_log_end', {}); + sendFrame(ws, 'dd:container_log_end', { requestId: request.data.requestId }); + sendFrame(ws, 'dd:container_log_end', { containerId: 'strict-container' }); + sendFrame(ws, 'dd:container_log_end', { + requestId: 'unknown', + containerId: 'strict-container', + }); + sendFrame(ws, 'dd:container_log_end', { + requestId: request.data.requestId, + containerId: 'other', + }); + sendFrame(ws, 'dd:container_log_error', {}); + sendFrame(ws, 'dd:container_log_error', { requestId: request.data.requestId }); + sendFrame(ws, 'dd:container_log_error', { containerId: 'strict-container' }); + sendFrame(ws, 'dd:container_log_error', { + requestId: 'unknown', + containerId: 'strict-container', + }); + sendFrame(ws, 'dd:container_log_error', { + requestId: request.data.requestId, + containerId: 'other', + }); + await new Promise((resolve) => setTimeout(resolve, 0)); + + expect(chunked).toHaveBeenCalledWith({ stream: 'stdout', logs: '' }); + expect(ended).not.toHaveBeenCalled(); + expect(failed).not.toHaveBeenCalled(); + + sendFrame(ws, 'dd:container_log_end', { + requestId: request.data.requestId, + containerId: 'strict-container', + }); + sendFrame(ws, 'dd:container_log_end', { + requestId: request.data.requestId, + containerId: 'strict-container', + }); + sendFrame(ws, 'dd:container_log_error', { + requestId: request.data.requestId, + containerId: 'strict-container', + }); + await new Promise((resolve) => setTimeout(resolve, 0)); + + expect(ended).toHaveBeenCalledWith(undefined); + expect(ended).toHaveBeenCalledTimes(1); + expect(failed).not.toHaveBeenCalled(); + }); + + test.each([ + { thrown: new Error('chunk callback failed'), message: 'chunk callback failed' }, + { thrown: 'non-error callback failure', message: 'non-error callback failure' }, + ])('cancels and reports a throwing chunk handler: $message', async ({ thrown, message }) => { + const { adapter, ws } = createAdapter(); + adapter.activate(); + const failed = vi.fn(); + adapter.streamContainerLogs( + 'throwing-container', + {}, + { + onChunk: () => { + throw thrown; + }, + onEnd: vi.fn(), + onError: failed, + }, + ); + const request = JSON.parse(ws.sentMessages.at(-1) ?? '{}') as { + data: { requestId: string }; + }; + + sendFrame(ws, 'dd:container_log_chunk', { + requestId: request.data.requestId, + containerId: 'throwing-container', + stream: 'stdout', + logs: 'boom', + }); + await new Promise((resolve) => setTimeout(resolve, 0)); + + expect(failed).toHaveBeenCalledWith(new Error(message)); + expect(JSON.parse(ws.sentMessages.at(-1) ?? '{}')).toEqual({ + type: 'dd:container_log_cancel', + data: { + requestId: request.data.requestId, + containerId: 'throwing-container', + }, + }); + }); + + test('uses the default stream error when the agent omits an error string', async () => { + const { adapter, ws } = createAdapter(); + adapter.activate(); + const failed = vi.fn(); + adapter.streamContainerLogs( + 'broken-container', + {}, + { onChunk: vi.fn(), onEnd: vi.fn(), onError: failed }, + ); + const request = JSON.parse(ws.sentMessages.at(-1) ?? '{}') as { + data: { requestId: string }; + }; + + sendFrame(ws, 'dd:container_log_error', { + requestId: request.data.requestId, + containerId: 'broken-container', + }); + await new Promise((resolve) => setTimeout(resolve, 0)); + + expect(failed).toHaveBeenCalledWith(new Error('container log stream failed')); + }); + + test('reports the stream capacity limit and returns an inert cancel handle', () => { + const { adapter, ws } = createAdapter(); + const internals = adapter as unknown as { + pendingRequests: Map; + }; + for (let index = 0; index < 1000; index += 1) { + internals.pendingRequests.set(`occupied-${index}`, {}); + } + const failed = vi.fn(); + const sentBefore = ws.sentMessages.length; + + const handle = adapter.streamContainerLogs( + 'over-capacity', + {}, + { onChunk: vi.fn(), onEnd: vi.fn(), onError: failed }, + ); + handle.cancel(); + + expect(failed).toHaveBeenCalledWith(new Error('concurrent request limit reached')); + expect(ws.sentMessages).toHaveLength(sentBefore); + }); + + test.each([ + { thrown: new Error('send failed'), message: 'send failed' }, + { thrown: 'socket exploded', message: 'socket exploded' }, + ])('reports and cleans up an initial stream send failure: $message', ({ thrown, message }) => { + const { adapter, ws } = createAdapter(); + vi.mocked(ws.send).mockImplementationOnce(() => { + throw thrown; + }); + const failed = vi.fn(); + + const handle = adapter.streamContainerLogs( + 'send-failure', + {}, + { onChunk: vi.fn(), onEnd: vi.fn(), onError: failed }, + ); + handle.cancel(); + + expect(failed).toHaveBeenCalledWith(new Error(message)); + }); +}); + describe('EdgeAgentAdapter — deleteContainer', () => { beforeEach(() => { vi.clearAllMocks(); diff --git a/app/agent/EdgeAgentAdapter.ts b/app/agent/EdgeAgentAdapter.ts index 4013edfbd..f7e46f7e9 100644 --- a/app/agent/EdgeAgentAdapter.ts +++ b/app/agent/EdgeAgentAdapter.ts @@ -59,6 +59,26 @@ interface PendingRequest { chunks?: Buffer[]; } +export interface ContainerLogStreamChunk { + stream: 'stdout' | 'stderr'; + logs: string; +} + +export interface ContainerLogStreamHandlers { + onChunk: (chunk: ContainerLogStreamChunk) => void; + onEnd: (reason?: string) => void; + onError: (error: Error) => void; +} + +export interface ContainerLogStreamHandle { + cancel: () => void; +} + +interface LiveContainerLogStream { + containerId: string; + handlers: ContainerLogStreamHandlers; +} + interface AgentComponentDescriptor { type: string; name: string; @@ -103,6 +123,7 @@ export class EdgeAgentAdapter { private readonly reconnected: boolean; private readonly execSessions: Map; private readonly pendingRequests: Map; + private readonly liveContainerLogStreams: Map; private pingInterval: ReturnType | undefined; private containerSyncWarnTimer: ReturnType | undefined; private connected = false; @@ -139,6 +160,7 @@ export class EdgeAgentAdapter { this.reconnected = options.reconnected ?? false; this.execSessions = new Map(); this.pendingRequests = new Map(); + this.liveContainerLogStreams = new Map(); this.containerRequestQueues = new Map(); this.messageListener = (raw: unknown): void => { @@ -319,6 +341,15 @@ export class EdgeAgentAdapter { case 'dd:container_log_response': this.handleContainerLogResponse(data); return; + case 'dd:container_log_chunk': + this.handleContainerLogChunk(data); + return; + case 'dd:container_log_end': + this.handleContainerLogEnd(data); + return; + case 'dd:container_log_error': + this.handleContainerLogError(data); + return; case 'dd:container_delete_response': this.handleContainerDeleteResponse(data); return; @@ -570,6 +601,16 @@ export class EdgeAgentAdapter { if (!containerId) { return; } + const requestId = typeof data.requestId === 'string' ? data.requestId : undefined; + const liveStream = requestId ? this.liveContainerLogStreams.get(requestId) : undefined; + if (requestId && liveStream?.containerId === containerId) { + this.liveContainerLogStreams.delete(requestId); + if (typeof data.logs === 'string' && data.logs.length > 0) { + liveStream.handlers.onChunk({ stream: 'stdout', logs: data.logs }); + } + liveStream.handlers.onEnd('legacy-response'); + return; + } const pending = this.takePendingContainerResponse('log', containerId, data); if (!pending) { return; @@ -577,6 +618,58 @@ export class EdgeAgentAdapter { pending.resolve(data.logs); } + private handleContainerLogChunk(data: Record): void { + const requestId = typeof data.requestId === 'string' ? data.requestId : undefined; + const containerId = typeof data.containerId === 'string' ? data.containerId : undefined; + const logs = typeof data.logs === 'string' ? data.logs : undefined; + if (!requestId || !containerId || logs === undefined) { + return; + } + const liveStream = this.liveContainerLogStreams.get(requestId); + if (!liveStream || liveStream.containerId !== containerId) { + return; + } + const stream = data.stream === 'stderr' ? 'stderr' : 'stdout'; + try { + liveStream.handlers.onChunk({ stream, logs }); + } catch (error: unknown) { + this.cancelContainerLogStream(requestId, containerId); + liveStream.handlers.onError( + error instanceof Error ? error : new Error(getErrorMessage(error)), + ); + } + } + + private handleContainerLogEnd(data: Record): void { + const requestId = typeof data.requestId === 'string' ? data.requestId : undefined; + const containerId = typeof data.containerId === 'string' ? data.containerId : undefined; + if (!requestId || !containerId) { + return; + } + const liveStream = this.liveContainerLogStreams.get(requestId); + if (!liveStream || liveStream.containerId !== containerId) { + return; + } + this.liveContainerLogStreams.delete(requestId); + liveStream.handlers.onEnd(typeof data.reason === 'string' ? data.reason : undefined); + } + + private handleContainerLogError(data: Record): void { + const requestId = typeof data.requestId === 'string' ? data.requestId : undefined; + const containerId = typeof data.containerId === 'string' ? data.containerId : undefined; + if (!requestId || !containerId) { + return; + } + const liveStream = this.liveContainerLogStreams.get(requestId); + if (!liveStream || liveStream.containerId !== containerId) { + return; + } + this.liveContainerLogStreams.delete(requestId); + liveStream.handlers.onError( + new Error(typeof data.error === 'string' ? data.error : 'container log stream failed'), + ); + } + private handleContainerDeleteResponse(data: Record): void { const containerId = typeof data.containerId === 'string' ? data.containerId : undefined; if (!containerId) { @@ -838,6 +931,72 @@ export class EdgeAgentAdapter { }); } + /** + * Open a continuous, cancellable container-log stream over the Drydock + * namespace. An older Portwing agent may answer with the one-shot + * dd:container_log_response; handleContainerLogResponse converts that into + * one stdout chunk followed by a legacy-response completion. + */ + streamContainerLogs( + containerId: string, + options: { + tail?: number; + since?: string | number; + follow?: boolean; + timestamps?: boolean; + }, + handlers: ContainerLogStreamHandlers, + ): ContainerLogStreamHandle { + if (this.pendingRequests.size + this.liveContainerLogStreams.size >= MAX_PENDING_REQUESTS) { + handlers.onError(new Error('concurrent request limit reached')); + return { cancel: () => {} }; + } + + const requestId = uuidv7(); + this.liveContainerLogStreams.set(requestId, { containerId, handlers }); + try { + this.ws.send( + JSON.stringify({ + type: 'dd:container_log_request', + data: { + containerId, + requestId, + ...options, + follow: options.follow ?? true, + stream: true, + }, + }), + ); + } catch (error: unknown) { + this.liveContainerLogStreams.delete(requestId); + handlers.onError(error instanceof Error ? error : new Error(getErrorMessage(error))); + } + + return { + cancel: () => { + this.cancelContainerLogStream(requestId, containerId); + }, + }; + } + + private cancelContainerLogStream(requestId: string, containerId: string): void { + const liveStream = this.liveContainerLogStreams.get(requestId); + if (!liveStream || liveStream.containerId !== containerId) { + return; + } + this.liveContainerLogStreams.delete(requestId); + try { + this.ws.send( + JSON.stringify({ + type: 'dd:container_log_cancel', + data: { requestId, containerId }, + }), + ); + } catch { + // connection may be closing + } + } + /** * Delete a container on the edge agent. * Returns a promise that resolves once the agent confirms deletion, or @@ -1115,6 +1274,10 @@ export class EdgeAgentAdapter { this.pendingRequests.delete(requestId); } this.containerRequestQueues.clear(); + for (const [requestId, stream] of this.liveContainerLogStreams.entries()) { + stream.handlers.onError(new Error('connection closed')); + this.liveContainerLogStreams.delete(requestId); + } // Close all exec sessions — session.close() sends exec_end before deleting (O5) for (const session of this.execSessions.values()) { diff --git a/app/api/api.ts b/app/api/api.ts index 3b32c6cde..23dd3e8b8 100644 --- a/app/api/api.ts +++ b/app/api/api.ts @@ -189,7 +189,9 @@ export function init(): express.Router { // Mount agents router.use('/agents', agentRouter.init()); - // Mount Portwing key management (edge agent auth registry) — experimental. + // Mount Portwing key management (edge agent auth registry). It is enabled by + // default after edge graduation and can be disabled explicitly for emergency + // rollback with DD_EXPERIMENTAL_PORTWING=false. if (getExperimentalPortwingEnabled()) { router.use('/portwing', portwingRouter.init()); } diff --git a/app/api/container/log-stream.test.ts b/app/api/container/log-stream.test.ts index 7b1c733bb..34840175d 100644 --- a/app/api/container/log-stream.test.ts +++ b/app/api/container/log-stream.test.ts @@ -1412,6 +1412,447 @@ describe('api/container/log-stream', () => { }); }); + describe('edge-agent container log streams', () => { + function createAuthenticatedEdgeGateway( + streamContainerLogs: ReturnType, + viewer: EventEmitter & { + send: ReturnType; + close: ReturnType; + bufferedAmount: number; + }, + agentResolver: + | (() => { edgeAdapter?: { streamContainerLogs: ReturnType } } | undefined) + | null = () => ({ + edgeAdapter: { streamContainerLogs }, + }), + ) { + const dependencies = { + getContainer: vi.fn(() => ({ + id: 'edge-container', + name: 'edge-container', + status: 'running', + watcher: 'docker', + agent: 'edge-agent', + })) as any, + getWatchers: vi.fn(() => ({})), + sessionMiddleware: (req: any, _res: unknown, next: (error?: unknown) => void) => { + req.session = { passport: { user: '{"username":"alice"}' } }; + req.sessionID = 'session-edge'; + next(); + }, + webSocketServer: { + handleUpgrade: vi.fn( + ( + _request: unknown, + _socket: unknown, + _head: unknown, + callback: (webSocket: unknown) => void, + ) => callback(viewer), + ), + }, + isRateLimited: vi.fn(() => false), + } as any; + if (agentResolver !== null) { + dependencies.getAgent = vi.fn(agentResolver); + } + return createContainerLogStreamGateway(dependencies); + } + + test('forwards correlated dd chunks from an edge agent to the authenticated viewer', async () => { + let handlers: + | { + onChunk: (chunk: { stream: 'stdout' | 'stderr'; logs: string }) => void; + onEnd: (reason?: string) => void; + onError: (error: Error) => void; + } + | undefined; + const cancel = vi.fn(); + const streamContainerLogs = vi.fn( + ( + _containerId: string, + _options: Record, + nextHandlers: NonNullable, + ) => { + handlers = nextHandlers; + return { cancel }; + }, + ); + const viewer = Object.assign(new EventEmitter(), { + send: vi.fn(), + close: vi.fn(), + bufferedAmount: undefined as unknown as number, + }); + const gateway = createAuthenticatedEdgeGateway(streamContainerLogs, viewer); + const handling = gateway.handleUpgrade( + createUpgradeRequest( + '/api/v1/containers/edge-container/logs/stream?tail=25&since=42&follow=true', + ) as any, + createUpgradeSocket() as any, + Buffer.alloc(0), + ); + await vi.waitFor(() => expect(streamContainerLogs).toHaveBeenCalledTimes(1)); + + expect(streamContainerLogs).toHaveBeenCalledWith( + 'edge-container', + { tail: 25, since: 42, follow: true, timestamps: true }, + expect.any(Object), + ); + handlers?.onChunk({ + stream: 'stderr', + logs: '2026-07-28T18:00:00.000000000Z warning\n', + }); + handlers?.onEnd('eof'); + await handling; + + const message = JSON.parse(String(viewer.send.mock.calls[0]?.[0])) as { + type: string; + ts: string; + line: string; + }; + expect(message).toMatchObject({ + type: 'stderr', + ts: '2026-07-28T18:00:00.000000000Z', + line: 'warning', + }); + expect(viewer.close).toHaveBeenCalledWith(1000, 'Stream ended'); + expect(cancel).not.toHaveBeenCalled(); + }); + + test('cancels the agent stream when the downstream viewer disconnects', async () => { + const cancel = vi.fn(); + const streamContainerLogs = vi.fn(() => ({ cancel })); + const viewer = Object.assign(new EventEmitter(), { + send: vi.fn(), + close: vi.fn(), + bufferedAmount: 0, + }); + const gateway = createAuthenticatedEdgeGateway(streamContainerLogs, viewer); + const handling = gateway.handleUpgrade( + createUpgradeRequest('/api/v1/containers/edge-container/logs/stream') as any, + createUpgradeSocket() as any, + Buffer.alloc(0), + ); + await vi.waitFor(() => expect(streamContainerLogs).toHaveBeenCalledTimes(1)); + + viewer.emit('close'); + await handling; + + expect(cancel).toHaveBeenCalledTimes(1); + }); + + test('evicts a slow viewer before its websocket buffer can grow without bound', async () => { + let handlers: + | { + onChunk: (chunk: { stream: 'stdout' | 'stderr'; logs: string }) => void; + } + | undefined; + const cancel = vi.fn(); + const streamContainerLogs = vi.fn( + ( + _containerId: string, + _options: Record, + nextHandlers: NonNullable, + ) => { + handlers = nextHandlers; + return { cancel }; + }, + ); + const viewer = Object.assign(new EventEmitter(), { + send: vi.fn(), + close: vi.fn(), + bufferedAmount: 2 * 1024 * 1024, + }); + const gateway = createAuthenticatedEdgeGateway(streamContainerLogs, viewer); + const handling = gateway.handleUpgrade( + createUpgradeRequest('/api/v1/containers/edge-container/logs/stream') as any, + createUpgradeSocket() as any, + Buffer.alloc(0), + ); + await vi.waitFor(() => expect(streamContainerLogs).toHaveBeenCalledTimes(1)); + + handlers?.onChunk({ stream: 'stdout', logs: 'too slow\n' }); + await handling; + + expect(cancel).toHaveBeenCalledTimes(1); + expect(viewer.close).toHaveBeenCalledWith(1013, 'Log viewer is too slow'); + expect(viewer.send).not.toHaveBeenCalled(); + }); + + test('keeps an edge stream error within the WebSocket close-reason limit', async () => { + let handlers: + | { + onError: (error: Error) => void; + } + | undefined; + const streamContainerLogs = vi.fn( + ( + _containerId: string, + _options: Record, + nextHandlers: NonNullable, + ) => { + handlers = nextHandlers; + return { cancel: vi.fn() }; + }, + ); + const viewer = Object.assign(new EventEmitter(), { + send: vi.fn(), + close: vi.fn((_code: number, reason: string) => { + if (Buffer.byteLength(reason, 'utf8') > 123) { + throw new RangeError('WebSocket close reason exceeds 123 bytes'); + } + }), + bufferedAmount: 0, + }); + const gateway = createAuthenticatedEdgeGateway(streamContainerLogs, viewer); + const handling = gateway.handleUpgrade( + createUpgradeRequest('/api/v1/containers/edge-container/logs/stream') as any, + createUpgradeSocket() as any, + Buffer.alloc(0), + ); + await vi.waitFor(() => expect(streamContainerLogs).toHaveBeenCalledTimes(1)); + + expect(() => handlers?.onError(new Error('x'.repeat(500)))).not.toThrow(); + await handling; + + const reason = String(viewer.close.mock.calls[0]?.[1]); + expect(Buffer.byteLength(reason, 'utf8')).toBeLessThanOrEqual(123); + expect(reason).toContain('Log stream error'); + }); + + test.each([ + { + resolver: () => undefined, + label: 'configured resolver', + }, + { + resolver: null, + label: 'default resolver', + }, + ])('closes when the edge agent is unavailable through the $label', async ({ resolver }) => { + const streamContainerLogs = vi.fn(); + const viewer = Object.assign(new EventEmitter(), { + send: vi.fn(), + close: vi.fn(), + bufferedAmount: 0, + }); + const gateway = createAuthenticatedEdgeGateway(streamContainerLogs, viewer, resolver); + + await gateway.handleUpgrade( + createUpgradeRequest('/api/v1/containers/edge-container/logs/stream') as any, + createUpgradeSocket() as any, + Buffer.alloc(0), + ); + + expect(viewer.close).toHaveBeenCalledWith(1011, 'Edge agent not available'); + expect(streamContainerLogs).not.toHaveBeenCalled(); + }); + + test('cancels when sending a decoded chunk to the viewer throws', async () => { + let handlers: + | { + onChunk: (chunk: { stream: 'stdout' | 'stderr'; logs: string }) => void; + onEnd: () => void; + onError: (error: Error) => void; + } + | undefined; + const cancel = vi.fn(); + const streamContainerLogs = vi.fn( + ( + _containerId: string, + _options: Record, + nextHandlers: NonNullable, + ) => { + handlers = nextHandlers; + return { cancel }; + }, + ); + const viewer = Object.assign(new EventEmitter(), { + send: vi.fn(() => { + throw new Error('viewer closed during send'); + }), + close: vi.fn(), + bufferedAmount: 0, + }); + const gateway = createAuthenticatedEdgeGateway(streamContainerLogs, viewer); + const handling = gateway.handleUpgrade( + createUpgradeRequest('/api/v1/containers/edge-container/logs/stream') as any, + createUpgradeSocket() as any, + Buffer.alloc(0), + ); + await vi.waitFor(() => expect(streamContainerLogs).toHaveBeenCalledTimes(1)); + + handlers?.onChunk({ + stream: 'stdout', + logs: '2026-07-28T18:00:00.000000000Z line\n', + }); + handlers?.onChunk({ stream: 'stdout', logs: 'ignored after cleanup\n' }); + handlers?.onEnd(); + handlers?.onError(new Error('ignored after cleanup')); + viewer.emit('close'); + await handling; + + expect(cancel).toHaveBeenCalledTimes(1); + expect(viewer.close).not.toHaveBeenCalled(); + }); + + test('cleans up once when sending a flushed partial line throws', async () => { + let handlers: + | { + onChunk: (chunk: { stream: 'stdout' | 'stderr'; logs: string }) => void; + onEnd: () => void; + } + | undefined; + const cancel = vi.fn(); + const streamContainerLogs = vi.fn( + ( + _containerId: string, + _options: Record, + nextHandlers: NonNullable, + ) => { + handlers = nextHandlers; + return { cancel }; + }, + ); + const viewer = Object.assign(new EventEmitter(), { + send: vi.fn(() => { + throw new Error('viewer closed during flush'); + }), + close: vi.fn(), + bufferedAmount: 0, + }); + const gateway = createAuthenticatedEdgeGateway(streamContainerLogs, viewer); + const handling = gateway.handleUpgrade( + createUpgradeRequest('/api/v1/containers/edge-container/logs/stream') as any, + createUpgradeSocket() as any, + Buffer.alloc(0), + ); + await vi.waitFor(() => expect(streamContainerLogs).toHaveBeenCalledTimes(1)); + + handlers?.onChunk({ stream: 'stdout', logs: 'partial without newline' }); + handlers?.onEnd(); + await handling; + + expect(cancel).toHaveBeenCalledTimes(1); + expect(viewer.close).not.toHaveBeenCalled(); + }); + + test('cancels on a downstream viewer error and ignores later agent callbacks', async () => { + let handlers: + | { + onChunk: (chunk: { stream: 'stdout' | 'stderr'; logs: string }) => void; + onEnd: () => void; + onError: (error: Error) => void; + } + | undefined; + const cancel = vi.fn(); + const streamContainerLogs = vi.fn( + ( + _containerId: string, + _options: Record, + nextHandlers: NonNullable, + ) => { + handlers = nextHandlers; + return { cancel }; + }, + ); + const viewer = Object.assign(new EventEmitter(), { + send: vi.fn(), + close: vi.fn(), + bufferedAmount: 0, + }); + const gateway = createAuthenticatedEdgeGateway(streamContainerLogs, viewer); + const handling = gateway.handleUpgrade( + createUpgradeRequest('/api/v1/containers/edge-container/logs/stream') as any, + createUpgradeSocket() as any, + Buffer.alloc(0), + ); + await vi.waitFor(() => expect(streamContainerLogs).toHaveBeenCalledTimes(1)); + + viewer.emit('error', new Error('viewer failed')); + handlers?.onChunk({ stream: 'stdout', logs: 'ignored\n' }); + handlers?.onEnd(); + handlers?.onError(new Error('ignored')); + viewer.emit('close'); + await handling; + + expect(cancel).toHaveBeenCalledTimes(1); + expect(viewer.send).not.toHaveBeenCalled(); + expect(viewer.close).not.toHaveBeenCalled(); + }); + + test('filters disabled stdout and stderr streams before sending', async () => { + let handlers: + | { + onChunk: (chunk: { stream: 'stdout' | 'stderr'; logs: string }) => void; + onEnd: () => void; + } + | undefined; + const streamContainerLogs = vi.fn( + ( + _containerId: string, + _options: Record, + nextHandlers: NonNullable, + ) => { + handlers = nextHandlers; + return { cancel: vi.fn() }; + }, + ); + const viewer = Object.assign(new EventEmitter(), { + send: vi.fn(), + close: vi.fn(), + bufferedAmount: 0, + }); + const gateway = createAuthenticatedEdgeGateway(streamContainerLogs, viewer); + const handling = gateway.handleUpgrade( + createUpgradeRequest( + '/api/v1/containers/edge-container/logs/stream?stdout=false&stderr=false', + ) as any, + createUpgradeSocket() as any, + Buffer.alloc(0), + ); + await vi.waitFor(() => expect(streamContainerLogs).toHaveBeenCalledTimes(1)); + + handlers?.onChunk({ stream: 'stdout', logs: 'stdout ignored\n' }); + handlers?.onChunk({ stream: 'stderr', logs: 'stderr ignored\n' }); + handlers?.onEnd(); + await handling; + + expect(viewer.send).not.toHaveBeenCalled(); + expect(viewer.close).toHaveBeenCalledWith(1000, 'Stream ended'); + }); + + test('uses an untruncated close reason for a short stream error', async () => { + let handlers: { onError: (error: Error) => void } | undefined; + const streamContainerLogs = vi.fn( + ( + _containerId: string, + _options: Record, + nextHandlers: NonNullable, + ) => { + handlers = nextHandlers; + return { cancel: vi.fn() }; + }, + ); + const viewer = Object.assign(new EventEmitter(), { + send: vi.fn(), + close: vi.fn(), + bufferedAmount: 0, + }); + const gateway = createAuthenticatedEdgeGateway(streamContainerLogs, viewer); + const handling = gateway.handleUpgrade( + createUpgradeRequest('/api/v1/containers/edge-container/logs/stream') as any, + createUpgradeSocket() as any, + Buffer.alloc(0), + ); + await vi.waitFor(() => expect(streamContainerLogs).toHaveBeenCalledTimes(1)); + + handlers?.onError(new Error('short')); + await handling; + + expect(viewer.close).toHaveBeenCalledWith(1011, 'Log stream error (short)'); + }); + }); + describe('attachContainerLogStreamWebSocketServer', () => { test('uses default ip-based key resolver when identity-aware keying is disabled', async () => { const webSocketUpgradeSpy = vi diff --git a/app/api/container/log-stream.ts b/app/api/container/log-stream.ts index a9ec76003..168d08fa7 100644 --- a/app/api/container/log-stream.ts +++ b/app/api/container/log-stream.ts @@ -2,6 +2,12 @@ import type { IncomingMessage } from 'node:http'; import type { Socket } from 'node:net'; import type { Readable } from 'node:stream'; import { type WebSocket, WebSocketServer } from 'ws'; +import type { + ContainerLogStreamChunk, + ContainerLogStreamHandle, + ContainerLogStreamHandlers, +} from '../../agent/EdgeAgentAdapter.js'; +import { getAgent } from '../../agent/manager.js'; import { getServerConfiguration } from '../../configuration/index.js'; import { formatLogDisplayTimestamp } from '../../log/display-timestamp.js'; import type { Container } from '../../model/container.js'; @@ -26,9 +32,12 @@ const RATE_LIMIT_WINDOW_MS = 15 * 60 * 1000; const RATE_LIMIT_MAX = 1000; const CLOSE_CODE_CONTAINER_NOT_RUNNING = 4001; const CLOSE_CODE_CONTAINER_NOT_FOUND = 4004; +const MAX_VIEWER_BUFFER_BYTES = 1024 * 1024; +const MAX_WEBSOCKET_CLOSE_REASON_BYTES = 123; type WebSocketLike = Pick & { off?: (event: 'close' | 'error', listener: () => void) => void; + bufferedAmount?: number; }; type WebSocketServerLike = { @@ -66,6 +75,22 @@ interface LogStreamContainerApi { interface ContainerLogStreamGatewayDependencies { getContainer: LogStreamContainerApi['getContainer']; getWatchers: () => Record; + getAgent?: (name: string) => + | { + edgeAdapter?: { + streamContainerLogs: ( + containerId: string, + options: { + tail?: number; + since?: string | number; + follow?: boolean; + timestamps?: boolean; + }, + handlers: ContainerLogStreamHandlers, + ) => ContainerLogStreamHandle; + }; + } + | undefined; sessionMiddleware?: SessionMiddleware; webSocketServer?: WebSocketServerLike; isRateLimited?: (key: string) => boolean; @@ -258,12 +283,140 @@ export function createDockerLogMessageDecoder() { }; } +function streamEdgeAgentLogsToWebSocket({ + webSocket, + container, + query, + getAgent: resolveAgent, +}: { + webSocket: WebSocketLike; + container: Container; + query: ParsedContainerLogStreamQuery; + getAgent: NonNullable; +}): Promise { + return new Promise((resolve) => { + const agent = resolveAgent(container.agent as string); + if (!agent?.edgeAdapter) { + webSocket.close(1011, 'Edge agent not available'); + resolve(); + return; + } + + const decoder = createDockerLogMessageDecoder(); + let handle: ContainerLogStreamHandle | undefined; + let cleaned = false; + + const cleanup = (cancelAgent: boolean) => { + if (cleaned) { + return; + } + cleaned = true; + webSocket.off?.('close', handleViewerClose); + webSocket.off?.('error', handleViewerError); + if (cancelAgent) { + handle?.cancel(); + } + resolve(); + }; + + const emitMessages = (messages: DockerLogMessage[]): boolean => { + for (const message of messages) { + if ( + (message.type === 'stdout' && !query.stdout) || + (message.type === 'stderr' && !query.stderr) + ) { + continue; + } + if ((webSocket.bufferedAmount ?? 0) > MAX_VIEWER_BUFFER_BYTES) { + webSocket.close(1013, 'Log viewer is too slow'); + cleanup(true); + return false; + } + try { + webSocket.send( + JSON.stringify({ + ...message, + displayTs: formatLogDisplayTimestamp(message.ts), + }), + ); + } catch { + cleanup(true); + return false; + } + } + return true; + }; + + const handleChunk = (chunk: ContainerLogStreamChunk) => { + if (cleaned) { + return; + } + emitMessages(decoder.push({ type: chunk.stream, payload: chunk.logs })); + }; + const handleEnd = () => { + if (cleaned) { + return; + } + if (emitMessages(decoder.flush())) { + webSocket.close(1000, 'Stream ended'); + } + cleanup(false); + }; + const handleError = (error: Error) => { + if (cleaned) { + return; + } + webSocket.close(1011, truncateWebSocketCloseReason(`Log stream error (${error.message})`)); + cleanup(false); + }; + const handleViewerClose = () => { + cleanup(true); + }; + const handleViewerError = () => { + cleanup(true); + }; + + webSocket.on('close', handleViewerClose); + webSocket.on('error', handleViewerError); + handle = agent.edgeAdapter.streamContainerLogs( + container.id, + { + tail: query.tail, + since: query.since, + follow: query.follow, + timestamps: true, + }, + { + onChunk: handleChunk, + onEnd: handleEnd, + onError: handleError, + }, + ); + }); +} + +function truncateWebSocketCloseReason(reason: string): string { + if (Buffer.byteLength(reason, 'utf8') <= MAX_WEBSOCKET_CLOSE_REASON_BYTES) { + return reason; + } + + let truncated = ''; + for (const character of reason) { + if (Buffer.byteLength(truncated + character, 'utf8') > MAX_WEBSOCKET_CLOSE_REASON_BYTES) { + break; + } + truncated += character; + } + return truncated; +} + async function streamContainerLogsToWebSocket({ webSocket, containerId, query, getContainer, getWatchers, + getAgent: resolveAgent, getErrorMessage, }: { webSocket: WebSocketLike; @@ -271,6 +424,7 @@ async function streamContainerLogsToWebSocket({ query: ParsedContainerLogStreamQuery; getContainer: ContainerLogStreamGatewayDependencies['getContainer']; getWatchers: ContainerLogStreamGatewayDependencies['getWatchers']; + getAgent: NonNullable; getErrorMessage: (error: unknown) => string; }): Promise { const container = getContainer(containerId); @@ -282,6 +436,15 @@ async function streamContainerLogsToWebSocket({ webSocket.close(CLOSE_CODE_CONTAINER_NOT_RUNNING, 'Container not running'); return; } + if (container.agent) { + await streamEdgeAgentLogsToWebSocket({ + webSocket, + container, + query, + getAgent: resolveAgent, + }); + return; + } const watcher = getWatchers()[`docker.${container.watcher}`]; if (!isLocalDockerWatcherApi(watcher) || !watcher.dockerApi) { @@ -396,6 +559,7 @@ export function createContainerLogStreamGateway( const { getContainer, getWatchers, + getAgent: resolveAgent = () => undefined, sessionMiddleware, webSocketServer = new WebSocketServer({ noServer: true }), isRateLimited = (() => { @@ -454,6 +618,7 @@ export function createContainerLogStreamGateway( query: parsedRequest.query, getContainer, getWatchers, + getAgent: resolveAgent, getErrorMessage: getLogStreamErrorMessage, }).finally(resolve); }); @@ -478,6 +643,7 @@ export function attachContainerLogStreamWebSocketServer(options: { const gateway = createContainerLogStreamGateway({ getContainer: storeContainer.getContainer, getWatchers: () => registry.getState().watcher, + getAgent, sessionMiddleware: options.sessionMiddleware, getRateLimitKey: createIdentityAwareUpgradeRateLimitKeyResolver(serverConfiguration), isRateLimited: options.isRateLimited, diff --git a/app/api/index.test.ts b/app/api/index.test.ts index 64637ff82..74fc51698 100644 --- a/app/api/index.test.ts +++ b/app/api/index.test.ts @@ -1984,9 +1984,7 @@ describe('API Index', () => { serverConfiguration: expect.objectContaining({ enabled: true }), isRateLimited: expect.any(Function), }); - expect(mockLog.info).toHaveBeenCalledWith( - 'portwing/1.0 edge endpoint enabled (experimental, DD_EXPERIMENTAL_PORTWING=true)', - ); + expect(mockLog.info).toHaveBeenCalledWith('portwing/1.0 edge endpoint enabled by default'); }); test('should NOT attach Portwing WS server when DD_EXPERIMENTAL_PORTWING is disabled', async () => { @@ -2019,7 +2017,7 @@ describe('API Index', () => { await indexRouter.init(); expect(mockLog.info).toHaveBeenCalledWith( - 'portwing/1.0 edge endpoint is disabled — set DD_EXPERIMENTAL_PORTWING=true to enable it', + 'portwing/1.0 edge endpoint disabled by DD_EXPERIMENTAL_PORTWING=false', ); }); }); diff --git a/app/api/index.ts b/app/api/index.ts index feb7389a3..697f73f76 100644 --- a/app/api/index.ts +++ b/app/api/index.ts @@ -400,15 +400,13 @@ export async function init() { isRateLimited, }); if (getExperimentalPortwingEnabled()) { - log.info('portwing/1.0 edge endpoint enabled (experimental, DD_EXPERIMENTAL_PORTWING=true)'); + log.info('portwing/1.0 edge endpoint enabled by default'); attachPortwingWsServer({ server, serverConfiguration: configuration as Record, isRateLimited, }); } else { - log.info( - 'portwing/1.0 edge endpoint is disabled — set DD_EXPERIMENTAL_PORTWING=true to enable it', - ); + log.info('portwing/1.0 edge endpoint disabled by DD_EXPERIMENTAL_PORTWING=false'); } } diff --git a/app/api/openapi/index.ts b/app/api/openapi/index.ts index b2a797eb5..d43b3bdbe 100644 --- a/app/api/openapi/index.ts +++ b/app/api/openapi/index.ts @@ -45,7 +45,8 @@ export const openApiDocument = { { name: 'Docs', description: 'API documentation endpoints' }, { name: 'Portwing', - description: 'Edge agent key registry — experimental, requires DD_EXPERIMENTAL_PORTWING=true', + description: + 'Stable edge-agent key registry — enabled by default; DD_EXPERIMENTAL_PORTWING=false is the emergency disable', }, ], security: [{ sessionAuth: [] }], diff --git a/app/api/openapi/paths/portwing.test.ts b/app/api/openapi/paths/portwing.test.ts index 782449885..4f326e6f9 100644 --- a/app/api/openapi/paths/portwing.test.ts +++ b/app/api/openapi/paths/portwing.test.ts @@ -18,9 +18,9 @@ describe('portwingPaths', () => { expect(getPath.summary).toBe('List all registered edge-agent keys'); }); - test('description mentions EXPERIMENTAL and DD_EXPERIMENTAL_PORTWING=true', () => { - expect(getPath.description).toContain('EXPERIMENTAL'); - expect(getPath.description).toContain('DD_EXPERIMENTAL_PORTWING=true'); + test('description documents default enable and emergency disable', () => { + expect(getPath.description).toMatch(/enabled by default/i); + expect(getPath.description).toContain('DD_EXPERIMENTAL_PORTWING=false'); }); test('200 response is a json array', () => { @@ -78,9 +78,9 @@ describe('portwingPaths', () => { expect(postPath.summary).toBe('Register a new authorized edge-agent key'); }); - test('description mentions EXPERIMENTAL and DD_EXPERIMENTAL_PORTWING=true', () => { - expect(postPath.description).toContain('EXPERIMENTAL'); - expect(postPath.description).toContain('DD_EXPERIMENTAL_PORTWING=true'); + test('description documents default enable and emergency disable', () => { + expect(postPath.description).toMatch(/enabled by default/i); + expect(postPath.description).toContain('DD_EXPERIMENTAL_PORTWING=false'); }); test('request body requires pubkeyBase64 and label', () => { @@ -147,9 +147,9 @@ describe('portwingPaths', () => { expect(deletePath.summary).toBe('Revoke a registered edge-agent key'); }); - test('description mentions EXPERIMENTAL and DD_EXPERIMENTAL_PORTWING=true', () => { - expect(deletePath.description).toContain('EXPERIMENTAL'); - expect(deletePath.description).toContain('DD_EXPERIMENTAL_PORTWING=true'); + test('description documents default enable and emergency disable', () => { + expect(deletePath.description).toMatch(/enabled by default/i); + expect(deletePath.description).toContain('DD_EXPERIMENTAL_PORTWING=false'); }); test('keyId path param has correct name and pattern', () => { diff --git a/app/api/openapi/paths/portwing.ts b/app/api/openapi/paths/portwing.ts index d26795f0d..dd4cefa7f 100644 --- a/app/api/openapi/paths/portwing.ts +++ b/app/api/openapi/paths/portwing.ts @@ -33,8 +33,8 @@ const agentKeyRecord = { additionalProperties: false, } as const; -const experimentalNote = - 'EXPERIMENTAL: only available when the server is started with DD_EXPERIMENTAL_PORTWING=true.'; +const availabilityNote = + 'Enabled by default; set DD_EXPERIMENTAL_PORTWING=false for an emergency disable.'; const keyIdPathParam = { name: 'keyId', @@ -50,7 +50,7 @@ export const portwingPaths = { tags: ['Portwing'], summary: 'List all registered edge-agent keys', operationId: 'listPortwingKeys', - description: `Returns all keys — active and revoked. ${experimentalNote}`, + description: `Returns all keys — active and revoked. ${availabilityNote}`, responses: { 200: jsonResponse('Array of agent key records', { type: 'array', @@ -63,7 +63,7 @@ export const portwingPaths = { tags: ['Portwing'], summary: 'Register a new authorized edge-agent key', operationId: 'createPortwingKey', - description: `Registers a new Ed25519 public key for edge agent authentication. ${experimentalNote}`, + description: `Registers a new Ed25519 public key for edge agent authentication. ${availabilityNote}`, requestBody: { required: true, content: { @@ -113,7 +113,7 @@ export const portwingPaths = { tags: ['Portwing'], summary: 'Revoke a registered edge-agent key', operationId: 'revokePortwingKey', - description: `Revokes the key and disconnects any live WebSocket session authenticated with it. ${experimentalNote}`, + description: `Revokes the key and disconnects any live WebSocket session authenticated with it. ${availabilityNote}`, parameters: [keyIdPathParam], responses: { 204: noContentResponse, diff --git a/app/configuration/index.test.ts b/app/configuration/index.test.ts index cc34df3ee..646f19777 100644 --- a/app/configuration/index.test.ts +++ b/app/configuration/index.test.ts @@ -109,9 +109,15 @@ test('getLocalWatcherEnabled should return false when disabled via env', async ( delete configuration.ddEnvVars.DD_LOCAL_WATCHER; }); -test('getExperimentalPortwingEnabled should default to false', () => { +test('getExperimentalPortwingEnabled should default to true after edge graduation', () => { delete configuration.ddEnvVars.DD_EXPERIMENTAL_PORTWING; + expect(configuration.getExperimentalPortwingEnabled()).toStrictEqual(true); +}); + +test('getExperimentalPortwingEnabled should allow an explicit emergency disable', () => { + configuration.ddEnvVars.DD_EXPERIMENTAL_PORTWING = 'false'; expect(configuration.getExperimentalPortwingEnabled()).toStrictEqual(false); + delete configuration.ddEnvVars.DD_EXPERIMENTAL_PORTWING; }); test('getExperimentalPortwingEnabled should return true when set to "true"', () => { diff --git a/app/configuration/index.ts b/app/configuration/index.ts index 0da4fd795..983a896af 100644 --- a/app/configuration/index.ts +++ b/app/configuration/index.ts @@ -230,7 +230,11 @@ function envFlagEnabled(value: string | undefined) { } export function getExperimentalPortwingEnabled() { - return envFlagEnabled(ddEnvVars.DD_EXPERIMENTAL_PORTWING); + const value = ddEnvVars.DD_EXPERIMENTAL_PORTWING; + if (value === undefined) { + return true; + } + return envFlagEnabled(value); } /** diff --git a/content/docs/current/api/index.mdx b/content/docs/current/api/index.mdx index a0a8e8551..bdbf7b7f7 100644 --- a/content/docs/current/api/index.mdx +++ b/content/docs/current/api/index.mdx @@ -92,9 +92,9 @@ If the header is missing or does not match the expected token, the API responds | `/api/v1/operations/:id/cancel` | `POST` | Cancel a queued or in-progress update operation (see [Container API](/docs/api/container#cancel-an-update-operation)) | | `/api/v1/containers/:id/release-notes` | `GET` | Structured release notes for a container's current update target (see [Container API](/docs/api/container#get-container-release-notes)) | | `/api/v1/containers/:id/intermediate-release-notes` | `GET` | Semver-only release notes between two tags, with `from` required and `to` defaulting to the pending update target (see [Container API](/docs/api/container#get-intermediate-release-notes)) | -| `/api/v1/portwing/keys` | `GET`, `POST` | Experimental Portwing edge-agent key registry, available only with `DD_EXPERIMENTAL_PORTWING=true` | -| `/api/v1/portwing/keys/:keyId` | `DELETE` | Revoke an experimental Portwing Ed25519 public key and disconnect live sessions authenticated with it | -| `/api/v1/portwing/ws` | `WS` | Experimental Portwing edge-agent WebSocket endpoint using the `portwing/1.0` protocol | +| `/api/v1/portwing/keys` | `GET`, `POST` | Stable Portwing edge-agent key registry, enabled by default (`DD_EXPERIMENTAL_PORTWING=false` is an emergency disable) | +| `/api/v1/portwing/keys/:keyId` | `DELETE` | Revoke a Portwing Ed25519 public key and disconnect live sessions authenticated with it | +| `/api/v1/portwing/ws` | `WS` | Stable Portwing edge-agent WebSocket endpoint using the `portwing/1.0` protocol | | `/api/v1/containers/update` | `POST` | Bulk queue updates for multiple containers (see [Container API](/docs/api/container#bulk-update-containers)) | | `/api/v1/containers/backups` | `GET` | [All image backups](/docs/api/container#list-all-backups) (optionally filter by `containerName`) | | `/api/v1/icons/:provider/:slug` | `GET` | Proxy and cache registry icon images | diff --git a/content/docs/current/api/portwing.mdx b/content/docs/current/api/portwing.mdx index c30eab65e..30005cc1a 100644 --- a/content/docs/current/api/portwing.mdx +++ b/content/docs/current/api/portwing.mdx @@ -5,7 +5,7 @@ description: "Ed25519 agent-key registry and the portwing/1.0 WebSocket edge end import { Callout } from 'fumadocs-ui/components/callout'; -Portwing support is experimental. Both the WebSocket gateway and the key-registry REST API below only exist when the server is started with `DD_EXPERIMENTAL_PORTWING=true` — the wire protocol and REST shapes may still change before this graduates out of the flag. +The Portwing WebSocket gateway and key-registry API are enabled by default and follow the published `portwing/1.0` compatibility policy. `DD_EXPERIMENTAL_PORTWING=false` remains available only as an emergency disable for new edge connections. Edge mode inverts the normal drydock connection model. In the standard setup the controller dials out to each remote agent. Edge mode is for agents that sit behind NAT or a firewall and cannot accept inbound connections: the `portwing` agent dials **out** to drydock over an encrypted WebSocket (`wss://`), and drydock registers the connection as if the agent had been reached normally. No inbound port on the agent host is needed. The chain is `sockguard → portwing (edge client) → drydock (this endpoint)`. @@ -195,6 +195,46 @@ A name collision with an already-connected agent under the same key (or the in-f `serverCompatLevel` is `1.4.0` — the compatibility level actually implemented by this server, not a target for a future release. If the hello's `drydockCompat` field has a different major version, the connection is still accepted; drydock only logs a warning so operators can check the compat matrix. +### Continuous container-log streaming + +The authenticated edge connection carries live Docker logs without opening another port on the agent. Drydock starts a stream with: + +```json +{ + "type": "dd:container_log_request", + "data": { + "requestId": "log-8f31", + "containerId": "8d3f…", + "stream": true, + "follow": true, + "tail": 200, + "timestamps": true, + "since": 0, + "until": 0 + } +} +``` + +A current Portwing agent responds with zero or more chunk frames, preserving the Docker stream: + +```json +{ + "type": "dd:container_log_chunk", + "data": { + "requestId": "log-8f31", + "containerId": "8d3f…", + "stream": "stderr", + "logs": "2026-07-28T12:00:00Z warning\n" + } +} +``` + +The agent finishes with `dd:container_log_end` (`requestId`, `containerId`, optional `reason`) or `dd:container_log_error` (`requestId`, `containerId`, `error`). When the browser closes its log WebSocket, Drydock sends `dd:container_log_cancel` with the same `requestId` and `containerId`; Portwing cancels the Docker request promptly. + +`requestId` is mandatory and correlates concurrent viewers. Drydock caps each downstream viewer at 1 MiB of queued data, while Portwing uses a bounded controller send queue and admits at most 128 live log streams. A slow consumer is closed instead of allowing unbounded memory growth. + +Mixed-version fleets remain compatible. An older Portwing agent ignores `stream` and returns the legacy `dd:container_log_response`; Drydock renders that response as one stdout chunk and then ends the viewer. A newer agent still uses the legacy response for requests that omit `stream: true`. + ### Protocol limits | Parameter | Value | diff --git a/content/docs/current/configuration/agents/index.mdx b/content/docs/current/configuration/agents/index.mdx index 5d9635d9a..8c0451073 100644 --- a/content/docs/current/configuration/agents/index.mdx +++ b/content/docs/current/configuration/agents/index.mdx @@ -16,14 +16,9 @@ The Controller connects to one or more Agents via HTTP/HTTPS. The Agent pushes r **The controller dials out to each agent — the agent does not need to know where the controller is.** Only the controller side gets `DD_AGENT_{name}_HOST`. The agent just needs its port reachable from the controller. (Both ends share the same secret so they trust each other.) -## Experimental edge-agent dial-out (Portwing) +## Edge-agent dial-out (Portwing) -When an agent is behind NAT or a firewall and cannot accept inbound controller requests, Drydock can accept an experimental Portwing edge-agent WebSocket instead. Enable the controller endpoint with: - -```yaml -environment: - - DD_EXPERIMENTAL_PORTWING=true -``` +When an agent is behind NAT or a firewall and cannot accept inbound controller requests, Drydock accepts a Portwing edge-agent WebSocket instead. The endpoint is enabled by default. `DD_EXPERIMENTAL_PORTWING=false` is retained only as an emergency disable; no enable flag is required. The WebSocket endpoint is `WS /api/v1/portwing/ws` (compatibility alias: `/api/portwing/ws`) and speaks `portwing/1.0`. Authentication uses Ed25519 public-key challenge-response, not the shared `DD_AGENT_SECRET` token. Register authorized agent keys through the authenticated API: @@ -58,7 +53,7 @@ secrets: file: ./portwing_ed25519.pem ``` -The Drydock controller enables the endpoint with `DD_EXPERIMENTAL_PORTWING=true` and needs the agent's public key registered before it connects: +The Drydock controller needs the agent's public key registered before it connects. No Portwing-specific environment variable is required: ```yaml services: @@ -66,15 +61,13 @@ services: image: codeswhat/drydock ports: - "3000:3000" - environment: - - DD_EXPERIMENTAL_PORTWING=true ``` Generate the agent's keypair with `portwing keygen -comment "edge-host-01"`, then register the printed public key with the controller (`POST /api/v1/portwing/keys`, or preload it via `DD_PORTWING_AUTHORIZED_KEYS`) before starting the agent — see the REST API table above. -Portwing support is experimental. Its REST API and wire protocol may change before it is promoted out of `DD_EXPERIMENTAL_PORTWING=true`. +Portwing's `portwing/1.0` wire protocol and key-registry API are stable under the published compatibility policy. Set `DD_EXPERIMENTAL_PORTWING=false` only when an emergency rollback requires disabling new edge connections. -**Container log downloads from an edge (Portwing) agent now honor the `timestamps` query parameter,** the same as regular (HTTP/SSE) agents. The `dd:container_log_request` wire message carries a `timestamps` field that a current Portwing agent reads, so the [container logs endpoint](/docs/api/container#download-container-logs) (and the UI "show timestamps" toggle) works over the edge tunnel. Tail-follow (`follow`/`until`) is honored by the agent as a short bounded window, but the one-shot log download endpoint doesn't expose it; continuous live tailing over an edge agent isn't wired up yet. An older Portwing agent that predates the `timestamps` wire field simply omits timestamps, degrading gracefully. +**Container logs work continuously over the authenticated edge tunnel.** Live viewers receive separate stdout/stderr chunks, may request timestamps, and cancel the upstream Docker stream when the viewer closes. Both ends enforce bounded queues so a stalled viewer or controller is disconnected instead of growing memory without limit. Older Portwing agents fall back to the bounded one-shot log response, so mixed-version fleets degrade gracefully. ## Quickstart diff --git a/scripts/portwing-fleet-soak.mjs b/scripts/portwing-fleet-soak.mjs new file mode 100644 index 000000000..b3b5b6581 --- /dev/null +++ b/scripts/portwing-fleet-soak.mjs @@ -0,0 +1,614 @@ +#!/usr/bin/env node + +import { spawn, spawnSync } from 'node:child_process'; +import { createHash, generateKeyPairSync } from 'node:crypto'; +import { + closeSync, + mkdirSync, + mkdtempSync, + openSync, + readFileSync, + rmSync, + statSync, + writeFileSync, +} from 'node:fs'; +import { createServer } from 'node:http'; +import { tmpdir } from 'node:os'; +import { dirname, join, resolve } from 'node:path'; +import { fileURLToPath, pathToFileURL } from 'node:url'; + +const DEFAULTS = { + agents: 8, + duration: '45s', + execPerAgent: 3, + rssGrowthThresholdBytes: 128 * 1024 * 1024, + heapGrowthThresholdBytes: 64 * 1024 * 1024, + output: 'portwing-fleet-soak.json', +}; + +function fail(message) { + throw new Error(`fleet-soak: ${message}`); +} + +function parsePositiveInteger(value, name) { + if (!/^[0-9]+$/.test(value)) { + fail(`invalid ${name}: ${value}`); + } + const parsed = Number(value); + if (!Number.isSafeInteger(parsed) || parsed < 1) { + fail(`invalid ${name}: ${value}`); + } + return parsed; +} + +function parseDuration(value) { + const match = /^([0-9]+)(ms|s|m|h)$/.exec(value); + if (!match) { + fail(`invalid --duration: ${value}`); + } + const amount = Number(match[1]); + const multipliers = { ms: 1, s: 1000, m: 60_000, h: 3_600_000 }; + const milliseconds = amount * multipliers[match[2]]; + if (milliseconds < 15_000) { + fail('--duration must be at least 15s'); + } + return milliseconds; +} + +function parseArgs(argv) { + const options = { ...DEFAULTS }; + for (let index = 0; index < argv.length; index += 1) { + const arg = argv[index]; + const next = () => { + index += 1; + if (index >= argv.length) { + fail(`missing value for ${arg}`); + } + return argv[index]; + }; + switch (arg) { + case '--portwing-bin': + options.portwingBin = resolve(next()); + break; + case '--mockdocker-bin': + options.mockDockerBin = resolve(next()); + break; + case '--portwing-repo': + options.portwingRepository = resolve(next()); + break; + case '--agents': + options.agents = parsePositiveInteger(next(), '--agents'); + break; + case '--duration': + options.duration = next(); + break; + case '--exec-per-agent': + options.execPerAgent = parsePositiveInteger(next(), '--exec-per-agent'); + break; + case '--rss-growth-threshold-bytes': + options.rssGrowthThresholdBytes = parsePositiveInteger( + next(), + '--rss-growth-threshold-bytes', + ); + break; + case '--heap-growth-threshold-bytes': + options.heapGrowthThresholdBytes = parsePositiveInteger( + next(), + '--heap-growth-threshold-bytes', + ); + break; + case '--output': + options.output = resolve(next()); + break; + case '--help': + case '-h': + process.stdout.write( + [ + 'Usage: scripts/portwing-fleet-soak.mjs --portwing-bin PATH --mockdocker-bin PATH [options]', + '', + ' --portwing-repo PATH', + ' --agents N', + ' --duration 45s|30m|4h', + ' --exec-per-agent N', + ' --rss-growth-threshold-bytes N', + ' --heap-growth-threshold-bytes N', + ' --output PATH', + '', + ].join('\n'), + ); + process.exit(0); + break; + default: + fail(`unknown argument: ${arg}`); + } + } + if (!options.portwingBin || !options.mockDockerBin) { + fail('--portwing-bin and --mockdocker-bin are required'); + } + for (const [name, path] of [ + ['portwing binary', options.portwingBin], + ['mockdocker binary', options.mockDockerBin], + ]) { + if (!statSync(path).isFile()) { + fail(`${name} is not a file: ${path}`); + } + } + options.durationMs = parseDuration(options.duration); + return options; +} + +function sleep(milliseconds) { + return new Promise((resolvePromise) => setTimeout(resolvePromise, milliseconds)); +} + +async function waitFor(description, predicate, timeoutMs = 20_000) { + const deadline = Date.now() + timeoutMs; + let lastError; + while (Date.now() < deadline) { + try { + const value = await predicate(); + if (value) { + return value; + } + } catch (error) { + lastError = error; + } + await sleep(50); + } + fail( + `timed out waiting for ${description}${lastError instanceof Error ? `: ${lastError.message}` : ''}`, + ); +} + +function childProcess(command, args, options) { + const child = spawn(command, args, { + ...options, + stdio: ['ignore', options.stdout ?? 'ignore', options.stderr ?? 'ignore'], + }); + child.on('error', (error) => { + process.stderr.write(`fleet-soak: child error (${command}): ${error.message}\n`); + }); + return child; +} + +function rssBytes(pid) { + if (!pid) { + return 0; + } + const result = spawnSync('ps', ['-o', 'rss=', '-p', String(pid)], { encoding: 'utf8' }); + if (result.status !== 0) { + return 0; + } + const kibibytes = Number(result.stdout.trim()); + return Number.isFinite(kibibytes) ? kibibytes * 1024 : 0; +} + +function threadCount(pid) { + if (!pid) { + return 0; + } + try { + return ( + readFileSync(`/proc/${pid}/status`, 'utf8') + .split('\n') + .map((line) => /^Threads:\s+([0-9]+)$/.exec(line)) + .find(Boolean)?.[1] ?? 0 + ); + } catch { + return 0; + } +} + +function gitCommit(cwd) { + const result = spawnSync('git', ['rev-parse', 'HEAD'], { cwd, encoding: 'utf8' }); + return result.status === 0 ? result.stdout.trim() : 'unknown'; +} + +function closeAdapterSessions(adapter, sessionIds) { + const sessions = adapter?.execSessions; + if (!(sessions instanceof Map)) { + return; + } + for (const sessionId of sessionIds) { + sessions.get(sessionId)?.close(); + } +} + +const options = parseArgs(process.argv.slice(2)); +const repositoryRoot = resolve(dirname(fileURLToPath(import.meta.url)), '..'); +const appRoot = join(repositoryRoot, 'app'); +process.chdir(join(appRoot, 'dist')); +const portwingRepository = options.portwingRepository; +const runDirectory = mkdtempSync(join(tmpdir(), 'portwing-fleet-soak-')); +const socketPath = join(runDirectory, 'docker.sock'); +const privateKeyPath = join(runDirectory, 'agent.key'); +const mockLogPath = join(runDirectory, 'mockdocker.log'); +const controller = createServer((_request, response) => { + response.writeHead(200, { 'Content-Type': 'application/json' }); + response.end('{"status":"ok"}\n'); +}); + +const children = new Map(); +let mockDocker; +let sampler; +let controllerAddress; +let getAgents; +let removeAgent; +let clearNonceCacheForTesting; +let assertionFailure; +const evidence = { + schemaVersion: 1, + startedAt: new Date().toISOString(), + passed: false, + configuration: { + agents: options.agents, + duration: options.duration, + execPerAgent: options.execPerAgent, + rssGrowthThresholdBytes: options.rssGrowthThresholdBytes, + heapGrowthThresholdBytes: options.heapGrowthThresholdBytes, + }, + revisions: { + drydock: gitCommit(repositoryRoot), + portwing: portwingRepository ? gitCommit(portwingRepository) : 'unknown', + }, + phases: { + initialConnections: 0, + execSessionsStarted: 0, + execOutputFrames: 0, + slowConsumerReconnects: 0, + reconnectStorms: 0, + reconnectsObserved: 0, + }, + resources: { + controllerHeap: {}, + agentRss: {}, + maxAgentThreads: 0, + }, + assertions: [], +}; + +function activeChildren() { + return [...children.values()].filter((child) => child.exitCode === null); +} + +function aggregateAgentRss() { + return activeChildren().reduce((total, child) => total + rssBytes(child.pid), 0); +} + +function sampleResources() { + const heap = process.memoryUsage().heapUsed; + const rss = aggregateAgentRss(); + evidence.resources.controllerHeap.maximum = Math.max( + evidence.resources.controllerHeap.maximum ?? 0, + heap, + ); + evidence.resources.agentRss.maximum = Math.max(evidence.resources.agentRss.maximum ?? 0, rss); + for (const child of activeChildren()) { + evidence.resources.maxAgentThreads = Math.max( + evidence.resources.maxAgentThreads, + Number(threadCount(child.pid)), + ); + } +} + +function recordAssertion(name, passed, detail) { + evidence.assertions.push({ name, passed, detail }); + if (!passed && !assertionFailure) { + assertionFailure = `${name}: ${detail}`; + } +} + +function spawnAgent(index) { + const name = `fleet-agent-${index}`; + const logPath = join(runDirectory, `${name}.log`); + const logFile = openSync(logPath, 'a', 0o600); + const child = childProcess(options.portwingBin, [], { + env: { + ...process.env, + AGENT_ID: name, + AGENT_NAME: name, + BIND_ADDRESS: '127.0.0.1', + DD_POLL_INTERVAL: '2', + DOCKER_SOCKET: socketPath, + DRYDOCK_URL: `http://127.0.0.1:${controllerAddress.port}`, + LOG_LEVEL: 'warn', + MAX_RECONNECT_DELAY: '2', + NO_COLOR: '1', + PORT: '0', + PRIVATE_KEY_FILE: privateKeyPath, + RECONNECT_DELAY: '1', + }, + stdout: logFile, + stderr: logFile, + }); + closeSync(logFile); + children.set(name, child); + return child; +} + +async function run() { + const gatewayModule = await import(pathToFileURL(join(appRoot, 'dist/api/portwing-ws.js')).href); + const managerModule = await import(pathToFileURL(join(appRoot, 'dist/agent/manager.js')).href); + const storeModule = await import(pathToFileURL(join(appRoot, 'dist/store/index.js')).href); + const { createPortwingWsGateway, clearNonceCacheForTesting: clearNonceCache } = gatewayModule; + ({ getAgents, removeAgent } = managerModule); + clearNonceCacheForTesting = clearNonceCache; + await storeModule.init({ memory: true }); + + const { privateKey, publicKey } = generateKeyPairSync('ed25519'); + writeFileSync(privateKeyPath, privateKey.export({ type: 'pkcs8', format: 'pem' }), { + mode: 0o600, + }); + const publicDer = publicKey.export({ type: 'spki', format: 'der' }); + const rawPublicKey = publicDer.subarray(12); + const keyId = createHash('sha256').update(rawPublicKey).digest().subarray(0, 8).toString('hex'); + const keyRecord = { + keyId, + pubkey: rawPublicKey.toString('base64'), + label: 'fleet-soak', + createdAt: new Date().toISOString(), + revokedAt: null, + }; + + const gateway = createPortwingWsGateway({ + serverConfiguration: {}, + getAgentKeys: { + getKey: (candidate) => (candidate === keyId ? keyRecord : null), + }, + }); + controller.on('upgrade', (request, socket, head) => { + gateway.handleUpgrade(request, socket, head); + }); + await new Promise((resolvePromise, reject) => { + controller.once('error', reject); + controller.listen(0, '127.0.0.1', () => { + controller.off('error', reject); + resolvePromise(); + }); + }); + controllerAddress = controller.address(); + + const mockLog = openSync(mockLogPath, 'a', 0o600); + mockDocker = childProcess(options.mockDockerBin, ['-socket', socketPath], { + stdout: mockLog, + stderr: mockLog, + }); + closeSync(mockLog); + await waitFor('mock Docker socket', () => { + try { + return statSync(socketPath).isSocket(); + } catch { + return false; + } + }); + + for (let index = 0; index < options.agents; index += 1) { + spawnAgent(index); + } + await waitFor( + `${options.agents} initial agents`, + () => getAgents().length === options.agents, + 30_000, + ); + evidence.phases.initialConnections = getAgents().length; + await sleep(1000); + + if (typeof global.gc === 'function') { + global.gc(); + } + evidence.resources.controllerHeap.baseline = process.memoryUsage().heapUsed; + evidence.resources.agentRss.baseline = aggregateAgentRss(); + const soakDeadline = Date.now() + options.durationMs; + sampleResources(); + sampler = setInterval(sampleResources, 1000); + + const firstExecSessionIds = new Map(); + for (const agent of getAgents()) { + const sessionIds = []; + for (let index = 0; index < options.execPerAgent; index += 1) { + const sessionId = await agent.edgeAdapter.startExec('c0000000001', ['sh', '-c', 'cat'], { + tty: true, + outputCallback: () => { + evidence.phases.execOutputFrames += 1; + }, + }); + sessionIds.push(sessionId); + evidence.phases.execSessionsStarted += 1; + } + firstExecSessionIds.set(agent, sessionIds); + } + await waitFor( + 'initial exec output from every session', + () => evidence.phases.execOutputFrames >= options.agents * options.execPerAgent, + ); + await sleep(Math.min(5000, Math.max(1000, Math.floor(options.durationMs / 8)))); + for (const [agent, sessionIds] of firstExecSessionIds) { + closeAdapterSessions(agent.edgeAdapter, sessionIds); + } + + const slowAgentName = 'fleet-agent-0'; + const slowAgent = getAgents().find((agent) => agent.name === slowAgentName); + if (!slowAgent) { + fail(`missing ${slowAgentName}`); + } + const slowAdapter = slowAgent.edgeAdapter; + const serverSocket = slowAdapter?.ws?._socket; + if (!serverSocket?.pause || !serverSocket?.resume) { + fail('production EdgeAgentAdapter WebSocket does not expose a pausable socket'); + } + serverSocket.pause(); + slowAdapter.streamContainerLogs( + 'c0000000001', + { follow: true, timestamps: true }, + { + onChunk: () => {}, + onEnd: () => {}, + onError: () => {}, + }, + ); + await sleep(3500); + serverSocket.resume(); + const reconnectedSlowAgent = await waitFor( + 'slow-consumer eviction and reconnect', + () => { + const candidate = getAgents().find((agent) => agent.name === slowAgentName); + return candidate && candidate !== slowAgent ? candidate : undefined; + }, + 25_000, + ); + evidence.phases.slowConsumerReconnects = reconnectedSlowAgent ? 1 : 0; + evidence.phases.reconnectsObserved += 1; + + const stormCycles = 2; + for (let cycle = 0; cycle < stormCycles; cycle += 1) { + const before = new Map(getAgents().map((agent) => [agent.name, agent])); + for (const agent of before.values()) { + agent.edgeAdapter?.ws?.close(1012, 'fleet soak reconnect storm'); + } + await waitFor( + `reconnect storm ${cycle + 1}`, + () => + [...before.entries()].every(([name, previous]) => { + const current = getAgents().find((agent) => agent.name === name); + return current && current !== previous; + }), + 30_000, + ); + evidence.phases.reconnectStorms += 1; + evidence.phases.reconnectsObserved += before.size; + } + + const sustainedExecSessionIds = new Map(); + for (const agent of getAgents()) { + const sessionIds = []; + for (let index = 0; index < options.execPerAgent; index += 1) { + sessionIds.push( + await agent.edgeAdapter.startExec('c0000000001', ['sh', '-c', 'cat'], { + tty: true, + outputCallback: () => { + evidence.phases.execOutputFrames += 1; + }, + }), + ); + evidence.phases.execSessionsStarted += 1; + } + sustainedExecSessionIds.set(agent, sessionIds); + } + + const remaining = Math.max(1000, soakDeadline - Date.now()); + await sleep(remaining); + for (const [agent, sessionIds] of sustainedExecSessionIds) { + closeAdapterSessions(agent.edgeAdapter, sessionIds); + } + await sleep(1000); + sampleResources(); + clearInterval(sampler); + sampler = undefined; + + if (typeof global.gc === 'function') { + global.gc(); + } + evidence.resources.controllerHeap.final = process.memoryUsage().heapUsed; + evidence.resources.agentRss.final = aggregateAgentRss(); + evidence.resources.controllerHeap.growth = + evidence.resources.controllerHeap.final - evidence.resources.controllerHeap.baseline; + evidence.resources.agentRss.growth = + evidence.resources.agentRss.final - evidence.resources.agentRss.baseline; + + recordAssertion( + 'real fleet connected', + evidence.phases.initialConnections === options.agents, + `${evidence.phases.initialConnections}/${options.agents} agents`, + ); + recordAssertion( + 'sustained concurrent exec completed', + evidence.phases.execOutputFrames >= options.agents * options.execPerAgent, + `${evidence.phases.execOutputFrames} output frames across ${evidence.phases.execSessionsStarted} sessions`, + ); + recordAssertion( + 'slow consumer triggered bounded reconnect', + evidence.phases.slowConsumerReconnects === 1, + `${evidence.phases.slowConsumerReconnects} observed reconnect`, + ); + recordAssertion( + 'reconnect storms recovered', + evidence.phases.reconnectStorms === stormCycles, + `${evidence.phases.reconnectStorms}/${stormCycles} storms recovered`, + ); + recordAssertion( + 'aggregate agent RSS growth bounded', + evidence.resources.agentRss.growth <= options.rssGrowthThresholdBytes, + `${evidence.resources.agentRss.growth} <= ${options.rssGrowthThresholdBytes} bytes`, + ); + recordAssertion( + 'controller heap growth bounded', + evidence.resources.controllerHeap.growth <= options.heapGrowthThresholdBytes, + `${evidence.resources.controllerHeap.growth} <= ${options.heapGrowthThresholdBytes} bytes`, + ); + + if (assertionFailure) { + fail(assertionFailure); + } + evidence.passed = true; +} + +async function cleanup() { + if (sampler) { + clearInterval(sampler); + } + try { + for (const agent of getAgents?.() ?? []) { + try { + agent.edgeAdapter?.ws?.close(1001, 'fleet soak complete'); + } catch { + // best effort + } + removeAgent?.(agent.name); + } + } catch { + // best effort + } + for (const child of activeChildren()) { + child.kill('SIGTERM'); + } + if (mockDocker?.exitCode === null) { + mockDocker.kill('SIGTERM'); + } + await Promise.race([ + Promise.allSettled( + [...children.values(), ...(mockDocker ? [mockDocker] : [])].map( + (child) => + new Promise((resolvePromise) => { + if (child.exitCode !== null) { + resolvePromise(); + return; + } + child.once('exit', resolvePromise); + }), + ), + ), + sleep(3000), + ]); + for (const child of [...children.values(), ...(mockDocker ? [mockDocker] : [])]) { + if (child.exitCode === null) { + child.kill('SIGKILL'); + } + } + await new Promise((resolvePromise) => controller.close(() => resolvePromise())); + clearNonceCacheForTesting?.(); + rmSync(runDirectory, { recursive: true, force: true }); +} + +try { + await run(); +} catch (error) { + evidence.error = error instanceof Error ? (error.stack ?? error.message) : String(error); + process.exitCode = 1; +} finally { + evidence.finishedAt = new Date().toISOString(); + mkdirSync(dirname(options.output), { recursive: true }); + writeFileSync(options.output, `${JSON.stringify(evidence, null, 2)}\n`, { mode: 0o600 }); + await cleanup(); +} + +process.stdout.write(`${JSON.stringify(evidence, null, 2)}\n`); From 05cd169a38b485803a33df1910cee2140897214c Mon Sep 17 00:00:00 2001 From: scttbnsn <80784472+scttbnsn@users.noreply.github.com> Date: Tue, 28 Jul 2026 15:07:37 -0400 Subject: [PATCH 2/2] =?UTF-8?q?=F0=9F=94=92=20security(ci):=20keep=20fleet?= =?UTF-8?q?=20soak=20dispatch=20inputs=20closed?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../workflows/quality-portwing-fleet-soak.yml | 23 +++---------------- 1 file changed, 3 insertions(+), 20 deletions(-) diff --git a/.github/workflows/quality-portwing-fleet-soak.yml b/.github/workflows/quality-portwing-fleet-soak.yml index 4b53ded20..135b24736 100644 --- a/.github/workflows/quality-portwing-fleet-soak.yml +++ b/.github/workflows/quality-portwing-fleet-soak.yml @@ -8,19 +8,6 @@ run-name: >- on: workflow_dispatch: - inputs: - duration: - description: "Soak duration (for example 5m or 1h)" - required: false - default: "5m" - agents: - description: "Number of real Portwing processes" - required: false - default: "8" - exec_per_agent: - description: "Concurrent exec sessions per agent" - required: false - default: "3" pull_request: branches: [main, 'dev/**'] paths: @@ -93,10 +80,6 @@ jobs: - name: Resolve soak parameters id: parameters - env: - INPUT_DURATION: ${{ github.event.inputs.duration }} - INPUT_AGENTS: ${{ github.event.inputs.agents }} - INPUT_EXEC_PER_AGENT: ${{ github.event.inputs.exec_per_agent }} run: | set -euo pipefail if [ "${GITHUB_EVENT_NAME}" = "schedule" ]; then @@ -112,9 +95,9 @@ jobs: rss_threshold="134217728" heap_threshold="67108864" else - duration="${INPUT_DURATION:-5m}" - agents="${INPUT_AGENTS:-8}" - exec_per_agent="${INPUT_EXEC_PER_AGENT:-3}" + duration="5m" + agents="8" + exec_per_agent="3" rss_threshold="268435456" heap_threshold="134217728" fi