add periodic consistency check spec

hermes-management/src/main/java/pl/allegro/tech/hermes/management/config/ConsistencyCheckerProperties.java

@@ -10,6 +10,7 @@ public class ConsistencyCheckerProperties {
     private int threadPoolSize = 2;
     private boolean periodicCheckEnabled = false;
     private Duration refreshInterval = Duration.ofMinutes(15);
+    private Duration initialRefreshDelay = Duration.ofMinutes(2);
 
     public int getThreadPoolSize() {
         return threadPoolSize;
@@ -36,4 +37,12 @@ public Duration getRefreshInterval() {
     public void setRefreshInterval(Duration refreshInterval) {
         this.refreshInterval = refreshInterval;
     }
+
+    public Duration getInitialRefreshDelay() {
+        return initialRefreshDelay;
+    }
+
+    public void setInitialRefreshDelay(Duration initialRefreshDelay) {
+        this.initialRefreshDelay = initialRefreshDelay;
+    }
 }

hermes-management/src/main/java/pl/allegro/tech/hermes/management/domain/consistency/DcConsistencyService.java

@@ -75,13 +75,17 @@ public DcConsistencyService(RepositoryManager repositoryManager,
                         .build()
         );
         if (properties.isPeriodicCheckEnabled()) {
-            scheduler.scheduleAtFixedRate(this::reportConsistency, 0, properties.getRefreshInterval().getSeconds(), TimeUnit.SECONDS);
+            scheduler.scheduleAtFixedRate(this::reportConsistency,
+                    properties.getInitialRefreshDelay().getSeconds(),
+                    properties.getRefreshInterval().getSeconds(),
+                    TimeUnit.SECONDS);
         }
     }
 
     @PreDestroy
     public void stop() {
         executor.shutdown();
+        scheduler.shutdown();
     }
 
     private void reportConsistency() {

hermes-management/src/test/groovy/pl/allegro/tech/hermes/management/domain/consistency/DcConsistencyServiceSpec.groovy

@@ -2,11 +2,16 @@ package pl.allegro.tech.hermes.management.domain.consistency
 
 import com.fasterxml.jackson.databind.ObjectMapper
 import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule
+import io.micrometer.core.instrument.simple.SimpleMeterRegistry
 import pl.allegro.tech.hermes.api.Group
 import pl.allegro.tech.hermes.api.Subscription
 import pl.allegro.tech.hermes.api.Topic
+import pl.allegro.tech.hermes.common.metric.MetricsFacade
 import pl.allegro.tech.hermes.management.config.ConsistencyCheckerProperties
 import spock.lang.Specification
+import spock.util.concurrent.PollingConditions
+
+import java.time.Duration
 
 import static pl.allegro.tech.hermes.test.helper.builder.GroupBuilder.group
 import static pl.allegro.tech.hermes.test.helper.builder.SubscriptionBuilder.subscription
@@ -15,6 +20,8 @@ import static pl.allegro.tech.hermes.test.helper.builder.TopicBuilder.topic
 class DcConsistencyServiceSpec extends Specification {
 
     def objectMapper = new ObjectMapper().registerModule(new JavaTimeModule())
+    def meterRegistry = new SimpleMeterRegistry()
+    def metricsFacade = new MetricsFacade(meterRegistry)
 
     def "should return empty list when given groups are consistent"() {
         given:
@@ -30,8 +37,8 @@ class DcConsistencyServiceSpec extends Specification {
                 .addGroup(group)
                 .addTopic(topic)
                 .addSubscription(subscription)
-        DcConsistencyService dcConsistencyService = new DcConsistencyService(repositoryManager, objectMapper,
-                new ConsistencyCheckerProperties())
+        DcConsistencyService dcConsistencyService = new DcConsistencyService(repositoryManager,
+                objectMapper, new ConsistencyCheckerProperties(), metricsFacade)
 
         when:
         def inconsistentGroups = dcConsistencyService.listInconsistentGroups([group.groupName] as Set)
@@ -48,8 +55,8 @@ class DcConsistencyServiceSpec extends Specification {
         repositoryManager.datacenter("dc2")
                 .addGroup(group("testGroup").build())
                 .addGroup(group("testGroup-dc2").build())
-        DcConsistencyService consistencyService = new DcConsistencyService(repositoryManager, objectMapper,
-                new ConsistencyCheckerProperties())
+        DcConsistencyService consistencyService = new DcConsistencyService(repositoryManager,
+                objectMapper, new ConsistencyCheckerProperties(), metricsFacade)
 
         when:
         def groups = consistencyService.listInconsistentGroups(["testGroup", "testGroup-dc1", "testGroup-dc2"] as Set)
@@ -68,7 +75,7 @@ class DcConsistencyServiceSpec extends Specification {
                 .addGroup(group)
                 .addTopic(topic(group.groupName, "testTopic").withDescription("dc2").build())
         DcConsistencyService consistencyService = new DcConsistencyService(repositoryManager, objectMapper,
-                new ConsistencyCheckerProperties())
+                new ConsistencyCheckerProperties(), metricsFacade)
 
         when:
         def groups = consistencyService.listInconsistentGroups(["testGroup"] as Set)
@@ -90,7 +97,7 @@ class DcConsistencyServiceSpec extends Specification {
                 .addTopic(topic)
                 .addSubscription(subscription(topic, "testSubscription").withDescription("dc2").build())
         DcConsistencyService consistencyService = new DcConsistencyService(repositoryManager, objectMapper,
-                new ConsistencyCheckerProperties())
+                new ConsistencyCheckerProperties(), metricsFacade)
 
         when:
         def groups = consistencyService.listInconsistentGroups(["testGroup"] as Set)
@@ -108,12 +115,64 @@ class DcConsistencyServiceSpec extends Specification {
                 .addGroup(group("testGroup").build())
                 .addGroup(group("testGroup-dc2").build())
         DcConsistencyService consistencyService = new DcConsistencyService(repositoryManager, objectMapper,
-                new ConsistencyCheckerProperties())
+                new ConsistencyCheckerProperties(), metricsFacade)
 
         when:
         def groups = consistencyService.listAllGroupNames()
 
         then:
         groups == ["testGroup", "testGroup-dc1", "testGroup-dc2"] as Set
     }
+
+    def "should report storage as not consistent with periodic check"() {
+        given: "inconsistent storage state"
+        MockRepositoryManager repositoryManager = new MockRepositoryManager()
+        repositoryManager.datacenter("dc1")
+                .addGroup(group("testGroup").build())
+                .addGroup(group("testGroup-dc1").build())
+        repositoryManager.datacenter("dc2")
+                .addGroup(group("testGroup").build())
+                .addGroup(group("testGroup-dc2").build())
+
+        and: "enabled periodic consistency checks"
+        def properties = new ConsistencyCheckerProperties()
+        properties.setPeriodicCheckEnabled(true)
+        properties.setInitialRefreshDelay(Duration.ofMillis(0))
+
+        when: "consistency service is created"
+        DcConsistencyService consistencyService = new DcConsistencyService(repositoryManager,
+                objectMapper,
+                properties,
+                metricsFacade)
+
+        then: "storage is reported as not consistent"
+        new PollingConditions(timeout: 10).eventually {
+            meterRegistry.get("storage.consistency").gauge().value() == 0.0d
+        }
+    }
+
+    def "should report storage as consistent with periodic check"() {
+        given: "consistent storage state"
+        MockRepositoryManager repositoryManager = new MockRepositoryManager()
+        repositoryManager.datacenter("dc1")
+                .addGroup(group("testGroup").build())
+        repositoryManager.datacenter("dc2")
+                .addGroup(group("testGroup").build())
+
+        and: "enabled periodic consistency checks"
+        def properties = new ConsistencyCheckerProperties()
+        properties.setPeriodicCheckEnabled(true)
+        properties.setInitialRefreshDelay(Duration.ofMillis(0))
+
+        when: "consistency service is created"
+        DcConsistencyService consistencyService = new DcConsistencyService(repositoryManager,
+                objectMapper,
+                properties,
+                metricsFacade)
+
+        then: "storage is reported as consistent"
+        new PollingConditions(timeout: 10).eventually {
+            meterRegistry.get("storage.consistency").gauge().value() == 1.0d
+        }
+    }
 }