This is an automated email from the ASF dual-hosted git repository.
dominikriemer pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/streampipes.git
The following commit(s) were added to refs/heads/dev by this push:
new 521dac6092 feat: Support structured logging configurations (#5198)
521dac6092 is described below
commit 521dac609250edf9840204aa35a22f2221468a27
Author: Dominik Riemer <[email protected]>
AuthorDate: Fri Nov 6 17:19:18 2026 +0100
feat: Support structured logging configurations (#5198)
---
pom.xml | 12 ++++
streampipes-service-base/pom.xml | 7 ++
.../service/base/PreContextStartup.java | 33 +++++++++
.../service/base/PreContextStartupListener.java | 58 +++++++++++++++
.../service/base/StreamPipesServiceBase.java | 14 ++--
.../src/main/resources/logback.xml | 38 ----------
.../base/PreContextStartupListenerTest.java | 84 ++++++++++++++++++++++
.../service/core/StreamPipesCoreApplication.java | 11 +--
.../src/main/resources/application.properties | 3 +-
.../StreamPipesExtensionsServiceBase.java | 42 ++++++-----
.../src/main/resources/application.properties | 3 +-
11 files changed, 231 insertions(+), 74 deletions(-)
diff --git a/pom.xml b/pom.xml
index 8b1588d00f..aab9a316a0 100644
--- a/pom.xml
+++ b/pom.xml
@@ -540,6 +540,10 @@
<groupId>ch.qos.reload4j</groupId>
<artifactId>reload4j</artifactId>
</exclusion>
+ <exclusion>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-reload4j</artifactId>
+ </exclusion>
<exclusion>
<groupId>io.netty</groupId>
<artifactId>netty-transport-native-epoll</artifactId>
@@ -603,6 +607,14 @@
<groupId>com.google.inject</groupId>
<artifactId>guice</artifactId>
</exclusion>
+ <exclusion>
+ <groupId>ch.qos.reload4j</groupId>
+ <artifactId>reload4j</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-reload4j</artifactId>
+ </exclusion>
</exclusions>
</dependency>
<dependency>
diff --git a/streampipes-service-base/pom.xml b/streampipes-service-base/pom.xml
index 92bc8cd723..f5dc67693f 100644
--- a/streampipes-service-base/pom.xml
+++ b/streampipes-service-base/pom.xml
@@ -86,6 +86,13 @@
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-actuator</artifactId>
</dependency>
+
+ <!-- Test dependencies -->
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter-api</artifactId>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<build>
<plugins>
diff --git
a/streampipes-service-base/src/main/java/org/apache/streampipes/service/base/PreContextStartup.java
b/streampipes-service-base/src/main/java/org/apache/streampipes/service/base/PreContextStartup.java
new file mode 100644
index 0000000000..af744385e0
--- /dev/null
+++
b/streampipes-service-base/src/main/java/org/apache/streampipes/service/base/PreContextStartup.java
@@ -0,0 +1,33 @@
+/*
+ * 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.streampipes.service.base;
+
+import java.net.UnknownHostException;
+
+/**
+ * Startup work that has to finish before the Spring context is created. It
runs after
+ * Spring Boot has initialized logging, so everything it logs uses the
configured format.
+ */
+@FunctionalInterface
+public interface PreContextStartup {
+
+ /**
+ * @return the networking configuration the service is started with
+ */
+ BaseNetworkingConfig run() throws UnknownHostException;
+}
diff --git
a/streampipes-service-base/src/main/java/org/apache/streampipes/service/base/PreContextStartupListener.java
b/streampipes-service-base/src/main/java/org/apache/streampipes/service/base/PreContextStartupListener.java
new file mode 100644
index 0000000000..c68b513663
--- /dev/null
+++
b/streampipes-service-base/src/main/java/org/apache/streampipes/service/base/PreContextStartupListener.java
@@ -0,0 +1,58 @@
+/*
+ * 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.streampipes.service.base;
+
+import
org.springframework.boot.context.event.ApplicationEnvironmentPreparedEvent;
+import org.springframework.context.ApplicationListener;
+import org.springframework.core.env.PropertiesPropertySource;
+
+import java.net.UnknownHostException;
+import java.util.Properties;
+
+/**
+ * Has no explicit order on purpose: it must run after Spring Boot's
LoggingApplicationListener,
+ * which handles the same event at a high precedence.
+ */
+class PreContextStartupListener implements
ApplicationListener<ApplicationEnvironmentPreparedEvent> {
+
+ private static final String PROPERTY_SOURCE_NAME =
"streampipesServiceConfig";
+
+ private final PreContextStartup startup;
+
+ PreContextStartupListener(PreContextStartup startup) {
+ this.startup = startup;
+ }
+
+ @Override
+ public void onApplicationEvent(ApplicationEnvironmentPreparedEvent event) {
+ BaseNetworkingConfig networkingConfig;
+ try {
+ networkingConfig = startup.run();
+ } catch (UnknownHostException e) {
+ throw new IllegalStateException(
+ "Could not auto-resolve host address - please manually provide the
hostname"
+ + " using the `SP_HOST` environment variable", e);
+ }
+
+ Properties config = new Properties();
+ ServiceBaseConfig.addPrometheusConfig(config);
+ ServiceBaseConfig.addPortConfig(networkingConfig.getPort(), config);
+
+ event.getEnvironment().getPropertySources().addLast(new
PropertiesPropertySource(PROPERTY_SOURCE_NAME, config));
+ }
+}
diff --git
a/streampipes-service-base/src/main/java/org/apache/streampipes/service/base/StreamPipesServiceBase.java
b/streampipes-service-base/src/main/java/org/apache/streampipes/service/base/StreamPipesServiceBase.java
index 6505e9ab20..a9fe7ef68e 100644
---
a/streampipes-service-base/src/main/java/org/apache/streampipes/service/base/StreamPipesServiceBase.java
+++
b/streampipes-service-base/src/main/java/org/apache/streampipes/service/base/StreamPipesServiceBase.java
@@ -21,7 +21,6 @@ import org.apache.commons.lang3.RandomStringUtils;
import org.springframework.boot.SpringApplication;
import java.net.UnknownHostException;
-import java.util.Properties;
public abstract class StreamPipesServiceBase {
@@ -29,18 +28,13 @@ public abstract class StreamPipesServiceBase {
public void startStreamPipesService(Class<?> serviceClass,
BaseNetworkingConfig
networkingConfig) throws UnknownHostException {
- runApplication(serviceClass, networkingConfig.getPort());
+ startStreamPipesService(serviceClass, () -> networkingConfig);
}
- private void runApplication(Class<?> serviceClass,
- Integer port) {
+ public void startStreamPipesService(Class<?> serviceClass,
+ PreContextStartup startup) {
SpringApplication app = new SpringApplication(serviceClass);
-
- Properties config = new Properties();
- ServiceBaseConfig.addPrometheusConfig(config);
- ServiceBaseConfig.addPortConfig(port, config);
-
- app.setDefaultProperties(config);
+ app.addListeners(new PreContextStartupListener(startup));
app.run();
}
diff --git a/streampipes-service-base/src/main/resources/logback.xml
b/streampipes-service-base/src/main/resources/logback.xml
deleted file mode 100644
index 1331d020df..0000000000
--- a/streampipes-service-base/src/main/resources/logback.xml
+++ /dev/null
@@ -1,38 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<!--
- ~ 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.
- ~
- -->
-
-<configuration>
-
- <include resource="org/springframework/boot/logging/logback/defaults.xml"/>
-
- <appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
- <encoder>
- <pattern>${CONSOLE_LOG_PATTERN}</pattern>
- <charset>utf8</charset>
- </encoder>
- </appender>
-
- <logger name="org.lightcouch" level="ERROR"/>
- <logger name="org.apache.kafka" level="ERROR"/>
-
-
- <root level="INFO">
- <appender-ref ref="CONSOLE"/>
- </root>
-</configuration>
diff --git
a/streampipes-service-base/src/test/java/org/apache/streampipes/service/base/PreContextStartupListenerTest.java
b/streampipes-service-base/src/test/java/org/apache/streampipes/service/base/PreContextStartupListenerTest.java
new file mode 100644
index 0000000000..8f7709c8a1
--- /dev/null
+++
b/streampipes-service-base/src/test/java/org/apache/streampipes/service/base/PreContextStartupListenerTest.java
@@ -0,0 +1,84 @@
+/*
+ * 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.streampipes.service.base;
+
+import org.junit.jupiter.api.Test;
+import org.slf4j.LoggerFactory;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.WebApplicationType;
+import org.springframework.context.annotation.Configuration;
+
+import java.net.UnknownHostException;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class PreContextStartupListenerTest {
+
+ @Configuration(proxyBeanMethods = false)
+ static class EmptyConfiguration {
+ }
+
+ private SpringApplication application(PreContextStartup startup) {
+ var application = new SpringApplication(EmptyConfiguration.class);
+ application.setWebApplicationType(WebApplicationType.NONE);
+ application.addListeners(new PreContextStartupListener(startup));
+ return application;
+ }
+
+ @Test
+ void startupRunsOnceLoggingIsInitialized() {
+ // Spring Boot suppresses all log output from the start of the run until
logging is initialized
+ var loggingEnabled = new AtomicBoolean(false);
+
+ try (var ignored = application(() -> {
+
loggingEnabled.set(LoggerFactory.getLogger(PreContextStartupListenerTest.class).isInfoEnabled());
+ return new BaseNetworkingConfig("localhost", 8090);
+ }).run()) {
+ assertTrue(loggingEnabled.get());
+ }
+ }
+
+ @Test
+ void portFromStartupIsAvailableToTheContext() {
+ try (var context = application(() -> new BaseNetworkingConfig("localhost",
8123)).run()) {
+ assertEquals("8123",
context.getEnvironment().getProperty("server.port"));
+ }
+ }
+
+ @Test
+ void configuredPortWinsOverPortFromStartup() {
+ try (var context = application(() -> new BaseNetworkingConfig("localhost",
8123))
+ .run("--server.port=9000")) {
+ assertEquals("9000",
context.getEnvironment().getProperty("server.port"));
+ }
+ }
+
+ @Test
+ void unresolvableHostFailsStartupWithHint() {
+ var application = application(() -> {
+ throw new UnknownHostException("unknown");
+ });
+
+ var exception = assertThrows(IllegalStateException.class,
application::run);
+
+ assertTrue(exception.getMessage().contains("SP_HOST"));
+ }
+}
diff --git
a/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/StreamPipesCoreApplication.java
b/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/StreamPipesCoreApplication.java
index 6edc4ab024..145ded42b8 100644
---
a/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/StreamPipesCoreApplication.java
+++
b/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/StreamPipesCoreApplication.java
@@ -84,7 +84,6 @@ import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import java.io.IOException;
-import java.net.UnknownHostException;
import java.time.Duration;
import java.util.List;
import java.util.Objects;
@@ -162,15 +161,11 @@ public class StreamPipesCoreApplication extends
StreamPipesServiceBase {
public void initialize(SupportedProtocols supportedProtocols,
List<Supplier<TransformationEngine>>
transformationEngines) {
- try {
+ startStreamPipesService(StreamPipesCoreApplication.class, () -> {
registerProtocols(supportedProtocols);
registerTransformationEngines(transformationEngines);
- BaseNetworkingConfig networkingConfig =
BaseNetworkingConfig.defaultResolution(8030);
- startStreamPipesService(StreamPipesCoreApplication.class,
networkingConfig);
- } catch (UnknownHostException e) {
- LOG.error("Could not auto-resolve host address - please manually provide
the hostname"
- + " using the `SP_HOST` environment variable");
- }
+ return BaseNetworkingConfig.defaultResolution(8030);
+ });
}
protected void
registerTransformationEngines(List<Supplier<TransformationEngine>>
transformationEngines) {
diff --git a/streampipes-service-core/src/main/resources/application.properties
b/streampipes-service-core/src/main/resources/application.properties
index 26febd012f..63435cc25c 100644
--- a/streampipes-service-core/src/main/resources/application.properties
+++ b/streampipes-service-core/src/main/resources/application.properties
@@ -35,7 +35,8 @@
streampipes.storage.cache.core-configuration.enabled=${SP_STORAGE_CACHE_CORE_CON
spring.cache.type=caffeine
spring.cache.cache-names=dataExplorerWidgets,permissions,adapters,dashboards,pipelines,dataLakeMeasures,roles,userGroups,privileges,users,coreConfiguration
spring.cache.caffeine.spec=maximumSize=10000,expireAfterWrite=10m,recordStats
-logging.config=classpath:logback.xml
+logging.level.org.lightcouch=ERROR
+logging.level.org.apache.kafka=ERROR
spring.output.ansi.enabled=always
springdoc.api-docs.enabled=${SP_API_DOCS_ENABLED:true}
springdoc.api-docs.path=/apidocs
diff --git
a/streampipes-service-extensions/src/main/java/org/apache/streampipes/service/extensions/StreamPipesExtensionsServiceBase.java
b/streampipes-service-extensions/src/main/java/org/apache/streampipes/service/extensions/StreamPipesExtensionsServiceBase.java
index 43b89eeaf5..f2a8aefe5b 100644
---
a/streampipes-service-extensions/src/main/java/org/apache/streampipes/service/extensions/StreamPipesExtensionsServiceBase.java
+++
b/streampipes-service-extensions/src/main/java/org/apache/streampipes/service/extensions/StreamPipesExtensionsServiceBase.java
@@ -62,6 +62,7 @@ import java.net.UnknownHostException;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Supplier;
import java.util.stream.Collectors;
@@ -84,12 +85,19 @@ public abstract class StreamPipesExtensionsServiceBase
extends StreamPipesServic
private volatile SpServiceRegistration currentServiceRegistration;
public void init() {
- SpServiceDefinition serviceDef = provideServiceDefinition();
- init(serviceDef);
+ startService(this::provideServiceDefinition);
}
public void init(SpServiceDefinition serviceDef) {
- try {
+ startService(() -> serviceDef);
+ }
+
+ private void startService(Supplier<SpServiceDefinition> serviceDefSupplier) {
+ var serviceDefinition = new AtomicReference<SpServiceDefinition>();
+
+ startStreamPipesService(this.getClass(), () -> {
+ SpServiceDefinition serviceDef = serviceDefSupplier.get();
+ serviceDefinition.set(serviceDef);
BaseNetworkingConfig networkingConfig =
BaseNetworkingConfig.defaultResolution(serviceDef.getDefaultPort());
String serviceId = serviceDef.getServiceGroup() + "-" +
AUTO_GENERATED_SERVICE_ID;
serviceDef.setServiceId(serviceId);
@@ -100,14 +108,12 @@ public abstract class StreamPipesExtensionsServiceBase
extends StreamPipesServic
GraalJsScriptEngine::new
));
- startExtensionsService(this.getClass(), serviceDef, networkingConfig);
- ServiceLoadDataReportGenerator.getInstance().initialize();
+ registerExtensionsService(serviceDef, networkingConfig);
+ return networkingConfig;
+ });
- } catch (UnknownHostException e) {
- LOG.error(
- "Could not auto-resolve host address - "
- + "please manually provide the hostname using the SP_HOST
environment variable");
- }
+ this.afterServiceRegistered(serviceDefinition.get(),
this.currentServiceRegistration);
+ ServiceLoadDataReportGenerator.getInstance().initialize();
}
protected void
registerTransformationEngines(List<Supplier<TransformationEngine>>
transformationEngines) {
@@ -125,6 +131,16 @@ public abstract class StreamPipesExtensionsServiceBase
extends StreamPipesServic
public void startExtensionsService(Class<?> serviceClass,
SpServiceDefinition serviceDef,
BaseNetworkingConfig networkingConfig)
throws UnknownHostException {
+ this.startStreamPipesService(serviceClass, () -> {
+ registerExtensionsService(serviceDef, networkingConfig);
+ return networkingConfig;
+ });
+
+ this.afterServiceRegistered(serviceDef, this.currentServiceRegistration);
+ }
+
+ private void registerExtensionsService(SpServiceDefinition serviceDef,
+ BaseNetworkingConfig
networkingConfig) {
this.extensionTransportMode = ExtensionServiceTransportMode.from(
Environments.getEnvironment().getExtensionTransportMode().getValueOrDefault()
);
@@ -164,12 +180,6 @@ public abstract class StreamPipesExtensionsServiceBase
extends StreamPipesServic
registerService(req);
this.registerConfigs(serviceDef.getServiceGroup(),
serviceDef.getServiceName(), serviceDef.getKvConfigs());
- this.startStreamPipesService(
- serviceClass,
- networkingConfig
- );
-
- this.afterServiceRegistered(serviceDef, req);
}
private void registerService(SpServiceRegistration serviceRegistration) {
diff --git
a/streampipes-service-extensions/src/main/resources/application.properties
b/streampipes-service-extensions/src/main/resources/application.properties
index 8814c22a59..e0006c18f1 100644
--- a/streampipes-service-extensions/src/main/resources/application.properties
+++ b/streampipes-service-extensions/src/main/resources/application.properties
@@ -16,5 +16,6 @@
#
#
-logging.config=classpath:logback.xml
+logging.level.org.lightcouch=ERROR
+logging.level.org.apache.kafka=ERROR
spring.output.ansi.enabled=always
