Esegui il deployment di Apache Kafka su GKE utilizzando Confluent

prepara l'ambiente

In questo tutorial utilizzi Cloud Shell per gestire le risorse ospitate su Google Cloud. Il software è preinstallato in Cloud Shell che ti servono per questo tutorial, kubectl, il gcloud CLI, Helm, e Terraform.

Per configurare l'ambiente con Cloud Shell, segui questi passaggi:

1. Avvia una sessione di Cloud Shell dalla console Google Cloud. clic su Attiva Cloud Shell nella Console Google Cloud: Verrà avviata una sessione nel riquadro inferiore della console Google Cloud.

2. Imposta le variabili di ambiente:

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

   Sostituisci PROJECT_ID: il tuo Google Cloud con ID progetto.

3. Clona il repository GitHub:

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

4. Passa alla directory di lavoro:

   cd kubernetes-engine-samples/streaming
   

Crea l'infrastruttura del tuo cluster

In questa sezione eseguirai uno script Terraform per creare un'istanza un cluster GKE a livello di regione e alta disponibilità. I passaggi seguenti consentono l'accesso pubblico al piano di controllo. Per limitare l'accesso, crea un'istanza cluster privato.

Puoi installare l'operatore utilizzando un'istanza Standard o Autopilot in un cluster Kubernetes.

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}

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

Outputs:

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

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}

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

Outputs:

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

Connettiti al cluster

Configura kubectl per comunicare con il cluster:

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

Esegui il deployment dell'operatore CFK nel cluster

In questa sezione, eseguirai il deployment dell'operatore Confluent for Kubernetes (CFK) utilizzando una grafico Helm ed eseguire il deployment di un cluster Kafka.

1. Aggiungi il repository Confluent Helm Chart:

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

2. Aggiungi uno spazio dei nomi per l'operatore CFK e il cluster Kafka:

   kubectl create ns kafka
   

3. Esegui il deployment dell'operatore del cluster CFK utilizzando Helm:

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

   Per consentire a CFK di gestire le risorse in tutti gli spazi dei nomi, aggiungi il parametro --set-namespaced=false al comando Helm.

4. Verifica che il deployment dell'operatore Confluent sia stato eseguito correttamente usando Helm:

   helm ls -n kafka
   

   L'output è simile al seguente:

   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
   

Esegui il deployment di Kafka

In questa sezione eseguirai il deployment di Kafka in una configurazione di base e quindi proverai vari di configurazione avanzata per risolvere problemi di disponibilità, requisiti di osservabilità.

Configurazione di base

La configurazione di base per l'istanza Kafka include quanto segue componenti:

• Tre repliche di broker Kafka, con un minimo di due repliche disponibili necessaria per la coerenza del cluster.
• Tre repliche di nodi ZooKeeper, che formano un cluster.
• Due listener Kafka: uno senza autenticazione e l'altro che utilizza TLS l'autenticazione con un certificato generato da CFK.
• Java MaxHeapSize e MinHeapSize impostati su 4 GB per Kafka.
• l'allocazione delle risorse di CPU di 1 richiesta di CPU e di 2 limiti di CPU; e Richieste di memoria da 5 GB e limiti per Kafka (4 GB per il servizio principale e 0,5 GB per l'esportazione delle metriche) e 3 GB per Zookeeper (2 GB per il servizio principale e 0,5 GB per l'esportatore di metriche).
• 100 GB di spazio di archiviazione allocati a ogni pod mediante premium-rwo storageClass, 100 per i dati Kafka e 90/10 per Zookeeper Data/Log.
• Tolleranze, nodeAffinità e podAntiAffinità configurati per per ogni carico di lavoro, assicurando una corretta distribuzione tra i nodi, rispettivi pool di nodi e zone diverse.
• Comunicazione all'interno del cluster protetta da certificati autofirmati mediante un Autorità di certificazione da te fornita.

Questa configurazione rappresenta la configurazione minima richiesta per creare in un cluster Kafka pronto per la produzione. Le seguenti sezioni illustrano la configurazioni per affrontare aspetti quali la sicurezza del cluster, gli elenchi di controllo dell'accesso (ACL), gestione di argomenti, gestione dei certificati e altro ancora.

Crea un cluster Kafka di base

1. Genera una coppia di CA:

   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 fornisce certificati generati automaticamente per Confluent Componenti della piattaforma da utilizzare per la crittografia della rete TLS. Devi generare e gestire fornire un'autorità di certificazione (CA).

2. Crea un secret Kubernetes per l'autorità di certificazione:

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

   Il nome del secret è valore predefinito

3. Crea un nuovo cluster Kafka utilizzando la configurazione di base:

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

   Questo comando crea una risorsa personalizzata Kafka e una risorsa personalizzata Zookeeper dell'operatore CFK che includono richieste e limiti di CPU e memoria, richieste di archiviazione, incompatibilità e affinità per distribuire tra i vari nodi Kubernetes.

4. Attendi alcuni minuti mentre Kubernetes avvia i carichi di lavoro richiesti:

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

5. Verifica che i carichi di lavoro Kafka siano stati creati:

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

   L'output è simile al seguente:

   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'operatore crea le seguenti risorse:

• Due StatefulSet per Kafka e ZooKeeper.
• Tre pod per le repliche dei broker Kafka.
• Tre pod per le repliche di ZooKeeper.
• Due risorse PodDisruptionBudget, garantendone al massimo una non disponibile per garantire la coerenza del cluster.
• Il servizio my-cluster, che funge da server di bootstrap per Kafka che si connettono dall'interno del cluster Kubernetes. Tutto Kafka interno in questo servizio sono disponibili listener.
• Il servizio zookeeper che consente ai broker Kafka di connettersi a Nodi ZooKeeper come client.

Autenticazione e gestione degli utenti

Questa sezione mostra come attivare l'autenticazione e l'autorizzazione per proteggere i listener Kafka e condividere le credenziali con i client.

Confluent per Kubernetes supporta vari metodi di autenticazione per Kafka, come come:

• Autenticazione SASL/PLAIN: I client utilizzano un nome utente e una password per l'autenticazione. Il nome utente e sono archiviate lato server in un secret Kubernetes.
• SASL/PLAIN con autenticazione LDAP: I client utilizzano un nome utente e una password per l'autenticazione. Le credenziali sono archiviati in un server LDAP.
• Autenticazione mTLS: I client utilizzano i certificati TLS per l'autenticazione.

Limitazioni

• CFK non fornisce risorse personalizzate per la gestione utenti. Tuttavia, puoi archiviare le credenziali nei secret e fare riferimento ai secret nelle specifiche del listener.
• Anche se non esiste una risorsa personalizzata per gestire direttamente gli ACL, Confluent per Kubernetes fornisce indicazioni sulla configurazione degli ACL utilizzando l'interfaccia a riga di comando Kafka.

Crea un utente

Questa sezione mostra come eseguire il deployment di un operatore CFK che dimostri di Google Cloud, tra cui:

• Un cluster Kafka con autenticazione basata su password (SASL/PLAIN) abilitata su uno degli ascoltatori
• A KafkaTopiccon 3 repliche
• Credenziali utente con autorizzazioni di lettura e scrittura

1. Crea un secret con credenziali utente:

   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\"}"
   

   Le credenziali devono essere archiviate nel seguente formato:

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

2. Configura il cluster Kafka per l'uso di un listener basato su password autenticazione Autenticazione SCRAM-SHA-512 sulla porta 9094:

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

3. configura un argomento e un pod del client per interagire con il tuo cluster Kafka e esegui i comandi 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 monta il secret my-user-credentials nel pod del client come Volume.

4. Quando il pod del client è pronto, connettiti e inizia a produrre che utilizzano i messaggi utilizzando le credenziali fornite:

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

5. Genera un messaggio utilizzando le credenziali my-user, quindi utilizza il valore per verificarne la ricezione.

   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
   

   L'output è simile al seguente:

   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
   

   Digita CTRL+C per interrompere il processo del consumatore. Se ricevi un Connect refused , attendi qualche minuto e riprova.

6. Esci dalla shell del pod

   exit
   

Backup e ripristino di emergenza

Utilizzando l'operatore Confluent puoi implementare strategie di backup efficienti seguendo determinati schemi.

Puoi utilizzare la modalità Backup per GKE per eseguire il backup:

• e i file manifest delle risorse Kubernetes.
• Risorse personalizzate delle API confluenti e relative definizioni estratte dal Server API Kubernetes del cluster in fase di backup.
• I volumi che corrispondono alle risorse PersistentVolumeClaim presenti e i file manifest.

Per ulteriori informazioni su come eseguire il backup e ripristinare i cluster Kafka utilizzando Backup per GKE, consulta Preparati per il ripristino di emergenza.

Puoi anche eseguire un backup manuale del cluster Kafka. Devi eseguire il backup di:

• La configurazione Kafka, che include tutte le risorse personalizzate API Confluent come KafkaTopicsoConnect
• I dati, che sono archiviati negli oggetti PersistentVolume dei broker Kafka

L'archiviazione dei manifest delle risorse Kubernetes, incluse le configurazioni Confluent, I repository Git possono eliminare la necessità di un backup separato per Kafka perché le risorse possono essere riapplicate a un nuovo cluster Kubernetes necessaria.

Per salvaguardare il recupero dei dati Kafka in scenari in cui un'istanza del server Kafka, oppure Il cluster Kubernetes in cui viene eseguito il deployment di Kafka è andato perso. Ti consigliamo di configura la classe di archiviazione Kubernetes utilizzata per il provisioning dei volumi per Kafka broker con l'opzione reclaimPolicy impostata su Retain. Abbiamo consigliato anche che prendi snapshot dei volumi dei broker Kafka.

Il manifest seguente descrive un oggetto StorageClass che utilizza l'oggetto reclaimPolicy opzione Retain:

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

L'esempio seguente mostra l'oggetto StorageClass aggiunto all'elemento spec di un Kafka risorsa personalizzata del cluster:

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

Con questa configurazione, il provisioning degli oggetti PersistentVolume viene eseguito usando la classe di archiviazione non vengono eliminati anche quando viene eliminato il PersistentVolumeClaim corrispondente.

Per recuperare l'istanza Kafka su un nuovo cluster Kubernetes utilizzando configurazione e dati dell'istanza dell'intermediario:

1. Applica le risorse personalizzate Confluent esistenti (Kafka, KafkaTopic, Zookeeper e così via) in un nuovo cluster Kubernetes
2. Aggiorna gli oggetti PersistentVolumeClaim con il nome del nuovo broker Kafka ai vecchi PersistentVolume usando la proprietà spec.volumeName su PersistentVolumeClaim.