Connector

Connector 是与外部平台的托管集成，支持两个方向：

接入源 —— 平台把数据推进 MoleSignal（Kinesis Firehose、Cloudflare Logpush、Heroku log drain）。
出向 sink —— 流水线把产出投递出去到 S3 或 Kafka。

配置一个 connector

在 Pipelines → Connectors 下创建 connector，或通过 API。每个 connector 有一个 kind 和一个 config_json，后者的结构随 kind 而定。敏感字段（push_token、access_key、secret_key …）在 API 响应中被掩码。

curl -X POST http://localhost:5080/api/v1/connectors \
  -H "authorization: Bearer $MS_JWT" \
  -H 'content-type: application/json' \
  -d '{
        "name": "cf-prod",
        "kind": "cloudflare_logpush",
        "config_json": { "push_token": "ms_cf_9f3c…", "target_stream": "cloudflare" }
      }'

接入源（推送）

三种推送源共用一套模型：创建一个带 push_token 和 target_stream 的 connector，然后把平台指向对应端点。这些端点不使用 JWT——请求用 connector token 鉴权，按平台允许的方式传入：

方式	适用平台
`X-Connector-Token` 头	可配置自定义头的平台（如 Cloudflare）
`X-Amz-Firehose-Access-Key` 头	Kinesis Firehose（自动发送）
`?token=` 查询参数	无法设置自定义头的平台（如 Heroku）

带 Content-Encoding: gzip 的 body 会被自动解压。事件走与其他接入相同的路径——schema-on-write、流水线、脱敏一并生效。

Kinesis Firehose
Cloudflare Logpush
Heroku

用 kind: "aws_kinesis_firehose" 创建 connector，然后给 Firehose 投递流添加一个 HTTP endpoint 目标：

URL —— https://molesignal.example.com/api/v1/_kinesis_firehose
Access key —— connector 的 push_token（Firehose 以 X-Amz-Firehose-Access-Key 发送）

MoleSignal 对每条 record base64 解码、按换行切分，并把每行解析为 JSON（纯文本则回退到 message 字段），并返回 Firehose 所要求的 200 ACK。

用 kind: "cloudflare_logpush" 创建 connector，然后创建一个 Logpush job，目标为：

URL —— https://molesignal.example.com/api/v1/_cloudflare
Header —— X-Connector-Token: <push_token>

Cloudflare 发送 gzip 压缩的 NDJSON，每行成为一条事件。端点回 204 No Content。

用 kind: "heroku_drain" 创建 connector，然后添加 drain——Heroku 设不了自定义头，故 token 放在 URL 里：

heroku drains:add "https://molesignal.example.com/api/v1/_heroku?token=<push_token>"

每行日志成为一条带 message 和 source: "heroku" 的事件。端点回 204 No Content。

target_stream 在首次投递时创建，schema 随新字段出现而演化，与任意其他数据流一致。省略 target_stream 则回退到按源区分的默认流名（kinesis、cloudflare、heroku）。

拉取源

CloudWatch Logs 是拉取源：MoleSignal 按周期主动轮询 AWS API，而非被动接收推送。创建一个 connector 后，MoleSignal 周期调用 FilterLogEvents（SigV4 签名，不引 AWS SDK），并按 connector 维度推进 checkpoint，使每轮只读新事件。

curl -X POST http://localhost:5080/api/v1/connectors \
  -H "authorization: Bearer $MS_JWT" \
  -H 'content-type: application/json' \
  -d '{
        "name": "cw-prod",
        "kind": "aws_cloudwatch_logs",
        "config_json": {
          "region": "ap-southeast-1",
          "log_group": "/aws/lambda/checkout",
          "access_key": "AKIA…",
          "secret_key": "…",
          "target_stream": "cloudwatch"
        }
      }'

字段	说明
`region` / `log_group`	必填——要读取的 CloudWatch 区域与 log group。
`access_key` / `secret_key`	具备 `logs:FilterLogEvents` 权限的 IAM 凭据。
`session_token`	可选，用于 STS 临时凭据。
`target_stream`	目标 stream（默认 `default`）。

轮询器以单例运行（alert-manager / standalone 节点），无论集群规模都只有一个节点轮询。每条事件成为一条带 message、log_stream、log_group 的日志。

出向 sink

S3 与 Kafka connector 是 sink：在流水线中选中对应 connector，即可在流水线运行时把变换后的事件投递出去。

Kind	载荷
`s3`	批量 JSON Lines `PUT` 到 `s3://<bucket>/<prefix><id>.jsonl`。
`kafka`	每条事件作为一条 JSON 记录 produce 到指定 topic。

权限

pipelines.read 用于列出和检查 Connector。pipelines.edit 用于创建、更新或删除 Connector。推送 Receiver 使用 Connector 专用 Token 身份验证，不能用普通登录 JWT 替代。

接入 API

Kinesis、Cloudflare、Heroku 推送接收端点的细节。

快速上手

观察

可靠性

AI 运维

数据

协作

管理

部署与运维

配置一个 connector

接入源（推送）

拉取源

出向 sink

权限

接入 API

​配置一个 connector

​接入源（推送）

​拉取源

​出向 sink

​权限

接入 API

配置一个 connector

接入源（推送）

拉取源

出向 sink

权限