CAMEL-9683: A new toService EIP that uses a client discovery to lookup alive services and pick a service ip/port to use when calling the service from Camel route. Allows to plugin different providers.
Project: http://git-wip-us.apache.org/repos/asf/camel/repo Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/f42f6e3a Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/f42f6e3a Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/f42f6e3a Branch: refs/heads/kube-lb Commit: f42f6e3a5c0416d5a004d9e56f049c45de4a55be Parents: 7465a91 Author: Claus Ibsen <davscl...@apache.org> Authored: Tue Apr 26 12:23:45 2016 +0200 Committer: Claus Ibsen <davscl...@apache.org> Committed: Tue Apr 26 12:23:45 2016 +0200 ---------------------------------------------------------------------- .../processor/KubernetesProcessorFactory.java | 67 +++++++++++++++----- .../processor/RoundRobinBalancer.java | 44 +++++++++++++ .../processor/ServiceCallRouteTest.java | 2 + .../src/test/resources/log4j.properties | 2 +- 4 files changed, 98 insertions(+), 17 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/camel/blob/f42f6e3a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/processor/KubernetesProcessorFactory.java ---------------------------------------------------------------------- diff --git a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/processor/KubernetesProcessorFactory.java b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/processor/KubernetesProcessorFactory.java index f0b231b..05d979b 100644 --- a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/processor/KubernetesProcessorFactory.java +++ b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/processor/KubernetesProcessorFactory.java @@ -43,24 +43,25 @@ public class KubernetesProcessorFactory implements ProcessorFactory { } @Override + @SuppressWarnings("unchecked") public Processor createProcessor(RouteContext routeContext, ProcessorDefinition<?> definition) throws Exception { if (definition instanceof ServiceCallDefinition) { - ServiceCallDefinition ts = (ServiceCallDefinition) definition; + ServiceCallDefinition sc = (ServiceCallDefinition) definition; // discovery must either not be set, or if set then must be us - if (ts.getDiscovery() != null && !"kubernetes".equals(ts.getDiscovery())) { + if (sc.getDiscovery() != null && !"kubernetes".equals(sc.getDiscovery())) { return null; } - String name = ts.getName(); - String namespace = ts.getNamespace(); - String uri = ts.getUri(); - ExchangePattern mep = ts.getPattern(); + String name = sc.getName(); + String namespace = sc.getNamespace(); + String uri = sc.getUri(); + ExchangePattern mep = sc.getPattern(); - ServiceCallConfigurationDefinition config = ts.getServiceCallConfiguration(); + ServiceCallConfigurationDefinition config = sc.getServiceCallConfiguration(); ServiceCallConfigurationDefinition configRef = null; - if (ts.getServiceCallConfigurationRef() != null) { - configRef = CamelContextHelper.mandatoryLookup(routeContext.getCamelContext(), ts.getServiceCallConfigurationRef(), ServiceCallConfigurationDefinition.class); + if (sc.getServiceCallConfigurationRef() != null) { + configRef = CamelContextHelper.mandatoryLookup(routeContext.getCamelContext(), sc.getServiceCallConfigurationRef(), ServiceCallConfigurationDefinition.class); } // extract the properties from the configuration from the model @@ -80,16 +81,13 @@ public class KubernetesProcessorFactory implements ProcessorFactory { namespace = kc.getNamespace(); } - // lookup the load balancer to use - ServiceCallLoadBalancer lb = ts.getLoadBalancer(); - if (lb == null && ts.getServiceCallConfigurationRef() != null) { - lb = CamelContextHelper.mandatoryLookup(routeContext.getCamelContext(), ts.getLoadBalancerRef(), ServiceCallLoadBalancer.class); - } + // lookup the load balancer to use (configured on EIP takes precedence vs configured on configuration) + ServiceCallLoadBalancer lb = configureLoadBalancer(routeContext, sc); if (lb == null && config != null) { - lb = config.getLoadBalancer(); + lb = configureLoadBalancer(routeContext, config); } if (lb == null && configRef != null) { - lb = CamelContextHelper.mandatoryLookup(routeContext.getCamelContext(), configRef.getLoadBalancerRef(), ServiceCallLoadBalancer.class); + lb = configureLoadBalancer(routeContext, configRef); } KubernetesServiceCallProcessor processor = new KubernetesServiceCallProcessor(name, namespace, uri, mep, kc); @@ -100,4 +98,41 @@ public class KubernetesProcessorFactory implements ProcessorFactory { } } + private ServiceCallLoadBalancer configureLoadBalancer(RouteContext routeContext, ServiceCallDefinition sd) { + ServiceCallLoadBalancer lb = null; + + if (sd != null) { + lb = sd.getLoadBalancer(); + if (lb == null && sd.getLoadBalancerRef() != null) { + String ref = sd.getLoadBalancerRef(); + // special for ref is referring to built-in + if ("random".equalsIgnoreCase(ref)) { + lb = new RandomLoadBalancer(); + } else if ("roundrobin".equalsIgnoreCase(ref)) { + lb = new RoundRobinBalancer(); + } else { + lb = CamelContextHelper.mandatoryLookup(routeContext.getCamelContext(), ref, ServiceCallLoadBalancer.class); + } + } + } + + return lb; + } + + private ServiceCallLoadBalancer configureLoadBalancer(RouteContext routeContext, ServiceCallConfigurationDefinition config) { + ServiceCallLoadBalancer lb = config.getLoadBalancer(); + if (lb == null && config.getLoadBalancerRef() != null) { + String ref = config.getLoadBalancerRef(); + // special for ref is referring to built-in + if ("random".equalsIgnoreCase(ref)) { + lb = new RandomLoadBalancer(); + } else if ("roundrobin".equalsIgnoreCase(ref)) { + lb = new RoundRobinBalancer(); + } else { + lb = CamelContextHelper.mandatoryLookup(routeContext.getCamelContext(), ref, ServiceCallLoadBalancer.class); + } + } + return lb; + } + } http://git-wip-us.apache.org/repos/asf/camel/blob/f42f6e3a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/processor/RoundRobinBalancer.java ---------------------------------------------------------------------- diff --git a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/processor/RoundRobinBalancer.java b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/processor/RoundRobinBalancer.java new file mode 100644 index 0000000..2a2a401 --- /dev/null +++ b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/processor/RoundRobinBalancer.java @@ -0,0 +1,44 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 org.apache.camel.component.kubernetes.processor; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; + +import org.apache.camel.spi.ServiceCallLoadBalancer; + +public class RoundRobinBalancer implements ServiceCallLoadBalancer<Server> { + + private int counter = -1; + + @Override + public Server chooseServer(Collection<Server> servers) { + List<Server> list = new ArrayList<>(servers); + + int size = list.size(); + if (++counter >= size) { + counter = 0; + } + return list.get(counter); + } + + @Override + public String toString() { + return "RoundRobinBalancer"; + } +} http://git-wip-us.apache.org/repos/asf/camel/blob/f42f6e3a/components/camel-kubernetes/src/test/java/org/apache/camel/component/kubernetes/processor/ServiceCallRouteTest.java ---------------------------------------------------------------------- diff --git a/components/camel-kubernetes/src/test/java/org/apache/camel/component/kubernetes/processor/ServiceCallRouteTest.java b/components/camel-kubernetes/src/test/java/org/apache/camel/component/kubernetes/processor/ServiceCallRouteTest.java index b6b675a..ae26cb6 100644 --- a/components/camel-kubernetes/src/test/java/org/apache/camel/component/kubernetes/processor/ServiceCallRouteTest.java +++ b/components/camel-kubernetes/src/test/java/org/apache/camel/component/kubernetes/processor/ServiceCallRouteTest.java @@ -45,6 +45,8 @@ public class ServiceCallRouteTest extends CamelTestSupport { config.setUsername("admin"); config.setPassword("admin"); config.setNamespace("default"); + // lets use the built-in round robin (random is default) + config.setLoadBalancerRef("roundrobin"); from("direct:start") .serviceCall("cdi-camel-jetty", null, config) http://git-wip-us.apache.org/repos/asf/camel/blob/f42f6e3a/components/camel-kubernetes/src/test/resources/log4j.properties ---------------------------------------------------------------------- diff --git a/components/camel-kubernetes/src/test/resources/log4j.properties b/components/camel-kubernetes/src/test/resources/log4j.properties index 767860e..f88e05c 100644 --- a/components/camel-kubernetes/src/test/resources/log4j.properties +++ b/components/camel-kubernetes/src/test/resources/log4j.properties @@ -18,7 +18,7 @@ # # The logging properties used # -log4j.rootLogger=INFO, out +log4j.rootLogger=INFO, file log4j.logger.org.apache.camel.component.kubernetes=DEBUG
