நாங்கள் ELK மற்றும் Exchange உடன் நண்பர்கள். பகுதி 2

நண்பர்கள் பரிமாற்றம் மற்றும் ELK (ஆரம்பத்தில்) எப்படி உருவாக்குவது என்பது பற்றிய எனது கதையைத் தொடர்கிறேன் இங்கே) இந்த கலவையானது அதிக எண்ணிக்கையிலான பதிவுகளை தயக்கமின்றி செயலாக்கும் திறன் கொண்டது என்பதை நினைவூட்டுகிறேன். இந்த நேரத்தில் Logstash மற்றும் Kibana கூறுகளுடன் எவ்வாறு Exchange வேலை செய்வது என்பது பற்றி பேசுவோம்.

ELK அடுக்கில் உள்ள லாக்ஸ்டாஷ், பதிவுகளை புத்திசாலித்தனமாக செயலாக்கவும், ஆவணங்களின் வடிவத்தில் மீள்நிலையில் வைக்க அவற்றைத் தயாரிக்கவும் பயன்படுத்தப்படுகிறது, அதன் அடிப்படையில் கிபானாவில் பல்வேறு காட்சிப்படுத்தல்களை உருவாக்குவது வசதியானது.

நிறுவல்

இரண்டு நிலைகளைக் கொண்டுள்ளது:

• OpenJDK தொகுப்பை நிறுவுதல் மற்றும் கட்டமைத்தல்.
• Logstash தொகுப்பை நிறுவுதல் மற்றும் கட்டமைத்தல்.

OpenJDK தொகுப்பை நிறுவுதல் மற்றும் கட்டமைத்தல்

OpenJDK தொகுப்பு பதிவிறக்கம் செய்யப்பட்டு ஒரு குறிப்பிட்ட கோப்பகத்தில் திறக்கப்பட வேண்டும். இந்த கோப்பகத்திற்கான பாதையானது Windows இயங்குதளத்தின் $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 ஜாவா செயல்முறைக்கு ரேம் ஒதுக்குவதற்கு பொறுப்பான திருத்தங்கள். சேவையகத்தின் RAM இல் பாதியைக் குறிப்பிட பரிந்துரைக்கிறேன். போர்டில் 16 ஜிபி ரேம் இருந்தால், இயல்புநிலை விசைகள்:

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

போர்ட் 9600 இல் Logstash வெற்றிகரமாக தொடங்கப்பட்டது.

இறுதி நிறுவல் படி: விண்டோஸ் சேவையாக Logstash ஐ துவக்கவும். உதாரணமாக, தொகுப்பைப் பயன்படுத்தி இதைச் செய்யலாம் என்.எஸ்.எஸ்.எம்:

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

தவறு சகிப்புத்தன்மை

மூல சேவையகத்திலிருந்து மாற்றப்படும் போது பதிவுகளின் பாதுகாப்பு நிலையான வரிசைகள் பொறிமுறையால் உறுதி செய்யப்படுகிறது.

அது எப்படி வேலை செய்கிறது

பதிவு செயலாக்கத்தின் போது வரிசைகளின் தளவமைப்பு: உள்ளீடு → வரிசை → வடிகட்டி + வெளியீடு.

உள்ளீட்டு செருகுநிரல் ஒரு பதிவு மூலத்திலிருந்து தரவைப் பெறுகிறது, அதை ஒரு வரிசையில் எழுதுகிறது, மேலும் தரவு பெறப்பட்டதற்கான உறுதிப்படுத்தலை மூலத்திற்கு அனுப்புகிறது.

வரிசையில் இருந்து வரும் செய்திகள் Logstash மூலம் செயலாக்கப்பட்டு, வடிகட்டி மற்றும் வெளியீடு செருகுநிரல் வழியாக அனுப்பப்படும். பதிவு அனுப்பப்பட்டதாக வெளியீட்டில் இருந்து உறுதிப்படுத்தலைப் பெறும்போது, ​​வரிசையிலிருந்து செயலாக்கப்பட்ட பதிவை Logstash நீக்குகிறது. Logstash நிறுத்தப்பட்டால், எந்த உறுதிப்படுத்தலும் பெறப்படாத அனைத்து செயலாக்கப்படாத செய்திகளும் செய்திகளும் வரிசையில் இருக்கும், மேலும் Logstash அடுத்த முறை தொடங்கும் போது அவற்றைச் செயல்படுத்தும்.

சரிசெய்தல்

கோப்பில் உள்ள விசைகள் மூலம் சரிசெய்யக்கூடியது C:Logstashconfiglogstash.yml:

• queue.type: (சாத்தியமான மதிப்புகள் - persisted и memory (default)).
• path.queue: (இயல்புநிலையாக C:Logstashqueue இல் சேமிக்கப்படும் வரிசை கோப்புகள் கொண்ட கோப்புறைக்கான பாதை).
• queue.page_capacity: (அதிகபட்ச வரிசை பக்க அளவு, இயல்புநிலை மதிப்பு 64mb).
• queue.drain: (உண்மை/தவறு - 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 இல் filebit இலிருந்து பதிவுகளைப் பெறுவதற்கான சேனலை நீங்கள் மடிக்கலாம். பீட்ஸ் செருகுநிரல் அமைப்புகளைப் பற்றி மேலும் படிக்கவும் இங்கே.

வடிகட்டி

Exchange உருவாக்கும் செயலாக்கத்திற்கு ஆர்வமுள்ள அனைத்து உரை பதிவுகளும் பதிவு கோப்பில் விவரிக்கப்பட்டுள்ள புலங்களுடன் 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 கட்டமைப்பு உங்களை பயன்படுத்த அனுமதிக்கிறது நிபந்தனை அறிக்கைகள், எனவே filebeat குறிச்சொல்லுடன் குறியிடப்பட்ட பதிவுகளை மட்டுமே பிரித்தெடுக்கும் செருகுநிரலுக்கு அனுப்ப முடியும். 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 ஒரு கட்டுமான தளம் போல் இருக்கும், இது கடிதங்களின் பெறுநர்களை கணக்கிட பகுப்பாய்வு செய்ய அனுமதிக்காது.

பதிவு செயலாக்க செயல்முறைக்கு ஒரு சிறிய மந்திரத்தை சேர்க்க வேண்டிய நேரம் இது.

எண் புலங்களை மாற்றுதல்

dissect செருகுநிரலுக்கு ஒரு விருப்பம் உள்ளது 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" ]
}

வெளியீடு

செயலாக்கப்பட்ட பதிவுகளை பதிவு பெறுநருக்கு அனுப்ப வெளியீட்டுப் பிரிவு பயன்படுத்தப்படுகிறது. எலாஸ்டிக் நேரடியாக அனுப்பினால், ஒரு சொருகி பயன்படுத்தப்படுகிறது elasticsearch, இது உருவாக்கப்பட்ட ஆவணத்தை அனுப்புவதற்கான சேவையக முகவரி மற்றும் குறியீட்டு பெயர் டெம்ப்ளேட்டைக் குறிப்பிடுகிறது:

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 ஐ எவ்வாறு நிறுவுவது?
• லாக்ஸ்டாஷைப் பதிவிறக்கவும்
• மீள்தன்மை நீக்கப்பட்ட விருப்பமான UseConcMarkSweepGC #36828 ஐப் பயன்படுத்துகிறது
• என்.எஸ்.எஸ்.எம்
• நிலையான வரிசைகள்
• உள்ளீடு சொருகி அடிக்கிறது
• லாக்ஸ்டாஷ் நண்பரே, எனது செயின்சா எங்கே? எனது பதிவுகளை நான் பிரிக்க வேண்டும்
• வடிகட்டி செருகுநிரலைப் பிரிக்கவும்
• conditionals
• வடிகட்டி செருகுநிரலை மாற்றவும்
• தேதி வடிகட்டி செருகுநிரல்
• மீள் தேடல் வெளியீட்டு செருகுநிரல்

ஆதாரம்: www.habr.com