This is an automated email from the ASF dual-hosted git repository.
glcj pushed a commit to branch feature/merlot
in repository https://gitbox.apache.org/repos/asf/plc4x-extras.git
The following commit(s) were added to refs/heads/feature/merlot by this push:
new 50319aa Implement and document Loki appender for Karaf event bus
(#649)
50319aa is described below
commit 50319aa1ae2531b303f04131687fc2a56a642f73
Author: Lerb123 <[email protected]>
AuthorDate: Wed May 20 11:50:05 2026 -0400
Implement and document Loki appender for Karaf event bus (#649)
* feat: Implement Loki appender for Karaf event bus
* fix: Implement Loki appender for Karaf event bus, added comments.
* feat: Functional test for Loki Appender using WireMock
* feat: CI Pipeline for MerlotLokiAppender
* Correction to the Maven configuration to prevent tests from being
skipped; add .mvn
* Correction to Maven configuration: local cache is disabled
* feat:Kafka collector for topics on the Phoebus alarm server
---
plc4j/tools/merlot/.mvn/maven.config | 0
plc4j/tools/merlot/anahata.md | 3 -
plc4j/tools/merlot/merlot.iml | 15 --
.../org.apache.plc4x.merlot.archiver.iml | 8 +
.../org.apache.plc4x.merlot.archiver/pom.xml | 99 ++++++++--
.../cfg/org.apache.plc4x.merlot.loki.appender.cfg | 39 ++++
.../merlot/archiver/impl/MerlotLokiAppender.java | 195 ++++++++++++++++++++
.../OSGI-INF/blueprint/archiver-service.xml | 35 +++-
.../java/runner/MerlotLokiAppenderRunnerTest.java | 16 ++
.../stepdefinition/LokiAppenderDefinition.java | 71 +++++++
.../src/test/resources/features/loki.feature | 6 +
.../src/main/feature/feature.xml | 10 +-
.../merlot/org.apache.plc4x.merlot.kafka/pom.xml | 2 +-
....apache.plc4x.merlot.kafka.collector-alarm.cfg} | 6 +-
.../core/MerlotKafkaManagedServiceFactory.java | 50 +++--
.../impl/MerlotKafkaDecanterCollectorImpl.java | 205 ++++++++++++++++-----
.../resources/OSGI-INF/blueprint/kafka-service.xml | 6 +-
plc4j/tools/merlot/org.apache.plc4x.merlot/pom.xml | 3 +-
plc4j/tools/merlot/pom.xml | 1 +
19 files changed, 661 insertions(+), 109 deletions(-)
diff --git a/plc4j/tools/merlot/.mvn/maven.config
b/plc4j/tools/merlot/.mvn/maven.config
new file mode 100644
index 0000000..e69de29
diff --git a/plc4j/tools/merlot/anahata.md b/plc4j/tools/merlot/anahata.md
deleted file mode 100644
index 7a959db..0000000
--- a/plc4j/tools/merlot/anahata.md
+++ /dev/null
@@ -1,3 +0,0 @@
-# Anahata Project Notes
-
-This file is for Anahata AI Assistant's notes regarding the 'merlot' project.
diff --git a/plc4j/tools/merlot/merlot.iml b/plc4j/tools/merlot/merlot.iml
deleted file mode 100644
index 452f210..0000000
--- a/plc4j/tools/merlot/merlot.iml
+++ /dev/null
@@ -1,15 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<module version="4">
- <component name="FacetManager">
- <facet type="web" name="Web">
- <configuration>
- <descriptors>
- <deploymentDescriptor name="web.xml"
url="file://$MODULE_DIR$/org.apache.plc4x.merlot.ui/src/main/webapp/WEB-INF/web.xml"
/>
- </descriptors>
- <webroots>
- <root
url="file://$MODULE_DIR$/org.apache.plc4x.merlot.ui/src/main/webapp"
relative="/" />
- </webroots>
- </configuration>
- </facet>
- </component>
-</module>
\ No newline at end of file
diff --git
a/plc4j/tools/merlot/org.apache.plc4x.merlot.archiver/org.apache.plc4x.merlot.archiver.iml
b/plc4j/tools/merlot/org.apache.plc4x.merlot.archiver/org.apache.plc4x.merlot.archiver.iml
new file mode 100644
index 0000000..359c469
--- /dev/null
+++
b/plc4j/tools/merlot/org.apache.plc4x.merlot.archiver/org.apache.plc4x.merlot.archiver.iml
@@ -0,0 +1,8 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<module version="4">
+ <component name="AdditionalModuleElements">
+ <content url="file://$MODULE_DIR$" dumb="true">
+ <sourceFolder url="file://$MODULE_DIR$/src/test/src/test/resources"
type="java-test-resource" />
+ </content>
+ </component>
+</module>
\ No newline at end of file
diff --git a/plc4j/tools/merlot/org.apache.plc4x.merlot.archiver/pom.xml
b/plc4j/tools/merlot/org.apache.plc4x.merlot.archiver/pom.xml
index 878cce8..5edea7a 100644
--- a/plc4j/tools/merlot/org.apache.plc4x.merlot.archiver/pom.xml
+++ b/plc4j/tools/merlot/org.apache.plc4x.merlot.archiver/pom.xml
@@ -21,10 +21,10 @@
<modelVersion>4.0.0</modelVersion>
<parent>
- <artifactId>merlot</artifactId>
- <groupId>org.apache.plc4x</groupId>
- <version>0.13.0-SNAPSHOT</version>
- </parent>
+ <artifactId>merlot</artifactId>
+ <groupId>org.apache.plc4x</groupId>
+ <version>0.13.0-SNAPSHOT</version>
+ </parent>
<groupId>org.apache.plc4x.merlot.archiver</groupId>
<artifactId>org.apache.plc4x.merlot.archiver</artifactId>
@@ -34,6 +34,19 @@
<name>PLC4J: Merlot :: archiver :: Historical archive </name>
<description>decanter OSGi blueprint bundle project.</description>
+ <properties>
+
+ <lombok.version>1.18.46</lombok.version>
+ <junit.version>4.13.2</junit.version>
+ <junit-jupiter.version>6.0.3</junit-jupiter.version>
+ <mockito-junit-jupiter.version>5.23.0</mockito-junit-jupiter.version>
+ <wiremock.version>3.13.2</wiremock.version>
+ <wiremock-junit5.version>1.1.0</wiremock-junit5.version>
+ <junit-platform.version>6.0.3</junit-platform.version>
+ <cucumber-java.version>7.34.3</cucumber-java.version>
+
<cucumber-junit-platform-engine.version>7.34.3</cucumber-junit-platform-engine.version>
+ </properties>
+
<build>
<plugins>
<plugin>
@@ -62,8 +75,8 @@
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<configuration>
- <source>17</source>
- <target>17</target>
+ <source>21</source>
+ <target>21</target>
<maxmem>256M</maxmem>
</configuration>
</plugin>
@@ -107,10 +120,10 @@
<file>src/main/cfg/org.apache.plc4x.merlot.collector.log-alarm.cfg</file>
<type>cfg5</type>
</artifact>
-<!-- <artifact>
-
<file>src/main/cfg/org.apache.plc4x.merlot.pvcomplex.cfg</file>
+ <artifact>
+
<file>src/main/cfg/org.apache.plc4x.merlot.loki.appender.cfg</file>
<type>cfg7</type>
- </artifact>
-->
+ </artifact>
</artifacts>
</configuration>
</execution>
@@ -227,19 +240,19 @@
<dependency>
<groupId>org.junit.jupiter</groupId>
<artifactId>junit-jupiter-api</artifactId>
- <version>5.10.3</version>
+ <version>${junit-jupiter.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.junit.jupiter</groupId>
<artifactId>junit-jupiter-params</artifactId>
- <version>5.10.3</version>
+ <version>${junit-jupiter.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.junit.jupiter</groupId>
<artifactId>junit-jupiter-engine</artifactId>
- <version>5.10.3</version>
+ <version>${junit-jupiter.version}</version>
<scope>test</scope>
</dependency>
<dependency>
@@ -266,15 +279,73 @@
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
- <version>4.13.2</version>
+ <version>${junit.version}</version>
<scope>test</scope>
<type>jar</type>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
- <version>1.18.42</version>
+ <version>${lombok.version}</version>
<type>jar</type>
</dependency>
+ <dependency>
+ <groupId>org.apache.karaf.decanter</groupId>
+ <artifactId>org.apache.karaf.decanter.api</artifactId>
+ <version>2.10.0</version>
+ <type>bundle</type>
+ </dependency>
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-junit-jupiter</artifactId>
+ <version>${mockito-junit-jupiter.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.wiremock</groupId>
+ <artifactId>wiremock</artifactId>
+ <version>${wiremock.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>de.mkammerer.wiremock-junit5</groupId>
+ <artifactId>wiremock-junit5</artifactId>
+ <version>${wiremock-junit5.version}</version>
+ <scope>test</scope>
+ <type>jar</type>
+ </dependency>
+
+ <dependency>
+ <groupId>org.junit.platform</groupId>
+ <artifactId>junit-platform-launcher</artifactId>
+ <version>${junit-platform.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.junit.platform</groupId>
+ <artifactId>junit-platform-suite-api</artifactId>
+ <version>${junit-platform.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>io.cucumber</groupId>
+ <artifactId>cucumber-java</artifactId>
+ <version>${cucumber-java.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>io.cucumber</groupId>
+ <artifactId>cucumber-junit-platform-engine</artifactId>
+ <version>${cucumber-junit-platform-engine.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.junit.platform</groupId>
+ <artifactId>junit-platform-suite-engine</artifactId>
+ <version>${junit-platform.version}</version> <scope>test</scope>
+ </dependency>
</dependencies>
</project>
diff --git
a/plc4j/tools/merlot/org.apache.plc4x.merlot.archiver/src/main/cfg/org.apache.plc4x.merlot.loki.appender.cfg
b/plc4j/tools/merlot/org.apache.plc4x.merlot.archiver/src/main/cfg/org.apache.plc4x.merlot.loki.appender.cfg
new file mode 100644
index 0000000..2ca9a72
--- /dev/null
+++
b/plc4j/tools/merlot/org.apache.plc4x.merlot.archiver/src/main/cfg/org.apache.plc4x.merlot.loki.appender.cfg
@@ -0,0 +1,39 @@
+################################################################################
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+################################################################################
+#Server Parameters
+#loki.url: Loki server IP address and port
+#loki.username: User configured on the Loki server
+#loki.password: Password configured on the Loki server
+
+################################################################################
+#loki.url=http://127.0.0.1:3100/loki/api/v1/push
+
+# Authentication (optional)
+#loki.username=loki
+#loki.password=loki
+
+#Allowed topics (separated by commas). General topics: merlot/*, phoebus/*,
iotdb/*, decanter/*
+#loki.topics=decanter/collect/log,decanter/collect/metric,phoebus/data/test,merlot/archiver/test
+
+# Static labels (always added)
+#loki.label.job=merlot
+#loki.label.env=dev
+#loki.label.app=plc4x-archiver
+#loki.label.logLevel=INFO
+#loki.label.component=archiver
diff --git
a/plc4j/tools/merlot/org.apache.plc4x.merlot.archiver/src/main/java/org/apache/plc4x/merlot/archiver/impl/MerlotLokiAppender.java
b/plc4j/tools/merlot/org.apache.plc4x.merlot.archiver/src/main/java/org/apache/plc4x/merlot/archiver/impl/MerlotLokiAppender.java
new file mode 100644
index 0000000..4ccf410
--- /dev/null
+++
b/plc4j/tools/merlot/org.apache.plc4x.merlot.archiver/src/main/java/org/apache/plc4x/merlot/archiver/impl/MerlotLokiAppender.java
@@ -0,0 +1,195 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.plc4x.merlot.archiver.impl;
+
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.time.Duration;
+import java.util.*;
+import java.util.concurrent.CompletableFuture;
+import java.nio.charset.StandardCharsets;
+
+import org.osgi.service.cm.ConfigurationException;
+import org.osgi.service.cm.ManagedService;
+import org.osgi.service.event.Event;
+import org.osgi.service.event.EventHandler;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class MerlotLokiAppender implements EventHandler, ManagedService {
+
+ private static final Logger LOGGER =
LoggerFactory.getLogger(MerlotLokiAppender.class);
+
+ private String url;
+ private String username;
+ private String password;
+ private final Set<String> allowedTopics = new HashSet<>();
+ private final Map<String, String> labels = new HashMap<>();
+
+ private HttpClient httpClient;
+
+ public MerlotLokiAppender() {
+ }
+
+ public void init() {
+ LOGGER.info("Starting the Merlot-Loki appender");
+ httpClient = HttpClient.newBuilder()
+ .connectTimeout(Duration.ofSeconds(5))
+ .build();
+ }
+
+ public void destroy() {
+ LOGGER.info("Deleting the Merlot-Loki appender module");
+ httpClient.close();
+ httpClient.shutdownNow();
+ }
+
+ //Inject HttpClient for use with WireMock
+ public void setHttpClient(HttpClient httpClient) {
+ this.httpClient = httpClient;
+ }
+
+ ////Inject Url for use with WireMock
+ public void setUrl(String url) {
+ this.url = url;
+ }
+
+ @Override
+ public void handleEvent(Event event) {
+ String topic = event.getTopic();
+
+ //The topic must match those listed in the “loki.topics” property of
the corresponding cfg file
+ if (!allowedTopics.isEmpty() && !allowedTopics.contains(topic)) {
+ return;
+ }
+ //For events without loki.label.<name> tags, general tags are assigned
as the primary tags (cfg file)
+ Map<String, String> eventLabels = new HashMap<>(this.labels);
+
+ StringBuilder msgBuilder = new StringBuilder();
+ msgBuilder.append("[").append(topic).append("] ");
+
+ for (String prop : event.getPropertyNames()) {
+ Object value = event.getProperty(prop);
+
+ //Static tags will be sent as message content to Loki.
+ if (prop.startsWith("loki.label.")) {
+ String labelName = prop.substring("loki.label.".length());
+ eventLabels.put(labelName, value.toString());
+ } else if (!prop.equals("event.topics") &&
!prop.equals("service.id") && !prop.equals("subject")) {
+ msgBuilder.append(prop).append("=").append(value).append(" ");
+ }
+ }
+
+ sendLokiServer(msgBuilder.toString().trim(), eventLabels);
+ }
+
+ public void sendLokiServer(String message, Map<String, String>
dynamicLabels) {
+ if (this.url == null || this.url.isEmpty()) {
+ return;
+ }
+
+ try {
+ long timeNano = System.currentTimeMillis() * 1_000_000;
+
+ //Label classifier
+ StringBuilder labelsJson = new StringBuilder("{");
+
+ StringJoiner sj = new StringJoiner(",");
+
+ dynamicLabels.forEach((k, v) -> {
+ sj.add("\"" + escapeJson(k) + "\":\"" + escapeJson(v) + "\"");
+ });
+
+ labelsJson.append(sj.toString()).append("}");
+
+ //Payload to be sent to the Loki server, with the tags from the
event as well as those saved in the cfg file.
+ String payload = "{\"streams\": [{\"stream\": " + labelsJson + ",
\"values\": [ [\"" + timeNano + "\", \"" + escapeJson(message) + "\"] ]}]}";
+
+ //The message must be sent as JSON
+ HttpRequest.Builder requestBuilder = HttpRequest.newBuilder()
+ .uri(URI.create(this.url))
+ .header("Content-Type", "application/json")
+ .POST(HttpRequest.BodyPublishers.ofString(payload,
StandardCharsets.UTF_8));
+
+ //If the server requires authentication, it must be encrypted
+ if (this.username != null && this.password != null) {
+ String auth = this.username + ":" + this.password;
+ String encoded =
Base64.getEncoder().encodeToString(auth.getBytes());
+ requestBuilder.header("Authorization", "Basic " + encoded);
+ }
+
+ //Send the request asynchronously to avoid blocking the Karaf
internal bus
+ httpClient.sendAsync(requestBuilder.build(),
HttpResponse.BodyHandlers.ofString())
+ .thenAccept(response -> {
+ if (response.statusCode() != 204 &&
response.statusCode() != 200) {
+ LOGGER.error("Loki rejected (Code {} ): {}",
response.statusCode(), response.body());
+ }
+ })
+ .exceptionally(e -> {
+ LOGGER.error("Loki Connection Error: {}",
e.getMessage());
+ return null;
+ });
+
+ } catch (Exception e) {
+ LOGGER.error("Critical Error in MerlotLokiAppender: {}",
e.getMessage());
+ }
+ }
+
+ private String escapeJson(String s) {
+ if (s == null) {
+ return "";
+ }
+ //Escape character correction
+ return s.replace("\\", "\\\\")
+ .replace("\"", "\\\"")
+ .replace("\n", "\\n")
+ .replace("\r", "")
+ .replace("\t", "\\t");
+ }
+
+ @Override
+ public void updated(Dictionary<String, ?> properties) throws
ConfigurationException {
+ if (properties == null || properties.isEmpty()) {
+ return;
+ }
+
+ //Connection parameters
+ this.url = (String) properties.get("loki.url");
+ this.username = (String) properties.get("loki.username");
+ this.password = (String) properties.get("loki.password");
+
+ // Topic configuration
+ this.allowedTopics.clear();
+ String strTopics = (String) properties.get("loki.topics");
+ if (strTopics != null) {
+
Arrays.stream(strTopics.split(",")).map(String::trim).forEach(this.allowedTopics::add);
+ }
+
+ // Configuration of dynamic, static labels
+ this.labels.clear();
+ Enumeration<String> keys = properties.keys();
+ while (keys.hasMoreElements()) {
+ String key = keys.nextElement();
+ if (key.startsWith("loki.label.")) {
+ String labelName = key.substring("loki.label.".length());
+ this.labels.put(labelName, properties.get(key).toString());
+ }
+ }
+ }
+}
diff --git
a/plc4j/tools/merlot/org.apache.plc4x.merlot.archiver/src/main/resources/OSGI-INF/blueprint/archiver-service.xml
b/plc4j/tools/merlot/org.apache.plc4x.merlot.archiver/src/main/resources/OSGI-INF/blueprint/archiver-service.xml
index d25c314..4e1e90a 100644
---
a/plc4j/tools/merlot/org.apache.plc4x.merlot.archiver/src/main/resources/OSGI-INF/blueprint/archiver-service.xml
+++
b/plc4j/tools/merlot/org.apache.plc4x.merlot.archiver/src/main/resources/OSGI-INF/blueprint/archiver-service.xml
@@ -30,7 +30,11 @@
<reference
id="refEventAdminService"
interface="org.osgi.service.event.EventAdmin" availability="mandatory"
timeout="1200"/>
-
+ <reference id="refMarshaller"
+ interface="org.apache.karaf.decanter.api.marshaller.Marshaller"
+ timeout="20000"
+ availability="mandatory">
+ </reference>
<bean id="MerlotGPClientBean"
class="org.apache.plc4x.merlot.archiver.impl.MerlotGPClientImpl"
init-method="init"
@@ -60,7 +64,17 @@
<argument ref="blueprintBundleContext" />
<argument ref="MerlotGPClientBean" />
<argument ref="refEventAdminService" />
- </bean>
+ </bean>
+
+ <!--Loki Appender-->
+ <bean id="MerlotLokiAppenderBean"
+ class="org.apache.plc4x.merlot.archiver.impl.MerlotLokiAppender"
+ init-method="init"
+ destroy-method="destroy"
+ scope="singleton"
+ activation="eager">
+
+ </bean>
<bean id="MerlotPvHtcCollectorImplBean"
class="org.apache.plc4x.merlot.archiver.impl.MerlotPvHtcCollectorImpl"
@@ -81,7 +95,7 @@
<argument ref="MerlotGPClientBean" />
</bean>
-<!-- <bean id="MerlotPvStructureRtCollectorBean"
+ <!-- <bean id="MerlotPvStructureRtCollectorBean"
class="org.apache.plc4x.merlot.archiver.impl.MerlotPVStructureRtCollector"
init-method="init"
destroy-method="destroy"
@@ -189,13 +203,26 @@
</service-properties>
</service>
+ <service ref="MerlotLokiAppenderBean" auto-export="interfaces">
+ <service-properties>
+ <entry key="service.pid"
value="org.apache.plc4x.merlot.loki.appender"/>
+ <entry key="event.topics">
+ <array value-type="java.lang.String">
+ <value>merlot/*</value>
+ <value>phoebus/*</value>
+ <value>iotdb/*</value>
+ <value>decanter/*</value>
+ </array>
+ </entry>
+ </service-properties>
+ </service>
<service ref="MerlotPvRtCollectorImplBean" auto-export="interfaces">
<service-properties>
<entry key="service.pid" value="org.apache.plc4x.merlot.pvrt"/>
</service-properties>
</service>
-<!-- <service ref="MerlotPvStructureRtCollectorBean"
auto-export="interfaces">
+ <!-- <service ref="MerlotPvStructureRtCollectorBean"
auto-export="interfaces">
<service-properties>
<entry key="service.pid"
value="org.apache.plc4x.merlot.pvcomplex"/>
</service-properties>
diff --git
a/plc4j/tools/merlot/org.apache.plc4x.merlot.archiver/src/test/java/runner/MerlotLokiAppenderRunnerTest.java
b/plc4j/tools/merlot/org.apache.plc4x.merlot.archiver/src/test/java/runner/MerlotLokiAppenderRunnerTest.java
new file mode 100644
index 0000000..6c610e5
--- /dev/null
+++
b/plc4j/tools/merlot/org.apache.plc4x.merlot.archiver/src/test/java/runner/MerlotLokiAppenderRunnerTest.java
@@ -0,0 +1,16 @@
+package runner;
+
+import org.junit.platform.suite.api.ConfigurationParameter;
+import org.junit.platform.suite.api.IncludeEngines;
+import org.junit.platform.suite.api.SelectClasspathResource;
+import org.junit.platform.suite.api.Suite;
+
+@Suite
+@IncludeEngines("cucumber")
+@SelectClasspathResource("features")
+@ConfigurationParameter(key = "cucumber.glue", value = "stepdefinition")
+@ConfigurationParameter(key = "cucumber.snippet-type", value = "camelcase")
+public class MerlotLokiAppenderRunnerTest {
+}
+
+
diff --git
a/plc4j/tools/merlot/org.apache.plc4x.merlot.archiver/src/test/java/stepdefinition/LokiAppenderDefinition.java
b/plc4j/tools/merlot/org.apache.plc4x.merlot.archiver/src/test/java/stepdefinition/LokiAppenderDefinition.java
new file mode 100644
index 0000000..5692b1c
--- /dev/null
+++
b/plc4j/tools/merlot/org.apache.plc4x.merlot.archiver/src/test/java/stepdefinition/LokiAppenderDefinition.java
@@ -0,0 +1,71 @@
+package stepdefinition;
+
+import com.github.tomakehurst.wiremock.WireMockServer;
+import com.github.tomakehurst.wiremock.client.WireMock;
+import io.cucumber.java.en.Given;
+import io.cucumber.java.en.Then;
+import io.cucumber.java.en.When;
+import org.apache.plc4x.merlot.archiver.impl.MerlotLokiAppender;
+import org.apache.plc4x.merlot.archiver.impl.MerlotPvRtCollectorImpl;
+import org.slf4j.LoggerFactory;
+
+import java.net.http.HttpClient;
+import java.util.Map;
+
+import static com.github.tomakehurst.wiremock.client.WireMock.*;
+import static com.github.tomakehurst.wiremock.client.WireMock.containing;
+import static com.github.tomakehurst.wiremock.client.WireMock.matching;
+import static com.github.tomakehurst.wiremock.client.WireMock.postRequestedFor;
+
+public class LokiAppenderDefinition {
+ private static final org.slf4j.Logger LOGGER =
LoggerFactory.getLogger(LokiAppenderDefinition.class);
+
+ private final MerlotLokiAppender mLoki = new MerlotLokiAppender();
+ private final WireMockServer wireMockServer = new WireMockServer(3100);
+ private final HttpClient httpClient = HttpClient.newHttpClient();
+
+ private Map<String, String> labels;
+
+
+ @Given("Labels supplied for the message {string}, {string}, {string},
{string}")
+ public void labelsSuppliedForTheMessage(String label1, String label2,
String label3, String label4) {
+ mLoki.setHttpClient(httpClient);
+ LOGGER.info("Labels: {} {} {} {}", label1, label2, label3, label4);
+ //Labels
+ labels = Map.of(label1, label2, label3, label4);
+
+ }
+ @Given("Url Loki Server {string}")
+ public void url(String url) {
+ mLoki.setUrl(url);
+ wireMockServer.start();
+ WireMock.configureFor("localhost", 3100);
+ }
+
+ @When("The server returned an HTTP {int} response")
+ public void theServerReturnedAnHTTPResponse(Integer code) {
+ //Stub for POST requests to the URL /loki/api/v1/push
+ wireMockServer.stubFor(WireMock.post(urlEqualTo("/loki/api/v1/push"))
+ .willReturn(aResponse()
+ .withStatus(code)
+ .withBody("Successfully pushed")));
+ //Send data to the simulated “Loki Server”
+ mLoki.sendLokiServer("Test Message", labels);
+
+
+ }
+ @Then("The instance displays “Log sent”")
+ public void theInstanceDisplaysLogSent() {
+ // Verify that at least one HTTP request was received during the test
+ verify(postRequestedFor(urlEqualTo("/loki/api/v1/push"))
+ .withHeader("Content-Type", matching("application/json.*"))
+ .withRequestBody(containing("Test Message"))
+ .withRequestBody(containing("\"job\":\"test\"")));
+
+
+ //Stop server
+ wireMockServer.stop();
+ LOGGER.info("Successfully pushed");
+ }
+
+}
diff --git
a/plc4j/tools/merlot/org.apache.plc4x.merlot.archiver/src/test/resources/features/loki.feature
b/plc4j/tools/merlot/org.apache.plc4x.merlot.archiver/src/test/resources/features/loki.feature
new file mode 100644
index 0000000..4f828ab
--- /dev/null
+++
b/plc4j/tools/merlot/org.apache.plc4x.merlot.archiver/src/test/resources/features/loki.feature
@@ -0,0 +1,6 @@
+Feature: Sending data to the Loki server
+ Scenario: Build Message
+ Given Labels supplied for the message "job", "test", "env", "dev"
+ And Url Loki Server "http://localhost:3100/loki/api/v1/push";
+ When The server returned an HTTP 204 response
+ Then The instance displays “Log sent”
\ No newline at end of file
diff --git
a/plc4j/tools/merlot/org.apache.plc4x.merlot.features/src/main/feature/feature.xml
b/plc4j/tools/merlot/org.apache.plc4x.merlot.features/src/main/feature/feature.xml
index 987c791..d249b00 100755
---
a/plc4j/tools/merlot/org.apache.plc4x.merlot.features/src/main/feature/feature.xml
+++
b/plc4j/tools/merlot/org.apache.plc4x.merlot.features/src/main/feature/feature.xml
@@ -36,6 +36,7 @@
<feature prerequisite="true">pax-web-karaf</feature>
<feature>plc4x-epics-gpclient</feature>
+ <feature>plc4x-kafka</feature>
<bundle>wrap:mvn:org.osgi/org.osgi.service.device/1.1.0/$Export-Package=*;version="1.1.0",!*</bundle>
@@ -257,7 +258,7 @@
<configfile
finalname="/etc/org.apache.plc4x.merlot.decanter-iotdb.cfg">mvn:org.apache.plc4x.merlot.archiver/org.apache.plc4x.merlot.archiver/${project.version}/cfg</configfile>
<configfile
finalname="/etc/org.apache.plc4x.merlot.pvhtc-iotdb.cfg">mvn:org.apache.plc4x.merlot.archiver/org.apache.plc4x.merlot.archiver/${project.version}/cfg4</configfile>
<configfile
finalname="/etc/org.apache.plc4x.merlot.pvrt-grafana.cfg">mvn:org.apache.plc4x.merlot.archiver/org.apache.plc4x.merlot.archiver/${project.version}/cfg6</configfile>
- <!--<configfile
finalname="/etc/org.apache.plc4x.merlot.pvcomplex.cfg">mvn:org.apache.plc4x.merlot.archiver/org.apache.plc4x.merlot.archiver/${project.version}/cfg7</configfile>-->
+ <configfile
finalname="/etc/org.apache.plc4x.merlot.loki.appender.cfg">mvn:org.apache.plc4x.merlot.archiver/org.apache.plc4x.merlot.archiver/${project.version}/cfg7</configfile>
<configfile
finalname="/etc/org.apache.plc4x.merlot.iotdb.cfg">mvn:org.apache.plc4x.merlot.archiver/org.apache.plc4x.merlot.archiver/${project.version}/cfg3</configfile>
<configfile
finalname="/etc/org.apache.plc4x.merlot.collector.log-alarm.cfg">mvn:org.apache.plc4x.merlot.archiver/org.apache.plc4x.merlot.archiver/${project.version}/cfg3</configfile>
</feature>
@@ -728,6 +729,13 @@
<bundle
start-level="30">mvn:org.apache.activemq/artemis-features/2.30.0</bundle>
</feature>
+ <feature name='plc4x-kafka' description='${project.name}'
version='${project.version}'>
+ <bundle
start-level="10">wrap:mvn:org.apache.kafka/kafka-clients/${kafka.version}/$Bundle-SymbolicName=kafka-client&Bundle-Version=${kafka.version}&Bundle-ManifestVersion="2"&Export-Package=*;version="${kafka.version}",!*</bundle>
+ <bundle
start-level="10">wrap:mvn:org.apache.karaf.decanter.collector/org.apache.karaf.decanter.collector.utils/2.12.0/$Bundle-SymbolicName=decanter-collector-utils&Bundle-Version=2.12.0&Bundle-ManifestVersion="2"&Export-Package=*;version="2.12.0",!*</bundle>
+ <bundle
start-level="30">mvn:org.apache.plc4x.merlot.kafka/org.apache.plc4x.merlot.kafka/${project.version}</bundle>
+ <configfile
finalname="/etc/org.apache.plc4x.merlot.kafka.collector-alarm.cfg">mvn:org.apache.plc4x.merlot.kafka/org.apache.plc4x.merlot.kafka/${project.version}/cfg</configfile>
+ </feature>
+
<feature name="plc4x-iotdb" version="${project.version}">
<!--<bundle
start-level="60">mvn:org.apache.iotdb/iotdb-session/${iotdb.session.version}</bundle>-->
<bundle
start-level="60">wrap:mvn:org.apache.iotdb/iotdb-session/${iotdb.session.version}/$Bundle-SymbolicName=iotdb-session&Bundle-Version=${iotdb.session.version}&Bundle-ManifestVersion="2"&Export-Package=*;version="${iotdb.session.version}",!*</bundle>
diff --git a/plc4j/tools/merlot/org.apache.plc4x.merlot.kafka/pom.xml
b/plc4j/tools/merlot/org.apache.plc4x.merlot.kafka/pom.xml
index c42666e..ad3e576 100644
--- a/plc4j/tools/merlot/org.apache.plc4x.merlot.kafka/pom.xml
+++ b/plc4j/tools/merlot/org.apache.plc4x.merlot.kafka/pom.xml
@@ -97,7 +97,7 @@
<configuration>
<artifacts>
<artifact>
-
<file>src/main/cfg/org.apache.plc4x.merlot.kafka.collector.cfg</file>
+
<file>src/main/cfg/org.apache.plc4x.merlot.kafka.collector-alarm.cfg</file>
<type>cfg</type>
</artifact>
</artifacts>
diff --git
a/plc4j/tools/merlot/org.apache.plc4x.merlot.kafka/src/main/cfg/org.apache.plc4x.merlot.kafka.collector.cfg
b/plc4j/tools/merlot/org.apache.plc4x.merlot.kafka/src/main/cfg/org.apache.plc4x.merlot.kafka.collector-alarm.cfg
similarity index 96%
rename from
plc4j/tools/merlot/org.apache.plc4x.merlot.kafka/src/main/cfg/org.apache.plc4x.merlot.kafka.collector.cfg
rename to
plc4j/tools/merlot/org.apache.plc4x.merlot.kafka/src/main/cfg/org.apache.plc4x.merlot.kafka.collector-alarm.cfg
index 15ebd09..3bc044d 100644
---
a/plc4j/tools/merlot/org.apache.plc4x.merlot.kafka/src/main/cfg/org.apache.plc4x.merlot.kafka.collector.cfg
+++
b/plc4j/tools/merlot/org.apache.plc4x.merlot.kafka/src/main/cfg/org.apache.plc4x.merlot.kafka.collector-alarm.cfg
@@ -76,10 +76,10 @@
# For SASL, you have to configure Java System property as explained in
http://kafka.apache.org/documentation.html#security_ssl
# message type (text to use the unmarshaller or raw to not use it)
-message.type=text
+#message.type=text
# Unmarshaller to use (json is recommended)
-unmarshaller.target=(dataFormat=json)
+#unmarshaller.target=(dataFormat=json)
# custom fields
-fields.remove.request-size-max=remove
\ No newline at end of file
+#fields.remove.request-size-max=remove
\ No newline at end of file
diff --git
a/plc4j/tools/merlot/org.apache.plc4x.merlot.kafka/src/main/java/org/apache/plc4x/merlot/kafka/core/MerlotKafkaManagedServiceFactory.java
b/plc4j/tools/merlot/org.apache.plc4x.merlot.kafka/src/main/java/org/apache/plc4x/merlot/kafka/core/MerlotKafkaManagedServiceFactory.java
index 5a41e00..65ed744 100644
---
a/plc4j/tools/merlot/org.apache.plc4x.merlot.kafka/src/main/java/org/apache/plc4x/merlot/kafka/core/MerlotKafkaManagedServiceFactory.java
+++
b/plc4j/tools/merlot/org.apache.plc4x.merlot.kafka/src/main/java/org/apache/plc4x/merlot/kafka/core/MerlotKafkaManagedServiceFactory.java
@@ -25,8 +25,6 @@ import
org.apache.plc4x.merlot.kafka.api.MerlotDecanterCollector;
import org.apache.plc4x.merlot.kafka.impl.MerlotKafkaDecanterCollectorImpl;
import org.osgi.framework.BundleContext;
import org.osgi.framework.Constants;
-import org.osgi.framework.InvalidSyntaxException;
-import org.osgi.framework.ServiceReference;
import org.osgi.framework.ServiceRegistration;
import org.osgi.service.cm.ConfigurationException;
import org.osgi.service.cm.ManagedServiceFactory;
@@ -36,13 +34,11 @@ import org.slf4j.LoggerFactory;
public class MerlotKafkaManagedServiceFactory implements ManagedServiceFactory
{
- private static final Logger LOGGER =
LoggerFactory.getLogger(MerlotKafkaManagedServiceFactory.class);
- private static String FILTER_COLLECTOR = "(&(" + Constants.OBJECTCLASS +
"=" + MerlotDecanterCollector.class.getName() + ")" +
- "(" + Constants.SERVICE_PID + "=*))";
-
- private final BundleContext ctx;
+ private static final Logger LOGGER =
LoggerFactory.getLogger(MerlotKafkaManagedServiceFactory.class);
+
+ private final BundleContext ctx;
private final EventAdmin dispatcher;
- private final Unmarshaller unmarshaller;
+ private Unmarshaller unmarshaller;
private Map<String,ServiceRegistration> services = new HashMap<>();
public MerlotKafkaManagedServiceFactory(BundleContext ctx, EventAdmin
dispatcher, Unmarshaller unmarshaller) {
@@ -50,17 +46,21 @@ public class MerlotKafkaManagedServiceFactory implements
ManagedServiceFactory {
this.dispatcher = dispatcher;
this.unmarshaller = unmarshaller;
}
-
+
@Override
public String getName() {
- return "Prueba";
+ return "Merlot Kafka Managed Service Factory";
}
@Override
public void updated(String pid, Dictionary<String, ?> properties) throws
ConfigurationException {
- System.out.println("Registering service: " + pid);
+ LOGGER.info("Registering service: {}", pid);
deleted(pid);
- MerlotKafkaDecanterCollectorImpl bundle = new
MerlotKafkaDecanterCollectorImpl(dispatcher, unmarshaller);
+ MerlotKafkaDecanterCollectorImpl bundle = new
MerlotKafkaDecanterCollectorImpl(dispatcher, unmarshaller);
+
+ bundle.activate(pid, (Dictionary<String, Object>) properties);
+ bundle.init();
+
Hashtable<String, String> serviceProperties = new Hashtable<>();
serviceProperties.put(Constants.SERVICE_PID, pid);
ServiceRegistration registration =
ctx.registerService(MerlotDecanterCollector.class, bundle, serviceProperties);
@@ -70,13 +70,25 @@ public class MerlotKafkaManagedServiceFactory implements
ManagedServiceFactory {
@Override
public void deleted(String pid) {
LOGGER.info("Removing service: " + pid);
- if (services.containsKey(pid)) {
- MerlotDecanterCollector bundle = (MerlotDecanterCollector)
ctx.getService(services.get(pid).getReference());
- bundle.destroy();
- services.get(pid).unregister();
+ ServiceRegistration registration = services.remove(pid);
+ if (registration != null) {
+ try {
+ MerlotDecanterCollector collector = (MerlotDecanterCollector)
ctx.getService(registration.getReference());
+ if (collector != null) {
+ collector.destroy();
+ }
+ registration.unregister();
+ } catch (Exception e) {
+ LOGGER.error("Error al eliminar el servicio {}", pid, e);
+ }
}
}
-
-
-
+
+ public void destroy() {
+ LOGGER.info("Destroying MerlotKafkaManagedServiceFactory, cleaning up
{} services", services.size());
+ for (String pid : services.keySet().toArray(new String[0])) {
+ deleted(pid);
+ }
+ }
+
}
diff --git
a/plc4j/tools/merlot/org.apache.plc4x.merlot.kafka/src/main/java/org/apache/plc4x/merlot/kafka/impl/MerlotKafkaDecanterCollectorImpl.java
b/plc4j/tools/merlot/org.apache.plc4x.merlot.kafka/src/main/java/org/apache/plc4x/merlot/kafka/impl/MerlotKafkaDecanterCollectorImpl.java
index 3075260..c6782c2 100644
---
a/plc4j/tools/merlot/org.apache.plc4x.merlot.kafka/src/main/java/org/apache/plc4x/merlot/kafka/impl/MerlotKafkaDecanterCollectorImpl.java
+++
b/plc4j/tools/merlot/org.apache.plc4x.merlot.kafka/src/main/java/org/apache/plc4x/merlot/kafka/impl/MerlotKafkaDecanterCollectorImpl.java
@@ -24,10 +24,17 @@ import java.util.Dictionary;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
+import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.errors.WakeupException;
import org.apache.karaf.decanter.api.marshaller.Unmarshaller;
import org.apache.karaf.decanter.collector.utils.PropertiesPreparator;
import org.apache.plc4x.merlot.kafka.api.MerlotDecanterCollector;
@@ -39,35 +46,53 @@ import org.slf4j.LoggerFactory;
public class MerlotKafkaDecanterCollectorImpl implements
MerlotDecanterCollector, Runnable {
- private static final Logger LOGGER =
LoggerFactory.getLogger(MerlotKafkaDecanterCollectorImpl.class);
-
+ private static final Logger LOGGER =
LoggerFactory.getLogger(MerlotKafkaDecanterCollectorImpl.class);
+
private String topic;
private String eventAdminTopic;
- private boolean consuming = false;
+ private volatile boolean consuming = false;
+ private final AtomicBoolean shutdownInitiated = new AtomicBoolean(false);
private String messageType;
-
- private Dictionary<String, Object> properties;
+
+ private Dictionary<String, Object> properties;
private KafkaConsumer<String, String> consumer;
-
+
private final EventAdmin dispatcher;
- private final Unmarshaller unmarshaller;
+ private Unmarshaller unmarshaller;
+ private ExecutorService executor;
public MerlotKafkaDecanterCollectorImpl(EventAdmin dispatcher,
Unmarshaller unmarshaller) {
this.dispatcher = dispatcher;
this.unmarshaller = unmarshaller;
}
-
+
@Override
public void init() {
- consuming = true;
- Executors.newSingleThreadExecutor().execute(this);
+ consuming = true;
+ this.executor = Executors.newSingleThreadExecutor();
+ this.executor.execute(this);
}
@Override
public void destroy() {
consuming = false;
- }
-
+ shutdownInitiated.set(true);
+ if (consumer != null) {
+ consumer.wakeup();
+ }
+ if (executor != null) {
+ executor.shutdown();
+ try {
+ if (!executor.awaitTermination(2, TimeUnit.SECONDS)) {
+ executor.shutdownNow();
+ }
+ } catch (InterruptedException e) {
+ executor.shutdownNow();
+ Thread.currentThread().interrupt();
+ }
+ }
+ }
+
public void activate(String pid, Dictionary<String, Object> properties) {
this.properties = properties;
topic = getValue(properties, "topic", "decanter");
@@ -139,64 +164,154 @@ public class MerlotKafkaDecanterCollectorImpl implements
MerlotDecanterCollector
String sslKeystoreType = getValue(properties, "ssl.keystore.type",
null);
if (sslKeystoreType != null)
- config.put("ssl.keystore.type", sslKeystoreType);
-
+ config.put("ssl.keystore.type", sslKeystoreType);
+
ClassLoader originClassLoader =
Thread.currentThread().getContextClassLoader();
try {
Thread.currentThread().setContextClassLoader(null);
consumer = new KafkaConsumer<String, String>(config);
String[] topics = topic.split(",");
- for (String t:topics){
- t = t.replaceAll("\\s+","");
+ for (int i = 0; i < topics.length; i++) {
+ topics[i] = topics[i].replaceAll("\\s+", "");
}
consumer.subscribe(Arrays.asList(topics));
} finally {
Thread.currentThread().setContextClassLoader(originClassLoader);
- }
+ }
}
-
+
@Override
public void run() {
- while (consuming) {
- try {
- consume();
- } catch (Exception e) {
- LOGGER.warn(e.getMessage(), e);
+ try {
+ while (consuming && !shutdownInitiated.get()) {
+ try {
+ consume();
+ } catch (WakeupException e) {
+ } catch (Exception e) {
+ LOGGER.info(e.getMessage(), e);
+ }
+ }
+ } finally {
+ if (consumer != null) {
+ try {
+ consumer.close();
+ } catch (Exception e) {
+ LOGGER.warn("Error closing Kafka consumer", e);
+ }
}
}
- }
-
- private void consume() throws UnsupportedEncodingException {
+ }
+
+ private void consume() {
ConsumerRecords<String, String> records =
consumer.poll(Duration.ofSeconds(1));
+
if (records.isEmpty()) {
return;
}
+
Map<String, Object> data = new HashMap<>();
- data.put("type", "kafka");
-
+ data.put("loki.label.job", "MerlotAlarmCollector");
+
for (ConsumerRecord<String, String> record : records) {
- String value = record.value();
- if (messageType.equalsIgnoreCase("text")) {
- ByteArrayInputStream is = new
ByteArrayInputStream(value.getBytes("utf-8"));
- data.putAll(unmarshaller.unmarshal(is));
- } else {
- data.put("payload", value);
+ if (!consuming) {
+ return;
}
- }
- try {
- PropertiesPreparator.prepare(data, properties);
- } catch (Exception e) {
- LOGGER.warn("Can't prepare data for the dispatcher", e);
+ //Data headers
+ String key = record.key();
+
+ //Alarm values
+ String value = record.value();
+
+ LOGGER.info("Key: {} Value: {}", key, value);
+
+ String pathPV = getPathPV(key);
+
+ //Loki paramaters
+ data.put("loki.label.topicalarm", getTopicAlarm(key));
+ data.put("alarm.pathpvname", pathPV);
+ data.put("loki.label.pvname",
pathPV.substring(pathPV.indexOf("//") + 2));
+ data.put("loki.label.component", getComponent(key));
+ data.put("loki.label.severity", getSeverity(value));
+ data.put("alarm.value", getValueAlarm(value));
+
+
+ //Send event bus karaf
+ Event event = new Event(eventAdminTopic, data);
+ dispatcher.postEvent(event);
}
+ }
- Event event = new Event(eventAdminTopic, data);
- dispatcher.postEvent(event);
- }
-
+
+ //Initial parameters
private String getValue(Dictionary<String, Object> config, String key,
String defaultValue) {
String value = (String)config.get(key);
return (value != null) ? value : defaultValue;
- }
-
+ }
+
+
+ //Kafka message parameters
+ public static String getTopicAlarm(String keyText) {
+ if (keyText == null) return null;
+ String regex = ":/([^/]+)/";
+ Matcher matcher = Pattern.compile(regex).matcher(keyText);
+
+ if (matcher.find()) {
+ return matcher.group(1);
+ }
+
+ return null;
+ }
+ public static String getPathPV(String keyText) {
+ if (keyText == null) return null;
+
+ int indexEndProtocol = keyText.indexOf(":\\/\\/");
+ if (indexEndProtocol == -1) {
+ indexEndProtocol = keyText.indexOf("://");
+ }
+
+ if (indexEndProtocol != -1) {
+ int indexLastSlash = keyText.lastIndexOf("/", indexEndProtocol);
+
+ if (indexLastSlash != -1) {
+ return keyText.substring(indexLastSlash + 1).replace("\\/\\/",
"//");
+ }
+ }
+ return null;
+ }
+ public static String getComponent(String keyText) {
+ if (keyText == null) return null;
+ String regex = "^[^:/]+:/[^/]+/(.+)/[a-zA-Z0-9]+:[\\\\/]{2}";
+
+ Matcher matcher = Pattern.compile(regex).matcher(keyText);
+
+ if (matcher.find()) {
+ return matcher.group(1);
+ }
+
+ return null;
+ }
+ public static String getSeverity(String valueText) {
+ if (valueText == null) return null;
+
+ String regex = "\"severity\"\\s*:\\s*\"([^\"]+)\"";
+
+ Matcher matcher = Pattern.compile(regex).matcher(valueText);
+
+ if (matcher.find()) {
+ return matcher.group(1);
+ }
+ return null;
+ }
+ public static String getValueAlarm(String valueText) {
+ if (valueText == null) return null;
+ String regex = "\"value\"\\s*:\\s*\"([^\"]+)\"";
+
+ Matcher matcher = Pattern.compile(regex).matcher(valueText);
+
+ if (matcher.find()) {
+ return matcher.group(1);
+ }
+ return null;
+ }
}
diff --git
a/plc4j/tools/merlot/org.apache.plc4x.merlot.kafka/src/main/resources/OSGI-INF/blueprint/kafka-service.xml
b/plc4j/tools/merlot/org.apache.plc4x.merlot.kafka/src/main/resources/OSGI-INF/blueprint/kafka-service.xml
index 69a97ec..125deed 100644
---
a/plc4j/tools/merlot/org.apache.plc4x.merlot.kafka/src/main/resources/OSGI-INF/blueprint/kafka-service.xml
+++
b/plc4j/tools/merlot/org.apache.plc4x.merlot.kafka/src/main/resources/OSGI-INF/blueprint/kafka-service.xml
@@ -27,11 +27,13 @@
interface="org.apache.karaf.decanter.api.marshaller.Unmarshaller"
timeout="20000"
availability="mandatory">
- </reference>
+ </reference>
<bean id="serviceBean"
class="org.apache.plc4x.merlot.kafka.core.MerlotKafkaManagedServiceFactory"
- scope="singleton">
+ scope="singleton"
+ destroy-method="destroy">
+
<argument ref="blueprintBundleContext"/>
<argument ref="refEventAdmin"/>
<argument ref="refMarshaller"/>
diff --git a/plc4j/tools/merlot/org.apache.plc4x.merlot/pom.xml
b/plc4j/tools/merlot/org.apache.plc4x.merlot/pom.xml
index 9677963..d042bb5 100644
--- a/plc4j/tools/merlot/org.apache.plc4x.merlot/pom.xml
+++ b/plc4j/tools/merlot/org.apache.plc4x.merlot/pom.xml
@@ -201,8 +201,7 @@
</startupFeatures>
<bootFeatures>
- <feature>org.apache.plc4x.merlot.features</feature>
- <feature>decanter-appender-loki</feature>
+ <feature>org.apache.plc4x.merlot.features</feature>
<feature>wrap</feature>
<feature>classpath</feature>
<feature>shell</feature>
diff --git a/plc4j/tools/merlot/pom.xml b/plc4j/tools/merlot/pom.xml
index c101b64..3e333ea 100644
--- a/plc4j/tools/merlot/pom.xml
+++ b/plc4j/tools/merlot/pom.xml
@@ -100,6 +100,7 @@ under the License.
<disruptor.version>4.0.0</disruptor.version>
<maven-failsafe-plugin.version>3.5.3</maven-failsafe-plugin.version>
+ <kafka.version>3.9.2</kafka.version>
<!-- EPICS -->
<epics.core.version>7.0.9</epics.core.version>
<epics.util.version>1.0.7</epics.util.version>
