Εισαγωγή

Κατά την ανάπτυξη ενός άλλου συστήματος, αντιμετωπίσαμε την ανάγκη επεξεργασίας ενός μεγάλου αριθμού διαφόρων αρχείων καταγραφής. Ως όργανο επιλέχθηκε το ΕΛΚ. Αυτό το άρθρο θα μιλήσει για την εμπειρία μας στη ρύθμιση αυτής της στοίβας.

Δεν θέτουμε στόχο να περιγράψουμε όλες τις δυνατότητές του, αλλά θέλουμε να επικεντρωθούμε στην επίλυση πρακτικών προβλημάτων. Αυτό οφείλεται στο γεγονός ότι με αρκετά μεγάλο όγκο τεκμηρίωσης και έτοιμες εικόνες, υπάρχουν πολλές παγίδες, τουλάχιστον τις βρήκαμε.

Αναπτύξαμε τη στοίβα μέσω του docker-compose. Επιπλέον, είχαμε ένα καλογραμμένο docker-compose.yml που μας επέτρεπε να ανεβάσουμε το stack χωρίς σχεδόν κανένα πρόβλημα. Και μας φάνηκε ότι η νίκη ήταν ήδη κοντά, τώρα θα τη στρίψουμε λίγο για να ταιριάζει στις ανάγκες μας και αυτό είναι.

Δυστυχώς, μια προσπάθεια συντονισμού του συστήματος για λήψη και επεξεργασία αρχείων καταγραφής από την εφαρμογή μας δεν ήταν επιτυχής αμέσως. Ως εκ τούτου, αποφασίσαμε ότι αξίζει να μελετήσουμε κάθε στοιχείο ξεχωριστά και στη συνέχεια να επιστρέψουμε στις συνδέσεις τους.

Ας ξεκινήσουμε λοιπόν με το logstash.

Περιβάλλον, ανάπτυξη, λειτουργία Logstash σε κοντέινερ

Για την ανάπτυξη, χρησιμοποιούμε docker-compose, τα πειράματα που περιγράφονται εδώ πραγματοποιήθηκαν σε MacOS και Ubuntu 18.0.4.

Η εικόνα logstash που είχαμε στο αρχικό μας docker-compose.yml είναι docker.elastic.co/logstash/logstash:6.3.2

Θα το χρησιμοποιήσουμε για πειράματα.

Για την εκτέλεση του logstash, γράψαμε ένα ξεχωριστό docker-compose.yml. Φυσικά, ήταν δυνατή η εκκίνηση της εικόνας από τη γραμμή εντολών, αλλά τελικά λύσαμε μια συγκεκριμένη εργασία, όπου όλα από το docker-compose ξεκινούν για εμάς.

Εν συντομία σχετικά με τα αρχεία διαμόρφωσης

Όπως προκύπτει από την περιγραφή, το logstash μπορεί να εκτελεστεί ως για ένα κανάλι, σε αυτήν την περίπτωση, χρειάζεται να μεταφέρει το αρχείο *.conf ή για πολλά κανάλια, οπότε πρέπει να μεταφέρει το αρχείο pipelines.yml, το οποίο με τη σειρά του , θα αναφέρεται στα αρχεία .conf για κάθε κανάλι.
Πήραμε το δεύτερο μονοπάτι. Μας φάνηκε πιο ευέλικτο και επεκτάσιμο. Επομένως, δημιουργήσαμε το pipelines.yml και φτιάξαμε έναν κατάλογο pipelines στον οποίο θα βάλουμε αρχεία .conf για κάθε κανάλι.

Μέσα στο κοντέινερ υπάρχει ένα άλλο αρχείο ρυθμίσεων - logstash.yml. Δεν το αγγίζουμε, το χρησιμοποιούμε ως έχει.

Άρα η δομή του καταλόγου μας είναι:

Προς το παρόν, υποθέτουμε ότι αυτό είναι tcp στη θύρα 5046 για λήψη δεδομένων εισόδου και θα χρησιμοποιήσουμε το stdout για έξοδο.

Εδώ είναι μια τόσο απλή διαμόρφωση για την πρώτη εκτέλεση. Γιατί το αρχικό καθήκον είναι η εκτόξευση.

Έχουμε λοιπόν αυτό το docker-compose.yml

version: '3'

networks:
  elk:

volumes:
  elasticsearch:
    driver: local

services:

  logstash:
    container_name: logstash_one_channel
    image: docker.elastic.co/logstash/logstash:6.3.2
    networks:
      	- elk
    ports:
      	- 5046:5046
    volumes:
      	- ./config/pipelines.yml:/usr/share/logstash/config/pipelines.yml:ro
	- ./config/pipelines:/usr/share/logstash/config/pipelines:ro

Τι βλέπουμε εδώ;

1. Τα δίκτυα και οι τόμοι ελήφθησαν από το αρχικό docker-compose.yml (αυτό όπου εκκινείται ολόκληρη η στοίβα) και νομίζω ότι δεν επηρεάζουν πολύ τη συνολική εικόνα εδώ.
2. Δημιουργούμε μία υπηρεσία (υπηρεσίες) logstash, από την εικόνα docker.elastic.co/logstash/logstash:6.3.2 και της δίνουμε το όνομα logstash_one_channel.
3. Προωθούμε τη θύρα 5046 μέσα στο κοντέινερ, στην ίδια εσωτερική θύρα.
4. Αντιστοιχίζουμε το αρχείο διαμόρφωσης του σωλήνα ./config/pipelines.yml στο αρχείο /usr/share/logstash/config/pipelines.yml μέσα στο κοντέινερ, όπου το logstash θα το πάρει και θα το κάνει μόνο για ανάγνωση, για κάθε ενδεχόμενο.
5. Αντιστοιχίζουμε τον κατάλογο ./config/pipelines, όπου έχουμε τα αρχεία διαμόρφωσης του σωλήνα, στον κατάλογο /usr/share/logstash/config/pipelines και τον κάνουμε επίσης μόνο για ανάγνωση.

αρχείο piping.yml

- pipeline.id: HABR
  pipeline.workers: 1
  pipeline.batch.size: 1
  path.config: "./config/pipelines/habr_pipeline.conf"

Περιγράφει ένα κανάλι με το αναγνωριστικό HABR και τη διαδρομή προς το αρχείο διαμόρφωσής του.

Και τέλος το αρχείο "./config/pipelines/habr_pipeline.conf"

input {
  tcp {
    port => "5046"
   }
  }
filter {
  mutate {
    add_field => [ "habra_field", "Hello Habr" ]
    }
  }
output {
  stdout {
      
    }
  }

Δεν θα μπούμε στην περιγραφή του προς το παρόν, προσπαθούμε να εκτελέσουμε:

docker-compose up

Τι βλέπουμε;

Το δοχείο έχει ξεκινήσει. Μπορούμε να ελέγξουμε τη δουλειά του:

echo '13123123123123123123123213123213' | nc localhost 5046

Και βλέπουμε την απάντηση στην κονσόλα κοντέινερ:

Ταυτόχρονα όμως βλέπουμε:

logstash_one_channel | [2019-04-29T11:28:59,790][ERROR][logstash.licensechecker.licensereader] Δεν είναι δυνατή η ανάκτηση πληροφοριών άδειας χρήσης από τον διακομιστή αδειών {:message=>"Elasticsearch Unreachable: [http://elasticsearch:9200/][Manticore ::ResolutionFailure]elasticsearch", ...

logstash_one_channel | [2019-04-29T11:28:59,894][INFO ][logstash.pipeline ] Ο αγωγός ξεκίνησε με επιτυχία {:pipeline_id=>".monitoring-logstash", :thread=>"# »}

logstash_one_channel | [2019-04-29T11:28:59,988][INFO ][logstash.agent ] Σωληνώσεις σε λειτουργία {:count=>2, :running_pipelines=>[:HABR, :".monitoring-logstash"], :non_running_pipelines=>[ ]}
logstash_one_channel | [2019-04-29T11:29:00,015][σφάλμα][logstash.inputs.metrics ] Το X-Pack είναι εγκατεστημένο στο Logstash αλλά όχι στο Elasticsearch. Εγκαταστήστε το X-Pack στο Elasticsearch για να χρησιμοποιήσετε τη δυνατότητα παρακολούθησης. Άλλες λειτουργίες ενδέχεται να είναι διαθέσιμες.
logstash_one_channel | [2019-04-29T11:29:00,526][INFO ][logstash.agent ] Ξεκίνησε με επιτυχία Logstash API endpoint {:port=>9600}
logstash_one_channel | [2019-04-29T11:29:04,478][INFO ][logstash.outputs.elasticsearch] Εκτέλεση ελέγχου υγείας για να δείτε εάν λειτουργεί μια σύνδεση Elasticsearch {:healthcheck_url=>http://elasticsearch:9200/, :path=> "/"}
logstash_one_channel | [2019-04-29T11:29:04,487][ΠΡΟΕΙΔΟΠΟΙΗΣΗ ][logstash.outputs.elasticsearch] Προσπάθησε να επαναφέρει τη σύνδεση σε νεκρή παρουσία ES, αλλά παρουσιάστηκε σφάλμα. {:url=>"ελαστική αναζήτηση:9200/", :error_type=>LogStash::Outputs::ElasticSearch::HttpClient::Pool::HostUnreachableError, :error=>"Elasticsearch Unreachable: [http://elasticsearch:9200/:][Resoureail] elasticsearch"}
logstash_one_channel | [2019-04-29T11:29:04,704][INFO ][logstash.licensechecker.licensereader] Εκτέλεση ελέγχου υγείας για να δείτε εάν λειτουργεί μια σύνδεση Elasticsearch {:healthcheck_url=>http://elasticsearch:9200/, :path=> "/"}
logstash_one_channel | [2019-04-29T11:29:04,710][ΠΡΟΕΙΔΟΠΟΙΗΣΗ ][logstash.licensechecker.licensereader] Προσπάθησε να επαναφέρει τη σύνδεση σε νεκρή παρουσία ES, αλλά παρουσιάστηκε σφάλμα. {:url=>"ελαστική αναζήτηση:9200/", :error_type=>LogStash::Outputs::ElasticSearch::HttpClient::Pool::HostUnreachableError, :error=>"Elasticsearch Unreachable: [http://elasticsearch:9200/:][Resoureail] elasticsearch"}

Και το κούτσουρο μας σέρνεται συνεχώς.

Εδώ επισήμανα με πράσινο το μήνυμα ότι η διοχέτευση ξεκίνησε με επιτυχία, με κόκκινο το μήνυμα σφάλματος και με κίτρινο το μήνυμα σχετικά με την προσπάθεια επικοινωνίας ελαστική αναζήτηση: 9200.
Αυτό συμβαίνει λόγω του γεγονότος ότι στο logstash.conf που περιλαμβάνεται στην εικόνα, υπάρχει έλεγχος για τη διαθεσιμότητα του elasticsearch. Εξάλλου, το logstash υποθέτει ότι λειτουργεί ως μέρος της στοίβας Elk, και το διαχωρίσαμε.

Μπορείς να δουλέψεις, αλλά δεν είναι βολικό.

Η λύση είναι να απενεργοποιήσετε αυτόν τον έλεγχο μέσω της μεταβλητής περιβάλλοντος XPACK_MONITORING_ENABLED.

Ας κάνουμε μια αλλαγή στο docker-compose.yml και ας το εκτελέσουμε ξανά:

version: '3'

networks:
  elk:

volumes:
  elasticsearch:
    driver: local

services:

  logstash:
    container_name: logstash_one_channel
    image: docker.elastic.co/logstash/logstash:6.3.2
    networks:
      - elk
    environment:
      XPACK_MONITORING_ENABLED: "false"
    ports:
      - 5046:5046
   volumes:
      - ./config/pipelines.yml:/usr/share/logstash/config/pipelines.yml:ro
      - ./config/pipelines:/usr/share/logstash/config/pipelines:ro

Τώρα, όλα είναι καλά. Το δοχείο είναι έτοιμο για πειράματα.

Μπορούμε να πληκτρολογήσουμε ξανά στη διπλανή κονσόλα:

echo '13123123123123123123123213123213' | nc localhost 5046

Και δες:

logstash_one_channel | {
logstash_one_channel |         "message" => "13123123123123123123123213123213",
logstash_one_channel |      "@timestamp" => 2019-04-29T11:43:44.582Z,
logstash_one_channel |        "@version" => "1",
logstash_one_channel |     "habra_field" => "Hello Habr",
logstash_one_channel |            "host" => "gateway",
logstash_one_channel |            "port" => 49418
logstash_one_channel | }

Εργαστείτε σε ένα κανάλι

Έτσι ξεκινήσαμε. Τώρα μπορείτε πραγματικά να αφιερώσετε χρόνο για να διαμορφώσετε απευθείας το logstash. Ας μην αγγίξουμε το αρχείο pipelines.yml προς το παρόν, ας δούμε τι μπορούμε να πετύχουμε δουλεύοντας με ένα κανάλι.

Πρέπει να πω ότι η γενική αρχή της εργασίας με το αρχείο διαμόρφωσης καναλιού περιγράφεται καλά στο επίσημο εγχειρίδιο, εδώ εδώ
Αν θέλετε να διαβάσετε στα ρωσικά, χρησιμοποιήσαμε αυτό άρθρο(αλλά η σύνταξη του ερωτήματος είναι παλιά εκεί, πρέπει να το λάβετε υπόψη).

Ας πάμε διαδοχικά από την ενότητα Εισαγωγή. Έχουμε ήδη δει το έργο στο tcp. Τι άλλο μπορεί να είναι ενδιαφέρον εδώ;

Δοκιμή μηνυμάτων χρησιμοποιώντας τον καρδιακό παλμό

Υπάρχει μια τόσο ενδιαφέρουσα δυνατότητα δημιουργίας αυτόματων δοκιμαστικών μηνυμάτων.
Για να το κάνετε αυτό, πρέπει να συμπεριλάβετε την προσθήκη heartbean στην ενότητα εισαγωγής.

input {
  heartbeat {
    message => "HeartBeat!"
   }
  } 

Το ανάβουμε, αρχίζουμε να λαμβάνουμε μια φορά το λεπτό

logstash_one_channel | {
logstash_one_channel |      "@timestamp" => 2019-04-29T13:52:04.567Z,
logstash_one_channel |     "habra_field" => "Hello Habr",
logstash_one_channel |         "message" => "HeartBeat!",
logstash_one_channel |        "@version" => "1",
logstash_one_channel |            "host" => "a0667e5c57ec"
logstash_one_channel | }

Θέλουμε να λαμβάνουμε πιο συχνά, πρέπει να προσθέσουμε την παράμετρο interval.
Έτσι θα λαμβάνουμε ένα μήνυμα κάθε 10 δευτερόλεπτα.

input {
  heartbeat {
    message => "HeartBeat!"
    interval => 10
   }
  }

Λήψη δεδομένων από ένα αρχείο

Αποφασίσαμε επίσης να δούμε τη λειτουργία αρχείου. Εάν λειτουργεί καλά με το αρχείο, τότε είναι πιθανό να μην απαιτείται πράκτορας, τουλάχιστον για τοπική χρήση.

Σύμφωνα με την περιγραφή, ο τρόπος λειτουργίας θα πρέπει να είναι παρόμοιος με το tail -f, δηλ. διαβάζει νέες γραμμές ή, προαιρετικά, διαβάζει ολόκληρο το αρχείο.

Τι θέλουμε λοιπόν να πάρουμε:

1. Θέλουμε να λαμβάνουμε γραμμές που προσαρτώνται σε ένα αρχείο καταγραφής.
2. Θέλουμε να λαμβάνουμε δεδομένα που είναι γραμμένα σε πολλά αρχεία καταγραφής, ενώ μπορούμε να διαχωρίσουμε τι ελήφθη από πού.
3. Θέλουμε να βεβαιωθούμε ότι κατά την επανεκκίνηση του logstash, δεν θα λάβει ξανά αυτά τα δεδομένα.
4. Θέλουμε να ελέγξουμε ότι εάν το logstash είναι απενεργοποιημένο και τα δεδομένα συνεχίσουν να εγγράφονται σε αρχεία, τότε όταν το εκτελέσουμε, θα λάβουμε αυτά τα δεδομένα.

Για να πραγματοποιήσουμε το πείραμα, ας προσθέσουμε μια ακόμη γραμμή στο docker-compose.yml, ανοίγοντας τον κατάλογο όπου τοποθετούμε τα αρχεία.

version: '3'

networks:
  elk:

volumes:
  elasticsearch:
    driver: local

services:

  logstash:
    container_name: logstash_one_channel
    image: docker.elastic.co/logstash/logstash:6.3.2
    networks:
      - elk
    environment:
      XPACK_MONITORING_ENABLED: "false"
    ports:
      - 5046:5046
   volumes:
      - ./config/pipelines.yml:/usr/share/logstash/config/pipelines.yml:ro
      - ./config/pipelines:/usr/share/logstash/config/pipelines:ro
      - ./logs:/usr/share/logstash/input

Και αλλάξτε την ενότητα εισαγωγής στο habr_pipeline.conf

input {
  file {
    path => "/usr/share/logstash/input/*.log"
   }
  }

Αρχίζουμε:

docker-compose up

Για να δημιουργήσουμε και να γράψουμε αρχεία καταγραφής, θα χρησιμοποιήσουμε την εντολή:


echo '1' >> logs/number1.log

{
logstash_one_channel |            "host" => "ac2d4e3ef70f",
logstash_one_channel |     "habra_field" => "Hello Habr",
logstash_one_channel |      "@timestamp" => 2019-04-29T14:28:53.876Z,
logstash_one_channel |        "@version" => "1",
logstash_one_channel |         "message" => "1",
logstash_one_channel |            "path" => "/usr/share/logstash/input/number1.log"
logstash_one_channel | }

Ναι, λειτουργεί!

Ταυτόχρονα, βλέπουμε ότι έχουμε προσθέσει αυτόματα το πεδίο διαδρομής. Έτσι, στο μέλλον, θα μπορούμε να φιλτράρουμε τις εγγραφές από αυτό.

Ας δοκιμάσουμε ξανά:

echo '2' >> logs/number1.log

{
logstash_one_channel |            "host" => "ac2d4e3ef70f",
logstash_one_channel |     "habra_field" => "Hello Habr",
logstash_one_channel |      "@timestamp" => 2019-04-29T14:28:59.906Z,
logstash_one_channel |        "@version" => "1",
logstash_one_channel |         "message" => "2",
logstash_one_channel |            "path" => "/usr/share/logstash/input/number1.log"
logstash_one_channel | }

Και τώρα σε άλλο αρχείο:

 echo '1' >> logs/number2.log

{
logstash_one_channel |            "host" => "ac2d4e3ef70f",
logstash_one_channel |     "habra_field" => "Hello Habr",
logstash_one_channel |      "@timestamp" => 2019-04-29T14:29:26.061Z,
logstash_one_channel |        "@version" => "1",
logstash_one_channel |         "message" => "1",
logstash_one_channel |            "path" => "/usr/share/logstash/input/number2.log"
logstash_one_channel | }

Εξαιρετική! Το αρχείο παραλήφθηκε, η διαδρομή καθορίστηκε σωστά, όλα είναι καλά.

Σταματήστε το logstash και κάντε επανεκκίνηση. Ας περιμένουμε. Σιωπή. Εκείνοι. Δεν λαμβάνουμε ξανά αυτά τα αρχεία.

Και τώρα το πιο τολμηρό πείραμα.

Βάζουμε logstash και εκτελούμε:

echo '3' >> logs/number2.log
echo '4' >> logs/number1.log

Εκτελέστε ξανά το logstash και δείτε:

logstash_one_channel | {
logstash_one_channel |            "host" => "ac2d4e3ef70f",
logstash_one_channel |     "habra_field" => "Hello Habr",
logstash_one_channel |         "message" => "3",
logstash_one_channel |        "@version" => "1",
logstash_one_channel |            "path" => "/usr/share/logstash/input/number2.log",
logstash_one_channel |      "@timestamp" => 2019-04-29T14:48:50.589Z
logstash_one_channel | }
logstash_one_channel | {
logstash_one_channel |            "host" => "ac2d4e3ef70f",
logstash_one_channel |     "habra_field" => "Hello Habr",
logstash_one_channel |         "message" => "4",
logstash_one_channel |        "@version" => "1",
logstash_one_channel |            "path" => "/usr/share/logstash/input/number1.log",
logstash_one_channel |      "@timestamp" => 2019-04-29T14:48:50.856Z
logstash_one_channel | }

Ζήτω! Όλα μαζεύτηκαν.

Όμως, είναι απαραίτητο να προειδοποιήσουμε για τα ακόλουθα. Εάν αφαιρεθεί το κοντέινερ logstash (docker stop logstash_one_channel && docker rm logstash_one_channel), δεν θα παραληφθεί τίποτα. Η θέση του αρχείου μέχρι την οποία διαβάστηκε αποθηκεύτηκε μέσα στο κοντέινερ. Αν ξεκινήσετε από το μηδέν, τότε θα δέχεται μόνο νέες γραμμές.

Ανάγνωση υπαρχόντων αρχείων

Ας υποθέσουμε ότι εκτελούμε το logstash για πρώτη φορά, αλλά έχουμε ήδη αρχεία καταγραφής και θα θέλαμε να τα επεξεργαστούμε.
Εάν εκτελέσουμε το logstash με την ενότητα εισαγωγής που χρησιμοποιήσαμε παραπάνω, δεν θα λάβουμε τίποτα. Μόνο οι νέες γραμμές θα υποβάλλονται σε επεξεργασία από το logstash.

Για να τραβήξετε γραμμές από υπάρχοντα αρχεία, προσθέστε μια πρόσθετη γραμμή στην ενότητα εισαγωγής:

input {
  file {
    start_position => "beginning"
    path => "/usr/share/logstash/input/*.log"
   }
  }

Επιπλέον, υπάρχει μια απόχρωση, αυτό επηρεάζει μόνο νέα αρχεία που δεν έχει δει ακόμη το logstash. Για τα ίδια αρχεία που βρίσκονταν ήδη στο οπτικό πεδίο του logstash, έχει ήδη θυμηθεί το μέγεθός τους και πλέον θα παίρνει μόνο νέες εγγραφές σε αυτά.

Ας σταματήσουμε σε αυτό μελετώντας την ενότητα εισαγωγής. Υπάρχουν πολλές περισσότερες επιλογές, αλλά προς το παρόν, έχουμε αρκετές για περαιτέρω πειράματα.

Δρομολόγηση και μετασχηματισμός δεδομένων

Ας προσπαθήσουμε να λύσουμε το παρακάτω πρόβλημα, ας πούμε ότι έχουμε μηνύματα από ένα κανάλι, μερικά από αυτά είναι ενημερωτικά και άλλα είναι μηνύματα σφάλματος. Διαφέρουν ως προς την ετικέτα. Κάποια είναι INFO, άλλα είναι ERROR.

Πρέπει να τα χωρίσουμε στην έξοδο. Εκείνοι. Γράφουμε ενημερωτικά μηνύματα σε ένα κανάλι και μηνύματα λάθους σε άλλο.

Για να το κάνετε αυτό, μεταβείτε από την ενότητα εισόδου στο φιλτράρισμα και την έξοδο.

Χρησιμοποιώντας την ενότητα φίλτρου, θα αναλύσουμε το εισερχόμενο μήνυμα, παίρνοντας ένα κατακερματισμό (ζεύγη κλειδιού-τιμής) από αυτό, με το οποίο μπορούμε ήδη να εργαστούμε, π.χ. αναλύστε σύμφωνα με τις συνθήκες. Και στην ενότητα εξόδου, θα επιλέξουμε μηνύματα και θα στείλουμε το καθένα στο δικό του κανάλι.

Ανάλυση μηνύματος με grok

Για να αναλύσετε τις συμβολοσειρές κειμένου και να λάβετε ένα σύνολο πεδίων από αυτές, υπάρχει μια ειδική προσθήκη στην ενότητα φίλτρου - grok.

Χωρίς να βάζω τον στόχο μου να δώσω μια λεπτομερή περιγραφή του εδώ (για αυτό αναφέρομαι επίσημη τεκμηρίωση), θα δώσω το απλό παράδειγμά μου.

Για να το κάνετε αυτό, πρέπει να αποφασίσετε για τη μορφή των γραμμών εισόδου. Τα έχω σαν αυτό:

1 Μήνυμα ΠΛΗΡΟΦΟΡΙΩΝ1
2 Μήνυμα ΣΦΑΛΜΑΤΟΣ2

Εκείνοι. Πρώτα αναγνωριστικό, μετά ΠΛΗΡΟΦΟΡΙΕΣ/ΣΦΑΛΜΑ και μετά κάποια λέξη χωρίς κενά.
Δεν είναι δύσκολο, αλλά αρκετά για να κατανοήσουμε την αρχή της λειτουργίας.

Έτσι, στην ενότητα φίλτρου, στο πρόσθετο grok, πρέπει να ορίσουμε ένα μοτίβο για την ανάλυση των συμβολοσειρών μας.

Θα μοιάζει με αυτό:

filter {
  grok {
    match => { "message" => ["%{INT:message_id} %{LOGLEVEL:message_type} %{WORD:message_text}"] }
   }
  } 

Βασικά, είναι μια κανονική έκφραση. Χρησιμοποιούνται έτοιμα μοτίβα, όπως INT, LOGLEVEL, WORD. Η περιγραφή τους, καθώς και άλλα μοτίβα, μπορείτε να δείτε εδώ. εδώ

Τώρα, περνώντας από αυτό το φίλτρο, η συμβολοσειρά μας θα μετατραπεί σε κατακερματισμό τριών πεδίων: message_id, message_type, message_text.

Θα εμφανιστούν στην ενότητα εξόδου.

Δρομολόγηση μηνυμάτων στην ενότητα εξόδου με την εντολή if

Στην ενότητα εξόδου, όπως θυμόμαστε, επρόκειτο να χωρίσουμε τα μηνύματα σε δύο ροές. Μερικά - τα οποία είναι iNFO, θα τα εξάγουμε στην κονσόλα και με σφάλματα θα τα εξάγουμε σε ένα αρχείο.

Πώς μπορούμε να μοιραστούμε αυτά τα μηνύματα; Η κατάσταση του προβλήματος υποδηλώνει ήδη μια λύση - σε τελική ανάλυση, έχουμε ήδη ένα αποκλειστικό πεδίο message_type, το οποίο μπορεί να λάβει μόνο δύο τιμές INFO και ERROR. Σε αυτό θα κάνουμε μια επιλογή χρησιμοποιώντας τη δήλωση if.

if [message_type] == "ERROR" {
        # Здесь выводим в файл
       } else
     {
      # Здесь выводим в stdout
    }

Η περιγραφή της εργασίας με πεδία και τελεστές βρίσκεται σε αυτήν την ενότητα επίσημο εγχειρίδιο.

Τώρα, για το ίδιο το συμπέρασμα.

Έξοδος κονσόλας, όλα είναι ξεκάθαρα εδώ - stdout {}

Αλλά η έξοδος στο αρχείο - θυμηθείτε ότι τα τρέχουμε όλα αυτά από το κοντέινερ και για να είναι προσβάσιμο το αρχείο στο οποίο γράφουμε το αποτέλεσμα από έξω, πρέπει να ανοίξουμε αυτόν τον κατάλογο στο docker-compose.yml.

Итого:

Η ενότητα εξόδου του αρχείου μας μοιάζει με αυτό:


output {
  if [message_type] == "ERROR" {
    file {
          path => "/usr/share/logstash/output/test.log"
          codec => line { format => "custom format: %{message}"}
         }
    } else
     {stdout {
             }
     }
  }

Προσθέστε έναν ακόμη τόμο στο docker-compose.yml για έξοδο:

version: '3'

networks:
  elk:

volumes:
  elasticsearch:
    driver: local

services:

  logstash:
    container_name: logstash_one_channel
    image: docker.elastic.co/logstash/logstash:6.3.2
    networks:
      - elk
    environment:
      XPACK_MONITORING_ENABLED: "false"
    ports:
      - 5046:5046
   volumes:
      - ./config/pipelines.yml:/usr/share/logstash/config/pipelines.yml:ro
      - ./config/pipelines:/usr/share/logstash/config/pipelines:ro
      - ./logs:/usr/share/logstash/input
      - ./output:/usr/share/logstash/output

Ξεκινάμε, προσπαθούμε, βλέπουμε τη διαίρεση σε δύο ρεύματα.

Πηγή: www.habr.com