Merge pull request #4 from CatoTH/integrate-prometheus

Prometheus integration

.env.tpl

@@ -0,0 +1,5 @@
+ANTRAGSGRUEN_INSTALLATIONS_0_ID=std
+ANTRAGSGRUEN_INSTALLATIONS_0_PUBLIC_KEY=MII...
+ANTRAGSGRUEN_WS_ORIGINS=http://*.antragsgruen.test
+ACTUATOR_USER=admin
+ACTUATOR_PASSWORD=admin

.gitignore

@@ -4,7 +4,10 @@ target/
 !**/src/test/**/target/
 
 src/main/resources/static/stomp.umd.min.js
+src/main/resources/public.key
+docker/prometheus/prometheus.yml
 node_modules/
+.env
 
 ### IntelliJ IDEA ###
 .idea

README.md

@@ -115,13 +115,20 @@ Compiling and running:
 
 ### Running with Docker (JRE)
 
-A dummy docker-compose.yml is provided that builds and runs the application.
+A dummy docker-compose.yml is provided that builds and runs the application. To set it up:
+- copy [.env.tpl](.env.tpl) to `.env` and modify environment variables to your needs. In particular, enter the JWT private key.
+- copy [prometheus.demo.yml](docker/prometheus/prometheus.demo.yml) to `prometheus.yml`. Set the credentials in there to the same as `ACTUATOR_USER` / `ACTUATOR_PASSWORD` in `.env`.
 
 ```shell
 docker compose -f docker-compose.jdk.yml build
 docker compose -f docker-compose.jdk.yml up
 ```
 
+This will expose services:
+- http://localhost:8080/ : The main webservice application (not meant to be accessed directly)
+- http://localhost:3000/ : Grafana, to access Prometheus logs. (Grafana will not be configured at all. So to access the metrics, it will be necessary to set up a Prometheus datasource pointing at `http://prometheus:9090`.)
+
+
 ## Testing
 
 ### Running spotbugs && checkstyle

docker-compose.jdk.yml

@@ -5,15 +5,14 @@ services:
       dockerfile: docker/jdk/Dockerfile
       #target: build - enable this to use JDK image
     environment:
-      - ANTRAGSGRUEN_WS_ORIGINS=http://*.antragsgruen.test
       - RABBITMQ_HOST=rabbitmq
-      - ACTUATOR_USER=admin
-      - ACTUATOR_PASSWORD=admin
+    env_file: .env
     depends_on:
       rabbitmq:
         condition: service_healthy
     ports:
       - 127.0.0.1:8080:8080
+
   rabbitmq:
     image: 'rabbitmq:3-management-alpine'
     environment:
@@ -26,4 +25,31 @@ services:
       test: rabbitmq-diagnostics -q ping
       interval: 5s
       timeout: 10s
-      retries: 3
+      retries: 3
+
+  prometheus:
+    image: prom/prometheus:v2.26.1
+    #ports:
+    #  - 9090:9090
+    volumes:
+      - prometheus-data:/prometheus
+      - ./docker/prometheus:/etc/prometheus
+    environment:
+      - ACTUATOR_USER=admin
+      - ACTUATOR_PASSWORD=admin
+    command: --config.file=/etc/prometheus/prometheus.yml
+    depends_on:
+      - live
+
+  grafana:
+    image: grafana/grafana:7.5.6
+    ports:
+      - 3000:3000
+    volumes:
+      - grafana-data:/var/lib/grafana
+    depends_on:
+      - prometheus
+
+volumes:
+  prometheus-data:
+  grafana-data:

docker/prometheus/prometheus.demo.yml

@@ -0,0 +1,18 @@
+global:
+  scrape_interval: 10s
+
+scrape_configs:
+  - job_name: 'prometheus'
+    static_configs:
+      - targets:
+        - 'localhost:9090'
+
+  - job_name: 'antragsgruen_live'
+    metrics_path: /actuator/prometheus
+    # https://github.com/prometheus/prometheus/issues/10554 - no support for environment variables
+    basic_auth:
+      username: admin
+      password: admin
+    static_configs:
+      - targets:
+          - 'live:8080'

pom.xml

@@ -5,7 +5,7 @@
 	<parent>
 		<groupId>org.springframework.boot</groupId>
 		<artifactId>spring-boot-starter-parent</artifactId>
-		<version>3.3.4</version>
+		<version>3.3.5</version>
 		<relativePath/> <!-- lookup parent from repository -->
 	</parent>
 	<groupId>de.antragsgruen</groupId>
@@ -29,7 +29,7 @@
 		<dependency>
 			<groupId>org.projectlombok</groupId>
 			<artifactId>lombok</artifactId>
-			<version>1.18.34</version>
+			<version>1.18.36</version>
 			<scope>provided</scope>
 		</dependency>
 		<dependency>
@@ -40,6 +40,11 @@
 			<groupId>org.springframework.boot</groupId>
 			<artifactId>spring-boot-starter-actuator</artifactId>
 		</dependency>
+		<dependency>
+			<groupId>io.micrometer</groupId>
+			<artifactId>micrometer-registry-prometheus</artifactId>
+			<scope>runtime</scope>
+		</dependency>
 
 		<!-- Test dependencies -->
 		<dependency>
@@ -50,7 +55,7 @@
 		<dependency>
 			<groupId>net.javacrumbs.json-unit</groupId>
 			<artifactId>json-unit-assertj</artifactId>
-			<version>3.4.1</version>
+			<version>3.5.0</version>
 			<scope>test</scope>
 		</dependency>
 	</dependencies>

spotbugs-ignore.xml

@@ -10,6 +10,7 @@
         <Or>
             <Package name="~de\.antragsgruen\.live\.rabbitmq\.dto.*"/>
             <Package name="~de\.antragsgruen\.live\.websocket\.dto.*"/>
+            <Package name="~de\.antragsgruen\.live\.metrics.*"/>
         </Or>
     </Match>
 

src/main/java/de/antragsgruen/live/LiveApplication.java

@@ -2,8 +2,10 @@
 
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.scheduling.annotation.EnableScheduling;
 
 @SpringBootApplication()
+@EnableScheduling
 public class LiveApplication {
 
     public static void main(String[] args) {

src/main/java/de/antragsgruen/live/metrics/ActiveWebsocketConnectionMetric.java

@@ -0,0 +1,48 @@
+package de.antragsgruen.live.metrics;
+
+import de.antragsgruen.live.multisite.ConsultationScope;
+import de.antragsgruen.live.websocket.TopicPermissionChecker;
+import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.Tag;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.messaging.simp.user.SimpUserRegistry;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Service;
+
+import java.util.List;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+@Service
+@RequiredArgsConstructor
+@Slf4j
+public class ActiveWebsocketConnectionMetric {
+    private final SimpUserRegistry userRegistry;
+    private final MeterRegistry registry;
+
+    private static final String METRIC_NAME = "antragsgruen_live.ws.active_connections";
+
+    @Scheduled(fixedRateString = "${antragsgruen.metrics.interval.ms}")
+    public void processGauge() {
+        userRegistry.findSubscriptions(subscription -> true)
+                .stream()
+                .map(subscription -> TopicPermissionChecker.consultationScopeFromPathParts(subscription.getDestination().split("/")))
+                .collect(Collectors.groupingBy(Function.identity(), Collectors.counting()))
+                .forEach(this::trackMetric);
+    }
+
+    private void trackMetric(ConsultationScope scope, Long subscribers) {
+        log.debug("Logging active websocket metrics: " + scope + " - " + subscribers);
+
+        registry.gauge(
+                METRIC_NAME,
+                List.of(
+                        Tag.of("installation", scope.installation()),
+                        Tag.of("site", scope.site()),
+                        Tag.of("consultation", scope.consultation())
+                ),
+                subscribers
+        );
+    }
+}

src/main/java/de/antragsgruen/live/metrics/ReceivedRabbitMQMessagesMetric.java

@@ -0,0 +1,41 @@
+package de.antragsgruen.live.metrics;
+
+import de.antragsgruen.live.multisite.ConsultationScope;
+import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.Tag;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+
+import java.util.List;
+
+@Service
+@RequiredArgsConstructor
+@Slf4j
+public class ReceivedRabbitMQMessagesMetric {
+    private final MeterRegistry registry;
+
+    private static final String METRIC_NAME = "antragsgruen_live.rabbitmq.msg_count";
+
+    private static final String TYPE_SPEECH = "speech";
+    private static final String TYPE_USER = "user";
+
+    public void onSpeechEvent(ConsultationScope scope) {
+        this.receivedMessage(scope, TYPE_SPEECH);
+    }
+
+    public void onUserEvent(ConsultationScope scope) {
+        this.receivedMessage(scope, TYPE_USER);
+    }
+
+    private void receivedMessage(ConsultationScope scope, String type) {
+        log.debug("Logging RabbitMQ message count: " + scope + " - " + type);
+
+        registry.counter(METRIC_NAME, List.of(
+                Tag.of("installation", scope.installation()),
+                Tag.of("site", scope.site()),
+                Tag.of("consultation", scope.consultation()),
+                Tag.of("type", type)
+        )).increment();
+    }
+}

src/main/java/de/antragsgruen/live/metrics/package-info.java

@@ -0,0 +1,6 @@
+@NonNullApi
+@NonNullFields
+package de.antragsgruen.live.metrics;
+
+import org.springframework.lang.NonNullApi;
+import org.springframework.lang.NonNullFields;

src/main/java/de/antragsgruen/live/rabbitmq/SpeechMessageReceiver.java

@@ -1,6 +1,7 @@
 package de.antragsgruen.live.rabbitmq;
 
 import de.antragsgruen.live.SpeechAdminHandler;
+import de.antragsgruen.live.metrics.ReceivedRabbitMQMessagesMetric;
 import de.antragsgruen.live.multisite.ConsultationScope;
 import de.antragsgruen.live.rabbitmq.dto.MQSpeechQueue;
 import de.antragsgruen.live.SpeechUserHandler;
@@ -23,6 +24,7 @@ public final class SpeechMessageReceiver {
 
     @NonNull private SpeechUserHandler speechUserHandler;
     @NonNull private SpeechAdminHandler speechAdminHandler;
+    @NonNull private ReceivedRabbitMQMessagesMetric receivedRabbitMQMessagesMetric;
 
     @RabbitListener(queues = {"${antragsgruen.rabbitmq.queue.speech}"})
     public void receiveMessage(MQSpeechQueue event, @Header(AmqpHeaders.RECEIVED_ROUTING_KEY) String routingKey) {
@@ -37,6 +39,7 @@ public void receiveMessage(MQSpeechQueue event, @Header(AmqpHeaders.RECEIVED_ROU
                 routingKeyParts[RK_PART_CONSULTATION]
         );
 
+        receivedRabbitMQMessagesMetric.onSpeechEvent(scope);
         speechUserHandler.onSpeechEvent(scope, event);
         speechAdminHandler.onSpeechEvent(scope, event);
     }

src/main/java/de/antragsgruen/live/rabbitmq/UserMessageReceiver.java

@@ -1,5 +1,6 @@
 package de.antragsgruen.live.rabbitmq;
 
+import de.antragsgruen.live.metrics.ReceivedRabbitMQMessagesMetric;
 import de.antragsgruen.live.multisite.ConsultationScope;
 import de.antragsgruen.live.rabbitmq.dto.MQUserEvent;
 import de.antragsgruen.live.websocket.Sender;
@@ -23,6 +24,7 @@ public final class UserMessageReceiver {
     private static final int RK_PART_USER = 4;
 
     @NonNull private Sender sender;
+    @NonNull private ReceivedRabbitMQMessagesMetric receivedRabbitMQMessagesMetric;
 
     @RabbitListener(queues = {"${antragsgruen.rabbitmq.queue.user}"})
     public void receiveMessage(MQUserEvent event, @Header(AmqpHeaders.RECEIVED_ROUTING_KEY) String routingKey) {
@@ -37,8 +39,9 @@ public void receiveMessage(MQUserEvent event, @Header(AmqpHeaders.RECEIVED_ROUTI
                 routingKeyParts[RK_PART_CONSULTATION]
         );
 
-        log.warn("Received user message: " + routingKey + " => " + event.username());
+        log.debug("Received user message: " + routingKey + " => " + event.username());
 
+        receivedRabbitMQMessagesMetric.onUserEvent(scope);
         sender.sendToUser(scope, routingKeyParts[RK_PART_USER], Sender.ROLE_USER, Sender.USER_CHANNEL_DEFAULT, event.username());
     }
 }

src/main/java/de/antragsgruen/live/websocket/TopicPermissionChecker.java

@@ -14,7 +14,7 @@
 public class TopicPermissionChecker {
     private static final String ROLE_SPEECH_ADMIN = "ROLE_SPEECH_ADMIN";
 
-    public static final int USER_PARTS_LENGTH = 7;
+    public static final int USER_PARTS_LENGTH = 7; // Also used for /admin/ topics
     public static final int USER_PART_ROLE = 1;
     public static final int USER_PART_INSTALLATION = 2;
     public static final int USER_PART_SITE = 3;
@@ -44,25 +44,33 @@ public boolean canSubscribeToDestination(JwtAuthenticationToken jwtToken, @Nulla
             return false;
         }
 
-        ConsultationScope scope = null;
-        boolean additionalPermissionsPassed = true;
+        ConsultationScope scope = TopicPermissionChecker.consultationScopeFromPathParts(pathParts);
 
-        if ("topic".equals(pathParts[TOPIC_PART_TOPIC]) && pathParts.length == TOPIC_PARTS_LENGTH) {
-            scope = new ConsultationScope(pathParts[TOPIC_PART_INSTALLATION], pathParts[TOPIC_PART_SITE], pathParts[TOPIC_PART_CONSULTATION]);
-        }
+        boolean additionalPermissionsPassed = true;
         if (Sender.ROLE_USER.equals(pathParts[USER_PART_ROLE]) && pathParts.length == USER_PARTS_LENGTH) {
-            scope = new ConsultationScope(pathParts[USER_PART_INSTALLATION], pathParts[USER_PART_SITE], pathParts[USER_PART_CONSULTATION]);
             additionalPermissionsPassed = pathParts[USER_PART_USER].equals(jwtToken.getName());
         }
         if (Sender.ROLE_ADMIN.equals(pathParts[USER_PART_ROLE]) && pathParts.length == USER_PARTS_LENGTH) {
-            scope = new ConsultationScope(pathParts[USER_PART_INSTALLATION], pathParts[USER_PART_SITE], pathParts[USER_PART_CONSULTATION]);
             additionalPermissionsPassed = jwtHasRoleForTopic(jwtToken, pathParts[USER_PART_MODULE])
                     && pathParts[USER_PART_USER].equals(jwtToken.getName());
         }
 
         return (scope != null && jwtIsForCorrectConsultation(jwtToken, scope) && additionalPermissionsPassed);
     }
 
+    public static ConsultationScope consultationScopeFromPathParts(String[] pathParts) {
+        if ("topic".equals(pathParts[TOPIC_PART_TOPIC]) && pathParts.length == TOPIC_PARTS_LENGTH) {
+            return new ConsultationScope(pathParts[TOPIC_PART_INSTALLATION], pathParts[TOPIC_PART_SITE], pathParts[TOPIC_PART_CONSULTATION]);
+        }
+        if (Sender.ROLE_USER.equals(pathParts[USER_PART_ROLE]) && pathParts.length == USER_PARTS_LENGTH) {
+            return new ConsultationScope(pathParts[USER_PART_INSTALLATION], pathParts[USER_PART_SITE], pathParts[USER_PART_CONSULTATION]);
+        }
+        if (Sender.ROLE_ADMIN.equals(pathParts[USER_PART_ROLE]) && pathParts.length == USER_PARTS_LENGTH) {
+            return new ConsultationScope(pathParts[USER_PART_INSTALLATION], pathParts[USER_PART_SITE], pathParts[USER_PART_CONSULTATION]);
+        }
+        return null;
+    }
+
     private boolean jwtIsForCorrectConsultation(JwtAuthenticationToken jwtToken, ConsultationScope scope) {
         Object payload = jwtToken.getTokenAttributes().get("payload");
         if (!(payload instanceof Map<?, ?> payloadMap)) {