K8s中FlinkDeployment的Service自动创建及自定义注解配置咨询

问题背景 #

我正在使用flink-kubernetes-operator 1.9-SNAPSHOT和Flink 1.18.1版本，希望为FlinkDeployment对应的Service资源添加以下自定义注解：

annotations:
  routing.mesh.net/enabled: "false"
  networking.istio.io/exportTo: "~"

但我无法找到添加这些注解的正确位置。执行helm plan时未发现任何Service相关内容，尝试在FlinkDeployment的spec.jobManager.service下添加注解时，出现错误：strict decoding error: unknown field "spec.jobManager.service"。现咨询以下问题：

1. 每个Flink作业的JobManager对应的Service是由谁/如何创建的？
2. 如何为该Service添加自定义配置？
3. 是否有我遗漏的相关详细官方文档？

附我的FlinkDeployment配置内容：

apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
  name: appzookeeper-control-parser
  namespace: cp-ce-cloud-infra-toc
  labels:
    ...
spec:
  mode: native
  image: "..."
  flinkVersion: v1_18
  serviceAccount: toc-flink-cluste-osd
  flinkConfiguration:
    execution.checkpointing.interval: "60s"
    heartbeat.timeout: "300000"
    taskmanager.memory.process.size: "2048m"
    high-availability.type: "NONE"
    state.checkpoints.dir: ...
    state.backend.fs.checkpointdir: ...
    state.savepoints.dir: ...
    blob.server.port: "6124"
    jobmanager.rpc.port: "6123"
    taskmanager.rpc.port: "6122"
    taskmanager.data.port: "6121"
    metrics.internal.query-service.port: "6120"
  podTemplate:
    apiVersion: v1
    kind: Pod
    metadata:
      name: pod-template
      labels:
        ...
    spec:
      automountServiceAccountToken: true
      securityContext:
        runAsUser: 9999
        runAsNonRoot: true
        fsGroup: 2000
      containers:
        - name: flink-main-container
          securityContext:
            capabilities:
              drop:
                - all
            runAsUser: 9999
            runAsNonRoot: true
            allowPrivilegeEscalation: false
          env:
            ...
            - name: JAVA_TOOL_OPTIONS
              value: -Dlog4j2.formatMsgNoLookups=true
            - name: TOC_FLINK_JOB_DEBUG_ENABLED
              value: "true"
            - name: JDK_JAVA_OPTIONS
              value: -agentlib:jdwp=transport=dt_socket,server=y,suspend=n,address=5005
            - name: JOB_NAME
              value: appzookeeper-control-parser
          volumeMounts:
            - mountPath: /opt/flink/log
              name: flink-logs
            - mountPath: /opt/flink/downloads
              name: downloads
            - mountPath: /data/flink/rocksdb
              name: flink-rocksdb-localdir
      volumes:
        - name: flink-logs
          emptyDir:
            sizeLimit: 5Gi
        - name: downloads
          emptyDir:
            sizeLimit: 5Gi
        - name: flink-rocksdb-localdir
          emptyDir:
            sizeLimit: 2Gi
  jobManager:
    replicas: 1
    resource:
      memory: "1g"
      cpu: 1
    service:
      annotations:
        routing.mesh.company.net/enabled: "false"
        networking.istio.io/exportTo: "~"
    podTemplate:
      apiVersion: v1
      kind: Pod
      metadata:
        name: job-manager-pod-template
        annotations:
          rsyslog.k8s-integration.company.com/inject: enabled
          identity.eks.company.com/use-native-identity: "true"
          opencensus.k8s-integration.company.com/inject: enabled
          logging.company.com/exclude.flink-main-container: "true"

        labels:
          app.kubernetes.io/instance: render-helm-output-manifest
          component: jobmanager
          ...
      spec:
        topologySpreadConstraints:
          - maxSkew: 1
            topologyKey: topology.kubernetes.io/zone
            whenUnsatisfiable: ScheduleAnyway
            labelSelector:
              matchLabels:
                app.kubernetes.io/instance: render-helm-output-manifest
                component: jobmanager
                ...
  taskManager:
    resource:
      memory: "2g"
      cpu: 1
    podTemplate:
      apiVersion: v1
      kind: Pod
      metadata:
        name: task-manager-pod-template
        annotations:
          rsyslog.k8s-integration.company.com/inject: enabled

        labels:
          app.kubernetes.io/instance: render-helm-output-manifest
          component: taskmanager
          ...
      spec:
        affinity:
          podAffinity:
            preferredDuringSchedulingIgnoredDuringExecution:
              - weight: 100
                podAffinityTerm:
                  labelSelector:
                    matchExpressions:
                      - key: component
                        operator: In
                        values:
                          - taskmanager
                      - key: p_service_instance
                        operator: In
                        values:
                          - cloud-toc-flink-cluster
                  topologyKey: topology.kubernetes.io/zone
  job:
    jarURI: local:///opt/flink/lib/pipeline-1.18.1-jar-with-dependencies.jar
    entryClass: com.company.toc.logparsers.job.MultilineLogParserJob
    upgradeMode: stateless
    parallelism: 1
    state: running
    allowNonRestoredState: True
    args:
      ...

问题解答 #

1. JobManager Service的创建主体与方式 #

JobManager对应的Service是由Flink Kubernetes Operator直接创建和管理的。当提交FlinkDeployment资源后，operator会根据CRD定义自动生成对应的Service资源，用于暴露JobManager的RPC和UI端口。这个Service属于operator运行时动态创建的资源，而非helm直接部署的内容，因此不会出现在helm plan的输出中。

2. 为Service添加自定义注解的正确方式 #

你遇到的strict decoding error是因为当前1.9-SNAPSHOT版本的FlinkDeployment CRD尚未支持spec.jobManager.service字段。要添加自定义注解，可采用以下两种方式：

方式一：通过Flink配置传递（单作业生效）

在spec.flinkConfiguration中添加带特定前缀的配置项，operator会自动将这些配置映射为Service的注解：

flinkConfiguration:
  kubernetes.jobmanager.service.annotations.routing.mesh.net/enabled: "false"
  kubernetes.jobmanager.service.annotations.networking.istio.io/exportTo: "~"

方式二：修改operator的Helm配置（全局生效）

如果需要为所有由该operator管理的作业统一添加注解，可在安装operator时通过helm values配置：

jobManager:
  service:
    annotations:
      routing.mesh.net/enabled: "false"
      networking.istio.io/exportTo: "~"

注意这种方式会对所有作业生效，无法单独配置单个作业。

3. 相关官方文档说明 #

当前1.9-SNAPSHOT版本处于开发阶段，部分功能文档可能未同步更新。可参考以下内容：

• Flink官方Kubernetes部署文档中关于kubernetes.jobmanager.service.annotations.*前缀配置的说明，这些配置会被operator用于生成Service注解。
• 通过kubectl get crd flinkdeployments.flink.apache.org -o yaml查看CRD定义文件，确认支持的字段结构，判断是否包含spec.jobManager.service字段。

内容的提问来源于stack exchange，提问作者ifellonscissors