Logstash-Java-Input-Plug-in für Apache PLC4X

In diesem Artikel beschreiben wir die Entwicklung, zusammen mit den Herausforderungen, des Logstash-Java-Input-Plug-ins für Apache PLC4X, das im Rahmen des Open-Source-Projekts Apache PLC4X in Absprache mit Elastic entwickelt wurde.

Architektonische Übersicht über das Logstash-Java-Input-Plug-in

Zunächst werfen wir einen Blick auf die grundlegende Architektur und Systemlandschaft und werden im Anschluss die Umsetzung des Plug-ins betrachten. In der nachfolgenden Abbildung ist eine Übersicht über das Einsatzgebiet des Plug-ins zu sehen.

Wir verwenden im Plug-in das Framework Apache PLC4X, um Daten von IIoT-Geräten (Industrial Internet of Things) mittels Pull-Prinzip abzufragen. Diese Daten werden an Logstash übergeben, welches diese anschließend über ein Output-Plug-in der Pipeline mittels Push-Prinzip weiterleitet. Hier haben wir beispielsweise ElasticSearch gewählt.
Neben der Abfrage von mehreren IIoT-Geräten in einer Pipeline ist es ebenfalls möglich, mehrere Pipelines zu definieren, um die Abfragen auf mehrere Pipelines zu verteilen. Dies ist z. B. bei verschiedenen Gerätetypen oder Standorten sinnvoll.

Die Entwicklung des Plug-ins und der zugehörige Build-Prozess (als Teil von Apache PLC4X)

Für die Entwicklung eines Logstash-Java-Input-Plug-ins gibt es von Elastic eine Anleitung, die hier gefunden werden kann.
Dort wird erklärt, dass als erstes der Quelltext von logstash heruntergeladen und gebaut werden muss. Anschließend wird im Projekt des Java-Logstash-Input-Plug-ins der Pfad zu logstash in der Datei gradle.properties eingetragen. Dieser Schritt ist notwendig, solange die entsprechende JAR-Datei von logstash noch nicht auf dem Maven Central Repository verfügbar ist. Aktuell ist noch unklar, wann dies der Fall ist – es existiert aber bereits ein Issue im entsprechenden GitHub-Projekt von Elastic.

Wird das Logstash-Java-Input-Plug-in anschließend gebaut, verwendet das Skript gradle.build die Datei rubyUtils.gradle von logstash. In diesen Utils wird JRuby heruntergeladen und anschließend verwendet. Für das Herunterladen und Entpacken von JRuby wird der Befehl commandLine ‚./mvnw‘, ‚clean‘, ‚install‘, ‚-Pdist‘, ‚-Pcomplete‘ verwendet. Gradle führt Maven an dieser Stelle über einen Maven-Wrapper aus. Dies ist im Rahmen des Apache-PLC4X-Projekts ungünstig. Denn PLC4X verwendet Maven selbst als Build-Tool. Daher ergäbe sich bei einfacher Einbindung von Gradle aus der Elastic-Anleitung folgender Build-Prozess:

Hier ist eine unnötige Verkettung von Build-Tools zu sehen, da Maven Gradle ausführt und dieses im Anschluss wiederum Maven aufruft. Daher haben wir beschlossen, den Gradle-Teil des Build-Prozesses ebenfalls in Maven umzusetzen, sodass sich folgender Ablauf ergibt:

Abschließend stellen wir noch die Umsetzung des Build-Prozesses mit Maven und den zugehörigen Plug-ins vor.

groovy-maven-plugin: In diesem Schritt wird die Version ohne SNAPSHOT berechnet und als künstliche Eigenschaft in Maven (current-full-version) gesetzt. JRuby benötigt die Version in diesem Format, da es ansonsten mit einer Fehlermeldung abbricht.
download-maven-plugin: Dieses Plug-in lädt und entpackt JRuby, das zum Erstellen des Artefakts benötigt wird.
maven-resources-plugin: Dieses Plug-in bearbeitet alle Ressourcen (z. B. ersetze ${current-full-version} durch die Version).
maven-shade-plugin: Erstellt eine Fat JAR aus den Klassen und Dateien des Plug-ins sowie allen Abhängigkeiten.
exec-maven-plugin: Mit diesem Plug-in werden JRuby und Gem aufgerufen, die anschließend das Artefakt logstash-input-plc4x-X.X.X.gem erstellen.
build-helper-maven-plugin: Mit diesem Plug-in wird das Artefakt in Maven hinzugefügt, sodass es ebenfalls in der install-Phase von Maven berücksichtigt wird.

Beispielkonfiguration des Plug-ins

Voraussetzung für die Verwendung des Plug-ins ist die Installation in Logstash. Anschließend kann das Plug-in wie folgt initialisiert und konfiguriert werden.

1## logstash pipeline config - input
2input {
3    ## use plc4x plugin (logstash-input-plc4x)
4    plc4x {
5        ## define sources
6        sources => {
7            source1 => "opcua:tcp://opcua-server:4840/"
8            source2 => "opcua:tcp://opcua-server1:4840/"
9            source3 => "..."
10        }
11        ## define jobs
12        jobs => {
13            job1 => {
14                # pull rate in milliseconds 
15                rate => 2
16                # sources queried by job1
17                sources => ["source1", ...]
18                # defined queries [logstash_internal_fieldname => "IIoT query"]
19                queries =>  {
20                    PreStage => "ns=2;i=3"
21                    MidStage => "ns=2;i=4"
22                    PostStage => "ns=2;i=5"
23                    Motor => "ns=2;i=6"
24                    ConvoyerBeltTimestamp => "ns=2;i=7"
25                    RobotArmTimestamp => "ns=2;i=8"
26                }
27            }
28            job2 => { .... }
29        }
30    }
31}
32 
33## logstash pipeline config - filter
34filter {
35    ...
36}
37 
38## logstash pipeline config - output
39output {
40    ...
41}

Im Abschnitt input wird das Plug-in durch plc4x initialisiert. Innerhalb des plc4x-Abschnitts können sources und jobs definiert werden. Sources sind die Connection-Strings für die jeweiligen PLCs (hier beispielsweise OPC UA Server).
Im jobs-Abschnitt können einzelne jobs definiert werden. Diese enthalten queries (Abfragen) für bestimmte sources. Die Abfragen sind in der Form X => „Y“ angegeben. X ist der interne Feldname für Logstash. Y ist die Abfrage, die zum PLC gesendet wird.
Für jeden Job wird eine Rate angegeben, die die Zeitspanne zwischen den einzelnen Abfragen definiert. Im Beispiel werden die fünf definierten Felder alle 300 Millisekunden von source1 abgefragt.

Quelltext des Java-Logstash-Plug-ins für PLC4X

Das Plug-in besteht aus einer Klasse, die wir für diesen Blogbeitrag in drei Abschnitte aufgeteilt haben.

Im ersten Teil des Plug-ins werden die sources eingelesen und anschließend dem Builder des Scrapers von PLC4X hinzugefügt. Der Scraper ist ein Teil von PLC4X, um Daten regelmäßig von beliebigen PLCs abzufragen.

1// parse sources
2for (String sourceName : sources.keySet()) {
3    Object o = sources.get(sourceName);
4    if(o instanceof String) {
5        String source = (String)o;
6        builder.addSource(sourceName, source);
7    } else {
8        logger.severe("URL of source " + sourceName + "has the wrong typ!");
9    }
10}

Der zweite Abschnitt des Quellcodes ist für das Einlesen und Erstellen der jobs zuständig.
In der äußeren for-Schleife wird über alle definierten jobs iteriert. Für jeden einzelnen Job werden anschließend die Rate, die sources als auch die Abfragen eingelesen und dem jobBuilder (ebenfalls Teil des Scrapers) von PLC4X zugewiesen.

1// parse jobs
2for (String jobName : jobs.keySet()) {
3    Object o = jobs.get(jobName);
4    if (o instanceof  Map) {
5        Map job = (Map<String, Object>) o;
6        JobConfigurationTriggeredImplBuilder jobBuilder = builder.job(
7            jobName, String.format("(SCHEDULED,%s)", job.get("rate")));
8        for (String source : ((List<String>) job.get("sources"))) {
9            jobBuilder.source(source);
10        }
11        Map<String, Object> queries = (Map<String, Object>) job.get("queries");
12        for (String queryName : queries.keySet()) {
13 
14            String fieldAlias = queryName;
15            String fieldAddress = (String) queries.get(queryName);
16            jobBuilder.field(fieldAlias, fieldAddress);
17        }
18        jobBuilder.build();
19    } else {
20        logger.severe("Jobs of wrong Type!");
21    }
22}

Nachdem nun alle sources und jobs eingelesen sind, können wir den Scraper ausführen. Bei der Initialisierung des Scrapers übergeben wir eine Lambda-Funktion, die nach erfolgreicher Abfrage der Felder aufgerufen wird. In dieser Funktion werden die erhaltenen Werte geloggt und die Ergebnisse anschließend vom consumer akzeptiert (der consumer ist Teil des Logstash-API und nimmt die Daten entgegen).

1// start scraper
2ScraperConfigurationTriggeredImpl scraperConfig = builder.build();
3try {
4    plcDriverManager = new PooledPlcDriverManager();
5    triggerCollector = new TriggerCollectorImpl(plcDriverManager);
6    scraper = new TriggeredScraperImpl(scraperConfig, (jobName, sourceName, results) -> {
7        HashMap<String, Object> resultMap = new HashMap<String, Object>();
8        resultMap.put("jobName", jobName);
9        resultMap.put("sourceName", sourceName);
10        resultMap.put("values", results);
11 
12        //TODO: add guard for debug mode, so only in debug mode its run
13        for (Map.Entry<String, Object> result : results.entrySet()) {
14            // Get field-name and -value from the results.
15            String fieldName = result.getKey();
16            Object fieldValue = result.getValue();
17            logger.finest("fieldName: " + fieldName);
18            logger.finest("fieldValue: " + fieldValue);
19        }
20        consumer.accept(resultMap);
21    }, triggerCollector);
22    scraper.start();
23    triggerCollector.start();
24} catch (ScraperException e) {
25    logger.severe("Error starting the scraper: "+ e);
26}

Fazit und Lessons Learned

• Benennung von Dateien und Klassen
  Die Logstash-Java-Input-Plug-ins sind bei der Benennung der Dateien und Konfigurationen sehr sensitiv. Die Namen der Gemspec-, JAR- sowie Gem-Dateien müssen alle derselben Benennung folgen (z. B. logstash-input-plc4x). Ebenfalls müssen der Klassenname und einige Konfigurationen (beispielsweise Inhalte der Ruby-Konfiguration, also der .rb-Dateien) bezüglich der Benennung übereinstimmen. Zu Beginn der Entwicklung hatten wir uns überlegt, die aritfact-id des Projekts zu verwenden. Dabei hatten wir das Problem, dass wir diese nicht überall durch das maven-resource-plugin ersetzen lassen konnten. Dadurch stimmten die Benennungen teilweise nicht überein und es kam zu Problemen bei Installation und Ausführung des Plug-ins. Nachdem wir uns für eine statische Benennung des Plug-ins entschieden und alle relevanten Stellen identifiziert hatten, ließ sich das Plug-in in Logstash installieren und ausführen. Diese Lösung hat auch den Vorteil, dass die artifact-id plc4j-logstash-plugin mit dem Namensschema von PLC4X und der Plug-in Name logstash-input-plc4x mit dem Namensschema für Plug-ins in Logstash übereinstimmt.
• Installationsdauer des Plug-ins
  Die Installation des Plug-ins in Logstash benötigt meist ~2 Minuten. Dies ist im Fehlerfall (wie z. B. bei der Problematik mit der Benennung) sehr unglücklich, da so zwischen Änderung und Test dieser einige Minuten vergehen.
• Build-Tools: Maven, Gradle, JRuby, …
  Ein weiteres Learning dieses Projekts besteht darin, dass wir in zukünftigen Projekten die Verwendung mehrerer Build-Tools, genau wie hier, hinterfragen werden. Ansonsten entstehen gegebenenfalls ungünstige Verkettungen von Build-Tools wie oben beschrieben (Maven und Gradle). An dieser Stelle möchten wir uns auch bei Christofer Dutz bedanken, der uns bei dieser Problematik unterstützt hat.

Der Quellcode des gesamten Plug-ins ist hier im PLC4X-Projekt zu finden.