Apache Druid Kubernetes Overlord 扩展：以 Kubernetes Job 替代 Middle Manager 的无 MM 任务调度实践

• 数据库
• OLAP
• 大数据
• 后端

导读

本文面向在 Kubernetes 上部署 Apache Druid 的运维与开发人员，系统讲解druid-kubernetes-overlord-extensions（Kubernetes Task Scheduling）扩展：它允许 Overlord 将 Druid 任务直接调度为 Kubernetes Job 运行，从而彻底移除 Middle Manager / Indexer 组件，实现"无 MM（MM-less）"的实时摄入架构。阅读本文你将掌握该扩展的工作原理、Overlord 端完整配置、Pod 适配器（Pod Adapter）与 Pod 模板选择策略、动态执行配置 API、RBAC 权限要求，以及从传统 Middle Manager 集群零停机迁移到 K8s 任务调度的k8sAndWorker混合模式。

该扩展的官方完整文档位于 docs/development/extensions-contrib/k8s-jobs.md，扩展源码位于 extensions-contrib/kubernetes-overlord-extensions。官方将其标注为 实验性（EXPERIMENTAL） 功能，主要原因是尚未在大量长期运行的 Druid 集群上进行过广泛验证，生产使用前请充分评估。

工作原理：任务如何变成 Kubernetes Job

在传统架构中，Overlord 将任务派发给 Middle Manager 或 Indexer 上的 worker 进程执行。而本扩展改变了这一模式：Overlord 根据指定的 Pod 适配器（Pod Adapter），为每个任务构建一个 Pod Spec，并提交为 Kubernetes Job。每个任务对应一个 K8s Job，Job 中的主容器以 "internal peon"（内部 peon）方式运行任务。

该设计带来的关键特性是任务的天然可恢复性：

• Job 与 Druid 部署本身完全解耦，重启 Pod 或进行升级不会影响正在运行的任务；
• 任务会继续运行，当 Overlord 重新启动后，会重新发现并追踪这些 Job。

从源码看，KubernetesTaskRunner（KubernetesTaskRunner.java）是核心执行器，其工作方式如下：

1. 每任务一线程：Runner 内部使用容量可配置的线程池（Execs.multiThreaded(config.getCapacity(), "k8s-task-runner-%d")），每个线程负责追踪一个 Job——先调用KubernetesPeonLifecycle把 Job 提交到 K8s，再等待其回报成功/失败状态；
2. 队列排队：当线程池没有空闲线程时，任务进入队列并保持WAITING状态，直到有任务完成腾出线程；
3. 启动时恢复：start()方法会调用client.getPeonJobs()列出正在运行的 Job，并通过joinAsync()重新挂接每个 Job 的状态，这正是"Overlord 重启后继续追踪任务"的实现细节（源码 KubernetesTaskRunner.java）；
4. 定期清理：Runner 内置一个定时清理线程，按照taskCleanupInterval周期检查并删除早于taskCleanupDelay的已完成 Job。

KubernetesTaskRunnerFactory（KubernetesTaskRunnerFactory.java）通过 Guice 在 KubernetesOverlordModule.java 中注册：druid.indexer.runner.type=k8s绑定到KubernetesTaskRunnerFactory，druid.indexer.runner.type=k8sAndWorker绑定到KubernetesAndWorkerTaskRunnerFactory。

配置：让 Overlord 使用 K8s 任务调度

加载扩展

首先，在 Overlord 进程的扩展加载列表中包含druid-kubernetes-overlord-extensions，具体方法参见 加载扩展。

核心运行时属性

扩展依赖以下关键配置（参考 KubernetesTaskRunnerConfig.java 中的字段定义）：

几点重要说明：

• capacity 的初始值建议：设置为切换 K8s 之前所有 Middle Manager 任务槽位（task slots）的总和。K8s task runner 为每个创建的 Job 使用一个线程，该值设置过大会在 Overlord 上造成内存问题（线程数量过多）；
• capacity同时作为线程池大小、总槽位数（getTotalTaskSlotCount）与已用槽位计算的上限，源码见 KubernetesTaskRunner.java；
• 若完全运行在无 Middle Manager 的集群中，还需要设置druid.processing.intermediaryData.storage.type=deepstore，让任务中间数据落 deep storage。

最小配置示例

druid.indexer.runner.type=k8s druid.indexer.runner.namespace=default druid.indexer.runner.capacity=10 druid.indexer.task.encapsulatedTask=true druid.processing.intermediaryData.storage.type=deepstore

动态执行配置 API（无需重启 Overlord）

Druid 运维人员可以动态调整扩展的某些特性，无需重启 Overlord 服务即可生效。动态配置的核心是 Pod 模板选择（Pod Template Selection），启用前需要先配置 [Custom Template Pod Adapter](#custom-template-pod-adapter 自定义模板 Pod 适配器)。

动态配置对象KubernetesTaskRunnerDynamicConfig在源码中对应配置键k8s.taskrunner.config，默认实现为DefaultKubernetesTaskRunnerDynamicConfig，默认选择策略为TaskTypePodTemplateSelectStrategy（见 KubernetesTaskRunnerDynamicConfig.java）。

使用这些 API 前，请确保拥有资源类型为CONFIG、资源名为"CONFIG"的读写权限，权限说明参见 用户认证与授权。

获取动态配置

GET /druid/indexer/v1/k8s/taskrunner/executionconfig

curl "http://ROUTER_IP:ROUTER_PORT/druid/indexer/v1/k8s/taskrunner/executionconfig"

GET /druid/indexer/v1/k8s/taskrunner/executionconfig HTTP/1.1 Host: http://ROUTER_IP:ROUTER_PORT

成功时返回当前动态执行配置的 JSON 对象：

{ "type": "default", "podTemplateSelectStrategy": { "type": "selectorBased", "selectors": [ { "selectionKey": "podSpec1", "context.tags": { "userProvidedTag": ["tag1", "tag2"] }, "dataSource": ["wikipedia"] }, { "selectionKey": "podSpec2", "type": ["index_kafka"] } ] } }

更新动态配置

POST /druid/indexer/v1/k8s/taskrunner/executionconfig

该端点支持以下可选请求头，用于在配置历史中记录变更者与变更说明：

• X-Druid-Author：String，配置变更的作者；
• X-Druid-Comment：String，本次更新的描述。

curl "http://ROUTER_IP:ROUTER_PORT/druid/indexer/v1/k8s/taskrunner/executionconfig" \ --header 'Content-Type: application/json' \ --data '{ "type": "default", "podTemplateSelectStrategy": { "type": "selectorBased", "selectors": [ { "selectionKey": "podSpec1", "context.tags": { "userProvidedTag": ["tag1", "tag2"] }, "dataSource": ["wikipedia"] }, { "selectionKey": "podSpec2", "type": ["index_kafka"] } ] } }'

POST /druid/indexer/v1/k8s/taskrunner/executionconfig HTTP/1.1 Host: http://ROUTER_IP:ROUTER_PORT Content-Type: application/json { "type": "default", "podTemplateSelectStrategy": { "type": "selectorBased", "selectors": [ { "selectionKey": "podSpec1", "context.tags": { "userProvidedTag": ["tag1", "tag2"] }, "dataSource": ["wikipedia"] }, { "selectionKey": "podSpec2", "type": ["index_kafka"] } ] } }

成功请求返回 HTTP200 OK与空响应体。

获取动态配置历史

GET /druid/indexer/v1/k8s/taskrunner/executionconfig/history

支持以下可选查询参数过滤结果：

• interval：String，以 ISO 8601 格式限定的时间区间，以/分隔，例如2023-07-13/2023-07-19。默认区间为一周，可通过在 Coordinator 的runtime.properties中设置druid.audit.manager.auditHistoryMillis调整；
• count：Integer，只返回最近n条记录。

curl "http://ROUTER_IP:ROUTER_PORT/druid/indexer/v1/k8s/taskrunner/executionconfig/history"

GET /druid/indexer/v1/k8s/taskrunner/executionconfig/history HTTP/1.1 Host: http://ROUTER_IP:ROUTER_PORT

若没有历史记录则返回空数组，示例响应：

[ { "key": "k8s.taskrunner.config", "type": "k8s.taskrunner.config", "auditInfo": { "author": "", "comment": "", "ip": "127.0.0.1" }, "payload": "{\"type\": \"default\",\"podTemplateSelectStrategy\":{\"type\": \"taskType\"}", "auditTime": "2024-06-13T20:59:51.622Z" } ]

Pod 适配器：如何构建 Job 的 Pod Spec

Pod 适配器决定了 Kubernetes Job 的 Pod 模板如何构建。适配器类型在KubernetesTaskRunnerFactory.buildTaskAdapter()中根据druid.indexer.runner.k8s.adapter.type属性选择（源码 KubernetesTaskRunnerFactory.java）。

Overlord Single Container Pod Adapter

该适配器直接取 Overlord Pod 的 podSpec，据此创建 Kubernetes Job，是默认的 Pod 适配器实现。显式启用方式：

druid.indexer.runner.k8s.adapter.type: overlordSingleContainer

Overlord Multi Container Pod Adapter

该适配器同样基于 Overlord Pod 的 podSpec 创建 Job，但会使用 kubexit 管理主容器（运行 Druid peon 的容器）与 Overlord Pod Spec 中定义的其它 sidecar 之间的依赖顺序。如果你的 sidecar 是 Splunk、Istio 等，该适配器可以正确处理它们。启用方式：

druid.indexer.runner.k8s.adapter.type: overlordMultiContainer

Multi Container 适配器的关键限制——必须显式声明 command：要让 sidecar 支持正常工作，Docker 镜像中的入口点/命令必须在 Pod Spec 中显式定义，否则扩展无法解析。以下写法不可行：

ENTRYPOINT: ["foo.sh"]

container: name: foo args: - arg1 - arg2

因为扩展无法推断 command 是什么。即使是 Istio 这类由 service mesh 动态注入的 sidecar 也同样需要显式声明。正确写法是在 sidecar spec 中显式给出 command：

container: name: foo command: foo.sh args: - arg1 - arg2

（Dockerfile 可以保持不变。）

对于上述两个适配器，可以通过以下属性为 K8s Job/Pod 添加可选标签与注解（值均为 JSON 对象字符串）：

druid.indexer.runner.labels: '{"key":"value"}' druid.indexer.runner.annotations: '{"key":"value"}'

迁移注意：原先为 Middle Manager 任务配置的所有其它配置都需要迁移到 Overlord 下，且 javaOpts 必须写成数组形式：

druid.indexer.runner.javaOptsArray

druid.indexer.runner.javaOpts已不再支持。

Custom Template Pod Adapter（自定义模板 Pod 适配器）

自定义模板适配器允许按任务类型指定 Pod 模板文件，以更灵活地定义 Pod。它要求 Overlord 文件系统上存在 Pod Template 文件，该模板作为 K8s Job Pod Spec 的基础，可覆盖 labels、环境变量、资源、注解甚至基础镜像。启用方式：

druid.indexer.runner.k8s.adapter.type: customTemplateAdapter druid.indexer.runner.k8s.podTemplate.base: /path/to/basePodSpec.yaml

示例 Pod Template

以下示例使用常规的 druid docker 镜像：

apiVersion: "v1" kind: "PodTemplate" template: metadata: annotations: sidecar.istio.io/proxyCPU: "512m" # to handle a injected istio sidecar labels: app.kubernetes.io/name: "druid-realtime-backend" spec: affinity: {} containers: - command: - sh - -c - | /peon.sh /druid/data 1 env: - name: CUSTOM_ENV_VARIABLE value: "hello" image: apache/druid:{{DRUIDVERSION}} name: main ports: - containerPort: 8091 name: druid-tls-port protocol: TCP - containerPort: 8100 name: druid-port protocol: TCP resources: limits: cpu: "1" memory: 2400M requests: cpu: "1" memory: 2400M volumeMounts: - mountPath: /opt/druid/conf/druid/cluster/master/coordinator-overlord # runtime props are still mounted in this location because that's where peon.sh looks for configs name: nodetype-config-volume readOnly: true - mountPath: /druid/data name:>kind: ConfigMap metadata: name: druid-tiny-cluster-peons-config namespace: default apiVersion: v1 data: jvm.config: |- -server -XX:MaxDirectMemorySize=1000M -Duser.timezone=UTC -Dfile.encoding=UTF-8 -Dlog4j.debug -Djava.util.logging.manager=org.apache.logging.log4j.jul.LogManager -Djava.io.tmpdir=/druid/data -Xmx1024M -Xms1024M log4j2.xml: |- <?xml version="1.0" encoding="UTF-8" ?> <Configuration status="WARN"> <Appenders> <Console name="Console" target="SYSTEM_OUT"> <PatternLayout pattern="%d{ISO8601} %p [%t] %c - %m%n"/> </Console> </Appenders> <Loggers> <Root level="info"> <AppenderRef ref="Console"/> </Root> </Loggers> </Configuration> runtime.properties: | druid.port=8100 druid.service=druid/peon druid.server.http.numThreads=5 druid.indexer.task.baseTaskDir=/druid/data druid.indexer.runner.type=k8s druid.peon.mode=remote druid.indexer.task.encapsulatedTask=true

Pod 模板选择策略

Custom Template Pod Adapter 可以使用动态执行配置来决定某个任务该用哪个 Pod 模板。

基于任务类型选择（TaskTypePodTemplateSelectStrategy）

该策略根据任务类型选择 Pod 模板，是默认的 Pod 模板选择策略。对应源码 TaskTypePodTemplateSelectStrategy.java：若模板集合中存在以任务类型为 key 的模板则使用它，否则回退到base模板。

显式指定该策略（即动态配置中的默认值）：

{ "type": "default" }

任务类型专属的 Pod 模板通过运行时属性指定，{taskType}为任务类型名称，例如index_parallel：

druid.indexer.runner.k8s.podTemplate.{taskType}: /path/to/taskSpecificPodSpec.yaml

环境变量转义说明：如果使用默认镜像的环境变量解析特性来设置运行时属性，指定 Pod 模板时需要多加一层转义下划线。例如，设置运行时属性druid.indexer.runner.k8s.podTemplate.index_kafka时，环境变量应写为druid_indexer_runner_k8s_podTemplate_index__kafka（index_kafka中的下划线转义为双下划线）。

基于任务类型的模板选择配置示例：

druid.indexer.runner.k8s.podTemplate.base=/path/to/basePodSpec.yaml druid.indexer.runner.k8s.podTemplate.index_kafka=/path/to/kafkaPodSpec.yaml

基于一个或多个条件选择（SelectorBasedPodTemplateSelectStrategy）

该策略评估selectors中的一系列条件来决定使用哪个 Pod 模板运行任务。Pod 模板在运行时属性中按druid.indexer.runner.k8s.podTemplate.<selectionKey>=...配置。对应源码为 SelectorBasedPodTemplateSelectStrategy.java 与 Selector.java。

{ "type": "selectorBased", "selectors": [ { "selectionKey": "podSpec1", "context.tags": { "userProvidedTag": ["tag1", "tag2"] }, "dataSource": ["wikipedia"] }, { "selectionKey": "podSpec2", "type": ["index_kafka"] } ] }

选择规则：

• 按顺序处理：Selectors 按顺序求值，Druid 选择第一个匹配的 selector 对应的模板；
• 兜底回退：任务不匹配任何 selector 时使用basePod 模板；
• 条件全匹配：selector 内的所有条件都必须满足才算匹配。一个 selector 可以匹配：
  • type：任务的类型；
  • dataSource：任务的目标数据源；
  • context.tags：任务 context 中传入的标签。

从 Selector.java 的evaluate()实现可以看到：context.tags条件要求任务 context 中tags字段非空，且指定标签 key 的值命中条件值集合；type与dataSource条件均为集合包含判断；任一条件不满足即不匹配。

完整示例

先设置运行时属性定义可供选择的 Pod Spec：

druid.indexer.runner.k8s.podTemplate.base=/path/to/basePodSpec.yaml druid.indexer.runner.k8s.podTemplate.podSpec1=/path/to/podSpecWithHighMemRequests.yaml druid.indexer.runner.k8s.podTemplate.podSpec2=/path/to/podSpecWithLowCpuRequests.yaml

再通过动态执行配置定义选择策略：

{ "type": "default", "podTemplateSelectStrategy": { "type": "selectorBased", "selectors": [ { "selectionKey": "podSpec1", "context.tags": { "userProvidedTag": ["tag1", "tag2"] }, "dataSource": ["wikipedia"] }, { "selectionKey": "podSpec2", "type": ["index_kafka"] } ] } }

Druid 将按如下规则选择模板：

1. 当以下两个条件同时满足时使用podSpecWithHighMemRequests.yaml：
   1. 任务 context 包含 key 为userProvidedTag、值为tag1或tag2的标签；
   2. 任务面向wikipedia数据源。
2. 任务类型为index_kafka时使用podSpecWithLowCpuRequests.yaml；
3. 其余任务一律使用basePodSpec.yaml。

注意：若存在一个针对wikipedia数据源、带userProvidedTag: tag1标签的index_kafka任务，由于第一个 selector 先匹配，Druid 会选择podSpecWithHighMemRequests.yaml（而非 podSpec2）。

完整属性参考

下表为 K8s task runner 的全部运行时属性（默认值与约束可对照源码 KubernetesTaskRunnerConfig.java 验证）：

补充说明（源码层面）：disableClientProxy在 KubernetesOverlordModule.java 的makeKubernetesClient()中生效——当其为 true 时，构建 fabric8 KubernetesConfig并将 http/https 代理置空。peonMonitors的设计动机是：ForkingTaskRunner 从 MM 继承 monitors，而在 k8s 模式下 peon 会从 Overlord 继承 monitors，如果 Overlord 配置了例如TaskCountStatsMonitor，peon 进程可能因无法注入该 monitor 而失败。

新增指标

注意事项（Gotchas）

• 统一命名空间：属于同一个 Druid 集群的所有 Druid Pod 必须位于同一个 Kubernetes 命名空间内；
• RBAC 权限：Overlord 的 service account 必须具备与 Kubernetes 交互所需的 role binding。示例如下：

kind: Role apiVersion: rbac.authorization.k8s.io/v1 metadata: namespace: <druid-namespace> name: druid-k8s-task-scheduler rules: - apiGroups: ["batch"] resources: ["jobs"] verbs: ["get", "watch", "list", "delete", "create"] - apiGroups: [""] resources: ["pods", "pods/log"] verbs: ["get", "watch", "list", "delete", "create"] --- kind: RoleBinding apiVersion: rbac.authorization.k8s.io/v1 metadata: name: druid-k8s-binding namespace: <druid-namespace> subjects: - kind: ServiceAccount name: <druid-overlord-k8s-service-account> namespace: <druid-namespace> roleRef: kind: Role name: druid-k8s-task-scheduler apiGroup: rbac.authorization.k8s.io

从权限列表可以看出 Overlord 需要管理batch/jobs以及pods、pods/log资源，这正对应KubernetesPeonClient提交 Job、查看日志与清理 Job 的底层操作。

迁移模式：Kubernetes and Worker Task Runner（零停机迁移）

如果集群当前仍有任务运行在 Middle Manager 或 Indexer 上，并希望零停机迁移到无 MM 摄入，可以使用迁移模式：该模式能够同时从 Middle Manager/Indexer 与 Kubernetes 读取任务，并将任务写入 Middle Manager 或 Kubernetes 中的任意一方。

启用方式（将druid.indexer.runner.type: k8s替换为）：

druid.indexer.runner.type: k8sAndWorker

迁移模式附加配置

源码层面的实现要点（见 KubernetesAndWorkerTaskRunnerConfig.java 与 KubernetesOverlordModule.java）：

• workerType只允许httpRemote（HttpRemoteTaskRunnerFactory，基于 HTTP 的远程任务运行器，不依赖 Zookeeper）或remote（RemoteTaskRunnerFactory，基于 Zookeeper）二者之一，配置其它值会在启动时抛出校验异常；
• runnerStrategy.type由RunnerStrategyProvider动态注入：它读取druid.indexer.runner.k8sAndWorker.runnerStrategy.{strategy}前缀下的配置，并通过JsonConfigProvider构建对应的RunnerStrategy实现（KubernetesRunnerStrategy/WorkerRunnerStrategy/TaskTypeRunnerStrategy，位于 runnerstrategy 包）；
• 迁移模式下 Overlord 需要同时具备两套 runner 的组件：KubernetesTaskRunnerFactory与HttpRemoteTaskRunnerFactory/RemoteTaskRunnerFactory均已在模块中注册。

总结

druid-kubernetes-overlord-extensions为 Kubernetes 上的 Druid 提供了"无 Middle Manager"的摄入执行路径：任务以 K8s Job 形式运行，天然与 Druid 部署解耦并可在 Overlord 重启后恢复；通过三种 Pod 适配器（单容器、多容器、自定义模板）灵活定义 Job 的 Pod Spec；通过动态执行配置 API 在线调整 Pod 模板选择策略；k8sAndWorker混合模式则支撑从传统 Middle Manager 集群零停机迁移。在使用时请特别注意capacity与线程/内存的关系、多容器适配器必须显式声明 command、统一命名空间与 RBAC 权限等约束。

延伸阅读

• 扩展的完整官方文档
• 扩展 README
• KubernetesOverlordModule 源码（模块与 Guice 绑定）
• KubernetesTaskRunner 源码（任务执行与恢复核心）
• KubernetesTaskRunnerConfig 源码（全部运行时属性）
• 动态配置与 Pod 模板选择策略实现
• 测试用例目录（覆盖配置解析、适配器与选择策略）
• 扩展加载方式
• Kubernetes 上的集群部署指南

• 数据库
• OLAP
• 大数据
• 后端