Merge pull request #628 from swisspost/develop

PR for release
swisspost · Jan 13, 2025 · fdb311e · fdb311e
2 parents 9bc4263 + 6be58ea
commit fdb311e
Show file tree

Hide file tree

Showing 44 changed files with 484 additions and 224 deletions.
diff --git a/gateleen-cache/pom.xml b/gateleen-cache/pom.xml
@@ -6,7 +6,7 @@
     <parent>
         <groupId>org.swisspush.gateleen</groupId>
         <artifactId>gateleen</artifactId>
-        <version>2.1.14-SNAPSHOT</version>
+        <version>2.1.15-SNAPSHOT</version>
     </parent>
 
     <artifactId>gateleen-cache</artifactId>

diff --git a/gateleen-core/pom.xml b/gateleen-core/pom.xml
@@ -6,7 +6,7 @@
     <parent>
         <groupId>org.swisspush.gateleen</groupId>
         <artifactId>gateleen</artifactId>
-        <version>2.1.14-SNAPSHOT</version>
+        <version>2.1.15-SNAPSHOT</version>
     </parent>
 
     <artifactId>gateleen-core</artifactId>

diff --git a/gateleen-delegate/pom.xml b/gateleen-delegate/pom.xml
@@ -6,7 +6,7 @@
     <parent>
         <groupId>org.swisspush.gateleen</groupId>
         <artifactId>gateleen</artifactId>
-        <version>2.1.14-SNAPSHOT</version>
+        <version>2.1.15-SNAPSHOT</version>
     </parent>
 
     <artifactId>gateleen-delegate</artifactId>

diff --git a/gateleen-delta/pom.xml b/gateleen-delta/pom.xml
@@ -6,7 +6,7 @@
     <parent>
         <groupId>org.swisspush.gateleen</groupId>
         <artifactId>gateleen</artifactId>
-        <version>2.1.14-SNAPSHOT</version>
+        <version>2.1.15-SNAPSHOT</version>
     </parent>
 
     <artifactId>gateleen-delta</artifactId>

diff --git a/gateleen-expansion/pom.xml b/gateleen-expansion/pom.xml
@@ -6,7 +6,7 @@
     <parent>
         <groupId>org.swisspush.gateleen</groupId>
         <artifactId>gateleen</artifactId>
-        <version>2.1.14-SNAPSHOT</version>
+        <version>2.1.15-SNAPSHOT</version>
     </parent>
 
     <artifactId>gateleen-expansion</artifactId>

diff --git a/gateleen-hook-js/pom.xml b/gateleen-hook-js/pom.xml
@@ -4,7 +4,7 @@
     <parent>
         <groupId>org.swisspush.gateleen</groupId>
         <artifactId>gateleen</artifactId>
-        <version>2.1.14-SNAPSHOT</version>
+        <version>2.1.15-SNAPSHOT</version>
     </parent>
     <artifactId>gateleen-hook-js</artifactId>
     <packaging>jar</packaging>

diff --git a/gateleen-hook/pom.xml b/gateleen-hook/pom.xml
@@ -6,7 +6,7 @@
     <parent>
         <groupId>org.swisspush.gateleen</groupId>
         <artifactId>gateleen</artifactId>
-        <version>2.1.14-SNAPSHOT</version>
+        <version>2.1.15-SNAPSHOT</version>
     </parent>
 
     <artifactId>gateleen-hook</artifactId>

diff --git a/gateleen-hook/src/main/java/org/swisspush/gateleen/hook/HookHandler.java b/gateleen-hook/src/main/java/org/swisspush/gateleen/hook/HookHandler.java
@@ -1568,7 +1568,7 @@ private void extractAndAddStaticHeadersToHook(final JsonObject jsonHook, final H
         // For backward compatibility we still parse the old "staticHeaders" - but now create a manipulator chain accordingly
         JsonObject staticHeaders = jsonHook.getJsonObject(STATIC_HEADERS);
         if (staticHeaders != null) {
-            log.warn("you use the deprecated \"staticHeaders\" syntax in your hook ({}). Please migrate to the more flexible \"headers\" syntax", jsonHook);
+            log.info("you use the deprecated \"staticHeaders\" syntax in your hook ({}). Please migrate to the more flexible \"headers\" syntax", jsonHook);
             hook.setHeaderFunction(HeaderFunctions.parseStaticHeadersFromJson(staticHeaders));
         }
     }

diff --git a/gateleen-kafka/README_kafka.md b/gateleen-kafka/README_kafka.md
@@ -48,7 +48,7 @@ The following topic configuration values are required:
 Besides these required configuration values, additional string values can be added. See documentation from Apache Kafka [here](https://kafka.apache.org/documentation/#producerconfigs).
 
 ## Usage
-To use the gateleen-kafka module, the [KafkaHandler](src/main/java/org/swisspush/gateleen/kafka/KafkaHandler.java) class has to be initialized as described in the _configuration_ section. Also the [KafkaHandler](src/main/java/org/swisspush/gateleen/kafka/KafkaHandler.java) has to be integrated in the "MainVerticle" handling all
+To use the gateleen-kafka module, the [KafkaHandler](src/main/java/org/swisspush/gateleen/kafka/KafkaHandler.java) class has to be initialized as described in the _configuration_ section. Also, the [KafkaHandler](src/main/java/org/swisspush/gateleen/kafka/KafkaHandler.java) has to be integrated in the "MainVerticle" handling all
 incoming requests. See [Playground Server](../gateleen-playground/src/main/java/org/swisspush/gateleen/playground/Server.java) and [Runconfig](../gateleen-runconfig/src/main/java/org/swisspush/gateleen/runconfig/RunConfig.java).
 
 The following sequence diagram shows the setup of the "MainVerticle". The `streamingPath` (KafkaHandler) is configured to `/playground/server/streaming/`
@@ -448,4 +448,30 @@ This sequence diagrams shows the process when messages are sent to Kafka:
                           │ <────────────────────│ │                           │                                │                                   │                                      │                                   
                           │                      │ │                           │                                │                                   │                                      │                                   
                           │                      └┬┘                           │                                │                                   │                                      │                                   
-```
+```
+
+### Micrometer metrics
+The kafka feature is monitored with micrometer. The following metrics are available:
+* gateleen_kafka_send_success_messages_total
+* gateleen_kafka_send_fail_messages_total
+* gateleen_kafka_validation_fail_messages_total
+
+Additional tags are provided to specify the topic.
+
+Example metrics:
+
+```
+# HELP gateleen_kafka_send_success_messages_total Amount of successfully sent kafka messages
+# TYPE gateleen_kafka_send_success_messages_total counter
+gateleen_kafka_send_success_messages_total{topic="my-topic-1",} 0.0
+gateleen_kafka_send_fail_messages_total{topic="my-topic-1",} 455.0
+gateleen_kafka_send_success_messages_total{topic="my-topic-2",} 256.0
+gateleen_kafka_send_success_messages_total{topic="my-topic-3",} 6.0
+gateleen_kafka_send_fail_messages_total{topic="my-topic-4",} 222.0
+# HELP gateleen_kafka_validation_fail_messages_total Amount of failed kafka message validations
+# TYPE gateleen_kafka_validation_fail_messages_total counter
+gateleen_kafka_validation_fail_messages_total{topic="my-topic-6",} 212.0
+```
+
+To enable `gateleen_kafka_send_success_messages_total` and `gateleen_kafka_send_fail_messages_total` metrics, set a `MeterRegistry` instance by calling `setMeterRegistry(MeterRegistry meterRegistry)` method in `KafkaMessageSender` class.
+To enable `gateleen_kafka_validation_fail_messages_total` metrics, set a `MeterRegistry` instance by calling `setMeterRegistry(MeterRegistry meterRegistry)` method in `KafkaMessageValidator` class.
diff --git a/gateleen-kafka/pom.xml b/gateleen-kafka/pom.xml
@@ -6,7 +6,7 @@
     <parent>
         <groupId>org.swisspush.gateleen</groupId>
         <artifactId>gateleen</artifactId>
-        <version>2.1.14-SNAPSHOT</version>
+        <version>2.1.15-SNAPSHOT</version>
     </parent>
 
     <artifactId>gateleen-kafka</artifactId>
@@ -22,6 +22,10 @@
             <artifactId>gateleen-validation</artifactId>
             <version>${project.version}</version>
         </dependency>
+        <dependency>
+            <groupId>io.micrometer</groupId>
+            <artifactId>micrometer-core</artifactId>
+        </dependency>
         <!-- TEST dependencies -->
         <dependency>
             <groupId>org.swisspush.gateleen</groupId>

diff --git a/gateleen-kafka/src/main/java/org/swisspush/gateleen/kafka/KafkaHandler.java b/gateleen-kafka/src/main/java/org/swisspush/gateleen/kafka/KafkaHandler.java
@@ -21,14 +21,11 @@
 import org.swisspush.gateleen.core.validation.ValidationStatus;
 import org.swisspush.gateleen.validation.ValidationException;
 
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.regex.Pattern;
 
-import static org.swisspush.gateleen.core.exception.GateleenExceptionFactory.newGateleenThriftyExceptionFactory;
-
 /**
  * Handler class for all Kafka related requests.
  *
@@ -52,46 +49,10 @@ public class KafkaHandler extends ConfigurationResourceConsumer {
     private final KafkaMessageSender kafkaMessageSender;
     private final Map<String, Object> properties;
     private final KafkaProducerRecordBuilder kafkaProducerRecordBuilder;
-    private KafkaMessageValidator kafkaMessageValidator;
+    private final KafkaMessageValidator kafkaMessageValidator;
 
     private boolean initialized = false;
 
-    /** @deprecated Use {@link #builder()} */
-    @Deprecated
-    public KafkaHandler(ConfigurationResourceManager configurationResourceManager, KafkaProducerRepository repository,
-                        KafkaMessageSender kafkaMessageSender, String configResourceUri, String streamingPath) {
-        this(configurationResourceManager, null, repository, kafkaMessageSender,
-                configResourceUri, streamingPath);
-    }
-
-    /** @deprecated Use {@link #builder()} */
-    @Deprecated
-    public KafkaHandler(ConfigurationResourceManager configurationResourceManager, KafkaMessageValidator kafkaMessageValidator,
-                        KafkaProducerRepository repository, KafkaMessageSender kafkaMessageSender, String configResourceUri,
-                        String streamingPath) {
-        this(configurationResourceManager, kafkaMessageValidator, repository, kafkaMessageSender,
-                configResourceUri, streamingPath, new HashMap<>());
-    }
-
-    /** @deprecated Use {@link #builder()} */
-    @Deprecated
-    public KafkaHandler(ConfigurationResourceManager configurationResourceManager, KafkaProducerRepository repository,
-    KafkaMessageSender kafkaMessageSender, String configResourceUri, String streamingPath, Map<String, Object> properties) {
-
-        this(configurationResourceManager, null, repository, kafkaMessageSender,
-                configResourceUri, streamingPath, properties);
-    }
-
-    /** @deprecated Use {@link #builder()} */
-    @Deprecated
-    public KafkaHandler(ConfigurationResourceManager configurationResourceManager, KafkaMessageValidator kafkaMessageValidator, KafkaProducerRepository repository,
-                        KafkaMessageSender kafkaMessageSender, String configResourceUri, String streamingPath, Map<String, Object> properties) {
-        this(Vertx.vertx(), newGateleenThriftyExceptionFactory(), configurationResourceManager,
-            kafkaMessageValidator, repository, kafkaMessageSender, configResourceUri, streamingPath,
-            properties);
-        log.warn("TODO: Do NOT use this DEPRECATED constructor! It creates instances that it should not create!");
-    }
-
     public KafkaHandler(
         Vertx vertx,
         GateleenExceptionFactory exceptionFactory,
@@ -140,8 +101,6 @@ private Future<Void> initializeKafkaConfiguration(Buffer configuration) {
         Promise<Void> promise = Promise.promise();
         final List<KafkaConfiguration> kafkaConfigurations = KafkaConfigurationParser.parse(configuration, properties);
 
-
-
         repository.closeAll().future().onComplete((event -> {
             for (KafkaConfiguration kafkaConfiguration : kafkaConfigurations) {
                 repository.addKafkaProducer(kafkaConfiguration);

diff --git a/gateleen-kafka/src/main/java/org/swisspush/gateleen/kafka/KafkaMessageSender.java b/gateleen-kafka/src/main/java/org/swisspush/gateleen/kafka/KafkaMessageSender.java
@@ -1,5 +1,7 @@
 package org.swisspush.gateleen.kafka;
 
+import io.micrometer.core.instrument.Counter;
+import io.micrometer.core.instrument.MeterRegistry;
 import io.vertx.core.CompositeFuture;
 import io.vertx.core.Future;
 import io.vertx.core.Promise;
@@ -9,7 +11,9 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.function.Function;
 
 import static java.util.stream.Collectors.toList;
@@ -18,17 +22,32 @@ public class KafkaMessageSender {
 
     private static final Logger log = LoggerFactory.getLogger(KafkaMessageSender.class);
 
+    private MeterRegistry meterRegistry;
+    private final Map<String, Counter> successSendCounterMap = new HashMap<>();
+    private final Map<String, Counter> failSendCounterMap = new HashMap<>();
+
+    public static final String SUCCESS_SEND_MESSAGES_METRIC = "gateleen.kafka.send.success.messages";
+    public static final String SUCCESS_SEND_MESSAGES_METRIC_DESCRIPTION = "Amount of successfully sent kafka messages";
+    public static final String FAIL_SEND_MESSAGES_METRIC = "gateleen.kafka.send.fail.messages";
+    public static final String FAIL_SEND_MESSAGES_METRIC_DESCRIPTION = "Amount of failed kafka message sendings";
+    public static final String TOPIC = "topic";
+
+    public void setMeterRegistry(MeterRegistry meterRegistry) {
+        this.meterRegistry = meterRegistry;
+        successSendCounterMap.clear();
+        failSendCounterMap.clear();
+    }
+
     Future<Void> sendMessages(KafkaProducer<String, String> kafkaProducer,
                               List<KafkaProducerRecord<String, String>> messages) {
         Promise<Void> promise = Promise.promise();
         log.debug("Start processing {} messages for kafka", messages.size());
 
-        @SuppressWarnings("rawtypes") //https://github.com/eclipse-vertx/vert.x/issues/2627
-        List<Future> futures = messages.stream()
+        List<Future<Void>> futures = messages.stream()
                 .map(message -> KafkaMessageSender.this.sendMessage(kafkaProducer, message))
                 .collect(toList());
 
-        CompositeFuture.all(futures).<Void>mapEmpty().onComplete(result -> {
+        Future.all(futures).<Void>mapEmpty().onComplete(result -> {
             if (result.succeeded()) {
                 promise.complete();
                 log.debug("Batch messages successfully sent to Kafka.");
@@ -44,7 +63,45 @@ private Future<Void> sendMessage(KafkaProducer<String, String> kafkaProducer, Ka
         return kafkaProducer.send(message).compose((Function<RecordMetadata, Future<Void>>) metadata -> {
             log.debug("Message successfully sent to kafka topic '{}' on partition {} with offset {}. Timestamp: {}",
                     metadata.getTopic(), metadata.getPartition(), metadata.getOffset(), metadata.getTimestamp());
+            incrementSuccessCount(metadata.getTopic());
             return Future.succeededFuture();
-        }).onFailure(throwable -> log.warn("Failed to send message with key '{}' to kafka. Cause: {}", message.key(), throwable));
+        }).onFailure(throwable -> {
+            log.warn("Failed to send message with key '{}' to kafka. Cause: {}", message.key(), throwable);
+            incrementFailCount1(message.topic());
+        });
+    }
+
+    private synchronized void incrementSuccessCount(String topic) {
+        Counter counter = successSendCounterMap.get(topic);
+        if(counter != null) {
+            counter.increment();
+            return;
+        }
+
+        if(meterRegistry != null) {
+            Counter newCounter = Counter.builder(SUCCESS_SEND_MESSAGES_METRIC)
+                    .description(SUCCESS_SEND_MESSAGES_METRIC_DESCRIPTION)
+                    .tag(TOPIC, topic)
+                    .register(meterRegistry);
+            newCounter.increment();
+            successSendCounterMap.put(topic, newCounter);
+        }
+    }
+
+    private synchronized void incrementFailCount1(String topic) {
+        Counter counter = failSendCounterMap.get(topic);
+        if(counter != null) {
+            counter.increment();
+            return;
+        }
+
+        if(meterRegistry != null) {
+            Counter newCounter = Counter.builder(FAIL_SEND_MESSAGES_METRIC)
+                    .description(FAIL_SEND_MESSAGES_METRIC_DESCRIPTION)
+                    .tag(TOPIC, topic)
+                    .register(meterRegistry);
+            newCounter.increment();
+            failSendCounterMap.put(topic, newCounter);
+        }
     }
 }
diff --git a/gateleen-kafka/src/main/java/org/swisspush/gateleen/kafka/KafkaMessageValidator.java b/gateleen-kafka/src/main/java/org/swisspush/gateleen/kafka/KafkaMessageValidator.java
@@ -1,5 +1,7 @@
 package org.swisspush.gateleen.kafka;
 
+import io.micrometer.core.instrument.Counter;
+import io.micrometer.core.instrument.MeterRegistry;
 import io.vertx.core.CompositeFuture;
 import io.vertx.core.Future;
 import io.vertx.core.buffer.Buffer;
@@ -14,6 +16,7 @@
 import org.swisspush.gateleen.validation.ValidationUtil;
 import org.swisspush.gateleen.validation.Validator;
 
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -26,11 +29,23 @@ public class KafkaMessageValidator {
     private final Validator validator;
     private final Logger log = LoggerFactory.getLogger(KafkaHandler.class);
 
+    private MeterRegistry meterRegistry;
+    private final Map<String, Counter> failedToValidateCounterMap = new HashMap<>();
+
+    public static final String FAIL_VALIDATION_MESSAGES_METRIC = "gateleen.kafka.validation.fail.messages";
+    public static final String FAIL_VALIDATION_MESSAGES_METRIC_DESCRIPTION = "Amount of failed kafka message validations";
+    public static final String TOPIC = "topic";
+
     public KafkaMessageValidator(ValidationResourceManager validationResourceManager, Validator validator) {
         this.validationResourceManager = validationResourceManager;
         this.validator = validator;
     }
 
+    public void setMeterRegistry(MeterRegistry meterRegistry) {
+        this.meterRegistry = meterRegistry;
+        failedToValidateCounterMap.clear();
+    }
+
     public Future<ValidationResult> validateMessages(HttpServerRequest request, List<KafkaProducerRecord<String, String>> kafkaProducerRecords) {
         if (kafkaProducerRecords.isEmpty()) {
             return Future.succeededFuture(new ValidationResult(ValidationStatus.VALIDATED_POSITIV));
@@ -49,6 +64,8 @@ public Future<ValidationResult> validateMessages(HttpServerRequest request, List
 
         SchemaLocation schemaLocation = optionalSchemaLocation.get();
 
+        String topic = kafkaProducerRecords.get(0).topic();
+
         @SuppressWarnings("rawtypes") //https://github.com/eclipse-vertx/vert.x/issues/2627
         List<Future> futures = kafkaProducerRecords.stream()
                 .map(message -> validator.validateWithSchemaLocation(schemaLocation, Buffer.buffer(message.value()), log))
@@ -57,10 +74,31 @@ public Future<ValidationResult> validateMessages(HttpServerRequest request, List
         return CompositeFuture.all(futures).compose(compositeFuture -> {
             for (Object o : compositeFuture.list()) {
                 if (((ValidationResult) o).getValidationStatus() != ValidationStatus.VALIDATED_POSITIV) {
+                    incrementValidationFailCount(topic);
                     return Future.succeededFuture((ValidationResult) o);
                 }
             }
             return Future.succeededFuture(new ValidationResult(ValidationStatus.VALIDATED_POSITIV));
+        }, throwable -> {
+            incrementValidationFailCount(topic);
+            return Future.failedFuture(throwable);
         });
     }
+
+    private void incrementValidationFailCount(String topic) {
+        Counter counter = failedToValidateCounterMap.get(topic);
+        if(counter != null) {
+            counter.increment();
+            return;
+        }
+
+        if(meterRegistry != null) {
+            Counter newCounter = Counter.builder(FAIL_VALIDATION_MESSAGES_METRIC)
+                    .description(FAIL_VALIDATION_MESSAGES_METRIC_DESCRIPTION)
+                    .tag(TOPIC, topic)
+                    .register(meterRegistry);
+            newCounter.increment();
+            failedToValidateCounterMap.put(topic, newCounter);
+        }
+    }
 }