📚 课程目录第4章事件源开发与配置

4. 事件源开发与配置：内置事件源类型、自定义事件源开发、事件源与Kubernetes资源绑定

事件源（Event Source）是 Knative Eventing 的入口组件，负责将外部系统或集群内部产生的事件转换为 CloudEvents 标准格式，并转发到 Broker 或 Channel。本章围绕事件源的三种形态展开：Knative 内置的通用事件源、自定义事件源的开发模式，以及事件源与 Kubernetes 原生资源的绑定机制。

4.1 内置事件源类型与选型

Knative 官方维护了多个开箱即用的事件源，覆盖消息队列、定时任务、HTTP 请求、云服务事件等常见场景。下表列出生产环境中使用频率最高的内置事件源及其核心参数：

事件源名称	触发方式	适用场景	关键配置字段
ApiServerSource	Kubernetes API 事件	监听 Pod/Deployment/ConfigMap 等资源变更	resources, mode, serviceAccountName
PingSource	定时/周期触发	定时任务、周期性数据同步、健康检查	schedule, jsonData, timezone
KafkaSource	Kafka 消息消费	高吞吐事件流、微服务异步解耦	bootstrapServers, topics, consumerGroup
SinkBinding	绑定已有 Pod 为事件源	将现有工作负载快速接入 Eventing	subject, sink, ceOverrides
GitHubSource	Webhook 回调	CI/CD 流水线、代码仓库事件驱动	owner, repository, accessToken, eventTypes

选型建议： 在控制面资源调度场景中，ApiServerSource 是最核心的事件源。它能够实时捕获 Pod 调度失败、节点资源不足、PVC 绑定延迟等 Kubernetes 原生事件，是构建调度反馈回路的基础。PingSource 适合周期性资源巡检，KafkaSource 则用于对接已有的消息中间件事件流。

4.2 内置事件源配置示例：ApiServerSource

以下 YAML 配置创建一个 ApiServerSource，监听集群中所有 Namespace 下 Pod 的 Failed 和 Evicted 事件，并将事件发送到名为 default 的 Broker：

apiVersion: sources.knative.dev/v1
kind: ApiServerSource
metadata:
  name: pod-failure-source
spec:
  serviceAccountName: eventing-controller
  mode: Resource
  resources:
    - apiVersion: v1
      kind: Event
      selector:
        reason: "Failed|Evicted"
        involvedObject:
          apiVersion: v1
          kind: Pod
  sink:
    ref:
      apiVersion: eventing.knative.dev/v1
      kind: Broker
      name: default

  关键点：
  mode: Resource 表示监听资源对象本身的变化；mode: Ref 则只传递资源引用。
selector 支持按 reason、involvedObject 等字段过滤，避免事件洪泛。
必须为 ApiServerSource 关联一个具有 get、watch、list 权限的 ServiceAccount。

4.3 自定义事件源开发模式

当内置事件源无法满足特定协议或业务逻辑时，需要开发自定义事件源。Knative 提供了两种开发路径：

基于 Receive Adapter 模式：编写一个独立的 Adapter 容器，负责从外部系统拉取或接收事件，转换为 CloudEvents 后通过 HTTP POST 发送到 Sink。这是官方推荐的标准方式。
基于 SinkBinding 模式：如果已有现成的服务或 Pod 能够产生事件，只需通过 SinkBinding 为其注入 K_SINK 环境变量，即可将事件转发到目标 Sink，无需额外开发 Adapter。

下图展示了 Receive Adapter 模式的自定义事件源架构：

4.4 自定义事件源开发实战：Prometheus 告警事件源

以开发一个 Prometheus Alertmanager 事件源为例，展示自定义事件源的完整实现。该事件源监听 Alertmanager Webhook，将告警转换为 CloudEvents 并发送到 Broker。

步骤 1：定义 CustomResourceDefinition

apiVersion: apiextensions.k8s.io/v1
kind: CustomResourceDefinition
metadata:
  name: prometheussources.sources.example.com
spec:
  group: sources.example.com
  names:
    kind: PrometheusSource
    plural: prometheussources
  scope: Namespaced
  versions:
    - name: v1
      served: true
      storage: true
      schema:
        openAPIV3Schema:
          type: object
          properties:
            spec:
              type: object
              properties:
                alertmanagerURL:
                  type: string
                sink:
                  type: object
                  properties:
                    ref:
                      type: object
                      properties:
                        apiVersion: { type: string }
                        kind: { type: string }
                        name: { type: string }

步骤 2：实现 Receive Adapter 核心逻辑（Go 语言片段）

package adapter

import (
    "context"
    "net/http"
    cloudevents "github.com/cloudevents/sdk-go/v2"
)

type PrometheusAdapter struct {
    SinkURI string
    CE      cloudevents.Client
}

func (a *PrometheusAdapter) HandleAlert(w http.ResponseWriter, r *http.Request) {
    // 解析 Alertmanager Webhook 请求
    alert := parseAlert(r.Body)

    // 构建 CloudEvent
    event := cloudevents.NewEvent()
    event.SetSource("prometheus/alertmanager")
    event.SetType("prometheus.alert.firing")
    event.SetData(cloudevents.ApplicationJSON, alert)

    // 发送到 Sink
    ctx := cloudevents.ContextWithTarget(context.Background(), a.SinkURI)
    if result := a.CE.Send(ctx, event); !cloudevents.IsACK(result) {
        http.Error(w, "failed to send event", http.StatusInternalServerError)
        return
    }
    w.WriteHeader(http.StatusOK)
}

开发注意事项：

Receive Adapter 必须从环境变量 K_SINK 读取目标地址，不要硬编码。
事件源 CRD 的 spec.sink 字段支持 ref（引用 Knative 资源）和 uri（直接 URL）两种模式。
建议在 Adapter 中实现健康检查端点 /healthz 和就绪检查端点 /readyz。

4.5 事件源与 Kubernetes 资源绑定

事件源与 Kubernetes 资源的绑定通过 SinkBinding 和 ApiServerSource 的 resources 字段实现。绑定机制确保事件源能够感知目标资源的生命周期变化，并自动调整事件路由。

下表对比了两种绑定模式：

绑定方式	绑定对象	注入方式	典型用途
SinkBinding	任意 Pod / Deployment / StatefulSet	环境变量 `K_SINK` + `K_CE_OVERRIDES`	将现有工作负载快速变为事件源
ApiServerSource.resources	Kubernetes 原生资源（Pod/Node/Event 等）	API Server Watch 机制	监听集群资源状态变更

SinkBinding 绑定示例： 将已有的日志采集 Pod 绑定到事件系统，使其产生的日志事件自动发送到 Broker。

apiVersion: sources.knative.dev/v1
kind: SinkBinding
metadata:
  name: log-collector-binding
spec:
  subject:
    apiVersion: apps/v1
    kind: Deployment
    selector:
      matchLabels:
        app: log-collector
  sink:
    ref:
      apiVersion: eventing.knative.dev/v1
      kind: Broker
      name: default
  ceOverrides:
    extensions:
      source-type: "log-collector"

绑定生效后，Knative 会自动在目标 Pod 的环境变量中注入 K_SINK=http://broker-ingress.knative-eventing.svc.cluster.local/default/default，应用只需读取该环境变量即可发送事件。

4.6 事件源生命周期管理与调试

事件源作为 Knative Eventing 的自定义资源，其生命周期遵循 Kubernetes 控制器模式。以下命令用于调试事件源状态：

# 查看所有事件源
kubectl get podsources --all-namespaces

# 查看 ApiServerSource 详细状态
kubectl get apiserversource pod-failure-source -o yaml

# 查看事件源对应的 Receive Adapter Pod
kubectl get pods -l eventing.knative.dev/sourceName=pod-failure-source

# 查看事件源发送的事件是否到达 Broker
kubectl get events --field-selector involvedObject.kind=Broker

调试技巧： 如果事件源未按预期发送事件，首先检查 Receive Adapter Pod 的日志：kubectl logs -l eventing.knative.dev/sourceName=<source-name> --tail=50。常见问题包括：ServiceAccount 权限不足、Sink 地址不可达、CloudEvents 格式不匹配。

下图总结了事件源从创建到事件投递的完整生命周期：

4.7 生产环境最佳实践

事件源幂等性： 确保事件源在重启或网络抖动后不会重复发送相同事件，或在消费者侧实现去重逻辑。
资源隔离： 为不同优先级的事件源设置独立的 Receive Adapter Deployment，避免相互影响。
事件过滤前置： 在事件源侧使用 selector 或 ceOverrides 进行初步过滤，减少下游 Broker 和 Trigger 的负载。
监控与告警： 为事件源暴露 Prometheus 指标（事件发送总数、失败数、延迟），并设置告警规则。
版本管理： 自定义事件源的 CRD 和 Adapter 镜像应遵循语义化版本，升级时注意向后兼容。