અમે ELK અને એક્સચેન્જના મિત્રો છીએ. ભાગ 2

મિત્રો એક્સચેન્જ અને ELK કેવી રીતે બનાવવું તે વિશે હું મારી વાર્તા ચાલુ રાખું છું (શરૂઆત અહીં). ચાલો હું તમને યાદ કરાવું કે આ સંયોજન ખચકાટ વિના ખૂબ મોટી સંખ્યામાં લોગ પર પ્રક્રિયા કરવામાં સક્ષમ છે. આ વખતે આપણે એક્સચેન્જને લોગસ્ટેશ અને કિબાના ઘટકો સાથે કેવી રીતે કામ કરવું તે વિશે વાત કરીશું.

ELK સ્ટેકમાં લોગસ્ટેશનો ઉપયોગ લોગ પર બુદ્ધિપૂર્વક પ્રક્રિયા કરવા અને દસ્તાવેજોના રૂપમાં તેમને સ્થિતિસ્થાપકમાં પ્લેસમેન્ટ માટે તૈયાર કરવા માટે થાય છે, જેના આધારે કિબાનામાં વિવિધ વિઝ્યુલાઇઝેશન બનાવવાનું અનુકૂળ છે.

સ્થાપન

બે તબક્કાઓ સમાવે છે:

• OpenJDK પેકેજ ઇન્સ્ટોલ અને ગોઠવી રહ્યું છે.
• Logstash પેકેજ ઇન્સ્ટોલ અને ગોઠવી રહ્યું છે.

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 પેકેજ ઇન્સ્ટોલ અને ગોઠવી રહ્યું છે

Logstash વિતરણ સાથે આર્કાઇવ ફાઇલ ડાઉનલોડ કરો અહીંથી. આર્કાઇવ ડિસ્કના રુટ પર અનપેક થયેલ હોવું આવશ્યક છે. ફોલ્ડરમાં અનપૅક કરો C:Program Files તે મૂલ્યવાન નથી, Logstash સામાન્ય રીતે શરૂ થવાનો ઇનકાર કરશે. પછી તમારે ફાઇલ દાખલ કરવાની જરૂર છે jvm.options જાવા પ્રક્રિયા માટે RAM ફાળવવા માટે જવાબદાર સુધારાઓ. હું સર્વરની અડધી RAM નો ઉલ્લેખ કરવાની ભલામણ કરું છું. જો તેમાં બોર્ડ પર 16 GB ની RAM હોય, તો ડિફોલ્ટ કીઓ છે:

-Xms1g
-Xmx1g

સાથે બદલવું આવશ્યક છે:

-Xms8g
-Xmx8g

વધુમાં, લીટી બહાર ટિપ્પણી કરવાની સલાહ આપવામાં આવે છે -XX:+UseConcMarkSweepGC. આ વિશે વધુ અહીં. આગળનું પગલું એ logstash.conf ફાઇલમાં ડિફોલ્ટ રૂપરેખાંકન બનાવવાનું છે:

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

આ રૂપરેખાંકન સાથે, Logstash કન્સોલમાંથી ડેટા વાંચે છે, તેને ખાલી ફિલ્ટરમાંથી પસાર કરે છે અને તેને કન્સોલ પર પાછું આઉટપુટ કરે છે. આ ગોઠવણીનો ઉપયોગ 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}

Logstash પોર્ટ 9600 પર સફળતાપૂર્વક લોન્ચ થયું.

અંતિમ ઇન્સ્ટોલેશન પગલું: લોગસ્ટેશને Windows સેવા તરીકે લોંચ કરો. આ કરી શકાય છે, ઉદાહરણ તરીકે, પેકેજનો ઉપયોગ કરીને એનએસએસએમ:

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

સહનશીલતા દોષ

લોગની સલામતી જ્યારે સોર્સ સર્વરમાંથી ટ્રાન્સફર કરવામાં આવે છે ત્યારે પર્સિસ્ટન્ટ ક્યૂઝ મિકેનિઝમ દ્વારા સુનિશ્ચિત કરવામાં આવે છે.

તે કેવી રીતે કામ કરે છે

લોગ પ્રોસેસિંગ દરમિયાન કતારોનું લેઆઉટ છે: ઇનપુટ → કતાર → ફિલ્ટર + આઉટપુટ.

ઇનપુટ પ્લગઇન લોગ સ્ત્રોતમાંથી ડેટા મેળવે છે, તેને કતારમાં લખે છે અને પુષ્ટિ મોકલે છે કે ડેટા સ્ત્રોતને પ્રાપ્ત થયો છે.

કતારમાંથી સંદેશાઓ લોગસ્ટેશ દ્વારા પ્રક્રિયા કરવામાં આવે છે, ફિલ્ટર અને આઉટપુટ પ્લગઇનમાંથી પસાર થાય છે. જ્યારે આઉટપુટમાંથી પુષ્ટિ મળે છે કે લોગ મોકલવામાં આવ્યો છે, ત્યારે Logstash કતારમાંથી પ્રોસેસ્ડ લોગને દૂર કરે છે. જો Logstash બંધ થઈ જાય, તો બધા બિનપ્રક્રિયા કરાયેલા સંદેશાઓ અને સંદેશાઓ કે જેના માટે કોઈ પુષ્ટિ પ્રાપ્ત થઈ નથી તે કતારમાં રહે છે, અને Logstash આગલી વખતે જ્યારે તે શરૂ થશે ત્યારે તેની પ્રક્રિયા કરવાનું ચાલુ રાખશે.

ગોઠવણ

ફાઇલમાં કીઓ દ્વારા એડજસ્ટેબલ C:Logstashconfiglogstash.yml:

• queue.type: (શક્ય મૂલ્યો - persisted и memory (default)).
• path.queue: (કતાર ફાઇલો સાથે ફોલ્ડરનો પાથ, જે મૂળભૂત રીતે C:Logstashqueue માં સંગ્રહિત છે).
• 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

ગોઠવણ

Logstash રૂપરેખાંકનમાં સામાન્ય રીતે ત્રણ ભાગોનો સમાવેશ થાય છે, જે ઇનકમિંગ લોગની પ્રક્રિયાના વિવિધ તબક્કાઓ માટે જવાબદાર છે: પ્રાપ્ત કરવું (ઇનપુટ વિભાગ), પાર્સિંગ (ફિલ્ટર વિભાગ) અને સ્થિતિસ્થાપક (આઉટપુટ વિભાગ) પર મોકલવું. નીચે આપણે તેમાંના દરેકને નજીકથી જોઈશું.

ઇનપુટ

અમે ફાઇલબીટ એજન્ટો પાસેથી કાચા લૉગ્સ સાથે ઇનકમિંગ સ્ટ્રીમ પ્રાપ્ત કરીએ છીએ. તે આ પ્લગઇન છે જે અમે ઇનપુટ વિભાગમાં સૂચવીએ છીએ:

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

Logstash રૂપરેખાંકન તમને ઉપયોગ કરવાની મંજૂરી આપે છે શરતી નિવેદનો, તેથી અમે વિચ્છેદન પ્લગઇન પર ફાઇલબીટ ટૅગ સાથે ટૅગ કરેલા લૉગ્સ જ મોકલી શકીએ છીએ 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}"
  }
}

ઉપયોગી લિંક્સ:

• વિન્ડોઝ પર ઓપનજેડીકે 11 કેવી રીતે ઇન્સ્ટોલ કરવું?
• લોગસ્ટેશ ડાઉનલોડ કરો
• સ્થિતિસ્થાપક વંચિત વિકલ્પનો ઉપયોગ કરે છે UseConcMarkSweepGC #36828
• એનએસએસએમ
• સતત કતારો
• બીટ્સ ઇનપુટ પ્લગઇન
• લોગસ્ટેશ દોસ્ત, મારી ચેઇનસો ક્યાં છે? મારે મારા લોગનું વિચ્છેદન કરવું છે
• ફિલ્ટર પ્લગઇનને ડિસેક્ટ કરો
• કંડિશનલ્સ
• ફિલ્ટર પ્લગઇનને બદલો
• તારીખ ફિલ્ટર પ્લગઇન
• સ્થિતિસ્થાપક શોધ આઉટપુટ પ્લગઇન

સોર્સ: www.habr.com