نحن أصدقاء مع ELK و Exchange. الجزء 2

أواصل قصتي حول كيفية تكوين صداقات في Exchange و ELK (ابدأ هنا ). اسمحوا لي أن أذكركم أن هذه المجموعة قادرة على التعامل مع عدد كبير جدًا من السجلات دون تردد. هذه المرة سنتحدث عن كيفية جعل Exchange يعمل مع مكونات Logstash و Kibana.



يتم استخدام Logstash في مكدس ELK لمعالجة السجلات بذكاء وتجهيزها لوضعها في Elastic في شكل مستندات ، والتي على أساسها يكون من الملائم بناء تصورات مختلفة في Kibana.

التركيب

يتكون من مرحلتين:

• تثبيت وتكوين حزمة OpenJDK.
• تثبيت وتكوين حزمة Logstash.

تثبيت وتكوين



حزمة OpenJDK يجب تنزيل حزمة OpenJDK وفك حزمها في دليل محدد. ثم يجب إدخال المسار إلى هذا الدليل في $ env: Path و $ env: متغيرات JAVA_HOME لنظام التشغيل Windows:











تحقق من إصدار 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)

تثبيت وتكوين حزمة Logstash

قم بتنزيل ملف الأرشيف مع توزيع Logstash من هنا . يجب تفريغ الأرشيف إلى جذر القرص. C:\Program Filesيجب ألا تفريغها في مجلد ، سيرفض Logstash البدء بشكل طبيعي. ثم تحتاج إلى إجراء jvm.optionsتغييرات على الملف المسؤول عن تخصيص ذاكرة الوصول العشوائي لعملية Java. أوصي بتحديد نصف ذاكرة الوصول العشوائي للخادم. إذا كان لديه 16 غيغابايت من ذاكرة الوصول العشوائي على متن الطائرة ، فإن المفاتيح الافتراضية هي:

-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 كخدمة Windows. يمكن القيام بذلك ، على سبيل المثال ، باستخدام حزمة NSSM :

PS C:\...\bin> .\nssm.exe install logstash
Service "logstash" installed successfully!

التسامح مع الخطأ

تضمن آلية قوائم الانتظار الثابتة سلامة السجلات أثناء الإرسال من الخادم المصدر.

كيف يعمل

تخطيط قوائم الانتظار أثناء معالجة السجل: الإدخال ← قائمة الانتظار ← التصفية + الإخراج.



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



تتم معالجة الرسائل من قائمة الانتظار بواسطة Logstash ، وتمرير عامل التصفية والمكوِّن الإضافي الناتج. عند تلقي تأكيد من إخراج إرسال السجل ، يقوم Logstash بإزالة السجل المعالج من قائمة الانتظار. إذا توقف Logstash ، فستظل جميع الرسائل والرسائل غير المعالجة التي لم يتم استلام تأكيد إرسالها في قائمة الانتظار ، وسيستمر Logstash في معالجتها في المرة التالية التي يبدأ فيها.

التخصيص

تنظمها مفاتيح في الملف C:\Logstash\config\logstash.yml:

• queue.type: (القيم المحتملة هي persistedو memory (default)).
• path.queue: (المسار إلى المجلد الذي يحتوي على ملفات قائمة الانتظار ، المخزنة افتراضيًا في C: \ Logstash \ queue).
• queue.page_capacity: (الحد الأقصى لحجم الصفحة لقائمة الانتظار ، الافتراضي هو 64 ميغا بايت).
• queue.drain: (صواب / خطأ - يمكّن / يعطل إيقاف معالجة قائمة الانتظار قبل إيقاف تشغيل Logstash. لا أوصي بتشغيله ، لأن هذا سيؤثر بشكل مباشر على سرعة إغلاق الخادم).
• queue.max_events: (الحد الأقصى لعدد الأحداث في قائمة الانتظار ، افتراضي - 0 (غير محدود)).
• queue.max_bytes: (الحد الأقصى لحجم قائمة الانتظار بالبايت ، الافتراضي هو 1024 ميجابايت (1 جيجابايت)).

إذا queue.max_eventsتم تكوينها queue.max_bytes، فسيتوقف استلام الرسائل في قائمة الانتظار عند الوصول إلى قيمة أي من هذه الإعدادات. اقرأ المزيد عن قوائم الانتظار الثابتة هنا .



مثال على جزء من logstash.yml المسؤول عن إعداد قائمة انتظار:

queue.type: persisted
queue.max_bytes: 10gb

التخصيص

يتكون تكوين Logstash عادةً من ثلاثة أجزاء ، مسؤولة عن المراحل المختلفة لمعالجة السجلات الواردة: الاستلام (قسم الإدخال) ، والتحليل (قسم التصفية) والإرسال إلى Elastic (قسم الإخراج). أدناه سوف نلقي نظرة فاحصة على كل منهم.

إدخال

يتم استلام الدفق الوارد مع السجلات الأولية من وكلاء Filebeat. هذا هو المكون الإضافي الذي نحدده في قسم الإدخال:

input {
  beats {
    port => 5044
  }
}

بعد هذا الإعداد ، يبدأ Logstash في الاستماع على المنفذ 5044 ، وعند تلقي السجلات ، يقوم بمعالجتها وفقًا للإعدادات في قسم التصفية. إذا لزم الأمر ، يمكنك التفاف القناة لتلقي السجلات من filebit في SSL. اقرأ المزيد عن إعدادات البرنامج المساعد Beats هنا .

منقي

جميع السجلات النصية المثيرة للاهتمام التي ينشئها Exchange للمعالجة هي بتنسيق 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 باستخدام العبارات الشرطية ، لذلك يمكننا فقط إرسال السجلات إلى المكون الإضافي للتشريح الذي تم تمييزه بعلامة filebeat 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) التي تحتوي على دخول من السجل، ويمكننا أن نضيف حقل التعسفي تلك الإرادة، على سبيل المثال، يحتوي على اسم التطبيق الذي نحن جمع السجلات.



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

• سيتم التعرف على الحقول الرقمية كنص ، مما يمنع إجراء العمليات عليها. وهي time-takenIIS سجل المجالات ، فضلا عن تتبع الحقول 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" ]
}

انتاج |

يتم استخدام قسم الإخراج لإرسال السجلات المعالجة إلى متلقي السجل. في حالة الإرسال مباشرة إلى Elastic ، يتم استخدام المكون الإضافي 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}"
  }
}

روابط مفيدة:

• كيفية تثبيت OpenJDK 11 على نظام Windows؟
• تنزيل Logstash
• يستخدم Elastic خيار depricated UseConcMarkSweepGC # 36828
• NSSM
• قوائم الانتظار المستمرة
• يدق المكون الإضافي الإدخال
• Logstash Dude ، أين منشارتي؟ أحتاج إلى تشريح سجلاتي
• تشريح مرشح البرنامج المساعد
• الشرطية
• تحويل البرنامج المساعد مرشح
• البرنامج المساعد لتصفية التاريخ
• البرنامج المساعد إخراج Elasticsearch