مقدمة

نشر النظام التالي ، في مواجهة الحاجة إلى معالجة عدد كبير من السجلات المختلفة. كما اختارت الأداة ELK. هذه المقالة سوف تناقش تجربتنا مع ضبط هذا المكدس.

نحن لا نضع هدفًا لوصف جميع إمكانياته ، لكننا نريد التركيز بدقة على حل المشكلات العملية. ويرجع ذلك إلى حقيقة أنه في وجود كمية كبيرة بما يكفي من الوثائق والصور الجاهزة ، هناك الكثير من المزالق ، على الأقل وجدناها.

نشرنا المكدس عبر عامل الميناء. علاوة على ذلك ، كان لدينا عامل ترسيم compose.yml مكتوب جيدًا ، والذي سمح لنا برفع الرصة دون أي مشاكل تقريبًا. وبدا لنا أن النصر كان قريبًا بالفعل ، والآن سنحرف قليلاً لتناسب احتياجاتنا وهذا كل شيء.

لسوء الحظ ، فإن محاولة ضبط النظام لتلقي السجلات ومعالجتها من تطبيقنا لم تتوج بالنجاح. لذلك ، قررنا أن الأمر يستحق استكشاف كل مكون على حدة ، ثم العودة إلى علاقاتهم.

لذلك ، بدأنا مع logstash.

البيئة ، النشر ، إطلاق Logstash في الحاوية

بالنسبة للنشر ، نستخدم عامل التثبيت ، وقد أجريت التجارب الموصوفة هنا على نظامي MacOS و Ubuntu 18.0.4.

صورة logstash التي تم تسجيلها معنا في عامل التحميل الأصلي - compose.yml هي docker.elastic.co/logstash/logstash:6.3.2

سوف نستخدمها للتجارب.

لتشغيل logstash ، كتبنا منفصلة عامل ميناء compose.yml. بالطبع ، كان من الممكن إطلاق الصورة من سطر الأوامر ، لكننا قمنا بحل مشكلة معينة ، حيث تم إطلاق كل شيء من عامل التأسيس.

لفترة وجيزة حول ملفات التكوين

على النحو التالي من الوصف ، يمكن تشغيل logstash على حد سواء لقناة واحدة ، في هذه الحالة ، تحتاج إلى نقل ملف * .conf أو لعدة قنوات ، في هذه الحالة ، يجب نقل ملف pipelines.yml ، والذي بدوره سيربط بالملفات .conf لكل قناة.
ذهبنا في الطريق الثاني. بدا لنا أكثر عالمية وقابلة للتطوير. لذلك ، أنشأنا pipelines.yml ، وقمنا بإنشاء دليل خطوط الأنابيب الذي سنضع فيه ملفات .conf لكل قناة.

داخل الحاوية يوجد ملف تكوين آخر - logstash.yml. نحن لا نلمسها ، استخدمها كما هي.

لذلك ، هيكل الدلائل لدينا:



للحصول على المدخلات ، في الوقت الحالي ، نعتقد أنه برنامج tcp على المنفذ 5046 ، ولأنه سوف نستخدم stdout.

هنا هو مثل هذا التكوين البسيط لأول مرة. منذ المهمة الأولى هي إطلاق.

لذلك ، لدينا هذا عامل ميناء 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 

ماذا نرى هنا؟

1. تم أخذ الشبكات والمجلدات من عامل الترسيم الأصلي compose.yml (واحد حيث تم إطلاق المكدس بالكامل) وأعتقد أنها لا تؤثر بشكل كبير على الصورة العامة هنا.
2. نقوم بإنشاء خدمة logstash واحدة من صورة docker.elastic.co/logstash/logstash:6.3.2 ومنحها اسم logstash_one_channel.
3. نقوم بإعادة توجيه المنفذ 5046 داخل الحاوية إلى نفس المنفذ الداخلي.
4. نقوم بتعيين ملف إعدادات قناتنا ./config/pipelines.yml إلى الملف /usr/share/logstash/config/pipelines.yml داخل الحاوية ، حيث سيقوم logstash باستلامه وجعله للقراءة فقط ، في حالة حدوث ذلك.
5. نعرض دليل ./config/pipelines ، حيث لدينا ملفات إعدادات القناة ، في دليل / usr / share / logstash / config / pipelines ونجعله أيضًا للقراءة فقط.

ملف 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 

ونحن نرى الجواب في وحدة التحكم في الحاوية:



لكن في الوقت نفسه ، نرى أيضًا:

logstash_one_channel | [2019-04-29T11: 28: 59،790] [خطأ] [logstash.licensechecker.licensereader] غير قادر على استرداد معلومات الترخيص من خادم الترخيص {: message => "Elasticsearch Unreachable: [http: // elasticsearch: 9200 /] [Manticore :: ResolutionFailure] elasticsearch "، ...

logstash_one_channel | [2019-04-29T11: 28: 59،894] [INFO] [logstash.pipeline] بدأ تشغيل خط الأنابيب بنجاح {: pipeline_id => ". Monitoring-logstash" ،: thread => "# <Thread: 0x119abb86 run>"}

logstash_one_channel | [2019-04-29T11: 28: 59،988] [INFO] [logstash.agent] خطوط الأنابيب التي تشغل {: count => 2 ،: running_pipelines => [: HABR ،: ". Monitoring-logstash"] ،: non_running_pipelines => [ ]}
logstash_one_channel | [2019-04-29T11: 29: 00،015] [خطأ] [logstash.inputs.metrics] تم تثبيت X-Pack على Logstash ولكن ليس على Elasticsearch. الرجاء تثبيت X-Pack على Elasticsearch لاستخدام ميزة المراقبة. ميزات أخرى قد تكون متاحة.
logstash_one_channel | [2019-04-29T11: 29: 00،526] [INFO] [logstash.agent] تم البدء بنجاح في نقطة نهاية واجهة برمجة تطبيقات Logstash {: port => 9600}
logstash_one_channel | [2019-04-29T11: 29: 04،478] [INFO] [logstash.outputs.elasticsearch] تشغيل التحقق من الصحة لمعرفة ما إذا كان اتصال Elasticsearch يعمل {: healthcheck_url => http: // elasticsearch: 9200 / ،: path => "/"}
ل ogstash_one_channel | [2019-04-29T11: 29: 04،487] [WARN] [logstash.outputs.elasticsearch] حاول إعادة إحياء الاتصال بمثال ES الميت ، لكن حصل خطأ. {: url => "elasticsearch: 9200 /" ،: error_type => LogStash :: المخرجات :: ElasticSearch :: HttpClient :: Pool :: HostUnreachableError ،: error => "Elasticsearch Unreachable: [http: // elasticsearch: 9200 / ] [Manticore :: ResolutionFailure] elasticsearch ”}
logstash_one_channel | [2019-04-29T11: 29: 04،704] [INFO] [logstash.licensechecker.licensereader] إجراء فحص صحي لمعرفة ما إذا كان اتصال Elasticsearch يعمل {: healthcheck_url => http: // elasticsearch: 9200 / ،: path => "/"}
logstash_one_channel | [2019-04-29T11: 29: 04،710] [WARN] [logstash.licensechecker.licensereader] حاول إعادة إحياء الاتصال بمثال ES الميت ، لكن حصل خطأ. {: url => "elasticsearch: 9200 /" ،: error_type => LogStash :: المخرجات :: ElasticSearch :: HttpClient :: Pool :: HostUnreachableError ،: error => "Elasticsearch Unreachable: [http: // elasticsearch: 9200 / ] [Manticore :: ResolutionFailure] elasticsearch ”}

وسجلنا يزحف طوال الوقت.

أبرزت هنا باللون الأخضر رسالة تفيد بأن خط الأنابيب بدأ بنجاح ، رسالة حمراء - رسالة خطأ وصفراء - رسالة حول محاولة الاتصال بـ elasticsearch : 9200.
يحدث هذا لأن logstash.conf المضمنة في الصورة لديه فحص للتحقق من توفر elasticsearch. بعد كل شيء ، يفترض logstash أنه يعمل كجزء من مكدس Elk ، وقمنا بفصله.

يمكنك العمل ، ولكن ليست مريحة.

يكمن الحل في تعطيل هذا الفحص من خلال متغير البيئة 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 | } 

العمل داخل قناة واحدة

لذلك ، بدأنا. الآن يمكنك بالفعل قضاء الوقت لتكوين logstash مباشرة. لن نلمس ملف pipelines.yml الآن ، وسنرى ما يمكنك الحصول عليه من خلال العمل مع قناة واحدة.

يجب أن أقول أن المبدأ العام للعمل مع ملف تكوين القناة موصوف جيدًا في الدليل الرسمي ، هنا
إذا كنت تريد القراءة باللغة الروسية ، فقد استخدمنا هذه المقالة هنا (ولكن بناء جملة الاستعلام قديم هناك ، يجب أن نأخذ ذلك في الاعتبار).

دعنا نذهب بالتتابع من قسم الإدخال. لقد رأينا بالفعل العمل على برنامج التعاون الفني. ماذا يمكن أن يكون موضع اهتمام هنا؟

اختبار الرسائل باستخدام نبضات

هناك فرصة مثيرة للاهتمام لإنشاء رسائل اختبار تلقائية.
للقيام بذلك ، تحتاج إلى تضمين البرنامج المساعد heartbean في قسم الإدخال.

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

استرداد البيانات من ملف

قررنا أيضًا رؤية وضع الملف. إذا كان يعمل بشكل طبيعي مع الملف ، فمن المحتمل أنه لن تكون هناك حاجة إلى وكيل ، حسنًا ، على الأقل للاستخدام المحلي.

وفقًا للوصف ، يجب أن يكون وضع التشغيل مشابهًا للذيل -f ، على سبيل المثال يقرأ الأسطر الجديدة أو ، كخيار ، يقرأ الملف بأكمله.

إذن ما نريد الحصول عليه:

1. نريد الحصول على خطوط يتم إلحاقها بملف سجل واحد.
2. نريد تلقي البيانات المكتوبة إلى عدة ملفات سجل ، مع التمكن من مشاركة ما جاء منها.
3. نريد التحقق من أنه عند إعادة تشغيل logstash ، لن تتلقى هذه البيانات مرة أخرى.
4. نريد التحقق من أنه في حالة تعطيل logstash ، واستمرار كتابة البيانات إلى الملفات ، فعند تشغيلها ، سنحصل على هذه البيانات.

لإجراء التجربة ، أضف سطرًا آخر إلى 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 | } 

! ممتاز تم التقاط الملف ، كان المسار صحيحًا ، كل شيء على ما يرام.

وقف logstash وإعادة تشغيل. لننتظر الصمت. أي نحن لا نتلقى هذه السجلات مرة أخرى.

والآن التجربة الأكثر جرأة.

نضع logstash وتنفيذ:

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

قم بتشغيل logstash مرة أخرى وانظر:

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

الصيحة! تم التقاط كل شيء.

ولكن ، يجب أن نحذر مما يلي. إذا تم حذف الحاوية التي تحتوي على logstash (عامل الإيقاف ، logstash_one_channel && docker rm logstash_one_channel) ، فلن يتم التقاط أي شيء. داخل الحاوية ، تم حفظ موضع الملف الذي تمت قراءته. إذا تم التشغيل من نقطة الصفر ، فسيتم قبول الخطوط الجديدة فقط.

قراءة الملفات الموجودة

لنفترض أننا قمنا بتشغيل logstash لأول مرة ، ولكن لدينا بالفعل سجلات ونود معالجتها.
إذا قمنا بتشغيل logstash باستخدام قسم الإدخال الذي استخدمناه أعلاه ، فلن نحصل على أي شيء. سيتم معالجة الأسطر الجديدة فقط بواسطة logstash.

من أجل سحب الخطوط من الملفات الموجودة ، أضف سطرًا إضافيًا إلى قسم الإدخال:

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

علاوة على ذلك ، هناك فارق بسيط ، وهذا يؤثر فقط على الملفات الجديدة التي لم يرها logstash حتى الآن. بالنسبة لنفس الملفات التي سقطت بالفعل في مجال عرض logstash ، فقد تذكر حجمها بالفعل وسيأخذ الآن إدخالات جديدة فيها فقط.

دعونا نتناول دراسة قسم المدخلات. هناك العديد من الخيارات ، لكن بالنسبة لنا ، يكفي إجراء مزيد من التجارب في الوقت الحالي.

التوجيه وتحويل البيانات

دعونا نحاول حل المشكلة التالية ، دعنا نقول أن لدينا رسائل من قناة واحدة ، بعضها إعلامي ، ورسالة خطأ جزئيًا. تختلف في العلامة. بعض المعلومات ، والبعض الآخر خطأ.

نحن بحاجة إلى فصلها في الإخراج. أي نكتب رسائل إعلامية في قناة واحدة ، ورسائل خطأ في قناة أخرى.

للقيام بذلك ، انتقل من قسم الإدخال إلى التصفية والإخراج.

باستخدام قسم التصفية ، سنقوم بتحليل الرسالة الواردة ، ونستخرج منها تجزئة (أزواج قيمة المفتاح) ، والتي يمكنك العمل معها بالفعل ، أي تفكيك حسب الظروف. وفي قسم الإخراج ، نختار الرسائل ونرسل كل منها إلى قناتنا.

تحليل رسالة باستخدام grok

من أجل تحليل السلاسل النصية والحصول على مجموعة من الحقول منها ، هناك مكون إضافي خاص في قسم التصفية - grok.

لا تهدف إلى إعطاء هنا وصفاً مفصلاً هنا (لهذا أشير إلى الوثائق الرسمية ) ، سأقدم مثالي البسيط.

للقيام بذلك ، تحتاج إلى تحديد تنسيق خطوط الإدخال. لدي لهم:

1 INFO message1
2 رسالة خطأ 2

أي يأتي المعرف أولاً ، ثم INFO / ERROR ، ثم بعض الكلمات بدون مسافات.
ليست صعبة ، ولكن يكفي لفهم كيف يعمل.

لذلك ، في قسم التصفية ، في البرنامج المساعد grok ، نحتاج إلى تحديد نمط لتحليل خطوطنا.

سيبدو مثل هذا:

 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 

نبدأ ، نحاول ، نرى التقسيم إلى قسمين.