Orchestrez vos workflows avec Kestra

Jusqu’ici nous avons vu comment créer des flows et remplir les propriétés de tâches de façon basique. Dans ce chapitre, nous allons explorer les différents moyens mis à disposition dans Kestra pour dynamiser les flow et éviter de nous répéter.

Créez des variables

Imaginons un cas simple où nous voulons utiliser une même valeur dans deux tâches différentes. Comme vous pourriez le faire dans un langage de programmation classique, Kestra permet d’écrire des variables à différents endroits :

• Au niveau du Flow, la propriété variables permet de déclarer des variables au niveau d’un flow.

  id: hello_world
  namespace: company.team
  
  variables:
    myvar: hello
    numeric_variable: 42
  
  tasks:
    - id: log
      type: io.kestra.plugin.core.debug.Return
      format: "{{ vars.myvar }} world {{ vars.numeric_variable }}"

  Ces variables peuvent être utilisées avec le mot-clé vars dans vos flux.

• Au niveau du Namespace dans la version Enterprise de Kestra, vous pouvez déclarer ces variables au niveau d’un Namespace et les utiliser dans les flux correspondants.

• Au niveau de la configuration vous pouvez déclarer des variables globales au niveau de la configuration de votre instance Kestra.

Créez des secrets

Parfois, nous avons besoin de variables pour nous connecter à des services externes – il s’agit souvent d’identifiants – autrement dit, des valeurs sensibles.

Par exemple, Julien a besoin de donner des informations de connexion au serveur gérant l’envoi d’email. Il s’agit d’informations sensibles, donc il prend toutes les précautions nécessaires pour protéger ces données.

Kestra possède une notion de secret qui permet de cacher ces valeurs et ne pas compromettre les différents accès. Dans la version open source de Kestra, les secrets sont des variables encodées en base64 au niveau de la configuration de Kestra.

Voici les étapes à suivre pour configurer un secret :

1. Prenez votre valeur et encodez-la en base64. Vous pouvez effectuer cette opération avec l’outil de votre choix - par exemple la ligne de commande suivante permet d’encoder une chaîne de caractères : echo -n "password" | base64.

2. Prenez cette valeur et ajoutez la ligne suivant dans la configuration Kestra - par exemple avec l’installation Docker-Compose :

     kestra:
       image: kestra/kestra:latest
       environment:
         SECRET_MYPASSWORD: <VOTRE_VALEUR_ENCODE_EN_BASE64>

3. Redémarrez votre instance, vous allez pouvoir utiliser ce secret avec la commande{{ secret('MYPASSWORD') }} dans n’importe quel de votre flow.

Notez bien la présence du suffix SECRET_ qui permet d’indiquer à Kestra qu’il s’agit d’une valeur secrète et non d’une variable d’environnement classique.

La gestion des secrets est simplifiée dans la version Enterprise de Kestra: vous pouvez utiliser un secret manager et l’interface graphique directement pour créer vos secrets au niveau des Namespace. Cette méthode permet de ne plus avoir besoin de redémarrer l’instance pour insérer des nouveaux secrets.

Voici à quoi ressemble l’exemple d’envoi d’email de Julien avec des secrets proprement configurés :

- id: send_email
  type: io.kestra.plugin.notifications.mail.MailSend
  from: Julien@company.com
  to: ceo@company.com
  username: '{{ secret("EMAIL_USERNAME") }}'
  password: '{{ secret("EMAIL_PASSWORD") }}'
  host: smtp.resend.com
  port: 465
  subject: "Here is your dataset"
  attachments:
    - name: data.csv
      uri: "{{ outputs.query.uri }}"
  htmlTextContent: "Please find attached your dataset as a CSV file"

Utilisez des pluginDefaults

C’est ce qu’offre la propriété pluginDefaults. Elle est paramétrable au niveau du flow, d’un namespace (édition Enterprise) ou de l’instance. Cette propriété permet d’attribuer des valeurs par défaut pour n’importe quelle tâche. Par exemple, nous pouvons paramétrer la tâche postgres.Query avec les propriétés de connexion par défaut pour éviter de les réécrire à chaque utilisation :

pluginDefaults:
    - type: io.kestra.plugin.jdbc.postgresql.Query
      values:
        url: jdbcostgresql://postgres:5432/kestra
        username: kestra
        password: k3str4

Nous n’avons plus besoin de remplir les propriétés lorsque nous utilisons la tâche :

id: postgres
namespace: kestra.sandbox
tasks:

  - id: query_1
    type: io.kestra.plugin.jdbc.postgresql.Query
    sql: SELECT * FROM pg_catalog.pg_namespace
    store: true


  - id: query_2
    type: io.kestra.plugin.jdbc.postgresql.Query
    sql: SELECT field1, AVG(field2) FROM schema.table GROUP BY field1
    store: true

pluginDefaults:
    - type: io.kestra.plugin.jdbc.postgresql.Query
      values:
        url: jdbc:postgresql://postgres:5432/kestra
        username: '{{ secret("DATABASE_USERNAME") }}'
        password: '{{ secret("DATABASE_PASSWORD") }}'

Julien de son côté peut aller encore plus loin: l’entreprise de Julien utilise AWS comme fournisseur cloud. Ainsi, il utilise de nombreuses tâches issues du plugin AWS.

Avec les pluginDefaults, Julien peut complètement automatiser l’identification aux services AWS :

    pluginDefaults:
      - type: io.kestra.plugin.aws
        values:
          accessKeyId: "{{ secret('AWS_ACCESS_KEY_ID') }}"
          secretKeyId: "{{ secret('AWS_SECRET_ACCESS_KEY') }}"
          region: "us-east-1"

Vous l’avez peut-être compris : il est possible d’attribuer des propriétés par défaut au niveau d’un plugin. Ainsi, toutes les tâches associées ayant les propriétés correspondantes hériteront des valeurs par défaut (dans notre exemple, les tâches liées au plugin AWS).

Cela est très puissant pour éviter toute erreur de configuration et ne plus se soucier des éléments qui sont au final peu importants d’un point de vue fonctionnel.

À vous de jouer !

Nous allons ici insérer des données dans une base de données en utilisant une base Postgres et des requêtes SQL basiques pour ingérer et enfin vérifier que l’opération s’est bien déroulée.

1. Créez un compte Neon pour avoir accès à une base de données PostgreSQL distante.

2. Créez une requête SQL qui crée un schéma, une table et insert des données dans la base de données CREATE SCHEMA, INSERT, QUERY.

3. Utilisez des secret pour remplir les propriétés de connexions à la tâche.

4. Vérifiez en exécutant un "task Query" que les données ont bien insérées ou en allant sur l’interface graphique de Neon.

Voici l'exercice :

id: exercise
namespace: open_class_room.kestra
description: |
  In this exercise we use several plugins to ingest data into a PostgreSQL database.
  
  **Prerequisites**:

    - PostgreSQL database. You can create a simple Postgre database with [Neon service](https://neon.tech/).

  
tasks:
  # 1. Run queries to create schema and tables in Postgres database
  # The connection url is jdbc:postgresql://<YOUR_POSTGRES_HOST_URL>:5432/<YOUR_DATABASE_NAME> 
  # The username and password comes from your Neon account
        # DROP SCHEMA IF EXISTS data CASCADE;
        # CREATE SCHEMA IF NOT EXISTS data;
        # CREATE TABLE IF NOT EXISTS data.salaries (
        #   "work_year" INT,
        #   "experience_level" VARCHAR,
        #   "employment_type" VARCHAR,
        #   "job_title" VARCHAR,
        #   "salary" INT,
        #   "salary_currency" VARCHAR,
        #   "salary_in_usd" INT,
        #   "employee_residence" VARCHAR,
        #   "remote_ratio" INT,
        #   "company_location" VARCHAR,
        #   "company_size" VARCHAR
        # );
  - id: ddl_queries
    type: 
    url: 
    username: 
    password: 
    sql: |


  - id: download
    type: io.kestra.plugin.core.http.Download
    uri: https://raw.githubusercontent.com/kestra-io/datasets/main/csv/salaries.csv


 # 2. Ingest data into the Postgres table
  - id: ingest
    type: io.kestra.plugin.jdbc.postgresql.CopyIn
    url: 
    username: 
    password: 
    format: CSV
    from: "{{ outputs.download.uri }}"
    table: 
    header: true

  # 3. Run the following query on the previous ingested Postgres table
  # SELECT job_title, AVG(salary) FROM data.salaries GROUP BY job_title
  - id: query
    type:
    store: true
    # ...

En résumé

• Les variables permettent de réutiliser des valeurs à travers vos flux.

• Les secrets sont des variables spéciales permettant de sécuriser des informations d’accès à des services externes.

• Les pluginDefault offrent la possibilité de donner des valeurs par défaut à vos tâches.

• Les pluginDefault sont très puissants pour simplifier et garder votre code simple.

Jusqu’ici, nous avons vu des concepts très liés à Kestra. Cependant, pour manipuler la donnée, nous avons souvent besoin d’intégrer des scripts écrits en Python ou SQL. Dans le chapitre suivant, nous allons voir comment intégrer n’importe quel type de scripts dans nos flux Kestra.