in services/src/main/java/org/apache/druid/cli/CliOverlord.java [197:473]
protected List<? extends Module> getModules(final boolean standalone)
{
return ImmutableList.of(
new Module()
{
@Override
public void configure(Binder binder)
{
validateCentralizedDatasourceSchemaConfig(properties);
if (standalone) {
binder.bindConstant()
.annotatedWith(Names.named("serviceName"))
.to(DEFAULT_SERVICE_NAME);
binder.bindConstant().annotatedWith(Names.named("servicePort")).to(8090);
binder.bindConstant().annotatedWith(Names.named("tlsServicePort")).to(8290);
JsonConfigProvider.bind(
binder,
CentralizedDatasourceSchemaConfig.PROPERTY_PREFIX,
CentralizedDatasourceSchemaConfig.class
);
binder.bind(CompactionStatusTracker.class).in(LazySingleton.class);
binder.bind(SegmentsMetadataManager.class)
.toProvider(SegmentsMetadataManagerProvider.class)
.in(ManageLifecycle.class);
binder.bind(CoordinatorConfigManager.class).in(LazySingleton.class);
}
JsonConfigProvider.bind(binder, "druid.coordinator.asOverlord", CoordinatorOverlordServiceConfig.class);
JsonConfigProvider.bind(binder, "druid.indexer.queue", TaskQueueConfig.class);
JsonConfigProvider.bind(binder, "druid.indexer.tasklock", TaskLockConfig.class);
JsonConfigProvider.bind(binder, "druid.indexer.task", TaskConfig.class);
JsonConfigProvider.bind(binder, "druid.indexer.task.default", DefaultTaskConfig.class);
binder.bind(RetryPolicyFactory.class).in(LazySingleton.class);
binder.bind(DruidOverlord.class).in(ManageLifecycle.class);
binder.bind(TaskMaster.class).in(ManageLifecycle.class);
binder.bind(TaskCountStatsProvider.class).to(TaskMaster.class);
binder.bind(TaskSlotCountStatsProvider.class).to(TaskMaster.class);
binder.bind(TaskLogStreamer.class)
.to(SwitchingTaskLogStreamer.class)
.in(LazySingleton.class);
binder.bind(new TypeLiteral<List<TaskLogStreamer>>() {})
.toProvider(new ListProvider<TaskLogStreamer>().add(TaskLogs.class))
.in(LazySingleton.class);
binder.bind(TaskLogStreamer.class)
.annotatedWith(Names.named("taskstreamer"))
.to(TaskRunnerTaskLogStreamer.class)
.in(LazySingleton.class);
binder.bind(TaskActionClientFactory.class).to(LocalTaskActionClientFactory.class).in(LazySingleton.class);
binder.bind(TaskActionToolbox.class).in(LazySingleton.class);
binder.bind(TaskLockbox.class).in(LazySingleton.class);
binder.bind(TaskQueryTool.class).in(LazySingleton.class);
binder.bind(IndexerMetadataStorageAdapter.class).in(LazySingleton.class);
binder.bind(CompactionScheduler.class).to(OverlordCompactionScheduler.class).in(ManageLifecycle.class);
binder.bind(ScheduledBatchTaskManager.class).in(LazySingleton.class);
binder.bind(SupervisorManager.class).in(LazySingleton.class);
binder.bind(ParallelIndexSupervisorTaskClientProvider.class).toProvider(Providers.of(null));
binder.bind(ShuffleClient.class).toProvider(Providers.of(null));
binder.bind(ChatHandlerProvider.class).toProvider(Providers.of(new NoopChatHandlerProvider()));
PolyBind.createChoice(
binder,
"druid.indexer.task.rowIngestionMeters.type",
Key.get(RowIngestionMetersFactory.class),
Key.get(DropwizardRowIngestionMetersFactory.class)
);
final MapBinder<String, RowIngestionMetersFactory> rowIngestionMetersHandlerProviderBinder =
PolyBind.optionBinder(binder, Key.get(RowIngestionMetersFactory.class));
rowIngestionMetersHandlerProviderBinder
.addBinding("dropwizard")
.to(DropwizardRowIngestionMetersFactory.class)
.in(LazySingleton.class);
binder.bind(DropwizardRowIngestionMetersFactory.class).in(LazySingleton.class);
PolyBind.optionBinder(binder, Key.get(TaskContextEnricher.class))
.addBinding(NoopTaskContextEnricher.TYPE)
.to(NoopTaskContextEnricher.class)
.in(LazySingleton.class);
PolyBind.createChoiceWithDefault(
binder,
"druid.indexer.task.contextenricher.type",
Key.get(TaskContextEnricher.class),
NoopTaskContextEnricher.TYPE
);
configureTaskStorage(binder);
configureIntermediaryData(binder);
configureAutoscale(binder);
binder.install(runnerConfigModule());
configureOverlordHelpers(binder);
if (standalone) {
binder.bind(RedirectFilter.class).in(LazySingleton.class);
binder.bind(RedirectInfo.class).to(OverlordRedirectInfo.class).in(LazySingleton.class);
binder.bind(JettyServerInitializer.class)
.to(OverlordJettyServerInitializer.class)
.in(LazySingleton.class);
}
Jerseys.addResource(binder, OverlordResource.class);
Jerseys.addResource(binder, SupervisorResource.class);
Jerseys.addResource(binder, HttpRemoteTaskRunnerResource.class);
Jerseys.addResource(binder, OverlordCompactionResource.class);
Jerseys.addResource(binder, OverlordDataSourcesResource.class);
binder.bind(AppenderatorsManager.class)
.to(DummyForInjectionAppenderatorsManager.class)
.in(LazySingleton.class);
if (standalone) {
LifecycleModule.register(binder, Server.class);
bindAnnouncer(
binder,
IndexingService.class,
DiscoverySideEffectsProvider.create()
);
}
Jerseys.addResource(binder, SelfDiscoveryResource.class);
LifecycleModule.registerKey(binder, Key.get(SelfDiscoveryResource.class));
binder.bind(LocalTmpStorageConfig.class)
.toProvider(new LocalTmpStorageConfig.DefaultLocalTmpStorageConfigProvider("overlord"))
.in(LazySingleton.class);
}
private void configureTaskStorage(Binder binder)
{
JsonConfigProvider.bind(binder, "druid.indexer.storage", TaskStorageConfig.class);
PolyBind.createChoice(
binder,
"druid.indexer.storage.type",
Key.get(TaskStorage.class),
Key.get(HeapMemoryTaskStorage.class)
);
final MapBinder<String, TaskStorage> storageBinder =
PolyBind.optionBinder(binder, Key.get(TaskStorage.class));
storageBinder.addBinding("local").to(HeapMemoryTaskStorage.class);
binder.bind(HeapMemoryTaskStorage.class).in(LazySingleton.class);
storageBinder.addBinding("metadata").to(MetadataTaskStorage.class).in(ManageLifecycle.class);
binder.bind(MetadataTaskStorage.class).in(LazySingleton.class);
}
private void configureIntermediaryData(Binder binder)
{
PolyBind.createChoice(
binder,
"druid.processing.intermediaryData.storage.type",
Key.get(IntermediaryDataManager.class),
Key.get(LocalIntermediaryDataManager.class)
);
final MapBinder<String, IntermediaryDataManager> biddy = PolyBind.optionBinder(
binder,
Key.get(IntermediaryDataManager.class)
);
biddy.addBinding("local").to(LocalIntermediaryDataManager.class);
biddy.addBinding("deepstore").to(DeepStorageIntermediaryDataManager.class).in(LazySingleton.class);
}
private Module runnerConfigModule()
{
return new Module()
{
@Override
public void configure(Binder binder)
{
JsonConfigProvider.bind(binder, "druid.worker", WorkerConfig.class);
PolyBind.createChoice(
binder,
"druid.indexer.runner.type",
Key.get(TaskRunnerFactory.class),
Key.get(HttpRemoteTaskRunnerFactory.class)
);
final MapBinder<String, TaskRunnerFactory> biddy = PolyBind.optionBinder(
binder,
Key.get(TaskRunnerFactory.class)
);
IndexingServiceModuleHelper.configureTaskRunnerConfigs(binder);
biddy.addBinding("local").to(ForkingTaskRunnerFactory.class);
binder.bind(ForkingTaskRunnerFactory.class).in(LazySingleton.class);
biddy.addBinding(RemoteTaskRunnerFactory.TYPE_NAME)
.to(RemoteTaskRunnerFactory.class)
.in(LazySingleton.class);
binder.bind(RemoteTaskRunnerFactory.class).in(LazySingleton.class);
biddy.addBinding(HttpRemoteTaskRunnerFactory.TYPE_NAME)
.to(HttpRemoteTaskRunnerFactory.class)
.in(LazySingleton.class);
binder.bind(HttpRemoteTaskRunnerFactory.class).in(LazySingleton.class);
JacksonConfigProvider.bind(binder, WorkerBehaviorConfig.CONFIG_KEY, WorkerBehaviorConfig.class, null);
JacksonConfigProvider.bind(
binder,
DruidCompactionConfig.CONFIG_KEY,
DruidCompactionConfig.class,
DruidCompactionConfig.empty()
);
}
@Provides
@ManageLifecycle
public TaskStorageDirTracker getTaskStorageDirTracker(WorkerConfig workerConfig, TaskConfig taskConfig)
{
return TaskStorageDirTracker.fromConfigs(workerConfig, taskConfig);
}
@Provides
@LazySingleton
@Named(ServiceStatusMonitor.HEARTBEAT_TAGS_BINDING)
public Supplier<Map<String, Object>> getHeartbeatSupplier(DruidOverlord overlord)
{
return () -> {
Map<String, Object> heartbeatTags = new HashMap<>();
heartbeatTags.put("leader", overlord.isLeader() ? 1 : 0);
return heartbeatTags;
};
}
};
}
private void configureAutoscale(Binder binder)
{
JsonConfigProvider.bind(binder, "druid.indexer.autoscale", ProvisioningSchedulerConfig.class);
JsonConfigProvider.bind(
binder,
"druid.indexer.autoscale",
PendingTaskBasedWorkerProvisioningConfig.class
);
JsonConfigProvider.bind(binder, "druid.indexer.autoscale", SimpleWorkerProvisioningConfig.class);
PolyBind.createChoice(
binder,
"druid.indexer.autoscale.strategy.type",
Key.get(ProvisioningStrategy.class),
Key.get(SimpleWorkerProvisioningStrategy.class)
);
final MapBinder<String, ProvisioningStrategy> biddy = PolyBind.optionBinder(
binder,
Key.get(ProvisioningStrategy.class)
);
biddy.addBinding("simple").to(SimpleWorkerProvisioningStrategy.class);
biddy.addBinding("pendingTaskBased").to(PendingTaskBasedWorkerProvisioningStrategy.class);
}
private void configureOverlordHelpers(Binder binder)
{
JsonConfigProvider.bind(binder, "druid.indexer.logs.kill", TaskLogAutoCleanerConfig.class);
Multibinder.newSetBinder(binder, OverlordDuty.class)
.addBinding()
.to(TaskLogAutoCleaner.class);
}
},
new IndexingServiceInputSourceModule(),
new IndexingServiceTaskLogsModule(),
new IndexingServiceTuningConfigModule(),
new InputSourceModule(),
new SupervisorModule(),
new LookupSerdeModule(),
new SamplerModule()
);
}