Zoned ServiceInstanceListSupplier (#658)

* Get ServiceInstance zone from metadata.

* Add ZonePreferenceServiceInstanceListSupplier.

* Add javadocs and license entries.

* Add tests.

* Add documentation.

* Documentation fix.

* Fix after code review.

Author: OlgaMaciaszek

docs/src/main/asciidoc/spring-cloud-commons.adoc

@@ -831,19 +831,20 @@ that retrieves available instances from Service Discovery using a <<discovery-cl
 
 === Spring Cloud LoadBalancer integrations
 
-In order to make it easy to use Spring Cloud LoadBalancer, we provide `ReactorLoadBalancerExchangeFilterFunction` that can be used
-with `WebClient` and `BlockingLoadBalancerClient` that works with `RestTemplate`. You can see more information and examples of usage
-in the following sections:
+In order to make it easy to use Spring Cloud LoadBalancer, we provide `ReactorLoadBalancerExchangeFilterFunction` that can be used with `WebClient` and `BlockingLoadBalancerClient` that works with `RestTemplate`.
+You can see more information and examples of usage in the following sections:
 
 * <<rest-template-loadbalancer-client,Spring RestTemplate as a Load Balancer Client>>
 * <<webclinet-loadbalancer-client, Spring WebClient as a Load Balancer Client>>
 * <<webflux-with-reactive-loadbalancer,Spring WebFlux WebClient with `ReactorLoadBalancerExchangeFilterFunction`>>
 
+[[loadbalancer-caching]]
 === Spring Cloud LoadBalancer Caching
 
 Apart from the basic `ServiceInstanceListSupplier` implementation that retrieves instances via `DiscoveryClient` each time it has to choose an instance, we provide two caching implementations.
 
 ==== https://github.com/ben-manes/caffeine[Caffeine]-backed LoadBalancer Cache Implementation
+
 If you have `com.github.ben-manes.caffeine:caffeine` in the classpath, Caffeine-based implementation will be used.
 See the <<loadbalancer-cache-configuration, LoadBalancerCacheConfiguration>> section for information on how to configure it.
 
@@ -871,25 +872,66 @@ The default setup includes `ttl` set to 30 seconds and the default `initialCapac
 You can also altogether disable loadBalancer caching by setting the value of `spring.cloud.loadbalancer.cache.enabled`
 to `false`.
 
-WARNING: Although the basic, non-cached, implementation is useful for prototyping and testing, it's much less efficient
-than the cached versions, so we recommend always using the cached version in production.
+WARNING: Although the basic, non-cached, implementation is useful for prototyping and testing, it's much less efficient than the cached versions, so we recommend always using the cached version in production.
+
+=== Zone-Based Load-Balancing
+
+To enable zone-based load-balancing, we provide the `ZonePreferenceServiceInstanceListSupplier`.
+We use `DiscoveryClient`-specific `zone` configuration (for example, `eureka.instance.metadata-map.zone`) to pick the zone that the client tries to filter available service instances for.
+
+NOTE: You can also override `DiscoveryClient`-specific zone setup by setting the value of `spring.cloud.loadbalancer.zone` property.
+
+NOTE: To determine the zone of a retrieved `ServiceInstance`, we check the value under the `"zone"` key in its metadata map.
+
+The `ZonePreferenceServiceInstanceListSupplier` filters retrieved instances and only returns the ones within the same zone.
+If the zone is `null` or there are no instances within the same zone, it returns all the retrieved instances.
+
+In order to use the zone-based load-balancing approach, you will have to instantiate a `ZonePreferenceServiceInstanceListSupplier` bean in a <<custom-loadbalancer-configuration,custom configuration>>.
+
+We use delegates to work with `ServiceInstanceListSupplier` beans.
+We suggest passing a `DiscoveryClientServiceInstanceListSupplier` delegate in the constructor of `ZonePreferenceServiceInstanceListSupplier` and, in turn, wrapping the latter with a `CachingServiceInstanceListSupplier` to leverage <<loadbalancer-caching, LoadBalancer caching mechanism>>.
+
+You could use this sample configuration to set it up:
+
+[[zoned-based-custom-loadbalancer-configuration]]
+[source,java,indent=0]
+----
+public class CustomLoadBalancerConfiguration {
+
+	@Bean
+	public ServiceInstanceListSupplier discoveryClientServiceInstanceListSupplier(
+			ReactiveDiscoveryClient discoveryClient, Environment environment,
+			ApplicationContext context) {
+		DiscoveryClientServiceInstanceListSupplier firstDelegate = new DiscoveryClientServiceInstanceListSupplier(
+				discoveryClient, environment);
+		ZonePreferenceServiceInstanceListSupplier delegate = new ZonePreferenceServiceInstanceListSupplier(firstDelegate,
+				environment);
+		ObjectProvider<LoadBalancerCacheManager> cacheManagerProvider = context
+				.getBeanProvider(LoadBalancerCacheManager.class);
+		if (cacheManagerProvider.getIfAvailable() != null) {
+			return new CachingServiceInstanceListSupplier(delegate,
+					cacheManagerProvider.getIfAvailable());
+		}
+		return delegate;
+	}
+
+}
+----
 
 [[spring-cloud-loadbalancer-starter]]
 === Spring Cloud LoadBalancer Starter
 
 We also provide a starter that allows you to easily add Spring Cloud LoadBalancer in a Spring Boot app.
-In order to use it, just add `org.springframework.cloud:spring-cloud-starter-loadbalancer` to your Spring
-Cloud dependencies in your build file.
+In order to use it, just add `org.springframework.cloud:spring-cloud-starter-loadbalancer` to your Spring Cloud dependencies in your build file.
 
 NOTE: Spring Cloud LoadBalancer starter includes
 https://docs.spring.io/spring-boot/docs/current/reference/html/boot-features-caching.html[Spring Boot Caching]
 and https://github.com/stoyanr[Evictor].
 
-WARNING: If you have both Ribbon and Spring Cloud LoadBalancer int the classpath, in order to maintain
-backward compatibility, Ribbon-based implementations will be used by default. In order
-to switch to using Spring Cloud LoadBalancer under the hood,
-make sure you set the property `spring.cloud.loadbalancer.ribbon.enabled` to `false`.
+WARNING: If you have both Ribbon and Spring Cloud LoadBalancer int the classpath, in order to maintain backward compatibility, Ribbon-based implementations will be used by default.
+In order to switch to using Spring Cloud LoadBalancer under the hood, make sure you set the property `spring.cloud.loadbalancer.ribbon.enabled` to `false`.
 
+[[custom-loadbalancer-configuration]]
 === Passing Your Own Spring Cloud LoadBalancer Configuration
 
 You can also use the `@LoadBalancerClient` annotation to pass your own load-balancer client configuration, passing the name of the load-balancer client and the configuration class, as follows:
@@ -898,7 +940,7 @@ You can also use the `@LoadBalancerClient` annotation to pass your own load-bala
 [source,java,indent=0]
 ----
 @Configuration
-@LoadBalancerClient(value = "stores", configuration = StoresLoadBalancerClientConfiguration.class)
+@LoadBalancerClient(value = "stores", configuration = CustomLoadBalancerConfiguration.class)
 public class MyConfiguration {
 
 	@Bean
@@ -910,6 +952,13 @@ public class MyConfiguration {
 ----
 ====
 
+You can use this feature to instantiate different implementations of `ServiceInstanceListSupplier` or `ReactorLoadBalancer`,
+either written by you, or provided by us as alternatives (for example `ZonePreferenceServiceInstanceListSupplier`) to override the default setup.
+
+You can see an example of a custom cofiguration <<zoned-based-custom-loadbalancer-configuration,here>>.
+
+NOTE: The annotation `value` arguments (`stores` in the example above) specifies the service id of the service that we should send the requests to with the given custom configuration.
+
 You can also pass multiple configurations (for more than one load-balancer client) through the `@LoadBalancerClients` annotation, as the following example shows:
 
 ====

spring-cloud-loadbalancer/src/main/java/org/springframework/cloud/loadbalancer/core/ZonePreferenceServiceInstanceListSupplier.java

@@ -0,0 +1,93 @@
+/*
+ * Copyright 2012-2019 the original author or authors.
+ *
+ * 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
+ *
+ *      https://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 org.springframework.cloud.loadbalancer.core;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import reactor.core.publisher.Flux;
+
+import org.springframework.cloud.client.ServiceInstance;
+import org.springframework.core.env.Environment;
+
+/**
+ * An implementation of {@link ServiceInstanceListSupplier} that filters instances
+ * retrieved by the delegate by zone. The zone is retrieved from the
+ * <code>spring.cloud.loadbalancer.zone</code> property. If the zone is not set or no instances are found for the
+ * requested zone, all instances retrieved by the delegate are returned.
+ *
+ * @author Olga Maciaszek-Sharma
+ * @since 2.2.1
+ */
+public class ZonePreferenceServiceInstanceListSupplier
+		implements ServiceInstanceListSupplier {
+
+	private final String ZONE = "zone";
+
+	private final ServiceInstanceListSupplier delegate;
+
+	private final Environment environment;
+
+	private String zone;
+
+	public ZonePreferenceServiceInstanceListSupplier(ServiceInstanceListSupplier delegate,
+			Environment environment) {
+		this.delegate = delegate;
+		this.environment = environment;
+	}
+
+	@Override
+	public String getServiceId() {
+		return delegate.getServiceId();
+	}
+
+	@Override
+	public Flux<List<ServiceInstance>> get() {
+		return delegate.get().map(this::filteredByZone);
+	}
+
+	private List<ServiceInstance> filteredByZone(List<ServiceInstance> serviceInstances) {
+		if (zone == null) {
+			zone = environment.getProperty("spring.cloud.loadbalancer.zone");
+		}
+		if (zone != null) {
+			List<ServiceInstance> filteredInstances = new ArrayList<>();
+			for (ServiceInstance serviceInstance : serviceInstances) {
+				String instanceZone = getZone(serviceInstance);
+				if (zone.equalsIgnoreCase(instanceZone)) {
+					filteredInstances.add(serviceInstance);
+				}
+			}
+			if (filteredInstances.size() > 0) {
+				return filteredInstances;
+			}
+		}
+		// If the zone is not set or there are no zone-specific instances available,
+		// we return all instances retrieved for given service id.
+		return serviceInstances;
+	}
+
+	private String getZone(ServiceInstance serviceInstance) {
+		Map<String, String> metadata = serviceInstance.getMetadata();
+		if (metadata != null) {
+			return metadata.get(ZONE);
+		}
+		return null;
+	}
+
+}

spring-cloud-loadbalancer/src/test/java/org/springframework/cloud/loadbalancer/core/ZonePreferenceServiceInstanceListSupplierTests.java

@@ -0,0 +1,124 @@
+/*
+ * Copyright 2012-2019 the original author or authors.
+ *
+ * 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
+ *
+ *      https://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 org.springframework.cloud.loadbalancer.core;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.junit.jupiter.api.Test;
+import reactor.core.publisher.Flux;
+
+import org.springframework.cloud.client.DefaultServiceInstance;
+import org.springframework.cloud.client.ServiceInstance;
+import org.springframework.core.env.Environment;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * Tests for {@link ZonePreferenceServiceInstanceListSupplier}.
+ *
+ * @author Olga Maciaszek-Sharma
+ */
+class ZonePreferenceServiceInstanceListSupplierTests {
+
+	private DiscoveryClientServiceInstanceListSupplier delegate = mock(
+			DiscoveryClientServiceInstanceListSupplier.class);
+
+	private Environment environment = mock(Environment.class);
+
+	private ZonePreferenceServiceInstanceListSupplier supplier = new ZonePreferenceServiceInstanceListSupplier(
+			delegate, environment);
+
+	private ServiceInstance first = serviceInstance("test-1", buildZoneMetadata("zone1"));
+
+	private ServiceInstance second = serviceInstance("test-2",
+			buildZoneMetadata("zone1"));
+
+	private ServiceInstance third = serviceInstance("test-3", buildZoneMetadata("zone2"));
+
+	private ServiceInstance fourth = serviceInstance("test-4",
+			buildZoneMetadata("zone3"));
+
+	private ServiceInstance fifth = serviceInstance("test-5", buildZoneMetadata(null));
+
+	@Test
+	void shouldFilterInstancesByZone() {
+		when(environment.getProperty("spring.cloud.loadbalancer.zone"))
+				.thenReturn("zone1");
+		when(delegate.get()).thenReturn(Flux.just(Arrays.asList(first, second, third, fourth, fifth)));
+
+		List<ServiceInstance> filtered = supplier.get().blockFirst();
+
+		assertThat(filtered).hasSize(2);
+		assertThat(filtered).contains(first, second);
+		assertThat(filtered).doesNotContain(third);
+		assertThat(filtered).doesNotContain(fourth);
+		assertThat(filtered).doesNotContain(fifth);
+	}
+
+	@Test
+	void shouldReturnAllInstancesIfNoZoneInstances() {
+		when(environment.getProperty("spring.cloud.loadbalancer.zone"))
+				.thenReturn("zone1");
+		when(delegate.get()).thenReturn(Flux.just(Arrays.asList(third, fourth)));
+
+		List<ServiceInstance> filtered = supplier.get().blockFirst();
+
+		assertThat(filtered).hasSize(2);
+		assertThat(filtered).contains(third, fourth);
+	}
+
+	@Test
+	void shouldNotThrowNPEIfNullInstanceMetadata() {
+		when(environment.getProperty("spring.cloud.loadbalancer.zone"))
+				.thenReturn("zone1");
+		when(delegate.get()).thenReturn(
+				Flux.just(Collections.singletonList(serviceInstance("test-6", null))));
+		assertThatCode(() -> supplier.get().blockFirst()).doesNotThrowAnyException();
+	}
+
+	@Test
+	void shouldReturnAllInstancesIfNoZone() {
+		when(environment.getProperty("spring.cloud.loadbalancer.zone")).thenReturn(null);
+		when(delegate.get())
+				.thenReturn(Flux.just(Arrays.asList(first, second, third, fourth)));
+
+		List<ServiceInstance> filtered = supplier.get().blockFirst();
+
+		assertThat(filtered).hasSize(4);
+		assertThat(filtered).contains(first, second, third, fourth);
+	}
+
+	private DefaultServiceInstance serviceInstance(String instanceId,
+			Map<String, String> metadata) {
+		return new DefaultServiceInstance("test", instanceId, "http://test.test", 9080,
+				false, metadata);
+	}
+
+	private Map<String, String> buildZoneMetadata(String zone) {
+		Map<String, String> metadata = new HashMap<>();
+		metadata.put("zone", zone);
+		return metadata;
+	}
+
+}