🥇 ہم ELK اور Exchange کے دوست ہیں۔ حصہ 2

میں اپنی کہانی جاری رکھتا ہوں کہ دوستی کا تبادلہ اور ELK کیسے بنایا جائے (شروع یہاں)۔ میں آپ کو یاد دلاتا ہوں کہ یہ امتزاج بغیر کسی ہچکچاہٹ کے بہت بڑی تعداد میں لاگ پر کارروائی کرنے کے قابل ہے۔ اس بار ہم اس بارے میں بات کریں گے کہ ایکسچینج کو Logstash اور Kibana اجزاء کے ساتھ کیسے کام کرنا ہے۔

ELK اسٹیک میں Logstash کا استعمال لاگز کو ذہانت سے پروسیس کرنے اور انہیں دستاویزات کی شکل میں لچکدار میں جگہ کے لیے تیار کرنے کے لیے کیا جاتا ہے، جس کی بنیاد پر کبانا میں مختلف ویژولائزیشنز بنانا آسان ہے۔

تنصیب

دو مراحل پر مشتمل ہے:

اوپن جے ڈی کے پیکج کو انسٹال اور کنفیگر کرنا۔
Logstash پیکیج کو انسٹال اور کنفیگر کرنا۔

اوپن جے ڈی کے پیکج کو انسٹال اور کنفیگر کرنا

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 کو ونڈوز سروس کے طور پر لانچ کریں۔ یہ کیا جا سکتا ہے، مثال کے طور پر، پیکج کا استعمال کرتے ہوئے این ایس ایس ایم:

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

غلطی کی رواداری

ماخذ سرور سے منتقل ہونے پر لاگز کی حفاظت کو مسلسل قطاروں کے طریقہ کار کے ذریعے یقینی بنایا جاتا ہے۔

یہ کیسے کام کرتا ہے

لاگ پروسیسنگ کے دوران قطاروں کی ترتیب یہ ہے: ان پٹ → قطار → فلٹر + آؤٹ پٹ۔

ان پٹ پلگ ان لاگ ماخذ سے ڈیٹا وصول کرتا ہے، اسے قطار میں لکھتا ہے، اور تصدیق بھیجتا ہے کہ ڈیٹا ماخذ کو موصول ہو گیا ہے۔

قطار سے آنے والے پیغامات لاگسٹاش کے ذریعے پروسیس کیے جاتے ہیں، فلٹر اور آؤٹ پٹ پلگ ان سے گزرتے ہیں۔ آؤٹ پٹ سے تصدیق موصول ہونے پر کہ لاگ بھیج دیا گیا ہے، Logstash پروسیس شدہ لاگ کو قطار سے ہٹا دیتا ہے۔ اگر Logstash رک جاتا ہے، تمام غیر پروسیس شدہ پیغامات اور پیغامات جن کے لیے کوئی تصدیق موصول نہیں ہوئی ہے، قطار میں رہتے ہیں، اور Logstash اگلی بار شروع ہونے پر ان پر کارروائی جاری رکھے گا۔

ایڈجسٹمنٹ

فائل میں چابیاں کی طرف سے سایڈست C:Logstashconfiglogstash.yml:

queue.type: (ممکنہ اقدار - persisted и memory (default)).
path.queue: (قطار فائلوں کے ساتھ فولڈر کا راستہ، جو C:Logstashqueue میں بطور ڈیفالٹ محفوظ ہے)۔
queue.page_capacity: (زیادہ سے زیادہ قطار کے صفحے کا سائز، ڈیفالٹ ویلیو 64 ایم بی ہے)۔
queue.drain: (true/false - Logstash کو بند کرنے سے پہلے قطار پروسیسنگ کو روکنے کو فعال/غیر فعال کرتا ہے۔ میں اسے فعال کرنے کی سفارش نہیں کرتا، کیونکہ یہ سرور کے بند ہونے کی رفتار کو براہ راست متاثر کرے گا)۔
queue.max_events: (قطار میں واقعات کی زیادہ سے زیادہ تعداد، ڈیفالٹ 0 ہے (لامحدود))۔
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 کنفیگریشن آپ کو استعمال کرنے کی اجازت دیتی ہے۔ مشروط بیانات، لہذا ہم صرف وہ لاگ ان بھیج سکتے ہیں جن کو فائل بیٹ ٹیگ کے ساتھ ٹیگ کیا گیا تھا dissect پلگ ان پر 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" }
  …
}

یہ یاد رکھنے کے قابل ہے کہ یہ طریقہ صرف اس صورت میں موزوں ہے جب فیلڈ میں یقینی طور پر سٹرنگ ہو۔ آپشن فیلڈز سے Null اقدار پر کارروائی نہیں کرتا ہے اور ایک استثناء پھینک دیتا ہے۔

ٹریکنگ لاگز کے لیے، بہتر ہے کہ اسی طرح کا کنورٹ طریقہ استعمال نہ کیا جائے، کیونکہ فیلڈز 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 mutate پلگ ان کا استعمال کرتے ہوئے، ہمیں مطلوبہ ٹائم زون رجسٹر کریں اور اس ٹائم اسٹیمپ کو اندر رکھیں 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}"
  }
}

کارآمد روابط:

ماخذ: www.habr.com

ہم ELK اور Exchange کے دوست ہیں۔ حصہ 2

تنصیب

Logstash پیکیج کو انسٹال اور کنفیگر کرنا

غلطی کی رواداری

یہ کیسے کام کرتا ہے

ایڈجسٹمنٹ

ایڈجسٹمنٹ

ان پٹ

فلٹر

عددی فیلڈز کو تبدیل کرنا

وصول کنندہ_پتہ کو انفرادی وصول کنندگان میں تقسیم کرنا

ٹائم اسٹیمپ کو تبدیل کرنا

آؤٹ پٹ

حتمی ترتیب

نیا تبصرہ شامل کریں جواب منسوخ کریں