Apache Hadoop-Logs erfassen

Parserversion: 3.0

Unterstützt in:

In diesem Dokument wird beschrieben, wie Sie Apache Hadoop-Logs mit dem Bindplane-Agenten in Google Security Operations aufnehmen. Der Parser extrahiert zuerst Felder aus Roh-Hadoop-Logs mithilfe von Grok-Mustern, die auf gängigen Hadoop-Logformaten basieren. Anschließend werden die extrahierten Felder den entsprechenden Feldern im Schema für einheitliche Datenmodelle (Unified Data Model, UDM) zugeordnet, Datentypkonvertierungen durchgeführt und die Daten mit zusätzlichem Kontext angereichert.

Hinweis

Prüfen Sie, ob folgende Voraussetzungen erfüllt sind:

  • Eine Google SecOps-Instanz
  • Ein Windows 2016- oder höher- oder Linux-Host mit systemd
  • Wenn der Agent hinter einem Proxy ausgeführt wird, müssen die Firewallports gemäß den Anforderungen des Bindplane-Agents geöffnet sein.
  • Privilegierter Zugriff auf die Apache Hadoop-Clusterkonfigurationsdateien

Google SecOps-Anmeldedaten für den Bindplane-Agenten abrufen

Wählen Sie eine der folgenden Methoden aus, um den Bindplane-Agent bei Google SecOps zu authentifizieren:

  • Neue Deployments: Verwenden Sie die Chronicle API (Option 1).
  • Alte Bereitstellungen: Verwenden Sie die Datei für die Authentifizierung bei der Aufnahme (Option 2) weiterhin, bis die Ingestion API am 20. Juli 2027 eingestellt wird.
  1. Melden Sie sich in der Google Cloud Console an und wählen Sie das Google Cloud -Projekt aus, das mit Ihrer Google SecOps-Instanz verknüpft ist.
  2. Rufen Sie IAM und Verwaltung > Dienstkonten auf.
  3. Klicken Sie auf Dienstkonto erstellen.
  4. Geben Sie die folgenden Konfigurationsdetails an:
    • Name des Dienstkontos: Geben Sie einen Namen für das Konto ein, z. B. bindplane-ingest.
    • Dienstkontobeschreibung: Geben Sie eine Beschreibung ein (z. B. Bindplane agent log ingestion).
  5. Klicken Sie auf Erstellen und fortfahren.
  6. Klicken Sie im Abschnitt Diesem Dienstkonto Zugriff auf das Projekt erteilen auf Rolle auswählen und wählen Sie Chronicle-Bearbeiter aus.
  7. Klicken Sie auf Weiter und dann auf Fertig.
  8. Klicken Sie auf das von Ihnen erstellte Dienstkonto und rufen Sie den Tab Schlüssel auf.
  9. Klicken Sie auf Schlüssel hinzufügen > Neuen Schlüssel erstellen.
  10. Wählen Sie JSON aus und klicken Sie auf Erstellen. Die Schlüsseldatei wird auf Ihren Computer heruntergeladen.
  11. Speichern Sie die Schlüsseldatei als ingestion-auth.json im Bindplane-Collector-Verzeichnis (/opt/observiq-otel-collector/ingestion-auth.json unter Linux oder C:\Program Files\observIQ OpenTelemetry Collector\ingestion-auth.json unter Windows) oder aktualisieren Sie creds_file_path in config.yaml, damit der Pfad mit dem Pfad übereinstimmt, in dem Sie die Datei speichern.
  12. Rufen Sie IAM und Verwaltung > Einstellungen auf und kopieren Sie die Projektnummer. Sie benötigen sie für die Exporter-Konfiguration.

Option 2: Authentifizierungsdatei für die Datenaufnahme herunterladen (Legacy)

  1. Melden Sie sich in der Google SecOps-Konsole an.
  2. Rufen Sie die SIEM-Einstellungen > Erfassungs-Agents auf.
  3. Laden Sie die Authentifizierungsdatei für die Datenaufnahme herunter.
  4. Speichern Sie die Datei sicher auf dem System, auf dem der Bindplane-Agent installiert werden soll.

Google SecOps-Kundennummer abrufen

  1. Melden Sie sich in der Google SecOps-Konsole an.
  2. Rufen Sie die SIEM-Einstellungen > Profil auf.
  3. Kopieren und speichern Sie die Kundennummer aus dem Bereich Organization Details (Organisationsdetails).

BindPlane-Agent installieren

Installieren Sie den Bindplane-Agenten gemäß der folgenden Anleitung auf Ihrem Windows- oder Linux-Betriebssystem.

Windows-Installation

  1. Öffnen Sie die Eingabeaufforderung oder PowerShell als Administrator.

  2. Führen Sie dazu diesen Befehl aus:

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

Linux-Installation

  1. Öffnen Sie ein Terminal mit Root- oder Sudo-Berechtigungen.

  2. Führen Sie dazu diesen Befehl aus:

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

Zusätzliche Installationsressourcen

Weitere Installationsoptionen finden Sie im Installationsleitfaden für den Bindplane-Agent.

Bindplane-Agenten für die Aufnahme von Syslog-Daten und das Senden an Google SecOps konfigurieren

  1. Rufen Sie die Konfigurationsdatei auf:

    • Suchen Sie die Datei config.yaml. Normalerweise befindet sie sich unter Linux im Verzeichnis /opt/observiq-otel-collector/ oder unter Windows im Installationsverzeichnis.
    • Öffnen Sie die Datei mit einem Texteditor (z. B. nano, vi oder Notepad).

  2. Bearbeiten Sie die Datei config.yamlso:

    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
    • Ersetzen Sie den Port und die IP-Adresse nach Bedarf in Ihrer Infrastruktur.
    • Ersetzen Sie <LOCATION> durch die Region Ihrer Google SecOps-Instanz (z. B. us, europe oder asia-southeast1). Der Exporteur stellt diesem Wert das Präfix chronicle.googleapis.com voran.
    • Ersetzen Sie <PROJECT_NUMBER> durch die Projektnummer unter IAM & Verwaltung > Einstellungen.
    • Ersetzen Sie <CUSTOMER_ID> durch die tatsächliche Kundennummer.
    • Aktualisieren Sie /path/to/ingestion-authentication-file.json auf den Dateipfad, unter dem die Dienstkontoschlüsseldatei (Option 1) in Schritt 1 gespeichert wurde.

Legacy-Ingestion-API-Exporter

  • Wenn Sie die Authentifizierungsdatei für die Datenaufnahme aus Option 2 verwenden, ersetzen Sie den Block exporters durch Folgendes. Legen Sie für creds_file_path den Pfad zur Authentifizierungsdatei für die Datenaufnahme fest. Wenn sich Ihre Instanz außerhalb der USA befindet, ersetzen Sie malachiteingestion-pa.googleapis.com durch Ihren regionalen Endpunkt (z. B. 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:

Starten Sie den Bindplane-Agent neu, um die Änderungen zu übernehmen.

  1. Führen Sie den folgenden Befehl aus, um den Bindplane-Agenten unter Linux neu zu starten:

    sudo systemctl restart observiq-otel-collector

  2. Um den Bindplane-Agent unter Windows neu zu starten, können Sie entweder die Konsole Dienste verwenden oder den folgenden Befehl eingeben:

    net stop observiq-otel-collector && net start observiq-otel-collector

Syslog-Weiterleitung in Apache Hadoop konfigurieren

Apache Hadoop verwendet Log4j für das Logging. Konfigurieren Sie den entsprechenden Syslog-Appender basierend auf Ihrer Log4j-Version, damit Hadoop-Daemons (NameNode, DataNode, ResourceManager, NodeManager usw.) Logs direkt an Ihren Syslog-Empfänger (Bindplane-Host) weiterleiten. Log4j wird über Dateien konfiguriert (keine Web-UI).

Option 1: Log4j 1.x-Konfiguration

  1. Suchen Sie die Datei log4j.properties (normalerweise in $HADOOP_CONF_DIR/log4j.properties).

  2. Fügen Sie der Datei die folgende SyslogAppender-Konfiguration hinzu:

    # 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

  3. Ersetzen Sie <BINDPLANE_HOST_IP> durch die IP-Adresse Ihres Bindplane-Hosts.

  4. Speichern Sie die Datei.

  5. Starten Sie die Hadoop-Daemons neu, um die Konfigurationsänderungen zu übernehmen.

Option 2: Log4j 2.x-Konfiguration

  1. Suchen Sie die Datei log4j2.xml (normalerweise in $HADOOP_CONF_DIR/log4j2.xml).

  2. Fügen Sie der Datei die folgende Syslog-Appender-Konfiguration hinzu:

    <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>

  3. Ersetzen Sie <BINDPLANE_HOST_IP> durch die IP-Adresse Ihres Bindplane-Hosts.

  4. Speichern Sie die Datei.

  5. Starten Sie die Hadoop-Daemons neu, um die Konfigurationsänderungen zu übernehmen.

UDM-Zuordnungstabelle

Logfeld UDM-Zuordnung Logik
zulässig security_result.action Wenn „false“, ist die Aktion „BLOCK“. Wenn „true“, ist die Aktion „ALLOW“.
auth_type additional.fields.key = "auth_type", additional.fields.value.string_value Aus dem Feld „ugi“ mit dem Grok-Muster „%{DATA:suser}@.*auth:%{WORD:auth_type}“ extrahiert. Klammern und „auth:“ werden entfernt.
Anruf additional.fields.key = "Call#", additional.fields.value.string_value Direkt zugeordnet.
call_context additional.fields.key = "callerContext", additional.fields.value.string_value Direkt zugeordnet.
cliIP principal.ip Wird nur zugeordnet, wenn das Feld „json_data“ vorhanden ist und erfolgreich als JSON geparst wurde.
CMD principal.process.command_line Direkt zugeordnet.
cluster_name target.hostname Wird als Zielhostname verwendet, falls vorhanden.
Tag metadata.event_timestamp.seconds Wird zusammen mit Monat, Jahr, Stunden, Minuten und Sekunden verwendet, um „event_timestamp“ zu erstellen.
Beschreibung metadata.description Direkt zugeordnet.
Fahrer additional.fields.key = "driver", additional.fields.value.string_value Direkt zugeordnet.
dst target.ip ODER target.hostname ODER target.file.full_path Wenn das Parsen als IP-Adresse erfolgreich war, wird sie der Ziel-IP-Adresse zugeordnet. Wenn der Wert mit „/user“ beginnt, wird er dem Zieldateipfad zugeordnet. Andernfalls wird der Ziel-Hostname zugeordnet.
dstport target.port Direkt zugeordnet und in Ganzzahl konvertiert.
enforcer security_result.rule_name Direkt zugeordnet.
event_count additional.fields.key = "event_count", additional.fields.value.string_value Direkt zugeordnet und in einen String konvertiert.
fname src.file.full_path Direkt zugeordnet.
Stunden metadata.event_timestamp.seconds Wird zusammen mit Monat, Tag, Jahr, Minuten und Sekunden verwendet, um „event_timestamp“ zu erstellen.
id additional.fields.key = "id", additional.fields.value.string_value Direkt zugeordnet.
ip principal.ip Wird dem Hauptkonto-IP zugeordnet, nachdem alle vorangestellten „/“-Zeichen entfernt wurden.
json_data Als JSON geparst. Extrahierte Felder werden den entsprechenden UDM-Feldern zugeordnet.
logType additional.fields.key = "logType", additional.fields.value.string_value Direkt zugeordnet.
Nachricht Wird verwendet, um verschiedene Felder mithilfe von Grok-Mustern zu extrahieren.
method network.http.method Direkt zugeordnet.
Minuten metadata.event_timestamp.seconds Wird zusammen mit Monat, Tag, Jahr, Stunden und Sekunden verwendet, um „event_timestamp“ zu erstellen.
Monat metadata.event_timestamp.seconds Wird mit Tag, Jahr, Stunden, Minuten und Sekunden verwendet, um „event_timestamp“ zu erstellen.
Beobachter observer.hostname ODER observer.ip Wenn die IP-Adresse erfolgreich geparst wurde, wird sie der IP-Adresse des Beobachters zugeordnet. Andernfalls wird der Hostname des Beobachters zugeordnet.
perm additional.fields.key = "perm", additional.fields.value.string_value Direkt zugeordnet.
Richtlinie security_result.rule_id Direkt zugeordnet und in String konvertiert.
Produkt metadata.product_name Direkt zugeordnet.
product_event metadata.product_event_type Direkt zugeordnet. Wenn „rename“, wird das Feld „dst“ dem Feld „target.file.full_path“ zugeordnet.
Proto network.application_protocol Direkt zugeordnet und in Großbuchstaben umgewandelt, wenn es sich nicht um „webhdfs“ handelt.
reason security_result.summary Direkt zugeordnet.
Repository additional.fields.key = "repo", additional.fields.value.string_value Direkt zugeordnet.
resType additional.fields.key = "resType", additional.fields.value.string_value Direkt zugeordnet.
Ergebnis additional.fields.key = "result", additional.fields.value.string_value Direkt zugeordnet und in einen String konvertiert.
Wiederholen additional.fields.key = "Retry#", additional.fields.value.string_value Direkt zugeordnet.
Sekunden metadata.event_timestamp.seconds Wird zusammen mit Monat, Tag, Jahr, Stunden und Minuten verwendet, um „event_timestamp“ zu erstellen.
seq_num additional.fields.key = "seq_num", additional.fields.value.string_value Direkt zugeordnet und in String konvertiert.
die Ausprägung security_result.severity Wird je nach Wert verschiedenen Schweregraden zugeordnet: „INFO“, „Info“, „info“ -> „INFORMATIONAL“; „Low“, „low“, „LOW“ -> „LOW“; „error“, „Error“, „WARN“, „Warn“ -> „MEDIUM“; „High“, „high“, „HIGH“ -> „HIGH“; „Critical“, „critical“, „CRITICAL“ -> „CRITICAL“.
shost principal.hostname Wird als Haupt-Hostname verwendet, wenn er sich von „src“ unterscheidet.
src principal.ip ODER principal.hostname ODER observer.ip Wenn die IP-Adresse erfolgreich geparst wurde, wird sie dem Hauptkonto und der Beobachter-IP zugeordnet. Andernfalls wird sie dem primären Hostnamen zugeordnet.
srcport principal.port Direkt zugeordnet und in eine Ganzzahl konvertiert.
Zusammenfassung security_result.summary Direkt zugeordnet.
suser principal.user.userid Direkt zugeordnet.
Tags additional.fields.key = "tags", additional.fields.value.string_value Direkt zugeordnet.
Thread additional.fields.key = "thread", additional.fields.value.string_value Direkt zugeordnet.
Tipp target.ip Direkt zugeordnet.
ugi target.hostname Wird als Zielhostname verwendet, wenn das Feld „log_data“ kein „·“ enthält.
URL target.url Direkt zugeordnet.
vendor metadata.vendor_name Direkt zugeordnet.
Version metadata.product_version Direkt zugeordnet.
Jahr metadata.event_timestamp.seconds Wird zusammen mit Monat, Tag, Stunde, Minute und Sekunde verwendet, um „event_timestamp“ zu erstellen.
– metadata.event_type Standardmäßig auf „NETWORK_CONNECTION“ festgelegt. Wird in „STATUS_UPDATE“ geändert, wenn kein Ziel angegeben ist.
– metadata.log_type Legen Sie diesen Wert auf „HADOOP“ fest.
– security_result.alert_state Auf „ALERTING“ festgelegt, wenn der Schweregrad „HIGH“ oder „CRITICAL“ ist.
– is_alert Wird auf „true“ gesetzt, wenn der Schweregrad „HIGH“ oder „CRITICAL“ ist.
– is_significant Wird auf „true“ gesetzt, wenn der Schweregrad „HIGH“ oder „CRITICAL“ ist.

Änderungsprotokoll

Änderungsprotokoll für diesen Parser ansehen

Benötigen Sie weitere Hilfe? Antworten von Community-Mitgliedern und Google SecOps-Experten erhalten