Samply.Beam 分布式通信引擎

!Logo

镜像概述和主要用途

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

该镜像包含Samply.Beam的代理组件，作为分布式通信引擎的一部分，用于在受严格安全限制的网络环境中实现可靠、高性能的通信。

核心功能和特性

• 分布式任务代理：跨网络边界的任务分发与结果收集
• 安全通信：端到端加密和签名，确保数据传输安全性
• 证书管理：集成证书颁发机构(CA)，处理证书的生成、验证和更新
• 灵活的通信模式：支持多种常见的分布式计算通信模式
• REST API：简单易用的API接口，便于应用程序集成
• 长轮询支持：高效的结果等待机制，减少网络流量
• 服务器发送事件(SSE)：实验性支持，适用于异步通知场景
• 健康检查：内置健康检查端点，便于系统监控
• 高安全性：专为医院等高强度安全环境设计，符合严格的网络安全要求

使用场景和适用范围

Samply.Beam特别适用于以下场景：

• 高安全性网络环境：如医院内部网络、***机构网络等
• 严格防火墙限制：禁止入站连接或需要通过HTTP代理的网络
• 分布式计算：需要跨多个节点协调工作的场景
• 联邦应用集成：不同组织或部门间的应用程序互联
• 数据交换：需要严格遵循隐私和安全规定的数据传输
• 跨组织协作：不同机构间需要安全交换信息的场景

系统架构

!Architecture Schema

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

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

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

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

使用方法和配置说明

基本概念

• 任务(Task)：需要完成的工作单元，由发起者创建并发送给一个或多个接收者
• 结果(Result)：任务处理完成后返回的结果
• BeamId：系统中每个组件的唯一标识符
• API Key：用于身份验证的密钥

任务生命周期

1. 创建任务
2. 监听相关任务
3. 声明任务处理
4. 返回处理结果
5. 等待任务完成

API 接口

创建任务

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

• 方法: POST
• URL: /v1/tasks
• 请求体: 任务对象(参见下方任务结构)
• 参数: 无

请求示例:

bash
curl -k -v --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" [***]

响应:

HTTP/1.1 201 Created
Location: /tasks/70c0aa90-bfcf-4312-a6af-42cbd57dc0b8
Content-Length: 0
Date: Mon, 27 Jun 2022 13:58:35 GMT

检索任务

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

• 方法: GET
• URL: /v1/tasks
• 参数:
  • from (可选): 仅获取由此ID创建的任务
  • to (可选): 仅获取定向到此ID的任务
  • filter (可选): 仅获取满足指定过滤条件的任务，支持filter=todo获取未完成任务

请求示例:

bash
curl -k -X GET -v -H "Authorization: ApiKey app.proxy2.broker.example.de AppKey" [***]

响应: 返回任务对象数组

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>
• 请求体: 结果对象(参见下方结果结构)
• 参数: 无

请求示例:

bash
curl -k -X PUT -v --json '{"from":"app.proxy2.broker.example.de","metadata":["任意","类型","都是","可能的"],"status":"succeeded","body":"处理结果","task":"70c0aa90-bfcf-4312-a6af-42cbd57dc0b8","to":["app.proxy1.broker.example.de"]}' -H "Authorization: ApiKey app.proxy2.broker.example.de AppKey" [***]

响应:

HTTP/1.1 204 No Content
Content-Length: 0
Date: Mon, 27 Jun 2022 13:58:35 GMT

检索结果

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

• 方法: GET
• URL: /v1/tasks/<task_id>/results
• 参数: 支持长轮询参数

请求示例:

bash
curl -k -X GET -v -H "Authorization: ApiKey app.proxy1.broker.example.de AppKey" [***]

响应: 返回结果对象数组

json
[
  {
    "from": "app.proxy2.broker.example.de",
    "to": ["app.proxy1.broker.example.de"],
    "task": "70c0aa90-bfcf-4312-a6af-42cbd57dc0b8",
    "status": "succeeded",
    "body": "处理结果",
    "metadata": ["任意","类型","都是","可能的"]
  }
]

健康检查

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

• 方法: GET
• URL: /v1/health
• 参数: 无

请求示例:

bash
curl -k -X GET [***]

响应 (Proxy):

HTTP/1.1 200 OK
Content-Length: 0
Date: Mon, 27 Jun 2022 14:26:45 GMT

响应 (Broker):

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

JSON 数据对象

任务(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": "broker可以读取并使用此字段，例如代表应用应用过滤器"
}

• id: 用于标识任务的UUID
• from: 提交应用程序的BeamID
• to: 允许检索任务和提交结果的工作者的BeamID
• body: 要完成的工作描述，Broker不解释
• failure_strategy: 建议每个客户端如何处理失败，可能的值为discard(丢弃)、retry(重试)
• failure_strategy.retry: 重试失败任务的次数(max_tries)和每次尝试之间的等待时间(backoff_millisecs)
• ttl: 生存时间，达到零后，broker将删除任务及其结果
• metadata: 可由broker读取的关联数据，可以是任意类型

结果(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外的所有状态都需要，包含任务的实际结果 payload 或错误消息
• metadata: 可由broker读取的关联数据，可以是任意类型

高级特性

长轮询API访问

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

• wait_count: API调用将阻塞，直到有这么多结果可用
• wait_time: 或直到此时间过去(如果没有特别说明，单位为秒)，以先到者为准

示例:

bash
# 阻塞直到有5个结果可用或30秒过去
curl -k -X GET -H "Authorization: ApiKey app.proxy1.broker.example.de AppKey" [***]

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

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

请求示例:

bash
curl -k -X GET -H "Authorization: ApiKey app.proxy1.broker.example.de AppKey" -H "Accept: text/event-stream" [***]

响应:

HTTP/1.1 200 OK
Content-Type: text/event-stream
Cache-Control: no-cache
Transfer-Encoding: chunked
Date: Thu, 09 Mar 2023 16:28:47 GMT

event: new_result
data: {"body":"处理结果","from":"app2.proxy1.broker","metadata":{"complex":"可以是复杂类型"},"status":"succeeded","task":"70c0aa90-bfcf-4312-a6af-42cbd57dc0b8","to":["app1.proxy1.broker"]}

...

Docker部署方案

Docker Run 命令

bash
# 运行Samply.Beam Proxy
docker run -d \
  --name samply-beam-proxy \
  -p 8080:8080 \
  -e RUST_LOG=info \
  -e PROXY_ID=proxy1.broker.example.de \
  -e BROKER_URL=[***] \
  -v /path/to/privkey.pem:/etc/beam/privkey.pem \
  -v /path/to/rootcert.pem:/etc/beam/rootcert.pem \
  samply/beam-proxy:latest \
  --privkey-file /etc/beam/privkey.pem \
  --rootcert-file /etc/beam/rootcert.pem

Docker Compose 配置

yaml
version: '3.8'

services:
  broker:
    image: samply/beam-broker:latest
    container_name: samply-beam-broker
    ports:
      - "8081:8080"
    environment:
      - RUST_LOG=info
      - BROKER_ID=broker.example.de
      - VAULT_URL=[***]
      - VAULT_TOKEN=your-vault-token
      - DATABASE_URL=postgres://user:password@db:5432/beam
    depends_on:
      - db
      - vault
    restart: unless-stopped

  proxy1:
    image: samply/beam-proxy:latest
    container_name: samply-beam-proxy1
    ports:
      - "8082:8080"
    environment:
      - RUST_LOG=info
      - PROXY_ID=proxy1.broker.example.de
      - BROKER_URL=[***]
    volumes:
      - ./proxy1/privkey.pem:/etc/beam/privkey.pem
      - ./proxy1/rootcert.pem:/etc/beam/rootcert.pem
    command: --privkey-file /etc/beam/privkey.pem --rootcert-file /etc/beam/rootcert.pem
    depends_on:
      - broker
    restart: unless-stopped

  proxy2:
    image: samply/beam-proxy:latest
    container_name: samply-beam-proxy2
    ports:
      - "8083:8080"
    environment:
      - RUST_LOG=info
      - PROXY_ID=proxy2.broker.example.de
      - BROKER_URL=[***]
    volumes:
      - ./proxy2/privkey.pem:/etc/beam/privkey.pem
      - ./proxy2/rootcert.pem:/etc/beam/rootcert.pem
    command: --privkey-file /etc/beam/privkey.pem --rootcert-file /etc/beam/rootcert.pem
    depends_on:
      - broker
    restart: unless-stopped

  db:
    image: postgres:14
    container_name: samply-beam-db
    environment:
      - POSTGRES_USER=user
      - POSTGRES_PASSWORD=password
      - POSTGRES_DB=beam
    volumes:
      - postgres_data:/var/lib/postgresql/data
    restart: unless-stopped

  vault:
    image: hashicorp/vault:latest
    container_name: samply-beam-vault
    ports:
      - "8200:8200"
    environment:
      - VAULT_DEV_ROOT_TOKEN_ID=your-vault-token
      - VAULT_DEV_LISTEN_ADDRESS=0.0.0.0:8200
    cap_add:
      - IPC_LOCK