ਅਸੀਂ ELK ਅਤੇ ਐਕਸਚੇਂਜ ਦੇ ਦੋਸਤ ਹਾਂ। ਭਾਗ 2

ਮੈਂ ਦੋਸਤ ਐਕਸਚੇਂਜ ਅਤੇ ELK (ਸ਼ੁਰੂਆਤ ਇੱਥੇ). ਮੈਨੂੰ ਤੁਹਾਨੂੰ ਯਾਦ ਦਿਵਾਉਣ ਦਿਓ ਕਿ ਇਹ ਸੁਮੇਲ ਬਿਨਾਂ ਕਿਸੇ ਝਿਜਕ ਦੇ ਬਹੁਤ ਸਾਰੇ ਲੌਗਸ ਨੂੰ ਪ੍ਰੋਸੈਸ ਕਰਨ ਦੇ ਸਮਰੱਥ ਹੈ। ਇਸ ਵਾਰ ਅਸੀਂ ਇਸ ਬਾਰੇ ਗੱਲ ਕਰਾਂਗੇ ਕਿ ਐਕਸਚੇਂਜ ਨੂੰ Logstash ਅਤੇ Kibana ਕੰਪੋਨੈਂਟ ਨਾਲ ਕਿਵੇਂ ਕੰਮ ਕਰਨਾ ਹੈ।

ELK ਸਟੈਕ ਵਿੱਚ ਲੌਗਸਟੈਸ਼ ਦੀ ਵਰਤੋਂ ਲੌਗਸ ਨੂੰ ਸਮਝਦਾਰੀ ਨਾਲ ਪ੍ਰਕਿਰਿਆ ਕਰਨ ਅਤੇ ਦਸਤਾਵੇਜ਼ਾਂ ਦੇ ਰੂਪ ਵਿੱਚ ਇਲਾਸਟਿਕ ਵਿੱਚ ਪਲੇਸਮੈਂਟ ਲਈ ਤਿਆਰ ਕਰਨ ਲਈ ਕੀਤੀ ਜਾਂਦੀ ਹੈ, ਜਿਸ ਦੇ ਆਧਾਰ 'ਤੇ ਕਿਬਾਨਾ ਵਿੱਚ ਵੱਖ-ਵੱਖ ਵਿਜ਼ੂਅਲਾਈਜ਼ੇਸ਼ਨਾਂ ਨੂੰ ਬਣਾਉਣਾ ਸੁਵਿਧਾਜਨਕ ਹੈ।

ਸੈਟਿੰਗ

ਦੋ ਪੜਾਵਾਂ ਦੇ ਸ਼ਾਮਲ ਹਨ:

• OpenJDK ਪੈਕੇਜ ਨੂੰ ਇੰਸਟਾਲ ਅਤੇ ਸੰਰਚਿਤ ਕਰਨਾ।
• ਲੌਗਸਟੈਸ਼ ਪੈਕੇਜ ਨੂੰ ਸਥਾਪਿਤ ਅਤੇ ਸੰਰਚਿਤ ਕਰਨਾ।

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)

ਲੌਗਸਟੈਸ਼ ਪੈਕੇਜ ਨੂੰ ਸਥਾਪਿਤ ਅਤੇ ਸੰਰਚਿਤ ਕਰਨਾ

ਲੌਗਸਟੈਸ਼ ਡਿਸਟਰੀਬਿਊਸ਼ਨ ਨਾਲ ਆਰਕਾਈਵ ਫਾਈਲ ਨੂੰ ਡਾਊਨਲੋਡ ਕਰੋ ਇੱਥੋਂ. ਆਰਕਾਈਵ ਨੂੰ ਡਿਸਕ ਦੇ ਰੂਟ ਤੱਕ ਅਨਪੈਕ ਕੀਤਾ ਜਾਣਾ ਚਾਹੀਦਾ ਹੈ। ਫੋਲਡਰ ਵਿੱਚ ਅਨਪੈਕ ਕਰੋ C:Program Files ਇਹ ਇਸਦੀ ਕੀਮਤ ਨਹੀਂ ਹੈ, ਲੌਗਸਟੈਸ਼ ਆਮ ਤੌਰ 'ਤੇ ਸ਼ੁਰੂ ਕਰਨ ਤੋਂ ਇਨਕਾਰ ਕਰ ਦੇਵੇਗਾ। ਫਿਰ ਤੁਹਾਨੂੰ ਫਾਈਲ ਵਿੱਚ ਦਾਖਲ ਹੋਣ ਦੀ ਜ਼ਰੂਰਤ ਹੈ jvm.options Java ਪ੍ਰਕਿਰਿਆ ਲਈ RAM ਨਿਰਧਾਰਤ ਕਰਨ ਲਈ ਜ਼ਿੰਮੇਵਾਰ ਫਿਕਸ। ਮੈਂ ਸਰਵਰ ਦੇ ਅੱਧੇ ਰੈਮ ਨੂੰ ਨਿਰਧਾਰਤ ਕਰਨ ਦੀ ਸਿਫ਼ਾਰਿਸ਼ ਕਰਦਾ ਹਾਂ। ਜੇਕਰ ਇਸ ਵਿੱਚ ਬੋਰਡ ਉੱਤੇ 16 GB RAM ਹੈ, ਤਾਂ ਡਿਫੌਲਟ ਕੁੰਜੀਆਂ ਹਨ:

-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 'ਤੇ ਸਫਲਤਾਪੂਰਵਕ ਲਾਂਚ ਹੋਇਆ।

ਅੰਤਮ ਇੰਸਟਾਲੇਸ਼ਨ ਪੜਾਅ: ਲੌਗਸਟੈਸ਼ ਨੂੰ ਵਿੰਡੋਜ਼ ਸੇਵਾ ਵਜੋਂ ਲਾਂਚ ਕਰੋ। ਇਹ ਕੀਤਾ ਜਾ ਸਕਦਾ ਹੈ, ਉਦਾਹਰਨ ਲਈ, ਪੈਕੇਜ ਦੀ ਵਰਤੋਂ ਕਰਕੇ ਐਨਐਸਐਸਐਮ:

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

ਨੁਕਸ ਸਹਿਣਸ਼ੀਲਤਾ

ਸਰੋਤ ਸਰਵਰ ਤੋਂ ਟ੍ਰਾਂਸਫਰ ਕੀਤੇ ਜਾਣ 'ਤੇ ਲੌਗਸ ਦੀ ਸੁਰੱਖਿਆ ਨੂੰ ਸਥਾਈ ਕਤਾਰਾਂ ਵਿਧੀ ਦੁਆਰਾ ਯਕੀਨੀ ਬਣਾਇਆ ਜਾਂਦਾ ਹੈ।

ਇਹ ਕਿਵੇਂ ਕੰਮ ਕਰਦਾ ਹੈ

ਲੌਗ ਪ੍ਰੋਸੈਸਿੰਗ ਦੌਰਾਨ ਕਤਾਰਾਂ ਦਾ ਖਾਕਾ ਹੈ: ਇਨਪੁਟ → ਕਤਾਰ → ਫਿਲਟਰ + ਆਉਟਪੁੱਟ।

ਇਨਪੁਟ ਪਲੱਗਇਨ ਇੱਕ ਲੌਗ ਸਰੋਤ ਤੋਂ ਡੇਟਾ ਪ੍ਰਾਪਤ ਕਰਦਾ ਹੈ, ਇਸਨੂੰ ਇੱਕ ਕਤਾਰ ਵਿੱਚ ਲਿਖਦਾ ਹੈ, ਅਤੇ ਪੁਸ਼ਟੀ ਕਰਦਾ ਹੈ ਕਿ ਡੇਟਾ ਸਰੋਤ ਨੂੰ ਪ੍ਰਾਪਤ ਹੋ ਗਿਆ ਹੈ।

ਕਤਾਰ ਤੋਂ ਸੁਨੇਹੇ ਲੌਗਸਟੈਸ਼ ਦੁਆਰਾ ਸੰਸਾਧਿਤ ਕੀਤੇ ਜਾਂਦੇ ਹਨ, ਫਿਲਟਰ ਅਤੇ ਆਉਟਪੁੱਟ ਪਲੱਗਇਨ ਦੁਆਰਾ ਪਾਸ ਕੀਤੇ ਜਾਂਦੇ ਹਨ। ਜਦੋਂ ਆਉਟਪੁੱਟ ਤੋਂ ਪੁਸ਼ਟੀ ਪ੍ਰਾਪਤ ਹੁੰਦੀ ਹੈ ਕਿ ਲੌਗ ਭੇਜਿਆ ਗਿਆ ਹੈ, ਲੌਗਸਟੈਸ਼ ਪ੍ਰਕਿਰਿਆ ਕੀਤੇ ਲੌਗ ਨੂੰ ਕਤਾਰ ਤੋਂ ਹਟਾ ਦਿੰਦਾ ਹੈ। ਜੇਕਰ ਲੌਗਸਟੈਸ਼ ਬੰਦ ਹੋ ਜਾਂਦਾ ਹੈ, ਤਾਂ ਸਾਰੇ ਅਣਪ੍ਰੋਸੈਸ ਕੀਤੇ ਸੁਨੇਹੇ ਅਤੇ ਸੁਨੇਹੇ ਜਿਨ੍ਹਾਂ ਲਈ ਕੋਈ ਪੁਸ਼ਟੀ ਪ੍ਰਾਪਤ ਨਹੀਂ ਹੋਈ ਹੈ, ਕਤਾਰ ਵਿੱਚ ਰਹਿੰਦੇ ਹਨ, ਅਤੇ ਲੌਗਸਟੈਸ਼ ਅਗਲੀ ਵਾਰ ਸ਼ੁਰੂ ਹੋਣ 'ਤੇ ਉਹਨਾਂ 'ਤੇ ਕਾਰਵਾਈ ਕਰਨਾ ਜਾਰੀ ਰੱਖੇਗਾ।

ਵਿਵਸਥਾ

ਫਾਈਲ ਵਿੱਚ ਕੁੰਜੀਆਂ ਦੁਆਰਾ ਅਡਜੱਸਟੇਬਲ C:Logstashconfiglogstash.yml:

• queue.type: (ਸੰਭਵ ਮੁੱਲ - persisted и memory (default)).
• path.queue: (ਕਤਾਰ ਫਾਈਲਾਂ ਵਾਲੇ ਫੋਲਡਰ ਦਾ ਮਾਰਗ, ਜੋ ਕਿ ਮੂਲ ਰੂਪ ਵਿੱਚ C:Logstashqueue ਵਿੱਚ ਸਟੋਰ ਕੀਤੇ ਜਾਂਦੇ ਹਨ)।
• queue.page_capacity: (ਵੱਧ ਤੋਂ ਵੱਧ ਕਤਾਰ ਪੰਨੇ ਦਾ ਆਕਾਰ, ਪੂਰਵ-ਨਿਰਧਾਰਤ ਮੁੱਲ 64mb ਹੈ)।
• queue.drain: (ਸੱਚ/ਗਲਤ - ਲੌਗਸਟੈਸ਼ ਨੂੰ ਬੰਦ ਕਰਨ ਤੋਂ ਪਹਿਲਾਂ ਕਤਾਰ ਪ੍ਰੋਸੈਸਿੰਗ ਨੂੰ ਰੋਕਣਾ ਸਮਰੱਥ/ਅਯੋਗ ਕਰਦਾ ਹੈ। ਮੈਂ ਇਸਨੂੰ ਸਮਰੱਥ ਕਰਨ ਦੀ ਸਿਫ਼ਾਰਸ਼ ਨਹੀਂ ਕਰਦਾ, ਕਿਉਂਕਿ ਇਹ ਸਰਵਰ ਬੰਦ ਹੋਣ ਦੀ ਗਤੀ ਨੂੰ ਸਿੱਧਾ ਪ੍ਰਭਾਵਿਤ ਕਰੇਗਾ)।
• queue.max_events: (ਕਤਾਰ ਵਿੱਚ ਇਵੈਂਟਾਂ ਦੀ ਅਧਿਕਤਮ ਸੰਖਿਆ, ਡਿਫੌਲਟ 0 ਹੈ (ਬੇਅੰਤ))।
• queue.max_bytes: (ਬਾਈਟ ਵਿੱਚ ਅਧਿਕਤਮ ਕਤਾਰ ਦਾ ਆਕਾਰ, ਡਿਫੌਲਟ - 1024mb (1gb))।

ਜੇਕਰ ਕੌਂਫਿਗਰ ਕੀਤਾ ਗਿਆ ਹੈ queue.max_events и queue.max_bytes, ਫਿਰ ਜਦੋਂ ਇਹਨਾਂ ਵਿੱਚੋਂ ਕਿਸੇ ਵੀ ਸੈਟਿੰਗ ਦਾ ਮੁੱਲ ਪੂਰਾ ਹੋ ਜਾਂਦਾ ਹੈ ਤਾਂ ਸੁਨੇਹੇ ਕਤਾਰ ਵਿੱਚ ਸਵੀਕਾਰ ਕੀਤੇ ਜਾਣੇ ਬੰਦ ਹੋ ਜਾਂਦੇ ਹਨ। ਸਥਾਈ ਕਤਾਰਾਂ ਬਾਰੇ ਹੋਰ ਜਾਣੋ ਇੱਥੇ.

ਕਤਾਰ ਸਥਾਪਤ ਕਰਨ ਲਈ ਜ਼ਿੰਮੇਵਾਰ logstash.yml ਦੇ ਹਿੱਸੇ ਦੀ ਇੱਕ ਉਦਾਹਰਨ:

queue.type: persisted
queue.max_bytes: 10gb

ਵਿਵਸਥਾ

ਲੌਗਸਟੈਸ਼ ਸੰਰਚਨਾ ਵਿੱਚ ਆਮ ਤੌਰ 'ਤੇ ਤਿੰਨ ਭਾਗ ਹੁੰਦੇ ਹਨ, ਆਉਣ ਵਾਲੇ ਲੌਗਾਂ ਦੀ ਪ੍ਰਕਿਰਿਆ ਦੇ ਵੱਖ-ਵੱਖ ਪੜਾਵਾਂ ਲਈ ਜ਼ਿੰਮੇਵਾਰ ਹੁੰਦੇ ਹਨ: ਪ੍ਰਾਪਤ ਕਰਨਾ (ਇਨਪੁਟ ਸੈਕਸ਼ਨ), ਪਾਰਸਿੰਗ (ਫਿਲਟਰ ਸੈਕਸ਼ਨ) ਅਤੇ ਇਲਾਸਟਿਕ (ਆਊਟਪੁੱਟ ਸੈਕਸ਼ਨ) ਨੂੰ ਭੇਜਣਾ। ਹੇਠਾਂ ਅਸੀਂ ਉਹਨਾਂ ਵਿੱਚੋਂ ਹਰੇਕ 'ਤੇ ਡੂੰਘਾਈ ਨਾਲ ਵਿਚਾਰ ਕਰਾਂਗੇ.

ਇੰਪੁੱਟ

ਅਸੀਂ ਫਾਈਲਬੀਟ ਏਜੰਟਾਂ ਤੋਂ ਕੱਚੇ ਲੌਗਸ ਦੇ ਨਾਲ ਆਉਣ ਵਾਲੀ ਸਟ੍ਰੀਮ ਪ੍ਰਾਪਤ ਕਰਦੇ ਹਾਂ। ਇਹ ਇਹ ਪਲੱਗਇਨ ਹੈ ਜੋ ਅਸੀਂ ਇਨਪੁਟ ਭਾਗ ਵਿੱਚ ਦਰਸਾਉਂਦੇ ਹਾਂ:

input {
  beats {
    port => 5044
  }
}

ਇਸ ਸੰਰਚਨਾ ਤੋਂ ਬਾਅਦ, ਲੌਗਸਟੈਸ਼ ਪੋਰਟ 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 ਸੰਰਚਨਾ ਤੁਹਾਨੂੰ ਵਰਤਣ ਲਈ ਸਹਾਇਕ ਹੈ ਸ਼ਰਤੀਆ ਬਿਆਨ, ਇਸਲਈ ਅਸੀਂ ਸਿਰਫ਼ ਉਹਨਾਂ ਲੌਗਸ ਨੂੰ ਭੇਜ ਸਕਦੇ ਹਾਂ ਜੋ ਕਿ ਫਾਈਲਬੀਟ ਟੈਗ ਨਾਲ ਟੈਗ ਕੀਤੇ ਗਏ ਸਨ ਡਿਸਸੈਕਟ ਪਲੱਗਇਨ ਨੂੰ 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}"
  }
}

ਉਪਯੋਗੀ ਲਿੰਕ:

• ਵਿੰਡੋਜ਼ ਉੱਤੇ ਓਪਨਜੇਡੀਕੇ 11 ਨੂੰ ਕਿਵੇਂ ਇੰਸਟਾਲ ਕਰਨਾ ਹੈ?
• Logstash ਡਾਊਨਲੋਡ ਕਰੋ
• ਲਚਕੀਲਾ ਵਾਂਝਾ ਵਿਕਲਪ ਵਰਤਦਾ ਹੈ UseConcMarkSweepGC #36828
• ਐਨਐਸਐਸਐਮ
• ਲਗਾਤਾਰ ਕਤਾਰਾਂ
• ਬੀਟਸ ਇਨਪੁਟ ਪਲੱਗਇਨ
• ਲੌਗਸਟੈਸ਼ ਦੋਸਤ, ਮੇਰਾ ਚੇਨਸਾ ਕਿੱਥੇ ਹੈ? ਮੈਨੂੰ ਮੇਰੇ ਲੌਗਸ ਨੂੰ ਕੱਟਣ ਦੀ ਲੋੜ ਹੈ
• ਫਿਲਟਰ ਪਲੱਗਇਨ ਨੂੰ ਵੱਖ ਕਰੋ
• ਸ਼ਰਤਾਂ
• ਫਿਲਟਰ ਪਲੱਗਇਨ ਨੂੰ ਬਦਲੋ
• ਮਿਤੀ ਫਿਲਟਰ ਪਲੱਗਇਨ
• Elasticsearch ਆਉਟਪੁੱਟ ਪਲੱਗਇਨ

ਸਰੋਤ: www.habr.com