Merge branch 'RED-8171' into 'main'

RED-8171 : `@Async` support

See merge request fforesight/tracing-commons!2

docker/README.md

@@ -0,0 +1 @@
+`docker-compose up -d` to start a local opentelemetry collector + zipkin for development purposes

docker/collector-config.yaml

@@ -0,0 +1,32 @@
+receivers:
+  otlp:
+    protocols:
+      grpc:
+      http:
+        cors:
+          allowed_origins:
+            - http://*
+            - https://*
+
+exporters:
+  zipkin:
+    endpoint: "http://zipkin-all-in-one:9411/api/v2/spans"
+  prometheus:
+    endpoint: "0.0.0.0:9464"
+
+processors:
+  batch:
+
+service:
+  telemetry:
+    logs:
+      level: "debug"
+  pipelines:
+    traces:
+      receivers: [otlp]
+      exporters: [zipkin]
+      processors: [batch]
+    metrics:
+      receivers: [otlp]
+      exporters: [prometheus]
+      processors: [batch]

docker/docker-compose.yaml

@@ -0,0 +1,18 @@
+version: "3"
+services:
+  collector:
+    image: otel/opentelemetry-collector-contrib:0.53.0
+    command: ["--config=/conf/collector-config.yaml"]
+    volumes:
+      - ./collector-config.yaml:/conf/collector-config.yaml
+    ports:
+      - "9464:9464"
+      - "4317:4317"
+      - "4318:4318"
+    depends_on:
+      - zipkin-all-in-one
+
+  zipkin-all-in-one:
+    image: openzipkin/zipkin:latest
+    ports:
+      - "9411:9411"

pom.xml

@@ -18,7 +18,7 @@
     <properties>
         <java.version>17</java.version>
         <spring-cloud.version>2022.0.4</spring-cloud.version>
-        <tenant-commons.version>0.19.0</tenant-commons.version>
+        <tenant-commons.version>0.20.0</tenant-commons.version>
     </properties>
 
 
@@ -87,6 +87,12 @@
             <artifactId>spring-rabbit-test</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>io.micrometer</groupId>
+            <artifactId>micrometer-tracing-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+
     </dependencies>
     <dependencyManagement>
         <dependencies>

src/main/java/com/knecon/fforesight/tracing/DefaultTracingAutoConfiguration.java

@@ -1,29 +1,45 @@
 package com.knecon.fforesight.tracing;
 
-import org.springframework.boot.actuate.autoconfigure.tracing.ConditionalOnEnabledTracing;
-import org.springframework.boot.autoconfigure.AutoConfiguration;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.ComponentScan;
-import org.springframework.http.server.observation.ServerRequestObservationContext;
-
+import io.micrometer.context.ContextRegistry;
 import io.micrometer.observation.ObservationPredicate;
+import io.micrometer.tracing.Tracer;
+import io.micrometer.tracing.contextpropagation.ObservationAwareSpanThreadLocalAccessor;
 import jakarta.annotation.PostConstruct;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.actuate.autoconfigure.tracing.ConditionalOnEnabledTracing;
+import org.springframework.boot.autoconfigure.AutoConfiguration;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Import;
+import org.springframework.context.annotation.PropertySource;
+import org.springframework.http.server.observation.ServerRequestObservationContext;
 
-@Slf4j
-@ComponentScan
 @AutoConfiguration
+@ConditionalOnBean({Tracer.class})
+@ConditionalOnEnabledTracing
+@Import({
+        TaskExecutionConfiguration.class,
+        TracingRabbitConfiguration.class,
+        WebMvcConfiguration.class
+})
+@PropertySource("classpath:tracing-task.properties")
+@Slf4j
 public class DefaultTracingAutoConfiguration {
 
+    @Autowired
+    Tracer tracer;
+
     @PostConstruct
     public void postConstruct() {
 
+        ContextRegistry.getInstance().registerThreadLocalAccessor(new ObservationAwareSpanThreadLocalAccessor(tracer));
+
         log.info("Tracing AutoConfiguration Loaded!");
     }
 
 
     @Bean
-    @ConditionalOnEnabledTracing
     ObservationPredicate actuatorServerContextPredicate() {
 
         // There might be a better solution soon: https://github.com/spring-projects/spring-boot/issues/34801

src/main/java/com/knecon/fforesight/tracing/TaskExecutionConfiguration.java

@@ -0,0 +1,44 @@
+package com.knecon.fforesight.tracing;
+
+import com.knecon.fforesight.tenantcommons.task.KneconTaskDecorator;
+import io.micrometer.context.ContextSnapshot;
+import lombok.extern.slf4j.Slf4j;
+import lombok.val;
+import org.springframework.beans.factory.ObjectProvider;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.task.TaskExecutorCustomizer;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.scheduling.annotation.EnableAsync;
+
+import java.util.function.Supplier;
+
+@Configuration(proxyBeanMethods = false)
+@EnableAsync
+@Slf4j
+public class TaskExecutionConfiguration {
+
+    private final static Supplier<KneconTaskDecorator> IDENTITY_KNECON_TASK_DECORATOR_SUPPLIER = () -> runna -> runna;
+
+    @Value("${spring.task.execution.pool.warnPercentageThreshold:80}")
+    private int executorWarnPercentageThreshold;
+
+    @Bean
+    public TaskExecutorCustomizer tracingTaskExecutorCustomizer(ObjectProvider<KneconTaskDecorator> taskDecorator) {
+        return taskExecutor -> {
+            taskExecutor.setTaskDecorator((KneconTaskDecorator) runnable -> {
+                val taskDecoratorUsed = taskDecorator.getIfUnique(IDENTITY_KNECON_TASK_DECORATOR_SUPPLIER);
+                val decoratedRunnable = taskDecoratorUsed.decorate(runnable);
+                val actualQueueSize = taskExecutor.getQueueSize() + 1;
+                val queueOccupancyPercentage = actualQueueSize * 100.f / taskExecutor.getMaxPoolSize();
+
+                if (queueOccupancyPercentage >= executorWarnPercentageThreshold) {
+                    log.warn("Executor pool [ " + taskExecutor + " ] queue size reached " + actualQueueSize + "/" + taskExecutor.getMaxPoolSize() +
+                            " entries awaiting execution triggering the warn level set for " + executorWarnPercentageThreshold +"% occupancy.");
+                }
+
+                return ContextSnapshot.captureAll(new Object[0]).wrap(decoratedRunnable);
+            });
+        };
+    }
+}

src/main/java/com/knecon/fforesight/tracing/TracingRabbitConfiguration.java

@@ -1,22 +1,15 @@
 package com.knecon.fforesight.tracing;
 
-import org.springframework.amqp.rabbit.config.ContainerCustomizer;
-import org.springframework.amqp.rabbit.core.RabbitTemplate;
-import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
-import org.springframework.boot.actuate.autoconfigure.tracing.ConditionalOnEnabledTracing;
+import com.knecon.fforesight.tenantcommons.RabbitTemplateMultiCustomizer;
+import com.knecon.fforesight.tenantcommons.SimpleMessageListenerContainerCustomizer;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
-import com.knecon.fforesight.tenantcommons.RabbitTemplateMultiCustomizer;
-import com.knecon.fforesight.tenantcommons.SimpleMessageListenerContainerCustomizer;
-
-import lombok.extern.slf4j.Slf4j;
-
 @Slf4j
-@Configuration
-@ConditionalOnEnabledTracing
-@ConditionalOnClass({RabbitTemplate.class, SimpleMessageListenerContainer.class, ContainerCustomizer.class})
+@Configuration(proxyBeanMethods = false)
+@ConditionalOnClass({ SimpleMessageListenerContainerCustomizer.class, RabbitTemplateMultiCustomizer.class })
 public class TracingRabbitConfiguration {
 
     @Bean

src/main/java/com/knecon/fforesight/tracing/WebMvcConfiguration.java

@@ -0,0 +1,17 @@
+package com.knecon.fforesight.tracing;
+
+import io.micrometer.context.ContextSnapshot;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.core.task.SimpleAsyncTaskExecutor;
+import org.springframework.web.servlet.config.annotation.AsyncSupportConfigurer;
+import org.springframework.web.servlet.config.annotation.WebMvcConfigurer;
+
+@Configuration(proxyBeanMethods = false)
+public class WebMvcConfiguration implements WebMvcConfigurer {
+
+    @Override
+    public void configureAsyncSupport(AsyncSupportConfigurer configurer) {
+
+        configurer.setTaskExecutor(new SimpleAsyncTaskExecutor(r -> new Thread(ContextSnapshot.captureAll().wrap(r))));
+    }
+}

src/main/resources/tracing-task.properties

@@ -0,0 +1,10 @@
+#
+# Configure task executor specs
+#
+# For more available props,
+#     see: https://github.com/spring-projects/spring-boot/blob/main/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/task/TaskExecutionProperties.java
+#
+spring.task.execution.pool.coreSize=4
+spring.task.execution.pool.maxSize=4
+spring.task.execution.pool.queueCapacity=10000
+spring.task.execution.threadNamePrefix=TracingAwareTaskExecutor-

src/test/java/com/knecon/fforesight/tracing/ObservationEnabledTest.java

@@ -0,0 +1,59 @@
+package com.knecon.fforesight.tracing;
+
+import com.knecon.fforesight.tenantcommons.RabbitTemplateMultiCustomizer;
+import com.knecon.fforesight.tenantcommons.SimpleMessageListenerContainerCustomizer;
+import lombok.extern.slf4j.Slf4j;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
+import org.springframework.boot.test.context.runner.ApplicationContextRunner;
+import org.springframework.boot.test.system.CapturedOutput;
+import org.springframework.boot.test.system.OutputCaptureExtension;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+
+@ExtendWith(OutputCaptureExtension.class)
+@Slf4j
+public class ObservationEnabledTest {
+
+
+    private final ApplicationContextRunner contextRunner = new ApplicationContextRunner()
+            .withPropertyValues("management.tracing.enabled=true")
+            .withUserConfiguration(SharedTestConfiguration.class, DefaultTracingAutoConfiguration.class);
+
+    @Test
+    public void testTracingAutoConfigurationLoaded(CapturedOutput output) {
+        this.contextRunner.run(context -> {
+            assertThat(output.getOut()).contains("Tracing AutoConfiguration Loaded!");
+        });
+    }
+
+    @Test
+    public void testRabbitTracingEnabled() {
+
+        this.contextRunner.run(context -> {
+            var rabbitTemplateCustomizer = context.getBean(RabbitTemplateMultiCustomizer.class);
+            var rabbitTemplate = mock(RabbitTemplate.class);
+
+            rabbitTemplateCustomizer.customize(rabbitTemplate);
+            verify(rabbitTemplate).setObservationEnabled(true);
+        });
+    }
+
+
+    @Test
+    public void testMessageListenerContainerTracingEnabled() {
+
+        this.contextRunner.run(context -> {
+            var containerConfigurer = context.getBean(SimpleMessageListenerContainerCustomizer.class);
+            var container = mock(SimpleMessageListenerContainer.class);
+
+            containerConfigurer.customize(container);
+            verify(container).setObservationEnabled(true);
+        });
+    }
+
+}

src/test/java/com/knecon/fforesight/tracing/SharedTestConfiguration.java

@@ -0,0 +1,16 @@
+package com.knecon.fforesight.tracing;
+
+import io.micrometer.tracing.Tracer;
+import io.micrometer.tracing.test.simple.SimpleTracer;
+import org.springframework.boot.autoconfigure.AutoConfiguration;
+import org.springframework.context.annotation.Bean;
+
+@AutoConfiguration
+public class SharedTestConfiguration {
+
+    @Bean
+    Tracer simpleTracer() {
+        return new SimpleTracer();
+    }
+
+}

src/test/java/com/knecon/fforesight/tracing/TaskExecutionTracingTest.java

@@ -0,0 +1,89 @@
+package com.knecon.fforesight.tracing;
+
+import io.micrometer.observation.Observation;
+import io.micrometer.observation.ObservationRegistry;
+import io.micrometer.tracing.Span;
+import io.micrometer.tracing.Tracer;
+import io.micrometer.tracing.test.simple.SimpleTracer;
+import lombok.extern.slf4j.Slf4j;
+import org.assertj.core.api.BDDAssertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.AutoConfigurations;
+import org.springframework.boot.autoconfigure.task.TaskExecutionAutoConfiguration;
+import org.springframework.boot.test.context.runner.ApplicationContextRunner;
+import org.springframework.boot.test.system.CapturedOutput;
+import org.springframework.boot.test.system.OutputCaptureExtension;
+import org.springframework.scheduling.annotation.Async;
+import org.springframework.stereotype.Component;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
+
+@ExtendWith(OutputCaptureExtension.class)
+@Slf4j
+public class TaskExecutionTracingTest {
+
+    private static final String ASYNC_SPAN_NAME = "Async Span";
+
+    private final ApplicationContextRunner contextRunner = new ApplicationContextRunner()
+            .withPropertyValues("management.tracing.enabled=true")
+            .withConfiguration(AutoConfigurations.of(TaskExecutionAutoConfiguration.class))
+            .withUserConfiguration(SharedTestConfiguration.class, AsyncLogic.class, DefaultTracingAutoConfiguration.class);
+    private final ObservationRegistry observationRegistry = ObservationRegistry.create();
+
+
+    @Test
+    public void asyncMethodTraceTest(CapturedOutput output) {
+        this.contextRunner.run(context -> {
+            SimpleTracer tracer = (SimpleTracer)context.getBean(Tracer.class);
+            var asyncLogic = context.getBean(AsyncLogic.class);
+
+            Observation firstSpan = Observation.createNotStarted("First span", observationRegistry).highCardinalityKeyValue("test", "test 1");
+            try (Observation.Scope scope = firstSpan.start().openScope()) {
+                log.info("Async in test with observation - before call");
+
+                Span secondSpan = tracer.nextSpan().name("Second span").tag("test", "test 2");
+                try (Tracer.SpanInScope scope2 = tracer.withSpan(secondSpan.start())) {
+                    log.info("Async in test with span - before call");
+                    Future<String> future = asyncLogic.asyncCall();
+                    String spanIdFromFuture = future.get(1, TimeUnit.SECONDS);
+                    log.info("Async in test with span - after call");
+                    BDDAssertions.then(spanIdFromFuture).isEqualTo(secondSpan.context().spanId());
+                }
+                finally {
+                    secondSpan.end();
+                }
+
+                log.info("Async in test with observation - after call");
+            }
+            finally {
+                firstSpan.stop();
+            }
+
+            var tracerSpans = tracer.getSpans().stream().toList();
+
+            assertThat(tracerSpans.size()).isEqualTo(2);
+            assertThat(tracerSpans.get(1).getName()).isEqualTo(ASYNC_SPAN_NAME);
+            assertThat(tracerSpans.get(1).getParentId()).isEqualTo(tracerSpans.get(0).getSpanId());
+        });
+    }
+
+    @Component
+    static class AsyncLogic {
+        @Autowired
+        Tracer tracer;
+
+        @Async("applicationTaskExecutor")
+        public Future<String> asyncCall() {
+            log.info("TASK EXECUTOR");
+            tracer.nextSpan().name(ASYNC_SPAN_NAME).tag("test", "test 3").start().end();
+            String spanId = tracer.currentSpan().context().spanId();
+            return CompletableFuture.supplyAsync(() -> spanId);
+        }
+    }
+}

src/test/java/com/knecon/fforesight/tracing/TracingTests.java

@@ -1,43 +0,0 @@
-package com.knecon.fforesight.tracing;
-
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.verify;
-
-import org.junit.jupiter.api.Test;
-import org.springframework.amqp.rabbit.config.ContainerCustomizer;
-import org.springframework.amqp.rabbit.core.RabbitTemplate;
-import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
-import org.springframework.boot.autoconfigure.AutoConfigurations;
-import org.springframework.boot.autoconfigure.amqp.RabbitTemplateCustomizer;
-import org.springframework.boot.test.context.runner.ApplicationContextRunner;
-
-import com.knecon.fforesight.tenantcommons.MultiTenancyAutoConfiguration;
-
-import lombok.extern.slf4j.Slf4j;
-
-@Slf4j
-public class TracingTests {
-
-    private final ApplicationContextRunner contextRunner = new ApplicationContextRunner().withConfiguration(AutoConfigurations.of(TracingRabbitConfiguration.class,
-            MultiTenancyAutoConfiguration.class));
-
-
-    @Test
-    public void testTracingEnabled() {
-
-        this.contextRunner.withPropertyValues("management.tracing.enabled=true").run(context -> {
-            var rabbitTemplateCustomizer = context.getBean(RabbitTemplateCustomizer.class);
-            var rabbitTemplate = mock(RabbitTemplate.class);
-
-            rabbitTemplateCustomizer.customize(rabbitTemplate);
-            verify(rabbitTemplate).setObservationEnabled(true);
-        }).run(context -> {
-            var containerConfigurer = context.getBean(ContainerCustomizer.class);
-            var container = mock(SimpleMessageListenerContainer.class);
-
-            containerConfigurer.configure(container);
-            verify(container).setObservationEnabled(true);
-        });
-    }
-
-}