[JAV-264]support zone aware load balance features

demo/demo-springmvc/springmvc-client/src/main/java/io/servicecomb/demo/springmvc/client/MicroserviceArray.java

@@ -0,0 +1,32 @@
+/*
+ * Copyright 2017 Huawei Technologies Co., Ltd
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.servicecomb.demo.springmvc.client;
+
+import io.servicecomb.serviceregistry.api.registry.Microservice;
+import java.util.List;
+
+public class MicroserviceArray {
+
+  private List<Microservice> services;
+
+  public List<Microservice> getServices() {
+    return services;
+  }
+
+  public void setServices(List<Microservice> services) {
+    this.services = services;
+  }
+}

demo/demo-springmvc/springmvc-client/src/main/java/io/servicecomb/demo/springmvc/client/ServiceCenterExample.java

@@ -0,0 +1,53 @@
+/*
+ * Copyright 2017 Huawei Technologies Co., Ltd
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.servicecomb.demo.springmvc.client;
+
+import java.net.URI;
+import org.springframework.http.HttpMethod;
+import org.springframework.http.RequestEntity;
+import org.springframework.http.ResponseEntity;
+import org.springframework.http.converter.json.MappingJackson2HttpMessageConverter;
+import org.springframework.util.LinkedMultiValueMap;
+import org.springframework.util.MultiValueMap;
+import org.springframework.web.client.RestTemplate;
+
+public class ServiceCenterExample {
+
+  public static void main(String[] args) throws Exception {
+    RestTemplate template = new RestTemplate();
+    template.getMessageConverters().add(new MappingJackson2HttpMessageConverter());
+    MultiValueMap<String, String> headers = new LinkedMultiValueMap<>();
+    headers.add("X-Tenant-Name", "default");
+
+    RequestEntity<String> requestEntity = new RequestEntity<String>(headers, HttpMethod.GET,
+        new URI("http://127.0.0.1:9980/registry/v3/microservices"));
+    ResponseEntity<String> stringResponseEntity = template.exchange(requestEntity, String.class);
+    System.out.println(stringResponseEntity.getBody());
+    ResponseEntity<MicroserviceArray> microseriveResponseEntity = template
+        .exchange(requestEntity, MicroserviceArray.class);
+    MicroserviceArray microserives = microseriveResponseEntity.getBody();
+    System.out.println(microserives.getServices().get(1).getServiceId());
+
+    // instance
+    headers.add("X-ConsumerId", microserives.getServices().get(1).getServiceId());
+    requestEntity = new RequestEntity<String>(headers, HttpMethod.GET,
+        new URI("http://127.0.0.1:9980/registry/v3/microservices/" + microserives.getServices().get(1).getServiceId()
+            + "/instances"));
+    ResponseEntity<String> microserviceInstanceResponseEntity = template.exchange(requestEntity, String.class);
+    System.out.println(microserviceInstanceResponseEntity.getBody());
+  }
+}

demo/demo-springmvc/springmvc-client/src/main/resources/microservice.yaml

@@ -17,7 +17,15 @@ cse:
     retryEnabled: true
     retryOnSame: 1
     retryOnNext: 1
-
+    serverListFilters: zoneaware
+    serverListFilter:
+      zoneaware:
+        className: io.servicecomb.loadbalance.filter.ZoneAwareServerListFilterExt
+  datacenter:
+    name: myDC
+    region: my-Region
+    availableZone: my-Zone
+
 #########SSL options
 ssl.protocols: TLSv1.2
 ssl.authPeer: true

demo/demo-springmvc/springmvc-server/src/main/resources/microservice.yaml

@@ -17,7 +17,10 @@ cse:
   tracing:
     enabled: true
     samplingRate: 0.5
-
+  datacenter:
+    name: myDC
+    region: my-Region
+    availableZone: my-Zone
 #########SSL options
 ssl.protocols: TLSv1.2
 ssl.authPeer: true

foundations/foundation-common/src/test/java/io/servicecomb/foundation/common/base/RetryableRunnableTest.java

@@ -24,6 +24,7 @@
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
@@ -62,20 +63,35 @@ public void retriesUnderlyingRunnableUntilSuccess() {
   @Test
   public void exitsWhenInterrupted() throws InterruptedException {
     ExecutorService executorService = Executors.newSingleThreadExecutor();
-    doThrow(exception).when(runnable).run();
+    CountDownLatch countDownLatch = new CountDownLatch(1);
+    DescriptiveRunnable runnable = new DescriptiveRunnable() {
+      @Override
+      public String description() {
+        return "runnable";
+      }
+
+      @Override
+      public void run() {
+        if(countDownLatch.getCount() == 1) {
+          countDownLatch.countDown();
+        }
+        throw new RuntimeException("oops");
+      }
+    };
+
+    RetryableRunnable retryableRunnable = new RetryableRunnable(runnable, 50);
 
     Future<?> retryable = executorService.submit(retryableRunnable);
     executorService.submit(blockedRunnable);
 
-    TimeUnit.MILLISECONDS.sleep(100);
+    countDownLatch.await();
 
     retryable.cancel(true);
 
-    TimeUnit.MILLISECONDS.sleep(100);
-
+    // this test have some pitfalls that can't shutdown the execution of retryableRunnable
     assertThat(retryable.isCancelled(), is(true));
 
     verify(blockedRunnable).run();
-    executorService.shutdown();
+    executorService.shutdownNow();
   }
 }

handlers/handler-loadbalance/src/main/java/io/servicecomb/loadbalance/filter/ZoneAwareServerListFilterExt.java

@@ -0,0 +1,72 @@
+/*
+ * Copyright 2017 Huawei Technologies Co., Ltd
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.servicecomb.loadbalance.filter;
+
+import com.netflix.loadbalancer.Server;
+import io.servicecomb.loadbalance.CseServer;
+import io.servicecomb.loadbalance.ServerListFilterExt;
+import io.servicecomb.serviceregistry.RegistryUtils;
+import io.servicecomb.serviceregistry.api.registry.MicroserviceInstance;
+import java.util.ArrayList;
+import java.util.List;
+
+public class ZoneAwareServerListFilterExt implements ServerListFilterExt {
+
+  @Override
+  public List<Server> getFilteredListOfServers(List<Server> list) {
+    List<Server> result = new ArrayList<>();
+    MicroserviceInstance myself = RegistryUtils.getMicroserviceInstance();
+    boolean find = false;
+    for (Server server : list) {
+      CseServer cseServer = (CseServer) server;
+      if (regionAndAZMatch(myself, cseServer.getInstance())) {
+        result.add(cseServer);
+        find = true;
+      }
+    }
+
+    if (!find) {
+      for (Server server : list) {
+        CseServer cseServer = (CseServer) server;
+        if (regionMatch(myself, cseServer.getInstance())) {
+          result.add(cseServer);
+          find = true;
+        }
+      }
+    }
+
+    if (!find) {
+      result = list;
+    }
+    return result;
+  }
+
+  private boolean regionAndAZMatch(MicroserviceInstance myself, MicroserviceInstance target) {
+    if (myself.getDataCenterInfo() != null && target.getDataCenterInfo() != null) {
+      return myself.getDataCenterInfo().getRegion().equals(target.getDataCenterInfo().getRegion()) &&
+          myself.getDataCenterInfo().getAvailableZone().equals(target.getDataCenterInfo().getAvailableZone());
+    }
+    return false;
+  }
+
+  private boolean regionMatch(MicroserviceInstance myself, MicroserviceInstance target) {
+    if (myself.getDataCenterInfo() != null && target.getDataCenterInfo() != null) {
+      return myself.getDataCenterInfo().getRegion().equals(target.getDataCenterInfo().getRegion());
+    }
+    return false;
+  }
+}

handlers/handler-loadbalance/src/test/java/io/servicecomb/loadbalance/filter/TestZoneAwareServerListFilterExt.java

@@ -0,0 +1,137 @@
+/*
+ * Copyright 2017 Huawei Technologies Co., Ltd
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.servicecomb.loadbalance.filter;
+
+import com.netflix.loadbalancer.Server;
+import io.servicecomb.loadbalance.CseServer;
+import io.servicecomb.serviceregistry.RegistryUtils;
+import io.servicecomb.serviceregistry.api.registry.DataCenterInfo;
+import io.servicecomb.serviceregistry.api.registry.MicroserviceInstance;
+import java.util.ArrayList;
+import java.util.List;
+import mockit.Expectations;
+import mockit.Mock;
+import mockit.MockUp;
+import mockit.Mocked;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class TestZoneAwareServerListFilterExt {
+
+  @Test
+  public void testZoneAwareServerListFilterExt(@Mocked RegistryUtils registryUtils) {
+    MicroserviceInstance myself = new MicroserviceInstance();
+    DataCenterInfo info = new DataCenterInfo();
+    info.setName("test");
+    info.setRegion("test-Region");
+    info.setAvailableZone("test-zone");
+    myself.setDataCenterInfo(info);
+
+    MicroserviceInstance allmatchInstance = new MicroserviceInstance();
+    info = new DataCenterInfo();
+    info.setName("test");
+    info.setRegion("test-Region");
+    info.setAvailableZone("test-zone");
+    allmatchInstance.setDataCenterInfo(info);
+
+    MicroserviceInstance regionMatchInstance = new MicroserviceInstance();
+    info = new DataCenterInfo();
+    info.setName("test");
+    info.setRegion("test-Region");
+    info.setAvailableZone("test-zone2");
+    regionMatchInstance.setDataCenterInfo(info);
+
+    MicroserviceInstance noneMatchInstance = new MicroserviceInstance();
+    info = new DataCenterInfo();
+    info.setName("test");
+    info.setRegion("test-Region2");
+    info.setAvailableZone("test-zone2");
+    noneMatchInstance.setDataCenterInfo(info);
+
+    new Expectations() {
+      {
+        RegistryUtils.getMicroserviceInstance();
+        result = myself;
+      }
+    };
+    ZoneAwareServerListFilterExt filter = new ZoneAwareServerListFilterExt();
+    List<Server> servers = new ArrayList<>();
+    CseServer noneMatchServer = new MockUp<CseServer>() {
+      @Mock
+      public String toString() {
+        return "noneMatchServer";
+      }
+
+      @Mock
+      public String getHost() {
+        return "noneMatchServer";
+      }
+
+      @Mock
+      public MicroserviceInstance getInstance() {
+        return noneMatchInstance;
+      }
+    }.getMockInstance();
+    CseServer regionMatchregionMatchServer = new MockUp<CseServer>() {
+      @Mock
+      public String toString() {
+        return "regionMatchregionMatchServer";
+      }
+
+      @Mock
+      public String getHost() {
+        return "regionMatchregionMatchServer";
+      }
+
+      @Mock
+      public MicroserviceInstance getInstance() {
+        return regionMatchInstance;
+      }
+    }.getMockInstance();
+
+    CseServer allmatchServer = new MockUp<CseServer>() {
+      @Mock
+      public String toString() {
+        return "allmatchServer";
+      }
+
+      @Mock
+      public String getHost() {
+        return "allmatchServer";
+      }
+
+      @Mock
+      public MicroserviceInstance getInstance() {
+        return allmatchInstance;
+      }
+    }.getMockInstance();
+
+    servers.add(noneMatchServer);
+    List<Server> result = filter.getFilteredListOfServers(servers);
+    Assert.assertEquals(result.size(), 1);
+    Assert.assertEquals(result.get(0), noneMatchServer);
+
+    servers.add(regionMatchregionMatchServer);
+    result = filter.getFilteredListOfServers(servers);
+    Assert.assertEquals(result.size(), 1);
+    Assert.assertEquals(result.get(0), regionMatchregionMatchServer);
+
+    servers.add(allmatchServer);
+    result = filter.getFilteredListOfServers(servers);
+    Assert.assertEquals(result.size(), 1);
+    Assert.assertEquals(result.get(0), allmatchServer);
+  }
+}