add traceid + metics: zipkin + prometheus, integrate it into cq invoc…

NotificationsSystemDocker/compose.yaml

@@ -28,6 +28,21 @@ services:
       - "1025:1025"
     networks:
       - notifications-network
+  zipkin:
+    image: openzipkin/zipkin:latest
+    container_name: zipkin-server
+    ports:
+      - "9411:9411"
+    networks:
+      - notifications-network
+  prometheus:
+    image: prom/prometheus:latest
+    ports:
+      - "9090:9090"
+    networks:
+      - notifications-network
+    volumes:
+      - ./prometheus.yml:/etc/prometheus/prometheus.yml
 
 networks:
   notifications-network:

NotificationsSystemDocker/prometheus.yml

@@ -0,0 +1,8 @@
+global:
+  scrape_interval: 15s
+
+scrape_configs:
+  - job_name: "notifications-system"
+    metrics_path: "/actuator/prometheus"
+    static_configs:
+      - targets: ["backend:8080"]

NotificationsSystemInfrastructure/pom.xml

@@ -53,6 +53,30 @@
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-mail</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-actuator</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-micrometer-tracing-brave</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.micrometer</groupId>
+            <artifactId>micrometer-tracing-bridge-brave</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-zipkin</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>net.ttddyy.observation</groupId>
+            <artifactId>datasource-micrometer-spring-boot</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.micrometer</groupId>
+            <artifactId>micrometer-registry-prometheus</artifactId>
+        </dependency>
 
         <dependency>
             <groupId>org.projectlombok</groupId>

NotificationsSystemInfrastructure/src/main/java/pl/marcinsobanski/notificationssystem/infrastructure/adapters/cqrs/CQCommandHandlerImpl.java

@@ -1,87 +1,61 @@
 package pl.marcinsobanski.notificationssystem.infrastructure.adapters.cqrs;
 
+import io.micrometer.observation.Observation;
+import io.micrometer.observation.ObservationRegistry;
+import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.commons.lang3.reflect.TypeUtils;
 import org.springframework.stereotype.Service;
 import pl.marcinsobanski.notificationssystem.api.cqrs.common.cqrs.Command;
-import pl.marcinsobanski.notificationssystem.api.cqrs.common.cqrs.CommandHandler;
 import pl.marcinsobanski.notificationssystem.api.cqrs.common.cqrs.Query;
-import pl.marcinsobanski.notificationssystem.api.cqrs.common.cqrs.QueryHandler;
 import pl.marcinsobanski.notificationssystem.application.cqrs.CQCommandHandler;
 
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
 @Service
 @Slf4j
-// TODO remove duplication
+@RequiredArgsConstructor
 public class CQCommandHandlerImpl implements CQCommandHandler {
 
-    private final Map<Class<Command<?>>, CommandHandler<?, ?>> commandHandlerMap;
-    private final Map<Class<Query<?>>, QueryHandler<?, ?>> queryHandlerMap;
-
-    public CQCommandHandlerImpl(
-            List<CommandHandler<?, ?>> commandHandlerBeans,
-            List<QueryHandler<?, ?>> queryHandlerBeans
-    ) {
-        final var commandHandlers = new HashMap<Class<Command<?>>, CommandHandler<?, ?>>();
-        for (CommandHandler<?, ?> commandHandlerBean : commandHandlerBeans) {
-            final var typeArguments = TypeUtils.getTypeArguments(commandHandlerBean.getClass(), CommandHandler.class);
-            final var commandClazz = (Class) typeArguments.entrySet().stream()
-                    .filter(typeVariableTypeEntry -> "COMMAND".equals(typeVariableTypeEntry.getKey().getName()))
-                    .map(Map.Entry::getValue)
-                    .findFirst()
-                    .orElseThrow();
-            commandHandlers.put(commandClazz, commandHandlerBean);
-        }
-        commandHandlerMap = Collections.unmodifiableMap(commandHandlers);
-
-        final var queryHandlers = new HashMap<Class<Query<?>>, QueryHandler<?, ?>>();
-        for (QueryHandler<?, ?> queryHandlerBean : queryHandlerBeans) {
-            final var typeArguments = TypeUtils.getTypeArguments(queryHandlerBean.getClass(), QueryHandler.class);
-            final var queryClazz = (Class) typeArguments.entrySet().stream()
-                    .filter(typeVariableTypeEntry -> "QUERY".equals(typeVariableTypeEntry.getKey().getName()))
-                    .map(Map.Entry::getValue)
-                    .findFirst()
-                    .orElseThrow();
-            queryHandlers.put(queryClazz, queryHandlerBean);
-        }
-        queryHandlerMap = Collections.unmodifiableMap(queryHandlers);
-    }
+    private final CQHandlersProvider cqHandlersProvider;
+    private final ObservationRegistry observationRegistry;
 
     @Override
     public <RESULT> RESULT executeCommand(Command<RESULT> command) {
-        final var commandExecutor = (CommandHandler<RESULT, Command<RESULT>>) commandHandlerMap.get(command.getClass());
+        final var commandExecutor = cqHandlersProvider.getCommandHandler(command);
         if (commandExecutor == null) {
             throw new IllegalArgumentException("Command handler for class " + command.getClass() + " not exist");
         }
-        log.info("Execute command {}", command);
-        try {
-            final var result = commandExecutor.handle(command);
-            log.info("Command {} executed: {}", command, result);
-            return result;
-        } catch (Exception e) {
-            log.error("Command {} execution failed", command, e);
-            throw e;
-        }
+        return Observation.createNotStarted("execute command", observationRegistry)
+                .lowCardinalityKeyValue("command.class", command.getClass().getSimpleName())
+                .observe(() -> {
+                    log.info("Execute command {}", command);
+                    try {
+                        final var result = commandExecutor.handle(command);
+                        log.info("Command {} executed: {}", command, result);
+                        return result;
+                    } catch (Exception e) {
+                        log.error("Command {} execution failed", command, e);
+                        throw e;
+                    }
+                });
     }
 
     @Override
     public <RESULT> RESULT executeQuery(Query<RESULT> query) {
-        final var queryExecutor = (QueryHandler<RESULT, Query<RESULT>>) queryHandlerMap.get(query.getClass());
+        final var queryExecutor = cqHandlersProvider.getQueryHandler(query);
         if (queryExecutor == null) {
             throw new IllegalArgumentException("Query handler for class " + query.getClass() + " not exist");
         }
-        log.info("Execute query {}", query);
-        try {
-            final var result = queryExecutor.handle(query);
-            log.info("Query {} executed: {}", query, result);
-            return result;
-        } catch (Exception e) {
-            log.error("Query {} execution failed", query, e);
-            throw e;
-        }
+        return Observation.createNotStarted("execute query", observationRegistry)
+                .lowCardinalityKeyValue("query.class", query.getClass().getName())
+                .observe(() -> {
+                    log.info("Execute query {}", query);
+                    try {
+                        final var result = queryExecutor.handle(query);
+                        log.info("Query {} executed: {}", query, result);
+                        return result;
+                    } catch (Exception e) {
+                        log.error("Query {} execution failed", query, e);
+                        throw e;
+                    }
+                });
     }
 }

NotificationsSystemInfrastructure/src/main/java/pl/marcinsobanski/notificationssystem/infrastructure/adapters/cqrs/CQHandlersProvider.java

@@ -0,0 +1,51 @@
+package pl.marcinsobanski.notificationssystem.infrastructure.adapters.cqrs;
+
+import org.apache.commons.lang3.reflect.TypeUtils;
+import org.springframework.stereotype.Service;
+import pl.marcinsobanski.notificationssystem.api.cqrs.common.cqrs.Command;
+import pl.marcinsobanski.notificationssystem.api.cqrs.common.cqrs.CommandHandler;
+import pl.marcinsobanski.notificationssystem.api.cqrs.common.cqrs.Query;
+import pl.marcinsobanski.notificationssystem.api.cqrs.common.cqrs.QueryHandler;
+
+import java.util.*;
+
+@Service
+public class CQHandlersProvider {
+
+    private final Map<Class<Command<?>>, CommandHandler<?, ?>> commandHandlerMap;
+    private final Map<Class<Query<?>>, QueryHandler<?, ?>> queryHandlerMap;
+
+    public CQHandlersProvider(
+            List<CommandHandler<?, ?>> commandHandlerBeans,
+            List<QueryHandler<?, ?>> queryHandlerBeans
+    ) {
+        final var commandHandlers = new HashMap<Class<Command<?>>, CommandHandler<?, ?>>();
+        commandHandlerBeans.forEach(commandHandlerBean -> commandHandlers.put((Class<Command<?>>) extract(commandHandlerBean, CommandHandler.class, "COMMAND"), commandHandlerBean));
+        commandHandlerMap = Collections.unmodifiableMap(commandHandlers);
+
+
+        final var queryHandlers = new HashMap<Class<Query<?>>, QueryHandler<?, ?>>();
+        queryHandlerBeans.forEach(queryHandlerBean -> queryHandlers.put((Class<Query<?>>) extract(queryHandlerBean, QueryHandler.class, "QUERY"), queryHandlerBean));
+        queryHandlerMap = Collections.unmodifiableMap(queryHandlers);
+    }
+
+    private Class<?> extract(Object cqBean, Class<?> cqHandlerClazz, String argument) {
+        final var typeArguments = TypeUtils.getTypeArguments(cqBean.getClass(), cqHandlerClazz);
+        return (Class) typeArguments.entrySet().stream()
+                .filter(typeVariableTypeEntry -> argument.equals(typeVariableTypeEntry.getKey().getName()))
+                .map(Map.Entry::getValue)
+                .findFirst()
+                .orElseThrow();
+    }
+
+    public <R, C extends Command<R>> CommandHandler<R, C> getCommandHandler(C command) {
+        final var clazz = command.getClass();
+        return (CommandHandler<R, C>) commandHandlerMap.get(clazz);
+    }
+
+    public <R, Q extends Query<R>> QueryHandler<R, Q> getQueryHandler(Q query) {
+        final var clazz = query.getClass();
+        return (QueryHandler<R, Q>) queryHandlerMap.get(clazz);
+    }
+
+}

NotificationsSystemInfrastructure/src/main/resources/application-docker.yaml

@@ -6,4 +6,15 @@ spring:
       ddl-auto: update
   mail:
     host: mailpit
-    port: 1025
+    port: 1025
+management:
+  endpoints:
+    web:
+      exposure:
+        include: health, prometheus
+  tracing:
+    sampling:
+      probability: 1.0
+    export:
+      zipkin:
+        endpoint: http://zipkin:9411/api/v2/spans

README.md

@@ -40,6 +40,8 @@ Url:
 * h2-console http://localhost:8080/h2-console
     * JDBC URL `jdbc:h2:file:/data/h2db`
     * user `admin`
+* zipkin http://localhost:9411
+* prometheus http://localhost:9090
 
 Important:
 

pom.xml

@@ -60,13 +60,21 @@
                 <artifactId>springdoc-openapi-starter-webmvc-ui</artifactId>
                 <version>3.0.1</version>
             </dependency>
+
             <dependency>
                 <groupId>org.springframework.boot</groupId>
                 <artifactId>spring-boot-dependencies</artifactId>
                 <version>4.0.2</version>
                 <type>pom</type>
                 <scope>import</scope>
             </dependency>
+            <dependency>
+                <groupId>net.ttddyy.observation</groupId>
+                <artifactId>datasource-micrometer-bom</artifactId>
+                <version>2.1.1</version>
+                <type>pom</type>
+                <scope>import</scope>
+            </dependency>
         </dependencies>
     </dependencyManagement>