Usando Kustomize para instalar Apache Flink en Kubernetes

En esta ocasión, vamos a usar Kustomize para instalar Apache Flink en Kubernetes (K8s). Ya hice una introducción a Kustomize en una entrada anterior que recomiendo leer si no está claro qué es y para qué sirve Kustomize, y aunque hablé de Kustomize en combinación con Helm, aquí no voy utilizar Helm.

Queremos tener Flink desplegado en un clúster de K8s. Tenemos tres clúster en este ejemplo, un clúster local que consiste en un K8s en nuestra propia máquina donde vamos a probar (minikube, microk8s, kind o cualquier otro), un clúster de pruebas llamado test donde vamos a realizar pruebas más a fondo, como rendimiento, etc. y, finalmente, un clúster de producción, es decir, el que van a usar nuestros clientes.

Lo primero es saber qué necesitamos para tener un Flink en K8s. Tras una búsqueda en Google, llegamos a esta página. Ahí, se indica que necesitamos tres recursos, por un lado, el despliegue del jobmanager, que será el gestor del clúster de Flink, quien decide dónde y cuando se ejecutan nuestras tareas de Flink. A su vez, se necesita exponer un servicio para el jobmanager en el clúster, que sería nuestro segundo recurso. Finalmente, tenemos el taskmanager, donde se ejecutan las tareas de Flink.

Voy a dejar aquí cada uno de ellos y a explicar paso a paso todo el proceso, pero antes, voy a dejar el árbol del proyecto completo aquí:

.
├── base
│   ├── jobmanager-deployment.yaml
│   ├── jobmanager-service.yaml
│   ├── kustomization.yaml
│   └── taskmanager-deployment.yaml
└── overlays
    ├── local
    │   ├── flink-conf.yaml
    │   ├── flinkconfig_patch.yaml
    │   ├── healthcheck_patch.yaml
    │   └── kustomization.yaml
    ├── production
    │   ├── cpulimit_patch.yaml
    │   ├── flink-conf.yaml
    │   ├── flinkconfig_patch.yaml
    │   ├── healthcheck_patch.yaml
    │   ├── kustomization.yaml
    │   ├── log4j-console.properties
    │   ├── memorylimit_patch.yaml
    │   └── replicacount_patch.yaml
    └── test
        ├── cpulimit_patch.yaml
        ├── flink-conf.yaml
        ├── flinkconfig_patch.yaml
        ├── healthcheck_patch.yaml
        ├── kustomization.yaml
        ├── memorylimit_patch.yaml
        └── replicacount_patch.yaml

También está en Github, por si alguien quiere clonarlo directamente.

Recursos base

En primer lugar, necesitamos definir nuestro recursos base, es decir, el jobmanager, su servicio y el taskmanager:

Tenemos, por tanto, una configuración básica para cada recurso, decimos el número de réplicas que queremos, en este caso, una, tenemos definidos los puertos, la imagen y versión a utilizar, las típicas etiquetas y poco más.

Podríamos desplegar cada uno de los recursos tal y como están definidos y todo funcionaría.

Todos estos YAML van a ir a una carpeta que vamos a llamar base. Para cualquier duda sobre cómo queda la jerarquía de ficheros podéis acudir al árbol que puse arriba.

En dicha carpeta base nos falta por crear un fichero llamado kustomization.yaml que es requerido por Kustomize. Él leerá dicho fichero durante el proceso de construcción de los YAML definitivos según entorno. Tiene el siguiente contenido:

resources:
- jobmanager-deployment.yaml
- jobmanager-service.yaml
- taskmanager-deployment.yaml

Simplemente, se indica los recursos de la carpeta base que debe parchear, es decir, modificar los valores según cada entorno.

Ahora, vamos a ir entorno por entorno creando los parches o modificaciones de las plantillas base, de acuerdo a cómo queremos que quede.

Entorno local

El primer entorno que necesitamos configurar es el entorno local. El más sencillo en teoría.

Lo que vamos a hacer en este caso es añadir un fichero de configuración que necesita Flink, y añadir un recurso HTTP que expone Flink para poder comprobar la salud del jobmanager.

Estamos por tanto en una nueva carpeta llamada overlays y, dentro de ella, una carpeta para nuestro entorno local que llamaremos local.

En ella, vamos a crear en primer lugar el fichero kustomization.yaml:

bases:
- ../../base
patches:
- healthcheck_patch.yaml
- flinkconfig_patch.yaml
configMapGenerator:
- name: flink-config
  files:
  - flink-conf.yaml

Indicamos en primer lugar donde se encuentran los YAML base. En segundo lugar, los ficheros con los parches y, finalmente, le indicamos que nos genere un ConfigMap con el contenido del fichero flink-config.yaml.

El fichero healthcheck_patch.yaml lo hemos definido de la siguiente forma:

Utilizamos el mecanismo livenessProbe de K8s para que compruebe la salud del jobmanager de acuerdo a los parámetros indicados, en este caso, el recurso y el puerto, el tipo de método HTTP a utilizar, y cada cuántos segundos debe lanzar la petición y cuántos segundos debe esperar.

El otro fichero, flinkconfig_patch.yaml, es el fichero donde vamos a montar como volumen la configuración de flink del fichero flink-cong.yaml, el cual se definirá en el clúster como recurso de tipo ConfigMap:

El último fichero que nos queda por mostrar es el propio flink.config.yaml:

Con esto, hemos cubierto las necesidades que tenemos en este entorno.

Quizás, lo ideal sea tener la configuración lo más parecida a producción, aunque con valores diferentes por entornos. Es lo más recomendable para evitar problemas en entornos superiores. Pero, para este ejemplo, vamos a ir añadiendo cosas en cada entorno, aunque tengamos que hacer el esfuerzo de aceptar pulpo como animal de compañía.

Vamos ahora a generar las plantillas finales, es decir, con los parches aplicados sobre la base, con la siguiente instrucción:

$ kustomize build overlays/local

Vamos a mostrar aquí la salida pero fijándonos sólo en cómo ha cambiado el recurso Deployment del jobmanager, el que hemos parcheado:

apiVersion: extensions/v1beta1
kind: Deployment
metadata:
  name: flink-jobmanager
spec:
  replicas: 1
  template:
    metadata:
      labels:
        app: flink
        component: jobmanager
    spec:
      containers:
      - args:
        - jobmanager
        env:
        - name: FLINK_CONF_DIR
          value: /etc/flink
        - name: JOB_MANAGER_RPC_ADDRESS
          value: flink-jobmanager
        image: flink:1.8.1
        imagePullPolicy: IfNotPresent
        livenessProbe:
          httpGet:
            path: /overview
            port: 8081
          initialDelaySeconds: 30
          periodSeconds: 10
        name: jobmanager
        ports:
        - containerPort: 6123
          name: rpc
        - containerPort: 6124
          name: blob
        - containerPort: 6125
          name: query
        - containerPort: 8081
          name: ui
        volumeMounts:
        - mountPath: /etc/flink
          name: flink-config
      volumes:
      - configMap:
          name: flink-config-f49d4294dm
        name: flink-config

Este es el resultado final, lo que teníamos en la base junto al livenessProbe y el volumen con la configuración.

Además, aparece un nuevo recurso que no teníamos al principio ni está en la carpeta base:

apiVersion: v1
data:
  flink-conf.yaml: |
    blob.server.port: 6124
    jobmanager.rpc.address: flink-jobmanager
    jobmanager.rpc.port: 6123
    jobmanager.heap.mb: 1024
    taskmanager.heap.mb: 1024
    taskmanager.numberOfTaskSlots: 1
kind: ConfigMap
metadata:
  name: flink-config-f49d4294dm

Esto es lo que comentabamos antes, Kustomize nos crea la definición YAML para el ConfigMap que tendrá como contenido del fichero flink-config.yaml.

Despliegue

Esto lo podemos desplegar “enganchando” la salida de la instrucción de Kustomize con el comando (declarativo) apply del cliente de K8s:

$ kustomize build overlays/local | kubectl apply -f -

Tras ejecutarlo vemos la siguiente salida del comando anterior, que nos indica que se han creado los recursos de la base más el ConfigMap:

configmap/flink-config-f49d4294dm unchanged
service/flink-jobmanager created
deployment.extensions/flink-jobmanager created
deployment.extensions/flink-taskmanager created

Vamos a comprobar que todo ha ido correctamente preguntando cuales son los pods que tenemos ejecutándose en el clúster:

$ kubectl get pods
NAME                                 READY   STATUS   RESTARTS   
flink-jobmanager-6cf45f78c4-hz4gc    1/1     Running  0          
flink-taskmanager-78984c8bc4-ntr2n   1/1     Running  0          

Como la configuración en nuestro caso para el número de réplicas es de una réplica para el taskmanager, sólo vemos un pod en dicho caso.

Podemos acceder a la consola de administración web del manager ejecutando la siguiente instrucción en una ventana del terminal independiente:

$ kubectl proxy

Y, accediendo a la siguiente URL:

http://localhost:8001/api/v1/namespaces/default/services/flink-jobmanager:ui/proxy

La consola podemos aplicarlo como comprobación en cada entorno, así que los pasos a seguir son estos. Por no dejar un post muy extenso voy a omitirlo en los siguientes entornos.

Entorno de pruebas

Lo primero que hay que indicar aquí, y que sirve para los siguientes entornos, es que en realidad no tengo 3 clústers diferentes. Estoy siempre trabajando en local, pero nos imaginamos en un nuevo esfuerzo que se trata de un clúster diferentes.

Estamos, por tanto, probando los despliegues y configuraciones de entornos de Kustomize en nuestra propia máquina local.

En un entorno de trabajo real, sí tendríamos un clúster diferente por entorno.

Dicho esto, vamos a pasar directamente a mostrar el fichero kustomization.yaml de este entorno, al que hemos llamado test y cuya carpeta recibe el mismo nombre:

Aquí la cosa se complica. Ahora necesitamos que el entorno tenga más restricciones y, además, que se parezca aún más al entorno de producción.

Tenemos, por tanto, además de lo que configuramos para local los requisitos de limitar el uso de los recursos del clúster de K8s que va a utilizar Flink, tanto de memoria como de CPU, así como, variar el número de réplicas del taskmanager.

Vamos a mostrar los ficheros que son diferentes en este caso, ya que el resto se mantienen sin cambios respecto del entorno local.

El primero y más sencillo es replicacount_patch.yaml, donde modificamos el campo replicas del YAML base. En este caso, queremos dos instancias del taskmanager.

Este segundo fichero, indicamos la limitación de uso de CPU por parte del pod.

El último sería memorylimit_patch.yaml, que es identico al anterior salvo porque ahora indicamos el límite de memoria en vez del límite de CPU. Se podría agrupar en uno pero lo hemos dejado separado en dos ficheros.

Despliegue

Igual que hicimos con el entorno local, vamos a desplegar esto en K8s. No voy a mostrar la salida de Kustomize ni los cambios sobre la base por recudir el tamaño del post.

Previamente, hemos borrado el despliegue anterior porque al estar trabajando en el mismo clúster local siempre, los despliegues van a “colisionar”. Aunque estamos usando apply y debería modificar el estado anterior, prefiero trabajar sobre el clúster limpio sólo para este ejemplo, ya que, en teoría son entornos “diferentes”:

$ kustomize build overlays/test | kubectl apply -f -

Vamos a comprobar, mediante el número de réplicas del taskmanager, que la configuración es la correcta para este entorno:

$ kubectl get pods
NAME                                 READY   STATUS    RESTARTS
flink-jobmanager-79d94d4dc9-67g84    1/1     Running   0          
flink-taskmanager-78984c8bc4-zc8h5   1/1     Running   0          
flink-taskmanager-78984c8bc4-twp9p   1/1     Running   0

La configuración se ha aplicado correctamente para este entorno.

Entorno de producción

Para este entorno y por no extendernos la mecánica es la misma, los cambios que vamos a aplicar aquí son sólo tener cuatro réplicas del taskmanager frente a dos del entorno de pruebas, y tener un nuevo fichero de configuración para los logs de Flink.

El cambio en el número de réplicas es directo, vamos al fichero replicacount_patch.yaml y cambiamos el campo replicas par que tenga 4:

...
spec:
  replicas: 4
...

En cuanto al fichero de configuración de los logs, necesitamos crear un fichero log4j-console.properties:

Ahora sólo tenemos que añadirlo reutilizando el ConfigMap anterior, vía fichero kustomization.yaml:

bases:
- ../../base
patches:
- healthcheck_patch.yaml
- flinkconfig_patch.yaml
- memorylimit_patch.yaml
- cpulimit_patch.yaml
- replicacount_patch.yaml
configMapGenerator:
- name: flink-config
  files:
  - flink-conf.yaml
  - log4j-console.properties

Vemos que aparece un nuevo campo al final:

...
  files:
  ..
  - log4j-console.properties

Despliegue

Con esto, ya tenemos todo. Por tanto, vamos a desplegar:

$ kustomize build overlays/production | kubectl apply -f -

El resultado lo vamos a comprobar viendo el número de réplicas del taskmanager:

$ kubectl get pods
NAME                                 READY   STATUS    RESTARTS
flink-jobmanager-79d94d4dc9-67g84    1/1     Running   0          
flink-taskmanager-78984c8bc4-zc8h5   1/1     Running   0          
flink-taskmanager-78984c8bc4-twp9p   1/1     Running   0
flink-taskmanager-78984c8bc4-jzf2x   1/1     Running   0          
flink-taskmanager-78984c8bc4-lbhrw   1/1     Running   0

Conlusión

Con Kustomize podemos tener nuestras aplicaciones definidas en formato YAML sin ningún tipo de anotación de tipo plantilla como ocurría con Helm. Además, podemos gestionar diferentes configuraciones por entorno como código.

Este ejemplo es extensible, según el caso, necesitaremos por ejemplo cambiar el usuario y contraseña del administrador en cada entorno para una base de datos MySQL. Además, como se muestra en la entrada de introducción a esta herramienta y cuyo link está al principio, se puede combinar con Helm para ahorrar tiempo, de esta forma, sólo hay que definir nuestra configuración según nuestros entornos y nuestros requisitos.

Como he comentado, lo ideal es que la configuración sea la misma en todos los entornos, aunque los valores para las réplicas, etc. serán diferentes. Así evitamos que surjan problemas en entornos superiores que no hemos detectado en local porque la configuración es diferente o no es tan completa.