KEDA 事件驱动自动缩放面试题库

graph LR
    A[外部事件源<br/>Kafka/RabbitMQ/Prometheus] --> B[Scalers<br/>指标采集器]
    B --> C[KEDA Metrics Server<br/>指标适配器]
    C --> D[HPA<br/>水平自动缩放器]
    D --> E[KEDA Operator<br/>控制器]
    E --> F[Deployment/StatefulSet<br/>工作负载]
    F -.->|scale-to-zero| E

graph LR
    A["1. 外部事件到达<br/>队列消息增加/流延迟上升/请求速率变化"] --> B["2. Scaler 从事件源采集指标值"]
    B --> C["3. KEDA Metrics Server 暴露为 External Metric"]
    C --> D["4. HPA 周期查询指标值，计算期望副本数"]
    D --> E["5. HPA 调整 Deployment/StatefulSet 副本数"]
    E --> F["6. 指标归零时，KEDA 可将工作负载缩至 0"]
    F --> G["7. 事件再次到达，KEDA 触发从 0→1 的扩容"]

graph LR
    A["用户创建 ScaledObject"] --> B["KEDA Operator 自动创建 HPA 资源"]
    B -.-> C["HPA 的 metric spec 指向 KEDA Metrics Server"]
    B --> D["HPA 通过 external.metrics.k8s.io API 查询指标"]
    D -.-> E["KEDA Metrics Server 根据 Scaler 返回实时值"]
    D --> F["HPA 计算期望副本数 → 调整 targetRef 的 replicas"]

graph LR
    A["事件消逝 → 指标归零"] --> B["HPA 将副本缩到 minReplicaCount<br/>通常为 1"]
    B --> C["KEDA Operator 检测到指标持续为零"]
    C --> D["KEDA 将 Deployment replicas 设置为 0<br/>绕过 HPA 限制"]
    D --> E["删除所有 Pod"]
    D --> F["内部标记该 ScaledObject 处于停用状态"]
    G["事件到达 → Scaler 检测到指标 > activationThreshold"] --> H["KEDA Operator 重新激活 ScaledObject"]
    H --> I["恢复 Deployment replicas 到 1"]
    H --> J["HPA 恢复接管后续缩放"]

apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
  name: kafka-consumer
spec:
  scaleTargetRef:
    name: my-deployment           # 目标工作负载
    kind: Deployment               # 支持 Deployment/StatefulSet/Custom Resource
    envSourceContainerName: app   # 环境变量来源容器
  pollingInterval: 30              # 轮询间隔（秒，默认 30）
  cooldownPeriod: 300              # 冷却期（秒，默认 300）
  minReplicaCount: 0               # 最小副本数（默认 0）
  maxReplicaCount: 30              # 最大副本数（默认 100）
  idleReplicaCount: 0              # 空闲副本数（闲置时保留）
  triggers:
  - type: kafka
    metadata:
      topic: my-topic
      bootstrapServers: kafka:9092
      consumerGroup: my-group
      lagThreshold: "10"
  advanced:                        # 高级配置
    horizontalPodAutoscalerConfig:
      behavior:
        scaleDown:
          stabilizationWindowSeconds: 60

apiVersion: keda.sh/v1alpha1
kind: ScaledJob
metadata:
  name: sqs-consumer
spec:
  jobTargetRef:                     # Job 模板定义
    template:
      spec:
        containers:
        - name: worker
          image: my-worker:latest
          command: ["./process"]
        restartPolicy: Never
    backoffLimit: 4
  pollingInterval: 10
  maxReplicaCount: 30
  successfulJobsHistoryLimit: 3
  failedJobsHistoryLimit: 2
  scalingStrategy:
    strategy: "custom"             # default / custom / accurate / eager
    customScalingQueueLengthDeduction: 1
    customScalingRunningJobPercentage: "0.5"
  triggers:
  - type: aws-sqs-queue
    metadata:
      queueURL: https://sqs.us-east-1.amazonaws.com/...
      queueLength: "5"

apiVersion: keda.sh/v1alpha1
kind: TriggerAuthentication
metadata:
  name: kafka-auth
spec:
  secretTargetRef:
  - parameter: sasl                    # Scaler 参数名
    name: kafka-secrets                # Secret 名称
    key: sasl                          # Secret 中的键
  - parameter: tls
    name: kafka-secrets
    key: tls
apiVersion: keda.sh/v1alpha1
kind: ClusterTriggerAuthentication
metadata:
  name: global-kafka-auth
spec:
  secretTargetRef:
  - parameter: sasl
    name: keda/kafka-secrets           # cluster-scoped 需包含命名空间
    key: sasl

graph LR
    A["ScaledObject 创建"] --> B["KEDA Operator 根据 triggers 配置注册 External Metric"]
    B -.-> C["metric name = keda-{namespace}-{scaledobject-name}-{trigger-index}"]
    B --> D["KEDA Metrics Server 实现 APIService"]
    D -.-> E["v1beta1.external.metrics.k8s.io"]
    D --> F["HPA 自动发现并查询 External Metric"]
    F -.-> G["spec: metrics: - type: External<br/>external: metric: name: keda-...<br/>target: averageValue: '10'"]
    F --> H["Metrics Server 调用对应 Scaler 的 GetMetrics 方法"]
    H -.-> I["返回当前队列深度/流延迟/请求数"]
    H --> J["HPA 计算 desired = ceil(current_value / target_value)"]
    J -.-> K["调整 Deployment replicas"]

# ScaledObject 生成的指标命名
# default 命名空间下名为 kafka-consumer 的 ScaledObject
# 第一个 trigger（index=0）
# → 外部指标名: keda-default-kafka-consumer-0

# 多个 trigger 时，每个 trigger 生成独立指标
# scaledObject 有 3 个 triggers → 3 个 external metrics
# advanced.scalingModifiers.formula 合并多指标

# KEDA Operator 自动创建（或更新）的 HPA
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: keda-hpa-kafka-consumer
  labels:
    scaledobject.keda.sh/name: kafka-consumer
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: my-deployment
  minReplicas: 0         # KEDA 允许 0
  maxReplicas: 30
  metrics:
  - type: External
    external:
      metric:
        name: keda-default-kafka-consumer-0
      target:
        type: AverageValue
        averageValue: "10"

graph LR
    subgraph 扩容阶段
        A[事件到达] --> B[激活阈值检查]
        B -->|超出 activationThreshold| C[从 idle 扩容到 1]
        C --> D[HPA 按指标持续扩容]
    end
    subgraph 缩容阶段
        E[指标下降] --> F[pollingInterval 检测到归零]
        F --> G[cooldownPeriod 等待期<br/>防止抖动]
        G --> H[缩容到 idleReplicaCount]
    end
    subgraph 完全空闲
        I[idleReplicaCount=0] --> J[完全缩零<br/>无 Pod 运行]
        K[idleReplicaCount=1] --> L[保留 1 个 Pod<br/>就绪但未处理]
    end
    D --> E
    H --> I
    H --> K

# 场景 1：毫秒级延迟敏感，但不希望空跑
apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
  name: latency-sensitive
spec:
  scaleTargetRef:
    name: my-app
  pollingInterval: 5        # 5s 采集一次，快速响应
  cooldownPeriod: 60        # 60s 后就缩
  idleReplicaCount: 1       # 保留 1 个热备
# 场景 2：批处理任务，完全不希望空跑
apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
  name: batch-worker
spec:
  scaleTargetRef:
    name: worker
  minReplicaCount: 0
  maxReplicaCount: 50
  pollingInterval: 30
  cooldownPeriod: 300
  idleReplicaCount: 0       # 完全缩零

apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
  name: multi-trigger
spec:
  scaleTargetRef:
    name: my-app
  triggers:
  - type: prometheus      # 指标: trig_one
    name: trig_one
    metadata:
      query: sum(rate(http_requests_total[1m]))
      threshold: "100"
  - type: rabbitmq        # 指标: trig_two
    name: trig_two
    metadata:
      queueName: my-queue
      queueLength: "10"
  advanced:
    scalingModifiers:
      formula: "(trig_one + trig_two) / 2"   # 复合公式
      target: "50"                            # 目标值
      activationTarget: "10"                  # 激活阈值
      metricType: "AverageValue"              # AverageValue / Value

graph LR
    A["触发模式：多 ScaledObject，各有独立 HPA"]
    A -.-> B["问题：同一工作负载被多个 HPA 管理<br/>行为不可预测"]
    C["聚合模式（推荐）：单 ScaledObject + scalingModifiers"]
    C -.-> D["单个 HPA，公式决定最终指标值"]
    E["独立模式：单 ScaledObject 多 trigger，无 modifiers"]
    E -.-> F["HPA 取所有指标中的最大值"]

graph LR
    A["无已有 HPA"] --> B["KEDA 自动创建 HPA"]
    C["有已有 HPA"] --> D["KEDA 检测到冲突，拒绝创建 ScaledObject"]
    D -.-> E["报错：ScaledObject 与已有 HPA 管理相同 targetRef"]

apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
  name: take-over
  annotations:
    scaledobject.keda.sh/transfer-hpa-ownership: "true"   # 启用接管
spec:
  scaleTargetRef:
    name: my-deployment
  advanced:
    horizontalPodAutoscalerConfig:
      name: my-existing-hpa                                 # 指定已有 HPA
  triggers:
  - type: prometheus
    metadata:
      serverAddress: http://prometheus:9090
      query: sum(rate(http_requests_total[1m]))
      threshold: "100"

graph LR
    A["1. 用户在 ScaledObject 添加 transfer-hpa-ownership 注解"] --> B["2. KEDA 验证 HPA 的 scaleTargetRef 与 ScaledObject 一致"]
    B --> C["3. 验证通过 → KEDA 接管 HPA 管理权"]
    C --> D["添加 label: scaledobject.keda.sh/name"]
    C --> E["更新 HPA metrics 指向 KEDA Metrics Server"]
    C --> F["保持 HPA behavior / 其他已有配置"]
    C --> G["4. 用户不可再直接编辑该 HPA"]
    G --> H["5. 删除 ScaledObject 时 → HPA 可选择保留或删除"]

# Custom Resource 必须实现 /scale 子资源
# 在 CRD 定义中：
apiVersion: apiextensions.k8s.io/v1
kind: CustomResourceDefinition
spec:
  versions:
  - name: v1
    subresources:
      scale:                          # 启用缩放子资源
        specReplicasPath: .spec.replicas
        statusReplicasPath: .status.replicas
        labelSelectorPath: .status.selector
# ScaledObject 引用自定义资源
apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
  name: custom-resource-scaler
spec:
  scaleTargetRef:
    name: my-cr-instance
    kind: MyCustomResource            # CRD 的 kind
    apiVersion: example.io/v1         # CRD 的 API 版本
    envSourceContainerName: app

apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
  name: controlled-scaler
spec:
  scaleTargetRef:
    name: my-app
  advanced:
    horizontalPodAutoscalerConfig:
      behavior:
        scaleDown:                              # 缩容策略
          stabilizationWindowSeconds: 300        # 稳定窗口（5分钟）
          policies:
          - type: Percent
            value: 10                           # 每分钟最多缩 10%
            periodSeconds: 60
          selectPolicy: Min                     # Min / Max / Disabled
        scaleUp:                                # 扩容策略
          stabilizationWindowSeconds: 0          # 无稳定窗口，立即扩容
          policies:
          - type: Percent
            value: 100                          # 每分钟最多翻倍扩容
            periodSeconds: 15
          - type: Pods
            value: 4                            # 每分钟最多加 4 Pod
            periodSeconds: 15
          selectPolicy: Max                     # 取两个策略中较大的限制

# 场景：扩容快、缩容慢、禁止缩到零
spec:
  advanced:
    horizontalPodAutoscalerConfig:
      behavior:
        scaleUp:
          stabilizationWindowSeconds: 0        # 无延迟，立即响应流量
          policies:
          - type: Percent
            value: 100
            periodSeconds: 15
          selectPolicy: Max
        scaleDown:
          stabilizationWindowSeconds: 600       # 10分钟稳定期
          policies:
          - type: Pods
            value: 1
            periodSeconds: 60
          selectPolicy: Min
# 场景：紧急扩容（CPU 突增）
spec:
  advanced:
    horizontalPodAutoscalerConfig:
      behavior:
        scaleDown:
          selectPolicy: Disabled               # 完全禁止缩容

# 暂停缩容
metadata:
  annotations:
    autoscaling.keda.sh/paused-scale-in: "true"

# 暂停扩容
metadata:
  annotations:
    autoscaling.keda.sh/paused-scale-out: "true"

# 完全暂停
metadata:
  annotations:
    autoscaling.keda.sh/paused: "true"

graph LR
    A["Kafka 消息到达（事件触发）"] --> B["KEDA Scaler 检测到消息数 > 阈值"]
    B --> C["KEDA Operator 根据 scalingStrategy 计算需要创建多少 Job"]
    C --> D["创建 Kubernetes Job（每消息一个任务）"]
    D -.-> E["每个 Job 处理一条消息后终止"]
    D --> F["Job 完成 → 清理<br/>保留在 successfulJobsHistoryLimit 内"]
    D --> G["Job 失败 → 重试（backoffLimit）<br/>保留在 failedJobsHistoryLimit 内"]

spec:
  scalingStrategy:
    strategy: "custom"
    customScalingQueueLengthDeduction: 1      # 每个 Job 处理 1 个消息
    customScalingRunningJobPercentage: "0.5"   # 运行中 Job 占队列比例

scalingStrategy:
  strategy: "default"

scalingStrategy:
  strategy: "custom"
  customScalingQueueLengthDeduction: 1        # 每个 Job 可消费的消息数
  customScalingRunningJobPercentage: "0.5"    # 运行中 Job 折算比例

desiredJobs = ceil(
    (pendingEvents / customScalingQueueLengthDeduction) -
    (runningJobs * customScalingRunningJobPercentage)
)

scalingStrategy:
  strategy: "accurate"

scalingStrategy:
  strategy: "eager"

apiVersion: keda.sh/v1alpha1
kind: ScaledJob
metadata:
  name: sqs-worker
spec:
  jobTargetRef:
    template:
      spec:
        containers:
        - name: worker
          image: worker:latest
        restartPolicy: Never
    backoffLimit: 4
  successfulJobsHistoryLimit: 5     # 保留最近 5 个成功 Job
  failedJobsHistoryLimit: 3         # 保留最近 3 个失败 Job
  maxReplicaCount: 50

graph LR
    A["定时检查：KEDA Operator 定期检查 ScaledJob 创建的 Job"]
    A --> B["成功 Job 数 > successfulJobsHistoryLimit"]
    B -.-> C["删除最旧的成功 Job<br/>保留最新的 N 个"]
    A --> D["失败 Job 数 > failedJobsHistoryLimit"]
    D -.-> E["删除最旧的失败 Job"]
    A --> F["Job 未完成 → 不参与清理"]

# 默认行为：ScaledJob 的 labels 全部传播到创建的 Job
# 排除特定标签（避免传播敏感或高基数标签）
metadata:
  annotations:
    scaledjob.keda.sh/job-excluded-labels: "team,environment"
  labels:
    app: worker
    team: backend
    environment: staging

graph LR
    A["每个 Scaler 定义一组需要的参数<br/>如 sasl、tls、password"] --> B["TriggerAuthentication 通过 parameter 字段将认证源的值映射到 Scaler 参数"]
    B --> C["secretTargetRef.parameter = sasl<br/>从 Secret 中取某个 key 的值<br/>传给 Scaler 的 sasl 参数"]
    C --> D["ScaledObject 的 trigger 中通过 authenticationRef 引用 TriggerAuthentication"]

# 示例：Scaler 需要 connectionString 和 queueName
# TriggerAuthentication 提供 connectionString 值
apiVersion: keda.sh/v1alpha1
kind: TriggerAuthentication
metadata:
  name: azure-queue-auth
spec:
  secretTargetRef:
  - parameter: connectionString     # 映射到 Azure Queue Scaler 的 connectionString 参数
    name: azure-storage-secret      # Secret 名称
    key: connectionString           # Secret 中的 key
# ScaledObject 引用
apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
  name: azure-consumer
spec:
  triggers:
  - type: azure-queue
    metadata:
      queueName: myqueue            # 非敏感参数放在 metadata
      # connectionString 由 TriggerAuthentication 提供
    authenticationRef:
      name: azure-queue-auth        # 引用认证

# Azure AD Workload Identity
apiVersion: keda.sh/v1alpha1
kind: TriggerAuthentication
metadata:
  name: azure-pod-identity-auth
spec:
  podIdentity:
    provider: azure-workload
    identityId: <uuid>              # 可选，指定托管标识 ID
# AWS IRSA
apiVersion: keda.sh/v1alpha1
kind: TriggerAuthentication
metadata:
  name: aws-irsa-auth
spec:
  podIdentity:
    provider: aws
    roleArn: arn:aws:iam::123456789012:role/keda-role

# 集群级认证配置，可在任意命名空间引用
apiVersion: keda.sh/v1alpha1
kind: ClusterTriggerAuthentication
metadata:
  name: global-rabbitmq-auth
spec:
  secretTargetRef:
  - parameter: host
    name: keda-secrets/rabbitmq     # cluster-scoped 需包含命名空间前缀
    key: host
# 跨命名空间引用
spec:
  triggers:
  - type: rabbitmq
    authenticationRef:
      name: global-rabbitmq-auth
      kind: ClusterTriggerAuthentication   # 指定集群级

# 前提：KEDA Operator 的 ServiceAccount 已配置 Azure AD 集成
apiVersion: keda.sh/v1alpha1
kind: TriggerAuthentication
metadata:
  name: azure-servicebus-auth
spec:
  podIdentity:
    provider: azure-workload
    identityId: /subscriptions/.../userAssignedIdentities/my-identity  # 可选
# ScaledObject 使用
apiVersion: keda.sh/v1alpha1
kind: ScaledObject
spec:
  triggers:
  - type: azure-servicebus
    metadata:
      queueName: myqueue
      namespace: sb-namespace
    authenticationRef:
      name: azure-servicebus-auth

# 前提：KEDA Operator ServiceAccount 已添加 IAM Role 注解
# kubectl annotate sa keda-operator -n keda eks.amazonaws.com/role-arn=arn:aws:iam::...
apiVersion: keda.sh/v1alpha1
kind: TriggerAuthentication
metadata:
  name: aws-sqs-auth
spec:
  podIdentity:
    provider: aws
    roleArn: arn:aws:iam::123456789012:role/keda-sqs-role
# or
spec:
  podIdentity:
    provider: aws-eks            # 已废弃，v3 移除
# 可选的 identityOwner 控制
spec:
  podIdentity:
    provider: aws
    identityOwner: keda           # keda（默认，使用 Operator 身份）
    identityOwner: workload       # workload（使用目标 Pod 身份）

# 前提：KEDA 运行在 GKE 并配置 Workload Identity 绑定
apiVersion: keda.sh/v1alpha1
kind: TriggerAuthentication
metadata:
  name: gcp-pubsub-auth
spec:
  podIdentity:
    provider: gcp
# ScaledObject 引用
spec:
  triggers:
  - type: gcp-pubsub
    metadata:
      topic: my-topic
      subscription: my-sub
    authenticationRef:
      name: gcp-pubsub-auth

graph LR
    A["KEDA Operator 默认使用自身 Pod 的身份进行外部访问"]
    A --> B["azure-workload: 使用 Operator ServiceAccount 的托管标识"]
    A --> C["aws/aws-eks: 使用 Operator ServiceAccount 绑定的 IAM Role"]
    A --> D["gcp: 使用 Operator 节点的 GCP Service Account"]
    A --> E["identityOwner: workload<br/>改用目标部署的 Pod 身份（仅 AWS）"]

apiVersion: keda.sh/v1alpha1
kind: TriggerAuthentication
metadata:
  name: vault-auth
spec:
  hashiCorpVault:
    address: http://vault.example.com:8200
    authentication: kubernetes              # Vault Kubernetes Auth 方法
    role: my-role                           # Vault Role（绑定 SA）
    mount: k8s                              # Vault auth mount path
    credential:
      serviceAccount: /var/run/secrets/kubernetes.io/serviceaccount/token  # 默认路径
    secrets:
    - parameter: password                   # 映射到 Scaler 的 password 参数
      key: db_password                      # Vault 中数据的 key
      path: secret/data/db/mysql            # Vault secret path
    - parameter: host
      key: host
      path: secret/data/kafka
apiVersion: keda.sh/v1alpha1
kind: ScaledObject
spec:
  triggers:
  - type: mysql
    metadata:
      dbName: mydb
    authenticationRef:
      name: vault-auth

apiVersion: keda.sh/v1alpha1
kind: TriggerAuthentication
metadata:
  name: vault-token-auth
spec:
  hashiCorpVault:
    address: http://vault.example.com:8200
    authentication: token                  # 使用 Vault Token
    credential:
      token: s.abcdef123456789             # 可直接指定 Token
    secrets:
    - parameter: connectionString
      key: connectionString
      path: secret/data/rabbitmq
# 从 Secret 读取 Token
spec:
  hashiCorpVault:
    authentication: token
    credential:
      tokenFromSecret:
        name: vault-token
        key: token

apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
  name: consumer
  namespace: ns-a
spec:
  triggers:
  - type: kafka
    metadata:
      topic: my-topic
      bootstrapServers: kafka:9092
      consumerGroup: my-group
    authenticationRef:                        # 每个 trigger 独立认证
      name: kafka-auth                        # TriggerAuthentication 名称（同命名空间）
      # kind 默认为 TriggerAuthentication（命名空间级）
  - type: rabbitmq
    metadata:
      queueName: my-queue
    authenticationRef:
      name: global-rabbit-auth
      kind: ClusterTriggerAuthentication     # 跨命名空间引用

# 一个 ScaledObject 的不同 trigger 可以使用不同的认证源
spec:
  triggers:
  - type: kafka
    authenticationRef:
      name: kafka-auth-vault               # 来自 Vault
  - type: prometheus
    authenticationRef:
      name: prom-auth-env                  # 来自环境变量
  - type: rabbitmq
    authenticationRef:
      name: global-rabbit-auth
      kind: ClusterTriggerAuthentication   # 集群级共享认证

TriggerAuthentication 的 parameter 必须与 Scaler 期望的参数名一致

常见 Scaler 参数名：
  - host / hosts / server / serverAddress
  - username / password
  - connectionString
  - apiKey
  - sasl / tls / ca / cert / key
  - awsRegion / awsAccessKeyId / awsSecretAccessKey
  - tenantId / clientId / clientSecret

TriggerAuthentication 的 parameter 值会覆盖 ScaledObject metadata 中的同名参数

# 通过 gRPC 协议实现自定义 Scaler
apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
  name: custom
spec:
  triggers:
  - type: external
    metadata:
      scalerAddress: custom-scaler:9000      # gRPC Scaler 地址
      # 自定义元数据
      myParam: value
    authenticationRef:
      name: custom-auth
# External Scaler 需实现以下 gRPC 接口：
# rpc IsActive(ctx, ScaledObjectRef) → IsActiveResponse
# rpc GetMetricSpec(ctx, ScaledObjectRef) → MetricSpec
# rpc GetMetrics(ctx, GetMetricRequest) → GetMetricResponse
# rpc StreamIsActive(ctx, ScaledObjectRef) → Stream → IsActiveResponse（可选，push 模式）

apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
  name: http-api
spec:
  scaleTargetRef:
    name: api-server
  triggers:
  - type: prometheus
    metadata:
      serverAddress: http://prometheus:9090
      metricName: http_requests_rate        # 指标名称（自定义）
      query: |
        sum(rate(http_requests_total{job="api"}[2m]))        
      threshold: "100"                      # 每个副本目标处理 100 req/s
      activationThreshold: "10"             # 低于 10 时不扩容
      namespace: default                    # 指标所属命名空间（用于隔离）
# 高级查询：按路径分流
  - type: prometheus
    metadata:
      serverAddress: http://prometheus:9090
      metricName: api_error_rate
      query: |
        sum(rate(http_requests_total{status=~"5.."}[1m]))
        /
        sum(rate(http_requests_total[1m]))        
      threshold: "0.01"                    # 错误率 > 1% 时扩容
# 使用 cortex/thanos 等多租户 Prometheus
  - type: prometheus
    metadata:
      serverAddress: http://thanos-query:9090
      metricName: queue_depth
      query: max(kafka_consumer_lag)
      threshold: "1000"
      namespace: monitoring
      headers: "X-Scope-OrgID: my-tenant"  # 自定义 HTTP Header

graph LR
    A["HPA 查询 KEDA → KEDA 执行 PromQL → 获取返回值的第一个数据点"] --> B["desiredReplicas = ceil(query_result / threshold)"]
    B --> C["query_result = 500, threshold = 100<br/>desired = ceil(500/100) = 5 replicas"]
    B --> D["query_result = 0（且未设 activationThreshold）<br/>desired = 0 → scale to zero"]
    B --> E["query_result < activationThreshold<br/>从 0 激活时不扩容"]

apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
  name: cron-scaler
spec:
  scaleTargetRef:
    name: my-app
  minReplicaCount: 1                        # 基础副本数
  maxReplicaCount: 50                       # 最大副本数
  triggers:
  - type: cron
    metadata:
      timezone: Asia/Shanghai
      start: 30 8 * * *                     # 每天 8:30 开始扩容
      end: 30 18 * * *                      # 每天 18:30 开始缩容
      desiredReplicas: "20"                 # 高峰期间保持 20 副本
# 多时间段配置
  - type: cron
    metadata:
      timezone: Asia/Shanghai
      start: 0 9 * * 1-5                    # 工作日 9:00
      end: 0 18 * * 1-5                     # 工作日 18:00
      desiredReplicas: "10"
  - type: cron
    metadata:
      timezone: Asia/Shanghai
      start: 0 10 * * 6,7                   # 周末 10:00
      end: 0 16 * * 6,7                     # 周末 16:00
      desiredReplicas: "5"

graph LR
    A["当前时间"]
    A --> B["在 start 和 end 之间<br/>Cron Scaler 返回 desiredReplicas"]
    B -.-> C["HPA 将副本数调整为 desiredReplicas<br/>受 maxReplicaCount 限制"]
    A --> D["不在时间段内<br/>Cron Scaler 返回 0"]
    D -.-> E["遵循 minReplicaCount<br/>如果 minReplicaCount > 0，保留基础副本"]

graph LR
    A["多 Cron 时间段重叠<br/>KEDA 取所有激活 Cron 中的最大值"]
    A --> B["Cron1: 9-18点, desiredReplicas=10"]
    A --> C["Cron2: 12-14点, desiredReplicas=20"]
    C -.-> D["12-14点期间: desiredReplicas = max(10, 20) = 20"]

apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
  name: kafka-consumer
spec:
  scaleTargetRef:
    name: consumer-app
  pollingInterval: 10                     # 高频采集 Lag
  cooldownPeriod: 60
  triggers:
  - type: kafka
    metadata:
      bootstrapServers: kafka-cluster:9092
      consumerGroup: my-group
      topic: my-topic
      lagThreshold: "100"                 # 每个副本允许的 Lag 上限
      activationLagThreshold: "10"        # 激活阈值（低于此不扩容）
      offsetResetPolicy: latest           # earliest / latest
      allowIdleConsumers: "true"          # 允许空闲消费者
      version: "2.8.0"                    # Kafka 版本
      excludePersistentLag: "false"       # 排除持久 Lag
      partitionLimitation: "2"            # 限制分区数
      # SASL 认证（建议使用 TriggerAuthentication）
      # sasl: ...
      # tls: ...
    authenticationRef:
      name: kafka-auth
# SASL + TLS 认证
apiVersion: keda.sh/v1alpha1
kind: TriggerAuthentication
metadata:
  name: kafka-auth
spec:
  secretTargetRef:
  - parameter: sasl
    name: kafka-secret
    key: sasl
  - parameter: tls
    name: kafka-secret
    key: tls

graph LR
    A["总 Lag = sum（所有分区的 consumer lag）"]
    A --> B["totalLag < activationLagThreshold（10）<br/>Scaler 返回 0 → 缩容<br/>到 idleReplicaCount 或 0"]
    A --> C["totalLag 在 10 ~ 100 之间<br/>已激活但不需要额外副本<br/>保持当前副本数"]
    A --> D["totalLag > lagThreshold（100）<br/>desiredReplicas = ceil(totalLag / lagThreshold)<br/>扩容到 2、3、4... 个副本"]

# 消费多个 topic（逗号分隔）
  - type: kafka
    metadata:
      topic: topic-a,topic-b,topic-c
      consumerGroup: multi-topic-group
# 正则匹配 topic
  - type: kafka
    metadata:
      topic: orders-.*
      consumerGroup: orders-group

service ExternalScaler {
  // 返回 Scaler 是否处于激活状态（决定是否从 0→1）
  rpc IsActive(ScaledObjectRef) returns (IsActiveResponse);

  // 返回 HPA 所需的指标规格（指标名、目标类型、目标值）
  rpc GetMetricSpec(ScaledObjectRef) returns (GetMetricSpecResponse);

  // 返回当前指标值（HPA 查询时调用）
  rpc GetMetrics(GetMetricsRequest) returns (GetMetricsResponse);

  // 可选：流式激活通知（替代轮询 IsActive）
  rpc StreamIsActive(ScaledObjectRef) returns (stream IsActiveResponse);
}

apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
  name: custom-scaler
spec:
  triggers:
  - type: external            # 内置 External Scaler
    metadata:
      scalerAddress: custom-scaler-svc:9000   # gRPC 服务地址
      # 自定义元数据，随请求发送到 gRPC Scaler
      endpoint: /api/queue-length
      apiKey: xxx
  - type: external-push       # Push 模式（无需轮询）
    metadata:
      scalerAddress: push-scaler:9000

# Push 模式：外部系统主动通知 KEDA，避免轮询开销
apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
  name: push-consumer
spec:
  triggers:
  - type: external-push
    metadata:
      scalerAddress: push-scaler:9000
# 外部系统通过 gRPC 调用 StreamIsActive 推送事件
# 每次事件到达 → KEDA 重新评估是否需要扩容

# ValidatingWebhook 校验的内容（示例）：
# 1. 防止多个 ScaledObject 指向同一个 targetRef
#    → 同一 Deployment 不能被两个 ScaledObject 管理
#    → 报错: "ScaledObject already exists for this target"
#
# 2. 校验 scaleTargetRef 引用的资源存在
#    → Deployment/StatefulSet 必须存在
#    → Custom Resource 必须实现 /scale 子资源
#
# 3. 校验 TriggerAuthentication 的 provider 配置
#    → podIdentity.provider 必须是合法值
#    → Vault 地址格式校验
#
# 4. 校验 Scaler 参数完整性
#    → 必填参数不能为空

# MutatingWebhook 自动设置的默认值：
# spec.pollingInterval = 30（未指定时）
# spec.cooldownPeriod = 300
# spec.minReplicaCount = 0
# spec.maxReplicaCount = 100
# spec.scalingStrategy.strategy = "default"

# KEDA 安装时默认启用 Webhooks
# 可配置失败策略：
apiVersion: admissionregistration.k8s.io/v1
kind: ValidatingWebhookConfiguration
metadata:
  name: keda-admission
spec:
  failurePolicy: Fail            # Fail / Ignore
  # Fail: Webhook 不可用时拒绝创建资源
  # Ignore: Webhook 不可用时允许创建（降级）

# KEDA Operator 默认在 8080 端口暴露指标
# ServiceMonitor 示例
apiVersion: monitoring.coreos.com/v1
kind: ServiceMonitor
metadata:
  name: keda-operator
  namespace: keda
spec:
  selector:
    matchLabels:
      app: keda-operator
  endpoints:
  - port: metrics
    interval: 15s
    path: /metrics
# Prometheus 直接采集
# keda-operator 的 Service 已包含 prometheus.io/scrape: "true" 注解

# KEDA 命名空间排除 Sidecar 自动注入
apiVersion: v1
kind: Namespace
metadata:
  name: keda
  annotations:
    sidecar.istio.io/inject: "false"          # 方式 1：命名空间注解
# 或为 KEDA Operator Pod 添加注解
apiVersion: apps/v1
kind: Deployment
metadata:
  name: keda-operator
  namespace: keda
spec:
  template:
    metadata:
      annotations:
        sidecar.istio.io/inject: "false"      # 方式 2：Pod 注解

# 允许 KEDA Metrics Server 跨命名空间访问指标
apiVersion: security.istio.io/v1beta1
kind: AuthorizationPolicy
metadata:
  name: keda-allow
  namespace: your-app
spec:
  selector:
    matchLabels:
      app: your-app
  action: ALLOW
  rules:
  - from:
    - source:
        namespaces: ["keda"]
    to:
    - operation:
        ports: ["8080"]

# 当工作负载启用 Strict mTLS 时
# KEDA 的 Prometheus Scaler 或其他 HTTP Scaler 可能需要跳过 TLS 验证
triggers:
- type: prometheus
  metadata:
    unsafeSsl: "true"          # 跳过 TLS 校验（仅内部网络可行）
# 或配置证书
triggers:
- type: prometheus
  metadata:
    ca: /path/to/ca.crt

graph LR
    A["KEDA Operator"] -->|OTLP| B["OpenTelemetry Collector"]
    B --> C["Jaeger/Zipkin（追踪）"]
    B --> D["Prometheus（指标）"]
    B --> E["Loki/Elasticsearch（日志）"]

# KEDA Operator 环境变量启用 OTel
apiVersion: apps/v1
kind: Deployment
metadata:
  name: keda-operator
  namespace: keda
spec:
  template:
    spec:
      containers:
      - name: keda-operator
        env:
        - name: KEDA_OTEL_ENABLED
          value: "true"
        - name: OTEL_EXPORTER_OTLP_ENDPOINT
          value: "http://otel-collector:4318"
        - name: OTEL_SERVICE_NAME
          value: "keda-operator"

# Helm values.yaml
# 部署配置
operator:
  name: keda-operator
  replicaCount: 2                          # 多副本高可用
  logLevel: info
  resources:
    requests:
      cpu: 100m
      memory: 128Mi
    limits:
      cpu: 500m
      memory: 256Mi
  watchNamespace: ""                        # 空 = 所有命名空间
  extraEnv:
  - name: KEDA_OTEL_ENABLED
    value: "false"                          # 按需启用 OTel

metricsServer:
  replicaCount: 2                           # Metrics Server 高可用
  resources:
    requests:
      cpu: 100m
      memory: 128Mi
    limits:
      cpu: 500m
      memory: 512Mi
  extraEnv:
  - name: METRICS_SERVER_ENABLE_PROFILING
    value: "false"

webhooks:
  replicaCount: 2                           # Webhook 高可用
  failurePolicy: Fail                       # Fail: 严格校验

# 全局配置
global:
  watchNamespace: ""                        # 监控所有命名空间
  kubeletPort: 10250
# 安装命令
helm install keda kedacore/keda \
  --namespace keda --create-namespace \
  --values values.yaml \
  --version 2.19.0

# 1. 检查 ScaledObject 状态
kubectl get scaledobject kafka-consumer -o jsonpath='{.status}' | jq .
# 关键字段:
# .status.conditions[] - 就绪状态
# .status.health[] - Scaler 健康状态
# .status.hpaName - 关联的 HPA 名称
# .status.lastActiveTime - 最后激活时间
# .status.externalMetricNames - 生成的外部指标名

# 2. 检查 HPA 事件
kubectl describe hpa keda-hpa-kafka-consumer
# 查看 Events 中的 Scaling 相关事件
# 查看 Metrics 字段是否正常显示当前/目标值

# 3. 检查外部指标是否存在
kubectl get --raw /apis/external.metrics.k8s.io/v1beta1 | jq '.resources[].name'
# 应包含 keda-<namespace>-<scaledobject-name>-<trigger-index>

# 4. 直接查询指标值
kubectl get --raw "/apis/external.metrics.k8s.io/v1beta1/namespaces/default/keda-default-kafka-consumer-0" | jq '.items[0].value'
# 应返回当前指标数值

# 5. 查看 Operator 日志
kubectl logs -n keda deploy/keda-operator --tail=100 | grep kafka-consumer
# 查找 ScaledObject 的调和日志

graph LR
    A[工作负载未缩放] --> B{检查 ScaledObject 状态}
    B -->|conditions.Ready=False| C[查看 status.health<br/>定位 Scaler 错误]
    B -->|conditions.Ready=True| D{检查 HPA 状态}
    D -->|metrics 不正常| E[检查 External Metrics API<br/>指标名和数值]
    D -->|metrics 正常| F{检查 Operator 日志}
    F -->|有错误日志| G[根据错误信息修复<br/>Scaler 连接/认证/权限]
    F -->|无错误| H[检查 Pod 资源限制<br/>HPA behavior 配置]
    C --> I[检查 TriggerAuthentication<br/>网络连通性<br/>Scaler 参数]
    E --> I

# 推荐：使用 Pod Identity，避免静态凭据
spec:
  podIdentity:
    provider: azure-workload     # 或 aws / gcp

# 备选：从 Secret 读取，不写明文
spec:
  secretTargetRef:
  - parameter: connectionString
    name: my-secret
    key: connStr

# 禁止：metadata 中明文传递
spec:
  triggers:
  - type: rabbitmq
    metadata:
      host: amqp://user:password@rabbitmq:5672   # ❌ 明文

# KEDA Operator 默认 ClusterRole 包含跨命名空间资源操作
# 多租户场景下需限制 watchNamespace
# 部署时使用 --watch-namespace 限定范围
helm install keda kedacore/keda \
  --set global.watchNamespace=app-ns-1

# 限制 KEDA 访问特定事件源
apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
  name: keda-egress
  namespace: keda
spec:
  podSelector:
    matchLabels:
      app: keda-operator
  policyTypes:
  - Egress
  egress:
  - to:
    - namespaceSelector: {}       # 允许访问事件源
    - podSelector:
        matchLabels:
          app: kafka-broker
    ports:
    - port: 9092
      protocol: TCP
  - to:                           # 允许 DNS
    ports:
    - port: 53
      protocol: UDP

# 1. 降低轮询频率
spec:
  pollingInterval: 60                  # 默认 30，低频场景可调大
  cooldownPeriod: 300
# 2. 启用指标缓存
spec:
  triggers:
  - type: kafka
    useCachedMetrics: true             # 减少 Scaler 调用频率
# 3. 限定 Operator 监控范围
# Helm values:
global:
  watchNamespace: app-team-a           # 只监控特定命名空间
# 4. 合并 Scaler（减少 HPA 数量）
spec:
  triggers:
  - type: prometheus
    name: cpu
  - type: prometheus
    name: queue
  advanced:
    scalingModifiers:
      formula: "max(cpu, queue)"       # 多个指标合并到同一个 HPA

# 多副本 + PDB
apiVersion: apps/v1
kind: Deployment
metadata:
  name: keda-operator
  namespace: keda
spec:
  replicas: 2
  strategy:
    rollingUpdate:
      maxSurge: 1
      maxUnavailable: 0
apiVersion: policy/v1
kind: PodDisruptionBudget
metadata:
  name: keda-operator
spec:
  minAvailable: 1
  selector:
    matchLabels:
      app: keda-operator