Contrôle automatique de la contre-pression dans la messagerie des connecteurs OpenCTI

La gestion des connecteurs dans OpenCTI est cruciale pour maintenir une ingestion fluide des données et éviter les surcharges système, en particulier avec RabbitMQ. Avant la version 6.2.12, la gestion des intervalles d'exécution des connecteurs pouvait être complexe, peu intuitive et parfois sujette à des erreurs en raison des différents types d'intervalles en place. Désormais, une nouvelle fonctionnalité de planification a été introduite pour automatiser et simplifier ce processus pour tous les connecteurs « External Import ».

Cet article explore en détail cette nouvelle fonctionnalité, ses variables d'environnement associées et les avantages qu'elle offre.

Objectif du planificateur

Vue d'ensemble fonctionnelle

Flux de travail complet du planificateur

Le planificateur permet une gestion plus efficace de l'exécution des connecteurs. Lorsqu'un connecteur est planifié pour s'exécuter, le planificateur vérifie d'abord la taille des messages accumulés dans RabbitMQ avant de démarrer le processus du connecteur, sauf lors de la première exécution. Si cette taille dépasse la capacité du serveur définie par la variable d'environnement queue_threshold, le connecteur passe en mode « Buffering » et reporte son exécution selon la variable d'environnement duration_period.

Cette fonctionnalité est essentielle pour prévenir les surcharges qui peuvent se produire dans les files d'attente, permettant aux connecteurs d'ingérer les données de manière plus fluide sans intervention manuelle. De plus, l'interface OpenCTI affiche désormais des informations plus précises, telles que « Last run », « Next run » et « Server capacity », qui sont mises à jour toutes les 40 secondes via pingAlive.

Nouvelles variables d'environnement

Deux nouvelles variables d'environnement ont été introduites pour améliorer cette fonctionnalité :

• duration_period : Spécifie la durée de la période d'exécution en utilisant le format ISO 8601.
• queue_threshold : Définit le seuil de la file d'attente (en Mo) auquel les connecteurs passent en mode « Buffering ». Par défaut, ce seuil est fixé à 500 Mo.

Ces variables permettent d'ajuster le comportement du planificateur en fonction des besoins spécifiques de votre environnement.

À quels types de connecteurs cette fonctionnalité est-elle limitée ?

Cette nouvelle fonctionnalité s'applique exclusivement aux connecteurs « External Import ». Ces connecteurs nécessitent une planification régulière pour l'importation de données depuis des API externes, contrairement à d'autres types de connecteurs qui ne nécessitent pas de ré-exécution planifiée.

Si vous souhaitez en savoir plus sur les différents types de connecteurs, nous vous invitons à consulter notre documentation : https://docs.opencti.io/latest/deployment/connectors/

Pourquoi est-ce important ?

Cas d'usage principal

La nouvelle fonctionnalité de planification est particulièrement cruciale pour les administrateurs de plateforme, qui doivent surveiller de près l'ingestion de données pour chaque connecteur. En effet, un connecteur peut envoyer une quantité massive de données à OpenCTI dans un laps de temps très court. Si ce processus n'est pas bien géré, il peut entraîner des problèmes importants.

Cas d'usage n°1 : Initialement, l'équipe plateforme devait surveiller constamment les files d'attente et identifier manuellement les connecteurs problématiques. Pour éviter de surcharger les files d'attente dans RabbitMQ, il était nécessaire d'arrêter ces connecteurs (sans réinitialiser leur état), tout en permettant aux workers de traiter les données restantes. Cette approche, bien que fonctionnelle, était lourde car elle consommait beaucoup de temps et de ressources.

Cas d'usage n°2 : Si un connecteur a du mal à ingérer les données et n'est pas arrêté à temps, les files d'attente RabbitMQ continueront de se remplir si l'intervalle du connecteur est trop court, conduisant finalement à une saturation complète de la mémoire physique du système hôte, tombant dans l'un des pires scénarios possibles. Cette situation peut provoquer des erreurs et des plantages dans RabbitMQ, obligeant l'équipe à vider manuellement les files d'attente surchargées et à redémarrer RabbitMQ. Cette intervention entraîne la perte de toutes les données en attente et nécessite de réinitialiser l'état du connecteur. L'équipe doit ensuite restaurer manuellement l'état le plus proche possible, redémarrer le connecteur et surveiller de près son ingestion pour éviter d'autres surcharges.

Solution : Avec l'ajout du « Scheduler », les administrateurs de plateforme n'ont plus besoin de surveiller constamment tous les connecteurs. Les connecteurs peuvent désormais passer automatiquement en mode « Buffering » lorsque le seuil prédéfini est atteint. Cela permet d'économiser beaucoup de temps et de ressources tout en assurant une gestion plus efficace et sécurisée des files d'attente RabbitMQ, le tout sans intervention manuelle lourde et potentiellement risquée.

Comment cela fonctionne-t-il ?

La gestion des intervalles d'exécution des connecteurs a été standardisée et simplifiée grâce à deux méthodes distinctes :

schedule_iso() (avec breaking change)

La méthode schedule_iso() utilise le format ISO 8601, permettant une planification standardisée et précise des intervalles d'exécution des connecteurs, tout en offrant une meilleure lisibilité pour les utilisateurs. Par exemple, « P1D » représente une période de 1 jour, tandis que « PT24H » indique une durée de 24 heures. Cette approche est désormais la norme pour toutes les nouvelles intégrations de connecteurs, garantissant clarté et cohérence dans la gestion des intervalles d'exécution des connecteurs.

• schedule_iso() nécessite deux arguments pour fonctionner, message_callback et duration_period
• • message_callback : Équivalent à l'initialisation du processus du connecteur.
  • duration_period : Sa valeur doit être au format ISO 8601, et la variable d'environnement correspondante doit exister dans la configuration. Ce format reconnu internationalement garantit une interprétation cohérente des durées, facilitant la lecture des futures intégrations et minimisant les erreurs.

Exemple d'implémentation

Dans le config.yml ou docker-compose.yml :

Copied!

1# config.yml
2duration_period: "PT5H" # Add this variable
3queue_threshold: 600 # Added variable if different from default value (500 Mo)
Copy

Copied!

1# docker-compose.yml
2- CONNECTOR_DURATION_PERIOD=PT5H # Add this variable
3- CONNECTOR_QUEUE_THRESHOLD=600  # Added variable if different from default value (500 Mo)
Copy

Dans le code du connecteur :

Copied!

1# Get Connector config
2self.duration_period = get_config_variable(
3	"CONNECTOR_DURATION_PERIOD", ["connector", "duration_period"], config
4)
Copy

Copied!

1def run(self):
2	# Use schedule_iso() method
3	# In the message_callback, you must indicate the connector process, here self.starter
4	# New duration_period that takes into account the standardized ISO 8601 format
5	self.helper.schedule_iso(message_callback=self.starter, duration_period=self.duration_period) # duration_period: "PT5H" => 5 Hours
Copy

schedule_unit() (sans breaking change)

La méthode schedule_unit() vous permet de réutiliser l'intervalle de connecteur existant en spécifiant l'unité de temps, comme les heures ou les minutes. Cette approche facilite une transition progressive vers le nouveau système de planification sans impacter immédiatement les connecteurs ou les opérations en cours, bien que cette méthode soit destinée à être progressivement abandonnée au fil du temps.

• schedule_unit() nécessite trois arguments pour fonctionner, message_callback, duration_period et time_unit
• • message_callback : Équivalent à l'initialisation du processus du connecteur.
  • duration_period : Contrairement à schedule_iso(), le duration_period n'a pas besoin d'être dans les variables d'environnement car il utilise l'intervalle existant (ex : interval_hours=6), normalement, il s'agit d'une valeur numérique.
  • time_unit : Cet argument spécifie l'unité de temps utilisée par duration_period ; dans notre exemple, time_unit doit être défini sur « HOURS ».

Exemple d'implémentation

Contrairement à schedule_iso(), il n'est pas nécessaire de configurer config.yml ou docker-compose.yml car cette méthode utilisera la variable d'intervalle de connecteur existante.

Dans le code du connecteur :

Copied!

1def run(self):
2	# Use schedule_unit() method
3	# In the message_callback, you must indicate the connector process, here self.starter
4	# duration_period reuses the existing connector interval, here self.interval_hours
5	# time_unit for Retro-compatible - Enum Valid (YEARS, WEEKS, DAYS, HOURS, MINUTES, SECONDS)
6	self.helper.schedule_unit(message_callback=self.starter, duration_period=self.interval_hours, time_unit=self.helper.TimeUnit.HOURS)
Copy

Informations supplémentaires sur le planificateur

Auparavant, les connecteurs géraient les intervalles d'exécution en utilisant une boucle while true combinée à time.sleep(self.interval) directement dans le code, l'intervalle étant personnalisable dans la configuration. Désormais, tous ces éléments seront supprimés, et il en va de même pour l'implémentation de « Run & Terminate » au sein du connecteur ; cette gestion sera entièrement prise en charge par le « Scheduler ».

Affichage dans l'interface utilisateur

Comportement et affichage pour les connecteurs en mode « Run & Terminate »

Diagramme fonctionnel du planificateur en mode « Run & Terminate »

Lorsqu'un connecteur est configuré en mode « Run & Terminate », la valeur de « Next run » est définie par « External schedule ». Les organisations qui choisissent ce mode gèrent généralement les intervalles d'exécution en utilisant une planification externe, leur offrant une flexibilité dans la gestion des cycles d'exécution du connecteur.

Vue d'ensemble dans l'interface utilisateur lorsque le connecteur est en mode « Run & Terminate »

Note : Un cas spécial a été implémenté. Si le duration_period dans la configuration est défini sur, par exemple : 0, « 0 », « P0D », « PT0S », etc., le connecteur présentera le même comportement qu'un mode « Run & Terminate ».

Affichage des détails pour les connecteurs en mode « Buffering »

Lorsque la capacité de la file d'attente RabbitMQ dépasse le seuil défini (par exemple, si queue_message_size est à 9,90 Mo et queue_threshold est configuré à 8 Mo), le connecteur passe automatiquement en mode « Buffering ».

En mode « Buffering », l'exécution du connecteur est suspendue jusqu'à ce que la capacité de la file d'attente passe en dessous du seuil spécifié. L'interface utilisateur affiche des indicateurs visuels pour signaler ce changement d'état, notamment un message d'avertissement et un changement de couleur dans la section « Server Capacity ».

Vue d'ensemble des alertes visuelles dans l'interface utilisateur lorsque le connecteur est en mode « Buffering »

Affichage des détails pour les connecteurs n'utilisant pas le planificateur

Les connecteurs qui ne prennent pas en charge le planificateur afficheront « Not Provided » dans l'interface utilisateur.

Un connecteur qui n'utilise pas le « Scheduler » et n'a pas d'état

Note : Si un last_run est présent dans l'état et est au format timestamp, il sera automatiquement converti et affiché avec la notation « (from State) » à côté du titre « Last run », indiquant à l'utilisateur que cette information provient de l'état et non du planificateur.

Avec un timestamp Unix standard

Avec un timestamp à virgule flottante

Conclusion

La nouvelle fonctionnalité de planification (Scheduler) introduite dans OpenCTI version 6.2.12 offre une gestion plus standardisée, sécurisée et automatisée pour les connecteurs « External Import ». Elle réduit les risques associés aux surcharges de files d'attente RabbitMQ et aux interventions manuelles, libérant un temps précieux pour que les utilisateurs puissent se concentrer sur des tâches à plus forte valeur ajoutée.

Avec les méthodes schedule_iso() et schedule_unit(), les utilisateurs peuvent ajuster l'exécution des connecteurs en fonction des besoins spécifiques de leur environnement, tout en permettant une transition progressive vers la nouvelle fonctionnalité. Il est important de noter que la méthode schedule_iso() doit désormais être privilégiée, tandis que la méthode schedule_unit() n'est présente que pour faciliter la transition de l'absence de planificateur vers l'utilisation de schedule_iso().

Cette amélioration offre également une vue d'ensemble plus compréhensible et intuitive des processus d'exécution des connecteurs dans l'interface utilisateur avec « Last run », « Next run » et « Server Capacity ». Vous pouvez déjà tester cette nouvelle fonctionnalité avec le template « external-import » ou directement avec les connecteurs qui intègrent déjà le planificateur : CISA KEV, Mandiant, SEKOIA, AlienVault, Recorded Future, CrowdStrike (v6.3).

Nous espérons que cet article vous a aidé à mieux comprendre les avantages et les possibilités offertes par cette nouvelle fonctionnalité. Intéressé par plus de conseils et d'astuces sur OpenCTI ? Rejoignez notre communauté Slack et connectez-vous avec d'autres utilisateurs pour partager des idées et des solutions.