Desplegar Apache Kafka en GKE con Strimzi

En esta guía se muestra cómo usar el operador Strimzi para desplegar clústeres de Apache Kafka.

Kafka es un sistema de mensajería distribuido de código abierto diseñado para gestionar datos de streaming de alto volumen, alto rendimiento y en tiempo real. Te permite crear pipelines de datos de streaming para transferir datos de forma fiable entre diferentes sistemas y aplicaciones, de modo que puedas llevar a cabo tareas de procesamiento y análisis.

Los operadores son extensiones de software que usan recursos personalizados para gestionar aplicaciones y sus componentes. Para obtener más información sobre los motivos para usar operadores, consulta Patrón de operador en la documentación de Kubernetes de código abierto. El operador de Strimzi ofrece flexibilidad en las opciones de implementación y te permite usar taints y tolerancias de Kubernetes para ejecutar Kafka en nodos dedicados.

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.

Esta solución es un buen punto de partida si quieres aprender a desplegar clústeres de Kafka con un operador de terceros para automatizar la gestión y reducir los errores. Si prefieres tener un control operativo más granular, consulta Desplegar clústeres de Kafka de alta disponibilidad en GKE.

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 consolaIcono de activación de Cloud Shell 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 strimzi-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 strimzi-cluster --region us-central1"

Conectarse 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 Strimzi en el clúster

En esta sección, desplegarás el operador de Strimzi mediante un gráfico de Helm. También hay varias formas de implementar Strimzi.

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

    helm repo add strimzi https://strimzi.io/charts/

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

    kubectl create ns kafka

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

    helm install strimzi-operator strimzi/strimzi-kafka-operator -n kafka

    Para desplegar el operador de clúster de Strimzi y los clústeres de Kafka en diferentes espacios de nombres, añade el parámetro --set watchNamespaces="{kafka-namespace,kafka-namespace-2,...}" al comando helm.

  4. Verifica que el operador de clúster de Strimzi 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
strimzi-operator    kafka      1       2023-06-27 11:22:15.850545 +0200 CEST    deployed    strimzi-kafka-operator-0.35.0    0.35.0

Desplegar Kafka

Una vez que el operador se haya desplegado en el clúster, podrás desplegar una instancia de clúster de 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 Strimzi.
  • MaxHeapSize y MinHeapSize de Java se han definido en 4 GB para Kafka y 2 GB para ZooKeeper.
  • Asignación de recursos de CPU de 1 solicitud de CPU y 2 límites de CPU tanto para Kafka como para ZooKeeper, junto con 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 2,5 GB para ZooKeeper (2 GB para el servicio principal y 0,5 GB para el exportador de métricas).
  • Entidad-operador con las siguientes solicitudes y límites:
    • tlsSidecar: 100 m/500 m de CPU y 128 Mi de memoria.
    • topicOperator: 100 m/500 m de CPU y 512 Mi de memoria.
    • userOperator: 500 m de CPU y 2 Gi de memoria.
  • Se asignan 100 GB de almacenamiento a cada pod mediante premium-rwo storageClass.
  • 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.
  • Comunicación dentro del clúster protegida por certificados autofirmados: autoridades de certificación (CAs) independientes para el clúster y los clientes (mTLS). También puedes configurar el uso de otra autoridad de certificación.

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 de los clústeres, 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. Crea un clúster de Kafka con la configuración básica:

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

    Este comando crea un recurso personalizado de Kafka del operador Strimzi que incluye solicitudes y límites de CPU y memoria, solicitudes de almacenamiento en bloque y una combinación de taints y afinidades para distribuir los pods aprovisionados entre los nodos de Kubernetes.

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

    kubectl wait kafka/my-cluster --for=condition=Ready --timeout=600s -n kafka

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

    kubectl get pod,service,deploy,pdb -l=strimzi.io/cluster=my-cluster -n kafka

    El resultado debería ser similar al siguiente:

    NAME                                            READY   STATUS  RESTARTS   AGE
pod/my-cluster-entity-operator-848698874f-j5m7f   3/3   Running   0        44m
pod/my-cluster-kafka-0                          1/1   Running   0        5m
pod/my-cluster-kafka-1                          1/1   Running   0        5m
pod/my-cluster-kafka-2                          1/1   Running   0        5m
pod/my-cluster-zookeeper-0                      1/1   Running   0        6m
pod/my-cluster-zookeeper-1                      1/1   Running   0        6m
pod/my-cluster-zookeeper-2                      1/1   Running   0        6m

NAME                                TYPE      CLUSTER-IP   EXTERNAL-IP   PORT(S)                             AGE
service/my-cluster-kafka-bootstrap  ClusterIP   10.52.8.80   <none>      9091/TCP,9092/TCP,9093/TCP          5m
service/my-cluster-kafka-brokers    ClusterIP   None         <none>      9090/TCP,9091/TCP,9092/TCP,9093/TCP   5m
service/my-cluster-zookeeper-client   ClusterIP   10.52.11.144   <none>      2181/TCP                            6m
service/my-cluster-zookeeper-nodes  ClusterIP   None         <none>      2181/TCP,2888/TCP,3888/TCP          6m

NAME                                       READY   UP-TO-DATE   AVAILABLE   AGE
deployment.apps/my-cluster-entity-operator   1/1   1          1         44m

NAME                                            MIN AVAILABLE   MAX UNAVAILABLE   ALLOWED DISRUPTIONS   AGE
poddisruptionbudget.policy/my-cluster-kafka     2             N/A             1                   5m
poddisruptionbudget.policy/my-cluster-zookeeper   2             N/A             1                   6m

El operador crea los siguientes recursos:

  • Dos StrimziPodSets para Kafka y ZooKeeper.
  • Tres pods para las réplicas del broker de Kafka.
  • Tres pods para réplicas de ZooKeeper.
  • Dos PodDisruptionBudgets, lo que garantiza una disponibilidad mínima de dos réplicas para la coherencia del clúster.
  • Un servicio llamado my-cluster-kafka-bootstrap, 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.
  • Un servicio sin interfaz gráfica llamado my-cluster-kafka-brokers que permite la resolución de DNS de las direcciones IP de los pods de broker de Kafka directamente. Este servicio se usa para la comunicación entre brokers.
  • Un servicio llamado my-cluster-zookeeper-client que permite que los brokers de Kafka se conecten a los nodos de ZooKeeper como clientes.
  • Un servicio sin interfaz gráfica llamado my-cluster-zookeeper-nodes que permite la resolución de DNS de las direcciones IP de los pods de ZooKeeper directamente. Este servicio se usa para conectar réplicas de ZooKeeper.
  • Una implementación llamada my-cluster-entity-operator que contiene los operadores topic-operator y user-operator y facilita la gestión de los recursos personalizados KafkaTopics y KafkaUsers.

También puedes configurar dos NetworkPolicies para facilitar la conectividad a los listeners de Kafka desde cualquier pod y espacio de nombres. Estas políticas también restringirían las conexiones a ZooKeeper a los brokers y habilitarían la comunicación entre los pods del clúster y los puertos de servicio internos exclusivos de la comunicación del clúster.

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.

Strimzi proporciona un método nativo de Kubernetes para la gestión de usuarios mediante un User Operator independiente y su recurso personalizado de Kubernetes correspondiente, KafkaUser, que define la configuración del usuario. La configuración de usuario incluye ajustes de autenticación y autorización, y aprovisiona al usuario correspondiente en Kafka.

Strimzi puede crear listeners y usuarios de Kafka que admitan varios mecanismos de autenticación, como la autenticación basada en nombre de usuario y contraseña (SCRAM-SHA-512) o TLS. También puedes usar la autenticación OAuth 2.0, que suele considerarse un método mejor que el uso de contraseñas o certificados para la autenticación debido a la seguridad y a la gestión de credenciales externas.

Desplegar un clúster de Kafka

En esta sección se muestra cómo implementar un operador de Strimzi 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 (SCRAM-SHA-512) habilitada en uno de los listeners.
  • KafkaTopic con 3 réplicas.
  • Un KafkaUser con una LCA que especifica que el usuario tiene permisos de lectura y escritura en el tema.

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

    kubectl apply -n kafka -f kafka-strimzi/manifests/03-auth/my-cluster.yaml

  2. Crea un Topic, un User y un pod de cliente para ejecutar comandos en el clúster de Kafka:

    kubectl apply -n kafka -f kafka-strimzi/manifests/03-auth/topic.yaml
kubectl apply -n kafka -f kafka-strimzi/manifests/03-auth/my-user.yaml

    El Secret my-user con las credenciales del usuario se monta en el pod del cliente como un volumen.

    Estas credenciales confirman que el usuario tiene permisos para publicar mensajes en el tema mediante el listener con la autenticación basada en contraseña (SCRAM-SHA-512) habilitada.

  3. Crea un pod de cliente:

    kubectl apply -n kafka -f kafka-strimzi/manifests/03-auth/kafkacat.yaml

  4. Espera unos minutos a que el pod del cliente se convierta en Ready y, a continuación, conéctate a él:

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

  5. Genera un nuevo mensaje con las credenciales de my-user e intenta consumirlo:

    echo "Message from my-user" |kcat \
  -b my-cluster-kafka-bootstrap.kafka.svc.cluster.local:9094 \
  -X security.protocol=SASL_SSL \
  -X sasl.mechanisms=SCRAM-SHA-512 \
  -X sasl.username=my-user \
  -X sasl.password=$(cat /my-user/password) \
  -t my-topic -P
kcat -b my-cluster-kafka-bootstrap.kafka.svc.cluster.local:9094 \
  -X security.protocol=SASL_SSL \
  -X sasl.mechanisms=SCRAM-SHA-512 \
  -X sasl.username=my-user \
  -X sasl.password=$(cat /my-user/password) \
  -t my-topic -C

    El resultado debería ser similar al siguiente:

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

    Escribe CTRL+C para detener el proceso del consumidor.

  6. Salir del shell del pod

    exit

Copias de seguridad y recuperación tras desastres

Aunque el operador Strimzi no ofrece funciones de copia de seguridad integradas, puede implementar estrategias de copia de seguridad eficientes siguiendo ciertos 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 Strimzi 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 crear una copia de seguridad de un clúster de Kafka que se haya implementado mediante el operador Strimzi. Deberías crear copias de seguridad de lo siguiente:

  • La configuración de Kafka, que incluye todos los recursos personalizados de la API de Strimzi, como KafkaTopics y KafkaUsers.
  • Los datos, que se almacenan en los PersistentVolumes de los brokers de Kafka.

Almacenar manifiestos de recursos de Kubernetes, incluidas las configuraciones de Strimzi, en repositorios de Git puede eliminar la necesidad de tener una copia de seguridad independiente para 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 como 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:
  kafka:
    # ...
    storage:
      type: persistent-claim
      size: 100Gi
      class: premium-rwo-retain

Con esta configuración, los PersistentVolumes aprovisionados mediante 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 Kafka de Strimzi (Kakfa, KafkaTopic, KafkaUser, 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.