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

Send heartbeats at 2/3 of ttl + List service nodes filtered by alive, when getting the server list

Browse Source
This commit is contained in:
nicu marasoiu
2015-03-11 15:25:35 +02:00
parent b870e790b5
commit 7ec2132162
12 changed files with 446 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
pom.xml
View File

@@ -18,12 +18,12 @@
<!-- lookup parent from repository -->
</parent>
<scm>
<url>https://github.com/spring-cloud/spring-cloud-consul</url>
<connection>scm:git:git://github.com/spring-cloud/spring-cloud-consul.git</connection>
<developerConnection>scm:git:ssh://git@github.com/spring-cloud/spring-cloud-consul.git</developerConnection>
<tag>HEAD</tag>
</scm>
<scm>
<url>https://github.com/spring-cloud/spring-cloud-consul</url>
<connection>scm:git:git://github.com/spring-cloud/spring-cloud-consul.git</connection>
<developerConnection>scm:git:ssh://git@github.com/spring-cloud/spring-cloud-consul.git</developerConnection>
<tag>HEAD</tag>
</scm>
<modules>
<module>spring-cloud-consul-core</module>
@@ -32,6 +32,7 @@
<module>spring-cloud-consul-bus</module>
<module>spring-cloud-consul-sample</module>
<module>spring-cloud-consul-tests</module>
<module>spring-cloud-consul-utils</module>
</modules>
<build>
@@ -56,9 +57,9 @@
<version>1.0.0.BUILD-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-consul-core</artifactId>
<version>1.0.0.BUILD-SNAPSHOT</version>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-consul-core</artifactId>
<version>1.0.0.BUILD-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
@@ -110,7 +111,7 @@
<dependency>
<groupId>com.ecwid.consul</groupId>
<artifactId>consul-api</artifactId>
<version>0.1</version>
<version>1.0.8</version>
<exclusions>
<exclusion>
<groupId>javax.servlet</groupId>

22
spring-cloud-consul-core/src/main/java/org/springframework/cloud/consul/model/SerfStatusEnum.java Normal file
View File

@@ -0,0 +1,22 @@
package org.springframework.cloud.consul.model;
/**
* Gossip pool (serf) statuses.
* Created by nicu on 10.03.2015.
*/
public enum SerfStatusEnum {
StatusAlive(1),
StatusLeaving(2),
StatusLeft(3),
StatusFailed(4);
private final int code;
SerfStatusEnum(int code) {
this.code=code;
}
public int getCode() {
return code;
}
}

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

@@ -13,6 +13,11 @@ public class ConsulDiscoveryClientConfiguration {
return new ConsulLifecycle();
}
@Bean
public TtlScheduler ttlScheduler() {
return new TtlScheduler();
}
/*@Bean
public ConsulLoadBalancerClient consulLoadBalancerClient() {
return new ConsulLoadBalancerClient();

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

@@ -19,6 +19,9 @@ public class ConsulLifecycle extends AbstractDiscoveryLifecycle {
@Autowired
private ConsulProperties consulProperties;
@Autowired
private TtlScheduler ttlScheduler;
@Override
protected void register() {
NewService service = new NewService();
@@ -30,8 +33,9 @@ public class ConsulLifecycle extends AbstractDiscoveryLifecycle {
Integer port = new Integer(getEnvironment().getProperty("server.port", "8080"));
service.setPort(port);
service.setTags(consulProperties.getTags());
//TODO: add support for Check
NewService.Check check = new NewService.Check();
check.setTtl(ttlScheduler.getTTL() + "s");
service.setCheck(check);
register(service);
}
@@ -49,6 +53,7 @@ public class ConsulLifecycle extends AbstractDiscoveryLifecycle {
protected void register(NewService service) {
log.info("Registering service with consul: {}", service.toString());
client.agentServiceRegister(service);
ttlScheduler.add(service);
}
@Override
@@ -57,7 +62,7 @@ public class ConsulLifecycle extends AbstractDiscoveryLifecycle {
}
@Override
protected void deregister(){
protected void deregister() {
deregister(getContext().getId());
}
@@ -67,6 +72,7 @@ public class ConsulLifecycle extends AbstractDiscoveryLifecycle {
}
private void deregister(String serviceId) {
ttlScheduler.remove(serviceId);
client.agentServiceDeregister(serviceId);
}

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

@@ -9,9 +9,13 @@ import com.netflix.loadbalancer.Server;
public class ConsulServer extends Server {
private final MetaInfo metaInfo;
private final String address;
private final String node;
public ConsulServer(final CatalogService service) {
super(service.getNode(), service.getServicePort());
address = service.getAddress();
node = service.getNode();
metaInfo = new MetaInfo() {
@Override
public String getAppName() {
@@ -39,4 +43,12 @@ public class ConsulServer extends Server {
public MetaInfo getMetaInfo() {
return metaInfo;
}
public String getAddress() {
return address;
}
public String getNode() {
return node;
}
}

126
spring-cloud-consul-discovery/src/main/java/org/springframework/cloud/consul/discovery/TtlScheduler.java Normal file
View File

@@ -0,0 +1,126 @@
package org.springframework.cloud.consul.discovery;
import com.ecwid.consul.v1.ConsulClient;
import com.ecwid.consul.v1.agent.model.NewService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import javax.annotation.PreDestroy;
import java.util.Collections;
import java.util.Set;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
/**
* Created by nicu on 11.03.2015.
*/
@Slf4j
public class TtlScheduler implements ApplicationContextAware {
private static final AtomicInteger _instances = new AtomicInteger();
private static final int DEFAULT_TTL = 3; // must be > 1
public static final int HEARTBEAT_INTERVAL_RATIO = 2 / 3;
private final ScheduledExecutorService ttlExecutor;
private volatile ScheduledFuture<?> scheduledFuture;
private final Thread shutdownThread;
private final Set<String> serviceIds;
private final AtomicBoolean isShuttingDown;
private final Runnable ttlPingThread;
private volatile int ttl;
private volatile int heartbeatInterval;
@Autowired
private ConsulClient client;
public TtlScheduler() {
if (_instances.addAndGet(1) > 1) {
throw new IllegalStateException("Expecting this to be used in singleton mode!");
}
//one thread to refresh ttl for each registered app to local agent is ok
ttlExecutor = Executors.newSingleThreadScheduledExecutor();
shutdownThread = new Thread(new Runnable() {
public void run() {
log.info("Shutting down the Executor Pool for TtlScheduler");
shutdownExecutorPool();
}
});
Runtime.getRuntime().addShutdownHook(shutdownThread);
serviceIds = Collections.newSetFromMap(new ConcurrentHashMap<String, Boolean>());
isShuttingDown = new AtomicBoolean();
ttlPingThread = new TtlPingThread();
}
@Override
public void setApplicationContext(ApplicationContext context) throws BeansException {
ttl = context.getEnvironment().getProperty("consul.ttl", Integer.class, DEFAULT_TTL);
ttl = Math.min(2, ttl);
// heartbeat at 2/3 ttl, but no later than ttl -1s and, (under lesser priority), no sooner than 1s from now
heartbeatInterval = Math.max(ttl - 1, Math.min(ttl * HEARTBEAT_INTERVAL_RATIO, 1));
scheduleTtlHeartbeat();
}
/**
* Add a service to the checks loop.
*/
public void add(final NewService service) {
serviceIds.add(service.getId());
}
public void remove(String serviceId) {
serviceIds.remove(serviceId);
}
public int getTTL() {
return ttl;
}
private void schedule(int millis, Runnable command) {
ttlExecutor.schedule(command, millis, TimeUnit.MILLISECONDS);
}
private void scheduleTtlHeartbeat() {
scheduledFuture = ttlExecutor.scheduleAtFixedRate(
ttlPingThread,
0, heartbeatInterval,
TimeUnit.SECONDS);
}
private void shutdownExecutorPool() {
isShuttingDown.set(true);
ttlExecutor.shutdown();
try {
Runtime.getRuntime().removeShutdownHook(shutdownThread);
} catch (IllegalStateException ignored) {
}
}
class TtlPingThread implements Runnable {
public void run() {
try {
heartbeatServices();
} catch (Throwable e) {
log.error("Exception while trying to send heartbeat from application to consul local agent in due TTL", e);
}
}
}
void heartbeatServices() {
for (String serviceId : serviceIds) {
if (!isShuttingDown.get()) {
client.agentCheckPass(serviceId);
}
}
}
@PreDestroy
public void shutdown() {
if (scheduledFuture != null) {
scheduledFuture.cancel(true);
}
}
}

39
spring-cloud-consul-utils/pom.xml Normal file
View File

@@ -0,0 +1,39 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<artifactId>spring-cloud-consul</artifactId>
<groupId>org.springframework.cloud</groupId>
<version>1.0.0.BUILD-SNAPSHOT</version>
<relativePath>../spring-cloud-consul-sample/pom.xml</relativePath>
</parent>
<modelVersion>4.0.0</modelVersion>
<name>Spring Cloud Consul Utils</name>
<artifactId>spring-cloud-consul-utils</artifactId>
<dependencies>
<dependency>
<groupId>com.ecwid.consul</groupId>
<artifactId>consul-api</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-consul-config</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-consul-discovery</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-consul-bus</artifactId>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
</dependency>
</dependencies>
</project>

58
spring-cloud-consul-utils/src/main/java/org/springframework/cloud/consul/alive/AliveFilteringContext.java Normal file
View File

@@ -0,0 +1,58 @@
package org.springframework.cloud.consul.alive;
import com.netflix.loadbalancer.Server;
import com.netflix.loadbalancer.ServerListFilter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.consul.discovery.ConsulServer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;
import java.util.ArrayList;
import java.util.List;
/**
* Injects a server list filter for giving servers hosting a service only for the live servers per serf status.
* @author nicu marasoiu on 10.03.2015.
*/
@Configuration
@ComponentScan
public class AliveFilteringContext {
@Bean
@Autowired
public ServerListFilter<Server> aliveServerListFilter(FilteringAgentClient filteringAgentClient) {
return adapter(new AliveServerListFilter(filteringAgentClient));
}
@Bean
@Autowired
public ServerListFilter<Server> ttlServerListFilter() {
return adapter(new ServiceCheckServerListFilter());
}
private ServerListFilter<Server> adapter(final ServerListFilter<ConsulServer> consulServerList) {
return new ServerListFilter<Server>() {
@Override
public List<Server> getFilteredListOfServers(List<Server> servers) {
return adapt2(consulServerList.getFilteredListOfServers(adapt1(servers)));
}
private List<ConsulServer> adapt1(List<Server> servers) {
List<ConsulServer> consulServers = new ArrayList<ConsulServer>(servers.size());
for (Server consulServer : servers) {
consulServers.add((ConsulServer) consulServer);
}
return consulServers;
}
private List<Server> adapt2(List<ConsulServer> consulServers) {
List<Server> servers = new ArrayList<Server>(consulServers.size());
for (ConsulServer consulServer : consulServers) {
servers.add(consulServer);
}
return servers;
}
};
}
}

37
spring-cloud-consul-utils/src/main/java/org/springframework/cloud/consul/alive/AliveServerListFilter.java Normal file
View File

@@ -0,0 +1,37 @@
package org.springframework.cloud.consul.alive;
import com.netflix.loadbalancer.ServerListFilter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.consul.discovery.ConsulServer;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;
/**
* Server filter: returns only alive servers.
* Each consul agent runs a serf agent which is a member of the serf gossip pool.
* The serf status (alive/failed/etc) is reflected in 2 consul APIs: in the agent API and in the catalog API.
* We prefer the agent API because it is most up to date (or perhaps we should intersect them and pick members that are liv in both).
* @author nicu marasoiu on 10.03.2015.
*/
public class AliveServerListFilter implements ServerListFilter<ConsulServer>{
private FilteringAgentClient filteringAgentClient;
@Autowired
public AliveServerListFilter(FilteringAgentClient filteringAgentClient) {
this.filteringAgentClient = filteringAgentClient;
}
@Override
public List<ConsulServer> getFilteredListOfServers(List<ConsulServer> servers) {
Set<String> liveNodes = filteringAgentClient.getAliveAgentsAddresses();
List<ConsulServer> filteredServers = new ArrayList<>();
for (ConsulServer server : servers) {
if (liveNodes.contains(server.getAddress())) {
filteredServers.add(server);
}
}
return filteredServers;
}
}

21
spring-cloud-consul-utils/src/main/java/org/springframework/cloud/consul/alive/FilteringAgentClient.java Normal file
View File

@@ -0,0 +1,21 @@
package org.springframework.cloud.consul.alive;
import com.ecwid.consul.v1.agent.model.Member;
import java.util.List;
import java.util.Set;
/**
* A CatalogClient which decorates some methods of CatalogClient with filtering, retaining info pertaining to live nodes.
* @author nicu on 10.03.2015.
*/
public interface FilteringAgentClient {
/**
* @return the set of alive gossip pool members (client or server consul agents).
*/
List<Member> getAliveAgents();
/**
* @return the set of alive gossip pool members addresses.
*/
Set<String> getAliveAgentsAddresses();
}

44
spring-cloud-consul-utils/src/main/java/org/springframework/cloud/consul/alive/FilteringAgentClientImpl.java Normal file
View File

@@ -0,0 +1,44 @@
package org.springframework.cloud.consul.alive;
import com.ecwid.consul.v1.ConsulClient;
import com.ecwid.consul.v1.agent.model.Member;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.consul.model.SerfStatusEnum;
import org.springframework.stereotype.Service;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
@Service
public class FilteringAgentClientImpl implements FilteringAgentClient {
public static final int ALIVE_STATUS = SerfStatusEnum.StatusAlive.getCode();
private final ConsulClient client;
@Autowired
public FilteringAgentClientImpl(ConsulClient client) {
this.client = client;
}
@Override
public List<Member> getAliveAgents() {
List<Member> members = client.getAgentMembers().getValue();
List<Member> liveMembers = new ArrayList<>(members.size());
for (Member peer : members) {
if (peer.getStatus() == ALIVE_STATUS) {
liveMembers.add(peer);
}
}
return liveMembers;
}
@Override
public Set<String> getAliveAgentsAddresses() {
Set<String> addresses = new HashSet<String>();
for (Member server : getAliveAgents()) {
addresses.add(server.getAddress());
}
return addresses;
}
}

62
spring-cloud-consul-utils/src/main/java/org/springframework/cloud/consul/alive/ServiceCheckServerListFilter.java Normal file
View File

@@ -0,0 +1,62 @@
package org.springframework.cloud.consul.alive;
import com.ecwid.consul.v1.ConsulClient;
import com.ecwid.consul.v1.QueryParams;
import com.ecwid.consul.v1.health.model.Check;
import com.netflix.loadbalancer.ServerListFilter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.consul.discovery.ConsulServer;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
/**
* Created by nicu on 12.03.2015.
*/
public class ServiceCheckServerListFilter implements ServerListFilter<ConsulServer> {
@Autowired
private ConsulClient client;
@Override
public List<ConsulServer> getFilteredListOfServers(List<ConsulServer> servers) {
Set<String> passingServiceIds = getPassingServiceIds(servers);
List<ConsulServer> okServers = new ArrayList<>(servers.size());
for(ConsulServer consulServer: servers){
String serviceId = consulServer.getMetaInfo().getInstanceId();
if(passingServiceIds.contains(serviceId)){
List<Check> nodeChecks = client.getHealthChecksForNode(consulServer.getNode(), QueryParams.DEFAULT).getValue();
boolean passingNodeChecks = true;
for(Check check: nodeChecks) {
if(check.getStatus() != Check.CheckStatus.PASSING) {
passingNodeChecks = false;
break;
}
}
if(passingNodeChecks){
okServers.add(consulServer);
}
}
}
return okServers;
}
private Set<String> getPassingServiceIds(List<ConsulServer> servers) {
Set<String> serviceIds = new HashSet<>(1);
for(ConsulServer server:servers){
serviceIds.add(server.getMetaInfo().getInstanceId());
}
for(String serviceId: serviceIds) {
List<Check> serviceChecks = client.getHealthChecksForService(serviceId, QueryParams.DEFAULT).getValue();
for(Check check: serviceChecks){
if(check.getStatus() != Check.CheckStatus.PASSING) {
serviceIds.remove(check.getServiceId());
}
}
}
return serviceIds;
}
}