Apache
/
druid
mirror of https://github.com/apache/druid.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'master' into stop_task 

Conflicts:
	merger/src/test/java/com/metamx/druid/merger/coordinator/RemoteTaskRunnerTest.java
This commit is contained in:
Fangjin Yang 2013-03-26 14:14:23 -07:00
parent a57951c7d6 e785b87be0
commit 537f527960
18 changed files with 82 additions and 49 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
client/pom.xml
View File

@ -28,7 +28,7 @@
<parent>
<groupId>com.metamx</groupId>
<artifactId>druid</artifactId>
<version>0.3.28-SNAPSHOT</version>
<version>0.3.29-SNAPSHOT</version>
</parent>
<dependencies>

2
common/pom.xml
View File

@ -28,7 +28,7 @@
<parent>
<groupId>com.metamx</groupId>
<artifactId>druid</artifactId>
<version>0.3.28-SNAPSHOT</version>
<version>0.3.29-SNAPSHOT</version>
</parent>
<dependencies>

4
druid-services/pom.xml
View File

@ -24,11 +24,11 @@
<artifactId>druid-services</artifactId>
<name>druid-services</name>
<description>druid-services</description>
<version>0.3.28-SNAPSHOT</version>
<version>0.3.29-SNAPSHOT</version>
<parent>
<groupId>com.metamx</groupId>
<artifactId>druid</artifactId>
<version>0.3.28-SNAPSHOT</version>
<version>0.3.29-SNAPSHOT</version>
</parent>
<dependencies>

2
examples/pom.xml
View File

@ -28,7 +28,7 @@
<parent>
<groupId>com.metamx</groupId>
<artifactId>druid</artifactId>
<version>0.3.28-SNAPSHOT</version>
<version>0.3.29-SNAPSHOT</version>
</parent>
<modules>

2
examples/rand/pom.xml
View File

@ -9,7 +9,7 @@
<parent>
<groupId>com.metamx</groupId>
<artifactId>druid-examples</artifactId>
<version>0.3.28-SNAPSHOT</version>
<version>0.3.29-SNAPSHOT</version>
</parent>
<dependencies>

2
examples/twitter/pom.xml
View File

@ -9,7 +9,7 @@
<parent>
<groupId>com.metamx</groupId>
<artifactId>druid-examples</artifactId>
<version>0.3.28-SNAPSHOT</version>
<version>0.3.29-SNAPSHOT</version>
</parent>
<dependencies>

2
index-common/pom.xml
View File

@ -28,7 +28,7 @@
<parent>
<groupId>com.metamx</groupId>
<artifactId>druid</artifactId>
<version>0.3.28-SNAPSHOT</version>
<version>0.3.29-SNAPSHOT</version>
</parent>
<dependencies>

2
indexer/pom.xml
View File

@ -28,7 +28,7 @@
<parent>
<groupId>com.metamx</groupId>
<artifactId>druid</artifactId>
<version>0.3.28-SNAPSHOT</version>
<version>0.3.29-SNAPSHOT</version>
</parent>
<dependencies>

7
merger/pom.xml
View File

@ -28,7 +28,7 @@
<parent>
<groupId>com.metamx</groupId>
<artifactId>druid</artifactId>
<version>0.3.28-SNAPSHOT</version>
<version>0.3.29-SNAPSHOT</version>
</parent>
<dependencies>
@ -160,11 +160,6 @@
<groupId>com.github.sgroschupf</groupId>
<artifactId>zkclient</artifactId>
</dependency>
<dependency>
<groupId>org.apache.zookeeper</groupId>
<artifactId>zookeeper</artifactId>
<version>3.4.3</version>
</dependency>
<!-- Tests -->
<dependency>

9
merger/src/main/java/com/metamx/druid/merger/coordinator/scaling/SimpleResourceManagementStrategy.java
View File

@ -70,6 +70,15 @@ public class SimpleResourceManagementStrategy implements ResourceManagementStrat
@Override
public boolean doProvision(Collection<TaskRunnerWorkItem> pendingTasks, Collection<ZkWorker> zkWorkers)
{
if (zkWorkers.size() >= workerSetupdDataRef.get().getMaxNumWorkers()) {
log.info(
"Cannot scale anymore. Num workers = %d, Max num workers = %d",
zkWorkers.size(),
workerSetupdDataRef.get().getMaxNumWorkers()
);
return false;
}
List<String> workerNodeIds = autoScalingStrategy.ipToIdLookup(
Lists.newArrayList(
Iterables.transform(

12
merger/src/main/java/com/metamx/druid/merger/coordinator/setup/WorkerSetupData.java
View File

@ -20,8 +20,6 @@
package com.metamx.druid.merger.coordinator.setup;
import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
@ -35,6 +33,7 @@ public class WorkerSetupData
private final String minVersion;
private final int minNumWorkers;
private final int maxNumWorkers;
private final EC2NodeData nodeData;
private final GalaxyUserData userData;
@ -42,12 +41,14 @@ public class WorkerSetupData
public WorkerSetupData(
@JsonProperty("minVersion") String minVersion,
@JsonProperty("minNumWorkers") int minNumWorkers,
@JsonProperty("maxNumWorkers") int maxNumWorkers,
@JsonProperty("nodeData") EC2NodeData nodeData,
@JsonProperty("userData") GalaxyUserData userData
)
{
this.minVersion = minVersion;
this.minNumWorkers = minNumWorkers;
this.maxNumWorkers = maxNumWorkers;
this.nodeData = nodeData;
this.userData = userData;
}
@ -64,6 +65,12 @@ public class WorkerSetupData
return minNumWorkers;
}
@JsonProperty
public int getMaxNumWorkers()
{
return maxNumWorkers;
}
@JsonProperty
public EC2NodeData getNodeData()
{
@ -82,6 +89,7 @@ public class WorkerSetupData
return "WorkerSetupData{" +
"minVersion='" + minVersion + '\'' +
", minNumWorkers=" + minNumWorkers +
", maxNumWorkers=" + maxNumWorkers +
", nodeData=" + nodeData +
", userData=" + userData +
'}';

2
merger/src/test/java/com/metamx/druid/merger/coordinator/RemoteTaskRunnerTest.java
View File

@ -331,7 +331,7 @@ public class RemoteTaskRunnerTest
pathChildrenCache,
scheduledExec,
new RetryPolicyFactory(new TestRetryPolicyConfig()),
new AtomicReference<WorkerSetupData>(new WorkerSetupData("0", 0, null, null)),
new AtomicReference<WorkerSetupData>(new WorkerSetupData("0", 0, 1, null, null)),
null
);

1
merger/src/test/java/com/metamx/druid/merger/coordinator/scaling/EC2AutoScalingStrategyTest.java
View File

@ -105,6 +105,7 @@ public class EC2AutoScalingStrategyTest
new WorkerSetupData(
"0",
0,
1,
new EC2NodeData(AMI_ID, INSTANCE_ID, 1, 1, Lists.<String>newArrayList(), "foo"),
new GalaxyUserData("env", "version", "type")
)

10
merger/src/test/java/com/metamx/druid/merger/coordinator/scaling/SimpleResourceManagementStrategyTest.java
View File

@ -60,7 +60,11 @@ public class SimpleResourceManagementStrategyTest
public void setUp() throws Exception
{
autoScalingStrategy = EasyMock.createMock(AutoScalingStrategy.class);
workerSetupData = new AtomicReference<WorkerSetupData>(null);
workerSetupData = new AtomicReference<WorkerSetupData>(
new WorkerSetupData(
"0", 0, 2, null, null
)
);
testTask = new TestTask(
"task1",
@ -251,7 +255,7 @@ public class SimpleResourceManagementStrategyTest
@Test
public void testDoSuccessfulTerminate() throws Exception
{
workerSetupData.set(new WorkerSetupData("0", 0, null, null));
workerSetupData.set(new WorkerSetupData("0", 0, 1, null, null));
EasyMock.expect(autoScalingStrategy.ipToIdLookup(EasyMock.<List<String>>anyObject()))
.andReturn(Lists.<String>newArrayList());
@ -281,7 +285,7 @@ public class SimpleResourceManagementStrategyTest
@Test
public void testSomethingTerminating() throws Exception
{
workerSetupData.set(new WorkerSetupData("0", 0, null, null));
workerSetupData.set(new WorkerSetupData("0", 0, 1, null, null));
EasyMock.expect(autoScalingStrategy.ipToIdLookup(EasyMock.<List<String>>anyObject()))
.andReturn(Lists.<String>newArrayList("ip")).times(2);

2
pom.xml
View File

@ -23,7 +23,7 @@
<groupId>com.metamx</groupId>
<artifactId>druid</artifactId>
<packaging>pom</packaging>
<version>0.3.28-SNAPSHOT</version>
<version>0.3.29-SNAPSHOT</version>
<name>druid</name>
<description>druid</description>
<scm>

2
realtime/pom.xml
View File

@ -28,7 +28,7 @@
<parent>
<groupId>com.metamx</groupId>
<artifactId>druid</artifactId>
<version>0.3.28-SNAPSHOT</version>
<version>0.3.29-SNAPSHOT</version>
</parent>
<properties>

2
server/pom.xml
View File

@ -28,7 +28,7 @@
<parent>
<groupId>com.metamx</groupId>
<artifactId>druid</artifactId>
<version>0.3.28-SNAPSHOT</version>
<version>0.3.29-SNAPSHOT</version>
</parent>
<dependencies>

66
server/src/main/java/com/metamx/druid/master/DruidMasterLogger.java
View File

@ -29,7 +29,6 @@ import com.metamx.druid.collect.CountingMap;
import com.metamx.emitter.service.ServiceEmitter;
import com.metamx.emitter.service.ServiceMetricEvent;
import javax.annotation.Nullable;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicLong;
@ -40,16 +39,20 @@ public class DruidMasterLogger implements DruidMasterHelper
{
private static final Logger log = new Logger(DruidMasterLogger.class);
private <T extends Number> void emitTieredStats(final ServiceEmitter emitter, final String formatString, final Map<String, T> statMap)
private <T extends Number> void emitTieredStats(
final ServiceEmitter emitter,
final String formatString,
final Map<String, T> statMap
)
{
if (statMap != null) {
if (statMap != null) {
for (Map.Entry<String, T> entry : statMap.entrySet()) {
String tier = entry.getKey();
Number value = entry.getValue();
emitter.emit(
new ServiceMetricEvent.Builder().build(
String.format(formatString, tier), value.doubleValue()
)
String.format(formatString, tier), value.doubleValue()
)
);
}
}
@ -82,30 +85,43 @@ public class DruidMasterLogger implements DruidMasterHelper
}
}
emitTieredStats(emitter, "master/%s/cost/raw",
stats.getPerTierStats().get("initialCost"));
emitTieredStats(
emitter, "master/%s/cost/raw",
stats.getPerTierStats().get("initialCost")
);
emitTieredStats(emitter, "master/%s/cost/normalization",
stats.getPerTierStats().get("normalization"));
emitTieredStats(
emitter, "master/%s/cost/normalization",
stats.getPerTierStats().get("normalization")
);
emitTieredStats(emitter, "master/%s/moved/count",
stats.getPerTierStats().get("movedCount"));
emitTieredStats(
emitter, "master/%s/moved/count",
stats.getPerTierStats().get("movedCount")
);
emitTieredStats(emitter, "master/%s/deleted/count",
stats.getPerTierStats().get("deletedCount"));
emitTieredStats(
emitter, "master/%s/deleted/count",
stats.getPerTierStats().get("deletedCount")
);
emitTieredStats(emitter, "master/%s/cost/normalized",
Maps.transformEntries(stats.getPerTierStats().get("normalizedInitialCostTimesOneThousand"),
new Maps.EntryTransformer<String, AtomicLong, Number>()
{
@Override
public Number transformEntry(
@Nullable String key, @Nullable AtomicLong value
)
{
return value.doubleValue() / 1000d;
}
}));
Map<String, AtomicLong> normalized = stats.getPerTierStats().get("normalizedInitialCostTimesOneThousand");
if (normalized != null) {
emitTieredStats(
emitter, "master/%s/cost/normalized",
Maps.transformEntries(
normalized,
new Maps.EntryTransformer<String, AtomicLong, Number>()
{
@Override
public Number transformEntry(String key, AtomicLong value)
{
return value.doubleValue() / 1000d;
}
}
)
);
}
Map<String, AtomicLong> unneeded = stats.getPerTierStats().get("unneededCount");
if (unneeded != null) {