Pergo fabulam quomodo facere amicos Exchange et ELK (principium ). Fac me monere te hanc coniunctionem posse sine dubitatione ligna expediendi plurimas. Hoc tempore loquemur quomodo commercium cum Logstash et Kibana componentibus comparare debebimus.
Logstash in ELK acervo ad tigna ingeniose processit et eas ad collocandum in Elastica forma documentorum praeparavit, ex quibus commodum est varias visualizationes in Kibana aedificare.
occasum
Constat duobus gradibus:
- Installing et figurans OpenJDK sarcina.
- Installing et figurans Logstash sarcina.
Installing et figurans OpenJDK sarcina
Sarcina OpenJDK in certum presulem induci et indagari debet. Deinde via ad hunc directorium ineundum est in $env:Path et $env:JAVA_HOME variabilium Windows systematis operantis:
Java scriptor versionem reprehendo:
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)
Installing ac Vestibulum Logstash sarcina
Download tabularium archivi cum Logstash distributio . Archivum ad radicem orbis inpeditum esse oportet. Unpack ut folder C:Program Files Non valet, Logstash regulariter incipere recusabit. Tunc debes tabellam intrare jvm.options responsabilis RAM destinat pro processu Javae. Commendo commendo dimidium servi RAM. Si 16 GB ipsius RAM in tabula habeat, claves default erunt;
-Xms1g
-Xmx1g
reponendum est:
-Xms8g
-Xmx8g
Praeterea visum est rectae explanare -XX:+UseConcMarkSweepGC. Plus de hoc . Proximus gradus est configurationem defaltam in file logstash.conf creare:
input {
stdin{}
}
filter {
}
output {
stdout {
codec => "rubydebug"
}
}
Cum hac configuratione, Logstash notitias e console legit, per spargtum inane transit, et ad consolatorium remittit. Hoc schemate utens probabit functionem Logstash. Hoc facere, modo interactive curramus;
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 feliciter in portum 9600 immissa.
Ultimum gradum institutionis: Logstash launch ut Fenestra opera. Id fieri potest, verbi gratia, adhibita sarcina :
PS C:...bin> .nssm.exe install logstash
Service "logstash" installed successfully!
culpa tolerantia
Salus lignorum a fonte servo translatis providetur per mechanismum Pertinax Queues.
Quam operatus est
Propositio queues in processus iniuriarum est: input → queue → filter + output.
Plugin input data ex fonte stipes accipit, eam ad queue scribit, et confirmat datam ad fontem receptam esse.
Mandata a queue a Logstash discursum sunt, per colum et output plugin transierunt. Cum confirmatur ab output quod stipes missum est, Logstash trabem processionaliter a queue removet. Si Logstash cessat, omnes nuntii et nuntii unprocessati, quibus confirmatio nulla recepta est, in queue manent, et Logstash perget ad eas processurae tempore quo incipit.
tionibus
Product per claves in tabella C:Logstashconfiglogstash.yml:
queue.type: (valores possibiles -persistedиmemory (default)).path.queue: (iter ad folder cum lima queue, quae in C: Logstashqueue per defaltam condita sunt).queue.page_capacity: (magnitudo maxima queue pagina, valor default 64mb).queue.drain: (Verum/falsum - dat/disables desinendi queue processus antequam Logstash claudatur. Non suadeo ut id, quod hoc directe afficit velocitatem servo shutdown).queue.max_events: (maximus numerus eventuum in queue, defectus est 0 (infinitus)).queue.max_bytes: (maxima queue size in bytes, default - 1024mb (1gb)).
Si configurati queue.max_events и queue.max_bytesergo nuntii desinunt in queue accepti cum valor cuiuslibet horum occasus pervenitur. Disce de Queues Pertinax .
Exemplum partis logstash.yml auctoris constituendi queue:
queue.type: persisted
queue.max_bytes: 10gb
tionibus
Configuratio Logstash plerumque constat tribus partibus, quae diversis gradibus processus ineuntes inveniunt: acceptio (sectio input), parsing (sectio filtri) et ad Elasticam (sectionem output mittendam). Infra propius consideremus singula.
initus
Ineuntes rivum accipimus cum lignis crudis ab agentibus filebeat. Hoc plugin est quod in sectione initus indicamus:
input {
beats {
port => 5044
}
}
Post hanc configurationem, Logstash incipit audire portum 5044, et cum tigna accepta, eas secundum sectionis colum uncinis procedit. Si opus est, canalem involvere potes ad recipienda ligna ex filebit in SSL. Read more about beats plugin settings .
Filter
Omnia texta omnia quae commoda sunt ad expediendas quas Exchange generat sunt in forma csv cum agris descriptis in ipso documento tabellae. Pro parsing csv records, Logstash nobis offert tria plugins: csv et grok. Primum est maxime sed capas parsing tantum tigna simplicissima.
Exempli gratia, sequens testimonium in duos dividet (ob praesentiae commatis intra campum), quam ob causam stipes non recte dividetur;
…,"MDB:GUID1, Mailbox:GUID2, Event:526545791, MessageClass:IPM.Note, CreationTime:2020-05-15T12:01:56.457Z, ClientType:MOMT, SubmissionAssistant:MailboxTransportSubmissionEmailAssistant",…
Parsing tigna adhiberi potest, exempli gratia, IIS. In hoc casu, sectio columella sic spectare potest:
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 configuratione concedit ut uti , ideo tantum possumus mittere trabes quae cum schedula fasciculi ad plugin dissectum Tagged IIS. Intus plugin invenimus valores campi nominibus suis, agrum originalem delemus messagequi introitum a trunco continebat, et adiicere possumus consuetudinem agri, qui vult, exempli gratia, nomen applicationis, ex quo ligna colligimus.
In causa investigationis, melius est plugin uti csv, et bene potest procedere in campis complexis;
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" }
}
}
Intus plugin invenimus valores campi nominibus suis, agrum originalem delemus message (Et agros tenant-id и schema-version) , qui introitum ex trunco continebat, et agrum consuetum addere possumus, qui, exempli gratia, nomen habet applicationis ex quo ligna colligimus.
Ad exitum e scaena eliquationis accipiemus documenta in approximatione prima, parata ad visualizationem in Kibana. Desunt quae sequuntur:
- Agrorum numerorum in textu agnoscentur, qui operationes in illis impediunt. agros scilicet
time-takenPugnam Mantinensem Apud log, tum agrisrecipient-countиtotal-bitesLog Semita. - Tabellae signaculum vexillum continebit tempus cum iniuriarum discursum est, non tempus quod in latere servo scriptum erat.
- agri
recipient-addressvidebunt quasi unum situs constructionis, quae non patitur analysi recipientes literarum numerare.
Tempus est addere parvam magicam ad processus processus iniuriarum.
Ordo numerorum convertens agros
Plugin dissectum habet optionem convert_datatypequae adhiberi possunt textum campum ad formam digitalem convertendi. Exempli gratia;
dissect {
…
convert_datatype => { "time-taken" => "int" }
…
}
Memorabile est hanc methodum tantum idoneam esse si ager certum filum contineat. Optio non processu Null valores ex agris exceptionem petit.
Ad ligna sequi melius est non simili ratione convertendi uti, cum agros recipient-count и total-bites vacare potest. Hos agros convertere melius est plugin uti :
mutate {
convert => [ "total-bytes", "integer" ]
convert => [ "recipient-count", "integer" ]
}
Scindendo recipient_address in singulis recipientibus
Hoc problema etiam solvi potest per plugin mutato:
mutate {
split => ["recipient_address", ";"]
}
Mutantur indicatione
In casu vestigandi, quaestio facillime solvitur a plugin quae scribes in agro iuvabit timestamp date and time in the required format from the field date-time:
date {
match => [ "date-time", "ISO8601" ]
timezone => "Europe/Moscow"
remove_field => [ "date-time" ]
}
In casu IIS ligna, notitia campi coniungere debebimus date и time utens plugin mutata, subcriptio temporis zonae nobis necessariae et hoc tempore stamp in timestamp per diem plugin:
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
In sectione output adhibetur ut acta processus ad stipes accipientis mitteret. In casu directe ad Elasticam mittendi, plugin adhibetur , qui dat speciem electronicae servientis et indicem nominis templates ad documentum generatum mittendum:
output {
elasticsearch {
hosts => ["127.0.0.1:9200", "127.0.0.2:9200"]
manage_template => false
index => "Exchange-%{+YYYY.MM.dd}"
}
}
Final configuratione
Configuratio finalis sic erit:
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}"
}
}
Utile links:
Source: www.habr.com