SpringCloud/spring-cloud-consul
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

cleanup ttl implementation

Browse Source
This commit is contained in:
Spencer Gibb
2015-03-13 15:19:38 -06:00
parent e35ac90108
commit d8a6e7e9ad
7 changed files with 65 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
spring-cloud-consul-discovery/src/main/java/org/springframework/cloud/consul/discovery/ConsulDiscoveryClientConfiguration.java
View File

@@ -1,5 +1,7 @@
package org.springframework.cloud.consul.discovery;
import com.ecwid.consul.v1.ConsulClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@@ -8,6 +10,10 @@ import org.springframework.context.annotation.Configuration;
*/
@Configuration
public class ConsulDiscoveryClientConfiguration {
@Autowired
private ConsulClient consulClient;
@Bean
public ConsulLifecycle consulLifecycle() {
return new ConsulLifecycle();
@@ -15,19 +21,14 @@ public class ConsulDiscoveryClientConfiguration {
@Bean
public TtlScheduler ttlScheduler() {
return new TtlScheduler();
return new TtlScheduler(heartbeatProperties(), consulClient);
}
@Bean
public TtlHeartbeatProperties ttlHeartbeatConfiguration() {
return new TtlHeartbeatProperties();
public HeartbeatProperties heartbeatProperties() {
return new HeartbeatProperties();
}
/*@Bean
public ConsulLoadBalancerClient consulLoadBalancerClient() {
return new ConsulLoadBalancerClient();
}*/
@Bean
public ConsulDiscoveryClient consulDiscoveryClient() {
return new ConsulDiscoveryClient();

4
spring-cloud-consul-discovery/src/main/java/org/springframework/cloud/consul/discovery/ConsulLifecycle.java
View File

@@ -23,7 +23,7 @@ public class ConsulLifecycle extends AbstractDiscoveryLifecycle {
private TtlScheduler ttlScheduler;
@Autowired
private TtlHeartbeatProperties ttlConfig;
private HeartbeatProperties ttlConfig;
@Override
protected void register() {
@@ -37,7 +37,7 @@ public class ConsulLifecycle extends AbstractDiscoveryLifecycle {
service.setPort(port);
service.setTags(consulProperties.getTags());
NewService.Check check = new NewService.Check();
check.setTtl(ttlConfig.getTtlAsString());
check.setTtl(ttlConfig.getTtl());
service.setCheck(check);
register(service);
}

4
spring-cloud-consul-discovery/src/main/java/org/springframework/cloud/consul/discovery/ConsulRibbonClientConfiguration.java
View File

@@ -21,6 +21,8 @@ import static com.netflix.client.config.CommonClientConfigKey.EnableZoneAffinity
import javax.annotation.PostConstruct;
import com.netflix.loadbalancer.Server;
import com.netflix.loadbalancer.ServerListFilter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
@@ -71,7 +73,7 @@ public class ConsulRibbonClientConfiguration {
}
@Bean
public ServiceCheckServerListFilter ribbonServerListFilter() {
public ServerListFilter<Server> ribbonServerListFilter() {
return new ServiceCheckServerListFilter(client);
}

31
spring-cloud-consul-discovery/src/main/java/org/springframework/cloud/consul/discovery/TtlHeartbeatProperties.java → spring-cloud-consul-discovery/src/main/java/org/springframework/cloud/consul/discovery/HeartbeatProperties.java
View File

@@ -1,37 +1,40 @@
package org.springframework.cloud.consul.discovery;
import lombok.Data;
import org.joda.time.Period;
import org.springframework.boot.context.properties.ConfigurationProperties;
import javax.annotation.PostConstruct;
import javax.validation.constraints.DecimalMax;
import javax.validation.constraints.DecimalMin;
import javax.validation.constraints.Max;
import javax.validation.constraints.Min;
import javax.validation.constraints.NotNull;
import lombok.Data;
import org.joda.time.Period;
import org.springframework.boot.context.properties.ConfigurationProperties;
@ConfigurationProperties(prefix = "consul.heartbeat")
@Data
public class TtlHeartbeatProperties {
public class HeartbeatProperties {
@Min(1)
@Max(10)
private volatile int ttl = 3;
private int ttlValue = 30;
@NotNull
private String ttlUnit = "s";
@DecimalMin("0.1")
@DecimalMax("0.9")
private volatile double intervalRatio = 2.0 / 3.0;
private double intervalRatio = 2.0 / 3.0;
private volatile Period heartbeatInterval;
private Period heartbeatInterval;
@PostConstruct
public void computeHeartbeatInterval() {
// heartbeat rate at ratio * ttl, but no later than ttl -1s and, (under lesser
// priority), no sooner than 1s from now
heartbeatInterval = new Period(Math.round(1000 * Math.max(ttl - 1,
Math.min(ttl * intervalRatio, 1))));
heartbeatInterval = new Period(Math.round(1000 * Math.max(ttlValue - 1,
Math.min(ttlValue * intervalRatio, 1))));
}
public String getTtlAsString() {
return ttl + "s";
public String getTtl() {
return ttlValue + ttlUnit;
}
}

30
spring-cloud-consul-discovery/src/main/java/org/springframework/cloud/consul/discovery/TtlScheduler.java
View File

@@ -4,7 +4,6 @@ import com.ecwid.consul.v1.ConsulClient;
import com.ecwid.consul.v1.agent.model.NewService;
import lombok.extern.slf4j.Slf4j;
import org.joda.time.DateTime;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import java.util.Map;
@@ -19,14 +18,16 @@ public class TtlScheduler {
public static final DateTime EXPIRED_DATE = new DateTime(0);
private final Map<String, DateTime> serviceHeartbeats = new ConcurrentHashMap<>();
private final AtomicBoolean heartbeatingNow = new AtomicBoolean();
@Autowired
private TtlHeartbeatProperties configuration;
private HeartbeatProperties configuration;
@Autowired
private ConsulClient client;
public TtlScheduler(HeartbeatProperties configuration, ConsulClient client) {
this.configuration = configuration;
this.client = client;
}
/**
* Add a service to the checks loop.
*/
@@ -38,16 +39,19 @@ public class TtlScheduler {
serviceHeartbeats.remove(serviceId);
}
@Scheduled(initialDelay = 0, fixedRateString = "${consul.heartbeat.fixedRate:201000}")
@Scheduled(initialDelay = 0, fixedRateString = "${consul.heartbeat.fixedRate:15000}")
private void heartbeatServices() {
if (heartbeatingNow.compareAndSet(false, true)) {
for (String serviceId : serviceHeartbeats.keySet()) {
DateTime latestHeartbeatDoneForService = serviceHeartbeats.get(serviceId);
if (latestHeartbeatDoneForService.plus(configuration.getHeartbeatInterval())
.isBefore(DateTime.now())) {
client.agentCheckPass(serviceId);
serviceHeartbeats.put(serviceId, DateTime.now());
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.info("Sending consul heartbeat for: "+serviceId);
serviceHeartbeats.put(serviceId, DateTime.now());
}
}
}

12
spring-cloud-consul-discovery/src/main/java/org/springframework/cloud/consul/discovery/filters/AliveServerListFilter.java
View File

@@ -4,6 +4,7 @@ import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import com.netflix.loadbalancer.Server;
import org.springframework.cloud.consul.discovery.ConsulServer;
import com.netflix.loadbalancer.ServerListFilter;
@@ -16,7 +17,7 @@ import com.netflix.loadbalancer.ServerListFilter;
* liv in both).
* @author nicu marasoiu on 10.03.2015.
*/
public class AliveServerListFilter implements ServerListFilter<ConsulServer> {
public class AliveServerListFilter implements ServerListFilter<Server> {
private FilteringAgentClient filteringAgentClient;
public AliveServerListFilter(FilteringAgentClient filteringAgentClient) {
@@ -24,11 +25,12 @@ public class AliveServerListFilter implements ServerListFilter<ConsulServer> {
}
@Override
public List<ConsulServer> getFilteredListOfServers(List<ConsulServer> servers) {
public List<Server> getFilteredListOfServers(List<Server> servers) {
Set<String> liveNodes = filteringAgentClient.getAliveAgentsAddresses();
List<ConsulServer> filteredServers = new ArrayList<>();
for (ConsulServer server : servers) {
if (liveNodes.contains(server.getAddress())) {
List<Server> filteredServers = new ArrayList<>();
for (Server server : servers) {
ConsulServer consulServer = ConsulServer.class.cast(server);
if (liveNodes.contains(consulServer.getAddress())) {
filteredServers.add(server);
}
}

18
spring-cloud-consul-discovery/src/main/java/org/springframework/cloud/consul/discovery/filters/ServiceCheckServerListFilter.java
View File

@@ -5,6 +5,7 @@ import java.util.HashSet;
import java.util.List;
import java.util.Set;
import com.netflix.loadbalancer.Server;
import org.springframework.cloud.consul.discovery.ConsulServer;
import com.ecwid.consul.v1.ConsulClient;
@@ -15,7 +16,7 @@ import com.netflix.loadbalancer.ServerListFilter;
/**
* Created by nicu on 12.03.2015.
*/
public class ServiceCheckServerListFilter implements ServerListFilter<ConsulServer> {
public class ServiceCheckServerListFilter implements ServerListFilter<Server> {
private ConsulClient client;
@@ -24,12 +25,13 @@ public class ServiceCheckServerListFilter implements ServerListFilter<ConsulServ
}
@Override
public List<ConsulServer> getFilteredListOfServers(List<ConsulServer> servers) {
public List<Server> getFilteredListOfServers(List<Server> servers) {
Set<String> passingServiceIds = getPassingServiceIds(servers);
List<ConsulServer> okServers = new ArrayList<>(servers.size());
for (ConsulServer consulServer : servers) {
String serviceId = consulServer.getMetaInfo().getInstanceId();
List<Server> okServers = new ArrayList<>(servers.size());
for (Server server : servers) {
String serviceId = server.getMetaInfo().getInstanceId();
if (passingServiceIds.contains(serviceId)) {
ConsulServer consulServer = ConsulServer.class.cast(server);
List<Check> nodeChecks = client.getHealthChecksForNode(
consulServer.getNode(), QueryParams.DEFAULT).getValue();
boolean passingNodeChecks = true;
@@ -40,16 +42,16 @@ public class ServiceCheckServerListFilter implements ServerListFilter<ConsulServ
}
}
if (passingNodeChecks) {
okServers.add(consulServer);
okServers.add(server);
}
}
}
return okServers;
}
private Set<String> getPassingServiceIds(List<ConsulServer> servers) {
private Set<String> getPassingServiceIds(List<Server> servers) {
Set<String> serviceIds = new HashSet<>(1);
for (ConsulServer server : servers) {
for (Server server : servers) {
serviceIds.add(server.getMetaInfo().getInstanceId());
}
for (String serviceId : serviceIds) {