Wie man MQTT-Nachrichten protokolliert und analysiert
mit dem ELK Stack?

Einleitung

DieELK Stack(Elasticsearch, Logstash und Kibana) ist eine mächtige Open-Source Log-Management- und Analyseplattform.Elasticsearch, sein Kern Bauteil, StützenWiederholung, so dass Daten über mehrere kopiert werden Knoten. Dies gewährleistetWie ist das?durch die Vermeidung von Datenverlusten bei Knotenpunkt Fehler.

Replication in Elasticsearch verbessertDer Präsidentvon die Suche Anfragen über Repliken. Es verbessert auchSkalierbarkeit, erlaubt Systeme bis mehr Anfragen effizient bearbeiten. Standardmäßig erstellt Elasticsearcheine In den Warenkorb Shard, aber dies kann auf Grund der Anforderungen angepasst werden. Richtige Replikationsstrategien HilfeLetzter Beitrag, Fehlertoleranz und Gesamtsystemsicherheit.

Elasticsearch

Elasticsearch ist einVerteilte Such- und Analysemaschinemit einer Leistung von mehr als 20 kW für die Handhabung großer Volumina Daten in Echtzeit. Es speichert Daten in einemNoSQLFormat mit JSON Dokumente und bietet schnelleVolltextsucheFähigkeiten. Daten werden mit einerInvertierter Index, die schnelle Retrieval und Analyse. Elasticsearch isSkalierbar, horizontal tragend Skalierung durch Hinzufügen von mehr Knoten zu einem Cluster. Es bietetVervielfältigung und Verfestigung, Daten sicherstellen Redundanz undDas ist gut. Wie man sieht. Gemeinsame In den Warenkorblog-Analyse, sicherheitsüberwachung und geschäft Intelligenz.

ZEITSCHRIFT

Logstash ist einDatenverarbeitung Pipelinedas sammelt, transformiert, und sendet Daten an verschiedene Ziele, einschließlich Elasticsearch. Es kann Daten vonMehr zum Themawie Logs, Metriken, Datenbanken und Nachrichtenwarten. Logstash verarbeitet Daten mit einem leistungsstarkenFilter System, die Umwandlung, Anreicherung und Parsing erlaubt. Es unterstützt verschiedene Plugins um verschiedene zu handhaben Datenformate, einschließlich JSON, CSV und syslog. Logstash wird häufig fürLog-Sammlung, Datentransformation, und Echtzeit-Datenverarbeitungvor der Lagerung.

Kibana

Kibana ist einVisualisierungen- und Analysewerkzeugentwickelt, um interagieren mit in Elasticsearch gespeicherten Daten. Es bietet eine intuitiveDas ist dasfür die Erstellung Visualisierungen wie Charts, Grafiken, und Karten. Kibana ermöglicht Echtzeit-Datenexploration, so dass es einfach ist, Protokolle zu analysieren und zu überwachen Systemleistung. Es umfasst Funktionen wieÜber uns Überwachung und Alarmierung.Benutzer können interaktive Dashboards erstellen und Alarme für Anomalien in Daten einrichten. Kibana ist weit verbreitet verwendet fürLog-Analyse, Sicherheitsüberwachung und Business Nachrichtenmeldung.



Bevor wir beginnen, stellen Sie sicher, dass Sie:

  • ALinux Server(Ubuntu 20.04 empfohlen)
  • MQTTRouteinstalliert und läuft
  • Java 8 oder später(erforderlich für ELK und Kafka)
  • Elastik, Logstash und Kibana (ELK Stack)installiert

Schritt 1: Installieren Sie Java 8 für ELK

  • Update der Paketliste :sudo apt update
  • Installieren Sie Sie OpenJDK 8:sudo apt install -y openjdk-8-jdk
  • Überprüfung der Instalation:java -version
  • Stellen Sie sicher, dass die Ausgabe installiert ist java8.


Wenn Java auf der Maschine installiert ist, wird die folgende Ausgabe angezeigt:

Text geknüpft

openjdk Version "1.8.0_442"
OpenJDK Runtime Environment (Bauen 1.8.0_442-8u442-b06~us1-0ubuntu1~20.04-b06)
OpenJDK 64-Bit Server VM (Bauen 25.442-b06, Mischmodus)

Schritt 2: Installieren und Konfigurieren des ELK Stacks

  • Elasticsearch installieren:

Systempakete:sudo apt update && sudo apt upgrade - Ja.


Elasticsearch herunterladen 8.17.3 :


! https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-8.17.3-amd64.deb


Elastiksuche installieren:-i Forschung-8.17.3-amd64.deb


Konfigurieren Sie suchen: Sie in das Verzeichnis und klicken Sie auf den Befehl, um die Konfiguration der Konfiguration Elastiksuche richtig - Ja.


Sudo Nano /etc/elasticsearch/elasticsearch.yml


♪ Netzwerkeinstellungen
Network.host: 0.0.0.0
http.port: 9200


♪ Clustereinstellungen
Cluster.name: my-cluster
node.name: node-1


# Sicherheitseinstellungen
xpack.security.enabled: true

Elastische Forschung starten und aktivieren:


Pressemitteilungen Forschung


sudo systemctl starten

Überprüfen Sie den Status, um zu sehen, ob Elasticsearch richtig läuft; das Ergebnis sollte wie unten gezeigt erscheinen:

  • elastiksearch.service - Elasticsearch

Beladen: laden (/lib/systemd/systemd/elasticsearch.service; aktivieren; Voreinstellung: aktiviert

Aktiv: aktiv (laufend) seit Tue 2025-03-25 11:59:26 IST; 16min vor

Dok.: http://www.elastic.co

HauptpID: 1554 (java)

Aufgaben: 92 (Begrenzung: 18932)

Speicher: 8,5G

CGroup: /system.slice/elasticsearch.service

•1554 /usr/share/elasticsearch/jdk/bin/java -Xm4m -Xmx64m -XX: Mehr Info -Dcli.name=server -Dcli.script=/usr/share/elasticsearch/bin/e>

Ă€_2134 /usr/share/elasticsearch/jdk/bin/java -Des.networkaddress.cache.ttl=60 -Des.networkaddress.cache.negative.ttl=10 -XX:+AlwaysPreTouc>

・2158 /usr/share/elasticsearch/modules/x-pack-ml/platform/linux-x86_64/bin/controller

25 11:58:11 host systemd[1]: Beginnend Elastikforschung...

25 11:58:47 host systemd-entrypoint[2134]: CompileCommand: Dontinlin java/lang/invoke/MethodHandle.setAsTypCache bool dontinline = true

25 11:58:47 host systemd-entrypoint[2134]: CompileCommand: Dontinlin java/lang/invoke/MethodHandle.asTypUncached bool dontinline = true

25 11:59:26 host systemd[1]: Started Elasticsearch.

Linien 1-16/16 (END)

Kennwort für elastische Forschung generieren. Durch Default Benutzername und Passwort ist notwendig, um die elastische Forschung Version 8.17.3 zu verbinden


Verwenden Sie den folgenden Befehl, um das Elasticsearch-Passwort zu generieren:


sudo /usr/share/elasticsearch/bin/elasticsearch-reset-password -u elastisch


Installation überprüfen:


Nachdem das Passwort erfolgreich generiert wird, führen Sie den folgenden Befehl aus, um es zu überprüfen:

elastischsearch Version 8.17.3 richtig installiert.
curl -k http://localhost:9200 -u elastisch:your_password


Wenn die elastische Forschung die folgenden Ergebnisse richtig konfiguriert hat :

{\cHFFFF}
"Name" : "Hostname",
"cluster_name" : "elasticsearch",
"cluster_uuid" : "kKzyHzjcTGGei6ikXndcFA",
"Version" {\cHFFFF}
"Zahl" : "8.17.3",
"build_flavor" : "Default",
"build_type" : "Deb",
"build_hash" : "a091390de485bd4b127884f7e565c0cad59b10d2",
"build_date" : "2025-02-28T10:07:26.089129809Z",
"build_snapshot" : falsch,
"lucene_version" : "9.12.0",
"minimum_wire_compatibility_version" : "7.17.0",
"minimum_index_compatibility_version" : "7.0.0"
},
"tagline" : "Du weißt, für die Suche"
}

TLS/SSL-Konfiguration in elastischer Forschung 8.17.3 :


Elasticsearch 8.17.3 umfasst integrierte Sicherheit und erfordert TLS/SSL für die Kommunikation. Folgen Sie diesen Schritten, um TLS/SSL für beide HTTP- und Transportschichten zu konfigurieren.


  • SSL-Zertifikat generieren :

Elasticsearch umfasst ein integriertes Werkzeug zur Erstellung von Zertifikaten. Verwenden Sie den folgenden Befehl, um ein Zertifikat zu generieren:


sudo /usr/share/elasticsearch/bin/elasticsearch-certutil zert --silent - Pem --keep-ca-key --out /etc/elasticsearch/certs.zip


Auszug der Zertifikate :


sudo apt install unzip -y
Sudo mkdir -p /etc/elastische Forschung/Zertifikate
sudo unzip /etc/elasticsearch/certs.zip -d /etc/elasticsearch/certs/
sudo chmod -R 750 /etc/elasticsearch/certs
sudo chown -R elastische Forschung:elastische Forschung /etc/elastische Forschung/Kräuter


    2. Konfigurieren Sie die elastische Suche für TLS/SSL:


Bearbeiten Sie die elastische Konfigurationsdatei mithilfe des folgenden Befehls:


Sudo Nano /etc/elasticsearch/elasticsearch.yml


Fügen oder aktualisieren Sie die Einstellungen:

Wenn die folgenden Eigenschaften nicht in elastischsearch.yml vorhanden sind, fügen Sie sie hinzu. Wenn sie bereits existieren, setzen oder aktivieren sie wahr.


# Sicherheitsfunktionen aktivieren
xpack.security.enabled: true
xpack.security.enrollment.enabled: true


# Verschlüsselung und gegenseitige Authentifizierung zwischen Clusterknoten aktivieren
xpack.security.transport.ssl:
aktiviert: wahr
verifikation_mode: zertifikat
keystore.path: certs/transport.p12
Truststore.path: certs/transport.p12


# Erstellen Sie nur einen neuen Cluster mit dem aktuellen Knoten
# Weitere Knoten können später noch dem Cluster beitreten
Cluster.initial_master_nodes: ["hostname"]


# HTTP API-Verbindungen von überall zulassen
# Verbindungen werden verschlüsselt und benötigen Benutzerauthentifizierung
http.host: 0.0.0.0


# Erlauben Sie anderen Knoten, den Cluster von überall zu verbinden
# Verbindungen werden verschlüsselt und gegenseitig authentifiziert
#transport.host: 0.0.0.0


Neuart Elasticsearch:

Neustart der Elastikforschung zur Anwendung der Änderungen :

Das neue System gestartet forschung. Service


ANHANG Sie danach, ob Elasticsearch mit dem folgenden Das ist ein Fehler. :

sudo systemctl status häufigsearch.service


Verifizieren Sie Elasticsearch TLS/SSL Konfiguration :

Standard Zertifikate im Verzeichnis

sudo /etc/elastikforschung/certs


ANHANG Sie den folgenden Befehl, um die TLS/SSL-Installation zu überprüfen:

curl -k --cacert /etc/elasticsearch/certs/ca.crt -u Bewältigung:Passwort 'https://localhost:9200'


Wenn Elasticsearch korrekt für TLS/SSL aufgenommen ist, erscheint das Ergebnis wie unten gezeigt:

(STÖHNT)
"Name" : "Hostname",
"cluster_name" : "elasticsearch",
"cluster_uuid" : "kKzyHzjcTGGei6ikXndcFA",
"Version" {\cHFFFF}
"Zahl" : "8.17.3",
"build_flavor" : "Default",
"build_type" : "Deb",
"build_hash" : "a091390de485bd4b127884f7e565c0cad59b10d2",
"build_date" : "2025-02-28T10:07:26.089129809Z",
"build_snapshot" : Ich weiß nicht.
"lucene_version" : "9.12.0",
"minimum_wire_compatibility_version" : "7.17.0",
"minimum_index_compatibility_version" : "7.0.0"
},
"tagline" : "Du weißt, für die Suche"
}

Logstash installieren:


Logstash 8.17.3 herunterladen


! https://artifacts.elastic.co/downloads/logstash/logstash-8.17.3-amd64.deb


Logstash 8.17.3 Installieren


sudo dpkg -i logstash-8.17.3-amd64.deb


Logstash Installation:


Überprüfen Sie, ob die Logstash-Version 8.17.3 mit dem folgenden Befehl installiert ist. Wenn die Version angezeigt wird, es bedeutet, dass Logstash korrekt auf unserer Maschine installiert ist.

/usr/share/logstash/bin/logstash --version


Das Ergebnis sieht aus wie:

Verwendung gebündelter JDK: /usr/share/logstash/jdk
Logstash 8.17.3


Starten und aktivieren logstash :

sicherheits- und sicherheitsmanagement

sudo systemctl start logstash


Sie den Status der Logstash-Installation :

sudo systemctl status logstash


Der Ausgang ist wie folgt:

Logstash. Service - logstash
beladen: beladen (/lib/systemd/systemd/systemd/systemd/logstash.service; aktivieren; Anbieter-Preset: aktivieren)
Aktiv: aktiv (laufend) seit Di 2025-03-25 12:55:40 IST; vor 11s
HauptpID: 4519 (java)
Aufgaben: 24 (Begrenzung: 18932)
Speicher: 344,9M
CGroup: /system.slice/logstash.service
— 4519 /usr/share/logstash/jdk/bin/java -Xms1g -Xmx1g -Djava.awt.headless=true -Dfile.encoding=UTF-8 -Djruby.compile.invokedynamic=true -XX:+HeapDumpOnOutOfMemoryError - >
25 12:55:40 hostname systemd[1]: Started logstash.
Mar 25 12:55:40 hostname logstash[4519]: Verwendung gebündelter JDK: /usr/share/logstash/jdk
Zeile 1-11/11 (END)

Logstash Logs in Logdatei überprüfen

ANHANG Logstash schreibt Protokolle zu /var/logstash/logstash-plain.log. Logs anzeigen mit:

sudo tail -f /var/logstash/logstash-plain.log


2. Um Logstash-Logs zu überprüfen, verwenden Sie Journalctl

sudo journalctl -u logstash -f


Kibana 8.17.3 herunterladen

Um Kibana 8.17.3 herunterladen, verwenden Sie haben die Befehl

wget https://artifacts.elastic.co/downloads/kibana/kibana-8.17.3-amd64.deb


Kibana 8.17.3 :

sudo dpkg -i kibana-8.17.3-amd64.deb


Kibana konfigurieren 8.17.3 :

Wiederholungen Sie den folgenden Schritten, um Kibana zu konfigurieren
Schritt-1:

Sie die Konfigurationsdatei kibana:Sudo Nano /etc/kibana/kibana.yml


Schritt 2:
Ungenehmigen und ändern Sie die folgende Einstellung:

Server.port: "5601"
server.host: "localhost"
: ["http://localhost:9200"]
Normalsearch.username: "kibana_system"
Suchbegriff: "Passwort"
Suchergebnis.maxSockets: 1024


Nach dem Ändern der Einstellungen speichern und Sie die Konfiguration.

Starten und aktivieren Kibana:


sicherheitsmanagement

sudo systemctl start kibana


Um zu überprüfen, ob Kibana richtig beansprucht ist, erscheinen die Ergebnisse wie gezeigt unten:


  • kibana.service - Kibana

Beladen: beladen (/lib/systemd/systemd/kibana.service; aktivieren; Anbieter Voreinstellung: aktiviert

Aktiv: aktiv (laufend) seit Tue 2025-03-25 15:14:22 IST; 9min vor

Dok.: http://www.elastic.co

HauptpID: 1554 (java)

HauptpID: 9690 (Node)

Aufgaben: 11 (Begrenzung: 18932)

Speicher: 598.2M

CGroup: /system.slice/kibana.service

・ — 9690 /usr/share/kibana/bin/../node/glibc-217/bin/node /usr/share/kibana/bin/../src/cli/dist

25 15:17:09 hostname kibana[9690]: [2025-03-25T15:17:09.848+05:30][ERROR][http] 400 Schlecht Anfrage

25 15:19:46 Hostname kibana[9690]: [2025-03-25T15:19:46.504+05:30][INFO ][plugins.securitySolution] Fetch Risk Motor Metriken

25 15:19:46 Hostname kibana[9690]: [2025-03-25T15:19:46.860+05:30][INFO ][plugins.cloudSecurityPosture] Cloud Security Telemetrie: Ressourcen-Payload erfolgreich aufgenommen

25 15:19:46 Hostname kibana[9690]: [2025-03-25T15:19:46.861+05:30][INFO][plugins.cloudSecurityPost] Cloud Security Telemetrie: Accounts Payload erfolgreich durchgeführt

25 15:19:46 Hostname kibana[9690]: [2025-03-25T15:19:46.862+05:30][INFO][plugins.cloudSecurityPost] Cloud Security Telemetrie: Regeln-Payload erfolgreich durchgeführt

25 15:19:47 Hostname kibana[9690]: [2025-03-25T15:19:47.193+05:30][INFO][plugins.cloudSecurityPost] Cloud Security Telemetrie: Alerts Nutzlast wurde erfolgreich angewandt

25 15:19:47 Hostname kibana[9690]: [2025-03-25T15:19:47.199+05:30][INFO][plugins.cloudSecurityPost] Cloud Security Telemetrie: Cloud Accounts Payload erfolgreich durchgeführt

25 15:19:47 Hostname kibana[9690]: [2025-03-25T15:19:47.336+05:30][INFO][plugins.cloudSecurityPost] Cloud Security Telemetrie: Muted Rules Payload erfolgreich durchgeführt

25 15:19:47 Hostname kibana[9690]: [2025-03-25T15:19:47.372+05:30][INFO][plugins.cloudSecurityPost] Cloud Security Telemetrie: Installationen-Payload wurde unterstützt

25 15:19:47 Hostname kibana[9690]: [2025-03-25T15:19:47.406+05:30][INFO][plugins.cloudSecurityPost] Cloud Security Telemetrie: Indices Nutzlast wurde erfolgreich angewandt


Kibana im Browser öffnen :


Kibana ausgewählt ist, öffnen Sie es mit der folgenden URL:


http://<server_ip>:5601


Zum Beispiel:http://localhost:5601


Kibana 8.17.3 mit TLS/SSL konfigurieren:


Wenn Elasticsearch bereits mitTLS/SSL (HTTPS), Sie müssen konfigurierenKibana zu kommunizieren ...Damit.


Jetzt sehen wir, wie man Kibana mit TLS/SSL einrichten kann:


Schritt 1:

Die Konfigurationsdatei:Sudo Nano /etc/kibana/kibana.yml


Schritt 2:
Kopieren Sie das Certficate aus dem Verzeichnis der Suchzerts in das kibana certs Verzeichnis:


cp /etc/elasticsearch/certs/elastic-stack-ca.crt /etc/kibana/


Setzen Sie die Korrektorberechtigungen für Zertifikate fest :


chown kibana:kibana /etc/kibana/elastic-stack-ca.crt


chmod 600 /etc/kibana/elastisch-stack-ca.crt


Sie haben dasigentum Dateitum und die Bedrohung:


ls -l /etc/kibana/elastic-stack-ca.crt


Die Ausgabe erscheint wie folgt:
-rwxr-x--- 1 kibana kibana 1915 25 13:56 /etc/kibana/elastic-stack-ca.crt


Schritt 3:
Ungenehmigen und ändern Sie die folgende Einstellung:


Server.port: 5601
server.host : "localhost"
„Software“:
Normalsearch.username: "kibana_system"
Suchbegriff: "Passwort"
Suchergebnis.maxSockets: 1024
Begriffsbestimmungen: [/etc/kibana/http_ca.crt]


Das ist nicht alles. Sie die kibana Konfigurationen.


Wenn du willst Sie habenKibana über HTTPSStatt http://localhost:5601 In den Warenkorb Sie TLS in kibana.yml:


Das ist ein Sie sind hier:


server.sl.enabled: true
server.ssl.certificate: "/etc/kibana/kibana.crt"
server.ssl.key: "/etc/kibana/kibana.key"


Schritt 4:
Starten Sie die Kibana, um die Bedingungen definiert:


sudo systemctl restart kibana. Service


Wenn Sie den Elasticsearch Benutzer nicht haben, verwenden Sie den Befehl unten, um manuell Das Passwort zurücksetzen.

Passwort manuell für den Benutzer festlegen


Verwenden Sie den folgenden Befehl in Ihrem Terminal:


Sudo /usr/share/elasticsearch/bin/elasticsearch-reset-password -u <username> -i


Schritt-4: Versuchen Sie, Kibana zu erreichen


Wenn Sie benutzenKibana HTTPS, verwenden:


http://localhost:5601


wennKibana SSL ist deaktiviert, verwenden:


http://localhost:5601


Wenn Kibana auf einemRemote Server, localhost durch die IP des Servers ersetzen:


http://your-server-ip:5601


MQTT herunterladen Brokerund in unserer Maschine laufen und es sieht aus wie die


Konfiguration des Netzwerks


DieOsterforschung Connectorin CrystalMQ erlaubt nahtlose Datenabruf vom Broker und effiziente Indexierung in Elasticsearch. Es stellt sicher sichere Authentifizierung, verschlüsselte Kommunikation und zuverlässige Datenspeicherung, so dass es wesentliche Komponente zur Verwaltung und Analyse von IoT-Daten.


  • Host:Geben Sie den Hostnamen oder die IP-Adresse der Elasticsearch ein Server.
  • Hafen:Geben Sie den Port an, auf dem der Elasticsearch Server ist Zuhören (Standard):9200)
  • Index:Geben Sie den Namen des Elasticsearch Index zum Speichern oder Daten abrufen.
  • Authentication:Wählen Sie die Authentifizierungsmethode:
    Grund Auth:Authenticate mit einem Benutzername und Passwort.
    PKI Auth:Authenticate mit Public Key Infrastructure (PKI) mit Client Zertifikate.
    JWT Auth:Authenticate mit einem JSON Web Token (JWT).
  • Zugang Token:Wenn SieJWT Authentifizierung,Geben Sie einen gültigen JWT-Zugangstoken ein.
  • SSL:ErmöglichenSSL/TLSfür eine sichere Verbindung. Wenn aktiviert, stellen Sie die erforderlichen Zertifikatsdateien zur Verfügung.
  • CA Certificate:Hochladen derBescheinigung Bescheinigung der Behörde (CA)bei Verwendung eines selbstsignierten Zertifikats.
  • Client Certificate & Key:Wenn SiePKI-Authentifizierung,laden Sie die gewünschteKundenzertifikat und Clientschlüssel

Authentisierung in Elastikforschung Anschluss


Elasticsearch unterstützt mehrere Authentifizierungsmethoden, um sicher zu stellen Zugang.


Grundlegende


Standardmäßig benötigt Elasticsearch eineBenutzername und Passwort vergessen?zur Authentifizierung. Ohne gültige Anmeldeinformationen wird die Verbindung scheitern, führt zu Authentifizierungsfehlern. Dieses Verfahren ist einfach und weit verbreitet zur Sicherung des Zugangs.


PKI (Public Key Infrastructure) Authentifizierung


PKI-Authentifizierung verwendetDigital Zertifikateanstatt Passwörter für sicheren Zugriff. In diesem Ansatz:


  • AKundenbetreuungund! Schlüsselwörtersind verpflichtet, eine vertrauenswürdige Verbindung mit Elasticsearch herzustellen.
  • Der Server überprüft das Zertifikat gegen eine vertrauenswürdigeBescheinigung Behörde (CA)um Authentizität zu gewährleisten.
  • Diese Methode verbessert die Sicherheit durch die Beseitigung der Notwendigkeit von Passwörtern und sichert verschlüsselt und authentifizierte Kommunikation.

JWT (JSON Web Token) Authentication


JWT-Authentifizierung bietetDas ist ein gutes Beispiel. ) Steuerung, SteuerungwennZugang zu Tokenwird anstelle von traditionelle Anmeldeinformationen.


  • Authentifizierungsanbieter wieGoogle oder Keycloakerzeugen Zugriff auf Token für Benutzer.
  • Elasticsearch validiert das Token mit einemDie Geschichte der Weltin der seine Konfiguration.
  • Dieses Verfahren ermöglicht(SSO)und sicher, skalierbar Authentifizierung.

2. Verschlüsselung: SSL/TLS in Elasticsearch


Um Daten im Transit zu schützen, unterstützt ElasticsearchSSL/TLS-Verschlüsselung,Sicherstellung der Kommunikation zwischen Clients und Servern bleibt privat und sicher.


SSL/TLS-Verschlüsselung einrichten


  • OpenSSLwird verwendet, um Sicherheitszertifikate zu erzeugen, die für verschlüsselte Kommunikation.
  • Elasticsearch erfordert spezifische Zertifikate:

    Ausgewählte CA Zertifikate- Eine vertrauenswürdige Bescheinigung, die andere Zertifikate.

    2. Kundenzertifikat- Erkennt den Kunden Elasticsearch.

    3. Wir haben es geschafft.- Ein privater Schlüssel zur Authentisierung der Kundenzertifikat.


Durch die korrekte Konfiguration der Authentifizierung und Verschlüsselung, die Elastische Forschung Connector in CrystalMQ bietet einensicher, skalierbar und effizientLösung für die Verwaltung von IoT-Daten.


Elasticsearch Connector

Im Elasticsearch Connector, grundlegende Authentifizierung (Benutzername und Passwort) ist erforderlich, so dass es als Standard konfiguriert werden kann.


Elasticsearch Connector Details


Geben Sie die Konfigurationen im Elasticsearch Connector ein, einschließlich des Elasticsearch Index, um die Verbindung herzustellen.


Sobald alle Konfigurationsfelder abgeschlossen sind, wählen Sie "Speichern", um die Konfigurationen in der Datenbank zu speichern.


Wenn der Elasticsearch-Steckverbinder erfolgreich angeschlossen ist, wird die Schaltfläche Aktiv/Disable toggle grün angezeigt. Wenn es ein Problem mit dem Steckverbinder gibt, wird die Schaltfläche rot angezeigt.


Elasticsearch Connector


In der obigen Konfiguration wird nur Basic Authentication verwendet, ohne TLS/SSL-Verschlüsselung zu ermöglichen.


Verwenden Sie Logstash, um Daten von Elasticsearch anzuzeigen. Mal sehen, wie wir die Logstash-Pipeline verwenden können, um diese Daten zu erfassen.


Logstash bietet verschiedene Methoden, um Daten auszugeben, je nachdem, wie Sie speichern, visualisieren oder weiterleiten möchten. Einige gemeinsame Ausgabeoptionen beinhalten:


  • Elasticsearch
  • Datei
  • Standardausgang (Ausgang)
  • Kafka
  • Redis
  • AWS S3 / Google Cloud Storage
  • InfluxDB
  • E-Mail senden
  • TCP / UDP
  • Http
  • Graphit
  • WebSocket
  • Google BigQuery

Schritt -1 :

Konfigurieren Sie Logstash, um Daten von Elasticsearch abzurufen:


  • Zurück zur Übersicht
    Erstellen einer Datei namensPressemitteilungenin der ZEITSCHRIFT
    Konfigurationsverzeichnis

  • Bearbeiten der Konfigurationsdatei
    ÖffnenPressemitteilungenund folgendes hinzufügen Konfiguration:

Eingang {
Elastische Forschung
hosts => ["http://your-elasticsearch-host:9200"]
Index => "Ihr-Index-Name"
Benutzer => "your-username"
Passwort => "your-password"
}
}
Ausgang {
stdout!
Codec => Json
}
}


Starten Sie die Logstash


Starten Sie Logstash mit dem folgenden Befehl:


Falls Logstash über einen Paketmanager installiert wurde:


sudo /usr/share/logstash/bin/logstash -f
/etc/logstash/conf.d/elasticsearch_logstash.conf


Zum Beispiel: sudo /usr/share/logstash/bin/logstash -f

~/Documents/elasticsearch_logstash.conf


Nach der Konfiguration von Logstash, verwenden Sie den Befehl unten, um zu überprüfen, ob die Konfiguration korrekt angewendet wurde.


sudo /usr/share/logstash/bin/logstash --path.settings /etc/logstash -f
path/to/elasticsearch_logstash.conf --config.test_and_exit


Ist die Konfiguration korrekt, so ist die Ausgabe wie folgt:

Verwendung gebündelter JDK: /usr/share/logstash/jdk

Senden von Logstash-Logs an /var/logstash, die jetzt über

log4j2.properties

[2025-03-26T11:40:50,375][WARN ][logstash.runner ] HINWEIS: Laufen

Logstash als Superuser wird stark entmutigt, da es eine Sicherheit Risiko.

Setzen Sie 'allow_superuser' auf false für bessere Sicherheit.

[2025-03-26T11:40:50,386][INFO][logstash.runner] Log4j Konfiguration

Pfad verwendet wird: /etc/logstash/log4j2.properties

[2025-03-26T11:40:50,388][INFO][logstash.runner] Starten von Logstash

"logstash.version"=>"8.17.3", "jruby.version"=>"jruby 9.4.9.0 (3.1.4) 2024-11-

04 547c6b150e OpenJDK 64-Bit Server VM 21.0.6+7-LTS auf 21.0.6+7-LTS +indy

(x86_64-linux)

[2025-03-26T11:40:50,392]

Fahnen: [-Xms1g, -Xmx1g, -Djava.awt.headless=true, -Dfile.encoding=UTF-8, - Ja.

Djruby.compile.invokedynamic=true, -

XX:+HeapDumpOnOutOfMemoryError, -

Djava.security.egd=file:/dev/urandom, - Ja.

Dlog4j2.isThreadContextMapUnheritable=true, -Dlogstash.jackson.stream-

read-constraints.max-string-length=200000000, - Ja.

Dlogstash.jackson.stream-read-constraints.max-number-length=10000, - Ja.

Djruby.regexp.interruptible=true, -Djdk.io.File.enableADS=true, --add-

export=jdk.compiler/com.sun.tools.javac.api=ALL-UNNAMED, --add-

export=jdk.compiler/com.sun.tools.javac.file=ALL-UNNAMED, --add-

export=jdk.compiler/com.sun.tools.javac.parser=ALL-UNNAMED, --add...

export=jdk.compiler/com.sun.tools.javac.tree=ALL-UNNAMED, --add...

export=jdk.compiler/com.sun.tools.javac.util=ALL-UNNAMED, --add-

open=java.base/java.security=ALL-UNNAMED, --add-

open=java.base/java.io=ALL-UNNAMED, --add-

open=java.base/java.nio.channels=ALL-UNNAMED, --add-

open=java.base/sun.nio.ch=ALL-UNNAMED, --add-

open=java.management/sun.management=ALL-UNNAMED, -

Dio.netty.allocator.maxOrder=11]

[2025-03-26T11:40:50,462][INFO]

[org.logstash.jackson.StreamReadConstraintsUtil Jackson] Standardwert

`logstash.jackson.stream-read-constraints.max-string-length`

`200000000`

[2025-03-26T11:40:50,463][INFO]

[org.logstash.jackson.StreamReadConstraintsUtil Jackson] Standardwert

`logstash.jackson.stream-read-constraints.max-number-length`

`10000 `

[2025-03-26T11:40:50,755][WARN ][logstash.config.source.multilocal]

Ignorieren der 'pipelines.yml'-Datei, weil Modul oder Befehlszeile Bitte

werden.

[2025-03-26T11:40:51,186][INFO ][org.reflexions.Reflections] In den Warenkorb

177 ms, um 1 URLs zu scannen, 152 Schlüssel und 530 Werte

[2025-03-26T11:40:51,846] [INFO][logstash. Javapipeline Pipeline `main' ist

`pipeline.ecs_kompatibilität: v8`-Einstellung. Alle Plugins in Das

Pipeline wird standardmäßig `ecs_compatibility => v8`, wenn nicht eindeutig

Das ist nicht alles.

Zurück zur Übersicht

[2025-03-26T11:40:51,848][INFO][logstash.runner] Verwendung

config.test_and_exit mode. Vertrauliches Ergebnis: OK. Aus ZEITSCHRIFT


Überprüfen Sie die Logstash-Protokolle mit dem Befehl unten, um zu überprüfen, ob die Daten von Elasticsearch abgerufen und in Logstash angezeigt werden.


Das Logstash-Log erscheint wie unten gezeigt:


Verwendung gebündelter JDK: /usr/share/logstash/jdk

[WARN ] 2025-03-25 18:16:42.180 [Haupt] Läufer - 'pipeline.buffer.type '

Die Einstellung wird nicht definiert. Vor dem Umzug auf 9.x Sie es auf 'Hap' und

Tuning-Hap-Größe nach oben, oder wählen Sie es auf "direkt" Verhalten.

[INFO ] 2025-03-25 18:16:42.181 [main] Läufer - Starting Logstash

"logstash.version"=>"8.17.3", "jruby.version"=>"jruby 9.4.9.0 (3.1.4) 2024-11-

04 547c6b150e OpenJDK 64-Bit Server VM 21.0.6+7-LTS auf 21.0.6+7-LTS +indy

(x86_64-linux)

[INFO ] 2025-03-25 18:16:42.185 [main] Läufer - JVM Bootstrap Flagge: [-Xms1g]

-Xmx1g, -Djava.awt.headless=true, -Dfile.encoding=UTF-8, -

Djruby.compile.invokedynamic=true, -

XX:+HeapDumpOnOutOfMemoryError, -

Djava.security.egd=file:/dev/urandom, - Ja.

Dlog4j2.isThreadContextMapUnheritable=true, -Dlogstash.jackson.stream-

read-constraints.max-string-length=200000000, - Ja.

Dlogstash.jackson.stream-read-constraints.max-number-length=10000, - Ja.

Djruby.regexp.interruptible=true, -Djdk.io.File.enableADS=true, --add-

export=jdk.compiler/com.sun.tools.javac.api=ALL-UNNAMED, --add-

export=jdk.compiler/com.sun.tools.javac.file=ALL-UNNAMED, --add-

export=jdk.compiler/com.sun.tools.javac.parser=ALL-UNNAMED, --add...

export=jdk.compiler/com.sun.tools.javac.tree=ALL-UNNAMED, --add...

export=jdk.compiler/com.sun.tools.javac.util=ALL-UNNAMED, --add-

open=java.base/java.security=ALL-UNNAMED, --add-

open=java.base/java.io=ALL-UNNAMED, --add-

open=java.base/java.nio.channels=ALL-UNNAMED, --add-

open=java.base/sun.nio.ch=ALL-UNNAMED, --add-

open=java.management/sun.management=ALL-UNNAMED, -

Dio.netty.allocator.maxOrder=11]

[INFO ] 2025-03-25 18:16:42.373 [main] StreamReadConstraints Util - Jackson.

Standardwert überschreiben `logstash.jackson.stream-read-constraints.max-

`200000000`

[INFO ] 2025-03-25 18:16:42.373 [main] StreamReadConstraints Util - Jackson.

Standardwert überschreiben `logstash.jackson.stream-read-constraints.max-

Nummernlänge, beschränkt auf `10000`

[WARN ] 2025-03-25 18:16:42.655 [LogStash::Runner] multilocal - Ignorieren

die 'pipelines.yml'-Datei, weil Module oder Befehlszeilenoptionen sind

!

[INFO ] 2025-03-25 18:16:43.449 [Api Webserver] Agent - erfolgreich Anfang

Logstash API Endpoint {:port=>9601, :ssl_enabled=>false}

[INFO ] 2025-03-25 18:16:43.805 [Converge PipelineAction::Create

!

Reflexionen - Reflexionen dauerte 169 ms, um 1 URLs zu scannen, die 152 Tasten produzierten und

[INFO ] 2025-03-25 18:16:44.521 [Converge PipelineAction::Create

!

javapipeline - Pipeline `main` ist mit

`pipeline.ecs_kompatibilität: v8` Einstellung. Alle Plugins in dieser Pipeline wird

Standard auf `ecs_compatibility => v8`, vorausgesetzt nicht angepasst

Sonst.

[INFO ] 2025-03-25 18:16:44.579 [[main]-pipeline-manager] javapipeline -

Startpipeline {:pipeline_id=>"main", "pipeline.workers"=>4,

"pipeline.batch.size"=>125, "pipeline.batch.delay"=>50,

"pipeline.max_inflight"=>500,

"pipeline.sources"=>[" /home/bevywise/Documents/easticsearch_logstash.

conf"], :thread=>"#<Thread:0x612a9b3c /usr/share/logstash/logstash-

core/lib/logstash/java_pipeline.rb:138 run>"}

[INFO ] 2025-03-25 18:16:45.277 [[main]-pipeline-manager] javapipeline -

Pipeline Java Ausführung Initialisierungszeit "Sekunden"=>0.7

[INFO] 2025-03-25 18:16:45.591 [[main]-pipeline-manager] Forschung - Ja.

`search_api => auto` aufgelöst auf `search_after ` {:elasticsearch=>"8.17.3"}

[INFO] 2025-03-25 18:16:45.593 [[main]-pipeline-manager] Forschung - Ja.

ECS-Kompatibilität ist aktiviert, aber `target` Option wurde nicht erfüllt. Das

kann dazu führen, dass Felder auf der obersten Ebene der Veranstaltung eingestellt werden, wo sie sind

wahrscheinlich mit dem Elastischen gemeinsamen Schema zusammenstoßen. Es wird empfohlen,

die `target`-Option, um Schemakonflikte zu vermeiden (wenn Ihre Daten EG

konform oder nicht konflikt, fühlen Sie sich frei, diese Nachricht zu ignorieren)

[INFO ] 2025-03-25 18:16:45.597 [[main]-pipeline-manager] javapipeline -

Pipeline gestartet {"pipeline.id"=>"main"}

[INFO ] 2025-03-25 18:16:45.600 [[main]<easticsearch] Nach der Suche - Erstellung

Zeitpunkt (PIT)

[INFO] 2025-03-25 18:16:45.619 Agent - Pipelines läuft

{:count=>1, :running_pipelines=>[:main], :non_running_pipelines=>[]}

[INFO ] 2025-03-25 18:16:45.630 [[main]<easticsearch] Nach der Suche - !

Start

[INFO ] 2025-03-25 18:16:45.703 [[main]<easticsearch] Nach der Suche - !

abgeschlossen

[INFO ] 2025-03-25 18:16:45.704 [[main]<easticsearch] Nach der Suche - Schluss

Zeitpunkt (PIT)

(STÖHNT)
"message" => "{\"sensor_id\": \"sensor_123\", \"Temperatur\": 25.5, \"humidity\": 60, \"Status\": \"aktiv\"}",
"topisch" => "test/search",
"client_id" => "Bevywise-fQ4kK3aI2uQ0",
"Zeitstempel" => 1742478868,
"Eigenschaften" => "{}",
"@version" => "1",
"@timestamp" => 2025-03-25T12:46:45.691895415Z
}

(STÖHNT)
"message" => "{\"sensor_id\": \"sensor123\", \"Temperatur\": 27, \"humidity\": 712, \"Status\": \"aktiv\"}",
"topisch" => "test/monitierend",
"client_id" => "Bevywise-qI8mQ7rI9uM3",
"Zeitstempel" => 1742817477,
"Eigenschaften" => "{}",
"@version" => "1",
"@timestamp" => 2025-03-25T12:46:45.693859511Z
}

(STÖHNT)
"message" => "{\"sensor_id\": \"sensor123\", \"Temperatur\": 27, \"humidity\": 712, \"Status\": \"aktiv\"}",
"topisch" => "test/monitierend",
"client_id" => "Bevywise-eP3cT2xW0vA4",
"Zeitstempel" => 1742817662
"Eigenschaften" => "{}",
"@version" => "1",
"@timestamp" => 2025-03-25T12:46:45.693957511Z
}


Verwenden Sie Kibana, um die Daten zu visualisieren, sobald Elasticsearch sie vom CrystalMQ Elasticsearch Connector erhalten hat.


Schritt-1:


Öffnen Sie den Browser url wiehttp://<host_ip>:5601


Schritt 2:Das ist ein Tag


Bevor Sie Daten visualisieren, müssen Sie ein Indexmuster definieren, um auf Ihre MQTT-Logs.


  • Gehen Sie zu "Stack Management" → "Index Patterns".
  • Klicken Sie auf "Indexmuster erstellen".
  • Geben Sie Ihren Indexnamen ein (z.B. Offset_mqtt-*).
  • Wählen Sie das Zeitstempelfeld aus (ggf.@timestamp).
  • Klicken Sie auf "Index-Muster erstellen".

Schritt 3:Diagramm für MQTT-Nachrichten über die Zeit


Bevor Sie Daten visualisieren, müssen Sie ein Indexmuster definieren, um auf Ihre MQTT-Logs.


  • Gehen Sie."Visualize Library"→ Klicken Sie auf"Kreate Visualisierung".
  • Wählen"Line Chart".
  • Wählen Sie Ihr Indexmuster (offset_mqtt-*).
  • Konfigurieren:
    X-Achse: Wählen Sie @Timestamp und eingestelltes Intervall (z.B. 1m, 1h).
    Y-Achse: Wählen Sie Zählen, um MQTT-Nachrichten zu zählen.
  • Klicken Sie auf"Save"und benennen Sie Ihre Visualisierung.

Schritt-4: Das Diagramm sieht wie folgt aus:


Balkendiagramm :


Bar Chart


Pie-Chart:


Pie Chart


Tabelle :


Table View


Jetzt verwenden wir Elasticsearch mit TLS/SSL-Verschlüsselung, um eine Verbindung aufzubauen und Nachrichten über Data Connectors in CrystalMQ zu senden:



Schritt-1: Überprüfen Sie, ob alle Zertifikatspfade in der Elasticsearch-Konfiguration korrekt bereitgestellt werden. Wenn Änderungen vorgenommen werden, starten Sie Elasticsearch neu, um sie anzuwenden.



Schritt-2: Öffnen Sie die Kibana-Konfigurationsdatei und überprüfen Sie, ob alle Einstellungen korrekt für TLS/SSL konfiguriert sind.



Schritt-3 : Run Crystal MQ und verwenden Sie den Elasticsearch Connector in Data Connectors, um mit der SSL/TLS-Verschlüsselung zu Elasticsearch zu verbinden.



Schritt-4: Für die SSL/TLS-Verschlüsselung ist das CA-Root-Zertifikat erforderlich, um mit Elasticsearch zu verbinden. Der Client-Zertifikat und der Client-Schlüssel sind optional – nur das CA-Root-Zertifikat reicht aus, um die Verbindung herzustellen.



Elasticsearch


Schritt-5: Nach dem Hochladen aller Zertifikate, speichern Sie die Konfiguration und aktivieren Sie die Verbinder mit der Schaltfläche Aktivieren/Disable toggle.



Enable Certificates


Schritt-6: Der Elasticsearch Connector mit TLS/SSL-Verschlüsselung wurde nun erfolgreich in CrystalMQ verbunden.



Verwenden Sie Logstash, um die Protokolle aus dem Elasticsearch Connector anzuzeigen:



Konfigurieren Sie Logstash, um Daten von Elasticsearch zu erhalten:

  • Zurück zur Übersicht
    Erstellen einer Datei namensPressemitteilungenin der ZEITSCHRIFT
    Konfigurationsverzeichnis
  • Bearbeiten der Konfigurationsdatei
    ÖffnenPressemitteilungenund folgendes hinzufügen Konfiguration:

Eingang {
Elastische Forschung
hosts => ["https://your_elasticsearch_host:9200"]
Benutzer => "elastisch"
Passwort => "your_password"
Index => "your_index_pattern"
ssl => wahr
cacert => "/path/to/elasticsearch-ca.crt"
}
}


Ausgang {
stdout!
Codec => Rubin
}
}


Speichern Sie die Konfiguration und verwenden Sie den Befehl unten, um zu überprüfen, ob es korrekt festgelegt wurde:


sudo /usr/share/logstash/bin/logstash --path.settings /etc/logstash -f

path/to/elasticsearch_logstash.conf --config.test_and_exit


Verwenden Sie Logstash, um die Daten von elastischer Suche mit TLS/SSL zu erfassen:


Wenn Sie Logstash über einen Paketmanager installiert haben:


sudo /usr/share/logstash/bin/logstash -f

/etc/logstash/conf.d/elasticsearch_logstash.conf


Zum Beispiel: sudo /usr/share/logstash/bin/logstash -f
~/Documents/elasticsearch_logstash.conf


Konfigurieren Sie jetzt die Kibana mit elastischer Suche TLS/SSL:



Beziehen Sie sich auf die obigen Schritte, um kibana.yml für Easticsearch mit TLS/SSL-Verschlüsselung zu konfigurieren.



Öffnen Sie den Browser unter http://<host_ip>:5601 und visualisieren Sie die Daten in Kibana anhand der oben genannten Anweisungen, mit Optionen wie Bar-Charts, Pier-Charts und Datenansichten.



In der Elasticsearch Version 8.17.3 verwenden wir die Grundlizenz, so dass PKI und JWT nicht verwendet werden können, um die Verbindung herzustellen.

Zur Verwendung der JWT- und PKI-Authentifizierung ist ein Upgrade auf die Platinum-Lizenz erforderlich.


Verwendung der PKI-Authentifizierung in CrystalMQ:


Bei der PKI-Authentifizierung sind das Clientzertifikat und der Clientschlüssel erforderlich, während das CA-Root-Zertifikat optional ist.


Speichern Sie die Konfiguration und aktivieren Sie die Schaltfläche Connector-Toggle, um mit der PKI-Authentifizierung mit Elasticsearch zu verbinden.


Elasticsearch ist jetzt mit PKI-Authentifizierung konfiguriert.


Wenn wir die Daten verschlüsseln möchten, können wir SSL/TLS mit PKI-Authentifizierung verwenden.


Zur Verschlüsselung der Daten kann SSL/TLS zusammen mit der PKI-Authentifizierung verwendet werden.


In der PKI-Authentifizierung mit SSL/TLS, dem Clientzertifikat und dem Clientschlüssel werden benötigt, während für SSL-nur Verbindungen nur das CA-Root-Zertifikat benötigt wird, um mit Elasticsearch zu verbinden.




Save Configurations


Verwendung von JWT-Authentifizierung in CrystalMQ:



In der JWT (JSON Web Tokens) Authentifizierung wird der Zugriffstoken verwendet, um mit Elasticsearch zu verbinden.


Verwenden Sie einen Authentifizierungsanbieter, wie Keycloak oder Google Console, um den Zugriffsschlüssel zu erhalten.


Verwenden Sie nun den Zugriffsschlüssel mit dem folgenden Befehl, um den Zugriffstoken zu erhalten:


curl --cacert /etc/keycloak.crt -X POST
"https://localhost:8443/realms/myrealm/protocol/openid-
Verbindung/token["](https://localhost:8443/realms/myrealm/protocol/openid-
Pressemitteilungen (

-H "Content-Typ: Anwendung/x-www-form-urlencoded" \

-d "grant_type=client_credentials" \

-d "client_id=elastischsearch" \

-d "client_secret=ACCESS_KEY"


In diesem Befehl wird der Keycloak-Authentifizierungsanbieter verwendet, um den Authentifizierungstoken zu erhalten.



Nach dem Zugriff auf Token, verwenden Sie es im Elasticsearch Datenstecker, um mit der JWT-Authentifizierung mit Elasticsearch zu verbinden.



Authentication


Wenn wir die Daten verschlüsseln möchten, können wir SSL/TLS mit PKI-Authentifizierung verwenden.



Upload Client Certificate

Wenn Sie SSL/TLS mit JWT verwenden, laden Sie das CA-Root-Zertifikat einfach hoch, um mit Elasticsearch zu verbinden.


MQTT optimiert Netzwerk!

Aktionär MQTT Einblicke mit ELK

MQTT-Broker-Nachrichten in Echtzeit überwachen und analysieren, um IoT zu verbessern
leistung und potenzielle Probleme sofort erkennen.