Migration de données de Splunk vers Elastic

Approche de migration

• Coût : Splunk est souvent considéré comme un outil coûteux, notamment pour de grands volumes de données. ELK, quant à lui, est open source et propose une option gratuite ainsi que des options commerciales plus abordables, ce qui peut être attractif pour les organisations cherchant à réduire leurs coûts.

• Flexibilité et personnalisation : Tout comme Splunk, ELK offre une grande flexibilité et une personnalisation poussée, permettant aux utilisateurs de créer des pipelines de traitement des données personnalisés avec Logstash, des visualisations sur mesure avec Kibana, et de s'adapter aux besoins spécifiques de leur infrastructure et de leurs cas d'utilisation.

• Scalabilité : ELK est bien connu pour sa scalabilité horizontale, ce qui signifie qu'il peut facilement être dimensionné pour gérer de grands volumes de données en ajoutant simplement de nouveaux nœuds au cluster Elasticsearch. Cela peut être particulièrement avantageux pour les environnements de grande échelle ou en constante croissance.

• Intégration avec l'écosystème open source : ELK s'intègre bien avec d'autres outils et technologies open source, permettant aux utilisateurs de créer des solutions complètes en combinant ELK avec d'autres composants open source, tels que Beats pour la collecte de données, ou des outils d'orchestration comme Kubernetes.

• Communauté et support : ELK bénéficie d'une communauté open source active et d'un large éventail de ressources en ligne, ce qui facilite l'apprentissage et la résolution de problèmes. De plus, Elastic, la société derrière ELK, propose également un support commercial pour les entreprises qui ont besoin d'une assistance supplémentaire.

• Fonctionnalités avancées : Bien que Splunk propose de nombreuses fonctionnalités avancées pour l'analyse et la visualisation des données, ELK continue d'étendre ses capacités avec des modules supplémentaires tels que Elastic SIEM, Elastic Observability et Machine Learning (détection d'anomalies et assistant IA). Il est également à noter que le modèle de licence d'ELK inclut toutes ces fonctionnalités dans la licence Enterprise par défaut, contrairement à Splunk, où tout est un « add-on » payant en plus de la licence de base.

La migration de Splunk vers ELK implique principalement la gestion des données. Cela peut être divisé en deux composants :

Données historiques : Généralement utilisées pour mener des enquêtes sur des cas d'utilisation spécifiques, tels que la fraude ou la fidélisation des clients (CVM).

Données en temps réel : Destinées à permettre une prise de décision rapide pour des cas d'utilisation en temps réel.

Pour réussir la migration, ces deux aspects doivent être pris en compte. Voici une approche générale du processus :

• Inventaire des sources de données : Identifier les types de données, la fréquence et le volume d'ingestion aidera considérablement à planifier efficacement la migration.

• Migration des données en temps réel : Il existe deux approches pour cela :

  • Ingestion à partir de Splunk en utilisant des heavy forwarders.
  • Ingestion directe en utilisant des agents/beats Elastic. L'ingestion sur Splunk diminuera progressivement à mesure que la stabilité et la maturité de l'ingestion sur ELK s'amélioreront. Pendant cette transition, il y aura une « double exécution » pour l'ingestion en temps réel entre Splunk et ELK.

• Migration des données historiques : Stabiliser l'ingestion sur ELK sera une étape importante pour l'ingestion des données historiques. Seules les données ingérées avant cette étape (données plus anciennes) seront migrées.

• Démantèlement de Splunk : Cela se produira une fois la migration des données historiques et l'ingestion des données en temps réel terminées.

Comment extraire et migrer les données historiques de Splunk vers Elastic en 2024

Pour migrer les données historiques de Splunk vers ELK, vous devez exécuter la commande suivante dans le répertoire /opt/splunk/bin du search head de Splunk :

./splunk search "index=main earliest=04/22/2024:10:00:00 
latest=04/23/2024:10:26:28" -maxout 0 -output json> splunklogs-json.log 

• Index=main fait référence à l'index ingéré dans Splunk.

• Earliest= correspond à l'horodatage des données les plus anciennes.

• Latest= correspond à l'horodatage des données les plus récentes dans vos enregistrements historiques.

• Json indique le format des fichiers ; nous avons choisi JSON car il est compatible avec ELK.

• splunklogs-json.log est le fichier qui contiendra toutes les données historiques de l'index.

Ingestion avec Logstash pour la migration de Splunk vers Elastic : Guide étape par étape

Vous aurez ensuite besoin d’un serveur Logstash pour ingérer ces fichiers dans Elasticsearch.

La configuration de Logstash pour cette ingestion est la suivante :

input { 
file { 
    path => "/etc/logstash/splunklogs-json.log" 
    start_position => "beginning" 
    type => "json" 
  } 
} 
filter { 
  json { 
    source => "message" 
   } 
} 
output { 
  stdout {} 
  elasticsearch { 
    hosts => ["https://ip-elk:9200"] 
    ssl => true 
    cacert => "/etc/logstash/certs/elasticsearch-ca.pem" 
    user => "user" 
    password => "******" 
    manage_template => false 
    data_stream => true 
    data_stream_type => "logs" 
    ecs_compatibility => "v8" 
  } 
} 

Effectuer la découverte des données sur ELK après la migration depuis Splunk

Tous les champs étiquetés par Splunk seront récupérés depuis Elasticsearch. Vous pouvez même reformater les données avec Logstash, en modifiant la structure des champs à votre guise. Mais même sans reformatage, tous les champs Splunk sont directement ingérés avec le suffixe résultat. Ainsi, en attendant que ces données expirent, vous devrez légèrement adapter vos requêtes de découverte électronique (eDiscovery) et de sondage.

Ingestion des données en temps réel

L'ingestion des données en temps réel est également possible en utilisant Logstash en plus des heavy forwarders. Dans cette section, nous examinerons l'approche mentionnée en 2.a.

Voici un exemple de traces syslog.

Prérequis

Pour transmettre des données syslog de Splunk à ELK en temps réel, vous avez besoin d'une infrastructure Splunk avec des heavy forwarders. Ce sont des instances Splunk capables de filtrer, router et transformer les logs.

Pourquoi ne pas utiliser directement le search head de Splunk ?
La transmission des logs peut présenter un risque de congestion sur les files d'attente de Splunk s'il y a de la latence ou des erreurs de transmission, notamment dans le cas de transmissions via TCP. Dans ce cas, Splunk arrêtera la réception. Il est donc préférable d'éviter les transmissions sur les search heads de Splunk, qui doivent principalement être utilisés pour les requêtes de recherche.

Pourquoi ne pas utiliser directement le Splunk universal forwarder ?
Les universal forwarders n'ont pas la capacité de filtrer les logs avant la transmission. Ils envoient tout le lot ou rien du tout.

Pourquoi avez-vous besoin de filtrage ?
Le filtrage est nécessaire car, dans le cas de la transmission en temps réel, vos données n'auront pas les balises Splunk pour identifier les différents champs et sources de données. De plus, pour faciliter le formatage dans un format lisible et utilisable au niveau d'ELK, il est essentiel de transformer vos données. Étant donné que la structure des logs change selon la source, ce travail est indispensable.

Configurer les Heavy Forwarders (HF) pour une migration fluide

[tcpout] 
defaultGroup = splunk_group 
[tcpout:splunk_group] 
server = 192.168.0.92:9997 
disabled = 0 

[syslog] 
defaultGroup=syslogGroup 
[syslog:syslogGroup] 
server = 192.168.0.92:5144 
sendCookedData = false 
type = tcp  

Optimisation des requêtes de recherche sur ELK après la migration depuis Splunk

Le fichier pointe vers deux destinations. Le serveur Splunk, qui est notre groupe par défaut, où nous enverrons tous les logs reçus par le Heavy Forwarder (HF).

Et le second groupe syslog, où nous enverrons uniquement les traces syslog dans notre exemple.

La transmission entre le HF de Splunk et le search head de Splunk utilise une communication codée s2s.

Cependant, Logstash ne comprend pas ce codage, d'où la nécessité d'envoyer des données non codées avec le paramètre :

sendCookedData = false.

Le fichier /opt/splunk/etc/system/local/props.conf est le suivant :

[syslog] 
TRANSFORMS-routing = routeAll, send_to_syslog 

Il indique deux transformations à appliquer : routeAll et send_to_syslog pour les traces syslog, qui sont définies dans le fichier transforms.conf ci-dessous.

Dans le fichier : /opt/splunk/etc/system/local/transforms.conf, il est indiqué ce qui suit :

[routeAll] 
REGEX = (.) 
DEST_KEY = _TCP_ROUTING 
FORMAT = splunk_group

[send_to_syslog] 
REGEX = (.) 
DEST_KEY = _SYSLOG_ROUTING 
FORMAT = syslogGroup
routeAll Transformation 

input { 
  tcp { 
    port => 5144 
   type => "syslog" 
} 
filter { 
  if [type] == "syslog" { 
    grok { 
      match => { "message" => "<%{NUMBER:facility}> %{IPORHOST:hostname} %{GREEDYDATA:syslog_message}" }    
    } 
    date { 
      match => [ "timestamp", "MMM  d HH:mm:ss", "MMM dd HH:mm:ss" ] 
    } 
 } 
} 
output{ 
  elasticsearch { 
    parameter elastic(look previous example)    
 }  
  } 

Requêtes de recherche sur ELK

Cette transformation nous a permis de récupérer les journaux sur Elasticsearch dans le format montré ci-dessous.

Les quelques champs présents dans cette capture résultent du fait que le filtre grok utilisé n'était pas entièrement personnalisé. Pour exploiter pleinement ces journaux, il est nécessaire d'extraire le maximum de champs afin d'optimiser et de faciliter les requêtes.

C'est une tâche à réaliser pour chaque type de source.

Même au sein de syslog, le format des journaux change selon que le log provient d'un serveur Linux/Windows, d'une source Cisco ou d'une autre. Chaque syslog permet de suivre un événement particulier avec des champs variables. Par conséquent, vous devez avoir autant de filtres grok qu'il y a de sources d'événements.

De plus, n'oubliez pas le filtrage en amont que vous devez effectuer dans Splunk pour envoyer des journaux spécifiques d'une source vers un pipeline Logstash sur un port dédié. De cette manière, vous pouvez imposer la structure que vous souhaitez. Dans cet exemple, nous avons envoyé les traces syslog vers le pipeline Logstash sur le port 5144.

Conclusion

L'ingestion des journaux en temps réel de Splunk vers Elastic est tout à fait possible, mais elle implique une complexité considérable. Cela nécessite une personnalisation importante de l'architecture Splunk, avec des heavy forwarders (HFs) obligatoires pour le filtrage initial des types de sources. Ensuite, il faut préparer plusieurs pipelines d'ingestion dans Logstash pour chaque source, avec plusieurs filtres grok pour extraire le maximum de champs. Tout ce travail est destiné à être abandonné une fois que Splunk sera entièrement démantelé.

Par conséquent, nous recommandons de déployer des agents Beats sur les sources dès les premières étapes du projet de migration pour extraire directement les données et les indexer dans des formats natifs à Elasticsearch. Cela optimisera les requêtes et réduira le temps consacré aux transformations de Splunk vers ELK. Cette phase initiale sera marquée par une double exécution de Splunk et d'ELK jusqu'à ce que l'ingestion en temps réel soit stabilisée. Cette étape représentera par la suite le jalon le plus récent pour les données historiques. Ensuite, nous procéderons à l'extraction de toutes les anciennes données jusqu'au jalon défini, marquant le début d'une ingestion en temps réel stable sur ELK, en suivant la méthode mentionnée dans la section concernant l'ingestion historique de ce document.