Skip to content

Fixes for property conversion, firehose registration, and the indexing service #253

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
fjy merged 1 commit into from
Sep 28, 2013
Merged

Fixes for property conversion, firehose registration, and the indexing service #253

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions indexing-service/src/main/java/io/druid/indexing/coordinator/ForkingTaskRunner.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -200,12 +200,12 @@ public TaskStatus call()
command.add("io.druid.cli.Main");
command.add("internal");
command.add("peon");
command.add(taskFile.toString());
command.add(statusFile.toString());
String nodeType = task.getNodeType();
if (nodeType != null) {
command.add(String.format("--nodeType %s", nodeType));
}
command.add(taskFile.toString());
command.add(statusFile.toString());

jsonMapper.writeValue(taskFile, task);

Expand Down
1 change: 1 addition & 0 deletions server/src/main/java/io/druid/client/selector/Server.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,5 +25,6 @@ public interface Server
{
public String getScheme();
public String getHost();
public String getAddress();
public int getPort();
}
2 changes: 1 addition & 1 deletion server/src/main/java/io/druid/curator/discovery/CuratorServiceAnnouncer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -62,7 +62,7 @@ public void announce(DruidNode service)
try {
instance = ServiceInstance.<Void>builder()
.name(serviceName)
.address(service.getHost())
.address(service.getHostNoPort())
.port(service.getPort())
.build();
}
Expand Down
6 changes: 3 additions & 3 deletions server/src/main/java/io/druid/curator/discovery/DiscoveryModule.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -147,9 +147,9 @@ public CuratorServiceAnnouncer getServiceAnnouncer(
final Injector injector,
final Set<KeyHolder<DruidNode>> nodesToAnnounce,
final Lifecycle lifecycle
)
) throws Exception
{
lifecycle.addHandler(
lifecycle.addMaybeStartHandler(
new Lifecycle.Handler()
{
private volatile List<DruidNode> nodes = null;
Expand Down Expand Up @@ -203,7 +203,7 @@ public ServiceDiscovery<Void> getServiceDiscovery(
.client(curator)
.build();

lifecycle.addHandler(
lifecycle.addMaybeStartHandler(
new Lifecycle.Handler()
{
@Override
Expand Down
6 changes: 6 additions & 0 deletions server/src/main/java/io/druid/curator/discovery/ServerDiscoverySelector.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -63,6 +63,12 @@ public Server pick()
{
@Override
public String getHost()
{
return String.format("%s:%d", getAddress(), getPort());
}

@Override
public String getAddress()
{
return instance.getAddress();
}
Expand Down
38 changes: 37 additions & 1 deletion services/src/main/java/io/druid/cli/CliOverlord.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,8 @@

package io.druid.cli;

import com.fasterxml.jackson.databind.jsontype.NamedType;
import com.fasterxml.jackson.databind.module.SimpleModule;
import com.google.common.collect.ImmutableList;
import com.google.inject.Binder;
import com.google.inject.Injector;
Expand All @@ -28,6 +30,10 @@
import com.google.inject.multibindings.MapBinder;
import com.google.inject.servlet.GuiceFilter;
import com.metamx.common.logger.Logger;
import druid.examples.flights.FlightsFirehoseFactory;
import druid.examples.rand.RandomFirehoseFactory;
import druid.examples.twitter.TwitterSpritzerFirehoseFactory;
import druid.examples.web.WebFirehoseFactory;
import io.airlift.command.Command;
import io.druid.guice.IndexingServiceModuleHelper;
import io.druid.guice.JacksonConfigProvider;
Expand All @@ -41,6 +47,8 @@
import io.druid.indexing.common.actions.LocalTaskActionClientFactory;
import io.druid.indexing.common.actions.TaskActionClientFactory;
import io.druid.indexing.common.actions.TaskActionToolbox;
import io.druid.indexing.common.index.EventReceiverFirehoseFactory;
import io.druid.indexing.common.index.StaticS3FirehoseFactory;
import io.druid.indexing.common.tasklogs.SwitchingTaskLogStreamer;
import io.druid.indexing.common.tasklogs.TaskLogStreamer;
import io.druid.indexing.common.tasklogs.TaskLogs;
Expand Down Expand Up @@ -69,6 +77,12 @@
import io.druid.indexing.coordinator.scaling.SimpleResourceManagementConfig;
import io.druid.indexing.coordinator.scaling.SimpleResourceManagementStrategy;
import io.druid.indexing.coordinator.setup.WorkerSetupData;
import io.druid.initialization.DruidModule;
import io.druid.segment.realtime.firehose.ClippedFirehoseFactory;
import io.druid.segment.realtime.firehose.IrcFirehoseFactory;
import io.druid.segment.realtime.firehose.KafkaFirehoseFactory;
import io.druid.segment.realtime.firehose.RabbitMQFirehoseFactory;
import io.druid.segment.realtime.firehose.TimedShutoffFirehoseFactory;
import io.druid.server.http.RedirectFilter;
import io.druid.server.http.RedirectInfo;
import io.druid.server.initialization.JettyServerInitializer;
Expand All @@ -84,6 +98,7 @@
import org.eclipse.jetty.servlets.GzipFilter;
import org.eclipse.jetty.util.resource.ResourceCollection;

import java.util.Arrays;
import java.util.List;

/**
Expand All @@ -105,7 +120,7 @@ public CliOverlord()
protected List<Object> getModules()
{
return ImmutableList.<Object>of(
new Module()
new DruidModule()
{
@Override
public void configure(Binder binder)
Expand Down Expand Up @@ -199,6 +214,27 @@ private void configureAutoscale(Binder binder)

JsonConfigProvider.bind(binder, "druid.indexer.autoscale", SimpleResourceManagementConfig.class);
}

@Override
public List<? extends com.fasterxml.jackson.databind.Module> getJacksonModules()
{
return Arrays.<com.fasterxml.jackson.databind.Module>asList(
new SimpleModule("RealtimeModule")
.registerSubtypes(
new NamedType(TwitterSpritzerFirehoseFactory.class, "twitzer"),
new NamedType(FlightsFirehoseFactory.class, "flights"),
new NamedType(RandomFirehoseFactory.class, "rand"),
new NamedType(WebFirehoseFactory.class, "webstream"),
new NamedType(KafkaFirehoseFactory.class, "kafka-0.7.2"),
new NamedType(RabbitMQFirehoseFactory.class, "rabbitmq"),
new NamedType(ClippedFirehoseFactory.class, "clipped"),
new NamedType(TimedShutoffFirehoseFactory.class, "timed"),
new NamedType(IrcFirehoseFactory.class, "irc"),
new NamedType(StaticS3FirehoseFactory.class, "s3"),
new NamedType(EventReceiverFirehoseFactory.class, "receiver")
)
);
}
}
);
}
Expand Down
37 changes: 36 additions & 1 deletion services/src/main/java/io/druid/cli/CliPeon.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,8 @@

package io.druid.cli;

import com.fasterxml.jackson.databind.jsontype.NamedType;
import com.fasterxml.jackson.databind.module.SimpleModule;
import com.google.common.base.Throwables;
import com.google.common.collect.ImmutableList;
import com.google.inject.Binder;
Expand All @@ -28,6 +30,10 @@
import com.google.inject.multibindings.MapBinder;
import com.metamx.common.lifecycle.Lifecycle;
import com.metamx.common.logger.Logger;
import druid.examples.flights.FlightsFirehoseFactory;
import druid.examples.rand.RandomFirehoseFactory;
import druid.examples.twitter.TwitterSpritzerFirehoseFactory;
import druid.examples.web.WebFirehoseFactory;
import io.airlift.command.Arguments;
import io.airlift.command.Command;
import io.airlift.command.Option;
Expand All @@ -45,18 +51,26 @@
import io.druid.indexing.common.actions.TaskActionClientFactory;
import io.druid.indexing.common.config.TaskConfig;
import io.druid.indexing.common.index.ChatHandlerProvider;
import io.druid.indexing.common.index.EventReceiverFirehoseFactory;
import io.druid.indexing.common.index.EventReceivingChatHandlerProvider;
import io.druid.indexing.common.index.NoopChatHandlerProvider;
import io.druid.indexing.common.index.StaticS3FirehoseFactory;
import io.druid.indexing.coordinator.TaskRunner;
import io.druid.indexing.coordinator.ThreadPoolTaskRunner;
import io.druid.indexing.worker.executor.ChatHandlerResource;
import io.druid.indexing.worker.executor.ExecutorLifecycle;
import io.druid.indexing.worker.executor.ExecutorLifecycleConfig;
import io.druid.initialization.DruidModule;
import io.druid.query.QuerySegmentWalker;
import io.druid.segment.loading.DataSegmentKiller;
import io.druid.segment.loading.S3DataSegmentKiller;
import io.druid.segment.loading.SegmentLoaderConfig;
import io.druid.segment.loading.StorageLocationConfig;
import io.druid.segment.realtime.firehose.ClippedFirehoseFactory;
import io.druid.segment.realtime.firehose.IrcFirehoseFactory;
import io.druid.segment.realtime.firehose.KafkaFirehoseFactory;
import io.druid.segment.realtime.firehose.RabbitMQFirehoseFactory;
import io.druid.segment.realtime.firehose.TimedShutoffFirehoseFactory;
import io.druid.server.initialization.JettyServerInitializer;
import org.eclipse.jetty.server.Server;

Expand Down Expand Up @@ -90,7 +104,7 @@ public CliPeon()
protected List<Object> getModules()
{
return ImmutableList.<Object>of(
new Module()
new DruidModule()
{
@Override
public void configure(Binder binder)
Expand Down Expand Up @@ -143,6 +157,27 @@ public void configure(Binder binder)

LifecycleModule.register(binder, Server.class);
}

@Override
public List<? extends com.fasterxml.jackson.databind.Module> getJacksonModules()
{
return Arrays.<com.fasterxml.jackson.databind.Module>asList(
new SimpleModule("RealtimeModule")
.registerSubtypes(
new NamedType(TwitterSpritzerFirehoseFactory.class, "twitzer"),
new NamedType(FlightsFirehoseFactory.class, "flights"),
new NamedType(RandomFirehoseFactory.class, "rand"),
new NamedType(WebFirehoseFactory.class, "webstream"),
new NamedType(KafkaFirehoseFactory.class, "kafka-0.7.2"),
new NamedType(RabbitMQFirehoseFactory.class, "rabbitmq"),
new NamedType(ClippedFirehoseFactory.class, "clipped"),
new NamedType(TimedShutoffFirehoseFactory.class, "timed"),
new NamedType(IrcFirehoseFactory.class, "irc"),
new NamedType(StaticS3FirehoseFactory.class, "s3"),
new NamedType(EventReceiverFirehoseFactory.class, "receiver")
)
);
}
}
);
}
Expand Down
16 changes: 15 additions & 1 deletion services/src/main/java/io/druid/cli/CliRealtimeExample.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,9 +35,16 @@
import io.druid.client.ServerView;
import io.druid.guice.NoopSegmentPublisherProvider;
import io.druid.guice.RealtimeModule;
import io.druid.indexing.common.index.EventReceiverFirehoseFactory;
import io.druid.indexing.common.index.StaticS3FirehoseFactory;
import io.druid.initialization.DruidModule;
import io.druid.segment.loading.DataSegmentPusher;
import io.druid.segment.realtime.SegmentPublisher;
import io.druid.segment.realtime.firehose.ClippedFirehoseFactory;
import io.druid.segment.realtime.firehose.IrcFirehoseFactory;
import io.druid.segment.realtime.firehose.KafkaFirehoseFactory;
import io.druid.segment.realtime.firehose.RabbitMQFirehoseFactory;
import io.druid.segment.realtime.firehose.TimedShutoffFirehoseFactory;
import io.druid.server.coordination.DataSegmentAnnouncer;
import io.druid.timeline.DataSegment;

Expand Down Expand Up @@ -88,7 +95,14 @@ public List<Module> getJacksonModules()
new NamedType(TwitterSpritzerFirehoseFactory.class, "twitzer"),
new NamedType(FlightsFirehoseFactory.class, "flights"),
new NamedType(RandomFirehoseFactory.class, "rand"),
new NamedType(WebFirehoseFactory.class, "webstream")
new NamedType(WebFirehoseFactory.class, "webstream"),
new NamedType(KafkaFirehoseFactory.class, "kafka"),
new NamedType(RabbitMQFirehoseFactory.class, "rabbitmq"),
new NamedType(ClippedFirehoseFactory.class, "clipped"),
new NamedType(TimedShutoffFirehoseFactory.class, "timed"),
new NamedType(IrcFirehoseFactory.class, "irc"),
new NamedType(StaticS3FirehoseFactory.class, "s3"),
new NamedType(EventReceiverFirehoseFactory.class, "receiver")
)
);
}
Expand Down
10 changes: 7 additions & 3 deletions services/src/main/java/io/druid/cli/convert/ConvertProperties.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -85,6 +85,8 @@ public class ConvertProperties implements Runnable
new Rename("druid.indexer.terminateResources.duration", "druid.indexer.autoscale.terminatePeriod"),
new Rename("druid.indexer.terminateResources.originDateTime", "druid.indexer.autoscale.originTime"),
new Rename("druid.indexer.autoscaling.strategy", "druid.indexer.autoscale.strategy"),
new Rename("druid.indexer.logs.s3bucket", "druid.indexer.logs.s3Bucket"),
new Rename("druid.indexer.logs.s3prefix", "druid.indexer.logs.s3Prefix"),
new Rename("druid.indexer.maxWorkerIdleTimeMillisBeforeDeletion", "druid.indexer.autoscale.workerIdleTimeout"),
new Rename("druid.indexer.maxScalingDuration", "druid.indexer.autoscale.scalingTimeout"),
new Rename("druid.indexer.numEventsToTrack", "druid.indexer.autoscale.numEventsToTrack"),
Expand Down Expand Up @@ -122,7 +124,7 @@ public void run()
}

File outFile = new File(outFilename);
if (!outFile.getParentFile().exists()) {
if (outFile.getParentFile() != null && !outFile.getParentFile().exists()) {
outFile.getParentFile().mkdirs();
}

Expand All @@ -144,8 +146,10 @@ public void run()
for (PropertyConverter converter : converters) {
if (converter.canHandle(property)) {
for (Map.Entry<String, String> entry : converter.convert(fromFile).entrySet()) {
++count;
updatedProps.setProperty(entry.getKey(), entry.getValue());
if (entry.getValue() != null) {
++count;
updatedProps.setProperty(entry.getKey(), entry.getValue());
}
}
handled = true;
}
Expand Down
7 changes: 6 additions & 1 deletion services/src/main/java/io/druid/cli/convert/Rename.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,11 @@ public boolean canHandle(String property)
@Override
public Map<String, String> convert(Properties properties)
{
return ImmutableMap.of(newProperty, properties.getProperty(property));
final String value = properties.getProperty(property);
if (value != null) {
return ImmutableMap.of(newProperty, value);
} else {
return ImmutableMap.of();
}
}
}
22 changes: 21 additions & 1 deletion services/src/main/java/io/druid/guice/RealtimeModule.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,15 +26,25 @@
import com.google.inject.Key;
import com.google.inject.TypeLiteral;
import com.google.inject.multibindings.MapBinder;
import druid.examples.flights.FlightsFirehoseFactory;
import druid.examples.rand.RandomFirehoseFactory;
import druid.examples.twitter.TwitterSpritzerFirehoseFactory;
import druid.examples.web.WebFirehoseFactory;
import io.druid.cli.QueryJettyServerInitializer;
import io.druid.indexing.common.index.EventReceiverFirehoseFactory;
import io.druid.indexing.common.index.StaticS3FirehoseFactory;
import io.druid.initialization.DruidModule;
import io.druid.query.QuerySegmentWalker;
import io.druid.segment.realtime.DbSegmentPublisher;
import io.druid.segment.realtime.FireDepartment;
import io.druid.segment.realtime.NoopSegmentPublisher;
import io.druid.segment.realtime.RealtimeManager;
import io.druid.segment.realtime.SegmentPublisher;
import io.druid.segment.realtime.firehose.ClippedFirehoseFactory;
import io.druid.segment.realtime.firehose.IrcFirehoseFactory;
import io.druid.segment.realtime.firehose.KafkaFirehoseFactory;
import io.druid.segment.realtime.firehose.RabbitMQFirehoseFactory;
import io.druid.segment.realtime.firehose.TimedShutoffFirehoseFactory;
import io.druid.server.initialization.JettyServerInitializer;
import org.eclipse.jetty.server.Server;

Expand Down Expand Up @@ -80,7 +90,17 @@ public List<? extends Module> getJacksonModules()
return Arrays.<Module>asList(
new SimpleModule("RealtimeModule")
.registerSubtypes(
new NamedType(KafkaFirehoseFactory.class, "kafka-0.7.2")
new NamedType(TwitterSpritzerFirehoseFactory.class, "twitzer"),
new NamedType(FlightsFirehoseFactory.class, "flights"),
new NamedType(RandomFirehoseFactory.class, "rand"),
new NamedType(WebFirehoseFactory.class, "webstream"),
new NamedType(KafkaFirehoseFactory.class, "kafka-0.7.2"),
new NamedType(RabbitMQFirehoseFactory.class, "rabbitmq"),
new NamedType(ClippedFirehoseFactory.class, "clipped"),
new NamedType(TimedShutoffFirehoseFactory.class, "timed"),
new NamedType(IrcFirehoseFactory.class, "irc"),
new NamedType(StaticS3FirehoseFactory.class, "s3"),
new NamedType(EventReceiverFirehoseFactory.class, "receiver")
)
);
}
Expand Down