हामी ELK र एक्सचेन्जका साथी हौं। भाग 2

म कसरी साथीहरू एक्सचेंज र ELK बनाउने बारे मेरो कथा जारी राख्छु (सुरुवात यहाँ)। म तपाईंलाई सम्झाउन दिनुहोस् कि यो संयोजन बिना कुनै हिचकिचाहट धेरै संख्यामा लगहरू प्रशोधन गर्न सक्षम छ। यस पटक हामी Logstash र Kibana कम्पोनेन्टहरूसँग एक्सचेन्ज कसरी काम गर्ने भन्ने बारे कुरा गर्नेछौं।

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 मा सफलतापूर्वक सुरु भयो।

अन्तिम स्थापना चरण: विन्डोज सेवाको रूपमा Logstash सुरु गर्नुहोस्। यो गर्न सकिन्छ, उदाहरणका लागि, प्याकेज प्रयोग गरेर NSSM:

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: (लाममा घटनाहरूको अधिकतम संख्या, पूर्वनिर्धारित ० हो (असीमित))।
• 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 प्लगइन प्रयोग गरेर हल गर्न सकिन्छ:

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

उपयोगी लिंकहरू:

• Windows मा OpenJDK 11 कसरी स्थापना गर्ने?
• Logstash डाउनलोड गर्नुहोस्
• इलास्टिकले डिप्रिकेटेड विकल्प UseConcMarkSweepGC #36828 प्रयोग गर्दछ
• NSSM
• लगातार लामहरू
• बिट इनपुट प्लगइन
• Logstash साथी, मेरो चेनसा कहाँ छ? मैले मेरो लगहरू विच्छेद गर्न आवश्यक छ
• फिल्टर प्लगइन विच्छेद गर्नुहोस्
• सर्तहरू
• म्यूटेट फिल्टर प्लगइन
• मिति फिल्टर प्लगइन
• इलास्टिकसर्च आउटपुट प्लगइन

स्रोत: www.habr.com