Déployer Apache Kafka sur GKE à l'aide de Confluent

Ce guide vous explique comment utiliser l'opérateur Confluent for Kubernetes (CFK) pour déployer des clusters Apache Kafka sur Google Kubernetes Engine (GKE).

Kafka est un système de messagerie Pub/Sub distribué Open Source permettant de gérer des données en streaming volumineuses, à haut débit et en temps réel. Kafka vous permet de créer des pipelines de données en streaming qui déplacent des données de manière fiable entre différents systèmes et applications à des fins de traitement et d'analyse.

Ce guide est destiné aux administrateurs de plate-forme, aux architectes cloud et aux professionnels des opérations qui souhaitent déployer des clusters Kafka sur GKE.

Vous pouvez également utiliser l'opérateur CFK pour déployer d'autres composants de Confluent Platform, tels que le centre Web Confluent Control, Schema Registry ou KsqlDB. Toutefois, ce guide ne se concentre que sur les déploiements de Kafka.

Remarque : L'opérateur CFK et la plupart des composants de la plateforme Confluent sont distribués sous une licence Confluent Enterprise, et les restrictions de licence exigent que les utilisateurs aient un abonnement actif après 30 jours d'essai pour continuer à utiliser l'opérateur. La licence Developer est également disponible pendant une durée illimitée. Toutefois, elle est limitée à une seule configuration d'agent par cluster.

Préparer l'environnement

Dans ce tutoriel, vous utilisez Cloud Shell pour gérer les ressources hébergées sur Google Cloud. Les logiciels dont vous avez besoin pour ce tutoriel sont préinstallés sur Cloud Shell, y compris kubectl, gcloud CLI, Helm et Terraform.

Pour configurer votre environnement avec Cloud Shell, procédez comme suit :

1. Lancez une session Cloud Shell depuis la console Google Cloud en cliquant sur Activer Cloud Shell dans la consoleGoogle Cloud . Une session s'ouvre dans le volet inférieur de la console Google Cloud .

2. Définissez les variables d'environnement :

   export PROJECT_ID=PROJECT_ID
   export KUBERNETES_CLUSTER_PREFIX=kafka
   export REGION=us-central1
   

   Remplacez PROJECT_ID: your Google Cloud par votre ID de projet.

3. Clonez le dépôt GitHub.

   git clone https://github.com/GoogleCloudPlatform/kubernetes-engine-samples
   

4. Accédez au répertoire de travail :

   cd kubernetes-engine-samples/streaming
   

Créer l'infrastructure de votre cluster

Dans cette section, vous allez exécuter un script Terraform pour créer un cluster GKE régional, privé et à disponibilité élevée. Les étapes suivantes autorisent l'accès public au plan de contrôle. Pour restreindre l'accès, créez un cluster privé.

Vous pouvez installer l'opérateur à l'aide d'un cluster standard ou Autopilot.

Standard

Le schéma suivant présente un cluster GKE standard régional privé déployé sur trois zones différentes :

Pour déployer cette infrastructure, exécutez les commandes suivantes à partir de Cloud Shell :

export GOOGLE_OAUTH_ACCESS_TOKEN=$(gcloud auth print-access-token)
terraform -chdir=kafka/terraform/gke-standard init
terraform -chdir=kafka/terraform/gke-standard apply -var project_id=${PROJECT_ID} \
  -var region=${REGION} \
  -var cluster_prefix=${KUBERNETES_CLUSTER_PREFIX}

Lorsque vous y êtes invité, saisissez yes. L'exécution de cette commande et le passage du cluster à l'état prêt peuvent prendre plusieurs minutes.

Terraform crée les ressources suivantes :

• Un réseau VPC et un sous-réseau privé pour les nœuds Kubernetes.
• Un routeur pour accéder à Internet via NAT.
• Un cluster GKE privé dans la région us-central1.
• Deux pools de nœuds avec l'autoscaling activé (1 à 2 nœuds par zone, 1 nœud par zone au minimum).
• Un ServiceAccount avec les autorisations de journalisation et de surveillance.
• Sauvegarde pour GKE pour la reprise après sinistre.
• Google Cloud Managed Service pour Prometheus pour la surveillance du cluster.

Le résultat ressemble à ce qui suit :

...
Apply complete! Resources: 14 added, 0 changed, 0 destroyed.

Outputs:

kubectl_connection_command = "gcloud container clusters get-credentials kafka-cluster --region us-central1"

Autopilot

Le schéma suivant présente un cluster GKE Autopilot régional privé :

Pour déployer l'infrastructure, exécutez les commandes suivantes à partir de Cloud Shell :

export GOOGLE_OAUTH_ACCESS_TOKEN=$(gcloud auth print-access-token)
terraform -chdir=kafka/terraform/gke-autopilot init
terraform -chdir=kafka/terraform/gke-autopilot apply -var project_id=${PROJECT_ID} \
  -var region=${REGION} \
  -var cluster_prefix=${KUBERNETES_CLUSTER_PREFIX}

Lorsque vous y êtes invité, saisissez yes. L'exécution de cette commande et le passage du cluster à l'état prêt peuvent prendre plusieurs minutes.

Terraform crée les ressources suivantes :

• Un réseau VPC et un sous-réseau privé pour les nœuds Kubernetes.
• Un routeur pour accéder à Internet via NAT.
• Un cluster GKE privé dans la région us-central1.
• Un ServiceAccount avec les autorisations de journalisation et de surveillance.
• Google Cloud Managed Service pour Prometheus pour la surveillance du cluster.

Le résultat ressemble à ce qui suit :

...
Apply complete! Resources: 12 added, 0 changed, 0 destroyed.

Outputs:

kubectl_connection_command = "gcloud container clusters get-credentials kafka-cluster --region us-central1"

Se connecter au cluster

Configurez kubectl pour communiquer avec le cluster :

gcloud container clusters get-credentials ${KUBERNETES_CLUSTER_PREFIX}-cluster --region ${REGION}

Déployer l'opérateur CFK sur votre cluster

Dans cette section, vous allez déployer l'opérateur Confluent for Kubernetes (CFK) à l'aide d'un chart Helm, puis déployer un cluster Kafka.

1. Ajoutez le dépôt de chart Helm Confluent :

   helm repo add confluentinc https://packages.confluent.io/helm
   

2. Ajoutez un espace de noms pour l'opérateur CFK et le cluster Kafka :

   kubectl create ns kafka
   

3. Déployez l'opérateur de cluster CFK à l'aide de Helm :

   helm install confluent-operator confluentinc/confluent-for-kubernetes -n kafka
   

   Pour permettre à CFK de gérer les ressources sur tous les espaces de noms, ajoutez le paramètre --set-namespaced=false à la commande Helm.

4. Vérifiez que l'opérateur Confluent a bien été déployé à l'aide de Helm :

   helm ls -n kafka
   

   Le résultat ressemble à ce qui suit :

   NAME                  NAMESPACE  REVISION UPDATED                                  STATUS      CHART                                APP VERSION
   confluent-operator    kafka      1        2023-07-07 10:57:45.409158 +0200 CEST    deployed    confluent-for-kubernetes-0.771.13    2.6.0
   

Déployer Kafka

Dans cette section, vous allez déployer Kafka dans une configuration de base puis tester divers scénarios de configuration avancés pour répondre aux exigences de disponibilité, de sécurité et d'observabilité.

Configuration de base

La configuration de base de l'instance Kafka comprend les composants suivants :

• Trois instances dupliquées des agents Kafka, avec au moins deux instances dupliquées disponibles pour la cohérence de cluster.
• Trois instances dupliquées des nœuds ZooKeeper, formant un cluster.
• Deux écouteurs Kafka : l'un sans authentification et l'autre utilisant l'authentification TLS avec un certificat généré par CFK.
• Java MaxHeapSize et MinHeapSize définis sur 4 Go pour Kafka
• Allocation de ressources de processeur de 1 requête de processeur et 2 limites de processeur, 5 Go de requêtes et limites de mémoire pour Kafka (4 Go pour le service principal et 0,5 Go pour l'exportateur de métriques) et 3 Go pour Zookeeper (2 Go pour le service principal et 0,5 Go pour l'exportateur de métriques).
• 100 Go d'espace de stockage alloué à chaque pod à l'aide de la StorageClass premium-rwo, 100 pour les données Kafka et 90/10 pour les données et le journal Zookeeper.
• Les tolérances, nodeAffinities et podAntiAffinities configurées pour chaque charge de travail afin de garantir une répartition appropriée des nœuds, en utilisant leurs pools de nœuds respectifs et des zones différentes.
• Communication au sein du cluster sécurisée par des certificats autosignés à l'aide d'une autorité de certification que vous fournissez.

Cette configuration représente la configuration minimale requise pour créer un cluster Kafka prêt pour la production. Les sections suivantes présentent des configurations personnalisées pour des aspects tels que la sécurité des clusters, les listes de contrôle d'accès (LCA), la gestion des sujets, la gestion des certificats, etc.

Créer un cluster Kafka de base

1. Générez une paire d'autorités de certification :

   openssl genrsa -out ca-key.pem 2048
   openssl req -new -key ca-key.pem -x509 \
     -days 1000 \
     -out ca.pem \
     -subj "/C=US/ST=CA/L=Confluent/O=Confluent/OU=Operator/CN=MyCA"
   

   Confluent for Kubernetes fournit des certificats générés automatiquement pour les composants de la plateforme Confluent à utiliser pour le chiffrement du réseau TLS. Vous devez générer et fournir une autorité de certification (CA).

2. Créez un secret Kubernetes pour l'autorité de certification :

   kubectl create secret tls ca-pair-sslcerts --cert=ca.pem --key=ca-key.pem -n kafka
   

   Le nom du secret est prédéfini.

3. Créez un cluster Kafka en utilisant la configuration de base :

   kubectl apply -n kafka -f kafka-confluent/manifests/01-basic-cluster/my-cluster.yaml
   

   Cette commande crée une ressource personnalisée Kafka et une ressource personnalisée ZooKeeper de l'opérateur CFK qui inclut les demandes et limites de processeurs et de mémoire, les requêtes de stockage de blocs, ainsi que les rejets et les affinités pour répartir les pods provisionnés sur les nœuds Kubernetes.

4. Attendez quelques minutes pendant que Kubernetes démarre les charges de travail requises :

   kubectl wait pods -l app=my-cluster --for condition=Ready --timeout=300s -n kafka
   

5. Vérifiez que les charges de travail Kafka ont bien été créées :

   kubectl get pod,svc,statefulset,deploy,pdb -n kafka
   

   Le résultat ressemble à ce qui suit :

   NAME                                    READY   STATUS  RESTARTS   AGE
   pod/confluent-operator-864c74d4b4-fvpxs   1/1   Running   0        49m
   pod/my-cluster-0                        1/1   Running   0        17m
   pod/my-cluster-1                        1/1   Running   0        17m
   pod/my-cluster-2                        1/1   Running   0        17m
   pod/zookeeper-0                         1/1   Running   0        18m
   pod/zookeeper-1                         1/1   Running   0        18m
   pod/zookeeper-2                         1/1   Running   0        18m
   
   NAME                          TYPE      CLUSTER-IP   EXTERNAL-IP   PORT(S)                                                        AGE
   service/confluent-operator    ClusterIP   10.52.13.164   <none>      7778/TCP                                                       49m
   service/my-cluster            ClusterIP   None         <none>      9092/TCP,8090/TCP,9071/TCP,7203/TCP,7777/TCP,7778/TCP,9072/TCP   17m
   service/my-cluster-0-internal   ClusterIP   10.52.2.242  <none>      9092/TCP,8090/TCP,9071/TCP,7203/TCP,7777/TCP,7778/TCP,9072/TCP   17m
   service/my-cluster-1-internal   ClusterIP   10.52.7.98   <none>      9092/TCP,8090/TCP,9071/TCP,7203/TCP,7777/TCP,7778/TCP,9072/TCP   17m
   service/my-cluster-2-internal   ClusterIP   10.52.4.226  <none>      9092/TCP,8090/TCP,9071/TCP,7203/TCP,7777/TCP,7778/TCP,9072/TCP   17m
   service/zookeeper             ClusterIP   None         <none>      2181/TCP,7203/TCP,7777/TCP,3888/TCP,2888/TCP,7778/TCP          18m
   service/zookeeper-0-internal  ClusterIP   10.52.8.52   <none>      2181/TCP,7203/TCP,7777/TCP,3888/TCP,2888/TCP,7778/TCP          18m
   service/zookeeper-1-internal  ClusterIP   10.52.12.44  <none>      2181/TCP,7203/TCP,7777/TCP,3888/TCP,2888/TCP,7778/TCP          18m
   service/zookeeper-2-internal  ClusterIP   10.52.12.134   <none>      2181/TCP,7203/TCP,7777/TCP,3888/TCP,2888/TCP,7778/TCP          18m
   
   NAME                        READY   AGE
   statefulset.apps/my-cluster   3/3   17m
   statefulset.apps/zookeeper  3/3   18m
   
   NAME                               READY   UP-TO-DATE   AVAILABLE   AGE
   deployment.apps/confluent-operator   1/1   1          1         49m
   
   NAME                                  MIN AVAILABLE   MAX UNAVAILABLE   ALLOWED DISRUPTIONS   AGE
   poddisruptionbudget.policy/my-cluster   N/A           1               1                   17m
   poddisruptionbudget.policy/zookeeper  N/A           1               1                   18m
   

L'opérateur crée les ressources suivantes :

• Deux StatefulSets pour Kafka et ZooKeeper.
• Trois pods pour les instances dupliquées d'agents Kafka.
• Trois pods pour les instances dupliquées ZooKeeper.
• Deux ressources PodDisruptionBudget, garantissant ainsi une instance répliquée indisponible au maximum pour la cohérence du cluster.
• Le service my-cluster, qui sert de serveur d'amorçage pour les clients Kafka se connectant à partir du cluster Kubernetes. Tous les écouteurs Kafka internes sont disponibles dans ce service.
• Le service zookeeper, qui permet aux agents Kafka de se connecter aux nœuds ZooKeeper en tant que clients.

Authentification et gestion des utilisateurs

Cette section explique comment activer l'authentification et l'autorisation pour sécuriser les écouteurs Kafka et partager des identifiants avec les clients.

Confluent fo Kubernetes est compatible avec diverses méthodes d'authentification pour Kafka, dont voici quelques exemples :

• Authentification SASL/PLAIN : les clients utilisent un nom d'utilisateur et un mot de passe pour l'authentification. Le nom d'utilisateur et le mot de passe sont stockés côté serveur dans un secret Kubernetes.
• Authentification SASL/PLAIN avec LDAP : les clients utilisent un nom d'utilisateur et un mot de passe pour l'authentification. Les identifiants sont stockés sur un serveur LDAP.
• Authentification mTLS : les clients utilisent des certificats TLS pour l'authentification.

Limites

• CFK ne fournit pas de ressources personnalisées pour la gestion des utilisateurs. Toutefois, vous pouvez stocker les identifiants dans des secrets et y faire référence dans les spécifications de l'écouteur.
• Bien qu'il n'existe pas de ressource personnalisée pour gérer directement les LCA, le service Confluent for Kubernetes officiel fournit des conseils sur la configuration des LCA à l'aide de la CLI Kafka.

Créer un compte utilisateur

Cette section vous explique comment déployer un opérateur CFK qui illustre les fonctionnalités de gestion des utilisateurs, y compris :

• Un cluster Kafka avec authentification basée sur les mots de passe (SASL/PLAIN) activé sur l'un des écouteurs
• Un KafkaTopic avec trois instances répliquées.
• Identifiants utilisateur avec autorisations de lecture et d'écriture

1. Créez un secret avec des identifiants utilisateur :

   export USERNAME=my-user
   export PASSWORD=$(openssl rand -base64 12)
   kubectl create secret generic my-user-credentials -n kafka \
     --from-literal=plain-users.json="{\"$USERNAME\":\"$PASSWORD\"}"
   

   Les identifiants doivent être stockés au format suivant :

   {
   "username1": "password1",
   "username2": "password2",
   ...
   "usernameN": "passwordN"
   }
   

2. Configurez le cluster Kafka pour utiliser un écouteur avec authentification par mot de passe SCRAM-SHA-512 sur le port 9094 :

   kubectl apply -n kafka -f kafka-confluent/manifests/02-auth/my-cluster.yaml
   

3. Configurez un sujet et un pod client pour interagir avec votre cluster Kafka et exécuter les commandes Kafka :

   kubectl apply -n kafka -f kafka-confluent/manifests/02-auth/my-topic.yaml
   kubectl apply -n kafka -f kafka-confluent/manifests/02-auth/kafkacat.yaml
   

   GKE installe le secret my-user-credentials sur le pod client en tant que volume.

4. Lorsque le pod client est prêt, connectez-vous au pod et commencez à produire et à utiliser des messages à l'aide des identifiants fournis :

   kubectl wait pod kafkacat --for=condition=Ready --timeout=300s -n kafka
   kubectl exec -it kafkacat -n kafka -- /bin/sh
   

5. Générez un message à l'aide des identifiants my-user, puis utilisez-le pour vérifier sa réception.

   export USERNAME=$(cat /my-user/plain-users.json|cut -d'"' -f 2)
   export PASSWORD=$(cat /my-user/plain-users.json|cut -d'"' -f 4)
   echo "Message from my-user" |kcat \
     -b my-cluster.kafka.svc.cluster.local:9094 \
     -X security.protocol=SASL_SSL \
     -X sasl.mechanisms=PLAIN \
     -X sasl.username=$USERNAME \
     -X sasl.password=$PASSWORD  \
     -t my-topic -P
   kcat -b my-cluster.kafka.svc.cluster.local:9094 \
     -X security.protocol=SASL_SSL \
     -X sasl.mechanisms=PLAIN \
     -X sasl.username=$USERNAME \
     -X sasl.password=$PASSWORD  \
     -t my-topic -C
   

   Le résultat ressemble à ce qui suit :

   Message from my-user
   % Reached end of topic my-topic [1] at offset 1
   % Reached end of topic my-topic [2] at offset 0
   % Reached end of topic my-topic [0] at offset 0
   

   Saisissez CTRL+C pour arrêter le processus destiné au consommateur. Si vous obtenez une erreur Connect refused, attendez quelques minutes, puis réessayez.

6. Quittez l'interface système du pod.

   exit
   

Sauvegardes et reprise après sinistre

L'opérateur Confluent vous permet de mettre en œuvre des stratégies de sauvegarde efficaces en suivant certains modèles.

Vous pouvez utiliser Sauvegarde pour GKE pour effectuer une sauvegarde des éléments suivants :

• Fichiers manifestes de ressources Kubernetes.
• Ressources personnalisées de l'API Confluent et leurs définitions extraites du serveur d'API Kubernetes du cluster à sauvegarder.
• Volumes correspondant aux ressources PersistentVolumeClaim trouvées dans les fichiers manifestes.

Pour en savoir plus sur la sauvegarde et la restauration des clusters Kafka à l'aide de Sauvegarde pour GKE, consultez la page Préparer la reprise après sinistre.

Vous pouvez également effectuer une sauvegarde manuelle de votre cluster Kafka. Vous devez sauvegarder :

• La configuration Kafka, qui inclut toutes les ressources personnalisées de l'API Confluent, telles que KafkaTopics ou Connect
• Les données, qui sont stockées dans les PersistentVolumes des agents Kafka

Le stockage des fichiers manifestes de ressources Kubernetes (y compris des configurations Confluent) dans des dépôts Git peut éliminer la nécessité d'une sauvegarde distincte pour la configuration Kafka, car les ressources peuvent être réappliquées à un nouveau cluster Kubernetes si nécessaire.

Pour protéger la récupération des données Kafka dans les scénarios où une instance de serveur Kafka ou un cluster Kubernetes où Kafka est déployé, nous vous recommandons de configurer la classe de stockage Kubernetes utilisée pour provisionner des volumes pour les agents Kafka avec l'option reclaimPolicy définie sur Retain. Nous vous recommandons également de prendre des instantanés des volumes de l'agent Kafka.

Le fichier manifeste suivant décrit une StorageClass qui utilise l'option de reclaimPolicy Retain :

apiVersion: storage.k8s.io/v1
kind: StorageClass
metadata:
  name: premium-rwo-retain
...
reclaimPolicy: Retain
volumeBindingMode: WaitForFirstConsumer

L'exemple suivant montre la ressource StorageClass ajoutée au spec d'une ressource personnalisée de cluster Kafka :

...
spec:
  ...
  dataVolumeCapacity: 100Gi
  storageClass:
  name: premium-rwo-retain

Avec cette configuration, les ressources PersistentVolume provisionnées à l'aide de la classe de stockage ne sont pas supprimées, même lorsque l'objet PersistentVolumeClaim correspondant est supprimé.

Pour récupérer l'instance Kafka sur un nouveau cluster Kubernetes à l'aide des données de configuration et d'instance d'agent existantes, procédez comme suit :

1. Appliquez les ressources personnalisées Confluent existantes (Kafka, KafkaTopic, Zookeeper, etc.) à un nouveau cluster Kubernetes.
2. Mettez à jour les PersistentVolumeClaims avec le nom des nouvelles instances d'agent Kafka pour utiliser les anciens PersistentVolumes en utilisant la propriété spec.volumeName de la PersistentVolumeClaim.