4. 事件源基础：Scaler 接口规范、事件源注册机制、自定义 Scaler 开发概述

KEDA 的核心价值在于其丰富的事件源生态。本章将深入剖析 KEDA 事件源的底层架构，包括 Scaler 接口规范、事件源注册机制以及自定义 Scaler 开发的完整流程。理解这些基础是后续配置流水线触发器的前提。

4.1 Scaler 接口规范：事件源的契约

在 KEDA 中，每一个事件源（如 Kafka、Prometheus、AWS SQS）都被抽象为一个 Scaler。所有 Scaler 都必须实现 KEDA 定义的 Scaler 接口，这是 KEDA 与外部系统交互的契约。

Scaler 接口定义在 KEDA 的 pkg/scalers 包中，核心方法如下：

方法签名 功能描述 返回值说明
GetMetricSpecForScaling(context.Context) []MetricSpec 返回 Scaler 暴露的指标名称和类型。KEDA 根据此信息向 HPA 注册指标。 返回 MetricSpec 切片，每个元素包含指标名（如 kafka-lag）和指标类型（如 ExternalMetric）。
GetMetrics(context.Context, MetricIdentifier, *ScalerConfig) ([]ExternalMetricValue, error) 核心方法。KEDA 定期调用此方法获取当前指标值（如 Kafka 消费者滞后数）。 返回 ExternalMetricValue 切片，包含指标名称、时间戳和当前值。
Run(ctx context.Context, activeThreshold int) (bool, error) 判断事件源是否处于“活跃”状态。用于决定是否将 Deployment 从 0 扩展到 1。 返回 bool：true 表示有事件需要处理，应保持至少一个 Pod 运行。
Close(context.Context) error 清理资源。当 Scaler 被销毁时调用，用于关闭连接、释放句柄。 返回 error，表示清理是否成功。
关键理解： GetMetrics 返回的指标值直接驱动 HPA 的扩缩容决策；而 Run 方法则决定了 KEDA 的“从零到一”缩放能力。两者缺一不可。

4.2 事件源注册机制：Scaler 的发现与加载

KEDA 采用 插件式架构 来管理 Scaler。所有内置 Scaler 在 KEDA Operator 启动时自动注册。注册机制的核心是 Scaler 工厂模式。

下图展示了 Scaler 的注册与调用流程：

KEDA Scaler 注册与调用流程 Scaler 注册与调用流程 ScaledObject 控制器 Scaler 工厂 (Factory) Scaler 实例 (如 KafkaScaler) GetMetrics / Run 调用 注册机制 1. 启动时扫描内置 Scaler 2. 根据 ScaledObject 的 trigger 类型匹配工厂 3. 工厂创建 Scaler 实例

注册流程的关键步骤：

  1. ScaledObject 解析：KEDA 控制器监听到 ScaledObject 资源变更，解析其 triggers 字段。
  2. 工厂匹配：根据 triggers[].type（如 kafka、prometheus）查找已注册的 Scaler 工厂。
  3. 实例化：工厂解析 triggers[].metadata 和 authenticationRef，创建具体的 Scaler 实例。
  4. 生命周期管理：KEDA 持有 Scaler 实例，定期调用 GetMetrics 和 Run，并在 ScaledObject 删除时调用 Close。

4.3 自定义 Scaler 开发概述

当内置 Scaler 无法满足需求时（如对接自研消息队列、内部监控系统），可以开发自定义 Scaler。KEDA 提供了两种方式：外部 Scaler (External Scaler) 和 内置 Scaler (Built-in Scaler)。

4.3.1 两种开发方式对比

特性 外部 Scaler (External Scaler) 内置 Scaler (Built-in Scaler)
部署方式 独立 gRPC 服务，作为 Sidecar 或独立 Deployment 运行 编译进 KEDA Operator 二进制文件
开发语言 任意语言（通过 gRPC 协议通信） Go 语言（必须遵循 KEDA 代码库规范）
维护成本 低，独立迭代，不影响 KEDA 核心 高，需要与 KEDA 版本同步，提交 PR 合并
性能 中等，有 gRPC 网络开销 高，进程内调用，无网络延迟
适用场景 快速原型、私有协议、非 Go 技术栈 通用性强、性能敏感、社区贡献

4.3.2 外部 Scaler 开发核心步骤

外部 Scaler 通过 gRPC 暴露 ExternalScaler 服务，KEDA 作为客户端调用。核心接口定义在 keda.external.scaler.externalscaler.proto 中：

// 核心 gRPC 服务定义
service ExternalScaler {
    // 返回 Scaler 的指标规范
    rpc GetMetricSpec(ScaledObjectRef) returns (GetMetricSpecResponse) {}
    // 获取当前指标值
    rpc GetMetrics(GetMetricsRequest) returns (GetMetricsResponse) {}
    // 判断是否活跃
    rpc IsActive(ScaledObjectRef) returns (IsActiveResponse) {}
    // 关闭 Scaler，清理资源
    rpc Close(ScaledObjectRef) returns (CloseResponse) {}
}

开发一个外部 Scaler 的典型流程：

  1. 定义 proto 文件：基于 KEDA 的 externalscaler.proto 生成服务端代码。
  2. 实现服务接口：实现 GetMetricSpec、GetMetrics、IsActive、Close 四个方法。
  3. 部署 gRPC 服务：将服务部署为 Kubernetes Deployment，并暴露 Service。
  4. 配置 ScaledObject：在 triggers 中使用 type: external，并指定 metadata.scalerAddress 指向 gRPC 服务。
最佳实践： 对于流水线构建触发器场景，推荐优先评估外部 Scaler 方案。它允许团队使用熟悉的语言（如 Python、Java）快速对接 Jenkins、GitLab CI 等系统的 Webhook 或 API，且不影响 KEDA 核心的稳定性。

4.3.3 内置 Scaler 开发核心步骤

内置 Scaler 需要直接修改 KEDA 源码。核心步骤包括：

  1. 创建 Scaler 文件：在 pkg/scalers/ 目录下创建 my_scaler.go，实现 Scaler 接口。
  2. 实现工厂函数：编写 NewMyScaler(config *ScalerConfig) (Scaler, error) 工厂函数。
  3. 注册 Scaler：在 pkg/scaling/scale_handler.go 的 buildScaler 方法中添加 case "my-scaler": 分支。
  4. 编译测试：重新编译 KEDA Operator 镜像，部署到集群验证。

内置 Scaler 的代码骨架示例：

package scalers

import (
    "context"
    "fmt"
    v2 "k8s.io/api/autoscaling/v2"
    "k8s.io/metrics/pkg/apis/external_metrics/v1beta1"
)

type MyScaler struct {
    // 自定义字段，如客户端连接、配置参数
    client *MyCustomClient
    threshold int
}

func NewMyScaler(config *ScalerConfig) (Scaler, error) {
    // 1. 解析 metadata
    threshold, err := config.GetInt("threshold")
    if err != nil {
        return nil, fmt.Errorf("missing threshold: %w", err)
    }
    // 2. 初始化客户端
    client, err := NewMyCustomClient(config)
    if err != nil {
        return nil, err
    }
    return &MyScaler{client: client, threshold: threshold}, nil
}

func (s *MyScaler) GetMetricSpecForScaling(ctx context.Context) []v2.MetricSpec {
    metricName := "my-custom-metric"
    metricSpec := v2.MetricSpec{
        External: &v2.ExternalMetricSource{
            Metric: v2.MetricIdentifier{
                Name: metricName,
            },
            Target: createMetricTarget(s.threshold),
        },
        Type: v2.ExternalMetricSourceType,
    }
    return []v2.MetricSpec{metricSpec}
}

func (s *MyScaler) GetMetrics(ctx context.Context, metricIdentifier v2.MetricIdentifier, config *ScalerConfig) ([]v1beta1.ExternalMetricValue, error) {
    // 调用自定义客户端获取指标值
    value, err := s.client.GetCurrentValue()
    if err != nil {
        return nil, err
    }
    return []v1beta1.ExternalMetricValue{
        {
            MetricName: metricIdentifier.Name,
            Value:      resource.NewQuantity(int64(value), resource.DecimalSI),
            Timestamp:  metav1.Now(),
        },
    }, nil
}

func (s *MyScaler) Run(ctx context.Context, activeThreshold int) (bool, error) {
    // 判断是否有事件待处理
    return s.client.HasPendingEvents()
}

func (s *MyScaler) Close(ctx context.Context) error {
    return s.client.Close()
}
注意事项： 内置 Scaler 开发需要深入理解 KEDA 的代码结构和测试框架。每次 KEDA 版本升级，内置 Scaler 可能需要同步适配。对于非核心场景，外部 Scaler 是更灵活的选择。

4.4 本章小结

← 上一章 📖 返回目录 下一章 →