हम ईएलके और एक्सचेंज के मित्र हैं। भाग 2

मैं दोस्त एक्सचेंज और ईएलके कैसे बनाएं (शुरुआत) के बारे में अपनी कहानी जारी रखता हूं यहां). मैं आपको याद दिला दूं कि यह संयोजन बिना किसी हिचकिचाहट के बहुत बड़ी संख्या में लॉग को संसाधित करने में सक्षम है। इस बार हम इस बारे में बात करेंगे कि एक्सचेंज को लॉगस्टैश और किबाना घटकों के साथ कैसे काम कराया जाए।

ईएलके स्टैक में लॉगस्टैश का उपयोग लॉग को समझदारी से संसाधित करने और उन्हें दस्तावेज़ों के रूप में इलास्टिक में प्लेसमेंट के लिए तैयार करने के लिए किया जाता है, जिसके आधार पर किबाना में विभिन्न विज़ुअलाइज़ेशन बनाना सुविधाजनक होता है।

स्थापना

दो चरणों से मिलकर बनता है:

• OpenJDK पैकेज को स्थापित और कॉन्फ़िगर करना।
• लॉगस्टैश पैकेज को स्थापित और कॉन्फ़िगर करना।

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)

लॉगस्टैश पैकेज को स्थापित और कॉन्फ़िगर करना

लॉगस्टैश वितरण के साथ संग्रह फ़ाइल डाउनलोड करें अत:. संग्रह को डिस्क के रूट पर अनपैक किया जाना चाहिए। फ़ोल्डर में अनपैक करें C:Program Files यह इसके लायक नहीं है, लॉगस्टैश सामान्य रूप से प्रारंभ करने से इंकार कर देगा। फिर आपको फ़ाइल में प्रवेश करना होगा jvm.options जावा प्रक्रिया के लिए रैम आवंटित करने के लिए जिम्मेदार फिक्स। मैं सर्वर की आधी रैम निर्दिष्ट करने की अनुशंसा करता हूँ। यदि इसमें 16 जीबी रैम है, तो डिफ़ॉल्ट कुंजियाँ हैं:

-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: (अधिकतम कतार पृष्ठ आकार, डिफ़ॉल्ट मान 64एमबी है)।
• queue.drain: (सही/गलत - लॉगस्टैश को बंद करने से पहले कतार प्रसंस्करण को रोकने में सक्षम/अक्षम करता है। मैं इसे सक्षम करने की अनुशंसा नहीं करता, क्योंकि यह सीधे सर्वर शटडाउन की गति को प्रभावित करेगा)।
• queue.max_events: (कतार में घटनाओं की अधिकतम संख्या, डिफ़ॉल्ट 0 है (असीमित))।
• queue.max_bytes: (अधिकतम कतार आकार बाइट्स में, डिफ़ॉल्ट - 1024 एमबी (1 जीबी))।

यदि कॉन्फ़िगर किया गया है queue.max_events и queue.max_bytes, तो इनमें से किसी भी सेटिंग का मान पहुंचने पर संदेश कतार में स्वीकार नहीं किए जाते हैं। सतत कतारों के बारे में और जानें यहां.

कतार स्थापित करने के लिए जिम्मेदार logstash.yml के भाग का एक उदाहरण:

queue.type: persisted
queue.max_bytes: 10gb

समायोजन

लॉगस्टैश कॉन्फ़िगरेशन में आमतौर पर तीन भाग होते हैं, जो आने वाले लॉग को संसाधित करने के विभिन्न चरणों के लिए जिम्मेदार होते हैं: प्राप्त करना (इनपुट अनुभाग), पार्सिंग (फ़िल्टर अनुभाग) और इलास्टिक (आउटपुट अनुभाग) को भेजना। नीचे हम उनमें से प्रत्येक पर करीब से नज़र डालेंगे।

निवेश

हम फ़ाइलबीट एजेंटों से कच्चे लॉग के साथ आने वाली स्ट्रीम प्राप्त करते हैं। यह वह प्लगइन है जिसे हम इनपुट अनुभाग में इंगित करते हैं:

input {
  beats {
    port => 5044
  }
}

इस कॉन्फ़िगरेशन के बाद, लॉगस्टैश पोर्ट 5044 को सुनना शुरू कर देता है, और लॉग प्राप्त करते समय, फ़िल्टर अनुभाग की सेटिंग्स के अनुसार उन्हें संसाधित करता है। यदि आवश्यक हो, तो आप एसएसएल में फ़ाइलबिट से लॉग प्राप्त करने के लिए चैनल को लपेट सकते हैं। बीट्स प्लगइन सेटिंग्स के बारे में और पढ़ें यहां.

फ़िल्टर

सभी टेक्स्ट लॉग जो प्रसंस्करण के लिए दिलचस्प हैं, जिन्हें एक्सचेंज जेनरेट करता है, लॉग फ़ाइल में वर्णित फ़ील्ड के साथ सीएसवी प्रारूप में हैं। सीएसवी रिकॉर्ड को पार्स करने के लिए, लॉगस्टैश हमें तीन प्लगइन्स प्रदान करता है: काटना, सीएसवी और ग्रोक। पहला वाला सबसे ज़्यादा है उपवास, लेकिन केवल सबसे सरल लॉग को पार्स करने का कार्य करता है।
उदाहरण के लिए, यह निम्नलिखित रिकॉर्ड को दो भागों में विभाजित कर देगा (फ़ील्ड के अंदर अल्पविराम की उपस्थिति के कारण), जिसके कारण लॉग को गलत तरीके से पार्स किया जाएगा:

…,"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, जिसमें लॉग से एक प्रविष्टि शामिल है, और हम एक कस्टम फ़ील्ड जोड़ सकते हैं, जिसमें, उदाहरण के लिए, उस एप्लिकेशन का नाम होगा जिससे हम लॉग एकत्र करते हैं।

ट्रैकिंग लॉग के मामले में, सीएसवी प्लगइन का उपयोग करना बेहतर है; यह जटिल फ़ील्ड को सही ढंग से संसाधित कर सकता है:

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

उत्पादन

आउटपुट अनुभाग का उपयोग संसाधित लॉग को लॉग रिसीवर को भेजने के लिए किया जाता है। सीधे इलास्टिक पर भेजने के मामले में, एक प्लगइन का उपयोग किया जाता है 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}"
  }
}

उपयोगी लिंक्स:

• विंडोज़ पर ओपनजेडीके 11 कैसे स्थापित करें?
• लॉगस्टैश डाउनलोड करें
• इलास्टिक डीप्रिकेटेड विकल्प यूज़कॉन्कमार्कस्वीपजीसी #36828 का उपयोग करता है
• एनएसएसएम
• लगातार कतारें
• बीट्स इनपुट प्लगइन
• लॉगस्टैश दोस्त, मेरी चेनसॉ कहाँ है? मुझे अपने लॉग का विश्लेषण करने की आवश्यकता है
• विच्छेदन फ़िल्टर प्लगइन
• सशर्त,
• फ़िल्टर प्लगइन को म्यूट करें
• दिनांक फ़िल्टर प्लगइन
• इलास्टिक्स खोज आउटपुट प्लगइन

स्रोत: www.habr.com