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

Merge branch 'nmarasoiu-both' into consul-client

Browse Source
This commit is contained in:
Spencer Gibb
2015-03-13 15:20:15 -06:00
parent a084b4b728 d8a6e7e9ad
commit 73da3b39d7
13 changed files with 344 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
pom.xml
View File

@@ -25,6 +25,13 @@
<tag>HEAD</tag>
</scm>
<repositories>
<repository>
<id>ecwid</id>
<url>http://nexus.ecwid.com/content/groups/public</url>
</repository>
</repositories>
<modules>
<module>spring-cloud-consul-core</module>
<module>spring-cloud-consul-config</module>
@@ -111,13 +118,19 @@
<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>
<artifactId>servlet-api</artifactId>
</exclusion>
</exclusions>
</dependency>
<!-- required by com.ecwid.consul but not as a pom dependency -->
<dependency>
<groupId>com.google.code.gson</groupId>
<artifactId>gson</artifactId>
<version>2.3.1</version>
</dependency>
<!-- force httpclient version -->
<dependency>
@@ -165,6 +178,11 @@
<artifactId>ribbon-httpclient</artifactId>
<version>${ribbon.version}</version>
</dependency>
<dependency>
<groupId>joda-time</groupId>
<artifactId>joda-time</artifactId>
<version>2.7</version>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>

5
spring-cloud-consul-core/pom.xml
View File

@@ -33,6 +33,11 @@
<groupId>com.ecwid.consul</groupId>
<artifactId>consul-api</artifactId>
</dependency>
<!-- required by com.ecwid.consul but not as a pom dependency -->
<dependency>
<groupId>com.google.code.gson</groupId>
<artifactId>gson</artifactId>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>

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;
}
}

4
spring-cloud-consul-discovery/pom.xml
View File

@@ -47,6 +47,10 @@
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>joda-time</groupId>
<artifactId>joda-time</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>

19
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,15 +10,24 @@ import org.springframework.context.annotation.Configuration;
*/
@Configuration
public class ConsulDiscoveryClientConfiguration {
@Autowired
private ConsulClient consulClient;
@Bean
public ConsulLifecycle consulLifecycle() {
return new ConsulLifecycle();
}
/*@Bean
public ConsulLoadBalancerClient consulLoadBalancerClient() {
return new ConsulLoadBalancerClient();
}*/
@Bean
public TtlScheduler ttlScheduler() {
return new TtlScheduler(heartbeatProperties(), consulClient);
}
@Bean
public HeartbeatProperties heartbeatProperties() {
return new HeartbeatProperties();
}
@Bean
public ConsulDiscoveryClient consulDiscoveryClient() {

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

@@ -19,6 +19,12 @@ public class ConsulLifecycle extends AbstractDiscoveryLifecycle {
@Autowired
private ConsulProperties consulProperties;
@Autowired
private TtlScheduler ttlScheduler;
@Autowired
private HeartbeatProperties ttlConfig;
@Override
protected void register() {
NewService service = new NewService();
@@ -30,8 +36,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(ttlConfig.getTtl());
service.setCheck(check);
register(service);
}
@@ -49,6 +56,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 +65,7 @@ public class ConsulLifecycle extends AbstractDiscoveryLifecycle {
}
@Override
protected void deregister(){
protected void deregister() {
deregister(getContext().getId());
}
@@ -67,6 +75,7 @@ public class ConsulLifecycle extends AbstractDiscoveryLifecycle {
}
private void deregister(String serviceId) {
ttlScheduler.remove(serviceId);
client.agentServiceDeregister(serviceId);
}

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

@@ -21,9 +21,12 @@ 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;
import org.springframework.cloud.consul.discovery.filters.ServiceCheckServerListFilter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@@ -69,6 +72,11 @@ public class ConsulRibbonClientConfiguration {
return serverList;
}
@Bean
public ServerListFilter<Server> ribbonServerListFilter() {
return new ServiceCheckServerListFilter(client);
}
@PostConstruct
public void preprocess() {
setProp(this.serviceId, DeploymentContextBasedVipAddresses.key(), this.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;
}
}

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

@@ -0,0 +1,40 @@
package org.springframework.cloud.consul.discovery;
import javax.annotation.PostConstruct;
import javax.validation.constraints.DecimalMax;
import javax.validation.constraints.DecimalMin;
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 HeartbeatProperties {
@Min(1)
private int ttlValue = 30;
@NotNull
private String ttlUnit = "s";
@DecimalMin("0.1")
@DecimalMax("0.9")
private double intervalRatio = 2.0 / 3.0;
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(ttlValue - 1,
Math.min(ttlValue * intervalRatio, 1))));
}
public String getTtl() {
return ttlValue + ttlUnit;
}
}

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

@@ -0,0 +1,58 @@
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.joda.time.DateTime;
import org.springframework.scheduling.annotation.Scheduled;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
/**
* Created by nicu on 11.03.2015.
*/
@Slf4j
public class TtlScheduler {
public static final DateTime EXPIRED_DATE = new DateTime(0);
private final Map<String, DateTime> serviceHeartbeats = new ConcurrentHashMap<>();
private HeartbeatProperties configuration;
private ConsulClient client;
public TtlScheduler(HeartbeatProperties configuration, ConsulClient client) {
this.configuration = configuration;
this.client = client;
}
/**
* Add a service to the checks loop.
*/
public void add(final NewService service) {
serviceHeartbeats.put(service.getId(), EXPIRED_DATE);
}
public void remove(String serviceId) {
serviceHeartbeats.remove(serviceId);
}
@Scheduled(initialDelay = 0, fixedRateString = "${consul.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.info("Sending consul heartbeat for: "+serviceId);
serviceHeartbeats.put(serviceId, DateTime.now());
}
}
}
}

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

@@ -0,0 +1,39 @@
package org.springframework.cloud.consul.discovery.filters;
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;
/**
* 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<Server> {
private FilteringAgentClient filteringAgentClient;
public AliveServerListFilter(FilteringAgentClient filteringAgentClient) {
this.filteringAgentClient = filteringAgentClient;
}
@Override
public List<Server> getFilteredListOfServers(List<Server> servers) {
Set<String> liveNodes = filteringAgentClient.getAliveAgentsAddresses();
List<Server> filteredServers = new ArrayList<>();
for (Server server : servers) {
ConsulServer consulServer = ConsulServer.class.cast(server);
if (liveNodes.contains(consulServer.getAddress())) {
filteredServers.add(server);
}
}
return filteredServers;
}
}

41
spring-cloud-consul-discovery/src/main/java/org/springframework/cloud/consul/discovery/filters/FilteringAgentClient.java Normal file
View File

@@ -0,0 +1,41 @@
package org.springframework.cloud.consul.discovery.filters;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import org.springframework.cloud.consul.model.SerfStatusEnum;
import com.ecwid.consul.v1.ConsulClient;
import com.ecwid.consul.v1.agent.model.Member;
public class FilteringAgentClient {
private static final int ALIVE_STATUS = SerfStatusEnum.StatusAlive.getCode();
private final ConsulClient client;
public FilteringAgentClient(ConsulClient client) {
this.client = client;
}
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;
}
public Set<String> getAliveAgentsAddresses() {
Set<String> addresses = new HashSet<>();
for (Member server : getAliveAgents()) {
addresses.add(server.getAddress());
}
return addresses;
}
}

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

@@ -0,0 +1,69 @@
package org.springframework.cloud.consul.discovery.filters;
import java.util.ArrayList;
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;
import com.ecwid.consul.v1.QueryParams;
import com.ecwid.consul.v1.health.model.Check;
import com.netflix.loadbalancer.ServerListFilter;
/**
* Created by nicu on 12.03.2015.
*/
public class ServiceCheckServerListFilter implements ServerListFilter<Server> {
private ConsulClient client;
public ServiceCheckServerListFilter(ConsulClient client) {
this.client = client;
}
@Override
public List<Server> getFilteredListOfServers(List<Server> servers) {
Set<String> passingServiceIds = getPassingServiceIds(servers);
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;
for (Check check : nodeChecks) {
if (check.getStatus() != Check.CheckStatus.PASSING) {
passingNodeChecks = false;
break;
}
}
if (passingNodeChecks) {
okServers.add(server);
}
}
}
return okServers;
}
private Set<String> getPassingServiceIds(List<Server> servers) {
Set<String> serviceIds = new HashSet<>(1);
for (Server 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;
}
}