diff --git a/server/src/main/java/io/druid/client/BrokerServerView.java b/server/src/main/java/io/druid/client/BrokerServerView.java index be6a6553ae7..767d34a6a8e 100644 --- a/server/src/main/java/io/druid/client/BrokerServerView.java +++ b/server/src/main/java/io/druid/client/BrokerServerView.java @@ -27,7 +27,7 @@ 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.ServerSelectorStrategy; +import io.druid.client.selector.TierSelectorStrategy; import io.druid.concurrent.Execs; import io.druid.guice.annotations.Client; import io.druid.query.DataSource; @@ -61,7 +61,7 @@ public class BrokerServerView implements TimelineServerView private final ObjectMapper smileMapper; private final HttpClient httpClient; private final ServerView baseView; - private final ServerSelectorStrategy serverSelectorStrategy; + private final TierSelectorStrategy tierSelectorStrategy; @Inject public BrokerServerView( @@ -69,14 +69,14 @@ public class BrokerServerView implements TimelineServerView ObjectMapper smileMapper, @Client HttpClient httpClient, ServerView baseView, - ServerSelectorStrategy serverSelectorStrategy + TierSelectorStrategy tierSelectorStrategy ) { this.warehouse = warehouse; this.smileMapper = smileMapper; this.httpClient = httpClient; this.baseView = baseView; - this.serverSelectorStrategy = serverSelectorStrategy; + this.tierSelectorStrategy = tierSelectorStrategy; this.clients = Maps.newConcurrentMap(); this.selectors = Maps.newHashMap(); @@ -171,7 +171,7 @@ public class BrokerServerView implements TimelineServerView ServerSelector selector = selectors.get(segmentId); if (selector == null) { - selector = new ServerSelector(segment, serverSelectorStrategy); + selector = new ServerSelector(segment, tierSelectorStrategy); VersionedIntervalTimeline timeline = timelines.get(segment.getDataSource()); if (timeline == null) { diff --git a/server/src/main/java/io/druid/client/selector/AbstractTierSelectorStrategy.java b/server/src/main/java/io/druid/client/selector/AbstractTierSelectorStrategy.java new file mode 100644 index 00000000000..ad04d3a7fa2 --- /dev/null +++ b/server/src/main/java/io/druid/client/selector/AbstractTierSelectorStrategy.java @@ -0,0 +1,62 @@ +/* + * 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.metamx.common.ISE; +import io.druid.timeline.DataSegment; + +import java.util.Comparator; +import java.util.Map; +import java.util.Set; +import java.util.TreeMap; + +/** + */ +public abstract class AbstractTierSelectorStrategy implements TierSelectorStrategy +{ + private final ServerSelectorStrategy serverSelectorStrategy; + + public AbstractTierSelectorStrategy(ServerSelectorStrategy serverSelectorStrategy) + { + this.serverSelectorStrategy = serverSelectorStrategy; + } + + @Override + public QueryableDruidServer pick( + TreeMap> prioritizedServers, DataSegment segment + ) + { + final Map.Entry> priorityServers = prioritizedServers.pollFirstEntry(); + + if (priorityServers == null) { + return null; + } + + final Set servers = priorityServers.getValue(); + switch (servers.size()) { + case 0: + throw new ISE("[%s] Something hella weird going on here. We should not be here", segment.getIdentifier()); + case 1: + return priorityServers.getValue().iterator().next(); + default: + return serverSelectorStrategy.pick(servers, segment); + } + } +} diff --git a/server/src/main/java/io/druid/client/selector/ConnectionCountServerSelectorStrategy.java b/server/src/main/java/io/druid/client/selector/ConnectionCountServerSelectorStrategy.java index 4376ac01137..ea377a0f85d 100644 --- a/server/src/main/java/io/druid/client/selector/ConnectionCountServerSelectorStrategy.java +++ b/server/src/main/java/io/druid/client/selector/ConnectionCountServerSelectorStrategy.java @@ -20,14 +20,11 @@ package io.druid.client.selector; import com.google.common.primitives.Ints; -import com.metamx.common.ISE; import io.druid.timeline.DataSegment; import java.util.Collections; import java.util.Comparator; -import java.util.Map; import java.util.Set; -import java.util.TreeMap; public class ConnectionCountServerSelectorStrategy implements ServerSelectorStrategy { @@ -41,25 +38,8 @@ public class ConnectionCountServerSelectorStrategy implements ServerSelectorStra }; @Override - public QueryableDruidServer pick( - TreeMap> prioritizedServers, DataSegment segment - ) + public QueryableDruidServer pick(Set servers, DataSegment segment) { - final Map.Entry> highestPriorityServers = prioritizedServers.pollLastEntry(); - - if (highestPriorityServers == null) { - return null; - } - - final Set servers = highestPriorityServers.getValue(); - final int size = servers.size(); - switch (size) { - case 0: - throw new ISE("[%s] Something hella weird going on here. We should not be here", segment.getIdentifier()); - case 1: - return highestPriorityServers.getValue().iterator().next(); - default: - return Collections.min(servers, comparator); - } + return Collections.min(servers, comparator); } } diff --git a/server/src/main/java/io/druid/client/selector/CustomTierSelectorStrategy.java b/server/src/main/java/io/druid/client/selector/CustomTierSelectorStrategy.java new file mode 100644 index 00000000000..1ed750a0dc9 --- /dev/null +++ b/server/src/main/java/io/druid/client/selector/CustomTierSelectorStrategy.java @@ -0,0 +1,48 @@ +package io.druid.client.selector; + +import com.google.api.client.util.Maps; +import com.google.common.primitives.Ints; +import com.google.inject.Inject; + +import java.util.Comparator; +import java.util.Map; + +/** + */ +public class CustomTierSelectorStrategy extends AbstractTierSelectorStrategy +{ + private final Comparator comparator; + + @Inject + public CustomTierSelectorStrategy( + ServerSelectorStrategy serverSelectorStrategy, + CustomTierSelectorStrategyConfig config + ) + { + super(serverSelectorStrategy); + + final Map lookup = Maps.newHashMap(); + int pos = 0; + for (Integer integer : config.getPriorities()) { + lookup.put(integer, pos); + pos++; + } + + this.comparator = new Comparator() + { + @Override + public int compare(Integer o1, Integer o2) + { + int pos1 = lookup.get(o1); + int pos2 = lookup.get(o2); + return Ints.compare(pos1, pos2); + } + }; + } + + @Override + public Comparator getComparator() + { + return comparator; + } +} diff --git a/server/src/main/java/io/druid/client/selector/CustomTierSelectorStrategyConfig.java b/server/src/main/java/io/druid/client/selector/CustomTierSelectorStrategyConfig.java new file mode 100644 index 00000000000..d5bc0201c70 --- /dev/null +++ b/server/src/main/java/io/druid/client/selector/CustomTierSelectorStrategyConfig.java @@ -0,0 +1,38 @@ +/* + * 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.JsonProperty; +import com.google.api.client.util.Lists; + +import java.util.List; + +/** + */ +public class CustomTierSelectorStrategyConfig +{ + @JsonProperty + private List priorities = Lists.newArrayList(); + + public List getPriorities() + { + return priorities; + } +} diff --git a/server/src/main/java/io/druid/client/selector/HighestPriorityTierSelectorStrategy.java b/server/src/main/java/io/druid/client/selector/HighestPriorityTierSelectorStrategy.java new file mode 100644 index 00000000000..a2dd22ec792 --- /dev/null +++ b/server/src/main/java/io/druid/client/selector/HighestPriorityTierSelectorStrategy.java @@ -0,0 +1,51 @@ +/* + * 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 com.google.inject.Inject; + +import java.util.Comparator; + +/** + */ +public class HighestPriorityTierSelectorStrategy extends AbstractTierSelectorStrategy +{ + private static final Comparator comparator = new Comparator() + { + @Override + public int compare(Integer o1, Integer o2) + { + return -Ints.compare(o1, o2); + } + }; + + @Inject + public HighestPriorityTierSelectorStrategy(ServerSelectorStrategy serverSelectorStrategy) + { + super(serverSelectorStrategy); + } + + @Override + public Comparator getComparator() + { + return comparator; + } +} diff --git a/server/src/main/java/io/druid/client/selector/LowestPriorityTierSelectorStrategy.java b/server/src/main/java/io/druid/client/selector/LowestPriorityTierSelectorStrategy.java new file mode 100644 index 00000000000..00e6988a50e --- /dev/null +++ b/server/src/main/java/io/druid/client/selector/LowestPriorityTierSelectorStrategy.java @@ -0,0 +1,51 @@ +/* + * 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 com.google.inject.Inject; + +import java.util.Comparator; + +/** + */ +public class LowestPriorityTierSelectorStrategy extends AbstractTierSelectorStrategy +{ + private static final Comparator comparator = new Comparator() + { + @Override + public int compare(Integer o1, Integer o2) + { + return Ints.compare(o1, o2); + } + }; + + @Inject + public LowestPriorityTierSelectorStrategy(ServerSelectorStrategy serverSelectorStrategy) + { + super(serverSelectorStrategy); + } + + @Override + public Comparator getComparator() + { + return 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 index 0e295211484..9489851d6ad 100644 --- a/server/src/main/java/io/druid/client/selector/RandomServerSelectorStrategy.java +++ b/server/src/main/java/io/druid/client/selector/RandomServerSelectorStrategy.java @@ -20,36 +20,18 @@ package io.druid.client.selector; import com.google.common.collect.Iterators; -import com.metamx.common.ISE; import io.druid.timeline.DataSegment; -import java.util.Map; import java.util.Random; import java.util.Set; -import java.util.TreeMap; public class RandomServerSelectorStrategy implements ServerSelectorStrategy { private static final Random random = new Random(); @Override - public QueryableDruidServer pick(TreeMap> prioritizedServers, DataSegment segment) + public QueryableDruidServer pick(Set servers, DataSegment segment) { - final Map.Entry> highestPriorityServers = prioritizedServers.pollLastEntry(); - - if (highestPriorityServers == null) { - return null; - } - - final Set servers = highestPriorityServers.getValue(); - final int size = servers.size(); - switch (size) { - case 0: - throw new ISE("[%s] Something hella weird going on here. We should not be here", segment.getIdentifier()); - case 1: - return highestPriorityServers.getValue().iterator().next(); - default: - return Iterators.get(servers.iterator(), random.nextInt(size)); - } + 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 232a722e563..71c776d8ac4 100644 --- a/server/src/main/java/io/druid/client/selector/ServerSelector.java +++ b/server/src/main/java/io/druid/client/selector/ServerSelector.java @@ -37,11 +37,11 @@ public class ServerSelector implements DiscoverySelector private final Set servers = Sets.newHashSet(); private final DataSegment segment; - private final ServerSelectorStrategy strategy; + private final TierSelectorStrategy strategy; public ServerSelector( DataSegment segment, - ServerSelectorStrategy strategy + TierSelectorStrategy strategy ) { this.segment = segment; @@ -79,7 +79,7 @@ public class ServerSelector implements DiscoverySelector public QueryableDruidServer pick() { synchronized (this) { - TreeMap> prioritizedServers = Maps.newTreeMap(); + final TreeMap> prioritizedServers = new TreeMap<>(strategy.getComparator()); for (QueryableDruidServer server : servers) { Set theServers = prioritizedServers.get(server.getServer().getPriority()); if (theServers == null) { diff --git a/server/src/main/java/io/druid/client/selector/ServerSelectorStrategy.java b/server/src/main/java/io/druid/client/selector/ServerSelectorStrategy.java index 065253061d8..b7551a71356 100644 --- a/server/src/main/java/io/druid/client/selector/ServerSelectorStrategy.java +++ b/server/src/main/java/io/druid/client/selector/ServerSelectorStrategy.java @@ -24,7 +24,6 @@ import com.fasterxml.jackson.annotation.JsonTypeInfo; import io.druid.timeline.DataSegment; import java.util.Set; -import java.util.TreeMap; @JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type", defaultImpl = RandomServerSelectorStrategy.class) @JsonSubTypes(value = { @@ -33,5 +32,5 @@ import java.util.TreeMap; }) public interface ServerSelectorStrategy { - public QueryableDruidServer pick(TreeMap> prioritizedServers, DataSegment segment); + public QueryableDruidServer pick(Set servers, DataSegment segment); } diff --git a/server/src/main/java/io/druid/client/selector/TierSelectorStrategy.java b/server/src/main/java/io/druid/client/selector/TierSelectorStrategy.java new file mode 100644 index 00000000000..cadbd435478 --- /dev/null +++ b/server/src/main/java/io/druid/client/selector/TierSelectorStrategy.java @@ -0,0 +1,43 @@ +/* + * 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 io.druid.timeline.DataSegment; + +import java.util.Comparator; +import java.util.Set; +import java.util.TreeMap; + +/** + */ +@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type", defaultImpl = HighestPriorityTierSelectorStrategy.class) +@JsonSubTypes(value = { + @JsonSubTypes.Type(name = "highestPriority", value = HighestPriorityTierSelectorStrategy.class), + @JsonSubTypes.Type(name = "lowestPriority", value = LowestPriorityTierSelectorStrategy.class), + @JsonSubTypes.Type(name = "custom", value = CustomTierSelectorStrategy.class) +}) +public interface TierSelectorStrategy +{ + public Comparator getComparator(); + + public QueryableDruidServer pick(TreeMap> prioritizedServers, DataSegment segment); +} diff --git a/server/src/test/java/io/druid/client/CachingClusteredClientTest.java b/server/src/test/java/io/druid/client/CachingClusteredClientTest.java index 582d0c648cc..7e151f3818d 100644 --- a/server/src/test/java/io/druid/client/CachingClusteredClientTest.java +++ b/server/src/test/java/io/druid/client/CachingClusteredClientTest.java @@ -38,6 +38,7 @@ import com.metamx.common.guava.nary.TrinaryFn; import io.druid.client.cache.Cache; import io.druid.client.cache.CacheConfig; import io.druid.client.cache.MapCache; +import io.druid.client.selector.HighestPriorityTierSelectorStrategy; import io.druid.client.selector.QueryableDruidServer; import io.druid.client.selector.RandomServerSelectorStrategy; import io.druid.client.selector.ServerSelector; @@ -152,7 +153,7 @@ public class CachingClusteredClientTest "*", Arrays.asList( new FieldAccessPostAggregator("avg_imps_per_row", "avg_imps_per_row"), - new ConstantPostAggregator("constant", 2, 2 ) + new ConstantPostAggregator("constant", 2, 2) ) ), new ArithmeticPostAggregator( @@ -160,7 +161,7 @@ public class CachingClusteredClientTest "/", Arrays.asList( new FieldAccessPostAggregator("avg_imps_per_row", "avg_imps_per_row"), - new ConstantPostAggregator("constant", 2, 2 ) + new ConstantPostAggregator("constant", 2, 2) ) ) ); @@ -585,7 +586,8 @@ public class CachingClusteredClientTest } @Test - public void testTopNOnPostAggMetricCaching() { + public void testTopNOnPostAggMetricCaching() + { final TopNQueryBuilder builder = new TopNQueryBuilder() .dataSource(DATA_SOURCE) .dimension(TOP_DIM) @@ -914,7 +916,10 @@ public class CachingClusteredClientTest ); serverExpectations.get(lastServer).addExpectation(expectation); - ServerSelector selector = new ServerSelector(expectation.getSegment(), new RandomServerSelectorStrategy()); + ServerSelector selector = new ServerSelector( + expectation.getSegment(), + new HighestPriorityTierSelectorStrategy(new RandomServerSelectorStrategy()) + ); selector.addServer(new QueryableDruidServer(lastServer, null)); final PartitionChunk chunk; @@ -1097,16 +1102,16 @@ public class CachingClusteredClientTest (DateTime) objects[i], new TimeseriesResultValue( ImmutableMap.builder() - .put("rows", objects[i + 1]) - .put("imps", objects[i + 2]) - .put("impers", objects[i + 2]) - .put("avg_imps_per_row",avg_impr) - .put("avg_imps_per_row_half",avg_impr / 2) - .put("avg_imps_per_row_double",avg_impr * 2) - .build() - ) + .put("rows", objects[i + 1]) + .put("imps", objects[i + 2]) + .put("impers", objects[i + 2]) + .put("avg_imps_per_row", avg_impr) + .put("avg_imps_per_row_half", avg_impr / 2) + .put("avg_imps_per_row_double", avg_impr * 2) + .build() ) - ); + ) + ); } return retVal; } @@ -1186,14 +1191,14 @@ public class CachingClusteredClientTest final double rows = ((Number) objects[index + 1]).doubleValue(); values.add( ImmutableMap.builder() - .put(TOP_DIM, objects[index]) - .put("rows", rows) - .put("imps", imps) - .put("impers", imps) - .put("avg_imps_per_row", imps / rows) - .put("avg_imps_per_row_double", ((imps * 2) / rows)) - .put("avg_imps_per_row_half", (imps / (rows * 2))) - .build() + .put(TOP_DIM, objects[index]) + .put("rows", rows) + .put("imps", imps) + .put("impers", imps) + .put("avg_imps_per_row", imps / rows) + .put("avg_imps_per_row_double", ((imps * 2) / rows)) + .put("avg_imps_per_row_half", (imps / (rows * 2))) + .build() ); index += 3; } diff --git a/server/src/test/java/io/druid/client/DirectDruidClientTest.java b/server/src/test/java/io/druid/client/DirectDruidClientTest.java index da14f869f07..aba91657686 100644 --- a/server/src/test/java/io/druid/client/DirectDruidClientTest.java +++ b/server/src/test/java/io/druid/client/DirectDruidClientTest.java @@ -28,6 +28,7 @@ import com.metamx.http.client.HttpClient; import com.metamx.http.client.Request; import com.metamx.http.client.RequestBuilder; import io.druid.client.selector.ConnectionCountServerSelectorStrategy; +import io.druid.client.selector.HighestPriorityTierSelectorStrategy; import io.druid.client.selector.QueryableDruidServer; import io.druid.client.selector.ServerSelector; import io.druid.jackson.DefaultObjectMapper; @@ -87,7 +88,7 @@ public class DirectDruidClientTest 0, 0L ), - new ConnectionCountServerSelectorStrategy() + new HighestPriorityTierSelectorStrategy(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 308ed761414..c10a3b05eb2 100644 --- a/services/src/main/java/io/druid/cli/CliBroker.java +++ b/services/src/main/java/io/druid/cli/CliBroker.java @@ -31,7 +31,9 @@ import io.druid.client.cache.Cache; import io.druid.client.cache.CacheConfig; import io.druid.client.cache.CacheMonitor; import io.druid.client.cache.CacheProvider; +import io.druid.client.selector.CustomTierSelectorStrategyConfig; import io.druid.client.selector.ServerSelectorStrategy; +import io.druid.client.selector.TierSelectorStrategy; import io.druid.curator.discovery.DiscoveryModule; import io.druid.guice.Jerseys; import io.druid.guice.JsonConfigProvider; @@ -83,7 +85,9 @@ 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.cache", CacheConfig.class); - JsonConfigProvider.bind(binder, "druid.broker.balancer", ServerSelectorStrategy.class); + JsonConfigProvider.bind(binder, "druid.broker.select.tier", TierSelectorStrategy.class); + JsonConfigProvider.bind(binder, "druid.broker.select.tier.custom", CustomTierSelectorStrategyConfig.class); + JsonConfigProvider.bind(binder, "druid.broker.select.server", ServerSelectorStrategy.class); binder.bind(QuerySegmentWalker.class).to(ClientQuerySegmentWalker.class).in(LazySingleton.class);