ഞങ്ങൾ ELK, എക്സ്ചേഞ്ച് എന്നിവയുമായി സുഹൃത്തുക്കളാണ്. ഭാഗം 2

ഫ്രണ്ട്സ് എക്സ്ചേഞ്ചും ELK ഉം എങ്ങനെ ഉണ്ടാക്കാം എന്നതിനെക്കുറിച്ചുള്ള എന്റെ കഥ ഞാൻ തുടരുന്നു (ആരംഭം ഇവിടെ). ഈ കോമ്പിനേഷന് ഒരു മടിയും കൂടാതെ വളരെ വലിയ ലോഗുകൾ പ്രോസസ്സ് ചെയ്യാൻ പ്രാപ്തമാണെന്ന് ഞാൻ നിങ്ങളെ ഓർമ്മിപ്പിക്കട്ടെ. ലോഗ്സ്റ്റാഷ്, കിബാന ഘടകങ്ങൾ ഉപയോഗിച്ച് എക്സ്ചേഞ്ച് എങ്ങനെ പ്രവർത്തിക്കാം എന്നതിനെക്കുറിച്ച് ഇത്തവണ നമ്മൾ സംസാരിക്കും.

ELK സ്റ്റാക്കിലെ ലോഗ്സ്റ്റാഷ്, ലോഗുകൾ ബുദ്ധിപരമായി പ്രോസസ്സ് ചെയ്യുന്നതിനും ഡോക്യുമെന്റുകളുടെ രൂപത്തിൽ ഇലാസ്റ്റിക് പ്ലെയ്‌സ്‌മെന്റിനായി തയ്യാറാക്കുന്നതിനും ഉപയോഗിക്കുന്നു, അതിന്റെ അടിസ്ഥാനത്തിൽ കിബാനയിൽ വിവിധ ദൃശ്യവൽക്കരണങ്ങൾ നിർമ്മിക്കുന്നത് സൗകര്യപ്രദമാണ്.

ഇൻസ്റ്റലേഷൻ

രണ്ട് ഘട്ടങ്ങൾ ഉൾക്കൊള്ളുന്നു:

• OpenJDK പാക്കേജ് ഇൻസ്റ്റാൾ ചെയ്യുകയും ക്രമീകരിക്കുകയും ചെയ്യുന്നു.
• ലോഗ്സ്റ്റാഷ് പാക്കേജ് ഇൻസ്റ്റാൾ ചെയ്യുകയും ക്രമീകരിക്കുകയും ചെയ്യുന്നു.

OpenJDK പാക്കേജ് ഇൻസ്റ്റാൾ ചെയ്യുകയും ക്രമീകരിക്കുകയും ചെയ്യുന്നു

OpenJDK പാക്കേജ് ഡൗൺലോഡ് ചെയ്യുകയും ഒരു പ്രത്യേക ഡയറക്ടറിയിലേക്ക് അൺപാക്ക് ചെയ്യുകയും വേണം. തുടർന്ന് ഈ ഡയറക്‌ടറിയിലേക്കുള്ള പാത വിൻഡോസ് ഓപ്പറേറ്റിംഗ് സിസ്റ്റത്തിന്റെ $env:Path, $env:JAVA_HOME വേരിയബിളുകളിൽ നൽകണം:

നമുക്ക് ജാവ പതിപ്പ് പരിശോധിക്കാം:

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)

ലോഗ്സ്റ്റാഷ് പാക്കേജ് ഇൻസ്റ്റാൾ ചെയ്യുകയും ക്രമീകരിക്കുകയും ചെയ്യുന്നു

Logstash വിതരണത്തോടൊപ്പം ആർക്കൈവ് ഫയൽ ഡൗൺലോഡ് ചെയ്യുക ഇവിടെ നിന്ന്. ആർക്കൈവ് ഡിസ്കിന്റെ റൂട്ടിലേക്ക് അൺപാക്ക് ചെയ്തിരിക്കണം. ഫോൾഡറിലേക്ക് അൺപാക്ക് ചെയ്യുക C:Program Files ഇത് വിലമതിക്കുന്നില്ല, ലോഗ്സ്റ്റാഷ് സാധാരണയായി ആരംഭിക്കാൻ വിസമ്മതിക്കും. അപ്പോൾ നിങ്ങൾ ഫയലിൽ പ്രവേശിക്കേണ്ടതുണ്ട് jvm.options ജാവ പ്രോസസ്സിനായി റാം അനുവദിക്കുന്നതിനുള്ള ഉത്തരവാദിത്തം പരിഹരിക്കുന്നു. സെർവറിന്റെ റാമിന്റെ പകുതി വ്യക്തമാക്കാൻ ഞാൻ ശുപാർശ ചെയ്യുന്നു. ബോർഡിൽ 16 GB റാം ഉണ്ടെങ്കിൽ, സ്ഥിരസ്ഥിതി കീകൾ ഇവയാണ്:

-Xms1g
-Xmx1g

പകരം വയ്ക്കണം:

-Xms8g
-Xmx8g

കൂടാതെ, വരിയിൽ അഭിപ്രായമിടുന്നത് ഉചിതമാണ് -XX:+UseConcMarkSweepGC. ഇതിനെക്കുറിച്ച് കൂടുതൽ ഇവിടെ. logstash.conf ഫയലിൽ ഒരു ഡിഫോൾട്ട് കോൺഫിഗറേഷൻ സൃഷ്ടിക്കുക എന്നതാണ് അടുത്ത ഘട്ടം:

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

ഈ കോൺഫിഗറേഷൻ ഉപയോഗിച്ച്, Logstash കൺസോളിൽ നിന്നുള്ള ഡാറ്റ വായിക്കുകയും ഒരു ശൂന്യമായ ഫിൽട്ടറിലൂടെ കടന്നുപോകുകയും അത് കൺസോളിലേക്ക് തിരികെ നൽകുകയും ചെയ്യുന്നു. ഈ കോൺഫിഗറേഷൻ ഉപയോഗിക്കുന്നത് ലോഗ്സ്റ്റാഷിന്റെ പ്രവർത്തനക്ഷമത പരിശോധിക്കും. ഇത് ചെയ്യുന്നതിന്, നമുക്ക് ഇത് ഇന്ററാക്ടീവ് മോഡിൽ പ്രവർത്തിപ്പിക്കാം:

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}

പോർട്ട് 9600-ൽ ലോഗ്സ്റ്റാഷ് വിജയകരമായി സമാരംഭിച്ചു.

അവസാന ഇൻസ്റ്റാളേഷൻ ഘട്ടം: ഒരു വിൻഡോസ് സേവനമായി ലോഗ്സ്റ്റാഷ് സമാരംഭിക്കുക. ഉദാഹരണത്തിന്, പാക്കേജ് ഉപയോഗിച്ച് ഇത് ചെയ്യാൻ കഴിയും എൻ‌എസ്‌എസ്എം:

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

തെറ്റ് സഹിഷ്ണുത

സോഴ്സ് സെർവറിൽ നിന്ന് ട്രാൻസ്ഫർ ചെയ്യുമ്പോൾ ലോഗുകളുടെ സുരക്ഷ പെർസിസ്റ്റന്റ് ക്യൂസ് മെക്കാനിസം ഉറപ്പാക്കുന്നു.

ഇത് എങ്ങനെ പ്രവർത്തിക്കുന്നു

ലോഗ് പ്രോസസ്സിംഗ് സമയത്ത് ക്യൂകളുടെ ലേഔട്ട് ഇതാണ്: ഇൻപുട്ട് → ക്യൂ → ഫിൽട്ടർ + ഔട്ട്പുട്ട്.

ഇൻപുട്ട് പ്ലഗിൻ ഒരു ലോഗ് ഉറവിടത്തിൽ നിന്ന് ഡാറ്റ സ്വീകരിക്കുന്നു, അത് ഒരു ക്യൂവിലേക്ക് എഴുതുന്നു, കൂടാതെ ഉറവിടത്തിലേക്ക് ഡാറ്റ ലഭിച്ചുവെന്ന് സ്ഥിരീകരണം അയയ്ക്കുന്നു.

ക്യൂവിൽ നിന്നുള്ള സന്ദേശങ്ങൾ ലോഗ്സ്റ്റാഷ് പ്രോസസ്സ് ചെയ്യുന്നു, ഫിൽട്ടറിലൂടെയും ഔട്ട്പുട്ട് പ്ലഗിനിലൂടെയും കടന്നുപോകുന്നു. ലോഗ് അയച്ചതായി ഔട്ട്പുട്ടിൽ നിന്ന് സ്ഥിരീകരണം ലഭിക്കുമ്പോൾ, ലോഗ്സ്റ്റാഷ് ക്യൂവിൽ നിന്ന് പ്രോസസ്സ് ചെയ്ത ലോഗ് നീക്കം ചെയ്യുന്നു. Logstash നിർത്തുകയാണെങ്കിൽ, സ്ഥിരീകരണം ലഭിക്കാത്ത എല്ലാ പ്രോസസ്സ് ചെയ്യാത്ത സന്ദേശങ്ങളും സന്ദേശങ്ങളും ക്യൂവിൽ നിലനിൽക്കും, അടുത്ത തവണ ആരംഭിക്കുമ്പോൾ Logstash അവ പ്രോസസ്സ് ചെയ്യുന്നത് തുടരും.

കമപ്പെടുത്തല്

ഫയലിലെ കീകൾ ഉപയോഗിച്ച് ക്രമീകരിക്കാവുന്നതാണ് C:Logstashconfiglogstash.yml:

• queue.type: (സാധ്യമായ മൂല്യങ്ങൾ - persisted и memory (default)).
• path.queue: (സി: ലോഗ്സ്റ്റാഷ്ക്യൂവിൽ സ്ഥിരസ്ഥിതിയായി സംഭരിച്ചിരിക്കുന്ന ക്യൂ ഫയലുകളുള്ള ഫോൾഡറിലേക്കുള്ള പാത).
• queue.page_capacity: (പരമാവധി ക്യൂ പേജ് വലുപ്പം, സ്ഥിര മൂല്യം 64mb ആണ്).
• queue.drain: (true/false - Logstash ഷട്ട് ഡൗൺ ചെയ്യുന്നതിന് മുമ്പ് ക്യൂ പ്രോസസ്സിംഗ് നിർത്തുന്നത് പ്രവർത്തനക്ഷമമാക്കുന്നു/അപ്രാപ്‌തമാക്കുന്നു. ഇത് പ്രവർത്തനക്ഷമമാക്കാൻ ഞാൻ ശുപാർശ ചെയ്യുന്നില്ല, കാരണം ഇത് സെർവർ ഷട്ട്ഡൗൺ വേഗതയെ നേരിട്ട് ബാധിക്കും).
• queue.max_events: (ക്യൂവിലെ പരമാവധി എണ്ണം ഇവന്റുകൾ, ഡിഫോൾട്ട് 0 ആണ് (പരിധിയില്ലാത്തത്)).
• queue.max_bytes: (പരമാവധി ക്യൂ വലുപ്പം ബൈറ്റുകളിൽ, സ്ഥിരസ്ഥിതി - 1024mb (1gb)).

ക്രമീകരിച്ചിട്ടുണ്ടെങ്കിൽ queue.max_events и queue.max_bytes, ഈ ക്രമീകരണങ്ങളിൽ ഏതെങ്കിലും മൂല്യം എത്തുമ്പോൾ സന്ദേശങ്ങൾ ക്യൂവിലേക്ക് സ്വീകരിക്കുന്നത് നിർത്തും. സ്ഥിരമായ ക്യൂകളെക്കുറിച്ച് കൂടുതലറിയുക ഇവിടെ.

ക്യൂ സജ്ജീകരിക്കുന്നതിന് ഉത്തരവാദിയായ logstash.yml-ന്റെ ഭാഗത്തിന്റെ ഒരു ഉദാഹരണം:

queue.type: persisted
queue.max_bytes: 10gb

കമപ്പെടുത്തല്

ലോഗ്സ്റ്റാഷ് കോൺഫിഗറേഷൻ സാധാരണയായി മൂന്ന് ഭാഗങ്ങൾ ഉൾക്കൊള്ളുന്നു, ഇൻകമിംഗ് ലോഗുകൾ പ്രോസസ്സ് ചെയ്യുന്നതിന്റെ വിവിധ ഘട്ടങ്ങൾക്ക് ഉത്തരവാദിത്തമുണ്ട്: സ്വീകരിക്കൽ (ഇൻപുട്ട് വിഭാഗം), പാഴ്സിംഗ് (ഫിൽട്ടർ വിഭാഗം), ഇലാസ്റ്റിക് (ഔട്ട്പുട്ട് വിഭാഗം) ലേക്ക് അയയ്ക്കൽ. ചുവടെ ഞങ്ങൾ അവ ഓരോന്നും സൂക്ഷ്മമായി പരിശോധിക്കും.

ഇൻപുട്ട്

ഫയൽബീറ്റ് ഏജന്റുകളിൽ നിന്ന് റോ ലോഗുകൾ ഉള്ള ഇൻകമിംഗ് സ്ട്രീം ഞങ്ങൾക്ക് ലഭിക്കുന്നു. ഈ പ്ലഗിൻ ആണ് ഞങ്ങൾ ഇൻപുട്ട് വിഭാഗത്തിൽ സൂചിപ്പിക്കുന്നത്:

input {
  beats {
    port => 5044
  }
}

ഈ കോൺഫിഗറേഷനുശേഷം, Logstash പോർട്ട് 5044 കേൾക്കാൻ തുടങ്ങുന്നു, കൂടാതെ ലോഗുകൾ സ്വീകരിക്കുമ്പോൾ, ഫിൽട്ടർ വിഭാഗത്തിന്റെ ക്രമീകരണങ്ങൾക്കനുസരിച്ച് അവ പ്രോസസ്സ് ചെയ്യുന്നു. ആവശ്യമെങ്കിൽ, SSL-ൽ ഫയൽബിറ്റിൽ നിന്ന് ലോഗുകൾ സ്വീകരിക്കുന്നതിന് നിങ്ങൾക്ക് ചാനൽ പൊതിയാവുന്നതാണ്. ബീറ്റ്സ് പ്ലഗിൻ ക്രമീകരണങ്ങളെക്കുറിച്ച് കൂടുതൽ വായിക്കുക ഇവിടെ.

അരിപ്പ

എക്‌സ്‌ചേഞ്ച് സൃഷ്‌ടിക്കുന്ന പ്രോസസ്സിംഗിന് താൽപ്പര്യമുള്ള എല്ലാ ടെക്‌സ്‌റ്റ് ലോഗുകളും ലോഗ് ഫയലിൽ തന്നെ വിവരിച്ചിരിക്കുന്ന ഫീൽഡുകളുള്ള csv ഫോർമാറ്റിലാണ്. csv റെക്കോർഡുകൾ പാഴ്‌സിംഗ് ചെയ്യുന്നതിന്, Logstash ഞങ്ങൾക്ക് മൂന്ന് പ്ലഗിനുകൾ വാഗ്ദാനം ചെയ്യുന്നു: വിച്ഛേദിക്കുക, csv, grok. ആദ്യത്തേതാണ് ഏറ്റവും കൂടുതൽ സ്വിഫ്റ്റ്, എന്നാൽ ഏറ്റവും ലളിതമായ ലോഗുകൾ മാത്രം പാഴ്‌സുചെയ്യുന്നത് നേരിടുന്നു.
ഉദാഹരണത്തിന്, ഇത് ഇനിപ്പറയുന്ന റെക്കോർഡിനെ രണ്ടായി വിഭജിക്കും (ഫീൽഡിനുള്ളിൽ ഒരു കോമയുടെ സാന്നിധ്യം കാരണം), അതിനാലാണ് ലോഗ് തെറ്റായി പാഴ്‌സ് ചെയ്യുന്നത്:

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

ലോഗുകൾ പാഴ്‌സ് ചെയ്യുമ്പോൾ ഇത് ഉപയോഗിക്കാം, ഉദാഹരണത്തിന്, IIS. ഈ സാഹചര്യത്തിൽ, ഫിൽട്ടർ വിഭാഗം ഇതുപോലെയാകാം:

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

ലോഗ്സ്റ്റാഷ് കോൺഫിഗറേഷൻ നിങ്ങളെ ഉപയോഗിക്കാൻ അനുവദിക്കുന്നു സോപാധിക പ്രസ്താവനകൾ, അതിനാൽ ഫയൽബീറ്റ് ടാഗ് ഉപയോഗിച്ച് ടാഗ് ചെയ്ത ലോഗുകൾ മാത്രമേ ഞങ്ങൾക്ക് ഡിസെക്റ്റ് പ്ലഗിനിലേക്ക് അയയ്ക്കാൻ കഴിയൂ. IIS. പ്ലഗിനിനുള്ളിൽ ഞങ്ങൾ ഫീൽഡ് മൂല്യങ്ങൾ അവയുടെ പേരുകളുമായി പൊരുത്തപ്പെടുത്തുന്നു, യഥാർത്ഥ ഫീൽഡ് ഇല്ലാതാക്കുക message, അതിൽ ലോഗിൽ നിന്നുള്ള ഒരു എൻട്രി അടങ്ങിയിരിക്കുന്നു, കൂടാതെ ഞങ്ങൾക്ക് ഒരു ഇഷ്‌ടാനുസൃത ഫീൽഡ് ചേർക്കാൻ കഴിയും, ഉദാഹരണത്തിന്, ഞങ്ങൾ ലോഗുകൾ ശേഖരിക്കുന്ന ആപ്ലിക്കേഷന്റെ പേര് അടങ്ങിയിരിക്കും.

ട്രാക്കിംഗ് ലോഗുകളുടെ കാര്യത്തിൽ, csv പ്ലഗിൻ ഉപയോഗിക്കുന്നതാണ് നല്ലത്; ഇതിന് സങ്കീർണ്ണമായ ഫീൽഡുകൾ ശരിയായി പ്രോസസ്സ് ചെയ്യാൻ കഴിയും:

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

പ്ലഗിനിനുള്ളിൽ ഞങ്ങൾ ഫീൽഡ് മൂല്യങ്ങൾ അവയുടെ പേരുകളുമായി പൊരുത്തപ്പെടുത്തുന്നു, യഥാർത്ഥ ഫീൽഡ് ഇല്ലാതാക്കുക message (കൂടാതെ വയലുകളും tenant-id и schema-version), അതിൽ ലോഗിൽ നിന്നുള്ള ഒരു എൻട്രി അടങ്ങിയിരിക്കുന്നു, കൂടാതെ ഞങ്ങൾക്ക് ഒരു ഇഷ്‌ടാനുസൃത ഫീൽഡ് ചേർക്കാൻ കഴിയും, ഉദാഹരണത്തിന്, ഞങ്ങൾ ലോഗുകൾ ശേഖരിക്കുന്ന ആപ്ലിക്കേഷന്റെ പേര് ഇതിൽ അടങ്ങിയിരിക്കും.

ഫിൽട്ടറിംഗ് ഘട്ടത്തിൽ നിന്ന് പുറത്തുകടക്കുമ്പോൾ, കിബാനയിൽ ദൃശ്യവൽക്കരണത്തിന് തയ്യാറായ ആദ്യ ഏകദേശരേഖയിൽ ഞങ്ങൾക്ക് ഡോക്യുമെന്റുകൾ ലഭിക്കും. ഇനിപ്പറയുന്നവ ഞങ്ങൾക്ക് നഷ്‌ടമാകും:

• സംഖ്യാ ഫീൽഡുകൾ ടെക്‌സ്‌റ്റായി അംഗീകരിക്കപ്പെടും, അത് അവയിലെ പ്രവർത്തനങ്ങളെ തടയുന്നു. അതായത്, വയലുകൾ time-taken IIS ലോഗ്, അതുപോലെ ഫീൽഡുകൾ recipient-count и total-bites ലോഗ് ട്രാക്കിംഗ്.
• സ്റ്റാൻഡേർഡ് ഡോക്യുമെന്റ് ടൈംസ്റ്റാമ്പിൽ ലോഗ് പ്രോസസ്സ് ചെയ്ത സമയം അടങ്ങിയിരിക്കും, സെർവർ വശത്ത് എഴുതിയ സമയമല്ല.
• ഫീൽഡ് recipient-address ഒരു നിർമ്മാണ സൈറ്റ് പോലെ കാണപ്പെടും, അത് അക്ഷരങ്ങളുടെ സ്വീകർത്താക്കളെ കണക്കാക്കാൻ വിശകലനം അനുവദിക്കുന്നില്ല.

ലോഗ് പ്രോസസ്സിംഗ് പ്രക്രിയയിൽ ഒരു ചെറിയ മാജിക് ചേർക്കേണ്ട സമയമാണിത്.

സംഖ്യാ ഫീൽഡുകൾ പരിവർത്തനം ചെയ്യുന്നു

ഡിസെക്റ്റ് പ്ലഗിന് ഒരു ഓപ്ഷൻ ഉണ്ട് convert_datatype, ഒരു ടെക്സ്റ്റ് ഫീൽഡ് ഡിജിറ്റൽ ഫോർമാറ്റിലേക്ക് പരിവർത്തനം ചെയ്യാൻ ഇത് ഉപയോഗിക്കാം. ഉദാഹരണത്തിന്, ഇതുപോലെ:

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

ഫീൽഡിൽ തീർച്ചയായും ഒരു സ്ട്രിംഗ് അടങ്ങിയിട്ടുണ്ടെങ്കിൽ മാത്രമേ ഈ രീതി അനുയോജ്യമാകൂ എന്നത് ഓർമിക്കേണ്ടതാണ്. ഓപ്‌ഷൻ ഫീൽഡുകളിൽ നിന്നുള്ള നൾ മൂല്യങ്ങൾ പ്രോസസ്സ് ചെയ്യുന്നില്ല കൂടാതെ ഒരു അപവാദം എറിയുന്നു.

ലോഗുകൾ ട്രാക്കുചെയ്യുന്നതിന്, ഫീൽഡുകൾ മുതൽ സമാനമായ പരിവർത്തന രീതി ഉപയോഗിക്കാതിരിക്കുന്നതാണ് നല്ലത് recipient-count и total-bites ശൂന്യമായിരിക്കാം. ഈ ഫീൽഡുകൾ പരിവർത്തനം ചെയ്യാൻ ഒരു പ്ലഗിൻ ഉപയോഗിക്കുന്നതാണ് നല്ലത് പരിവർത്തനം ചെയ്യുക:

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

സ്വീകർത്താവിന്റെ_വിലാസം വ്യക്തിഗത സ്വീകർത്താക്കളായി വിഭജിക്കുന്നു

മ്യൂട്ടേറ്റ് പ്ലഗിൻ ഉപയോഗിച്ചും ഈ പ്രശ്നം പരിഹരിക്കാൻ കഴിയും:

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

ടൈംസ്റ്റാമ്പ് മാറ്റുന്നു

ട്രാക്കിംഗ് ലോഗുകളുടെ കാര്യത്തിൽ, പ്ലഗിൻ ഉപയോഗിച്ച് പ്രശ്നം വളരെ എളുപ്പത്തിൽ പരിഹരിക്കപ്പെടും തീയതി, ഇത് ഫീൽഡിൽ എഴുതാൻ നിങ്ങളെ സഹായിക്കും timestamp ഫീൽഡിൽ നിന്ന് ആവശ്യമായ ഫോർമാറ്റിൽ തീയതിയും സമയവും date-time:

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

IIS ലോഗുകളുടെ കാര്യത്തിൽ, ഞങ്ങൾ ഫീൽഡ് ഡാറ്റ സംയോജിപ്പിക്കേണ്ടതുണ്ട് date и time മ്യൂട്ടേറ്റ് പ്ലഗിൻ ഉപയോഗിച്ച്, നമുക്ക് ആവശ്യമുള്ള സമയ മേഖല രജിസ്റ്റർ ചെയ്ത് ഈ ടൈം സ്റ്റാമ്പ് ഇടുക timestamp തീയതി പ്ലഗിൻ ഉപയോഗിക്കുന്നു:

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" ]
}

ഔട്ട്പുട്ട്

പ്രോസസ്സ് ചെയ്ത ലോഗുകൾ ലോഗ് റിസീവറിലേക്ക് അയയ്ക്കാൻ ഔട്ട്പുട്ട് വിഭാഗം ഉപയോഗിക്കുന്നു. ഇലാസ്റ്റിക് നേരിട്ട് അയയ്ക്കുന്ന സാഹചര്യത്തിൽ, ഒരു പ്ലഗിൻ ഉപയോഗിക്കുന്നു ഇലാസ്റ്റിക് തിരയൽ, ജനറേറ്റ് ചെയ്‌ത പ്രമാണം അയയ്‌ക്കുന്നതിനുള്ള സെർവർ വിലാസവും സൂചിക നാമ ടെംപ്ലേറ്റും ഇത് വ്യക്തമാക്കുന്നു:

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

അന്തിമ കോൺഫിഗറേഷൻ

അന്തിമ കോൺഫിഗറേഷൻ ഇതുപോലെ കാണപ്പെടും:

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

പ്രയോജനപ്രദമായ ലിങ്കുകൾ:

• വിൻഡോസിൽ OpenJDK 11 എങ്ങനെ ഇൻസ്റ്റാൾ ചെയ്യാം?
• Logstash ഡൗൺലോഡ് ചെയ്യുക
• ഇലാസ്റ്റിക് ഉപയോഗശൂന്യമായ ഓപ്ഷൻ UseConcMarkSweepGC #36828 ഉപയോഗിക്കുന്നു
• എൻ‌എസ്‌എസ്എം
• സ്ഥിരമായ ക്യൂകൾ
• ഇൻപുട്ട് പ്ലഗിൻ ബീറ്റ്സ്
• ലോഗ്സ്റ്റാഷ് സുഹൃത്തേ, എന്റെ ചെയിൻസോ എവിടെയാണ്? എനിക്ക് എന്റെ ലോഗുകൾ വിച്ഛേദിക്കേണ്ടതുണ്ട്
• ഫിൽട്ടർ പ്ലഗിൻ വിച്ഛേദിക്കുക
• വ്യവസ്ഥകൾ
• ഫിൽട്ടർ പ്ലഗിൻ മ്യൂട്ടേറ്റ് ചെയ്യുക
• തീയതി ഫിൽട്ടർ പ്ലഗിൻ
• ഇലാസ്റ്റിക് സെർച്ച് ഔട്ട്പുട്ട് പ്ലഗിൻ

അവലംബം: www.habr.com