Flume NG

Dans tous les projets informatiques, la problématique de logs est récurrente : stratégie, format, volume, etc. En cas de problème, ces derniers représentent un des premiers points d’entrée dans une application :

• Actions réalisées par un utilisateur ;
• Erreurs affectant un utilisateur.

Cependant, soumises à de fortes volumétries, la collecte et l’exploitation des logs peut vite devenir difficile, voire impossible. Comment alors exploiter cette mine d’informations qui dorment dans tous ces fichiers ?

C’est pour répondre à cette problématique que Flume NG a été créé.

Qu’est ce que Flume NG ?

Flume NG (Next Generation) est un projet Apache et une solution de collecte et d’agrégation de gros volumes d’évènements. Ce dernier permet notamment de pousser ses logs dans un référentiel centralisé (Filesystem, HDFS, HBase, etc). Il a été également pensé pour être scalable.

Flume NG est une refonte complète de Flume OG (Old Generation) offrant des fonctionnalités plus avancées et une utilisation simplifiée.

Flume est composé d’agents possédant les caractéristiques suivantes :

 

Comment fonctionne un agent ?

• Une source consomme des évènements provenant de sources externes (syslog, autre agent Flume, etc) et place ces derniers dans un ou plusieurs channel(s) ;
• Un channel (base de données, en mémoire, etc.) permet de stocker temporairement les évènements attendant d’être consommés ;
• Un sink va ensuite pousser ces évènements vers la destination souhaitée (HDFS, Flume agent source, etc.).

Voici une liste non exhaustive de possibilités de configuration des agents:

Sources :

• Avro : écouter sur un port TCP. Utilisé également pour communiquer entre différents agents Flume ;
• Spooling Directory Source : lire des fichiers arrivant dans un répertoire ;
• Syslog (TCP ou UDP) : capter les évènements d’un server syslog. Il est, par exemple, possible de configurer Tomcat pour envoyer les logs vers syslog et Flume les prendra en charge ;
• HTTP : un handler se charge de traduire les requêtes POST et GET (encore en phase d’expérimentation).

Channels :

• Memory : garde les évènements en mémoire ;
• JDBC : les évènements sont stockés en base de données ;
• File : les évènements sont stockés sur un filesystem.

Sinks :

• HDFS : pousse les évènements dans Hadoop ;
• Avro : permet de communiquer avec un autre agent ;
• IRC : pousse les évènements sur un channel IRC ;
• HBase : pousse les évènements dans une base HBase ;
• ElasticSearch : pousse les évènements dans une base ElasticSearch ;
• File : enregistre les évènements sur un ficher local.

Il est également possible de chainer plusieurs agents Flume :

Installation et configuration

L’installation de Flume est simple :

• Avoir un JDK installé ;
• Dézipper Flume et le mettre dans un répertoire de votre choix (ex : /usr/local/flume) ;
• Avoir installé Hadoop si vous souhaitez pousser les logs dans un HDFS (ex: /usr/local/hadoop), ou, a minima, posséder en local le jar hadoop-core.

Voici un exemple de configuration avec un agent Flume écoutant un Syslog pour mettre les logs dans un HDFS (PATH_FLUME/conf/flume.conf) :

agent.sources = syslog    
agent.channels = memoryChannel    
agent.channels.memoryChannel.type = memory 
agent.channels.memoryChannel.capacity = 10000
    
agent.sources.syslog.type = syslogtcp    
agent.sources.syslog.port = 5140    
agent.sources.syslog.host = 127.0.0.1    
agent.sources.syslog.channels = memoryChannel

agent.sinks = HDFSEventSink    
agent.sinks.HDFSEventSink.channel = memoryChannel 
agent.sinks.HDFSEventSink.type = hdfs 
agent.sinks.HDFSEventSink.hdfs.path = hdfs://PATH_TO_YOUR_HDFS

Ensuite pour démarrer votre agent vous devez saisir la ligne de commande suivante au niveau de votre répertoire d’installation (PATH_FLUME) :

  flume-ng agent –conf ./conf -f conf/flume.conf –n agent 

Votre agent Flume écoute les évènements sur le port 5140 et les pousse dans HDFS.

Customiser Flume

Flume offre la possibilité d’implémenter nos propres sources, channels et sinks à partir d’interfaces Java. Il suffit ensuite de mettre les implémentations dans le classpath de Flume et de l’utiliser dans le fichier de configuration :

agent.sources=mySource     
agent.channels=memoryChannel
agent.sources.mySource.type=com.xebia.MyCustomSource
agent.sources.mySource.channels=memoryChannel

Autres composants

Flume permet d’ajouter de l’intelligence aux agents grâce aux composants suivants :

• Interceptors : ils peuvent modifier, filtrer les évènements en ajoutant des en-têtes (timestamp, hostname, etc) :

agent.sources=mySource
agent.channels=memoryChannel
agent.sources.mySource.interceptors=myInterceptor
agent.sources.mySource.interceptors.myInterceptor.type=host
agent.sources.mySource.interceptors.myInterceptor.hostHeader=hostname

Dans cet exemple, notre intercepteur va ajouter dans l’en-tête une clé hostname et la valeur qui est l’adresse IP de la machine courante. Dans une configuration complexe de Flume (plusieurs machines), cet intercepteur est très pratique pour savoir par quelles machines les logs ont été traitées.

• Channel selectors : permet d’aiguiller les évènements vers des channels spécifiques :

agent.sources = mySource
agent.channels = channel1, channel2, channel3, channel4
agent.sources.mySource.selector.type = multiplexing
agent.sources.mySource.selector.header = hostname
agent.sources.mySource.selector.mapping.host1 = channel2, channel4
agent.sources.mySource.selector.mapping.host2 = channel3
agent.sources.mySource.selector.mapping.default = channel1

Notre source analyse le header et si la clé est hostname et que sa valeur est host1, les évènements seront alors envoyés vers les channels 2 et 4. Lorsque le sélecteur n’arrive pas à trouver de règle correspondante, les évènements sont envoyés vers le channel par défaut (channel1).

• Sink processors : permet de définir un groupe de sinks et de définir des stratégies de load balancing ou failover :

agent.sinkgroups = myGroup
agent.sinkgroups.myGroup.sinks = sink1, sink2
agent.sinkgroups.myGroup.processor.type = load_balance
agent.sinkgroups.myGroup.processor.selector = round_robin

Dans cet exemple, les deux sinks déclarés recevront des évènements chacun leur tour (round_robin).

Reporting

Il est possible de pousser des métriques Flume vers Ganglia. Il suffit d’ajouter les options suivantes au démarrage de vos agents :

-Dflume.monitoring.type=ganglia –Dflume.monitoring.hosts=hostname1:port1, hostname2:port2

Les métriques sont également accessibles au format JSON :

{
 "typeName1.componentName1" : {"metric1" :  "metricValue1", "metric2" :  "metricValue2"}
 "typeName2.componentName2" : {"metric3" :  "metricValue3", "metric4" :  "metricValue4"}
}

Au démarrage il suffit d’ajouter les paramètres suivants :

-Dflume.monitoring.type=http –Dflume.monitoring.port=port

Exemple de configuration plus complexe

Dans cet exemple, les hosts 1 et 2 reçoivent des évènements de Syslog et doivent les transmettre à des machines distantes (hosts 3 et 4) qui se chargent de les pousser dans un HDFS. Afin d’ajouter de la scalabilité, nous souhaitons mettre en place un load balancing entre les hosts 3 et 4.

Voici les configurations nécessaires :

Host 1 et Host 2:

agent.sources = syslog     
agent.channels = memoryChannel     
agent.channels.memoryChannel.type = memory    
agent.channels.memoryChannel.capacity = 10000 

agent.sources.syslog.type = syslogtcp     
agent.sources.syslog.port = 5140     
agent.sources.syslog.host = 127.0.0.1     
agent.sources.syslog.channels = memoryChannel

agent.sinkgroups = myGroup
agent.sinkgroups.myGroup.sinks = avroSink1, avroSink2
agent.sinkgroups.myGroup.processor.type = load_balance
agent.sinkgroups.myGroup.processor.selector = round_robin

agent.sinks = avroSink1, avroSink2
agent.sinks.avroSink1.channel = memoryChannel 
agent.sinks.avroSink1.type = avro
agent.sinks.avroSink1.hostname = host3 
agent.sinks.avroSink1.port = 5141
 
agent.sinks.avroSink2.channel = memoryChannel 
agent.sinks.avroSink2.type = avro
agent.sinks.avroSink2.hostname = host4
agent.sinks.avroSink2.port = 5142

Host 3 et Host 4 :

agent.sources = avroCollection1, avroCollection2
agent.channels = memoryChannel     
agent.channels.memoryChannel.type = memory    
agent.channels.memoryChannel.capacity = 10000 

agent.sources.avroCollection1.channels = memoryChannel
agent.sources.avroCollection1.type = avro
agent.sources.avroCollection1.bind = 127.0.0.1
agent.sources.avroCollection1.port = 5141

agent.sources.avroCollection2.channels = memoryChannel
agent.sources.avroCollection2.type = avro
agent.sources.avroCollection2.bind = 127.0.0.1
agent.sources.avroCollection2.port = 5142

agent.sinks = HDFSEventSink     
agent.sinks.HDFSEventSink.channel = memoryChannel
agent.sinks.HDFSEventSink.type = hdfs
agent.sinks.HDFSEventSink.hdfs.path = hdfs://PATH_TO_YOUR_HDFS

Avec cette configuration, nous voyons que la mise en place de ce type d’architecture devient triviale.

Pour aller plus loin, il est possible d’ajouter des intercepteurs ou bien aiguiller nos logs avec des Channels selectors vers les host 3 ou 4 selon des règles définies (IP de la machine par exemple).

Conclusion

Flume NG propose une solution de collecte de logs performante. De par la richesse des composants proposés (sources, channels et sinks), Flume permet des interactions avec une multitude de technologies (HDFS, HBase, ElasticSearch, etc.). Ajouté à sa simplicité de mise en place et ses possibilités de scalabilité, Flume se place comme un outil incontournable.