diff --git a/server/src/main/java/io/druid/client/BrokerServerView.java b/server/src/main/java/io/druid/client/BrokerServerView.java index a60cfb18fea..3e860a4f23c 100644 --- a/server/src/main/java/io/druid/client/BrokerServerView.java +++ b/server/src/main/java/io/druid/client/BrokerServerView.java @@ -25,8 +25,9 @@ import com.google.common.collect.Ordering; import com.google.inject.Inject; import com.metamx.common.logger.Logger; import com.metamx.http.client.HttpClient; -import io.druid.client.selector.QueryableDruidServer; import io.druid.client.selector.ServerSelector; +import io.druid.client.selector.QueryableDruidServer; +import io.druid.client.selector.ServerSelectorStrategy; import io.druid.concurrent.Execs; import io.druid.guice.annotations.Client; import io.druid.query.QueryRunner; @@ -57,19 +58,22 @@ public class BrokerServerView implements TimelineServerView private final ObjectMapper smileMapper; private final HttpClient httpClient; private final ServerView baseView; + private final ServerSelectorStrategy serverSelectorStrategy; @Inject public BrokerServerView( QueryToolChestWarehouse warehose, ObjectMapper smileMapper, @Client HttpClient httpClient, - ServerView baseView + ServerView baseView, + ServerSelectorStrategy serverSelectorStrategy ) { this.warehose = warehose; this.smileMapper = smileMapper; this.httpClient = httpClient; this.baseView = baseView; + this.serverSelectorStrategy = serverSelectorStrategy; this.clients = Maps.newConcurrentMap(); this.selectors = Maps.newHashMap(); @@ -164,7 +168,7 @@ public class BrokerServerView implements TimelineServerView ServerSelector selector = selectors.get(segmentId); if (selector == null) { - selector = new ServerSelector(segment); + selector = new ServerSelector(segment, serverSelectorStrategy); VersionedIntervalTimeline timeline = timelines.get(segment.getDataSource()); if (timeline == null) { diff --git a/server/src/main/java/io/druid/client/CachingClusteredClient.java b/server/src/main/java/io/druid/client/CachingClusteredClient.java index 29a3b7e753d..31a644cf887 100644 --- a/server/src/main/java/io/druid/client/CachingClusteredClient.java +++ b/server/src/main/java/io/druid/client/CachingClusteredClient.java @@ -40,8 +40,8 @@ import com.metamx.common.guava.Sequence; import com.metamx.common.guava.Sequences; import com.metamx.emitter.EmittingLogger; import io.druid.client.cache.Cache; -import io.druid.client.selector.QueryableDruidServer; import io.druid.client.selector.ServerSelector; +import io.druid.client.selector.QueryableDruidServer; import io.druid.guice.annotations.Smile; import io.druid.query.BySegmentResultValueClass; import io.druid.query.CacheStrategy; diff --git a/server/src/main/java/io/druid/client/selector/ConnectionCountServerSelectorStrategy.java b/server/src/main/java/io/druid/client/selector/ConnectionCountServerSelectorStrategy.java new file mode 100644 index 00000000000..8a75e5403cf --- /dev/null +++ b/server/src/main/java/io/druid/client/selector/ConnectionCountServerSelectorStrategy.java @@ -0,0 +1,44 @@ +/* + * Druid - a distributed column store. + * Copyright (C) 2012, 2013 Metamarkets Group Inc. + * + * This program is free software; you can redistribute it and/or + * modify it under the terms of the GNU General Public License + * as published by the Free Software Foundation; either version 2 + * of the License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. + */ + +package io.druid.client.selector; + +import com.google.common.primitives.Ints; + +import java.util.Collections; +import java.util.Comparator; +import java.util.Set; + +public class ConnectionCountServerSelectorStrategy implements ServerSelectorStrategy +{ + private static final Comparator comparator = new Comparator() + { + @Override + public int compare(QueryableDruidServer left, QueryableDruidServer right) + { + return Ints.compare(left.getClient().getNumOpenConnections(), right.getClient().getNumOpenConnections()); + } + }; + + @Override + public QueryableDruidServer pick(Set servers) + { + return Collections.min(servers, comparator); + } +} diff --git a/server/src/main/java/io/druid/client/selector/RandomServerSelectorStrategy.java b/server/src/main/java/io/druid/client/selector/RandomServerSelectorStrategy.java new file mode 100644 index 00000000000..fbc4a429f2e --- /dev/null +++ b/server/src/main/java/io/druid/client/selector/RandomServerSelectorStrategy.java @@ -0,0 +1,40 @@ +/* + * Druid - a distributed column store. + * Copyright (C) 2012, 2013 Metamarkets Group Inc. + * + * This program is free software; you can redistribute it and/or + * modify it under the terms of the GNU General Public License + * as published by the Free Software Foundation; either version 2 + * of the License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. + */ + +package io.druid.client.selector; + +import com.google.common.collect.Iterators; + +import java.util.Random; +import java.util.Set; + +public class RandomServerSelectorStrategy implements ServerSelectorStrategy +{ + Random random; + + public RandomServerSelectorStrategy() { + this.random = new Random(); + } + + @Override + public QueryableDruidServer pick(Set servers) + { + return Iterators.get(servers.iterator(), random.nextInt(servers.size())); + } +} diff --git a/server/src/main/java/io/druid/client/selector/ServerSelector.java b/server/src/main/java/io/druid/client/selector/ServerSelector.java index d10c561c72d..3a7464a05c8 100644 --- a/server/src/main/java/io/druid/client/selector/ServerSelector.java +++ b/server/src/main/java/io/druid/client/selector/ServerSelector.java @@ -31,24 +31,18 @@ import java.util.Set; */ public class ServerSelector implements DiscoverySelector { - private static final Comparator comparator = new Comparator() - { - @Override - public int compare(QueryableDruidServer left, QueryableDruidServer right) - { - return Ints.compare(left.getClient().getNumOpenConnections(), right.getClient().getNumOpenConnections()); - } - }; - private final Set servers = Sets.newHashSet(); private final DataSegment segment; + private final ServerSelectorStrategy strategy; public ServerSelector( - DataSegment segment + DataSegment segment, + ServerSelectorStrategy strategy ) { this.segment = segment; + this.strategy = strategy; } public DataSegment getSegment() @@ -86,7 +80,7 @@ public class ServerSelector implements DiscoverySelector switch (size) { case 0: return null; case 1: return servers.iterator().next(); - default: return Collections.min(servers, comparator); + default: return strategy.pick(servers); } } } diff --git a/server/src/main/java/io/druid/client/selector/ServerSelectorStrategy.java b/server/src/main/java/io/druid/client/selector/ServerSelectorStrategy.java new file mode 100644 index 00000000000..d4684d28e08 --- /dev/null +++ b/server/src/main/java/io/druid/client/selector/ServerSelectorStrategy.java @@ -0,0 +1,35 @@ +/* + * Druid - a distributed column store. + * Copyright (C) 2012, 2013 Metamarkets Group Inc. + * + * This program is free software; you can redistribute it and/or + * modify it under the terms of the GNU General Public License + * as published by the Free Software Foundation; either version 2 + * of the License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. + */ + +package io.druid.client.selector; + +import com.fasterxml.jackson.annotation.JsonSubTypes; +import com.fasterxml.jackson.annotation.JsonTypeInfo; + +import java.util.Set; + +@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type", defaultImpl = RandomServerSelectorStrategy.class) +@JsonSubTypes(value = { + @JsonSubTypes.Type(name = "random", value = RandomServerSelectorStrategy.class), + @JsonSubTypes.Type(name = "connectionCount", value = ConnectionCountServerSelectorStrategy.class) +}) +public interface ServerSelectorStrategy +{ + public QueryableDruidServer pick(Set servers); +} diff --git a/server/src/test/java/io/druid/client/selector/ServerSelectorTest.java b/server/src/test/java/io/druid/client/selector/ServerSelectorTest.java index d66648beb68..fc5da720ea7 100644 --- a/server/src/test/java/io/druid/client/selector/ServerSelectorTest.java +++ b/server/src/test/java/io/druid/client/selector/ServerSelectorTest.java @@ -74,7 +74,8 @@ public class ServerSelectorTest new NoneShardSpec(), 0, 0L - ) + ), + new ConnectionCountServerSelectorStrategy() ); DirectDruidClient client1 = new DirectDruidClient( diff --git a/services/src/main/java/io/druid/cli/CliBroker.java b/services/src/main/java/io/druid/cli/CliBroker.java index ec37aefbe82..10d124a3a74 100644 --- a/services/src/main/java/io/druid/cli/CliBroker.java +++ b/services/src/main/java/io/druid/cli/CliBroker.java @@ -30,6 +30,7 @@ import io.druid.client.TimelineServerView; import io.druid.client.cache.Cache; import io.druid.client.cache.CacheMonitor; import io.druid.client.cache.CacheProvider; +import io.druid.client.selector.ServerSelectorStrategy; import io.druid.curator.discovery.DiscoveryModule; import io.druid.guice.Jerseys; import io.druid.guice.JsonConfigProvider; @@ -81,6 +82,8 @@ public class CliBroker extends ServerRunnable binder.bind(Cache.class).toProvider(CacheProvider.class).in(ManageLifecycle.class); JsonConfigProvider.bind(binder, "druid.broker.cache", CacheProvider.class); + JsonConfigProvider.bind(binder, "druid.broker.balancer", ServerSelectorStrategy.class); + binder.bind(QuerySegmentWalker.class).to(ClientQuerySegmentWalker.class).in(LazySingleton.class); binder.bind(JettyServerInitializer.class).to(QueryJettyServerInitializer.class).in(LazySingleton.class);