Unhelathy topic/subscriptions tests rewritten to prometheus

integration/src/integration/java/pl/allegro/tech/hermes/integration/MetricsTest.java

@@ -2,6 +2,8 @@
 
 import com.googlecode.catchexception.CatchException;
 import jakarta.ws.rs.BadRequestException;
+import java.util.Collections;
+import java.util.List;
 import org.testng.annotations.BeforeMethod;
 import org.testng.annotations.Test;
 import pl.allegro.tech.hermes.api.MessageFilterSpecification;
@@ -14,6 +16,9 @@
 import pl.allegro.tech.hermes.frontend.config.GraphiteProperties;
 import pl.allegro.tech.hermes.integration.env.SharedServices;
 import pl.allegro.tech.hermes.integration.helper.PrometheusEndpoint;
+import pl.allegro.tech.hermes.integration.helper.PrometheusEndpoint.PrometheusSubscriptionResponse;
+import pl.allegro.tech.hermes.integration.helper.PrometheusEndpoint.PrometheusSubscriptionResponseBuilder;
+import pl.allegro.tech.hermes.integration.helper.PrometheusEndpoint.PrometheusTopicResponse;
 import pl.allegro.tech.hermes.integration.helper.graphite.GraphiteMockServer;
 import pl.allegro.tech.hermes.integration.shame.Unreliable;
 import pl.allegro.tech.hermes.test.helper.endpoint.RemoteServiceEndpoint;
@@ -27,6 +32,7 @@
 import static java.lang.Integer.MAX_VALUE;
 import static org.assertj.core.api.Assertions.assertThat;
 import static pl.allegro.tech.hermes.api.BatchSubscriptionPolicy.Builder.batchSubscriptionPolicy;
+import static pl.allegro.tech.hermes.integration.helper.PrometheusEndpoint.PrometheusSubscriptionResponseBuilder.*;
 import static pl.allegro.tech.hermes.test.helper.builder.SubscriptionBuilder.subscription;
 import static pl.allegro.tech.hermes.test.helper.builder.TopicBuilder.randomTopic;
 
@@ -54,7 +60,8 @@ public void shouldIncreaseTopicMetricsAfterMessageHasBeenPublished() {
         // given
         Topic topic = operations.buildTopic(randomTopic("group", "topic_metrics").build());
         operations.createSubscription(topic, "subscription", remoteService.getUrl());
-        prometheusEndpoint.returnTopicMetrics(topic.getName().getGroupName(), topic.getName().getName(), 10, 15);
+        prometheusEndpoint.returnTopicMetrics(topic.getName().getGroupName(), topic.getName().getName(),
+                new PrometheusTopicResponse(10, 15, 0));
 
         remoteService.expectMessages(TestMessage.simple().body());
         assertThat(publisher.publish(topic.getQualifiedName(), TestMessage.simple().body()).getStatus())
@@ -79,7 +86,7 @@ public void shouldIncreaseSubscriptionDeliveredMetricsAfterMessageDelivered() {
         // given
         Topic topic = operations.buildTopic(randomTopic("pl.group", "topic").build());
         operations.createSubscription(topic, "subscription", remoteService.getUrl());
-        prometheusEndpoint.returnSubscriptionMetrics(topic, "subscription", 15);
+        prometheusEndpoint.returnSubscriptionMetrics(topic, "subscription", builder().withRate(15).build());
 
         remoteService.expectMessages(TestMessage.simple().body());
         assertThat(publisher.publish(topic.getQualifiedName(), TestMessage.simple().body()).getStatus())
@@ -122,7 +129,7 @@ public void shouldReadSubscriptionDeliveryRate() {
         Topic topic = operations.buildTopic("pl.allegro.tech.hermes", "topic");
         String subscriptionName = "pl.allegro.tech.hermes.subscription";
         operations.createSubscription(topic, subscriptionName, remoteService.getUrl());
-        prometheusEndpoint.returnSubscriptionMetrics(topic, subscriptionName, 15);
+        prometheusEndpoint.returnSubscriptionMetrics(topic, subscriptionName, builder().withRate(15).build());
 
         wait.until(() -> {
             // when

integration/src/integration/java/pl/allegro/tech/hermes/integration/helper/PrometheusEndpoint.java

@@ -1,14 +1,17 @@
 package pl.allegro.tech.hermes.integration.helper;
 
+import com.fasterxml.jackson.annotation.JsonInclude;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.github.tomakehurst.wiremock.WireMockServer;
 import com.github.tomakehurst.wiremock.client.WireMock;
-import java.util.List;
+import java.util.ArrayList;
 import pl.allegro.tech.hermes.api.Topic;
 import pl.allegro.tech.hermes.integration.env.EnvironmentAware;
 
+import java.util.List;
+
 import static com.github.tomakehurst.wiremock.client.WireMock.aResponse;
 import static com.github.tomakehurst.wiremock.client.WireMock.get;
 import static com.github.tomakehurst.wiremock.client.WireMock.urlMatching;
@@ -37,7 +40,7 @@ public class PrometheusEndpoint implements EnvironmentAware {
             ".*hermes_consumers_subscription_throughput_bytes_total" +
             ".*hermes_consumers_subscription_other_errors_total" +
             ".*hermes_consumers_subscription_batches_total" +
-            ".*hermes_consumers_subscription_http_status_codes_total.*";
+            ".*hermes_consumers_subscription_http_status_codes_total.*GROUP.*TOPIC.*SUBSCRIPTION.*";
 
     private final ObjectMapper objectMapper;
     private final WireMock prometheusListener;
@@ -47,8 +50,8 @@ public PrometheusEndpoint(WireMockServer prometheus) {
         this.objectMapper = new ObjectMapper();
     }
 
-    public void returnTopicMetrics(String group, String topic, int rate, int deliveryRate) {
-        String response = generateTopicsMetricsResponse(rate, deliveryRate);
+    public void returnTopicMetrics(String group, String topic, PrometheusTopicResponse topicStub) {
+        String response = generateTopicsMetricsResponse(topicStub.topicRate, topicStub.deliveredRate, topicStub.throughput);
         String query = TOPIC_QUERY_PATTERN
                 .replaceAll("GROUP", group)
                 .replaceAll("TOPIC", topic);
@@ -59,8 +62,8 @@ public void returnTopicMetrics(String group, String topic, int rate, int deliver
                         .withBody(response)));
     }
 
-    public void returnSubscriptionMetrics(Topic topic, String subscription, int rate) {
-        String response = generateSubscriptionResponse(rate);
+    public void returnSubscriptionMetrics(Topic topic, String subscription, PrometheusSubscriptionResponse stub) {
+        String response = generateSubscriptionResponse(stub);
         String query = SUBSCRIPTION_QUERY_PATTERN
                 .replaceAll("GROUP", topic.getName().getGroupName())
                 .replaceAll("TOPIC", topic.getName().getName())
@@ -72,128 +75,66 @@ public void returnSubscriptionMetrics(Topic topic, String subscription, int rate
                         .withBody(response)));
     }
 
-    private String generateTopicsMetricsResponse(int rate, int deliveryRate) {
+    public void returnSubscriptionMetricsWithDelay(Topic topic, String subscription, PrometheusSubscriptionResponse stub,
+                                                   int prometheusDelay) {
+        String response = generateSubscriptionResponse(stub);
+        String query = SUBSCRIPTION_QUERY_PATTERN
+                .replaceAll("GROUP", topic.getName().getGroupName())
+                .replaceAll("TOPIC", topic.getName().getName())
+                .replaceAll("SUBSCRIPTION", subscription);
+        prometheusListener.register(get(urlMatching(query))
+                .willReturn(aResponse()
+                        .withStatus(200)
+                        .withFixedDelay(prometheusDelay)
+                        .withHeader("Content-Type", "application/json")
+                        .withBody(response)));
+    }
+
+    public void returnServerErrorForAllTopics() {
+        String query = TOPIC_QUERY_PATTERN
+                .replaceAll("GROUP", "")
+                .replaceAll("TOPIC", "");
+        prometheusListener.register(get(urlMatching(query))
+                .willReturn(aResponse()
+                        .withStatus(500)
+                        .withHeader("Content-Type", "application/json")));
+    }
+
+    private String generateTopicsMetricsResponse(int rate, int deliveryRate, int throughput) {
         return writeToString(
                 new PrometheusResponse(
                         "success",
                         new PrometheusResponse.Data(
                                 "vector",
                                 List.of(
                                         new PrometheusResponse.Result(
-                                                new PrometheusResponse.MetricName(TOPIC_REQUESTS_TOTAL),
+                                                new PrometheusResponse.MetricName(TOPIC_REQUESTS_TOTAL, null),
                                                 List.of(TIMESTAMP, String.valueOf(rate))),
                                         new PrometheusResponse.Result(
-                                                new PrometheusResponse.MetricName(TOPIC_DELIVERED_TOTAL),
-                                                List.of(TIMESTAMP, String.valueOf(deliveryRate))))
+                                                new PrometheusResponse.MetricName(TOPIC_DELIVERED_TOTAL, null),
+                                                List.of(TIMESTAMP, String.valueOf(deliveryRate))),
+                                        new PrometheusResponse.Result(
+                                                new PrometheusResponse.MetricName(TOPIC_THROUGHPUT_TOTAL, null),
+                                                List.of(TIMESTAMP, String.valueOf(throughput))))
                         )));
     }
 
-    private String generateSubscriptionResponse(int rate) {
-        return writeToString(
-                new PrometheusResponse(
-                        "success",
-                        new PrometheusResponse.Data(
-                                "vector",
-                                List.of(
-                                        new PrometheusResponse.Result(
-                                                new PrometheusResponse.MetricName(SUBSCRIPTION_DELIVERED),
-                                                List.of(TIMESTAMP, String.valueOf(rate)))
-                                ))));
+    private String generateSubscriptionResponse(PrometheusSubscriptionResponse stub) {
+        List<PrometheusResponse.Result> results = new ArrayList<>();
+        results.add(
+                new PrometheusResponse.Result(
+                        new PrometheusResponse.MetricName(SUBSCRIPTION_DELIVERED, null),
+                        List.of(TIMESTAMP, String.valueOf(stub.rate)))
+        );
+        stub.statusCodes().forEach(s -> results.add(
+                new PrometheusResponse.Result(
+                        new PrometheusResponse.MetricName(SUBSCRIPTION_STATUS_CODES, s.code()),
+                        List.of(TIMESTAMP, String.valueOf(s.rate)))));
+        PrometheusResponse response = new PrometheusResponse(
+                "success", new PrometheusResponse.Data("vector", results));
+        return writeToString(response);
     }
 
-
-//    public void returnServerErrorForAllTopics() {
-//        graphiteListener.register(get(urlMatching(TOPIC_URL_PATTERN))
-//                .willReturn(aResponse()
-//                        .withStatus(500)
-//                        .withHeader("Content-Type", "application/json")
-//                )
-//        );
-//    }
-//
-//    public void returnMetric(SubscriptionMetricsStubDefinition metricsStubDefinition) {
-//        graphiteListener.register(get(urlMatching(metricsStubDefinition.toUrlPattern()))
-//                .willReturn(aResponse()
-//                        .withStatus(200)
-//                        .withHeader("Content-Type", "application/json")
-//                        .withBody(metricsStubDefinition.toBody())));
-//    }
-//
-//    public void returnMetricWithDelay(SubscriptionMetricsStubDefinition metricsStubDefinition, int responseDelayInMs) {
-//        graphiteListener.register(get(urlMatching(metricsStubDefinition.toUrlPattern()))
-//                .willReturn(aResponse()
-//                        .withFixedDelay(responseDelayInMs)
-//                        .withStatus(200)
-//                        .withHeader("Content-Type", "application/json")
-//                        .withBody(metricsStubDefinition.toBody())));
-//    }
-//
-//    private static class SubscriptionMetricsStubDefinition {
-//        private final String subscription;
-//        private final List<GraphiteStubResponse> responseBody;
-//
-//        private SubscriptionMetricsStubDefinition(String subscription, List<GraphiteStubResponse> responseBody) {
-//            this.subscription = subscription;
-//            this.responseBody = responseBody;
-//        }
-//
-//        private String toUrlPattern() {
-//            return "/.*sumSeries%28stats.tech.hermes\\.consumer\\.%2A\\.meter\\." + subscription + "\\.m1_rate%29.*";
-//        }
-//
-//        private String toBody() {
-//            try {
-//                return new ObjectMapper().writeValueAsString(responseBody);
-//            } catch (JsonProcessingException e) {
-//                throw new RuntimeException(e);
-//            }
-//        }
-//    }
-//
-//    public static class SubscriptionMetricsStubDefinitionBuilder {
-//        private final String subscription;
-//        private final List<GraphiteStubResponse> response = new ArrayList<>();
-//
-//        private SubscriptionMetricsStubDefinitionBuilder(String subscription) {
-//            this.subscription = subscription;
-//        }
-//
-//        public SubscriptionMetricsStubDefinitionBuilder withRate(int rate) {
-//            String target = "sumSeries(stats.tech.hermes.consumer.*.meter." + subscription + ".m1_rate)";
-//            response.add(new GraphiteStubResponse(target, dataPointOf(rate)));
-//            return this;
-//        }
-//
-//        public SubscriptionMetricsStubDefinitionBuilder withThroughput(int rate) {
-//            String target = "sumSeries(stats.tech.hermes.consumer.*.throughput." + subscription + ".m1_rate)";
-//            response.add(new GraphiteStubResponse(target, dataPointOf(rate)));
-//            return this;
-//        }
-//
-//        public SubscriptionMetricsStubDefinitionBuilder withStatusRate(int httpStatus, int rate) {
-//            String statusFamily = httpStatusFamily(httpStatus);
-//            String target = "sumSeries(stats.tech.hermes.consumer.*.status." + subscription + "." + statusFamily + ".m1_rate)";
-//            response.add(new GraphiteStubResponse(target, dataPointOf(rate)));
-//            return this;
-//        }
-//
-//        public SubscriptionMetricsStubDefinition build() {
-//            return new SubscriptionMetricsStubDefinition(subscription, response);
-//        }
-//
-//        private String httpStatusFamily(int statusCode) {
-//            return format("%dxx", statusCode / 100);
-//        }
-//
-//        private static List<List<Object>> dataPointOf(int rate) {
-//            return singletonList(asList(rate, TIMESTAMP));
-//        }
-//    }
-//
-//    public static SubscriptionMetricsStubDefinitionBuilder subscriptionMetricsStub(String subscription) {
-//        return new SubscriptionMetricsStubDefinitionBuilder(subscription);
-//    }
-
     String writeToString(Object o) {
         try {
             return objectMapper.writeValueAsString(o);
@@ -202,27 +143,58 @@ String writeToString(Object o) {
         }
     }
 
-    record PrometheusResponse(@JsonProperty("status") String status,
-                              @JsonProperty("data") Data data) {
+    public record PrometheusTopicResponse(int topicRate, int deliveredRate, int throughput) {
+    }
+
+    public record PrometheusSubscriptionResponse(int rate, List<SubscriptionStatusCode> statusCodes) {
+    }
+
+    public record SubscriptionStatusCode(String code, int rate) {
+    }
+
+    public static class PrometheusSubscriptionResponseBuilder {
+        private int rate = 0;
+        private final List<SubscriptionStatusCode> statusCodes = new ArrayList<>();
 
-        boolean isSuccess() {
-            return status.equals("success") && data.isVector();
+        private PrometheusSubscriptionResponseBuilder() {
         }
 
+        public static PrometheusSubscriptionResponseBuilder builder() {
+            return new PrometheusSubscriptionResponseBuilder();
+        }
+
+        public PrometheusSubscriptionResponseBuilder withRate(int rate) {
+            this.rate = rate;
+            return this;
+        }
+
+        public PrometheusSubscriptionResponseBuilder withRatedStatusCode(String statusCode, int rate) {
+            this.statusCodes.add(new SubscriptionStatusCode(statusCode, rate));
+            return this;
+        }
+
+        public PrometheusSubscriptionResponse build() {
+            return new PrometheusSubscriptionResponse(rate, statusCodes);
+        }
+    }
+
+    record PrometheusResponse(@JsonProperty("status") String status,
+                              @JsonProperty("data") Data data) {
+
         record Data(@JsonProperty("resultType") String resultType,
                     @JsonProperty("result") List<Result> results) {
-            boolean isVector() {
-                return resultType.equals("vector");
-            }
         }
 
         record Result(
                 @JsonProperty("metric") MetricName metricName,
                 @JsonProperty("value") List<String> values) {
         }
 
+        @JsonInclude(JsonInclude.Include.NON_NULL)
         record MetricName(
-                @JsonProperty(value = "__name__") String name) {
+                @JsonProperty(value = "__name__") String name,
+                @JsonProperty(value = "status_code") String statusCode
+        ) {
         }
     }
 }