आम्ही ईएलके आणि एक्सचेंजचे मित्र आहोत. भाग 2

मित्र एक्सचेंज आणि ELK कसे बनवायचे याबद्दल मी माझी कथा सुरू ठेवतो (सुरुवाती येथे). मी तुम्हाला आठवण करून देतो की हे संयोजन संकोच न करता खूप मोठ्या संख्येने लॉगवर प्रक्रिया करण्यास सक्षम आहे. यावेळी आम्ही लॉगस्टॅश आणि किबाना घटकांसह एक्सचेंज कसे कार्य करावे याबद्दल बोलू.

ईएलके स्टॅकमधील लॉगस्टॅशचा वापर लॉगवर बुद्धिमानपणे प्रक्रिया करण्यासाठी आणि कागदपत्रांच्या स्वरूपात लवचिक मध्ये प्लेसमेंटसाठी तयार करण्यासाठी केला जातो, ज्याच्या आधारावर किबानामध्ये विविध व्हिज्युअलायझेशन तयार करणे सोयीचे आहे.

सेटिंग

दोन टप्प्यांचा समावेश आहे:

• OpenJDK पॅकेज स्थापित करणे आणि कॉन्फिगर करणे.
• लॉगस्टॅश पॅकेज स्थापित करणे आणि कॉन्फिगर करणे.

OpenJDK पॅकेज स्थापित करणे आणि कॉन्फिगर करणे

OpenJDK पॅकेज एका विशिष्ट निर्देशिकेत डाउनलोड आणि अनपॅक केलेले असणे आवश्यक आहे. नंतर या निर्देशिकेचा मार्ग Windows ऑपरेटिंग सिस्टमच्या $env:Path आणि $env:JAVA_HOME व्हेरिएबल्समध्ये प्रविष्ट केला पाहिजे:

चला 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)

लॉगस्टॅश पॅकेज स्थापित करणे आणि कॉन्फिगर करणे

लॉगस्टॅश वितरणासह संग्रहण फाइल डाउनलोड करा येथून. संग्रहण डिस्कच्या रूटवर अनपॅक केलेले असणे आवश्यक आहे. फोल्डरमध्ये अनपॅक करा C:Program Files हे फायदेशीर नाही, लॉगस्टॅश सामान्यपणे सुरू करण्यास नकार देईल. मग आपल्याला फाइलमध्ये प्रवेश करणे आवश्यक आहे jvm.options Java प्रक्रियेसाठी RAM वाटप करण्यासाठी जबाबदार निराकरणे. मी सर्व्हरची अर्धी रॅम निर्दिष्ट करण्याची शिफारस करतो. बोर्डवर 16 GB RAM असल्यास, डीफॉल्ट की आहेत:

-Xms1g
-Xmx1g

सह पुनर्स्थित करणे आवश्यक आहे:

-Xms8g
-Xmx8g

याव्यतिरिक्त, ओळ बाहेर टिप्पणी सल्ला दिला आहे -XX:+UseConcMarkSweepGC. याबद्दल अधिक येथे. पुढील पायरी म्हणजे logstash.conf फाइलमध्ये डीफॉल्ट कॉन्फिगरेशन तयार करणे:

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

या कॉन्फिगरेशनसह, लॉगस्टॅश कन्सोलमधील डेटा वाचतो, तो रिकाम्या फिल्टरमधून जातो आणि तो परत कन्सोलवर आउटपुट करतो. हे कॉन्फिगरेशन वापरल्याने लॉगस्टॅशच्या कार्यक्षमतेची चाचणी होईल. हे करण्यासाठी, ते परस्परसंवादी मोडमध्ये चालवूया:

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!

चुकीची सहनशीलता

स्त्रोत सर्व्हरवरून हस्तांतरित केल्यावर लॉगची सुरक्षितता पर्सिस्टंट क्यू मेकॅनिझमद्वारे सुनिश्चित केली जाते.

हे कसे कार्य करते

लॉग प्रोसेसिंग दरम्यान रांगांचे लेआउट आहे: इनपुट → रांग → फिल्टर + आउटपुट.

इनपुट प्लगइन लॉग स्त्रोताकडून डेटा प्राप्त करतो, तो एका रांगेत लिहितो आणि स्त्रोताला डेटा प्राप्त झाल्याची पुष्टी पाठवतो.

रांगेतील संदेशांवर लॉगस्टॅशद्वारे प्रक्रिया केली जाते, फिल्टर आणि आउटपुट प्लगइनमधून पास केले जाते. लॉग पाठवला गेल्याची आउटपुट वरून पुष्टी प्राप्त करताना, लॉगस्टॅश प्रक्रिया केलेला लॉग रांगेतून काढून टाकतो. लॉगस्टॅश थांबल्यास, सर्व प्रक्रिया न केलेले संदेश आणि संदेश ज्यासाठी कोणतेही पुष्टीकरण प्राप्त झाले नाही ते रांगेत राहतील आणि लॉगस्टॅश पुढील वेळी सुरू झाल्यावर त्यावर प्रक्रिया करणे सुरू ठेवेल.

समायोजन

फाईलमधील की द्वारे समायोज्य C:Logstashconfiglogstash.yml:

• queue.type: (संभाव्य मूल्ये - persisted и memory (default)).
• path.queue: (डीफॉल्टनुसार C:Logstashqueue मध्ये संग्रहित केलेल्या रांगेतील फाइल्ससह फोल्डरचा मार्ग).
• queue.page_capacity: (कमाल रांगेचे पृष्ठ आकार, डीफॉल्ट मूल्य 64mb आहे).
• queue.drain: (सत्य/असत्य - लॉगस्टॅश बंद करण्यापूर्वी रांग प्रक्रिया थांबवणे सक्षम/अक्षम करते. मी ते सक्षम करण्याची शिफारस करत नाही, कारण याचा थेट परिणाम सर्व्हर बंद होण्याच्या गतीवर होईल).
• 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
  }
}

या कॉन्फिगरेशननंतर, लॉगस्टॅश पोर्ट 5044 ऐकण्यास प्रारंभ करते आणि लॉग प्राप्त करताना, फिल्टर विभागाच्या सेटिंग्जनुसार त्यावर प्रक्रिया करते. आवश्यक असल्यास, SSL मध्ये फाइलबिट वरून लॉग प्राप्त करण्यासाठी तुम्ही चॅनेल गुंडाळू शकता. बीट्स प्लगइन सेटिंग्जबद्दल अधिक वाचा येथे.

फिल्टर

एक्सचेंजद्वारे तयार केलेले सर्व मजकूर लॉग जे प्रक्रियेसाठी मनोरंजक आहेत ते लॉग फाइलमध्येच वर्णन केलेल्या फील्डसह csv फॉरमॅटमध्ये आहेत. csv रेकॉर्ड पार्स करण्यासाठी, Logstash आम्हाला तीन प्लगइन ऑफर करते: विच्छेदन, csv आणि grok. पहिला सर्वात जास्त आहे द्रुत, परंतु फक्त सर्वात सोप्या लॉगचे पार्सिंग सह copes.
उदाहरणार्थ, ते खालील रेकॉर्ड दोनमध्ये विभाजित करेल (फील्डमध्ये स्वल्पविरामाच्या उपस्थितीमुळे), ज्यामुळे लॉगचे चुकीचे विश्लेषण केले जाईल:

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

वैयक्तिक प्राप्तकर्त्यांमध्ये recipient_address विभाजित करणे

ही समस्या म्यूटेट प्लगइन वापरून देखील सोडविली जाऊ शकते:

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 कसे इंस्टॉल करावे?
• लॉगस्टॅश डाउनलोड करा
• लवचिक वंचित पर्याय वापरते UseConcMarkSweepGC #36828
• एनएसएसएम
• सततच्या रांगा
• बीट्स इनपुट प्लगइन
• लॉगस्टॅश मित्रा, माझा चेनसॉ कुठे आहे? मला माझ्या नोंदींचे विच्छेदन करावे लागेल
• फिल्टर प्लगइनचे विच्छेदन करा
• सशर्त
• म्यूटेट फिल्टर प्लगइन
• तारीख फिल्टर प्लगइन
• इलास्टिकसर्च आउटपुट प्लगइन

स्त्रोत: www.habr.com