diff --git a/internal/venice-common/src/main/java/com/linkedin/venice/ConfigKeys.java b/internal/venice-common/src/main/java/com/linkedin/venice/ConfigKeys.java index 1957740f1e9..2485990bac1 100644 --- a/internal/venice-common/src/main/java/com/linkedin/venice/ConfigKeys.java +++ b/internal/venice-common/src/main/java/com/linkedin/venice/ConfigKeys.java @@ -2564,6 +2564,15 @@ private ConfigKeys() { */ public static final String IDENTITY_PARSER_CLASS = "identity.parser.class"; + /** + * Comma-separated list of {@link com.linkedin.venice.controller.ControllerPlugin} implementation class names. + * Each class must have a public constructor taking + * ({@code VeniceParentHelixAdmin}, {@code AuthorizerService}, {@code VeniceControllerMultiClusterConfig}). + * Plugins are instantiated via reflection during {@code VeniceController} construction and started later + * when the controller's {@code start()} method is invoked. Only applies to parent controllers. + */ + public static final String CONTROLLER_PLUGIN_CLASS_NAMES = "controller.plugin.class.names"; + /** * Specifies a list of partitioners venice supported. * It contains a string of concatenated partitioner class names separated by comma. diff --git a/services/venice-controller/src/main/java/com/linkedin/venice/controller/ControllerPlugin.java b/services/venice-controller/src/main/java/com/linkedin/venice/controller/ControllerPlugin.java new file mode 100644 index 00000000000..78fa29e0d1e --- /dev/null +++ b/services/venice-controller/src/main/java/com/linkedin/venice/controller/ControllerPlugin.java @@ -0,0 +1,33 @@ +package com.linkedin.venice.controller; + +import java.io.Closeable; + + +/** + * A pluggable service that runs on the parent Venice controller. + * Implementations are provided externally (e.g., by downstream projects) and registered one of two ways: + * + * + *

The lifecycle is: + *

    + *
  1. {@link ControllerPluginFactory#create} — called during controller construction
    2. + *
  2. {@link #start()} — called when the controller starts
    3. + *
  3. {@link #close()} — called when the controller stops
    4. + *
+ */ +public interface ControllerPlugin extends Closeable { + /** + * Called once during controller startup. May start background threads. + */ + void start(); + + /** + * Returns a human-readable name for logging. + */ + String getName(); +} diff --git a/services/venice-controller/src/main/java/com/linkedin/venice/controller/ControllerPluginFactory.java b/services/venice-controller/src/main/java/com/linkedin/venice/controller/ControllerPluginFactory.java new file mode 100644 index 00000000000..6b2ad80d817 --- /dev/null +++ b/services/venice-controller/src/main/java/com/linkedin/venice/controller/ControllerPluginFactory.java @@ -0,0 +1,29 @@ +package com.linkedin.venice.controller; + +import com.linkedin.venice.authorization.AuthorizerService; + + +/** + * Factory to create a {@link ControllerPlugin}. Receives all controller context needed to + * initialize the plugin: the parent admin for store enumeration and leadership checks, + * the authorizer service, and the multi-cluster config for reading plugin-specific properties. + * + *

Factories are registered via {@link VeniceControllerContext.Builder#setControllerPluginFactories} + * and invoked during controller construction. A factory may return {@code null} to indicate + * the plugin should not be created (e.g., when disabled by config). + */ +@FunctionalInterface +public interface ControllerPluginFactory { + /** + * @param admin the parent controller admin + * @param authorizerService the controller's authorizer, or {@code null} when controller authorization is + * disabled. Implementations that require it must null-check and return {@code null} + * (skip the plugin) rather than dereferencing it. + * @param config the multi-cluster config, for reading plugin-specific properties + * @return the plugin to run, or {@code null} to skip creation + */ + ControllerPlugin create( + VeniceParentHelixAdmin admin, + AuthorizerService authorizerService, + VeniceControllerMultiClusterConfig config); +} diff --git a/services/venice-controller/src/main/java/com/linkedin/venice/controller/VeniceController.java b/services/venice-controller/src/main/java/com/linkedin/venice/controller/VeniceController.java index 9b1062805d6..036e9e777c7 100644 --- a/services/venice-controller/src/main/java/com/linkedin/venice/controller/VeniceController.java +++ b/services/venice-controller/src/main/java/com/linkedin/venice/controller/VeniceController.java @@ -1,5 +1,6 @@ package com.linkedin.venice.controller; +import static com.linkedin.venice.ConfigKeys.CONTROLLER_PLUGIN_CLASS_NAMES; import static com.linkedin.venice.ConfigKeys.ZOOKEEPER_ADDRESS; import com.linkedin.d2.balancer.D2Client; @@ -55,6 +56,7 @@ import com.linkedin.venice.system.store.ControllerClientBackedSystemSchemaInitializer; import com.linkedin.venice.utils.LogContext; import com.linkedin.venice.utils.PropertyBuilder; +import com.linkedin.venice.utils.ReflectUtils; import com.linkedin.venice.utils.RegionUtils; import com.linkedin.venice.utils.SslUtils; import com.linkedin.venice.utils.Utils; @@ -150,6 +152,7 @@ public static List> getMetricEntity private final PubSubPositionTypeRegistry pubSubPositionTypeRegistry; private final Optional> versionLifecycleEventListeners; private final Optional externalETLService; + private final List controllerPlugins; /** * Allocates a new {@code VeniceController} object. @@ -220,6 +223,7 @@ public VeniceController(VeniceControllerContext ctx) { this.versionLifecycleEventListeners = Optional.ofNullable(ctx.getVersionLifecycleEventListeners()); this.externalETLService = Optional.ofNullable(ctx.getExternalETLService()); this.controllerService = createControllerService(); + this.controllerPlugins = createControllerPlugins(ctx); this.adminServer = createAdminServer(false); this.secureAdminServer = sslEnabled ? createAdminServer(true) : null; this.topicCleanupService = createTopicCleanupService(); @@ -267,6 +271,77 @@ private VeniceControllerService createControllerService() { return veniceControllerService; } + private List createControllerPlugins(VeniceControllerContext ctx) { + if (!multiClusterConfigs.isParent()) { + return Collections.emptyList(); + } + Admin admin = controllerService.getVeniceHelixAdmin(); + if (!(admin instanceof VeniceParentHelixAdmin)) { + return Collections.emptyList(); + } + VeniceParentHelixAdmin parentAdmin = (VeniceParentHelixAdmin) admin; + AuthorizerService authService = ctx.getAuthorizerService(); + List plugins = new ArrayList<>(); + + // Path 1: Programmatic factories from VeniceControllerContext + List factories = ctx.getControllerPluginFactories(); + if (factories != null) { + for (ControllerPluginFactory factory: factories) { + try { + ControllerPlugin plugin = factory.create(parentAdmin, authService, multiClusterConfigs); + if (plugin != null) { + plugins.add(plugin); + LOGGER.info("Created controller plugin from factory: {}", plugin.getName()); + } + } catch (Exception e) { + // Fail fast: a registered factory that throws is a wiring bug. Surface it at startup rather + // than letting the controller come up with the plugin silently absent. + throw new VeniceException("Failed to create controller plugin from factory.", e); + } + } + } + + // Path 2: Reflection-based discovery from config + String pluginClassNames = + multiClusterConfigs.getCommonConfig().getProps().getString(CONTROLLER_PLUGIN_CLASS_NAMES, ""); + if (!pluginClassNames.isEmpty()) { + for (String className: pluginClassNames.split(",")) { + className = className.trim(); + if (className.isEmpty()) { + continue; + } + try { + Class loadedClass = ReflectUtils.loadClass(className); + if (!ControllerPlugin.class.isAssignableFrom(loadedClass)) { + throw new VeniceException( + "Configured controller plugin class does not implement " + ControllerPlugin.class.getName() + ": " + + className); + } + Class pluginClass = loadedClass.asSubclass(ControllerPlugin.class); + ControllerPlugin plugin = ReflectUtils.callConstructor( + pluginClass, + new Class[] { VeniceParentHelixAdmin.class, AuthorizerService.class, + VeniceControllerMultiClusterConfig.class }, + new Object[] { parentAdmin, authService, multiClusterConfigs }); + plugins.add(plugin); + LOGGER.info("Created controller plugin from config: {} ({})", plugin.getName(), className); + } catch (VeniceException e) { + // Already specific (e.g. wrong type); surface as-is without re-wrapping. + throw e; + } catch (Exception e) { + // Fail fast: this class name was set explicitly by an operator. A typo, missing class, or + // missing constructor must abort startup rather than silently no-op the configured plugin. + throw new VeniceException( + "Failed to create controller plugin from configured class: " + className + ". Check the " + + CONTROLLER_PLUGIN_CLASS_NAMES + " config.", + e); + } + } + } + + return plugins; + } + AdminSparkServer createAdminServer(boolean secure) { return new AdminSparkServer( secure ? multiClusterConfigs.getAdminSecurePort() : multiClusterConfigs.getAdminPort(), @@ -463,6 +538,17 @@ public void start() { systemStoreRepairService.ifPresent(AbstractVeniceService::start); disabledPartitionEnablerService.ifPresent(AbstractVeniceService::start); deferredVersionSwapService.ifPresent(AbstractVeniceService::start); + for (ControllerPlugin plugin: controllerPlugins) { + LOGGER.info("Starting controller plugin: {}", plugin.getName()); + try { + plugin.start(); + } catch (Exception e) { + // Fail fast: a configured plugin that cannot start is a deployment problem and must surface + // immediately. Abort controller startup (before service discovery registration below) rather + // than letting the controller come up healthy while the plugin is silently not running. + throw new VeniceException("Failed to start controller plugin: " + plugin.getName(), e); + } + } // register with service discovery at the end asyncRetryingServiceDiscoveryAnnouncer.register(); if (adminGrpcServer != null) { @@ -531,6 +617,9 @@ public void stop() { storeBackupVersionCleanupService.ifPresent(Utils::closeQuietlyWithErrorLogged); disabledPartitionEnablerService.ifPresent(Utils::closeQuietlyWithErrorLogged); deferredVersionSwapService.ifPresent(Utils::closeQuietlyWithErrorLogged); + for (ControllerPlugin plugin: controllerPlugins) { + Utils.closeQuietlyWithErrorLogged(plugin); + } if (adminGrpcServer != null) { adminGrpcServer.stop(); } diff --git a/services/venice-controller/src/main/java/com/linkedin/venice/controller/VeniceControllerContext.java b/services/venice-controller/src/main/java/com/linkedin/venice/controller/VeniceControllerContext.java index 8e75bbf5db0..2b389d01dfb 100644 --- a/services/venice-controller/src/main/java/com/linkedin/venice/controller/VeniceControllerContext.java +++ b/services/venice-controller/src/main/java/com/linkedin/venice/controller/VeniceControllerContext.java @@ -38,6 +38,7 @@ public class VeniceControllerContext { private PubSubClientsFactory pubSubClientsFactory; private List versionLifecycleEventListeners; private ExternalETLService externalETLService; + private List controllerPluginFactories; public List getPropertiesList() { return propertiesList; @@ -91,6 +92,10 @@ public ExternalETLService getExternalETLService() { return externalETLService; } + public List getControllerPluginFactories() { + return controllerPluginFactories; + } + public VeniceControllerContext(Builder builder) { this.propertiesList = builder.propertiesList; this.metricsRepository = builder.metricsRepository; @@ -105,6 +110,7 @@ public VeniceControllerContext(Builder builder) { this.d2Clients = builder.d2Clients; this.versionLifecycleEventListeners = builder.versionLifecycleEventListeners; this.externalETLService = builder.externalETLService; + this.controllerPluginFactories = builder.controllerPluginFactories; } public static class Builder { @@ -124,6 +130,7 @@ public static class Builder { private boolean isServiceDiscoveryAnnouncerSet; private List versionLifecycleEventListeners; private ExternalETLService externalETLService; + private List controllerPluginFactories = Collections.emptyList(); public Builder setPropertiesList(List propertiesList) { this.propertiesList = propertiesList; @@ -194,6 +201,11 @@ public Builder setExternalETLService(ExternalETLService externalETLService) { return this; } + public Builder setControllerPluginFactories(List controllerPluginFactories) { + this.controllerPluginFactories = controllerPluginFactories; + return this; + } + private void addDefaultValues() { if (metricsRepository == null && !isMetricsRepositorySet) { diff --git a/services/venice-controller/src/test/java/com/linkedin/venice/controller/VeniceControllerContextTest.java b/services/venice-controller/src/test/java/com/linkedin/venice/controller/VeniceControllerContextTest.java index 8e6272eb6b9..c0f2abe6fcc 100644 --- a/services/venice-controller/src/test/java/com/linkedin/venice/controller/VeniceControllerContextTest.java +++ b/services/venice-controller/src/test/java/com/linkedin/venice/controller/VeniceControllerContextTest.java @@ -24,6 +24,7 @@ public void testVeniceServerContextCanSetDefaults() { assertNotNull(veniceControllerContext.getMetricsRepository()); assertNotNull(veniceControllerContext.getServiceDiscoveryAnnouncers()); assertEquals(veniceControllerContext.getServiceDiscoveryAnnouncers(), Collections.emptyList()); + assertEquals(veniceControllerContext.getControllerPluginFactories(), Collections.emptyList()); } @Test @@ -36,6 +37,8 @@ public void testVeniceServerContextCanSetValues() { ClientConfig routerClientConfig = mock(ClientConfig.class); ICProvider icProvider = mock(ICProvider.class); SupersetSchemaGenerator externalSupersetSchemaGenerator = mock(SupersetSchemaGenerator.class); + List controllerPluginFactories = + Collections.singletonList(mock(ControllerPluginFactory.class)); VeniceControllerContext veniceControllerContext = new VeniceControllerContext.Builder().setPropertiesList(propertiesList) @@ -47,6 +50,7 @@ public void testVeniceServerContextCanSetValues() { .setExternalSupersetSchemaGenerator(externalSupersetSchemaGenerator) .setMetricsRepository(null) .setServiceDiscoveryAnnouncers(null) + .setControllerPluginFactories(controllerPluginFactories) .build(); assertEquals(veniceControllerContext.getPropertiesList(), propertiesList); @@ -57,5 +61,6 @@ public void testVeniceServerContextCanSetValues() { assertEquals(veniceControllerContext.getD2Client(), d2Client); assertEquals(veniceControllerContext.getRouterClientConfig(), routerClientConfig); assertEquals(veniceControllerContext.getIcProvider(), icProvider); + assertEquals(veniceControllerContext.getControllerPluginFactories(), controllerPluginFactories); } }