Übersicht über das Parsen von Protokollen

Unterstützt in:

In diesem Dokument finden Sie einen Überblick darüber, wie Google Security Operations Rohlogs in das Unified Data Model (UDM)-Format parset.

Google Security Operations kann Protokolldaten aus den folgenden Aufnahmequellen empfangen:

  • Google Security Operations-Weiterleitung
  • Google Security Operations API-Feed
  • Google Security Operations Ingestion API
  • Technologiepartner eines Drittanbieters

Im Allgemeinen senden Kunden Daten als Original-Rohprotokolle. Einzigartige Sicherheit bei Google Security Operations das Gerät identifiziert, das die Protokolle mithilfe des LogType generiert hat. LogType steht für beide:

  • den Anbieter und das Gerät, von denen das Log generiert wurde, z. B. Cisco Firewall, Linux DHCP-Server oder Bro DNS.
  • Der Parser wandelt das Rohprotokoll in ein strukturiertes einheitliches Datenmodell (Unified Data Model, UDM) um. Zwischen einem Parser und einem LogType besteht eine 1:1-Beziehung. Jeder Parser wandelt Daten um, die von einem einzelnen LogType empfangen wurden.

Google Security Operations bietet eine Reihe von Standardparsern, die die ursprünglichen Rohprotokolle lesen und Strukturierte UDM-Datensätze mithilfe von Daten im ursprünglichen Rohlog generieren Google Security Operations verwaltet diese Parser. Kunden können auch benutzerdefinierte Datenzuordnungsanweisungen definieren, indem sie einen kundenspezifischen Parser erstellen. Google Security Operations kontaktieren für Informationen zum Erstellen eines kundenspezifischen Parsers.

Workflow für Datenaufnahme und Normalisierung

Der Parser enthält Anweisungen zur Datenzuordnung. Sie definiert, wie Daten zugeordnet werden. aus dem ursprünglichen Rohprotokoll in ein oder mehrere Felder in der UDM-Datenstruktur.

Wenn keine Parsing-Fehler auftreten, erstellt Google Security Operations einen UDM-strukturierten Datensatz mithilfe von Daten aus dem Rohprotokoll. Der Prozess zum Konvertieren eines Rohlogs in einen UDM-Eintrag wird als Normalisierung bezeichnet.

Ein Standardparser könnte eine Teilmenge von Kernwerten aus dem Rohlog zuordnen. Normalerweise Diese Kernfelder sind für die Bereitstellung von Sicherheitsinformationen in Google Security Operations Nicht zugeordnete Werte bleiben im Rohprotokoll, werden aber nicht im UDM-Eintrag gespeichert.

Kunden können auch die Ingestion API verwenden, um Daten im strukturierten Unified Data Model-Format (UDM) zu senden.

Parsen von aufgenommenen Daten anpassen

Google Security Operations bietet Kunden folgende Möglichkeiten: Passen Sie das Parsen von Daten für eingehende Original-Protokolldaten an.

  • Kundenspezifische Parser: Kunden erstellen eine benutzerdefinierte Parserkonfiguration für einen bestimmten Protokolltyp, der ihren spezifischen Anforderungen entspricht. Ein kundenspezifischer Parser ersetzt Den Standardparser für den jeweiligen LogType. Google Security Operations kontaktieren für Informationen zum Erstellen eines kundenspezifischen Parsers.
  • Parser-Erweiterungen: Kunden können benutzerdefinierte Zuordnungsanweisungen hinzufügen in zur Parser-Standardkonfiguration hinzu. Jeder Kunde kann eigene Anweisungen für die Zuordnung erstellen. Diese Zuordnung Anweisungen zum Extrahieren und Transformieren zusätzlicher Felder aus ursprünglichen Rohlogs in UDM-Feldern. Eine Parsererweiterung ersetzt nicht den Standard- oder kundenspezifischen Parser.

Beispiel für ein Squid-Webproxy-Log

In diesem Abschnitt finden Sie ein Beispiel für ein Squid-Webproxy-Protokoll und eine Beschreibung, wie die Werte einem UDM-Eintrag zugeordnet werden. Eine Beschreibung aller Felder im UDM-Schema Siehe Liste der Felder für einheitliche Datenmodell

Das Beispiel-Squid-Webproxy-Log enthält durch Leerzeichen getrennte Werte. Jeder Datensatz steht für ein Ereignis und enthält die folgenden Daten: Zeitstempel, Dauer, Client, Ergebniscode/Ergebnisstatus, übertragene Bytes, Anfragemethode, URL, Nutzer, Hierarchiecode und Inhaltstyp. In diesem Beispiel sind die folgenden Felder extrahiert und einem UDM-Eintrag zugeordnet: Zeit, Client, Ergebnisstatus, Byte, und die URL angeben.

1588059648.129 23 192.168.23.4 TCP_HIT/200 904 GET www.google.com/images/sunlogo.png - HIER_DIRECT/203.0.113.52 image/jpeg

Beispiel für einen Squid-Webproxy

Beachten Sie beim Vergleich dieser Strukturen, dass nur ein Teil der ursprünglichen Logdaten sind im UDM-Eintrag enthalten. Bestimmte Felder sind Pflichtfelder, andere sind optional. Außerdem ist nur ein Teil der Abschnitte im UDM-Eintrag Daten enthalten. Wenn der Parser der UDM keine Daten aus dem ursprünglichen Log zuordnet wird dieser Abschnitt des UDM-Eintrags in Google Security Operations nicht angezeigt.

UDM zugeordnete Logwerte

Im Bereich metadata wird der Zeitstempel des Ereignisses gespeichert. Der Wert wurde umgerechnet, von EPOCH in RFC 3339-Format ändern. Diese Umwandlung ist optional. Der Zeitstempel kann im EPOCH-Format gespeichert werden. Dabei werden die Sekunden und Millisekunden in separaten Feldern gespeichert.

Die metadata.event_type enthält den Wert NETWORK_HTTP, bei dem es sich um einen Aufzählungswert handelt. der den Ereignistyp identifiziert. Der Wert von metadata.event_type bestimmt, welche zusätzlichen UDM-Felder erforderlich und optional sind. Die Werte product_name und vendor_name enthalten nutzerfreundliche Beschreibungen des Geräts, auf dem das ursprüngliche Protokoll aufgezeichnet wurde.

Der Wert metadata.event_type in einem UDM-Ereignisdatensatz entspricht nicht dem „log_type“, der beim Aufnehmen von Daten mit der Ingestion API definiert wurde. Diese beiden Attribute speichern und unterschiedliche Informationen.

Der Bereich network enthält Werte aus dem ursprünglichen Protokollereignis. Hinweis in dieser Beispiel, dass der Statuswert aus dem ursprünglichen Protokoll aus dem Feld „Ergebnis“ Code/Status bevor in den UDM-Eintrag geschrieben werden. Nur der „result_code“ wurde in den UDM-Eintrag aufgenommen.

UDM zugeordnete Logwerte

Im Abschnitt principal werden die Clientinformationen aus dem ursprünglichen Log gespeichert. Im Abschnitt target werden sowohl die vollständig qualifizierte URL als auch die IP-Adresse gespeichert.

Im Abschnitt security_result wird einer der Aufzählungswerte für die Aktion gespeichert, die im ursprünglichen Log aufgezeichnet wurde.

Dies ist der im JSON-Format formatierte UDM-Eintrag. Beachten Sie, dass nur Abschnitte, enthalten sind. Die Abschnitte src, observer, intermediary, about und extensions sind nicht enthalten.

{
        "metadata": {
            "event_timestamp": "2020-04-28T07:40:48.129Z",
            "event_type": "NETWORK_HTTP",
            "product_name": "Squid Proxy",
            "vendor_name": "Squid"
        },
        "principal": {
            "ip": "192.168.23.4"
        },
        "target": {
            "url": "www.google.com/images/sunlogo.png",
            "ip": "203.0.113.52"
        },
        "network": {
            "http": {
                "method": "GET",
                "response_code": 200,
                "received_bytes": 904
            }
        },
        "security_result": {
            "action": "UNKNOWN_ACTION"
        }
}

Schritte in Parser-Anweisungen

Anweisungen zur Datenzuordnung innerhalb eines Parsers folgen einem gemeinsamen Muster: folgt:

  1. Daten aus dem ursprünglichen Protokoll analysieren und extrahieren.
  2. Die extrahierten Daten bearbeiten. Dazu gehört auch die Verwendung bedingter Logik, Werte selektiv analysieren, Datentypen konvertieren, Teilzeichenfolgen in einer in Groß- oder Kleinbuchstaben umwandeln usw.
  3. Weisen Sie UDM-Feldern Werte zu.
  4. Gib den zugeordneten UDM-Eintrag mit dem Schlüssel „@output“ aus.

Daten aus dem ursprünglichen Protokoll parsen und extrahieren

Filteranweisung festlegen

Die filter-Anweisung ist die erste Anweisung in den Parseanweisungen. Alle zusätzlichen Parsing-Anweisungen sind in der Anweisung filter enthalten.

filter {

}

Variablen initialisieren, in denen extrahierte Werte gespeichert werden

Initialisieren Sie in der Anweisung filter Zwischenvariablen, die vom verwendet der Parser zum Speichern aus dem Log extrahierte Werte.

Diese Variablen werden jedes Mal verwendet, wenn ein einzelnes Protokoll geparst wird. Der Wert in jeder Zwischenvariablen wird später in den Parseanweisungen auf ein oder mehrere UDM-Felder festgelegt.

  mutate {
    replace => {
      "event.idm.read_only_udm.metadata.product_name" => "Webproxy"
      "event.idm.read_only_udm.metadata.vendor_name" => "Squid"
      "not_valid_log" => "false"
      "when" => ""
      "srcip" => ""
      "action" => ""
      "username" => ""
      "url" => ""
      "tgtip" => ""
      "method" => ""
    }
  }

Einzelne Werte aus dem Protokoll extrahieren

Google Security Operations bietet eine Reihe von auf Logstash basierenden Filtern zum Extrahieren von Feldern aus den Original-Protokolldateien. Je nach Format des Protokolls verwenden Sie ein oder mehrere Extraktionsfilter, um alle Daten aus dem Log zu extrahieren. Wenn der String wie folgt lautet:

  • Natives JSON: Die Parsersyntax ähnelt dem JSON-Filter, der JSON-formatierte Protokolle unterstützt. Verschachtelte JSON-Daten werden nicht unterstützt.
  • XML-Format, ähnelt die Parsersyntax der XML-Filter, der XML-formatierte Logs unterstützt.
  • Schlüssel/Wert-Paare, die Parsersyntax ähnelt dem Kv-Filter, der Nachrichten im Schlüssel/Wert-Format unterstützt.
  • CSV-Format, die Parsersyntax ähnelt dem CSV-Filter, der Nachrichten im CSV-Format unterstützt.
  • allen anderen Formaten verwenden, ähnelt die Parsersyntax der GROK-Filter mit integrierten GROK-Mustern . Hier werden Extraktionsanleitungen im Regex-Stil verwendet.

Google Security Operations bietet einen Teil der Funktionen, die in den einzelnen Filtern verfügbar sind. Google Security Operations bietet außerdem eine benutzerdefinierte Syntax für den Datenabgleich, die in den Filtern nicht verfügbar ist. Siehe Parser-Syntaxreferenz. finden Sie eine Beschreibung unterstützter Funktionen und benutzerdefinierter Funktionen.

Wir bleiben bei dem Beispiel mit dem Squid-Web-Proxy-Protokoll und der folgenden Datenextraktion -Anweisung umfasst eine Kombination aus Logstash Grok-Syntax und regulären Ausdrücke.

Die folgende Extraktionsanweisung speichert Werte in der Variablen:

  • when
  • srcip
  • action
  • returnCode
  • size
  • method
  • username
  • url
  • tgtip

In dieser Beispielanweisung wird auch das Schlüsselwort overwrite verwendet, um die extrahierten Werte zu speichern in jeder Variablen an. Wenn der Extraktionsprozess einen Fehler zurückgibt, wird durch die on_error-Anweisung not_valid_log auf True gesetzt.

grok {
   match => {
     "message" => [
       "%{NUMBER:when}\\s+\\d+\\s%{SYSLOGHOST:srcip} %{WORD:action}\\/%{NUMBER:returnCode} %{NUMBER:size} %{WORD:method} (?P<url>\\S+) (?P<username>.*?) %{WORD}\\/(?P<tgtip>\\S+).*"
     ]
   }
   overwrite => ["when","srcip","action","returnCode","size","method","url","username","tgtip"]
   on_error => "not_valid_log"
}

Extrahierte Werte bearbeiten und transformieren

Google Security Operations nutzt die Funktionen des mutate-Filter-Plug-ins von Logstash, um die Manipulation von Werten zu ermöglichen, die aus dem ursprünglichen Protokolls. Google Security Operations bietet einen Teil der Funktionen, die im Plug-in verfügbar sind. Eine Beschreibung der Funktionen finden Sie unter Parser-Syntax. unterstützte und benutzerdefinierte Funktionen wie:

  • Werte in einen anderen Datentyp umwandeln
  • Werte im String ersetzen
  • zwei Arrays zusammenführen oder einem Array einen String anhängen. Stringwerte werden vor dem Zusammenführen in ein Array konvertiert.
  • in Klein- oder Großbuchstaben umwandeln

Dieser Abschnitt enthält Beispiele für die Datentransformation, die auf dem Squid-Web-Proxy-Protokoll basieren. die ich zuvor vorgestellt habe.

Ereigniszeitstempel transformieren

Alle als UDM-Einträge gespeicherten Ereignisse müssen einen Ereigniszeitstempel haben. In diesem Beispiel wird geprüft, ob ein Wert für die Daten aus dem Protokoll extrahiert wurde. Anschließend wird der Wert mithilfe der Grok-Datumsfunktion dem Zeitformat UNIX zugeordnet.

if [when] != "" {
  date {
    match => [
      "when", "UNIX"
    ]
   }
 }

Wert für username transformieren

Mit der folgenden Beispielanweisung wird der Wert in der Variablen username konvertiert in Kleinbuchstaben.

mutate {
   lowercase => [ "username"]
   }

action-Wert transformieren

Im folgenden Beispiel wird der Wert in der Zwischenvariablen action ausgewertet und in ALLOW, BLOCK oder UNKNOWN_ACTION geändert. Dies sind gültige Werte für das UDM-Feld security_result.action. Das UDM von security_result.action ist ein Enum-Typ, der nur bestimmte Werte speichert.

if ([action] == "TCP_DENIED" or [action] == "TCP_MISS" or [action] == "Denied" or [action] == "denied" or [action] == "Dropped") {
      mutate {
        replace => {
          "action" => "BLOCK"
        }
      }
   } else if ([action] == "TCP_TUNNEL" or [action] == "Accessed" or [action] == "Built" or [action] == "Retrieved" or [action] == "Stored") {
     mutate {
        replace => {
          "action" => "ALLOW"
        }
     }
   } else {
      mutate {
        replace => {
          "action" => "UNKNOWN_ACTION" }
      }
   }

Ziel-IP-Adresse transformieren

Im folgenden Beispiel wird in der Zwischenvariablen tgtip nach einem Wert gesucht. Wenn der Wert gefunden wird, wird er mithilfe eines vordefinierten Groks mit einem IP-Adressmuster abgeglichen. Muster zu ändern. Wenn beim Abgleich des Werts mit einem IP-Adressmuster ein Fehler auftritt, Die Funktion on_error setzt die Eigenschaft not_valid_tgtip auf True. Wenn die Übereinstimmung erfolgreich ist, wird die Property not_valid_tgtip nicht festgelegt.

if [tgtip] not in [ "","-" ] {
   grok {
     match => {
       "tgtip" => [ "%{IP:tgtip}" ]
     }
     overwrite => ["tgtip"]
     on_error => "not_valid_tgtip"
   }

Datentyp von „returnCode“ und „size“ ändern

Im folgenden Beispiel wird der Wert in der Variablen size in uinteger und der Wert in der Variablen returnCode in integer umgewandelt. Das ist erforderlich, da die Variable size im UDM-Feld network.received_bytes gespeichert wird, in dem ein int64-Datentyp gespeichert wird. Die Variable returnCode wird im UDM-Feld network.http.response_code gespeichert, in dem ein int32-Datentyp gespeichert wird.

mutate {
  convert => {
    "returnCode" => "integer"
    "size" => "uinteger"
  }
}

UDM-Feldern in einem Ereignis Werte zuweisen

Nachdem die Werte extrahiert und vorverarbeitet wurden, weisen Sie sie Feldern in einer UDM zu. Ereignisdatensatz. Sie können einem UDM-Feld sowohl extrahierte als auch statische Werte zuweisen.

Wenn Sie event.disambiguation_key ausfüllen, muss dieses Feld eindeutig sein jedes Ereignis, das für das jeweilige Protokoll generiert wird. Wenn zwei Ereignisse die disambiguation_key identisch, führt dies zu unerwartetem Verhalten im System.

Die Parserbeispiele in diesem Abschnitt bauen auf dem Beispiel für Squid-Webproxy-Logs oben auf.

Ereigniszeitstempel speichern

Für jeden UDM-Ereignisabsatz muss ein Wert für das UDM-Feld metadata.event_timestamp festgelegt sein. Im folgenden Beispiel wird der aus dem Log extrahierte Ereigniszeitstempel im @timestamp. In Google Security Operations wird dieser Wert standardmäßig im UDM-Feld metadata.event_timestamp gespeichert.

mutate {
  rename => {
    "when" => "timestamp"
  }
}

Ereignistyp festlegen

Für jeden UDM-Ereignisdatensatz muss ein Wert für das UDM-Feld metadata.event_type festgelegt sein. Dieses Feld ist ein Aufzählungstyp. Der Wert dieses Feldes bestimmt, welche zusätzlichen UDM-Felder ausgefüllt werden müssen, damit der UDM-Eintrag gespeichert wird. Das Parsen und Normalisieren schlägt fehl, wenn eines der Pflichtfelder keine gültigen Daten enthält.

replace => {
    "event.idm.read_only_udm.metadata.event_type" => "NETWORK_HTTP"
   }
}

Speichern Sie die Werte username und method mit der replace-Anweisung

Die Werte in den Zwischenfeldern username und method sind Strings. Die folgendes Beispiel prüft, ob ein gültiger Wert vorhanden ist, und speichert, falls dies der Fall ist, den Wert username in das UDM-Feld principal.user.userid und den Wert method in das UDM-Feld network.http.method ein.

if [username] not in [ "-" ,"" ] {
  mutate {
    replace => {
      "event.idm.read_only_udm.principal.user.userid" => "%{username}"
    }
  }
}

if [method] != "" {
  mutate {
    replace => {
      "event.idm.read_only_udm.network.http.method" => "%{method}"
    }
  }
}

Speichern Sie action im UDM-Feld security_result.action.

Im vorherigen Abschnitt war der Wert in der Zwischenvariablen action ausgewertet und in einen der Standardwerte für das UDM-Feld security_result.action umgewandelt wurde.

Die UDM-Felder security_result und action speichern ein Array von Elementen. Sie müssen also beim Speichern dieser Daten einen etwas anderen Ansatz verfolgen. Wert.

Speichern Sie zuerst den transformierten Wert in einer Zwischen-security_result.action. ein. Das Feld security_result ist ein übergeordnetes Feld des Felds action.

mutate {
   merge => {
     "security_result.action" => "action"
   }
}

Speichern Sie als Nächstes das Zwischenfeld security_result.action im UDM-Feld security_result. Das UDM-Feld security_result speichert ein Array von Elementen. Der Wert wird diesem Feld angehängt.

# save the security_result field
mutate {
  merge => {
    "event.idm.read_only_udm.security_result" => "security_result"
  }
}

Ziel-IP-Adresse und Quell-IP-Adresse mit der merge-Anweisung speichern

Speichere die folgenden Werte im UDM-Ereignisprotokoll:

  • Wert in der Zwischenvariablen srcip für das UDM-Feld principal.ip.
  • Wert in der Zwischenvariablen tgtip für das UDM-Feld target.ip.

Die UDM-Felder principal.ip und target.ip speichern ein Array von Elementen, sodass werden an jedes Feld angehängt.

Die folgenden Beispiele zeigen verschiedene Ansätze zum Speichern dieser Werte. Während des Transformationsschritts wurde die Zwischenvariable tgtip mit einem IP-Adresse mit einem vordefinierten Grok-Muster. In der folgenden Beispielanweisung wird geprüft, ob die Eigenschaft not_valid_tgtip den Wert „wahr“ hat, was bedeutet, dass der tgtip-Wert nicht mit einem IP-Adressmuster abgeglichen werden konnte. Ist der Wert „false“, wird der Wert tgtip im UDM-Feld target.ip gespeichert.

if ![not_valid_tgtip] {
  mutate {
    merge => {
      "event.idm.read_only_udm.target.ip" => "tgtip"
    }
  }
 }

Die Zwischenvariable srcip wurde nicht transformiert. Die folgende Erklärung Prüft, ob ein Wert aus dem ursprünglichen Protokoll extrahiert wurde, und speichert, ob dies der Fall ist den Wert in das UDM-Feld principal.ip ein.

if [srcip] != "" {
  mutate {
    merge => {
      "event.idm.read_only_udm.principal.ip" => "srcip"
    }
  }
}

url, returnCode und size mit der rename-Anweisung speichern

In der folgenden Beispielanweisung werden die folgenden Werte mithilfe der rename-Anweisung gespeichert.

  • Die Variable url, die im UDM-Feld target.url gespeichert wurde.
  • Die Zwischenvariable returnCode, die im UDM-Feld network.http.response_code gespeichert wurde.
  • Die Zwischenvariable size, die im UDM-Feld network.received_bytes gespeichert ist.
mutate {
  rename => {
     "url" => "event.idm.read_only_udm.target.url"
     "returnCode" => "event.idm.read_only_udm.network.http.response_code"
     "size" => "event.idm.read_only_udm.network.received_bytes"
  }
}

UDM-Eintrag an die Ausgabe binden

Die letzte Anweisung in der Datenzuordnungsanleitung gibt die verarbeiteten Daten in einem UDM-Ereignisdatensatz aus.

mutate {
    merge => {
      "@output" => "event"
    }
  }

Vollständiger Parsercode

Dies ist der vollständige Parsercode. Die Reihenfolge der Anweisungen entspricht nicht den in derselben Reihenfolge wie in den vorherigen Abschnitten, aber dieselbe Ausgabe.

filter {

# initialize variables
  mutate {
    replace => {
      "event.idm.read_only_udm.metadata.product_name" => "Webproxy"
      "event.idm.read_only_udm.metadata.vendor_name" => "Squid"
      "not_valid_log" => "false"
      "when" => ""
      "srcip" => ""
      "action" => ""
      "username" => ""
      "url" => ""
      "tgtip" => ""
      "method" => ""
    }
  }

  # Extract fields from the raw log.
    grok {
      match => {
        "message" => [
          "%{NUMBER:when}\\s+\\d+\\s%{SYSLOGHOST:srcip} %{WORD:action}\\/%{NUMBER:returnCode} %{NUMBER:size} %{WORD:method} (?P<url>\\S+) (?P<username>.*?) %{WORD}\\/(?P<tgtip>\\S+).*"
        ]
      }
      overwrite => ["when","srcip","action","returnCode","size","method","url","username","tgtip"]
      on_error => "not_valid_log"
    }

  # Parse event timestamp
  if [when] != "" {
    date {
      match => [
        "when", "UNIX"
      ]
     }
   }

   # Save the value in "when" to the event timestamp
   mutate {
     rename => {
       "when" => "timestamp"
     }
   }

   # Transform and save username
   if [username] not in [ "-" ,"" ] {
     mutate {
       lowercase => [ "username"]
        }
      }
     mutate {
       replace => {
         "event.idm.read_only_udm.principal.user.userid" => "%{username}"
       }
     }


if ([action] == "TCP_DENIED" or [action] == "TCP_MISS" or [action] == "Denied" or [action] == "denied" or [action] == "Dropped") {
      mutate {
        replace => {
          "action" => "BLOCK"
        }
      }
   } else if ([action] == "TCP_TUNNEL" or [action] == "Accessed" or [action] == "Built" or [action] == "Retrieved" or [action] == "Stored") {
     mutate {
        replace => {
          "action" => "ALLOW"
        }
     }
   } else {
      mutate {
        replace => {
          "action" => "UNKNOWN_ACTION" }
      }
   }

  # save transformed value to an intermediary field
   mutate {
      merge => {
        "security_result.action" => "action"
      }
   }

    # save the security_result field
    mutate {
      merge => {
        "event.idm.read_only_udm.security_result" => "security_result"
      }
    }

   # check for presence of target ip. Extract and store target IP address.
   if [tgtip] not in [ "","-" ] {
     grok {
       match => {
         "tgtip" => [ "%{IP:tgtip}" ]
       }
       overwrite => ["tgtip"]
       on_error => "not_valid_tgtip"
     }

     # store  target IP address
     if ![not_valid_tgtip] {
       mutate {
         merge => {
           "event.idm.read_only_udm.target.ip" => "tgtip"
         }
       }
     }
   }

   # convert  the returnCode and size  to integer data type
   mutate {
     convert => {
       "returnCode" => "integer"
       "size" => "uinteger"
     }
   }

   # save  url, returnCode, and size
   mutate {
     rename => {
        "url" => "event.idm.read_only_udm.target.url"
        "returnCode" => "event.idm.read_only_udm.network.http.response_code"
        "size" => "event.idm.read_only_udm.network.received_bytes"
     }

     # set the event type to NETWORK_HTTP
     replace => {
        "event.idm.read_only_udm.metadata.event_type" => "NETWORK_HTTP"
     }
   }

   # validate and set source IP address
   if [srcip] != "" {
     mutate {
       merge => {
         "event.idm.read_only_udm.principal.ip" => "srcip"
       }
     }
   }

  # save  event to @output
   mutate {
     merge => {
       "@output" => "event"
     }
   }

} #end of filter