परिचय

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

हमने इसकी सभी क्षमताओं का वर्णन करने का लक्ष्य निर्धारित नहीं किया है, बल्कि हम विशेष रूप से व्यावहारिक समस्याओं को हल करने पर ध्यान केंद्रित करना चाहते हैं। यह इस तथ्य के कारण है कि भले ही काफी बड़ी मात्रा में दस्तावेज और तैयार छवियां मौजूद हैं, फिर भी इसमें काफी कमियां हैं, कम से कम हमने उन्हें ढूंढ लिया है।

हमने डॉकर-कंपोज़ के माध्यम से स्टैक को तैनात किया। इसके अलावा, हमारे पास एक अच्छी तरह से लिखा गया docker-compose.yml था, जिसने हमें लगभग बिना किसी समस्या के स्टैक बढ़ाने की अनुमति दी। और हमें ऐसा लग रहा था कि जीत पहले से ही करीब थी, अब हम अपनी आवश्यकताओं के अनुरूप इसमें थोड़ा बदलाव करेंगे और बस इतना ही।

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

तो, हमने लॉगस्टैश से शुरुआत की।

पर्यावरण, परिनियोजन, एक कंटेनर में लॉगस्टैश चलाना

परिनियोजन के लिए हम docker-compose का उपयोग करते हैं; यहां वर्णित प्रयोग MacOS और Ubuntu 18.0.4 पर किए गए थे।

लॉगस्टैश छवि जो हमारे मूल docker-compose.yml में पंजीकृत थी वह docker.elastic.co/logstash/logstash:6.3.2 है

हम इसका प्रयोग प्रयोगों के लिए करेंगे.

हमने लॉगस्टैश चलाने के लिए एक अलग docker-compose.yml लिखा। बेशक, छवि को कमांड लाइन से लॉन्च करना संभव था, लेकिन हम एक विशिष्ट समस्या का समाधान कर रहे थे, जहां हम डॉकर-कंपोज़ से सब कुछ चलाते हैं।

कॉन्फ़िगरेशन फ़ाइलों के बारे में संक्षेप में

विवरण के अनुसार, लॉगस्टैश को या तो एक चैनल के लिए चलाया जा सकता है, जिस स्थिति में इसे *.conf फ़ाइल को पास करने की आवश्यकता होती है, या कई चैनलों के लिए, जिस स्थिति में इसे पाइपलाइन.yml फ़ाइल को पास करने की आवश्यकता होती है, जो बदले में , प्रत्येक चैनल के लिए .conf फ़ाइलों से लिंक होगा।
हमने दूसरा रास्ता अपनाया. यह हमें अधिक सार्वभौमिक और मापनीय लगा। इसलिए, हमने पाइपलाइन.वाईएमएल बनाई, और एक पाइपलाइन निर्देशिका बनाई जिसमें हम प्रत्येक चैनल के लिए .conf फ़ाइलें डालेंगे।

कंटेनर के अंदर एक और कॉन्फ़िगरेशन फ़ाइल है - logstash.yml। हम इसे छूते नहीं हैं, हम इसे वैसे ही उपयोग करते हैं।

तो, हमारी निर्देशिका संरचना:

इनपुट डेटा प्राप्त करने के लिए, अभी हम मानते हैं कि यह पोर्ट 5046 पर टीसीपी है, और आउटपुट के लिए हम stdout का उपयोग करेंगे।

यहां पहले लॉन्च के लिए एक सरल कॉन्फ़िगरेशन है। क्योंकि प्रारंभिक कार्य लॉन्च करना है।

तो, हमारे पास यह docker-compose.yml है

version: '3'

networks:
  elk:

volumes:
  elasticsearch:
    driver: local

services:

  logstash:
    container_name: logstash_one_channel
    image: docker.elastic.co/logstash/logstash:6.3.2
    networks:
      	- elk
    ports:
      	- 5046:5046
    volumes:
      	- ./config/pipelines.yml:/usr/share/logstash/config/pipelines.yml:ro
	- ./config/pipelines:/usr/share/logstash/config/pipelines:ro

हम यहाँ क्या देखते हैं?

नेटवर्क और वॉल्यूम मूल docker-compose.yml (जहां संपूर्ण स्टैक लॉन्च किया गया है) से लिए गए थे और मुझे लगता है कि वे यहां समग्र तस्वीर को बहुत प्रभावित नहीं करते हैं।
हम docker.elastic.co/logstash/logstash:6.3.2 छवि से एक लॉगस्टैश सेवा बनाते हैं और इसे logstash_one_channel नाम देते हैं।
हम कंटेनर के अंदर पोर्ट 5046 को उसी आंतरिक पोर्ट पर अग्रेषित करते हैं।
हम अपनी पाइप कॉन्फ़िगरेशन फ़ाइल ./config/pipelines.yml को कंटेनर के अंदर फ़ाइल /usr/share/logstash/config/pipelines.yml पर मैप करते हैं, जहां लॉगस्टैश इसे उठाएगा और इसे केवल पढ़ने के लिए बना देगा, बस मामले में।
हम ./config/pipelines निर्देशिका को, जहां हमारे पास चैनल सेटिंग्स वाली फ़ाइलें हैं, /usr/share/logstash/config/pipelines निर्देशिका में मैप करते हैं और इसे केवल पढ़ने के लिए भी बनाते हैं।

पाइपलाइन.yml फ़ाइल

- pipeline.id: HABR
  pipeline.workers: 1
  pipeline.batch.size: 1
  path.config: "./config/pipelines/habr_pipeline.conf"

HABR पहचानकर्ता वाला एक चैनल और इसकी कॉन्फ़िगरेशन फ़ाइल का पथ यहां वर्णित है।

और अंत में फ़ाइल "./config/pipelines/habr_pipeline.conf"

input {
  tcp {
    port => "5046"
   }
  }
filter {
  mutate {
    add_field => [ "habra_field", "Hello Habr" ]
    }
  }
output {
  stdout {
      
    }
  }

आइए अभी इसके विवरण में न जाएं, आइए इसे चलाने का प्रयास करें:

docker-compose up

हम क्या देखते हैं?

कंटेनर शुरू हो गया है. हम इसके संचालन की जांच कर सकते हैं:

echo '13123123123123123123123213123213' | nc localhost 5046

और हम कंटेनर कंसोल में प्रतिक्रिया देखते हैं:

लेकिन साथ ही, हम यह भी देखते हैं:

लॉगस्टैश_वन_चैनल | [2019-04-29टी11:28:59,790][त्रुटि][लॉगस्टैश.लाइसेंसचेकर.लाइसेंसरीडर] लाइसेंस सर्वर से लाइसेंस जानकारी प्राप्त करने में असमर्थ {:संदेश=>"इलास्टिकसर्च पहुंच योग्य नहीं: [http://elasticsearch:9200/][मैन्टिकोर ::रिज़ॉल्यूशन विफलता] इलास्टिक्स खोज", ...

लॉगस्टैश_वन_चैनल | [2019-04-29टी11:28:59,894][सूचना ][लॉगस्टैश.पाइपलाइन ] पाइपलाइन सफलतापूर्वक शुरू हुई {:पाइपलाइन_आईडी=>"।निगरानी-लॉगस्टैश", :थ्रेड=>"# "}

लॉगस्टैश_वन_चैनल | [2019-04-29टी11:28:59,988][सूचना ][लॉगस्टैश.एजेंट ] पाइपलाइनें चल रही हैं {:count=>2, :running_pipelines=>[:HABR, :.monitoring-logstash'], :non_running_pipelines=>[ ]}
लॉगस्टैश_वन_चैनल | [2019-04-29T11:29:00,015][ERROR][logstash.inputs.metrics] एक्स-पैक लॉगस्टैश पर स्थापित है लेकिन इलास्टिक्स खोज पर नहीं। मॉनिटरिंग सुविधा का उपयोग करने के लिए कृपया इलास्टिक्स खोज पर एक्स-पैक स्थापित करें। अन्य सुविधाएं उपलब्ध हो सकती हैं.
लॉगस्टैश_वन_चैनल | [2019-04-29टी11:29:00,526][जानकारी ][लॉगस्टैश.एजेंट] लॉगस्टैश एपीआई एंडपॉइंट सफलतापूर्वक शुरू हुआ {:पोर्ट=>9600}
लॉगस्टैश_वन_चैनल | [2019-04-29टी11:29:04,478][सूचना ][logstash.outputs.elasticsearch] इलास्टिक्स खोज कनेक्शन काम कर रहा है या नहीं यह देखने के लिए स्वास्थ्य जांच चला रहा है {:healthcheck_url=>http://elasticsearch:9200/, :path=> "/"}
लॉगस्टैश_वन_चैनल | [2019-04-29T11:29:04,487][चेतावनी][logstash.outputs.elasticsearch] मृत ES इंस्टेंस से कनेक्शन को पुनर्जीवित करने का प्रयास किया गया, लेकिन एक त्रुटि मिली। {:यूआरएल=>“Elasticsearch:9200/", :error_type=>LogStash::Outputs::ElasticSearch::HttpClient::Pool::HostUnreachableError, :error=>"Elasticsearch Unreachable: [http://elasticsearch:9200/][Manticore::ResolutionFairure] इलास्टिक्स खोज"}
लॉगस्टैश_वन_चैनल | [2019-04-29टी11:29:04,704][सूचना ][logstash.licensechecker.licensereader] इलास्टिक्स खोज कनेक्शन काम कर रहा है या नहीं यह देखने के लिए स्वास्थ्य जांच चला रहा है {:healthcheck_url=>http://elasticsearch:9200/, :path=> "/"}
लॉगस्टैश_वन_चैनल | [2019-04-29टी11:29:04,710][चेतावनी ][logstash.licensechecker.licensereader] मृत ईएस इंस्टेंस से कनेक्शन को पुनर्जीवित करने का प्रयास किया गया, लेकिन एक त्रुटि मिली। {:url=>“Elasticsearch:9200/", :error_type=>LogStash::Outputs::ElasticSearch::HttpClient::Pool::HostUnreachableError, :error=>"Elasticsearch Unreachable: [http://elasticsearch:9200/][Manticore::ResolutionFairure] इलास्टिक्स खोज"}

और हमारा लॉग हर समय खराब हो रहा है।

यहां मैंने हरे रंग में उस संदेश को हाइलाइट किया है कि पाइपलाइन सफलतापूर्वक लॉन्च हो गई है, लाल रंग में त्रुटि संदेश और पीले रंग में संपर्क करने के प्रयास के बारे में संदेश को हाइलाइट किया है। Elasticsearch: 9200।
ऐसा इसलिए होता है क्योंकि छवि में शामिल logstash.conf में इलास्टिक्स खोज उपलब्धता की जांच होती है। आख़िरकार, लॉगस्टैश मानता है कि यह एल्क स्टैक के हिस्से के रूप में काम करता है, लेकिन हमने इसे अलग कर दिया।

यह काम करना संभव है, लेकिन यह सुविधाजनक नहीं है।

इसका समाधान XPACK_MONITORING_ENABLED पर्यावरण चर के माध्यम से इस चेक को अक्षम करना है।

आइए docker-compose.yml में बदलाव करें और इसे फिर से चलाएँ:

version: '3'

networks:
  elk:

volumes:
  elasticsearch:
    driver: local

services:

  logstash:
    container_name: logstash_one_channel
    image: docker.elastic.co/logstash/logstash:6.3.2
    networks:
      - elk
    environment:
      XPACK_MONITORING_ENABLED: "false"
    ports:
      - 5046:5046
   volumes:
      - ./config/pipelines.yml:/usr/share/logstash/config/pipelines.yml:ro
      - ./config/pipelines:/usr/share/logstash/config/pipelines:ro

अब सब ठीक हे। कंटेनर प्रयोग के लिए तैयार है.

हम अगले कंसोल में फिर से टाइप कर सकते हैं:

echo '13123123123123123123123213123213' | nc localhost 5046

और देखो:

logstash_one_channel | {
logstash_one_channel |         "message" => "13123123123123123123123213123213",
logstash_one_channel |      "@timestamp" => 2019-04-29T11:43:44.582Z,
logstash_one_channel |        "@version" => "1",
logstash_one_channel |     "habra_field" => "Hello Habr",
logstash_one_channel |            "host" => "gateway",
logstash_one_channel |            "port" => 49418
logstash_one_channel | }

एक चैनल के भीतर काम करना

तो हमने लॉन्च किया. अब आप वास्तव में लॉगस्टैश को कॉन्फ़िगर करने के लिए समय ले सकते हैं। आइए अभी पाइपलाइनों.yml फ़ाइल को न छुएं, आइए देखें कि एक चैनल के साथ काम करके हम क्या प्राप्त कर सकते हैं।

मुझे कहना होगा कि चैनल कॉन्फ़िगरेशन फ़ाइल के साथ काम करने का सामान्य सिद्धांत यहां आधिकारिक मैनुअल में अच्छी तरह से वर्णित है यहां
यदि आप रूसी में पढ़ना चाहते हैं, तो हमने इसका उपयोग किया है एक लेख(लेकिन वहां क्वेरी सिंटैक्स पुराना है, हमें इसे ध्यान में रखना होगा)।

चलिए इनपुट अनुभाग से क्रमिक रूप से चलते हैं। हम पहले ही टीसीपी पर काम देख चुके हैं। यहां और क्या दिलचस्प हो सकता है?

दिल की धड़कन का उपयोग करके संदेशों का परीक्षण करें

स्वचालित परीक्षण संदेश उत्पन्न करने का एक दिलचस्प अवसर है।
ऐसा करने के लिए, आपको इनपुट अनुभाग में हार्टबीन प्लगइन को सक्षम करना होगा।

input {
  heartbeat {
    message => "HeartBeat!"
   }
  }

इसे चालू करें, प्रति मिनट एक बार प्राप्त करना प्रारंभ करें

logstash_one_channel | {
logstash_one_channel |      "@timestamp" => 2019-04-29T13:52:04.567Z,
logstash_one_channel |     "habra_field" => "Hello Habr",
logstash_one_channel |         "message" => "HeartBeat!",
logstash_one_channel |        "@version" => "1",
logstash_one_channel |            "host" => "a0667e5c57ec"
logstash_one_channel | }

यदि हम अधिक बार प्राप्त करना चाहते हैं, तो हमें अंतराल पैरामीटर जोड़ना होगा।
इस प्रकार हमें हर 10 सेकंड में एक संदेश प्राप्त होगा।

input {
  heartbeat {
    message => "HeartBeat!"
    interval => 10
   }
  }

किसी फ़ाइल से डेटा पुनर्प्राप्त करना

हमने फ़ाइल मोड को देखने का भी निर्णय लिया। यदि यह फ़ाइल के साथ ठीक काम करता है, तो शायद किसी एजेंट की आवश्यकता नहीं है, कम से कम स्थानीय उपयोग के लिए।

विवरण के अनुसार, ऑपरेटिंग मोड टेल-एफ के समान होना चाहिए, अर्थात। नई पंक्तियाँ पढ़ता है या, विकल्प के रूप में, संपूर्ण फ़ाइल पढ़ता है।

तो हम क्या पाना चाहते हैं:

हम ऐसी पंक्तियाँ प्राप्त करना चाहते हैं जो एक लॉग फ़ाइल से जुड़ी हों।
हम वह डेटा प्राप्त करना चाहते हैं जो कई लॉग फ़ाइलों में लिखा गया है, जबकि जो प्राप्त हुआ है उसे कहां से अलग करने में सक्षम हैं।
हम यह सुनिश्चित करना चाहते हैं कि जब लॉगस्टैश पुनः प्रारंभ हो, तो उसे यह डेटा दोबारा प्राप्त न हो।
हम यह जांचना चाहते हैं कि यदि लॉगस्टैश बंद है, और डेटा फ़ाइलों में लिखा जाना जारी है, तो जब हम इसे चलाएंगे, तो हमें यह डेटा प्राप्त होगा।

प्रयोग करने के लिए, आइए docker-compose.yml में एक और पंक्ति जोड़ें, उस निर्देशिका को खोलें जिसमें हम फ़ाइलें डालते हैं।

version: '3'

networks:
  elk:

volumes:
  elasticsearch:
    driver: local

services:

  logstash:
    container_name: logstash_one_channel
    image: docker.elastic.co/logstash/logstash:6.3.2
    networks:
      - elk
    environment:
      XPACK_MONITORING_ENABLED: "false"
    ports:
      - 5046:5046
   volumes:
      - ./config/pipelines.yml:/usr/share/logstash/config/pipelines.yml:ro
      - ./config/pipelines:/usr/share/logstash/config/pipelines:ro
      - ./logs:/usr/share/logstash/input

और इनपुट सेक्शन को habr_pipeline.conf में बदलें

input {
  file {
    path => "/usr/share/logstash/input/*.log"
   }
  }

चलो शुरू करो:

docker-compose up

लॉग फ़ाइलें बनाने और लिखने के लिए हम कमांड का उपयोग करेंगे:

 echo '1' >> logs/number1.log

{
logstash_one_channel |            "host" => "ac2d4e3ef70f",
logstash_one_channel |     "habra_field" => "Hello Habr",
logstash_one_channel |      "@timestamp" => 2019-04-29T14:28:53.876Z,
logstash_one_channel |        "@version" => "1",
logstash_one_channel |         "message" => "1",
logstash_one_channel |            "path" => "/usr/share/logstash/input/number1.log"
logstash_one_channel | }

हाँ, यह काम करता है!

साथ ही, हम देखते हैं कि हमने स्वचालित रूप से पथ फ़ील्ड जोड़ दिया है। इसका मतलब है कि भविष्य में हम इसके द्वारा रिकॉर्ड फ़िल्टर कर सकेंगे.

फिर से कोशिश करते है:

echo '2' >> logs/number1.log

{
logstash_one_channel |            "host" => "ac2d4e3ef70f",
logstash_one_channel |     "habra_field" => "Hello Habr",
logstash_one_channel |      "@timestamp" => 2019-04-29T14:28:59.906Z,
logstash_one_channel |        "@version" => "1",
logstash_one_channel |         "message" => "2",
logstash_one_channel |            "path" => "/usr/share/logstash/input/number1.log"
logstash_one_channel | }

और अब दूसरी फ़ाइल पर:

 echo '1' >> logs/number2.log

{
logstash_one_channel |            "host" => "ac2d4e3ef70f",
logstash_one_channel |     "habra_field" => "Hello Habr",
logstash_one_channel |      "@timestamp" => 2019-04-29T14:29:26.061Z,
logstash_one_channel |        "@version" => "1",
logstash_one_channel |         "message" => "1",
logstash_one_channel |            "path" => "/usr/share/logstash/input/number2.log"
logstash_one_channel | }

महान! फ़ाइल उठा ली गई, पथ सही ढंग से निर्दिष्ट किया गया, सब कुछ ठीक है।

लॉगस्टैश बंद करें और फिर से प्रारंभ करें। चलो इंतजार करते हैं। मौन। वे। ये रिकॉर्ड हमें दोबारा नहीं मिलते.

और अब सबसे साहसी प्रयोग.

लॉगस्टैश स्थापित करें और निष्पादित करें:

echo '3' >> logs/number2.log
echo '4' >> logs/number1.log

लॉगस्टैश फिर से चलाएँ और देखें:

logstash_one_channel | {
logstash_one_channel |            "host" => "ac2d4e3ef70f",
logstash_one_channel |     "habra_field" => "Hello Habr",
logstash_one_channel |         "message" => "3",
logstash_one_channel |        "@version" => "1",
logstash_one_channel |            "path" => "/usr/share/logstash/input/number2.log",
logstash_one_channel |      "@timestamp" => 2019-04-29T14:48:50.589Z
logstash_one_channel | }
logstash_one_channel | {
logstash_one_channel |            "host" => "ac2d4e3ef70f",
logstash_one_channel |     "habra_field" => "Hello Habr",
logstash_one_channel |         "message" => "4",
logstash_one_channel |        "@version" => "1",
logstash_one_channel |            "path" => "/usr/share/logstash/input/number1.log",
logstash_one_channel |      "@timestamp" => 2019-04-29T14:48:50.856Z
logstash_one_channel | }

हुर्रे! सब कुछ उठा लिया गया.

लेकिन हमें आपको निम्नलिखित के बारे में चेतावनी देनी चाहिए। यदि लॉगस्टैश कंटेनर हटा दिया गया है (docker stop logstash_one_channel && docker rm logstash_one_channel), तो कुछ भी नहीं उठाया जाएगा। फ़ाइल की जिस स्थिति तक उसे पढ़ा गया था उसे कंटेनर के अंदर संग्रहीत किया गया था। यदि आप इसे स्क्रैच से चलाते हैं, तो यह केवल नई लाइनें स्वीकार करेगा।

मौजूदा फ़ाइलें पढ़ना

मान लीजिए कि हम पहली बार लॉगस्टैश लॉन्च कर रहे हैं, लेकिन हमारे पास पहले से ही लॉग हैं और हम उन्हें संसाधित करना चाहेंगे।
यदि हम ऊपर उपयोग किए गए इनपुट सेक्शन के साथ लॉगस्टैश चलाते हैं, तो हमें कुछ नहीं मिलेगा। केवल नई लाइनें लॉगस्टैश द्वारा संसाधित की जाएंगी।

मौजूदा फ़ाइलों से लाइनें खींचने के लिए, आपको इनपुट अनुभाग में एक अतिरिक्त लाइन जोड़नी चाहिए:

input {
  file {
    start_position => "beginning"
    path => "/usr/share/logstash/input/*.log"
   }
  }

इसके अलावा, एक बारीकियां है: यह केवल उन नई फ़ाइलों को प्रभावित करता है जिन्हें लॉगस्टैश ने अभी तक नहीं देखा है। उन्हीं फ़ाइलों के लिए जो पहले से ही लॉगस्टैश के दृश्य क्षेत्र में थीं, इसने पहले ही उनका आकार याद कर लिया है और अब केवल उनमें नई प्रविष्टियाँ लेगा।

आइए यहां रुकें और इनपुट अनुभाग का अध्ययन करें। अभी भी कई विकल्प हैं, लेकिन अभी हमारे लिए आगे के प्रयोगों के लिए यही पर्याप्त है।

रूटिंग और डेटा परिवर्तन

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

हमें उन्हें बाहर निकलने पर अलग करने की जरूरत है। वे। हम एक चैनल में सूचना संदेश लिखते हैं, और दूसरे में त्रुटि संदेश।

ऐसा करने के लिए, इनपुट अनुभाग से फ़िल्टर और आउटपुट पर जाएँ।

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

ग्रोक के साथ एक संदेश को पार्स करना

टेक्स्ट स्ट्रिंग्स को पार्स करने और उनसे फ़ील्ड का एक सेट प्राप्त करने के लिए, फ़िल्टर अनुभाग में एक विशेष प्लगइन है - ग्रोक।

यहां इसका विस्तृत विवरण देने का लक्ष्य निर्धारित किए बिना (इसके लिए मैं इसका उल्लेख करता हूं)। आधिकारिक दस्तावेज), मैं अपना सरल उदाहरण दूंगा।

ऐसा करने के लिए, आपको इनपुट स्ट्रिंग्स के प्रारूप पर निर्णय लेना होगा। मेरे पास वे इस प्रकार हैं:

1 सूचना संदेश1
2 त्रुटि संदेश2

वे। पहचानकर्ता पहले आता है, फिर सूचना/त्रुटि, फिर बिना रिक्त स्थान के कुछ शब्द।
यह कठिन नहीं है, लेकिन संचालन के सिद्धांत को समझने के लिए यह पर्याप्त है।

इसलिए, ग्रोक प्लगइन के फ़िल्टर अनुभाग में, हमें अपनी स्ट्रिंग्स को पार्स करने के लिए एक पैटर्न परिभाषित करना होगा।

यह इस तरह दिखेगा:

filter {
  grok {
    match => { "message" => ["%{INT:message_id} %{LOGLEVEL:message_type} %{WORD:message_text}"] }
   }
  }

मूलतः यह एक नियमित अभिव्यक्ति है. तैयार पैटर्न का उपयोग किया जाता है, जैसे INT, LOGLEVEL, WORD। उनका विवरण, साथ ही अन्य पैटर्न, यहां पाया जा सकता है यहां

अब, इस फ़िल्टर से गुजरते हुए, हमारी स्ट्रिंग तीन फ़ील्ड के हैश में बदल जाएगी: message_id, message_type, message_text।

उन्हें आउटपुट अनुभाग में प्रदर्शित किया जाएगा.

if कमांड का उपयोग करके संदेशों को आउटपुट सेक्शन में रूट करना

आउटपुट अनुभाग में, जैसा कि हमें याद है, हम संदेशों को दो धाराओं में विभाजित करने जा रहे थे। कुछ - जो iNFO हैं, कंसोल पर आउटपुट होंगे, और त्रुटियों के साथ, हम एक फ़ाइल में आउटपुट करेंगे।

हम इन संदेशों को कैसे अलग करें? समस्या की स्थिति पहले से ही एक समाधान सुझाती है - आखिरकार, हमारे पास पहले से ही एक समर्पित message_type फ़ील्ड है, जो केवल दो मान ले सकता है: INFO और ERROR। इसी आधार पर हम if कथन का उपयोग करके चुनाव करेंगे।

if [message_type] == "ERROR" {
        # Здесь выводим в файл
       } else
     {
      # Здесь выводим в stdout
    }

फ़ील्ड और ऑपरेटरों के साथ काम करने का विवरण इस अनुभाग में पाया जा सकता है आधिकारिक मैनुअल.

अब, वास्तविक निष्कर्ष के बारे में ही।

कंसोल आउटपुट, यहां सब कुछ स्पष्ट है - stdout {}

लेकिन फ़ाइल का आउटपुट - याद रखें कि हम यह सब एक कंटेनर से चला रहे हैं और जिस फ़ाइल में हम परिणाम लिखते हैं उसे बाहर से एक्सेस करने के लिए, हमें इस निर्देशिका को docker-compose.yml में खोलने की आवश्यकता है।

कुल:

हमारी फ़ाइल का आउटपुट अनुभाग इस तरह दिखता है:

 output {
  if [message_type] == "ERROR" {
    file {
          path => "/usr/share/logstash/output/test.log"
          codec => line { format => "custom format: %{message}"}
         }
    } else
     {stdout {
             }
     }
  }

docker-compose.yml में हम आउटपुट के लिए एक और वॉल्यूम जोड़ते हैं:

version: '3'

networks:
  elk:

volumes:
  elasticsearch:
    driver: local

services:

  logstash:
    container_name: logstash_one_channel
    image: docker.elastic.co/logstash/logstash:6.3.2
    networks:
      - elk
    environment:
      XPACK_MONITORING_ENABLED: "false"
    ports:
      - 5046:5046
   volumes:
      - ./config/pipelines.yml:/usr/share/logstash/config/pipelines.yml:ro
      - ./config/pipelines:/usr/share/logstash/config/pipelines:ro
      - ./logs:/usr/share/logstash/input
      - ./output:/usr/share/logstash/output

हम इसे लॉन्च करते हैं, इसे आज़माते हैं, और दो धाराओं में विभाजन देखते हैं।

स्रोत: www.habr.com

ईएलके का व्यावहारिक अनुप्रयोग। लॉगस्टैश सेट करना