第4章 Operator

4.1. Cluster Operator

Cluster Operator を使用して Kafka クラスターや他の Kafka コンポーネントをデプロイします。

Cluster Operator は YAML インストールファイルを使用してデプロイされます。Cluster Operator のデプロイメントに関する詳細は、「Cluster Operator のデプロイ」を参照してください。

Kafka で利用可能なデプロイメントオプションの詳細は、「Kafka Cluster の設定」を参照してください。

注記

OpenShift では、Kafka Connect デプロイメントに Source2Image 機能を組み込み、追加のコネクターを加えるための便利な方法として利用できます。

4.1.1. Cluster Operator

AMQ Streams では、Cluster Operator を使用して以下のクラスターをデプロイおよび管理します。

  • Kafka (ZooKeeper、Entity Operator、Kafka Exporter、Cruise Control を含む)
  • Kafka Connect
  • Kafka MirrorMaker
  • Kafka Bridge

クラスターのデプロイメントにはカスタムリソースが使用されます。

たとえば、以下のように Kafka クラスターをデプロイします。

  • クラスター設定のある Kafka リソースが OpenShift クラスター内で作成されます。
  • Kafka リソースに宣言された内容を基にして、該当する Kafka クラスターが Cluster Operator によってデプロイされます。

Cluster Operator で以下もデプロイできます (Kafka リソースの設定より)。

  • KafkaTopic カスタムリソースより Operator スタイルのトピック管理を提供する Topic Operator
  • KafkaUser カスタムリソースより Operator スタイルのユーザー管理を提供する User Operator

デプロイメントの Entity Operator 内の Topic Operator および User Operator 関数。

Cluster Operator のアーキテクチャー例

Cluster Operator

4.1.2. 調整

Operator は OpenShift クラスターから受信する必要なクラスターリソースに関するすべての通知に対応しますが、Operator が実行されていない場合や、何らかの理由で通知が受信されない場合、必要なリソースは実行中の OpenShift クラスターの状態と同期しなくなります。

フェイルオーバーを適切に処理するために、Cluster Operator によって定期的な調整プロセスが実行され、必要なリソースすべてで一貫した状態になるように、必要なリソースの状態を現在のクラスターデプロイメントと比較できます。[STRIMZI_FULL_RECONCILIATION_INTERVAL_MS] 変数を使用して、定期的な調整の期間を設定できます。

4.1.3. Cluster Operator の設定

Cluster Operator は、以下のサポートされる環境変数を使用して設定できます。

STRIMZI_NAMESPACE

Operator が操作する namespace のカンマ区切りのリスト。設定されていない場合や、空の文字列や * に設定された場合は、Cluster Operator はすべての namespace で操作します。Cluster Operator デプロイメントでは OpenShift Downward API を使用して、これを Cluster Operator がデプロイされる namespace に自動設定することがあります。以下に例を示します。

env:
  - name: STRIMZI_NAMESPACE
    valueFrom:
      fieldRef:
        fieldPath: metadata.namespace
STRIMZI_FULL_RECONCILIATION_INTERVAL_MS
任意設定、デフォルトは 120000 ミリ秒です。定期的な調整の間隔 (秒単位)。
STRIMZI_LOG_LEVEL
任意設定、デフォルトは INFO です。ロギングメッセージの出力レベル。設定可能な値: ERRORWARNINGINFODEBUG、および TRACE
STRIMZI_OPERATION_TIMEOUT_MS
任意設定、デフォルトは 300000 ミリ秒です。内部操作のタイムアウト (ミリ秒単位)。この値は、標準の OpenShift 操作の時間が通常よりも長いクラスターで (Docker イメージのダウンロードが遅い場合など) AMQ Streams を使用する場合に増やす必要があります。
STRIMZI_KAFKA_IMAGES
必須。Kafka バージョンから、そのバージョンの Kafka ブローカーが含まれる該当の Docker イメージへのマッピングが提供されます。必要な構文は、空白またはカンマ区切りの <version>=<image> ペアです。例: 2.4.1=registry.redhat.io/amq7/amq-streams-kafka-24-rhel7:1.5.0, 2.5.0=registry.redhat.io/amq7/amq-streams-kafka-25-rhel7:1.5.0これは、「コンテナーイメージ」に説明されているように、Kafka.spec.kafka.version プロパティーは指定されていても Kafka.spec.kafka.image プロパティーは指定されていない場合に使用されます。
STRIMZI_DEFAULT_KAFKA_INIT_IMAGE
任意設定、デフォルトは registry.redhat.io/amq7/amq-streams-rhel7-operator:1.5.0 です。「コンテナーイメージ」kafka-init-image として指定されたイメージがない場合に、初期設定作業 (ラックサポート) のブローカーの前に開始される init コンテナーのデフォルトとして使用するイメージ名。
STRIMZI_DEFAULT_TLS_SIDECAR_KAFKA_IMAGE
任意設定、デフォルトは registry.redhat.io/amq7/amq-streams-kafka-25-rhel7:1.5.0 です。「コンテナーイメージ」Kafka.spec.kafka.tlsSidecar.image として指定されたイメージがない場合に、Kafka の TLS サポートを提供するサイドカーコンテナーをデプロイする際にデフォルトとして使用するイメージ名。
STRIMZI_KAFKA_CONNECT_IMAGES
必須。Kafka バージョンから、そのバージョンの Kafka Connect が含まれる該当の Docker イメージへのマッピングが提供されます。必要な構文は、空白またはカンマ区切りの <version>=<image> ペアです。例: 2.4.1=registry.redhat.io/amq7/amq-streams-kafka-24-rhel7:1.5.0, 2.5.0=registry.redhat.io/amq7/amq-streams-kafka-25-rhel7:1.5.0これは、「コンテナーイメージ」に説明されているように、KafkaConnect.spec.version プロパティーは指定されていても KafkaConnect.spec.image プロパティーは指定されていない場合に使用されます。
STRIMZI_KAFKA_CONNECT_S2I_IMAGES
必須。Kafka バージョンから、そのバージョンの Kafka Connect が含まれる該当の Docker イメージへのマッピングが提供されます。必要な構文は、空白またはカンマ区切りの <version>=<image> ペアです。例: 2.4.1=registry.redhat.io/amq7/amq-streams-kafka-24-rhel7:1.5.0, 2.5.0=registry.redhat.io/amq7/amq-streams-kafka-25-rhel7:1.5.0これは、「コンテナーイメージ」に説明されているように、KafkaConnectS2I.spec.version プロパティーは指定されていても KafkaConnectS2I.spec.image プロパティーは指定されていない場合に使用されます。
STRIMZI_KAFKA_MIRROR_MAKER_IMAGES
必須。Kafka バージョンから、そのバージョンの Kafka Mirror Maker が含まれる該当の Docker イメージへのマッピングが提供されます。必要な構文は、空白またはカンマ区切りの <version>=<image> ペアです。例: 2.4.1=registry.redhat.io/amq7/amq-streams-kafka-24-rhel7:1.5.0, 2.5.0=registry.redhat.io/amq7/amq-streams-kafka-25-rhel7:1.5.0これは、「コンテナーイメージ」に説明されているように、KafkaMirrorMaker.spec.version プロパティーは指定されていても KafkaMirrorMaker.spec.image プロパティーは指定されていない場合に使用されます。
STRIMZI_DEFAULT_TOPIC_OPERATOR_IMAGE
任意設定、デフォルトは registry.redhat.io/amq7/amq-streams-rhel7-operator:1.5.0 です。Kafka リソースの 「コンテナーイメージ」Kafka.spec.entityOperator.topicOperator.image として指定されたイメージがない場合に、Topic Operator のデプロイ時にデフォルトとして使用するイメージ名。
STRIMZI_DEFAULT_USER_OPERATOR_IMAGE
任意設定、デフォルトは registry.redhat.io/amq7/amq-streams-rhel7-operator:1.5.0 です。Kafka リソースの 「コンテナーイメージ」Kafka.spec.entityOperator.userOperator.image として指定されたイメージがない場合に、User Operator のデプロイ時にデフォルトとして使用するイメージ名。
STRIMZI_DEFAULT_TLS_SIDECAR_ENTITY_OPERATOR_IMAGE
任意設定、デフォルトは registry.redhat.io/amq7/amq-streams-kafka-25-rhel7:1.5.0 です。「コンテナーイメージ」Kafka.spec.entityOperator.tlsSidecar.image として指定されたイメージがない場合に、Entity Operator の TLS サポートを提供するサイドカーコンテナーをデプロイする際にデフォルトとして使用するイメージ名。
STRIMZI_IMAGE_PULL_POLICY
任意設定。AMQ Streams の Cluster Operator によって管理されるすべての Pod のコンテナーに適用される ImagePullPolicy。有効な値は、AlwaysIfNotPresent、および Never です。指定のない場合、OpenShift のデフォルトが使用されます。ポリシーを変更すると、すべての Kafka、Kafka Connect、および Kafka MirrorMaker クラスターのローリングアップデートが実行されます。
STRIMZI_IMAGE_PULL_SECRETS
任意設定。Secret 名のカンマ区切りのリスト。ここで参照されるシークレットには、コンテナーイメージがプルされるコンテナーレジストリーへのクレデンシャルが含まれます。シークレットは、Cluster Operator によって作成されるすべての PodsimagePullSecrets フィールドで使用されます。このリストを変更すると、Kafka、Kafka Connect、および Kafka MirrorMaker のすべてのクラスターのローリングアップデートが実行されます。
STRIMZI_KUBERNETES_VERSION

任意設定。API サーバーから検出された OpenShift バージョン情報をオーバーライドします。以下に例を示します。

env:
  - name: STRIMZI_KUBERNETES_VERSION
    value: |
           major=1
           minor=16
           gitVersion=v1.16.2
           gitCommit=c97fe5036ef3df2967d086711e6c0c405941e14b
           gitTreeState=clean
           buildDate=2019-10-15T19:09:08Z
           goVersion=go1.12.10
           compiler=gc
           platform=linux/amd64
KUBERNETES_SERVICE_DNS_DOMAIN

任意設定。デフォルトの OpenShift DNS サフィックスを上書きします。

デフォルトでは、OpenShfit クラスターで割り当てられるサービスに、デフォルトのサフィックス cluster.local を使用する DNS ドメイン名があります。

ブローカーが kafka-0 の場合の例は次のとおりです。

<cluster-name>-kafka-0.<cluster-name>-kafka-brokers.<namespace>.svc.cluster.local

DNS ドメイン名は、ホスト名の検証に使用される Kafka ブローカー証明書に追加されます。

クラスターで異なる DNS サフィックスを使用している場合、Kafka ブローカーとの接続を確立するために、KUBERNETES_SERVICE_DNS_DOMAIN 環境変数をデフォルトから現在使用中の DNS サフィックスに変更します。

4.1.4. ロールベースアクセス制御 (RBAC)

4.1.4.1. Cluster Operator のロールベースアクセス制御 (RBAC) のプロビジョニング

Cluster Operator が機能するには、KafkaKafkaConnect などのリソースや ConfigMapsPodsDeploymentsStatefulSetsServices などの管理リソースと対話するために OpenShift クラスター内でパーミッションが必要になり ます。このようなパーミッションは、OpenShift のロールベースアクセス制御 (RBAC) リソースに記述されます。

  • ServiceAccount
  • Role および ClusterRole
  • RoleBinding および ClusterRoleBinding

Cluster Operator は、ClusterRoleBinding を使用して独自の ServiceAccount で実行される他に、OpenShift リソースへのアクセスを必要とするコンポーネントの RBAC リソースを管理します。

また OpenShift には、ServiceAccount で動作するコンポーネントが、その ServiceAccount にはない他の ServiceAccounts の権限を付与しないようにするための特権昇格の保護機能も含まれています。Cluster Operator は、ClusterRoleBindings と、それが管理するリソースで必要な RoleBindings を作成できる必要があるため、Cluster Operator にも同じ権限が必要です。

4.1.4.2. 委譲された権限

Cluster Operator が必要な Kafka リソースのリソースをデプロイする場合、以下のように ServiceAccountsRoleBindings、および ClusterRoleBindings も作成します。

  • Kafka ブローカー Pod は cluster-name-kafka という ServiceAccount を使用します。

    • ラック機能が使用されると、strimzi-cluster-name-kafka-init ClusterRoleBinding は、strimzi-kafka-broker と呼ばれる ClusterRole 経由で、クラスター内のノードへの ServiceAccount アクセスを付与するために使用されます。
    • ラック機能が使用されていない場合は、バインディングは作成されません。
  • ZooKeeper Pod は cluster-name-zookeeper という ServiceAccount を使用します。
  • Entity Operator は、cluster-name-entity-operator という ServiceAccount を使用します。

    • Topic Operator はステータス情報のある OpenShift イベントを生成し、ServiceAccountstrimzi-entity-operator という ClusterRole にバインドされるようにします。strimzi-entity-operator はこのアクセス権限を strimzi-entity-operator RoleBinding 経由で付与します。
  • KafkaConnect および KafkaConnectS2I リソースの Pod は cluster-name-cluster-connect という ServiceAccount を使用します。
  • KafkaMirrorMaker の Pod は cluster-name-mirror-maker というServiceAccount を使用します。
  • KafkaBridge の Pod は cluster-name-bridge というServiceAccount を使用します。

4.1.4.3. ServiceAccount

Cluster Operator は ServiceAccount を使用して最適に実行されます。

Cluster Operator の ServiceAccount の例

apiVersion: v1
kind: ServiceAccount
metadata:
  name: strimzi-cluster-operator
  labels:
    app: strimzi

その後、Cluster Operator の Deployment で、これを spec.template.spec.serviceAccountName に指定する必要があります。

Cluster Operator の Deployment の部分的な例

apiVersion: apps/v1
kind: Deployment
metadata:
  name: strimzi-cluster-operator
  labels:
    app: strimzi
spec:
  replicas: 1
  selector:
    matchLabels:
      name: strimzi-cluster-operator
      strimzi.io/kind: cluster-operator
  template:
      # ...

strimzi-cluster-operator ServiceAccountserviceAccountName として指定されている 12 行目に注目してください。

4.1.4.4. ClusterRoles

Cluster Operator は、必要なリソースへのアクセス権限を付与する ClusterRoles を使用して操作する必要があります。OpenShift クラスターの設定によっては、クラスター管理者が ClusterRoles を作成する必要があることがあります。

注記

クラスター管理者の権限は ClusterRoles の作成にのみ必要です。Cluster Operator はクラスター管理者アカウントで実行されません。

ClusterRoles は、 最小権限の原則に従い、Kafka、Kafka Connect、および ZooKeeper クラスターを操作するために Cluster Operator が必要とする権限のみが含まれます。最初に割り当てられた一連の権限により、Cluster Operator で StatefulSetsDeploymentsPods、および ConfigMaps などの OpenShift リソースを管理できます。

Cluster Operator は ClusterRoles を使用して、namespace スコープリソースのレベルおよびクラスタースコープリソースのレベルで権限を付与します。

Cluster Operator の namespaced リソースのある ClusterRole

apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
  name: strimzi-cluster-operator-namespaced
  labels:
    app: strimzi
rules:
- apiGroups:
  - ""
  resources:
  # The cluster operator needs to access and manage service accounts to grant Strimzi components cluster permissions
  - serviceaccounts
  verbs:
  - get
  - create
  - delete
  - patch
  - update
- apiGroups:
  - "rbac.authorization.k8s.io"
  resources:
  # The cluster operator needs to access and manage rolebindings to grant Strimzi components cluster permissions
  - rolebindings
  verbs:
  - get
  - create
  - delete
  - patch
  - update
- apiGroups:
  - ""
  resources:
  # The cluster operator needs to access and manage config maps for Strimzi components configuration
  - configmaps
  # The cluster operator needs to access and manage services to expose Strimzi components to network traffic
  - services
  # The cluster operator needs to access and manage secrets to handle credentials
  - secrets
  # The cluster operator needs to access and manage persistent volume claims to bind them to Strimzi components for persistent data
  - persistentvolumeclaims
  verbs:
  - get
  - list
  - watch
  - create
  - delete
  - patch
  - update
- apiGroups:
  - "kafka.strimzi.io"
  resources:
  # The cluster operator runs the KafkaAssemblyOperator, which needs to access and manage Kafka resources
  - kafkas
  - kafkas/status
  # The cluster operator runs the KafkaConnectAssemblyOperator, which needs to access and manage KafkaConnect resources
  - kafkaconnects
  - kafkaconnects/status
  # The cluster operator runs the KafkaConnectS2IAssemblyOperator, which needs to access and manage KafkaConnectS2I resources
  - kafkaconnects2is
  - kafkaconnects2is/status
  # The cluster operator runs the KafkaConnectorAssemblyOperator, which needs to access and manage KafkaConnector resources
  - kafkaconnectors
  - kafkaconnectors/status
  # The cluster operator runs the KafkaMirrorMakerAssemblyOperator, which needs to access and manage KafkaMirrorMaker resources
  - kafkamirrormakers
  - kafkamirrormakers/status
  # The cluster operator runs the KafkaBridgeAssemblyOperator, which needs to access and manage BridgeMaker resources
  - kafkabridges
  - kafkabridges/status
  # The cluster operator runs the KafkaMirrorMaker2AssemblyOperator, which needs to access and manage KafkaMirrorMaker2 resources
  - kafkamirrormaker2s
  - kafkamirrormaker2s/status
  # The cluster operator runs the KafkaRebalanceAssemblyOperator, which needs to access and manage KafkaRebalance resources
  - kafkarebalances
  - kafkarebalances/status
  verbs:
  - get
  - list
  - watch
  - create
  - delete
  - patch
  - update
- apiGroups:
  - ""
  resources:
  # The cluster operator needs to access and delete pods, this is to allow it to monitor pod health and coordinate rolling updates
  - pods
  verbs:
  - get
  - list
  - watch
  - delete
- apiGroups:
  - ""
  resources:
  - endpoints
  verbs:
  - get
  - list
  - watch
- apiGroups:
  # The cluster operator needs the extensions api as the operator supports Kubernetes version 1.11+
  # apps/v1 was introduced in Kubernetes 1.14
  - "extensions"
  resources:
  # The cluster operator needs to access and manage deployments to run deployment based Strimzi components
  - deployments
  - deployments/scale
  # The cluster operator needs to access replica sets to manage Strimzi components and to determine error states
  - replicasets
  # The cluster operator needs to access and manage replication controllers to manage replicasets
  - replicationcontrollers
  # The cluster operator needs to access and manage network policies to lock down communication between Strimzi components
  - networkpolicies
  # The cluster operator needs to access and manage ingresses which allow external access to the services in a cluster
  - ingresses
  verbs:
  - get
  - list
  - watch
  - create
  - delete
  - patch
  - update
- apiGroups:
  - "apps"
  resources:
  # The cluster operator needs to access and manage deployments to run deployment based Strimzi components
  - deployments
  - deployments/scale
  - deployments/status
  # The cluster operator needs to access and manage stateful sets to run stateful sets based Strimzi components
  - statefulsets
  # The cluster operator needs to access replica-sets to manage Strimzi components and to determine error states
  - replicasets
  verbs:
  - get
  - list
  - watch
  - create
  - delete
  - patch
  - update
- apiGroups:
  - ""
  resources:
  # The cluster operator needs to be able to create events and delegate permissions to do so
  - events
  verbs:
  - create
- apiGroups:
  # OpenShift S2I requirements
  - apps.openshift.io
  resources:
  - deploymentconfigs
  - deploymentconfigs/scale
  - deploymentconfigs/status
  - deploymentconfigs/finalizers
  verbs:
  - get
  - list
  - watch
  - create
  - delete
  - patch
  - update
- apiGroups:
  # OpenShift S2I requirements
  - build.openshift.io
  resources:
  - buildconfigs
  - builds
  verbs:
  - create
  - delete
  - get
  - list
  - patch
  - watch
  - update
- apiGroups:
  # OpenShift S2I requirements
  - image.openshift.io
  resources:
  - imagestreams
  - imagestreams/status
  verbs:
  - create
  - delete
  - get
  - list
  - watch
  - patch
  - update
- apiGroups:
  - networking.k8s.io
  resources:
  # The cluster operator needs to access and manage network policies to lock down communication between Strimzi components
  - networkpolicies
  verbs:
  - get
  - list
  - watch
  - create
  - delete
  - patch
  - update
- apiGroups:
  - route.openshift.io
  resources:
  # The cluster operator needs to access and manage routes to expose Strimzi components for external access
  - routes
  - routes/custom-host
  verbs:
  - get
  - list
  - create
  - delete
  - patch
  - update
- apiGroups:
  - policy
  resources:
  # The cluster operator needs to access and manage pod disruption budgets this limits the number of concurrent disruptions
  # that a Strimzi component experiences, allowing for higher availability
  - poddisruptionbudgets
  verbs:
  - get
  - list
  - watch
  - create
  - delete
  - patch
  - update

2 番目の一連の権限には、クラスタースコープリソースに必要な権限が含まれます。

Cluster Operator のクラスタースコープリソースのある ClusterRole

apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
  name: strimzi-cluster-operator-global
  labels:
    app: strimzi
rules:
- apiGroups:
  - "rbac.authorization.k8s.io"
  resources:
  # The cluster operator needs to create and manage cluster role bindings in the case of an install where a user
  # has specified they want their cluster role bindings generated
  - clusterrolebindings
  verbs:
  - get
  - create
  - delete
  - patch
  - update
  - watch
- apiGroups:
  - storage.k8s.io
  resources:
  # The cluster operator requires "get" permissions to view storage class details
  # This is because only a persistent volume of a supported storage class type can be resized
  - storageclasses
  verbs:
  - get
- apiGroups:
  - ""
  resources:
  # The cluster operator requires "list" permissions to view all nodes in a cluster
  # The listing is used to determine the node addresses when NodePort access is configured
  # These addresses are then exposed in the custom resource states
  - nodes
  verbs:
  - list

strimzi-kafka-broker ClusterRole は、ラック機能に使用される Kafka Pod の init コンテナーが必要とするアクセス権限を表します。「委譲された権限」で説明したように、このアクセスを委譲できるようにするには、このロールも Cluster Operator に必要です。

Cluster Operator の ClusterRole により、OpenShift ノードへのアクセスを Kafka ブローカー Pod に委譲できます。

apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
  name: strimzi-kafka-broker
  labels:
    app: strimzi
rules:
- apiGroups:
  - ""
  resources:
  # The Kafka Brokers require "get" permissions to view the node they are on
  # This information is used to generate a Rack ID that is used for High Availability configurations
  - nodes
  verbs:
  - get

strimzi-topic-operatorClusterRole は、Topic Operator が必要とするアクセスを表します。「委譲された権限」で説明したように、このアクセスを委譲できるようにするには、このロールも Cluster Operator に必要です。

Cluster Operator の ClusterRole により、イベントへのアクセスを Topic Operator に委譲できます。

apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
  name: strimzi-entity-operator
  labels:
    app: strimzi
rules:
- apiGroups:
  - "kafka.strimzi.io"
  resources:
  # The entity operator runs the KafkaTopic assembly operator, which needs to access and manage KafkaTopic resources
  - kafkatopics
  - kafkatopics/status
  # The entity operator runs the KafkaUser assembly operator, which needs to access and manage KafkaUser resources
  - kafkausers
  - kafkausers/status
  verbs:
  - get
  - list
  - watch
  - create
  - patch
  - update
  - delete
- apiGroups:
  - ""
  resources:
  - events
  verbs:
  # The entity operator needs to be able to create events
  - create
- apiGroups:
  - ""
  resources:
  # The entity operator user-operator needs to access and manage secrets to store generated credentials
  - secrets
  verbs:
  - get
  - list
  - create
  - patch
  - update
  - delete

4.1.4.5. ClusterRoleBindings

Operator には ClusterRoleBindings と、ClusterRoleServiceAccount に関連付ける RoleBindings が必要です。ClusterRoleBindings は、クラスタースコープリロースが含まれる ClusterRoles に必要です。

Cluster Operator の ClusterRoleBinding の例

apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
metadata:
  name: strimzi-cluster-operator
  labels:
    app: strimzi
subjects:
- kind: ServiceAccount
  name: strimzi-cluster-operator
  namespace: myproject
roleRef:
  kind: ClusterRole
  name: strimzi-cluster-operator-global
  apiGroup: rbac.authorization.k8s.io

ClusterRoleBindings は、委譲に必要な ClusterRoles にも必要です。

Cluster Operator の RoleBinding の例

apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
metadata:
  name: strimzi-cluster-operator-kafka-broker-delegation
  labels:
    app: strimzi
# The Kafka broker cluster role must be bound to the cluster operator service account so that it can delegate the cluster role to the Kafka brokers.
# This must be done to avoid escalating privileges which would be blocked by Kubernetes.
subjects:
- kind: ServiceAccount
  name: strimzi-cluster-operator
  namespace: myproject
roleRef:
  kind: ClusterRole
  name: strimzi-kafka-broker
  apiGroup: rbac.authorization.k8s.io

namespaced リソースのみが含まれる ClusterRoles は、RoleBindings のみを使用してバインドされます。

apiVersion: rbac.authorization.k8s.io/v1
kind: RoleBinding
metadata:
  name: strimzi-cluster-operator
  labels:
    app: strimzi
subjects:
- kind: ServiceAccount
  name: strimzi-cluster-operator
  namespace: myproject
roleRef:
  kind: ClusterRole
  name: strimzi-cluster-operator-namespaced
  apiGroup: rbac.authorization.k8s.io
apiVersion: rbac.authorization.k8s.io/v1
kind: RoleBinding
metadata:
  name: strimzi-cluster-operator-entity-operator-delegation
  labels:
    app: strimzi
# The Entity Operator cluster role must be bound to the cluster operator service account so that it can delegate the cluster role to the Entity Operator.
# This must be done to avoid escalating privileges which would be blocked by Kubernetes.
subjects:
- kind: ServiceAccount
  name: strimzi-cluster-operator
  namespace: myproject
roleRef:
  kind: ClusterRole
  name: strimzi-entity-operator
  apiGroup: rbac.authorization.k8s.io

4.2. Topic Operator

Topic Operator はカスタムリソースを使用して Kafka トピックをを管理します。

以下のように Topic Operator がデプロイされます。

4.2.1. Topic Operator

Topic Operator は、OpenShift リソースより Kafka クラスターのトピックを管理する方法を提供します。

Topic Operator のアーキテクチャー例

Topic Operator

Topic Operator の役割は、対応する Kafka トピックと同期して Kafka トピックを記述する KafkaTopic OpenShift リソースのセットを保持することです。

KafkaTopic とトピックの関係は次のとおりです。

  • KafkaTopic が作成されると、Topic Operator によってトピックが作成されます。
  • KafkaTopic が削除されると、Topic Operator によってトピックが削除されます。
  • KafkaTopic が変更されると、Topick Operator によってトピックが更新されます。

上記と逆になるトピックと KafkaTopic の関係は次のとおりです。

  • トピックが Kafka クラスター内で作成されると、Operator によって KafkaTopic が作成されます。
  • トピックが Kafka クラスターから削除されると、Operator によって KafkaTopic が削除されます。
  • トピックが Kafka クラスターで変更されると、Operator によって KafkaTopic が更新されます。

このため、KafkaTopic をアプリケーションのデプロイメントの一部として宣言でき、トピックの作成は Topic Operator によって行われます。アプリケーションは、必要なトピックからの作成または消費のみに対処する必要があります。

トピックが再設定された場合や、別の Kafka ノードに再割り当てされた場合、KafkaTopic は常に最新の状態になります。

4.2.2. トピック処理用の Kafka クラスターの特定

KafkaTopic リソースには、このリソースが属する Kafka クラスターに適した名前 (Kafka リソースの名前から派生) を定義するラベルが含まれています。

apiVersion: kafka.strimzi.io/v1beta1
kind: KafkaTopic
metadata:
  name: my-topic
  labels:
    strimzi.io/cluster: my-cluster

ラベルは、KafkaTopic リソースを特定し、新しいトピックを作成するために、Topic Operator によって使用されます。また、以降のトピックの処理でも使用されます。

ラベルが Kafka クラスターと一致しない場合、Topic Operator は KafkaTopic を識別できず、トピックは作成されません。

4.2.3. Topic Operator について

Operator にとって解決しなければならない基本的な問題として、信頼できる唯一の情報源 (SSOT: single source of truth) がないことがあります。KafkaTopic リソースと Kafka 内のトピックの両方とも、Operator に関係なく変更される可能性があります面倒なことに、Topic Operator は KafkaTopic リソースと Kafka トピックで変更を常にリアルタイムで監視できるとは限りません (たとえば Operator が停止している場合もあります)。

これを解決するために、Operator は各トピックに関する情報の独自のプライベートコピーを維持します。Kafka クラスターまたは OpenShift で変更が生じると、他のシステムの状態とプライベートコピーの両方を確認し、すべての同期が保たれるように何を変更する必要があるかを判断します。同じことが Operator の起動時に必ず実行され、また Operator の稼働中にも定期的に行われます。

たとえば、Topic Operator が実行されていないときに KafkaTopicmy-topic が作成された場合を考えてみましょう。Operator は、起動時に「my-topic」のプライベートコピーを持たないので、Operator が前回稼働状態であった後に KafkaTopic が作成されたと推測できます。Operator によって「my-topic」に対応するトピックが作成され、さらに「my-topic」のメタデータのプライベートコピーが保存されます。

このプライベートコピーによって、Operator は、Kafka と OpenShift の両方でトピック設定が変更される場合に対処できますが、それができるのは変更に矛盾 (たとえば両方で同じトピックの config キーが異なる値に変更される場合など) がない場合に限ります。変更に矛盾がある場合、Kafka の設定が優先され、KafkaTopic はそれを反映する形で更新されます。

プライベートコピーは、Kafka が使用するのと同じ ZooKeeper アンサンブルに保持されます。これにより可用性の懸念が軽減されます。これは、ZooKeeper が実行中でなければ Kafka 自体を実行できないため、Operator がステートレスであっても可用性は下がらないからです。

4.2.4. リソース要求および制限のある Topic Operator の設定

CPU やメモリーなどのリソースを Topic Operator に割り当て、Topic Operator が消費できるリソースの量に制限を設定できます。

前提条件

  • Cluster Operator が稼働している必要があります。

手順

  1. 必要に応じてエディターで Kafka クラスター設定を更新します。

    oc edit を使用します。

    oc edit kafka my-cluster
  2. Kafka リソースの spec.entityOperator.topicOperator.resources プロパティーで、Topic Operator のリソース要求および制限を設定します。

    apiVersion: kafka.strimzi.io/v1beta1
    kind: Kafka
    spec:
      # kafka and zookeeper sections...
      entityOperator:
        topicOperator:
          resources:
            request:
              cpu: "1"
              memory: 500Mi
            limit:
              cpu: "1"
              memory: 500Mi
  3. 新しい設定を適用してリソースを作成または更新します。

    oc apply を使用します。

    oc apply -f kafka.yaml

その他のリソース

  • resources オブジェクトのスキーマの詳細は「<link xlink:href="https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.18/#resourcerequirements-v1-core">ResourceRequirements v1 core</link>」を参照してください。

4.3. User Operator

User Operator はカスタムリソースを使用して Kafka ユーザーを管理します。

以下のように User Operator がデプロイされます。

4.3.1. User Operator

User Operator は、Kafka ユーザーが記述される KafkaUser リソースを監視して Kafka クラスターの Kafka ユーザーを管理し、Kafka ユーザーが Kafka クラスターで適切に設定されるようにします。

たとえば、KafkaUser とユーザーの関係は次のようになります。

  • KafkaUser が作成されると、User Operator によって記述されるユーザーが作成されます。
  • KafkaUser が削除されると、User Operator によって記述されるユーザーが削除されます。
  • KafkaUser が変更されると、User Operator によって記述されるユーザーが更新されます。

User Operator は Topic Operator とは異なり、Kafka クラスターからの変更は OpenShift リソースと同期されません。アプリケーションで直接 Kafka トピックを Kafka で作成することは可能ですが、ユーザーが User Operator と同時に直接 Kafka クラスターで管理されることは想定されません。

User Operator では、アプリケーションのデプロイメントの一部として KafkaUser リソースを宣言できます。ユーザーの認証および承認メカニズムを指定できます。たとえば、ユーザーがブローカーへのアクセスを独占しないようにするため、Kafka リソースの使用を制御する ユーザークォータ を設定することもできます。

ユーザーが作成されると、ユーザークレデンシャルが Secret に作成されます。アプリケーションはユーザーとそのクレデンシャルを使用して、認証やメッセージの生成または消費を行う必要があります。

User Operator は 認証のクレデンシャルを管理する他に、KafkaUser 宣言にユーザーのアクセス権限の記述を含めることで承認も管理します。

4.3.2. ユーザー処理用の Kafka クラスターの特定

KafkaUser リソースには、このリソースが属する Kafka クラスターに適した名前 (Kafka リソースの名前から派生) を定義するラベルが含まれています。

apiVersion: kafka.strimzi.io/v1beta1
kind: KafkaUser
metadata:
  name: my-user
  labels:
    strimzi.io/cluster: my-cluster

このラベルは、KafkaUser リソースを特定し、新しいユーザーを作成するために、User Operator によって使用されます。また、以降のユーザーの処理でも使用されます。

ラベルが Kafka クラスターと一致しない場合、User Operator は kafkaUser を識別できず、ユーザーは作成されません。

4.3.3. リソース要求および制限のある User Operator の設定

CPU やメモリーなどのリソースを User Operator に割り当て、User Operator が消費できるリソースの量に制限を設定できます。

前提条件

  • Cluster Operator が稼働している必要があります。

手順

  1. 必要に応じてエディターで Kafka クラスター設定を更新します。

    oc edit kafka my-cluster
  2. Kafka リソースの spec.entityOperator.userOperator.resources プロパティーで、User Operator のリソース要求および制限を設定します。

    apiVersion: kafka.strimzi.io/v1beta1
    kind: Kafka
    spec:
      # kafka and zookeeper sections...
      entityOperator:
        userOperator:
          resources:
            request:
              cpu: "1"
              memory: 500Mi
            limit:
              cpu: "1"
              memory: 500Mi

    ファイルを保存し、エディターを終了します。Cluster Operator によって変更が自動的に適用されます。

その他のリソース

  • resources オブジェクトのスキーマの詳細は「<link xlink:href="https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.18/#resourcerequirements-v1-core">ResourceRequirements v1 core</link>」を参照してください。

4.4. Operator の監視

4.4.1. Prometheus メトリクス

AMQ Streams の operator は Prometheus メトリクスを公開します。メトリクスは自動で有効になり、以下の情報が含まれます。

  • 調整の数
  • operator が処理しているカスタムリソースの数
  • 調整の期間
  • operator からの JVM メトリクス

この他に、Grafana ダッシュボードのサンプルが提供されます。

Prometheus の詳細は、「Kafka へのメトリクスの導入」を参照してください。