Kami berteman dengan ELK dan Exchange. Bagian 2

Saya melanjutkan cerita saya tentang bagaimana membuat Exchange dan ELK teman (mulai dari sini ). Izinkan saya mengingatkan Anda bahwa kombinasi ini mampu menangani log dalam jumlah yang sangat besar tanpa ragu-ragu. Kali ini kita akan berbicara tentang cara membuat Exchange bekerja dengan komponen Logstash dan Kibana.



Logstash di tumpukan ELK digunakan untuk memproses log secara cerdas dan mempersiapkannya untuk penempatan di Elastis dalam bentuk dokumen, yang menjadi dasar untuk membangun berbagai visualisasi di Kibana.

Instalasi

Terdiri dari dua tahap:

• Menginstal dan mengkonfigurasi paket OpenJDK.
• Menginstal dan mengkonfigurasi paket Logstash.

Instalasi dan Konfigurasi



Paket OpenJDK Paket OpenJDK harus diunduh dan diurai ke dalam direktori tertentu. Kemudian jalur ke direktori ini harus dimasukkan ke dalam variabel $ env: Path dan $ env: JAVA_HOME dari sistem operasi Windows:











Periksa versi Java:

PS C:\> java -version
openjdk version "13.0.1" 2019-10-15
OpenJDK Runtime Environment (build 13.0.1+9)
OpenJDK 64-Bit Server VM (build 13.0.1+9, mixed mode, sharing)

Menginstal dan mengkonfigurasi paket Logstash

Unduh file arsip dengan distribusi Logstash dari sini . Arsip harus dibongkar ke root disk. C:\Program FilesAnda tidak boleh membukanya ke dalam folder , Logstash akan menolak untuk memulai secara normal. Kemudian Anda perlu membuat jvm.optionsperubahan pada file yang bertanggung jawab mengalokasikan RAM untuk proses Java. Saya sarankan untuk menentukan setengah dari RAM server. Jika dia memiliki 16 GB RAM, maka kunci defaultnya adalah:

-Xms1g
-Xmx1g

harus diganti dengan:

-Xms8g
-Xmx8g

Selain itu, disarankan untuk mengomentari baris tersebut -XX:+UseConcMarkSweepGC. Baca lebih lanjut di sini . Langkah selanjutnya adalah membuat konfigurasi default di file logstash.conf:

input {
 stdin{}
}
 
filter {
}
 
output {
 stdout {
 codec => "rubydebug"
 }
}

Dengan konfigurasi ini, Logstash membaca data dari konsol, meneruskannya melalui filter kosong, dan mencetaknya kembali ke konsol. Menerapkan konfigurasi ini akan menguji fungsionalitas Logstash. Untuk melakukan ini, jalankan secara interaktif:

PS C:\...\bin> .\logstash.bat -f .\logstash.conf
...
[2019-12-19T11:15:27,769][INFO ][logstash.javapipeline    ][main] Pipeline started {"pipeline.id"=>"main"}
The stdin plugin is now waiting for input:
[2019-12-19T11:15:27,847][INFO ][logstash.agent           ] Pipelines running {:count=>1, :running_pipelines=>[:main], :non_running_pipelines=>[]}
[2019-12-19T11:15:28,113][INFO ][logstash.agent           ] Successfully started Logstash API endpoint {:port=>9600}

Logstash berhasil diluncurkan pada port 9600.



Langkah terakhir penginstalan adalah meluncurkan Logstash sebagai layanan Windows. Ini dapat dilakukan, misalnya, menggunakan paket NSSM :

PS C:\...\bin> .\nssm.exe install logstash
Service "logstash" installed successfully!

toleransi kesalahan

Mekanisme Persistent Queues memastikan keamanan log selama transmisi dari server sumber.

bagaimana cara kerjanya

Tata letak antrian selama pemrosesan log: input → queue → filter + output.



Plugin input menerima data dari sumber log, menuliskannya ke antrian dan mengirimkan konfirmasi penerimaan data ke sumber.



Pesan dari antrian diproses oleh Logstash, melewati filter dan plugin keluaran. Setelah menerima konfirmasi dari keluaran untuk mengirim log, Logstash menghapus log yang diproses dari antrian. Jika Logstash berhenti, maka semua pesan yang belum diproses dan pesan yang belum menerima konfirmasi pengiriman tetap berada dalam antrean, dan Logstash akan terus memprosesnya saat dimulai berikutnya.

Kustomisasi

Diatur oleh kunci dalam file C:\Logstash\config\logstash.yml:

• queue.type: (nilai yang mungkin adalah persisteddan memory (default)).
• path.queue: (jalur ke folder dengan file antrian, yang disimpan di C: \ Logstash \ queue secara default).
• queue.page_capacity: (ukuran halaman maksimum dari antrian, defaultnya adalah 64mb).
• queue.drain: (true / false - mengaktifkan / menonaktifkan menghentikan pemrosesan antrian sebelum mematikan Logstash. Saya tidak menyarankan untuk menyalakannya, karena ini secara langsung akan memengaruhi kecepatan mematikan server).
• queue.max_events: (jumlah maksimum acara dalam antrian, default - 0 (tidak terbatas)).
• queue.max_bytes: (ukuran antrian maksimum dalam byte, default adalah 1024mb (1gb)).

Jika queue.max_eventsdan dikonfigurasi queue.max_bytes, pesan akan berhenti diterima dalam antrian ketika nilai dari pengaturan ini tercapai. Baca lebih lanjut tentang Antrian Persisten di sini .



Contoh dari bagian logstash.yml yang bertanggung jawab untuk menyiapkan antrian:

queue.type: persisted
queue.max_bytes: 10gb

Kustomisasi

Konfigurasi Logstash biasanya terdiri dari tiga bagian, yang bertanggung jawab untuk berbagai tahapan pemrosesan log masuk: penerimaan (bagian masukan), penguraian (bagian filter) dan pengiriman ke Elastis (bagian keluaran). Di bawah ini kita akan melihat lebih dekat pada masing-masingnya.

Memasukkan

Arus masuk dengan log mentah diterima dari agen pemukul file. Plugin inilah yang kami tentukan di bagian input:

input {
  beats {
    port => 5044
  }
}

Setelah pengaturan ini, Logstash mulai mendengarkan pada port 5044, dan ketika menerima log, memprosesnya sesuai dengan pengaturan di bagian filter. Jika perlu, Anda dapat menggabungkan saluran untuk menerima log dari filebit di SSL. Baca lebih lanjut tentang pengaturan plugin beats di sini .

Saring

Semua log teks menarik yang dihasilkan Exchange untuk diproses berada dalam format csv dengan bidang yang dijelaskan dalam file log itu sendiri. Untuk mem-parsing data csv, Logstash menawarkan tiga plugin: dissect , csv, dan grok. Yang pertama adalah yang tercepat , tetapi hanya dapat mengurai log yang paling sederhana.

Misalnya, ini akan membagi rekaman berikut menjadi dua (karena adanya koma di dalam bidang), yang akan menyebabkan log diuraikan secara tidak benar:

…,"MDB:GUID1, Mailbox:GUID2, Event:526545791, MessageClass:IPM.Note, CreationTime:2020-05-15T12:01:56.457Z, ClientType:MOMT, SubmissionAssistant:MailboxTransportSubmissionEmailAssistant",…

Ini dapat digunakan saat mengurai log, misalnya, IIS. Dalam kasus ini, bagian filter mungkin terlihat seperti ini:

filter {
  if "IIS" in [tags] {
    dissect {
      mapping => {
        "message" => "%{date} %{time} %{s-ip} %{cs-method} %{cs-uri-stem} %{cs-uri-query} %{s-port} %{cs-username} %{c-ip} %{cs(User-Agent)} %{cs(Referer)} %{sc-status} %{sc-substatus} %{sc-win32-status} %{time-taken}"
      }
      remove_field => ["message"]
      add_field => { "application" => "exchange" }
    }
  }
} 

Konfigurasi Logstash memungkinkan pernyataan bersyarat , jadi kita hanya dapat mengirim log ke plugin dissect yang telah ditandai dengan tag filebeat IIS. Di dalam plugin, kami membandingkan nilai bidang dengan namanya, menghapus bidang asli messageyang berisi entri dari log, dan kami dapat menambahkan bidang sewenang-wenang yang akan, misalnya, berisi nama aplikasi tempat kami mengumpulkan log.



Dalam kasus log pelacakan, lebih baik menggunakan plugin csv, ini dapat memproses bidang kompleks dengan benar:

filter {
  if "Tracking" in [tags] {
    csv {
      columns => ["date-time","client-ip","client-hostname","server-ip","server-hostname","source-context","connector-id","source","event-id","internal-message-id","message-id","network-message-id","recipient-address","recipient-status","total-bytes","recipient-count","related-recipient-address","reference","message-subject","sender-address","return-path","message-info","directionality","tenant-id","original-client-ip","original-server-ip","custom-data","transport-traffic-type","log-id","schema-version"]
      remove_field => ["message", "tenant-id", "schema-version"]
      add_field => { "application" => "exchange" }
    }
}

Di dalam plugin, kami mencocokkan nilai bidang dengan namanya, menghapus bidang asli message(serta bidang tenant-iddan schema-version) yang berisi entri dari log, dan kami dapat menambahkan bidang arbitrer yang akan, misalnya, berisi nama aplikasi tempat kami mengumpulkan log.



Saat keluar dari tahap pemfilteran, kita akan mendapatkan dokumen dalam perkiraan pertama, siap untuk dirender di Kibana. Kami akan melewatkan yang berikut ini:

• Bidang numerik akan dikenali sebagai teks, mencegah operasi dilakukan pada bidang tersebut. Yakni, bidang time-takenlog IIS, serta bidang Pelacakan recipient-countdan total-biteslog.
• Stempel waktu dokumen standar akan berisi waktu pemrosesan log, bukan waktu perekaman sisi server.
• Bidang recipient-addressakan terlihat seperti konstruksi tunggal, yang tidak memungkinkan analisis dengan menghitung penerima surat.

Sekarang saatnya menambahkan keajaiban ke proses pemrosesan log.

Mengonversi bidang numerik

Plugin pembedahan memiliki opsi convert_datatypeyang dapat Anda gunakan untuk mengonversi bidang teks ke format digital. Misalnya seperti ini:

dissect {
  …
  convert_datatype => { "time-taken" => "int" }
  …
}

Perlu diingat bahwa metode ini hanya cocok jika bidang pasti berisi string. Opsi ini tidak memproses nilai null dari kolom dan dimasukkan ke dalam pengecualian.



Untuk log pelacakan mengkonversi metode serupa harus dihindari, karena lapangan recipient-countdan total-bitesbisa kosong. Lebih baik menggunakan plugin mutate untuk mengonversi bidang ini :

mutate {
  convert => [ "total-bytes", "integer" ]
  convert => [ "recipient-count", "integer" ]
}

Memisahkan penerima_alamat menjadi penerima individu

Tugas ini juga dapat diselesaikan menggunakan plugin mutate:

mutate {
  split => ["recipient_address", ";"]
}

Mengubah stempel waktu

Dalam kasus log pelacakan, tugas ini sangat mudah diselesaikan dengan plugin tanggal , yang akan membantu untuk menulis timestamptanggal dan waktu di bidang dalam format yang diperlukan dari lapangan date-time:

date {
  match => [ "date-time", "ISO8601" ]
  timezone => "Europe/Moscow"
  remove_field => [ "date-time" ]
}

Dalam kasus log IIS, kita perlu menggabungkan data lapangan datedan time, dengan menggunakan plugin mutate, daftarkan zona waktu yang kita butuhkan dan tempatkan cap waktu ini dengan timestampmenggunakan plugin tanggal:

mutate { 
  add_field => { "data-time" => "%{date} %{time}" }
  remove_field => [ "date", "time" ]
}
date { 
  match => [ "data-time", "YYYY-MM-dd HH:mm:ss" ]
  timezone => "UTC"
  remove_field => [ "data-time" ]
}

Keluaran

Bagian keluaran digunakan untuk mengirim log yang diproses ke penerima log. Dalam kasus pengiriman langsung ke Elastic, plugin elasticsearch digunakan , yang menentukan alamat server dan template nama indeks untuk mengirim dokumen yang dihasilkan:

output {
  elasticsearch {
    hosts => ["127.0.0.1:9200", "127.0.0.2:9200"]
    manage_template => false
    index => "Exchange-%{+YYYY.MM.dd}"
  }
}

Konfigurasi akhir

Konfigurasi terakhir akan terlihat seperti ini:

input {
  beats {
    port => 5044
  }
}
 
filter {
  if "IIS" in [tags] {
    dissect {
      mapping => {
        "message" => "%{date} %{time} %{s-ip} %{cs-method} %{cs-uri-stem} %{cs-uri-query} %{s-port} %{cs-username} %{c-ip} %{cs(User-Agent)} %{cs(Referer)} %{sc-status} %{sc-substatus} %{sc-win32-status} %{time-taken}"
      }
      remove_field => ["message"]
      add_field => { "application" => "exchange" }
      convert_datatype => { "time-taken" => "int" }
    }
    mutate { 
      add_field => { "data-time" => "%{date} %{time}" }
      remove_field => [ "date", "time" ]
    }
    date { 
      match => [ "data-time", "YYYY-MM-dd HH:mm:ss" ]
      timezone => "UTC"
      remove_field => [ "data-time" ]
    }
  }
  if "Tracking" in [tags] {
    csv {
      columns => ["date-time","client-ip","client-hostname","server-ip","server-hostname","source-context","connector-id","source","event-id","internal-message-id","message-id","network-message-id","recipient-address","recipient-status","total-bytes","recipient-count","related-recipient-address","reference","message-subject","sender-address","return-path","message-info","directionality","tenant-id","original-client-ip","original-server-ip","custom-data","transport-traffic-type","log-id","schema-version"]
      remove_field => ["message", "tenant-id", "schema-version"]
      add_field => { "application" => "exchange" }
    }
    mutate {
      convert => [ "total-bytes", "integer" ]
      convert => [ "recipient-count", "integer" ]
      split => ["recipient_address", ";"]
    }
    date {
      match => [ "date-time", "ISO8601" ]
      timezone => "Europe/Moscow"
      remove_field => [ "date-time" ]
    }
  }
}
 
output {
  elasticsearch {
    hosts => ["127.0.0.1:9200", "127.0.0.2:9200"]
    manage_template => false
    index => "Exchange-%{+YYYY.MM.dd}"
  }
}

Link yang berguna:

• Bagaimana cara menginstal OpenJDK 11 di Windows?
• Unduh Logstash
• Elastic menggunakan opsi depricated UseConcMarkSweepGC # 36828
• NSSM
• Antrian persisten
• Plugin masukan Beats
• Logstash Bung, mana gergaji saya? Saya perlu membedah log saya
• Membedah plugin filter
• Persyaratan
• Mutasi plugin filter
• Plugin filter tanggal
• Plugin keluaran Elasticsearch