11 - Le filtre Mutate : transformer les champs

Ce que tu vas apprendre

• Toutes les opérations du filtre mutate (rename, remove, convert, gsub, split, etc.)
• L'ordre d'exécution interne d'un bloc mutate
• Quand utiliser un seul bloc vs plusieurs blocs mutate
• Nettoyer les metadonnees Filebeat
• Les pièges courants

Prerequisites

• Avoir compris les filtres Logstash (voir article 03)

---

Le filtre que tu utiliseras dans chaque pipeline

Grok parse. Dissect decoupe. Mais apres le parsing, tu as souvent besoin de petites transformations : renommer un champ, changer un type, supprimer des metadonnees inutiles, mettre en minuscules. C'est le job de mutate.

Mutate est dans chaque pipeline que j'ai écrit. C'est le filtre le plus basique et le plus frequent.

Les opérations, une par une

rename : renommer un champ

filter {
  mutate {
    rename => {
      "clientip" => "client_ip"
      "[host][hostname]" => "server_name"
    }
  }
}

Utile pour harmoniser les noms de champs entre différentes sources. Grok appelle le champ clientip, mais ton schema Elasticsearch attend client_ip. Un rename suffit.

remove_field : supprimer un champ

filter {
  mutate {
    remove_field => ["message", "@version", "event", "host"]
  }
}

Les champs inutiles prennent de la place dans Elasticsearch. Apres le parsing, si tu n'as plus besoin de message (la ligne brute), supprime-le.

add_field : ajouter un champ

filter {
  mutate {
    add_field => {
      "environment" => "production"
      "processed_by" => "logstash-01"
      "[geo][country]" => "France"
    }
  }
}

Les champs imbriques se creent avec la notation entre crochets. [geo][country] créé un objet geo avec un champ country.

Tu peux utiliser des références a d'autres champs :

filter {
  mutate {
    add_field => { "log_id" => "%{service}-%{request_id}" }
  }
}

convert : changer le type

filter {
  mutate {
    convert => {
      "status_code" => "integer"
      "duration" => "float"
      "in_stock" => "boolean"
      "bytes" => "integer"
    }
  }
}

Types supportes : integer, float, string, boolean.

Grok extrait tout en string par défaut (sauf avec la syntaxe %{INT:field:int}). Mutate convert est le moyen standard de corriger les types.

Pour les booleens, les valeurs true, t, yes, y, 1 deviennent true. Tout le reste est false.

gsub : remplacer du texte avec une regex

filter {
  mutate {
    gsub => [
      # Supprimer les retours chariot Windows
      "message", "\r", "",

      # Remplacer les slashes par des tirets dans une date
      "date_field", "/", "-",

      # Masquer les emails
      "log_line", "[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}", "[EMAIL_MASKED]"
    ]
  }
}

La syntaxe de gsub est un tableau plat : [champ, regex, remplacement, champ, regex, remplacement, ...]. Trois éléments par opération.

lowercase et uppercase

filter {
  mutate {
    lowercase => ["level", "http_method"]
    uppercase => ["country_code"]
  }
}

lowercase transforme ERROR en error. uppercase transforme fr en FR.

strip : supprimer les espaces

filter {
  mutate {
    strip => ["username", "email"]
  }
}

Supprime les espaces en début et fin de chaîne. Comme trim() en JavaScript.

split : couper en tableau

filter {
  mutate {
    split => { "tags_string" => "," }
  }
}

Entree : tags_string = "docker,kubernetes,devops" Résultat : tags_string = ["docker", "kubernetes", "devops"]

join : fusionner un tableau en string

filter {
  mutate {
    join => { "tags" => ", " }
  }
}

L'inverse de split. ["docker", "kubernetes"] => "docker, kubernetes".

merge : fusionner deux champs

filter {
  mutate {
    merge => { "all_tags" => "extra_tags" }
  }
}

Si all_tags est ["a", "b"] et extra_tags est ["c"], le résultat est all_tags = ["a", "b", "c"].

copy : copier un champ

filter {
  mutate {
    copy => { "message" => "original_message" }
  }
}

Cree une copie independante. Utile pour garder la valeur brute avant de la modifier avec gsub ou d'autres opérations.

update : ecraser un champ existant

filter {
  mutate {
    update => { "level" => "UNKNOWN" }
  }
}

update ne fait quelque chose que si le champ existe deja. Si level n'existe pas, rien ne se passe. C'est la différence avec add_field, qui créé le champ dans tous les cas.

replace : remplacer la valeur

filter {
  mutate {
    replace => { "message" => "%{level} - %{service} - %{error_message}" }
  }
}

Comme update, mais créé le champ s'il n'existe pas. Accepte les références a d'autres champs avec %{...}.

L'ordre d'exécution dans un bloc mutate

C'est le piège principal. Dans un seul bloc mutate {}, les opérations ne s'executent pas dans l'ordre ou tu les ecris. Elles suivent un ordre fixe :

1. coerce
2. rename
3. update
4. replace
5. convert
6. gsub
7. uppercase
8. lowercase
9. strip
10. remove_field
11. split
12. join
13. merge
14. copy

Ca veut dire que si tu ecris :

filter {
  mutate {
    rename => { "old_name" => "new_name" }
    convert => { "new_name" => "integer" }
  }
}

Ca marche, parce que rename (étape 2) s'exécuté avant convert (étape 5).

Mais si tu ecris :

filter {
  mutate {
    lowercase => ["level"]
    gsub => ["level", "error", "ERREUR"]
  }
}

Ca ne fait pas ce que tu penses. gsub (étape 6) s'exécuté avant lowercase (étape 8). Le gsub cherche "error" sur la valeur originale (qui est peut-etre "ERROR"), ne trouve rien, et ensuite lowercase met tout en minuscules.

La solution : plusieurs blocs mutate

Pour garantir l'ordre, utilise des blocs separes :

filter {
  mutate {
    lowercase => ["level"]
  }

  mutate {
    gsub => ["level", "error", "erreur"]
  }
}

Les blocs mutate sont exécutés dans l'ordre du fichier. Le premier met en minuscules, le deuxieme fait le gsub sur la valeur deja en minuscules.

Ma regle : si l'ordre compte, un bloc par opération. Si l'ordre ne compte pas (rename + remove_field sur des champs différents), un seul bloc suffit.

Cas pratique : nettoyer les metadonnees Filebeat

Quand Filebeat envoie des événements a Logstash, il ajoute beaucoup de metadonnees. Voici un mutate typique pour nettoyer :

filter {
  # ... parsing avec Grok ou Dissect ...

  # Nettoyer les metadonnees Filebeat
  mutate {
    remove_field => [
      "agent",          # info sur l'agent Filebeat
      "ecs",            # Elastic Common Schema version
      "input",          # type d'input (log, stdin)
      "log",            # chemin du fichier source
      "@version",       # toujours "1", inutile
      "event"           # event.original (doublon de message)
    ]
  }

  # Renommer les champs Filebeat pour correspondre a notre schema
  mutate {
    rename => {
      "[host][name]" => "source_host"
    }
    remove_field => ["host"]
  }
}

Avant nettoyage, un événement Filebeat contient environ 15 champs de metadonnees. Apres, il n'a que les champs utiles. Ca economise du stockage et rend Kibana plus lisible.

Cas pratique : normaliser les donnees

Des logs de trois services différents arrivent avec des noms de champs différents :

# Service A : {"status": "200", "latency": "42"}
# Service B : {"code": 200, "response_time": 42.0}
# Service C : {"http_status": "200", "duration_ms": "42"}

Un mutate pour les normaliser :

filter {
  # Renommer les champs selon la source
  if [code] {
    mutate {
      rename => {
        "code" => "status_code"
        "response_time" => "duration_ms"
      }
    }
  } else if [http_status] {
    mutate {
      rename => {
        "http_status" => "status_code"
      }
    }
  } else if [status] {
    mutate {
      rename => {
        "status" => "status_code"
        "latency" => "duration_ms"
      }
    }
  }

  # Convertir en types corrects
  mutate {
    convert => {
      "status_code" => "integer"
      "duration_ms" => "float"
    }
  }
}

Apres ce filtre, les trois services produisent des événements avec les memes champs status_code (integer) et duration_ms (float). Les dashboards Kibana fonctionnent sans se soucier de la source.

Sur paltemps.fr, cette normalisation est la première étape apres le parsing. Quelle que soit la source, les champs ont le meme nom et le meme type a la sortie.

Résumé

• Mutate est le couteau suisse des transformations de champs
• Les opérations principales : rename, remove_field, add_field, convert, gsub, lowercase, uppercase, split, join, copy
• Dans un seul bloc mutate, les opérations suivent un ordre fixe (pas l'ordre d'écriture)
• Pour garantir l'ordre, utilise plusieurs blocs mutate
• Apres le parsing, nettoie les metadonnees inutiles (agent, ecs, input, log)
• Normalise les noms de champs entre différentes sources pour des dashboards coherents

---

Precedent : 10 - Le filtre Dissect | Suivant : 12 - Filtres Date et GeoIP

Sources

• Mutate Filter Plugin - elastic.co
• Elastic Common Schema (ECS) - elastic.co