🥇ከELK እና ልውውጥ ጋር ጓደኛሞች ነን። ክፍል 2

እንዴት ጓደኛ ማፍራት እንዳለብኝ ታሪኬን እቀጥላለሁ ልውውጥ እና ኤልኬ (መጀመሪያ እዚህ). ይህ ጥምረት በጣም ብዙ ቁጥር ያላቸውን ምዝግብ ማስታወሻዎች ያለምንም ማመንታት ማቀናበር የሚችል መሆኑን ላስታውስዎት። በዚህ ጊዜ ልውውጥ ከሎግስታሽ እና ከኪባና አካላት ጋር እንዴት እንደሚሰራ እንነጋገራለን ።

በ ELK ቁልል ውስጥ Logstash ምዝግብ ማስታወሻዎችን በብልህነት ለማስኬድ እና በ Elastic ውስጥ በሰነዶች መልክ ለመመደብ ለማዘጋጀት ይጠቅማል ፣ በዚህ መሠረት በኪባና ውስጥ የተለያዩ ምስላዊ ምስሎችን ለመገንባት ምቹ ነው።

ቅንብር

ሁለት ደረጃዎችን ያካትታል:

የ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 ጥቅል መጫን እና ማዋቀር

የማህደሩን ፋይል በሎግስታሽ ስርጭት ያውርዱ እዚህ. ማህደሩ ወደ ዲስኩ ስር መከፈት አለበት። ወደ አቃፊ ይንቀሉ C:Program Files ዋጋ የለውም, Logstash በመደበኛነት ለመጀመር እምቢ ይላል. ከዚያ ወደ ፋይሉ ውስጥ ማስገባት ያስፈልግዎታል jvm.options ለጃቫ ሂደት ራም የመመደብ ሃላፊነትን ያስተካክላል። የአገልጋዩን ራም ግማሹን እንዲገልጹ እመክራለሁ። በቦርዱ ላይ 16 ጊባ ራም ካለው ነባሪ ቁልፎቹ የሚከተሉት ናቸው

-Xms1g
-Xmx1g

መተካት ያለበት በ:

-Xms8g
-Xmx8g

በተጨማሪም, በመስመሩ ላይ አስተያየት መስጠት ተገቢ ነው -XX:+UseConcMarkSweepGC. ስለዚህ ጉዳይ የበለጠ እዚህ. ቀጣዩ ደረጃ በ logstash.conf ፋይል ውስጥ ነባሪ ውቅር መፍጠር ነው፡-

input {
 stdin{}
}
 
filter {
}
 
output {
 stdout {
 codec => "rubydebug"
 }
}

በዚህ ውቅር ሎግስታሽ ከኮንሶሉ ላይ ያለውን መረጃ አንብቦ በባዶ ማጣሪያ ውስጥ ያልፍና ወደ ኮንሶሉ መልሶ ያወጣዋል። ይህንን ውቅር በመጠቀም የሎግስታሽ ተግባርን ይፈትሻል። ይህንን ለማድረግ በይነተገናኝ ሁነታ እናስኬደው፡

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 ከቆመ፣ ምንም ማረጋገጫ ያልደረሳቸው ሁሉም ያልተከናወኑ መልእክቶች እና መልእክቶች ወረፋው ላይ ይቆያሉ፣ እና ሎግስታሽ በሚቀጥለው ጊዜ ሲጀምር ማሰራቱን ይቀጥላል።

በደንብ ማድረግ

በፋይሉ ውስጥ ባሉ ቁልፎች ማስተካከል ይቻላል C:Logstashconfiglogstash.yml:

queue.type: ( ሊሆኑ የሚችሉ እሴቶች - persisted и memory (default)).
path.queue: (በነባሪ በ C: Logstashqueue ውስጥ የተከማቹ የወረፋ ፋይሎች ወዳለው አቃፊ የሚወስደው መንገድ)።
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 ውቅር ብዙውን ጊዜ ሶስት ክፍሎችን ያቀፈ ነው, ለተለያዩ የገቢ ምዝግቦች ሂደት ደረጃዎች ኃላፊነት ያለው: መቀበል (የግቤት ክፍል), መተንተን (የማጣሪያ ክፍል) እና ወደ ላስቲክ (የውጤት ክፍል) መላክ. ከዚህ በታች እያንዳንዳቸውን በዝርዝር እንመለከታለን.

ግቤት

የሚመጣውን ዥረት በጥሬ ምዝግብ ማስታወሻዎች ከፋይልቢት ወኪሎች እንቀበላለን። በግቤት ክፍል ውስጥ የምናመለክተው ይህን ፕለጊን ነው፡-

input {
  beats {
    port => 5044
  }
}

ከዚህ ውቅር በኋላ Logstash ወደብ 5044 ማዳመጥ ይጀምራል, እና ምዝግብ ማስታወሻዎችን ሲቀበሉ, በማጣሪያው ክፍል ቅንጅቶች መሰረት ያስኬዳቸዋል. አስፈላጊ ከሆነ በኤስኤስኤል ውስጥ ከፋይልቢት ምዝግብ ማስታወሻዎችን ለመቀበል ቻናሉን መጠቅለል ይችላሉ። ስለ ቢትስ ተሰኪ ቅንብሮች የበለጠ ያንብቡ እዚህ.

ማጣሪያ

ልውውጥ የሚያመነጨውን ለመስራት የሚያስደስት የጽሑፍ ምዝግብ ማስታወሻዎች በራሱ በመዝገብ ፋይሉ ውስጥ ከተገለጹት መስኮች ጋር በ csv ቅርጸት ናቸው። የ csv መዝገቦችን ለመተንተን ሎግስታሽ ሶስት ተሰኪዎችን ይሰጠናል፡ ፈሳሽ, 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" }
  …
}

ሜዳው በእርግጠኝነት ሕብረቁምፊን ከያዘ ይህ ዘዴ ተስማሚ መሆኑን ማስታወስ ጠቃሚ ነው. አማራጩ ባዶ እሴቶችን ከሜዳ ላይ አያስኬድም እና የተለየ ነገርን ይጥላል።

የምዝግብ ማስታወሻዎችን ለመከታተል, ከመስኮቹ ጀምሮ, ተመሳሳይ የመቀየሪያ ዘዴን አለመጠቀም የተሻለ ነው 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" ]
}

ዉጤት

የውጤት ክፍሉ የተቀነባበሩ ምዝግብ ማስታወሻዎችን ወደ ሎግ ተቀባይ ለመላክ ይጠቅማል። በቀጥታ ወደ ላስቲክ ለመላክ፣ ፕለጊን ጥቅም ላይ ይውላል እንጦጦየመነጨውን ሰነድ ለመላክ የአገልጋይ አድራሻውን እና የመረጃ ጠቋሚውን አብነት የሚገልጽ፡-

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

ጠቃሚ አገናኞች:

ምንጭ: hab.com

እኛ ከኤልኬ እና ልውውጥ ጋር ጓደኛሞች ነን። ክፍል 2

ቅንብር

የ Logstash ጥቅል መጫን እና ማዋቀር

ስህተትን መታገስ

እንዴት እንደሚሰራ

በደንብ ማድረግ

በደንብ ማድረግ

ግቤት

ማጣሪያ

የቁጥር መስኮችን በመቀየር ላይ

የተቀባዩን_አድራሻ ወደ ግለሰብ ተቀባዮች በመከፋፈል

የጊዜ ማህተሙን መለወጥ

ዉጤት

የመጨረሻ ውቅር

አስተያየት ያክሉ ምላሽ መሰረዝ