From 650552dbab05da63f5710df27e0a4a4d72d92a00 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?St=C3=A9phane=20Leroy?= Date: Tue, 24 May 2016 19:38:10 +0200 Subject: [PATCH] Improve Consul TTL Check management Remove spring.cloud.consul.discovery.heartbeat.fixedRate property and use a TaskScheduler (with FixedRateTrigger) to send Consul checks Fixes gh-189 --- .../cloud/consul/discovery/TtlScheduler.java | 52 +++++++----- .../discovery/TtlSchedulerRemoveTest.java | 79 +++++++++++++++++++ .../consul/discovery/TtlSchedulerTest.java | 73 +++++++++++++++++ 3 files changed, 184 insertions(+), 20 deletions(-) create mode 100644 spring-cloud-consul-discovery/src/test/java/org/springframework/cloud/consul/discovery/TtlSchedulerRemoveTest.java create mode 100644 spring-cloud-consul-discovery/src/test/java/org/springframework/cloud/consul/discovery/TtlSchedulerTest.java diff --git a/spring-cloud-consul-discovery/src/main/java/org/springframework/cloud/consul/discovery/TtlScheduler.java b/spring-cloud-consul-discovery/src/main/java/org/springframework/cloud/consul/discovery/TtlScheduler.java index 306720bd..78cab588 100644 --- a/spring-cloud-consul-discovery/src/main/java/org/springframework/cloud/consul/discovery/TtlScheduler.java +++ b/spring-cloud-consul-discovery/src/main/java/org/springframework/cloud/consul/discovery/TtlScheduler.java @@ -18,23 +18,28 @@ package org.springframework.cloud.consul.discovery; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledFuture; -import lombok.extern.slf4j.Slf4j; - -import org.joda.time.DateTime; -import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.scheduling.TaskScheduler; +import org.springframework.scheduling.concurrent.ConcurrentTaskScheduler; import com.ecwid.consul.v1.ConsulClient; import com.ecwid.consul.v1.agent.model.NewService; +import lombok.extern.slf4j.Slf4j; + /** * Created by nicu on 11.03.2015. + * @author Stéphane LEROY */ @Slf4j public class TtlScheduler { - public static final DateTime EXPIRED_DATE = new DateTime(0); - private final Map serviceHeartbeats = new ConcurrentHashMap<>(); + private final Map serviceHeartbeats = new ConcurrentHashMap<>(); + + private final TaskScheduler scheduler = new ConcurrentTaskScheduler( + Executors.newSingleThreadScheduledExecutor()); private HeartbeatProperties configuration; @@ -49,27 +54,34 @@ public class TtlScheduler { * Add a service to the checks loop. */ public void add(final NewService service) { - serviceHeartbeats.put(service.getId(), EXPIRED_DATE); + ScheduledFuture task = scheduler.scheduleAtFixedRate(new ConsulHeartbeatTask( + service.getId()), configuration.computeHearbeatInterval() + .toStandardDuration().getMillis()); + serviceHeartbeats.put(service.getId(), task); } public void remove(String serviceId) { + ScheduledFuture task = serviceHeartbeats.get(serviceId); + if (task != null) { + task.cancel(true); + } serviceHeartbeats.remove(serviceId); } - @Scheduled(initialDelay = 0, fixedRateString = "${spring.cloud.consul.discovery.heartbeat.fixedRate:15000}") - private void heartbeatServices() { - for (String serviceId : serviceHeartbeats.keySet()) { - DateTime latestHeartbeatDoneForService = serviceHeartbeats.get(serviceId); - if (latestHeartbeatDoneForService.plus(configuration.getHeartbeatInterval()) - .isBefore(DateTime.now())) { - String checkId = serviceId; - if (!checkId.startsWith("service:")) { - checkId = "service:" + checkId; - } - client.agentCheckPass(checkId); - log.debug("Sending consul heartbeat for: " + serviceId); - serviceHeartbeats.put(serviceId, DateTime.now()); + private class ConsulHeartbeatTask implements Runnable { + private String checkId; + + ConsulHeartbeatTask(String serviceId) { + this.checkId = serviceId; + if (!checkId.startsWith("service:")) { + checkId = "service:" + checkId; } } + + @Override + public void run() { + client.agentCheckPass(checkId); + log.debug("Sending consul heartbeat for: " + checkId); + } } } \ No newline at end of file diff --git a/spring-cloud-consul-discovery/src/test/java/org/springframework/cloud/consul/discovery/TtlSchedulerRemoveTest.java b/spring-cloud-consul-discovery/src/test/java/org/springframework/cloud/consul/discovery/TtlSchedulerRemoveTest.java new file mode 100644 index 00000000..70c1f718 --- /dev/null +++ b/spring-cloud-consul-discovery/src/test/java/org/springframework/cloud/consul/discovery/TtlSchedulerRemoveTest.java @@ -0,0 +1,79 @@ +package org.springframework.cloud.consul.discovery; + +import java.util.List; + +import org.junit.FixMethodOrder; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.MethodSorters; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.test.SpringApplicationConfiguration; +import org.springframework.boot.test.WebIntegrationTest; +import org.springframework.cloud.client.discovery.EnableDiscoveryClient; +import org.springframework.cloud.consul.ConsulAutoConfiguration; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.Import; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +import com.ecwid.consul.v1.ConsulClient; +import com.ecwid.consul.v1.QueryParams; +import com.ecwid.consul.v1.Response; +import com.ecwid.consul.v1.health.model.Check; + +import static com.ecwid.consul.v1.health.model.Check.CheckStatus.CRITICAL; +import static com.ecwid.consul.v1.health.model.Check.CheckStatus.PASSING; +import static org.hamcrest.core.IsEqual.equalTo; +import static org.junit.Assert.assertThat; + +/** + * @author Stéphane Leroy + */ +@RunWith(SpringJUnit4ClassRunner.class) +@FixMethodOrder(MethodSorters.NAME_ASCENDING) +@SpringApplicationConfiguration(classes = TtlSchedulerRemoveTestConfig.class) +@WebIntegrationTest(value = { "spring.application.name=ttlSchedulerRemove", + "spring.cloud.consul.discovery.instanceId=ttlSchedulerRemove-id", + "spring.cloud.consul.discovery.heartbeat.enabled=true", + "spring.cloud.consul.discovery.heartbeat.ttlValue=2" }, randomPort = true) +public class TtlSchedulerRemoveTest { + + @Autowired + private ConsulClient consul; + + @Autowired + private TtlScheduler ttlScheduler; + + @Test + public void should_not_send_check_if_service_removed() throws InterruptedException { + Thread.sleep(1000); // wait for Ttlscheduler to send a check to consul. + Check serviceCheck = getCheckForService("ttlSchedulerRemove"); + assertThat("Service check is in wrong state", serviceCheck.getStatus(), + equalTo(PASSING)); + + // Remove service from TtlScheduler and wait for TTL to expired. + ttlScheduler.remove("ttlSchedulerRemove-id"); + Thread.sleep(2100); + serviceCheck = getCheckForService("ttlSchedulerRemove"); + assertThat("Service check is in wrong state", serviceCheck.getStatus(), + equalTo(CRITICAL)); + } + + private Check getCheckForService(String serviceId) { + Response> checkResponse = consul.getHealthChecksForService(serviceId, + QueryParams.DEFAULT); + if (checkResponse.getValue().size() > 0) { + return checkResponse.getValue().get(0); + } + return null; + } + +} + +@Configuration +@EnableDiscoveryClient +@EnableAutoConfiguration +@Import({ ConsulAutoConfiguration.class, ConsulDiscoveryClientConfiguration.class }) +class TtlSchedulerRemoveTestConfig { + +} diff --git a/spring-cloud-consul-discovery/src/test/java/org/springframework/cloud/consul/discovery/TtlSchedulerTest.java b/spring-cloud-consul-discovery/src/test/java/org/springframework/cloud/consul/discovery/TtlSchedulerTest.java new file mode 100644 index 00000000..f27fa734 --- /dev/null +++ b/spring-cloud-consul-discovery/src/test/java/org/springframework/cloud/consul/discovery/TtlSchedulerTest.java @@ -0,0 +1,73 @@ +package org.springframework.cloud.consul.discovery; + +import java.util.List; + +import org.junit.FixMethodOrder; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.MethodSorters; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.test.SpringApplicationConfiguration; +import org.springframework.boot.test.WebIntegrationTest; +import org.springframework.cloud.client.discovery.EnableDiscoveryClient; +import org.springframework.cloud.consul.ConsulAutoConfiguration; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.Import; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +import com.ecwid.consul.v1.ConsulClient; +import com.ecwid.consul.v1.QueryParams; +import com.ecwid.consul.v1.Response; +import com.ecwid.consul.v1.health.model.Check; + +import static com.ecwid.consul.v1.health.model.Check.CheckStatus.PASSING; +import static org.hamcrest.core.IsEqual.equalTo; +import static org.junit.Assert.assertThat; + +/** + * @author Stéphane Leroy + */ +@RunWith(SpringJUnit4ClassRunner.class) +@FixMethodOrder(MethodSorters.NAME_ASCENDING) +@SpringApplicationConfiguration(classes = TtlSchedulerTestConfig.class) +@WebIntegrationTest(value = { "spring.application.name=ttlScheduler", + "spring.cloud.consul.discovery.instanceId=ttlScheduler-id", + "spring.cloud.consul.discovery.heartbeat.enabled=true", + "spring.cloud.consul.discovery.heartbeat.ttlValue=2", "management.port=0" }, randomPort = true) +public class TtlSchedulerTest { + + @Autowired + private ConsulClient consul; + + @Test + public void should_send_a_check_before_ttl_for_all_services() + throws InterruptedException { + Thread.sleep(2100); // Wait for TTL to expired (TTL is set to 2 seconds) + + Check serviceCheck = getCheckForService("ttlScheduler"); + assertThat("Service check is in wrong state", serviceCheck.getStatus(), + equalTo(PASSING)); + Check serviceManagementCheck = getCheckForService("ttlScheduler-management"); + assertThat("Service management heck in wrong state", + serviceManagementCheck.getStatus(), equalTo(PASSING)); + } + + private Check getCheckForService(String serviceId) { + Response> checkResponse = consul.getHealthChecksForService(serviceId, + QueryParams.DEFAULT); + if (checkResponse.getValue().size() > 0) { + return checkResponse.getValue().get(0); + } + return null; + } + +} + +@Configuration +@EnableDiscoveryClient +@EnableAutoConfiguration +@Import({ ConsulAutoConfiguration.class, ConsulDiscoveryClientConfiguration.class }) +class TtlSchedulerTestConfig { + +}