Desplegar Apache Kafka en GKE con Confluent

En esta guía se muestra cómo usar el operador Confluent for Kubernetes (CFK) para desplegar clústeres de Apache Kafka en Google Kubernetes Engine (GKE).

Kafka es un sistema de mensajería de publicación y suscripción distribuido y de código abierto para gestionar datos de streaming de alto volumen, alto rendimiento y en tiempo real. Puedes usar Kafka para crear flujos de procesamiento de datos de streaming que transfieran datos de forma fiable entre diferentes sistemas y aplicaciones para procesarlos y analizarlos.

Esta guía está dirigida a administradores de plataformas, arquitectos de la nube y profesionales de operaciones que quieran desplegar clústeres de Kafka en GKE.

También puedes usar el operador CFK para desplegar otros componentes de Confluent Platform, como Confluent Control Center (basado en la Web), Schema Registry o KsqlDB. Sin embargo, esta guía se centra únicamente en las implementaciones de Kafka.

Nota: El operador de CFK y la mayoría de los componentes de la plataforma de Confluent se distribuyen con una licencia empresarial de Confluent. Las restricciones de la licencia requieren que los usuarios tengan una suscripción activa después de la prueba de 30 días para seguir usando el operador. La licencia de desarrollador también está disponible por un periodo indefinido, pero se limita a una sola configuración de broker por clúster.

Preparar el entorno

En este tutorial, usarás Cloud Shell para gestionar los recursos alojados en Google Cloud. Cloud Shell tiene preinstalado el software que necesitas para este tutorial, como kubectl, la CLI de gcloud, Helm y Terraform.

Para configurar tu entorno con Cloud Shell, sigue estos pasos:

1. Inicia una sesión de Cloud Shell desde la Google Cloud consola haciendo clic en Activar Cloud Shell en la Google Cloud consola. Se iniciará una sesión en el panel inferior de la consola Google Cloud .

2. Define las variables de entorno:

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

   Sustituye PROJECT_ID: your Google Cloud por tu ID de proyecto.

3. Clona el repositorio de GitHub:

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

4. Cambia al directorio de trabajo:

   cd kubernetes-engine-samples/streaming
   

Crear la infraestructura del clúster

En esta sección, ejecutarás una secuencia de comandos de Terraform para crear un clúster de GKE privado, regional y de alta disponibilidad. Los pasos siguientes permiten el acceso público al plano de control. Para restringir el acceso, crea un clúster privado.

Puedes instalar el operador con un clúster estándar o Autopilot.

Estándar

En el siguiente diagrama se muestra un clúster de GKE estándar regional privado desplegado en tres zonas diferentes:

Para desplegar esta infraestructura, ejecuta los siguientes comandos desde 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}

Cuando se te solicite, escribe yes. Este comando puede tardar varios minutos en completarse y el clúster en mostrar el estado "Listo".

Terraform crea los siguientes recursos:

• Una red de VPC y una subred privada para los nodos de Kubernetes.
• Un router para acceder a Internet a través de NAT.
• Un clúster de GKE privado en la región us-central1.
• 2 grupos de nodos con el autoescalado habilitado (de 1 a 2 nodos por zona y 1 nodo por zona como mínimo)
• Un ServiceAccount con permisos de registro y monitorización.
• Copia de seguridad de GKE para la recuperación tras fallos.
• Google Cloud Managed Service para Prometheus para la monitorización de clústeres.

El resultado debería ser similar al siguiente:

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

Outputs:

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

Autopilot

En el siguiente diagrama se muestra un clúster de Autopilot de GKE regional privado:

Para desplegar la infraestructura, ejecuta los siguientes comandos desde 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}

Cuando se te solicite, escribe yes. Este comando puede tardar varios minutos en completarse y el clúster en mostrar el estado "Listo".

Terraform crea los siguientes recursos:

• Red de VPC y subred privada para los nodos de Kubernetes.
• Un router para acceder a Internet a través de NAT.
• Un clúster de GKE privado en la región us-central1.
• Un ServiceAccount con permisos de registro y monitorización
• Google Cloud Managed Service para Prometheus para la monitorización de clústeres.

El resultado debería ser similar al siguiente:

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

Outputs:

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

Conéctate al clúster

Configura kubectl para que se comunique con el clúster:

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

Desplegar el operador de CFK en el clúster

En esta sección, desplegarás el operador de Confluent para Kubernetes (CFK) mediante un gráfico de Helm y, a continuación, desplegarás un clúster de Kafka.

1. Añade el repositorio de gráficos de Helm de Confluent:

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

2. Añade un espacio de nombres para el operador de CFK y el clúster de Kafka:

   kubectl create ns kafka
   

3. Despliega el operador de clúster de CFK con Helm:

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

   Para permitir que CFK gestione recursos en todos los espacios de nombres, añade el parámetro --set-namespaced=false al comando de Helm.

4. Verifica que el operador de Confluent se ha implementado correctamente con Helm:

   helm ls -n kafka
   

   El resultado debería ser similar al siguiente:

   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
   

Desplegar Kafka

En esta sección, desplegarás Kafka con una configuración básica y, después, probarás varios escenarios de configuración avanzada para cumplir los requisitos de disponibilidad, seguridad y observabilidad.

Configuración básica

La configuración básica de la instancia de Kafka incluye los siguientes componentes:

• Tres réplicas de brokers de Kafka, con un mínimo de dos réplicas disponibles para mantener la coherencia del clúster.
• Tres réplicas de nodos de ZooKeeper que forman un clúster.
• Dos listeners de Kafka: uno sin autenticación y otro que utiliza la autenticación TLS con un certificado generado por CFK.
• MaxHeapSize y MinHeapSize de Java definidos en 4 GB para Kafka.
• Asignación de recursos de CPU de 1 solicitud de CPU y 2 límites de CPU, y 5 GB de solicitudes y límites de memoria para Kafka (4 GB para el servicio principal y 0,5 GB para el exportador de métricas) y 3 GB para Zookeeper (2 GB para el servicio principal y 0,5 GB para el exportador de métricas).
• Se asignan 100 GB de almacenamiento a cada pod mediante premium-rwo storageClass, 100 para los datos de Kafka y 90/10 para los datos o registros de Zookeeper.
• Tolerancias, nodeAffinities y podAntiAffinities configurados para cada carga de trabajo, lo que asegura una distribución adecuada entre los nodos, utilizando sus respectivos grupos de nodos y diferentes zonas.
• La comunicación dentro del clúster está protegida por certificados autofirmados mediante una autoridad de certificación que proporciones.

Esta configuración representa la configuración mínima necesaria para crear un clúster de Kafka listo para producción. En las siguientes secciones se muestran configuraciones personalizadas para abordar aspectos como la seguridad del clúster, las listas de control de acceso (LCA), la gestión de temas y certificados, entre otros.

Crear un clúster de Kafka básico

1. Genera un par de autoridades de certificación:

   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 para Kubernetes proporciona certificados generados automáticamente para que los componentes de Confluent Platform los usen para el cifrado de red TLS. Debes generar y proporcionar una autoridad de certificación (AC).

2. Crea un secreto de Kubernetes para la autoridad de certificación:

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

   El nombre del secreto es predefined

3. Crea un clúster de Kafka con la configuración básica:

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

   Este comando crea un recurso personalizado de Kafka y un recurso personalizado de ZooKeeper del operador de CFK que incluyen solicitudes y límites de CPU y memoria, solicitudes de almacenamiento en bloque y taints y afinidades para distribuir los pods aprovisionados entre los nodos de Kubernetes.

4. Espera unos minutos mientras Kubernetes inicia las cargas de trabajo necesarias:

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

5. Verifica que se hayan creado las cargas de trabajo de Kafka:

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

   El resultado debería ser similar al siguiente:

   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
   

El operador crea los siguientes recursos:

• Dos StatefulSets para Kafka y ZooKeeper.
• Tres pods para las réplicas del broker de Kafka.
• Tres pods para réplicas de ZooKeeper.
• Dos recursos PodDisruptionBudget, lo que garantiza que haya un máximo de una réplica no disponible para mantener la coherencia del clúster.
• El servicio my-cluster, que actúa como servidor de arranque para los clientes de Kafka que se conectan desde el clúster de Kubernetes. Todos los listeners internos de Kafka están disponibles en este servicio.
• El servicio zookeeper, que permite que los brokers de Kafka se conecten a los nodos de ZooKeeper como clientes.

Autenticación y gestión de usuarios

En esta sección se muestra cómo habilitar la autenticación y la autorización para proteger los listeners de Kafka y compartir credenciales con los clientes.

Confluent para Kubernetes admite varios métodos de autenticación para Kafka, como los siguientes:

• Autenticación SASL/PLAIN: los clientes usan un nombre de usuario y una contraseña para autenticarse. El nombre de usuario y la contraseña se almacenan en el lado del servidor en un secreto de Kubernetes.
• SASL/PLAIN con autenticación LDAP: los clientes usan un nombre de usuario y una contraseña para autenticarse. Las credenciales se almacenan en un servidor LDAP.
• Autenticación mTLS: los clientes usan certificados TLS para la autenticación.

Limitaciones

• CFK no proporciona recursos personalizados para la gestión de usuarios. Sin embargo, puedes almacenar credenciales en Secrets y hacer referencia a Secrets en las especificaciones de los listeners.
• Aunque no hay ningún recurso personalizado para gestionar las listas de control de acceso directamente, la documentación oficial de Confluent para Kubernetes proporciona instrucciones sobre cómo configurar las listas de control de acceso mediante la CLI de Kafka.

Crear un usuario

En esta sección se muestra cómo implementar un operador de CFK que demuestre las funciones de gestión de usuarios, entre las que se incluyen las siguientes:

• Un clúster de Kafka con la autenticación basada en contraseñas (SASL/PLAIN) habilitada en uno de los listeners
• Un KafkaTopic con 3 réplicas
• Credenciales de usuario con permisos de lectura y escritura

1. Crea un secreto con las credenciales de usuario:

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

   Las credenciales deben almacenarse con el siguiente formato:

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

2. Configura el clúster de Kafka para que use un listener con autenticación SCRAM-SHA-512 basada en contraseñas en el puerto 9094:

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

3. Configura un tema y un pod de cliente para interactuar con tu clúster de Kafka y ejecutar comandos de 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 el secreto my-user-credentials en el pod del cliente como un volumen.

4. Cuando el pod del cliente esté listo, conéctate a él y empieza a producir y consumir mensajes con las credenciales proporcionadas:

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

5. Genera un mensaje con las credenciales de my-user y, a continuación, consume el mensaje para verificar que se ha recibido.

   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
   

   El resultado debería ser similar al siguiente:

   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
   

   Escribe CTRL+C para detener el proceso del consumidor. Si aparece un error Connect refused, espera unos minutos y vuelve a intentarlo.

6. Salir del shell del pod

   exit
   

Copias de seguridad y recuperación tras desastres

Con el operador de Confluent, puede implementar estrategias de copia de seguridad eficientes siguiendo determinados patrones.

Puedes usar Copia de seguridad de GKE para crear copias de seguridad de lo siguiente:

• Manifiestos de recursos de Kubernetes.
• Recursos personalizados de la API de Confluent y sus definiciones extraídas del servidor de la API de Kubernetes del clúster que se está copiando.
• Volúmenes que corresponden a recursos PersistentVolumeClaim encontrados en los manifiestos.

Para obtener más información sobre cómo crear copias de seguridad y restaurar clústeres de Kafka con Copia de seguridad de GKE, consulta Prepararse para la recuperación tras desastres.

También puedes hacer una copia de seguridad manual de tu clúster de Kafka. Deberías crear copias de seguridad de lo siguiente:

• La configuración de Kafka, que incluye todos los recursos personalizados de la API de Confluent, como KafkaTopicsoConnect
• Los datos, que se almacenan en los PersistentVolumes de los brokers de Kafka

Almacenar manifiestos de recursos de Kubernetes, incluidas las configuraciones de Confluent, en repositorios de Git puede eliminar la necesidad de tener una copia de seguridad independiente de la configuración de Kafka, ya que los recursos se pueden volver a aplicar a un nuevo clúster de Kubernetes cuando sea necesario.

Para proteger la recuperación de datos de Kafka en situaciones en las que se pierde una instancia de servidor de Kafka o un clúster de Kubernetes en el que se ha implementado Kafka, te recomendamos que configures la clase de almacenamiento de Kubernetes que se usa para aprovisionar volúmenes para los brokers de Kafka con la opción reclaimPolicy definida en Retain. También te recomendamos que hagas instantáneas de los volúmenes de los brokers de Kafka.

El siguiente archivo de manifiesto describe una StorageClass que usa la opción reclaimPolicy Retain:

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

En el siguiente ejemplo se muestra la StorageClass añadida al spec de un recurso personalizado de clúster de Kafka:

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

Con esta configuración, los PersistentVolumes aprovisionados con la clase de almacenamiento no se eliminan aunque se elimine el PersistentVolumeClaim correspondiente.

Para recuperar la instancia de Kafka en un nuevo clúster de Kubernetes con la configuración y los datos de la instancia de broker actuales, sigue estos pasos:

1. Aplicar los recursos personalizados de Confluent (Kafka, KafkaTopic, Zookeeper, etc.) a un nuevo clúster de Kubernetes
2. Actualiza los PersistentVolumeClaims con el nombre de las nuevas instancias de broker de Kafka a los PersistentVolumes antiguos mediante la propiedad spec.volumeName en PersistentVolumeClaim.