Le blog tech de Nicolas Steinmetz (Time Series, IoT, Web, Ops, Data)
Il y a quelques temps et sachant que j’utilisais n8n pour automatiser la génération des brèves du BigData Hebdo, Mathias m’a demandé s’il était possible de faire la même chose entre n8n et Warp 10 qu’avec node-red et Warp 10.
La réponse est oui mais voyons comment faire cela.
Pour ceux qui ne connaissent pas n8n, c’est un clone open source (sous licence fair-code) à des services comme Zapier ou IFTTT. Il permet d’automatiser des processus via la création de workflows. Ces workflows sont composés d’étapes et d’actions. n8n dispose d’un grand nombre de connecteurs vers les différents services existants, des opérateurs génériques (faire un appel http, appliquer une fonction), des opérateurs logiques (si, etc), des opérateurs de transformation de données, etc. Chacun de ces éléments est implémenté via une node. A chaque étape du workflow, une node est instanciée puis paramétrée. Les nodes peuvent être reliées entre-elles et la sortie d’une node peut alimenter la suivante.
Le workflow se veut basique et va être le suivant :
Ce n’est pas le workflow le plus passionnant du monde, mais cela permet de faire deux appels à l’API HTTP de Warp 10 :
/api/v0/exec
; vu le code, j’aurais pu passer par /api/V0/fetch
mais cela me permet de tester l’exécution de code WarpScript./api/v0/update
pour insérer une donnée dans une série. Cela permet de tester le passage du token d’authentification via un header.Pour commencer le workflow, la donnée de départ est la valeur en pourcentage du métrique “CPU Idle” d’un de mes serveurs.
En WarpScript, cela donne:
'<readToken>' 'readToken' STORE
[ $readToken 'crnt-ovh.cpu.usage_idle' { "host" "crnt-d10-gitlab" "cpu" "cpu-total" } NOW -1 ] FETCH
Et la réponse :
[
[{
"c": "crnt-ovh.cpu.usage_idle",
"l": {
"host": "crnt-d10-gitlab",
"cpu": "cpu-total",
"source": "telegraf",
".app": "io.warp10.bootstrap"
},
"a": {},
"la": 0,
"v": [
[1634505650000000, 91.675025]
]
}]
]
n8n dispose d’une node HTTP Request, qui comme son nom l’indique permet de faire des requêtes HTTP vers un serveur distant. Toutefois, il n’est pas possible de passer notre code WarpScript directement dans l’appel HTTP. Il faut créer un objet avec le code WarpScript et passer ensuite l’objet créé et le nom de la propriété contenant le code WarpScript à la node HTTP Request.
Pour stocker le code WarpScript dans un objet, il faut utiliser la node Set. Une fois la node Set ajoutée dans le workflow, aller dans Parameters > Add Value > Type: String
Saisir:
En cliquant sur “Execute Node”, on peut valider la variable (la partie grisée étant mon token) :
On peut maintenant ajouter une node HTTP Request dans le workflow et la relier à la node Set nouvellement créée. Ainsi, la node HTTP Request aura directement accès au résultat de la node Set.
Pour les ajustements à faire :
http://url.de.votre.instance.warp.10/api/v0/exec
En cliquant sur “Execute Node”, le résultat de la requête est visible (la partie grisée étant un bout de mon token) :
On retrouve notre objet JSON mais il est imbriqué dans des Array Javascript, on va applanir tout ça et extraire le timestamp et la valeur du cpu via l’ajout de deux nodes Function que l’on relie à la node HTTP Request. La node Function permet d’exécuter du code javascript sur les données et de réaliser des transformations que l’on ne peut pas forcément faire avec les autres nodes. Cela n’étant pas le coeur du sujet, cela ne sera pas détaillé.
A l’issue des deux exécutions, les données sont réduites à ce qui suit :
[{
"ts": 1634503660000000,
"cpu": 93.219488
}]
La node IF ne sera pas détaillée non plus ; elle sert juste à introduire un semblant de logique dans le workflow. En l’occurence, si la valeur de “cpu” >= 90, alors le test est considéré comme vrai et faux sinon. Dans le cas où c’est faux, une node noOp a été ajoutée pour matérialiser la fin du workflow.
Dans le cas où le test est vrai (valeur de “cpu” >= 90), on veut alors insérer le timestamp et la valeur dans une autre série sur une instance Warp 10. Comme précédemment, cela va se faire en deux fois:
On ajoute une node Set, ensuite dans Parameters > Add Value > Type: String
Saisir:
{{$json["ts"]}}// n8n{} {{$json["cpu"]}}
Ce qui nous donne l’écran suivant :
On revient à l’écran précédent en cliquant sur la croix à droite et en exécutant la node, on obtient :
Ensuite, il faut ajouter une nouvelle node HTTP Request avec le paramétrage suivant :
http://url.de.votre.instance.warp.10/api/v0/update
En haut du menu de gauche, une section “Credentials” est apparue ; dans la liste déroulante, cliquer sur “Create new” et remplissez le formulaire de la façon suivante:
Revener ensuite dans votre node HTTP Request dont on peut lancer l’exécution et on obtient :
Si je vais ensuite voir le contenu de ma série n8n :
'<readToken>' 'readToken' STORE
[ $readToken 'n8n' {} NOW -100 ] FETCH
J’obtiens comme réponse :
[
[{
"c": "n8n",
"l": {
".app": "io.warp10.bootstrap"
},
"a": {},
"la": 0,
"v": [
[1634503660000000, 93.219488],
[1634502790000000, 94.808468],
[1634501690000000, 93.7751],
[1634501550000000, 91.741742],
[1634478300000000, 92.774711]
]
}]
]
Avec une entrée pour chaque exécution du workflow sous réserve d’avoir un “CPU idle” >= 90%.
En conclusion, nous pouvons retenir que :
Le workflow était très basique pour permettre de montrer rapidement cette intégration. Des workflows plus complexes et riches sont laissés à votre imagination :
CérénIT vient de finaliser la migration pour un de ses clients d’un socle InfluxDB/Chronograf/Kapacitor vers InfluxDB2. Ce billet est l’occasion de revenir sur la partie alerting et de la migration de Kapacitor vers des alertes dans InfluxDB2.
Dans le cadre du socle InfluxDB/Chronograf/Kapacitor, le fonctionnement était le suivant :
Avant d’envisager la migration InfluxDB2, un point de vocabulaire :
Avec la migration InfluxDB2, nous avons voulu maintenir le même mécanisme. Toutefois :
Heureusement, la documentation mentionne la possibilité de faire des “custom checks” et un billet très détaillé intitulé “InfluxDB’s Checks and Notifications System” permet de mieux comprendre ce qu’il est possible de faire et donne quelques exemples de code.
Dès lors, il s’agit de :
Pour se faire, nous allons nous appuyer sur les mécanismes mis à disposition par Influxdata, à savoir les fonctions monitor.check(), monitor.from() et monitor.notify() et les mécanismes induits.
C’est ce que nous allons voir maintenant :
Le cycle de vie d’une alerte est le suivant :
monitor.check()
en définissant les informations d’identification du check, le type de check que l’on utilise (threshold, deadman, custom), les différents seuils dont on a besoin, le message à envoyer au endpoint, les données issues de la requête flux.monitor.check()
va alors stocker l’ensemble de ces données dans un measurement statuses
dans le bucket _monitoring
et il s’arrête là.monitor.from()
prend le relais, regarde s’il y a de nouveaux status depuis sa dernière exécution et en fonction des règles de notifications qui ont été définies, il va passer le relais monitor.notify()
.monitor.notify()
enverra une notification si la règle est validée et il insérera une entrée dans le measurement notifications
du bucket _monitoring
Une première version des alertes ont été implémentées sur cette logique. Des dashboards ont été réalisés pour suivre les status et les notifications. Cela fonctionne, pas de soucis ou presque.
Il se peut qu’il y ait un délai entre le moment où l’insertion issue du monitor.check()
se fait et le moment où le monitor.from()
s’exécute. Si monitor.from()
fait sa requête avant l’insertion de données, alors l’alerte ne sera pas immédiatement levée. Elle sera levée à la prochaine exécution de la task, ce qui peut être problématique dans certains cas. Pour une tâche qui s’exécute toutes les minutes, cela ne se voit pas ou presque. Pour une tâche toutes les 5 minutes, ça commence à se voir.
Une version intermédiare de la task est alors née : une fois le monitor.check()
exécuté, nous faisons appel à monitor.notify()
pour envoyer le message vers le endpoint.
Avantage :
Inconvénients :
notifications
de la même façon que précédemment (d’où les pointillés) vu que les données insérées dans le measurement statuses
n’existent pas encore. On perd la visibilité sur les notifications envoyées (mais on a toujours le suivi des statuts ; nous supposons que si on a le statut, alors on sait si la notification a été envoyée)Une variante non essayée à ce stade : elle consiste à faire cette notification au plus tôt mais de conserver le mécanisme de monitor.from() + monitor.notify()
pour avoir le measurement notifications
correctement mise à jour. A voir si les alertes ne sont pas perturbées par ce double appel à monitor.notify()
. Dans le cas présent, c’est l’application métier qui envoie les alertes après que la task InfluxDB ait appelé son API HTTP. Si chaque monitor.notify()
en vient à lever une alerte, cela est sans impact pour l’utilisateur. En effet, une fois qu’une alerte est levée, elle est considérée comme levée tant qu’elle n’est pas acquittée. Donc même si la task provoque 2 appels, seul le premier lévera l’alerte et la seconde ne fera rien de plus.
Enfin dernière variante (testée) : s’affranchir complètement de monitor.notify()
pour faire directement appel à http.endpoint() et http.post() et faire complètement l’impasse sur le suivi dans notifications
.
Tout est une histoire de compromis.
En conclusion, nous pouvons retenir que :
_monitoring
et les measurements statuses
et notifications
.monitor.check()
et monitor.from()
peuvent conduire à des décalages de levées d’alertesCérénIT a été contacté pour mener l’audit d’une instance InfluxDB 1.8 OSS utilisée dans un projet IoT lié à l’énergie. L’audit avait plusieurs objectifs :
De l’audit, on notera que :
Avant d’aller plus loin, précisons un peu cette notion de shard et les notions liées pour bien appréhender le sujet :
Nous pouvons représenter la logique instance > database > shard(s) > tsm files de la façon suivante :
Par défaut, InfluxDB applique les shard duration suivantes en fonction des retention policy :
Retention policy | Default shard duration |
---|---|
<= 2 days | 1 hour |
<= 6 months | 1 day |
> 6 months | 7 days |
Source : InfluxData - Shard group duration
Dès lors, une base de données avec une retention policy infinie aura une shard duration de 7 jours. Ainsi, si cette base contient 10 ans d’hisorique (soit 10 * 52 semaines = 520 semaines), elle contiendra 520 shards.
Du coup, InfluxData recommande les valeurs suivantes (au moins en 1.x ; on peut supposer que cela reste valable en 2.x):
Retention policy | Default shard duration |
---|---|
<= 1 day | 6 hour |
<= 7 days | 1 day |
<= 3 months | 7 days |
> 3 months | 30 days |
infinite | >= 52 weeks |
Source : Shard group duration recommendations
Selon cette perspective, la base de données avec 10 ans d’historique ne contiendra plus 520 shards mais 10 shards en prenant une shard duration de 52 semaines. L’écart entre la valeur par défaut et la valeur recommandée est plus que significatif.
Pour bien dimensionner vos shard duration, InfluxData recommande :
Pourquoi nous en arrivons là ? C’est assez simple :
Dès lors, un nombre important de shards va augmenter d’autant plus la consommation mémoire et le nombre de fichiers ouverts pour manipuler les données associées.
Si on recoupe ces données avec les recommendations pour InfluxDB Entreprise, à savoir 30/40 bases par data nodes et 1.000 shards par node, le bon réglage des retention policy et des shard durations n’est pas à négliger pour la bonne santé de votre instance.
En outre, s’il est possible de mettre à jour la retention policy et la shard duration en 1.x, cela ne s’appliquera que pour les nouveaux shards. Les anciens shards ne seront pas “restructurés” en fonction des nouvelles valeurs.
Pour mettre à jour les shards existants, il faut :
Ultime question, la version 2.x OSS change-t-elle la donne sur le sujet :
report
et report-disk
.En conclusion, ce qu’il faut retenir :
Mise à jour : le client reporte les gains suivants post restructuration des bases pour le serveur de recette et production :
podman machine
est supporté nativement sur Linux et MacOS/Intel et en remote client sur Windows/Intel.docker compose xxx
). Pour Windows & OSX, il est fourni avec Docker Desktop.tar
de NodeJS directement (ou indirectement), il est judicieux de mettre à jour votre version de npm
et node
et de vérifier vos dépendances.first()
et last()
ainsi que les nouvelles fonctions timestamp_floor()
et timestamp_ceil()
pour gérer les arrondis inférieurs/supérieurs. Enfin, l’API HTTP accepte des paramètres liés au “Out Of Order”.kubernetes_manifest
permet de définir ses propres manifests et donc permet d’utiliser des CRD non disponibles dans le provider officiel. Plus pratique que de générer les manifests via des templates et de faire du kubectl apply
par dessus.time_bucket_ng
qui permet de faire de nouvelles aggrégations et d’avoir prochainement un support des timezones. La seconde fonctionnalité expérimentale porte sur la capacité à bouger de la donnée entre plusieurs nodes en mode cluster.