Google utilise la technologie IA pour traduire le contenu dans votre langue préférée. Les traductions générées par IA peuvent contenir des erreurs.

Collecter les journaux Apache Hadoop

Version de l'analyseur : 3.0

Compatible avec :

Google SecOps SIEM

Ce document explique comment ingérer des journaux Apache Hadoop dans Google Security Operations à l'aide de l'agent Bindplane. L'analyseur extrait d'abord les champs des journaux Hadoop bruts à l'aide de modèles Grok basés sur les formats de journaux Hadoop courants. Ensuite, il mappe les champs extraits aux champs correspondants dans le schéma UDM (Unified Data Model), effectue des conversions de types de données et enrichit les données avec un contexte supplémentaire.

Avant de commencer

Assurez-vous de remplir les conditions suivantes :

Une instance Google SecOps
Un hôte Windows 2016 ou version ultérieure, ou Linux avec systemd
Si vous exécutez l'agent derrière un proxy, assurez-vous que les ports de pare-feu sont ouverts conformément aux exigences de l'agent BindPlane.
Accès privilégié aux fichiers de configuration du cluster Apache Hadoop

Obtenir des identifiants Google SecOps pour l'agent BindPlane

Pour authentifier l'agent Bindplane auprès de Google SecOps, choisissez l'une des méthodes suivantes :

Nouveaux déploiements : utilisez l'API Chronicle (option 1).
Anciens déploiements : continuez à utiliser le fichier d'authentification d'ingestion (option 2) jusqu'à l'arrêt de l'API Ingestion le 20 juillet 2027.

Option 1 : Créer un compte de service pour l'API Chronicle (recommandé)

Connectez-vous à la consoleGoogle Cloud et sélectionnez le projet Google Cloud associé à votre instance Google SecOps.
Accédez à IAM et administration > Comptes de service.
Cliquez sur Créer un compte de service.
Fournissez les informations de configuration suivantes :
- Nom du compte de service : saisissez un nom pour le compte (par exemple, bindplane-ingest).
- Description du compte de service : saisissez une description (par exemple, Bindplane agent log ingestion).
Cliquez sur Créer et continuer.
Dans la section Autoriser ce compte de service à accéder au projet, cliquez sur Sélectionner un rôle, puis sélectionnez Éditeur Chronicle.
Cliquez sur Continue (Continuer), puis sur Done (OK).
Cliquez sur le compte de service que vous avez créé, puis accédez à l'onglet Clés.
Cliquez sur Ajouter une clé > Créer une clé.
Sélectionnez JSON, puis cliquez sur Créer. Le fichier de clé est téléchargé sur votre ordinateur.
Enregistrez le fichier de clé sous le nom ingestion-auth.json dans le répertoire du collecteur Bindplane (/opt/observiq-otel-collector/ingestion-auth.json sous Linux ou C:\Program Files\observIQ OpenTelemetry Collector\ingestion-auth.json sous Windows), ou mettez à jour creds_file_path dans config.yaml pour qu'il corresponde au chemin d'accès où vous l'enregistrez.
Accédez à IAM et administration > Paramètres, puis copiez le numéro du projet. Vous en aurez besoin pour configurer l'exportateur.

Option 2 : Télécharger le fichier d'authentification d'ingestion (ancien)

Connectez-vous à la console Google SecOps.
Accédez à Paramètres > Gestion des données > Agents de collecte. Si vous utilisez une plate-forme SIEM uniquement, accédez à Paramètres > Agents de collecte.
Téléchargez le fichier d'authentification d'ingestion.
Enregistrez le fichier de manière sécurisée sur le système où l'agent BindPlane sera installé.

Obtenir le numéro client Google SecOps

Connectez-vous à la console Google SecOps.
Accédez à Paramètres > Administration > Profil. Si vous utilisez une plate-forme SIEM uniquement, accédez à Paramètres > Profil.
Copiez et enregistrez le numéro client de la section Détails de l'organisation.

Installer l'agent BindPlane

Installez l'agent BindPlane sur votre système d'exploitation Windows ou Linux en suivant les instructions ci-dessous.

Installation de Windows

Ouvrez l'invite de commande ou PowerShell en tant qu'administrateur.

Exécutez la commande suivante :

msiexec /i "https://github.com/observIQ/bindplane-agent/releases/latest/download/observiq-otel-collector.msi" /quiet

Installation sous Linux

Ouvrez un terminal avec des droits d'accès racine ou sudo.

Exécutez la commande suivante :

sudo sh -c "$(curl -fsSlL https://github.com/observiq/bindplane-agent/releases/latest/download/install_unix.sh)" install_unix.sh

Ressources d'installation supplémentaires

Pour connaître les autres options d'installation, consultez le guide d'installation de l'agent BindPlane.

Configurer l'agent Bindplane pour ingérer Syslog et l'envoyer à Google SecOps

Accédez au fichier de configuration :
- Trouvez le fichier config.yaml. En général, il se trouve dans le répertoire /opt/observiq-otel-collector/ sous Linux ou dans le répertoire d'installation sous Windows.
- Ouvrez le fichier à l'aide d'un éditeur de texte (par exemple, nano, vi ou Notepad).

Modifiez le fichier config.yaml comme suit :

receivers:
  udplog:
    # Replace the port and IP address as required
    listen_address: "0.0.0.0:514"

exporters:
  chronicle/chronicle_w_labels:
    protocol: https
    # Region of your instance, for example us, europe, asia-southeast1
    location: <LOCATION>
    # Project number of the Google Cloud project linked to the instance
    project: '<PROJECT_NUMBER>'
    compression: gzip
    # Adjust the path to the credentials file you downloaded in Step 1
    creds_file_path: '/path/to/ingestion-authentication-file.json'
    # Replace with your actual customer ID from Step 2
    customer_id: <CUSTOMER_ID>
    endpoint: chronicle.googleapis.com
    # Add optional ingestion labels for better organization
    log_type: 'HADOOP'
    raw_log_field: body
    ingestion_labels:

service:
  pipelines:
    logs/source0__chronicle_w_labels-0:
      receivers:
        - udplog
      exporters:
        - chronicle/chronicle_w_labels

Remplacez le port et l'adresse IP selon les besoins de votre infrastructure.
Remplacez <LOCATION> par la région de votre instance Google SecOps (par exemple, us, europe ou asia-southeast1). L'exportateur ajoute cette valeur en préfixe à chronicle.googleapis.com.
Remplacez <PROJECT_NUMBER> par le numéro de projet indiqué dans IAM et administration > Paramètres.
Remplacez <CUSTOMER_ID> par le numéro client réel.
Remplacez /path/to/ingestion-authentication-file.json par le chemin d'accès au fichier de clé du compte de service (option 1) enregistré à l'étape 1.

Ancien exportateur d'API d'ingestion

Si vous utilisez le fichier d'authentification d'ingestion de l'option 2, remplacez le bloc exporters par le code suivant. Définissez creds_file_path sur le chemin d'accès au fichier d'authentification d'ingestion. Si votre instance se trouve en dehors des États-Unis, remplacez malachiteingestion-pa.googleapis.com par votre point de terminaison régional (par exemple, europe-malachiteingestion-pa.googleapis.com).

exporters:
  chronicle/chronicle_w_labels:
    compression: gzip
    # Adjust the path to the credentials file you downloaded in Step 1
    creds_file_path: '/path/to/ingestion-authentication-file.json'
    # Replace with your actual customer ID from Step 2
    customer_id: <CUSTOMER_ID>
    endpoint: malachiteingestion-pa.googleapis.com
    # Add optional ingestion labels for better organization
    log_type: 'HADOOP'
    raw_log_field: body
    ingestion_labels:

Redémarrez l'agent BindPlane pour appliquer les modifications.

Pour redémarrer l'agent BindPlane sous Linux, exécutez la commande suivante :
```
sudo systemctl restart observiq-otel-collector
```
Pour redémarrer l'agent BindPlane sous Windows, vous pouvez utiliser la console Services ou saisir la commande suivante :
```
net stop observiq-otel-collector && net start observiq-otel-collector
```

Configurer le transfert Syslog sur Apache Hadoop

Apache Hadoop utilise Log4j pour la journalisation. Configurez l'appender Syslog approprié en fonction de votre version de Log4j afin que les daemons Hadoop (NameNode, DataNode, ResourceManager, NodeManager, etc.) transfèrent les journaux directement à votre récepteur syslog (hôte Bindplane). Log4j est configuré via des fichiers (pas d'UI Web).

Option 1 : Configuration de Log4j 1.x

Recherchez le fichier log4j.properties (généralement dans $HADOOP_CONF_DIR/log4j.properties).

Ajoutez la configuration SyslogAppender suivante au fichier :

# Syslog appender (UDP example)
log4j.appender.SYSLOG=org.apache.log4j.net.SyslogAppender
log4j.appender.SYSLOG.SyslogHost=<BINDPLANE_HOST_IP>:514
log4j.appender.SYSLOG.Facility=LOCAL0
log4j.appender.SYSLOG.FacilityPrinting=true
log4j.appender.SYSLOG.layout=org.apache.log4j.PatternLayout
log4j.appender.SYSLOG.layout.ConversionPattern=%d{ISO8601} level=%p logger=%c thread=%t msg=%m%n

# Example: send NameNode logs to syslog
log4j.logger.org.apache.hadoop.hdfs.server.namenode=INFO,SYSLOG
log4j.additivity.org.apache.hadoop.hdfs.server.namenode=false

# Or attach to root logger to send all Hadoop logs
# log4j.rootLogger=INFO, SYSLOG

Remplacez <BINDPLANE_HOST_IP> par l'adresse IP de votre hôte BindPlane.
Enregistrez le fichier.
Redémarrez les daemons Hadoop pour appliquer les modifications de configuration.

Option 2 : Configuration de Log4j 2.x

Recherchez le fichier log4j2.xml (généralement dans $HADOOP_CONF_DIR/log4j2.xml).

Ajoutez la configuration Syslog appender suivante au fichier :

<Configuration status="WARN">
  <Appenders>
    <!-- UDP example; for TCP use protocol="TCP" -->
    <Syslog name="SYSLOG" format="RFC 5424"
            host="<BINDPLANE_HOST_IP>"
            port="514"
            protocol="UDP"
            facility="LOCAL0"
            appName="hadoop"
            enterpriseNumber="18060"
            mdcId="mdc">
      <PatternLayout pattern="%d{ISO8601} level=%p logger=%c thread=%t msg=%m %X%n"/>
    </Syslog>
  </Appenders>

  <Loggers>
    <!-- Send NameNode logs to syslog -->
    <Logger name="org.apache.hadoop.hdfs.server.namenode" level="info" additivity="false">
      <AppenderRef ref="SYSLOG"/>
    </Logger>

    <!-- Or send all Hadoop logs -->
    <Root level="info">
      <AppenderRef ref="SYSLOG"/>
    </Root>
  </Loggers>
</Configuration>

Remplacez <BINDPLANE_HOST_IP> par l'adresse IP de votre hôte BindPlane.
Enregistrez le fichier.
Redémarrez les daemons Hadoop pour appliquer les modifications de configuration.

Table de mappage UDM

Champ du journal	Mappage UDM	Logique
Autorisés	security_result.action	Si la valeur est "false", l'action est "BLOCK". Si la valeur est "true", l'action est "ALLOW".
auth_type	additional.fields.key = "auth_type", additional.fields.value.string_value	Extrait du champ "ugi" à l'aide du modèle grok "%{DATA:suser}@.*auth:%{WORD:auth_type}". Les parenthèses et "auth:" sont supprimés.
call	additional.fields.key = "Call#", additional.fields.value.string_value	Mappage direct.
call_context	additional.fields.key = "callerContext", additional.fields.value.string_value	Mappage direct.
cliIP	principal.ip	Mappé uniquement lorsque le champ "json_data" existe et est analysé avec succès au format JSON.
cmd	principal.process.command_line	Mappage direct.
cluster_name	target.hostname	Utilisé comme nom d'hôte cible, le cas échéant.
jour	metadata.event_timestamp.seconds	Utilisé avec le mois, l'année, les heures, les minutes et les secondes pour construire event_timestamp.
description	metadata.description	Mappage direct.
conducteur	additional.fields.key = "driver", additional.fields.value.string_value	Mappage direct.
dst	target.ip OR target.hostname OR target.file.full_path	Si l'analyse en tant qu'adresse IP réussit, l'adresse IP est mappée à l'adresse IP cible. Si la valeur commence par "/user", elle est mappée au chemin d'accès du fichier cible. Sinon, mappé sur le nom d'hôte cible.
dstport	target.port	Mappé directement et converti en entier.
enforcer	security_result.rule_name	Mappage direct.
event_count	additional.fields.key = "event_count", additional.fields.value.string_value	Mappé directement et converti en chaîne.
fname	src.file.full_path	Mappage direct.
heures	metadata.event_timestamp.seconds	Utilisé avec le mois, le jour, l'année, les minutes et les secondes pour construire event_timestamp.
id	additional.fields.key = "id", additional.fields.value.string_value	Mappage direct.
ip	principal.ip	Mappé sur l'adresse IP principale après suppression de tout caractère "/" de début.
json_data		Analysé au format JSON. Les champs extraits sont mappés aux champs UDM correspondants.
logType	additional.fields.key = "logType", additional.fields.value.string_value	Mappage direct.
message		Utilisé pour extraire différents champs à l'aide de modèles grok.
method	network.http.method	Mappage direct.
minutes	metadata.event_timestamp.seconds	Utilisé avec le mois, le jour, l'année, les heures et les secondes pour construire event_timestamp.
mois	metadata.event_timestamp.seconds	Utilisé avec le jour, l'année, les heures, les minutes et les secondes pour construire event_timestamp.
observateur	observer.hostname OU observer.ip	Si l'analyse en tant qu'adresse IP réussit, mappé à l'adresse IP de l'observateur. Sinon, mappé sur le nom d'hôte de l'observateur.
perm	additional.fields.key = "perm", additional.fields.value.string_value	Mappage direct.
stratégie	security_result.rule_id	Mappé directement et converti en chaîne.
produit	metadata.product_name	Mappage direct.
product_event	metadata.product_event_type	Mappage direct. Si la valeur est "rename", le champ "dst" est mappé sur "target.file.full_path".
proto	network.application_protocol	Mappé directement et converti en majuscules s'il ne s'agit pas de "webhdfs".
reason	security_result.summary	Mappage direct.
dépôt	additional.fields.key = "repo", additional.fields.value.string_value	Mappage direct.
resType	additional.fields.key = "resType", additional.fields.value.string_value	Mappage direct.
résultat	additional.fields.key = "result", additional.fields.value.string_value	Mappé directement et converti en chaîne.
Réessayer	additional.fields.key = "Retry#", additional.fields.value.string_value	Mappage direct.
seconds	metadata.event_timestamp.seconds	Utilisé avec le mois, le jour, l'année, les heures et les minutes pour créer event_timestamp.
seq_num	additional.fields.key = "seq_num", additional.fields.value.string_value	Mappé directement et converti en chaîne.
de gravité,	security_result.severity	Mappé à différents niveaux de gravité en fonction de la valeur : "INFO", "Info", "info" -> "INFORMATIONAL" ; "Low", "low", "LOW" -> "LOW" ; "error", "Error", "WARN", "Warn" -> "MEDIUM" ; "High", "high", "HIGH" -> "HIGH" ; "Critical", "critical", "CRITICAL" -> "CRITICAL".
shost	principal.hostname	Utilisé comme nom d'hôte principal s'il est différent de "src".
src	principal.ip OR principal.hostname OR observer.ip	Si l'analyse en tant qu'adresse IP réussit, mappé à l'adresse IP du compte principal et de l'observateur. Sinon, mappé sur le nom d'hôte principal.
srcport	principal.port	Mappé directement et converti en entier.
résumé	security_result.summary	Mappage direct.
suser	principal.user.userid	Mappage direct.
tags	additional.fields.key = "tags", additional.fields.value.string_value	Mappage direct.
fil de discussion	additional.fields.key = "thread", additional.fields.value.string_value	Mappage direct.
tip	target.ip	Mappage direct.
ugi	target.hostname	Utilisé comme nom d'hôte cible si le champ"log_data" ne contient pas "·".
url	target.url	Mappage direct.
vendor	metadata.vendor_name	Mappage direct.
version	metadata.product_version	Mappage direct.
an	metadata.event_timestamp.seconds	Utilisé avec le mois, le jour, les heures, les minutes et les secondes pour construire event_timestamp.
N/A	metadata.event_type	La valeur par défaut est "NETWORK_CONNECTION". La valeur est remplacée par "STATUS_UPDATE" si aucune cible n'est identifiée.
N/A	metadata.log_type	Définissez cette option sur "HADOOP".
N/A	security_result.alert_state	Définissez cette option sur "ALERTING" si la gravité est "HIGH" ou "CRITICAL".
N/A	is_alert	Définie sur "true" si la gravité est "HIGH" ou "CRITICAL".
N/A	is_significant	Définie sur "true" si la gravité est "HIGH" ou "CRITICAL".

Journal des modifications

Afficher le journal des modifications de cet analyseur

Vous avez encore besoin d'aide ? Obtenez des réponses de membres de la communauté et de professionnels Google SecOps.