samply/beam-broker

Samply.Beam 分布式通信引擎

!https://raw.githubusercontent.com/samply/beam/main/doc/Logo.svg

镜像概述和主要用途

Samply.Beam是一个分布式任务代理，专为在严格的网络环境中进行高效通信而设计。它提供跨严格网络边界的常用通信模式、端到端加密和签名，以及基于易用REST API的证书管理和验证功能。

核心功能和特性

• 分布式通信层：提供可靠、高性能的跨网络通信能力
• 网络边界穿透：优化在严格网络环境（如医院内网）中的通信
• 安全特性：
  • 端到端加密和签名
  • 证书管理和验证
  • 基于BeamId的身份识别和认证
• 通信模式：支持常见分布式计算通信模式
• API接口：简单易用的REST API
• 灵活的任务处理：
  • 任务失败策略（重试、丢弃）
  • 任务过期时间(TTL)管理
  • 结果状态跟踪（已认领、成功、临时失败、永久失败）
• 高级通信功能：
  • 长轮询(Long Polling)支持
  • 服务器发送事件(SSE)支持（实验性）
• 健康检查：内置健康状态监控接口

使用场景和适用范围

Samply.Beam特别适合以下场景：

• 高安全性网络环境：如医院内网、***机构网络等严格受限环境
• 分布式计算系统：需要跨节点可靠通信的联邦应用
• 限制性网络边界：存在严格防火墙规则、禁止入站连接或使用特殊HTTP代理的网络
• 多机构协作：不同组织或站点间需要安全通信的场景
• 替代复杂解决方案：作为消息队列、***覆盖网络等复杂方案的轻量级替代

系统架构

!https://raw.githubusercontent.com/samply/beam/main/doc/Architecture.svg

Samply.Beam由两个中央运行组件和每个分布式节点上的一个代理组成：

• Samply.Broker：中央组件，负责促进连接、存储和转发任务与消息，并与中央证书机构通信
• Samply.Proxy：本地代理，处理与broker的所有通信，以及身份验证、加密和签名
• 证书机构：基于Hashicorp Vault的实例，管理所有用于签名和加密有效负载所需的证书

身份标识：BeamId

系统中的每个组件都通过其分层的BeamId唯一标识：

app3.proxy2.broker1.samply.de
<--------------------------->
            AppId
     <---------------------->
             ProxyId
            <--------------->
                 BrokerId

• BrokerId：必须是可通过DNS解析的FQDN（完全限定域名）
• ProxyId：通过代理证书表示，格式为CN=proxy2.broker2.samply.de
• AppId：通过与代理通信时使用的API密钥确定

使用方法和配置说明

基本概念

Samply.Beam基于任务和结果的通信模型：

• 任务(Task)：由应用创建，包含要执行的工作描述
• 结果(Result)：由任务接收方创建，包含任务执行状态和结果

数据对象结构

任务(Task)

json
{
  "id": "70c0aa90-bfcf-4312-a6af-42cbd57dc0b8",
  "from": "app7.proxy-hd.broker-project1.samply.de",
  "to": [
    "app1.proxy-hd.broker-project1.samply.de",
    "app5.proxy-ma.broker-project1.samply.de"
  ],
  "body": "需要完成的工作描述",
  "failure_strategy": {
    "retry": {
      "backoff_millisecs": 1000,
      "max_tries": 5
    }
  },
  "ttl": "30s",
  "metadata": "代理可读取并使用此字段，例如代表应用应用过滤器"
}

字段说明：

• id：任务的UUID标识符
• from：提交应用的BeamID（由Proxy根据身份验证信息自动设置）
• to：允许检索任务并提交结果的工作者BeamID列表
• body：工作描述，Broker不解释此内容
• failure_strategy：建议客户端如何处理失败，可选值：discard（丢弃）、retry（重试）
• failure_strategy.retry：重试策略，max_tries（最大重试次数）和backoff_millisecs（重试间隔毫秒数）
• ttl：生存时间，超时后Broker将删除任务及其结果
• metadata：代理可读取的关联数据，可为任意类型

结果(Result)

json
{
  "from": "app1.proxy-hd.broker-project1.samply.de",
  "to": [
    "app7.proxy-hd.broker-project1.samply.de"
  ],
  "task": "70c0aa90-bfcf-4312-a6af-42cbd57dc0b8",
  "status": "succeeded",
  "body": "成功完成的结果数据",
  "metadata": ["任意", "数据", "类型", "都是", "允许的"]
}

字段说明：

• from：提交结果的客户端BeamID
• to：结果的预期接收者BeamID
• task：关联任务的UUID
• status：工作结果状态，允许值：claimed（已认领）、tempfailed（临时失败）、permfailed（永久失败）、succeeded（成功）
• body：结果负载，除claimed状态外，所有状态都需要此字段
• metadata：代理可读取的关联数据，可为任意类型且不加密

API使用说明

创建任务

创建新任务，由指定的工作者处理。当前body大小限制为10MB。

• 方法: POST
• URL: /v1/tasks
• 请求体: 任务对象
• 认证头: Authorization: ApiKey <BeamID> <ApiKey>

示例：

bash
curl -k --json '{"body":"生命、宇宙和一切的终极问题的答案是什么？","failure_strategy":{"retry":{"backoff_millisecs":1000,"max_tries":5}},"from":"app.proxy1.broker.example.de","id":"70c0aa90-bfcf-4312-a6af-42cbd57dc0b8","metadata":"代理可读取此字段","to":["app.proxy2.broker.example.de"],"ttl":"60s"}' \
  -H "Authorization: ApiKey app.proxy1.broker.example.de AppKey" \
  https://proxy1.broker.example.de/v1/tasks

响应：

HTTP/1.1 201 Created
Location: /tasks/70c0aa90-bfcf-4312-a6af-42cbd57dc0b8
Content-Length: 0

注意：服务器可能忽略提交的id属性并自动生成自己的ID，应使用响应中Location头指定的URL来引用任务。

获取任务

工作者定期调用此端点来检索提交的任务。

• 方法: GET
• URL: /v1/tasks
• 参数:
  • from (可选): 仅获取由此ID创建的任务
  • to (可选): 仅获取定向到此ID的任务
  • filter (可选): 仅获取满足指定过滤条件的任务
    • filter=todo: 匹配要由查询客户端处理的未完成任务
• 认证头: Authorization: ApiKey <BeamID> <ApiKey>

示例：

bash
curl -k -H "Authorization: ApiKey app.proxy2.broker.example.de AppKey" \
  https://proxy2.broker.example.de/v1/tasks?filter=todo

响应：

json
[
  {
    "id": "70c0aa90-bfcf-4312-a6af-42cbd57dc0b8",
    "from": "app.proxy1.broker.example.de",
    "to": ["app.proxy2.broker.example.de"],
    "body": "生命、宇宙和一切的终极问题的答案是什么？",
    "failure_strategy": {"retry":{"backoff_millisecs":1000,"max_tries":5}},
    "ttl": "60s",
    "metadata": "代理可读取此字段"
  }
]

创建结果

创建或更新任务的结果。当前body大小限制为10MB。

• 方法: PUT
• URL: /v1/tasks/<task_id>/results/<app_id>
• 请求体: 结果对象
• 认证头: Authorization: ApiKey <BeamID> <ApiKey>

示例：

bash
curl -k -X PUT --json '{"from":"app.proxy2.broker.example.de","metadata":["任意","数据类型","都是","允许的"],"status":"succeeded","body":"答案是42","task":"70c0aa90-bfcf-4312-a6af-42cbd57dc0b8","to":["app.proxy1.broker.example.de"]}' \
  -H "Authorization: ApiKey app.proxy2.broker.example.de AppKey" \
  https://proxy2.broker.example.de/v1/tasks/70c0aa90-bfcf-4312-a6af-42cbd57dc0b8/results/app.proxy2.broker.example.de

响应：

HTTP/1.1 204 No Content
Content-Length: 0

获取结果

任务提交者调用此端点来检索结果。

• 方法: GET
• URL: /v1/tasks/<task_id>/results
• 认证头: Authorization: ApiKey <BeamID> <ApiKey>

示例：

bash
curl -k -H "Authorization: ApiKey app.proxy1.broker.example.de AppKey" \
  https://proxy1.broker.example.de/v1/tasks/70c0aa90-bfcf-4312-a6af-42cbd57dc0b8/results

响应：

json
[
  {
    "from": "app.proxy2.broker.example.de",
    "to": ["app.proxy1.broker.example.de"],
    "task": "70c0aa90-bfcf-4312-a6af-42cbd57dc0b8",
    "status": "succeeded",
    "body": "答案是42",
    "metadata": ["任意","数据类型","都是","允许的"]
  }
]

长轮询API访问

所有读取端点都支持长轮询，作为常规（重复）轮询的高效替代方案。

• 参数:
  • wait_count: API调用将阻塞，直到有这么多结果可用
  • wait_time: 阻塞的最长时间（如果未指定单位，默认为秒）

示例：

bash
# 阻塞直到有1个结果可用或30秒过去
curl -k -H "Authorization: ApiKey app.proxy1.broker.example.de AppKey" \
  https://proxy1.broker.example.de/v1/tasks/70c0aa90-bfcf-4312-a6af-42cbd57dc0b8/results?wait_count=1&wait_time=30s

服务器发送事件(SSE) API (实验性)

为了更好地支持异步用例，开发版本支持服务器发送事件(Server-Sent Events)用于结果检索。

• 方法: GET
• URL: /v1/tasks/<task_id>/results?wait_count=3
• 头: Accept: text/event-stream
• 参数: 与长轮询相同的参数

示例：

bash
curl -k -H "Accept: text/event-stream" \
  -H "Authorization: ApiKey app.proxy1.broker.example.de AppKey" \
  https://proxy1.broker.example.de/v1/tasks/70c0aa90-bfcf-4312-a6af-42cbd57dc0b8/results?wait_count=3

响应：

event: new_result
data: {"body":"答案是42","from":"app.proxy2.broker.example.de","metadata":["任意","数据类型","都是","允许的"],"status":"succeeded","task":"70c0aa90-bfcf-4312-a6af-42cbd57dc0b8","to":["app.proxy1.broker.example.de"]}

健康检查

监控Samply.Beam组件的运行状态。

• 方法: GET
• URL: /v1/health

Proxy响应：

HTTP/1.1 200 OK
Content-Length: 0

Broker响应：

json
{
  "summary": "healthy",
  "vault": {
    "status": "ok"
  }
}

Docker部署方案

开发环境

开发环境包含一个broker和两个proxy：

bash
# 启动开发环境
./dev/beamdev start

# 或在后台运行服务
./dev/beamdev start_bg
docker-compose logs -f

Docker Run示例

启动Broker

bash
docker run -d \
  --name beam-broker \
  -p 8080:8080 \
  -e RUST_LOG=info \
  -e BROKER_ID=broker.example.de \
  -e VAULT_ADDR=http://vault:8200 \
  -e VAULT_TOKEN=your-vault-token \
  samply/beam-broker:latest

启动Proxy

bash
docker run -d \
  --name beam-proxy \
  -p 8081:8080 \
  -e RUST_LOG=info \
  -e PROXY_ID=proxy1.broker.example.de \
  -e BROKER_URL=http://broker:8080 \
  -v ./privkey.pem:/app/privkey.pem \
  -v ./rootcert.pem:/app/rootcert.pem \
  samply/beam-proxy:latest

Docker Compose示例

yaml
version: '3.8'

services:
  vault:
    image: hashicorp/vault:latest
    ports:
      - "8200:8200"
    environment:
      - VAULT_DEV_ROOT_TOKEN_ID=dev-root-token
      - VAULT_DEV_LISTEN_ADDRESS=0.0.0.0:8200
    cap_add:
      - IPC_LOCK
    command: server -dev

  broker:
    image: samply/beam-broker:latest
    depends_on:
      - vault
    ports:
      - "8080:8080"
    environment:
      - RUST_LOG=info
      - BROKER_ID=broker.example.de
      - VAULT_ADDR=http://vault:8200
      - VAULT_TOKEN=dev-root-token
    restart: unless-stopped

  proxy1:
    image: samply/beam-proxy:latest
    depends_on:
      - broker
    ports:
      - "8081:8080"
    environment:
      - RUST_LOG=info
      - PROXY_ID=proxy1.broker.example.de
      - BROKER_URL=http://broker:8080
      - ROOTCERT_FILE=/app/rootcert.pem
      - PRIVKEY_FILE=/app/privkey.pem
    volumes:
      - ./proxy1/privkey.pem:/app/privkey.pem
      - ./proxy1/rootcert.pem:/app/rootcert.pem
    restart: unless-stopped

  proxy2:
    image: samply/beam-proxy:latest
    depends_on:
      - broker
    ports:
      - "8082:8080"
    environment:
      - RUST_LOG=info
      - PROXY_ID=proxy2.broker.example.de
      - BROKER_URL=http://broker:8080
      - ROOTCERT_FILE=/app/rootcert.pem
      - PRIVKEY_FILE=/app/privkey.pem
    volumes:
      - ./proxy2/privkey.pem:/app/privkey.pem
      - ./proxy2/rootcert.pem:/app/rootcert.pem
    restart: unless-stopped

配置参数和环境变量

Broker环境变量