Beispiel für den Import von Nginx-Protokollen in Elasticsearch

log_format main '$remote_addr $http_x_forwarded_for [$time_local] $server_name $request ' 
            „$status $body_bytes_sent $http_referer“ 
            '"$http_user_agent" '
            '"$Verbindung" '
            '"$http_cookie" '
            „$Anforderungszeit“
            „$upstream_response_time“;

tar -zxvf filebeat-6.2.4-linux-x86_64.tar.gz -C /usr/local
cd /usr/local;ln -s filebeat-6.2.4-linux-x86_64 filebeat
cd /usr/local/filebeat

./filebeat-Module aktivieren nginx

./filebeat-Modulliste

vim /usr/local/filebeat/blog_module_logstash.yml
Dateibeat.modules:
- Modul: nginx
 Zugang:
  aktiviert: true
  var.paths: ["/home/weblog/blog.cnfol.com_access.log"]
 #Fehler:
 # aktiviert: true
 # var.paths: ["/home/weblogerr/blog.cnfol.com_error.log"]


Ausgabe.logstash:
 Gastgeber: ["192.168.15.91:5044"]

./filebeat -c blog_module_logstash.yml -e

tar -zxvf logstash-6.2.4.tar.gz /usr/local
cd /usr/local;ln -s logstash-6.2.4 logstash
Erstellen Sie eine Pipeline-Datei für das Nginx-Protokoll cd /usr/local/logstash

Anbieter/Bundle/jruby/2.3.0/Gems/logstash-patterns-core-4.1.2/Muster

VORWORT (?:%{IPV4}[,]?[ ]?)+|%{WORT}

#Eingabe {
# stdin {}
#}
# Dateneingabe von Filebeat akzeptieren {
 schlägt {
 Port => 5044
 Host => "0.0.0.0"
 }
}

Filter {
 # Einen Debugging-Schalter hinzufügen mutate{add_field => {"[@metadata][debug]"=>true}}
 grok {
 # Nginx-Protokoll filtern #match => { "message" => "%{NGINXACCESS_TEST2}" }
 #Match => {"message" => ' %{iporhost: clientIp} # (? <http_x_forwarded_for> [^\ #]*) # \ [ %{httpdate: [@metadata] [WebTime]} {{nicht problem {{{nichtspace: Hostname: Hostname: {{} # %{Word {Word {Word {Word {{Word {Word {{Word {{Word {Word} {{{{{{{nicht : httpversion} #%{Nummer: Antwort} # (?:%{number: bytes} |-) # (?: "(?:%{NotSpace: Referrer} |-)" |%{NotSpace: Referrer} | {Nummer: Verbindung} |-) # (?: "(? <Cookies> [^ #]*)") # %{Nummer: Request_time: float} # (?: %{Number: upstream_response_time: float} |-) '}
 #Match => {"message" => '(?:%{iporhost: clientIp} |-) (? |-)%{word: methode}%{afripathparam: request} http/%{numme: httpversion}%{numme: response} (? : Verbindung} |-) "|%{Nummer: Verbindung} |-) (?:" (? <Cookies> [^#]*) ")%{Nummer: Request_Time: float} (?
    match => {"message" => '(? Nummer: httpversion}%{Nummer: Antwort} (? {Nummer: Request_time: float} (?:%{number: upstream_response_time: float} |-) '}
 }
 # Weisen Sie dem neuen Feld @read_tiimestamp den Standardwert @timestamp (die Zeit, zu der Beats Protokolle sammelt) zu
 Rubin { 
 #code => "event.set('@read_timestamp',event.get('@timestamp'))"
 #Ändern Sie die Zeitzone auf Ost 8 Code => "event.set('@read_timestamp',event.get('@timestamp').time.localtime + 8*60*60)"
 }
 # Formatieren Sie die Aufzeichnungszeit des Nginx-Protokolls. # Formatieren Sie die Zeit 20. Mai 2015:21:05:56 +0000
 Datum {
 Gebietsschema => "en"
 Übereinstimmung => ["[@metadata][webtime]","dd/MMM/yyyy:HH:mm:ss Z"]
 }
 # Wandelt das Bytefeld von einem String in eine Zahl um mutate {
 konvertieren => {"Bytes" => "Ganzzahl"}
 }
 # Analysieren Sie das Cookie-Feld in ein JSON
 #mutieren {
 # gsub => ["Cookies",'\;',',']
 #} 
 # Wenn CDN-Beschleunigung verwendet wird, gibt es mehrere IP-Adressen für http_x_forwarded_for. Die erste IP-Adresse ist die tatsächliche IP-Adresse des Benutzers.
 wenn[http_x_forwarded_for] =~ ", "{
     Rubin {
         Code => 'event.set("http_x_forwarded_for", event.get("http_x_forwarded_for").split(",")[0])'
        }
    }
 # Analysieren Sie die IP-Adresse und ermitteln Sie den geografischen Standort der IP-Adresse geoip {
 Quelle => "http_x_forwarded_for"
 # # Holen Sie sich nur die Felder Längen- und Breitengrad, Land, Stadt und Zeitzone der IP => ["Standort","Ländername","Stadtname","Regionsname"] 
 }
 # Analysieren Sie das Agent-Feld, um spezifische Informationen wie Browser- und Systemversion zu erhalten useragent {
 Quelle => "Agent"
 Ziel => "Benutzeragent"
 }
 #Geben Sie die zu löschenden Daten an#mutate{remove_field=>["message"]}
 # Legen Sie das Indexnamenpräfix entsprechend dem Protokollnamen ruby ​​​​{ fest
 Code => 'Ereignis.Setzen("@[Metadaten][index_pre]",Ereignis.Get("Quelle").Split("/")[-1])'
 } 
 # Formatieren Sie @timestamp auf 2019.04.23
 Rubin {
 Code => 'Ereignis.Setzen("@[Metadaten][Index_Tag]",Ereignis.Get("@Zeitstempel").Zeit.Lokalzeit.strftime("%Y.%m.%d"))'
 }
 # Legen Sie den Standardindexnamen für die Ausgabe von mutate { fest.
 Feld hinzufügen => {
  #"[@metadata][index]" => "%{@[metadata][index_pre]}_%{+JJJJ.MM.tt}"
  "[@metadata][index]" => "%{@[metadata][index_pre]}_%{@[metadata][index_day]}"
 }
 }
 # Analysieren Sie das Cookies-Feld in JSON
# mutieren {
# gsub => [
# "Kekse", ";", ",",
# "Kekse", "=", ":"
# ]
# #split => {"Cookies" => ","}
# }
# json_encode {
# Quelle => "Cookies"
# Ziel => "cookies_json"
# }
# mutieren {
# gsub => [
# "cookies_json", ',', ',',
# "cookies_json", ':', '":"'
# ]
# }
# json {
# Quelle => "cookies_json"
# Ziel => "Cookies2"
# }
 # Wenn beim Grok-Parsing ein Fehler auftritt, schreiben Sie den Fehler in einen separaten Index, wenn "_grokparsefailure" in [tags] {
 #if "_dateparsefailure" in [tags] {
 mutieren {
  ersetzen => {
  #"[@metadata][index]" => "%{@[metadata][index_pre]}_failure_%{+JJJJ.MM.tt}"
  "[@metadata][index]" => "%{@[metadata][index_pre]}_failure_%{@[metadata][index_day]}"
  }
 }
 # Wenn kein Fehler vorliegt, löschen Sie die Nachricht
 }anders{
 mutieren{remove_field=>["Nachricht"]}
 }
}

Ausgabe {
 wenn [@metadata][debug]{
 # Ausgabe an rubydebuyg und Ausgabemetadaten
 stdout{codec => rubydebug{metadata => true}}
 }anders{
 # Konvertieren Sie den Ausgabeinhalt in "."
 stdout{codec => Punkte} 
 # Ausgabe an das angegebene es
 elastische Suche
  Hosts => ["192.168.15.160:9200"]
  index => "%{[@metadata][index]}"
  Dokumenttyp => "Dokument"
 } 
 }
}

nohup bin/logstash -f test_pipline2.conf &