🥇ನಾವು ELK ಮತ್ತು ವಿನಿಮಯದೊಂದಿಗೆ ಸ್ನೇಹಿತರಾಗಿದ್ದೇವೆ. ಭಾಗ 2

ಸ್ನೇಹಿತರನ್ನು ವಿನಿಮಯ ಮಾಡಿಕೊಳ್ಳುವುದು ಮತ್ತು ELK ಅನ್ನು ಹೇಗೆ ಮಾಡುವುದು ಎಂಬುದರ ಕುರಿತು ನನ್ನ ಕಥೆಯನ್ನು ನಾನು ಮುಂದುವರಿಸುತ್ತೇನೆ (ಆರಂಭದಲ್ಲಿ ಇಲ್ಲಿ) ಈ ಸಂಯೋಜನೆಯು ಹೆಚ್ಚಿನ ಸಂಖ್ಯೆಯ ಲಾಗ್‌ಗಳನ್ನು ಹಿಂಜರಿಕೆಯಿಲ್ಲದೆ ಪ್ರಕ್ರಿಯೆಗೊಳಿಸಲು ಸಮರ್ಥವಾಗಿದೆ ಎಂದು ನಾನು ನಿಮಗೆ ನೆನಪಿಸುತ್ತೇನೆ. ಈ ಸಮಯದಲ್ಲಿ ನಾವು ಲಾಗ್‌ಸ್ಟಾಶ್ ಮತ್ತು ಕಿಬಾನಾ ಘಟಕಗಳೊಂದಿಗೆ ವಿನಿಮಯವನ್ನು ಹೇಗೆ ಪಡೆಯುವುದು ಎಂಬುದರ ಕುರಿತು ಮಾತನಾಡುತ್ತೇವೆ.

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 ಜಾವಾ ಪ್ರಕ್ರಿಯೆಗೆ RAM ಅನ್ನು ನಿಯೋಜಿಸುವ ಜವಾಬ್ದಾರಿಯನ್ನು ಸರಿಪಡಿಸುತ್ತದೆ. ಸರ್ವರ್‌ನ ಅರ್ಧದಷ್ಟು 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}

ಪೋರ್ಟ್ 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 ನಮಗೆ ಮೂರು ಪ್ಲಗಿನ್‌ಗಳನ್ನು ನೀಡುತ್ತದೆ: ect ೇದಿಸಿ, csv ಮತ್ತು grok. ಮೊದಲನೆಯದು ಹೆಚ್ಚು быстрый, ಆದರೆ ಸರಳವಾದ ದಾಖಲೆಗಳನ್ನು ಮಾತ್ರ ಪಾರ್ಸಿಂಗ್ ಮಾಡುವುದರೊಂದಿಗೆ copes.
ಉದಾಹರಣೆಗೆ, ಇದು ಕೆಳಗಿನ ದಾಖಲೆಯನ್ನು ಎರಡಾಗಿ ವಿಭಜಿಸುತ್ತದೆ (ಕ್ಷೇತ್ರದ ಒಳಗೆ ಅಲ್ಪವಿರಾಮದ ಉಪಸ್ಥಿತಿಯಿಂದಾಗಿ), ಅದಕ್ಕಾಗಿಯೇ ಲಾಗ್ ಅನ್ನು ತಪ್ಪಾಗಿ ಪಾರ್ಸ್ ಮಾಡಲಾಗುತ್ತದೆ:

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

ಲಾಗ್‌ಸ್ಟ್ಯಾಶ್ ಕಾನ್ಫಿಗರೇಶನ್ ನಿಮಗೆ ಬಳಸಲು ಅನುಮತಿಸುತ್ತದೆ ಷರತ್ತುಬದ್ಧ ಹೇಳಿಕೆಗಳು, ಆದ್ದರಿಂದ ನಾವು ಫೈಲ್‌ಬೀಟ್ ಟ್ಯಾಗ್‌ನೊಂದಿಗೆ ಟ್ಯಾಗ್ ಮಾಡಲಾದ ಲಾಗ್‌ಗಳನ್ನು ಮಾತ್ರ ಡಿಸೆಕ್ಟ್ ಪ್ಲಗಿನ್‌ಗೆ ಕಳುಹಿಸಬಹುದು 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}"
  }
}

ಉಪಯುಕ್ತ ಲಿಂಕ್‌ಗಳು:

ಮೂಲ: www.habr.com

ನಾವು ELK ಮತ್ತು ವಿನಿಮಯದೊಂದಿಗೆ ಸ್ನೇಹಿತರಾಗಿದ್ದೇವೆ. ಭಾಗ 2

ಸೆಟ್ಟಿಂಗ್

ಲಾಗ್‌ಸ್ಟಾಶ್ ಪ್ಯಾಕೇಜ್ ಅನ್ನು ಸ್ಥಾಪಿಸುವುದು ಮತ್ತು ಕಾನ್ಫಿಗರ್ ಮಾಡುವುದು

ದೋಷಸಹಿಷ್ಣುತೆ

ಇದು ಹೇಗೆ ಕಾರ್ಯನಿರ್ವಹಿಸುತ್ತದೆ

ಹೊಂದಾಣಿಕೆ

ಹೊಂದಾಣಿಕೆ

ಇನ್ಪುಟ್

ಫಿಲ್ಟರ್

ಸಂಖ್ಯಾ ಕ್ಷೇತ್ರಗಳನ್ನು ಪರಿವರ್ತಿಸುವುದು

ಸ್ವೀಕರಿಸುವವರ_ವಿಳಾಸವನ್ನು ಪ್ರತ್ಯೇಕ ಸ್ವೀಕೃತದಾರರನ್ನಾಗಿ ವಿಭಜಿಸುವುದು

ಟೈಮ್‌ಸ್ಟ್ಯಾಂಪ್ ಅನ್ನು ಬದಲಾಯಿಸುವುದು

ಔಟ್ಪುಟ್

ಅಂತಿಮ ಸಂರಚನೆ

ಕಾಮೆಂಟ್ ಅನ್ನು ಸೇರಿಸಿ Отменить ответ