Merge pull request #10103 from vespa-engine/hmusum/move-messagebus-setup-for-containers

Move setup of message bus to ApplicationContainerCluster

6 files changed, 71 insertions, 67 deletions

diff --git a/config-model/src/main/java/com/yahoo/vespa/model/admin/LogserverContainerCluster.java b/config-model/src/main/java/com/yahoo/vespa/model/admin/LogserverContainerCluster.java
index 641e3b42e13..458bd5903fc 100644
--- a/config-model/src/main/java/com/yahoo/vespa/model/admin/LogserverContainerCluster.java
+++ b/config-model/src/main/java/com/yahoo/vespa/model/admin/LogserverContainerCluster.java
@@ -36,6 +36,8 @@ public class LogserverContainerCluster extends ContainerCluster<LogserverContain
         builder.maxthreads(10);
     }
 
+    protected boolean messageBusEnabled() { return false; }
+
     private void addLogHandler() {
         Handler<?> logHandler = Handler.fromClassName(ContainerCluster.LOG_HANDLER_CLASS);
         logHandler.addServerBindings("*://*/logs");
diff --git a/config-model/src/main/java/com/yahoo/vespa/model/admin/clustercontroller/ClusterControllerContainerCluster.java b/config-model/src/main/java/com/yahoo/vespa/model/admin/clustercontroller/ClusterControllerContainerCluster.java
index 277335fc3ea..2de4e5f5950 100644
--- a/config-model/src/main/java/com/yahoo/vespa/model/admin/clustercontroller/ClusterControllerContainerCluster.java
+++ b/config-model/src/main/java/com/yahoo/vespa/model/admin/clustercontroller/ClusterControllerContainerCluster.java
@@ -27,4 +27,6 @@ public class ClusterControllerContainerCluster extends ContainerCluster<ClusterC
     @Override
     protected void doPrepare(DeployState deployState) { }
 
+    protected boolean messageBusEnabled() { return false; }
+
 }
diff --git a/config-model/src/main/java/com/yahoo/vespa/model/admin/metricsproxy/MetricsProxyContainerCluster.java b/config-model/src/main/java/com/yahoo/vespa/model/admin/metricsproxy/MetricsProxyContainerCluster.java
index af28c577143..091c5a3acb4 100644
--- a/config-model/src/main/java/com/yahoo/vespa/model/admin/metricsproxy/MetricsProxyContainerCluster.java
+++ b/config-model/src/main/java/com/yahoo/vespa/model/admin/metricsproxy/MetricsProxyContainerCluster.java
@@ -93,7 +93,6 @@ public class MetricsProxyContainerCluster extends ContainerCluster<MetricsProxyC
         this.parent = parent;
         applicationId = deployState.getProperties().applicationId();
 
-        setMessageBusEnabled(false);
         setRpcServerEnabled(true);
         addDefaultHandlersExceptStatus();
 
@@ -158,6 +157,8 @@ public class MetricsProxyContainerCluster extends ContainerCluster<MetricsProxyC
         builder.maxthreads(10);
     }
 
+    protected boolean messageBusEnabled() { return false; }
+
     private MetricSet getAdditionalDefaultMetrics() {
         return getAdmin()
                 .map(Admin::getAdditionalDefaultMetrics)
diff --git a/config-model/src/main/java/com/yahoo/vespa/model/container/ApplicationContainerCluster.java b/config-model/src/main/java/com/yahoo/vespa/model/container/ApplicationContainerCluster.java
index 303052d011f..473971c5e7a 100644
--- a/config-model/src/main/java/com/yahoo/vespa/model/container/ApplicationContainerCluster.java
+++ b/config-model/src/main/java/com/yahoo/vespa/model/container/ApplicationContainerCluster.java
@@ -2,13 +2,18 @@
 package com.yahoo.vespa.model.container;
 
 import com.yahoo.component.ComponentId;
+import com.yahoo.component.ComponentSpecification;
 import com.yahoo.config.FileReference;
 import com.yahoo.config.application.api.ComponentInfo;
 import com.yahoo.config.model.api.TlsSecrets;
 import com.yahoo.config.model.deploy.DeployState;
 import com.yahoo.config.model.producer.AbstractConfigProducer;
 import com.yahoo.container.BundlesConfig;
+import com.yahoo.container.bundle.BundleInstantiationSpecification;
+import com.yahoo.container.jdisc.ContainerMbusConfig;
+import com.yahoo.container.jdisc.messagebus.MbusServerProvider;
 import com.yahoo.jdisc.http.ServletPathsConfig;
+import com.yahoo.osgi.provider.model.ComponentModel;
 import com.yahoo.vespa.config.search.RankProfilesConfig;
 import com.yahoo.vespa.config.search.core.RankingConstantsConfig;
 import com.yahoo.vespa.defaults.Defaults;
@@ -39,7 +44,8 @@ public final class ApplicationContainerCluster extends ContainerCluster<Applicat
         BundlesConfig.Producer,
         RankProfilesConfig.Producer,
         RankingConstantsConfig.Producer,
-        ServletPathsConfig.Producer
+        ServletPathsConfig.Producer,
+        ContainerMbusConfig.Producer
 {
 
     private final Set<FileReference> applicationBundles = new LinkedHashSet<>();
@@ -52,6 +58,9 @@ public final class ApplicationContainerCluster extends ContainerCluster<Applicat
     private Optional<TlsSecrets> tlsSecrets;
     private final boolean enableGroupingSessionCache;
 
+    private MbusParams mbusParams;
+    private boolean messageBusEnabled = true;
+
     public ApplicationContainerCluster(AbstractConfigProducer<?> parent, String subId, String name, DeployState deployState) {
         super(parent, subId, name, deployState);
 
@@ -156,6 +165,20 @@ public final class ApplicationContainerCluster extends ContainerCluster<Applicat
         if (modelEvaluation != null) modelEvaluation.getConfig(builder);
     }
 
+    @Override
+    public void getConfig(ContainerMbusConfig.Builder builder) {
+        if (mbusParams != null) {
+            if (mbusParams.maxConcurrentFactor != null)
+                builder.maxConcurrentFactor(mbusParams.maxConcurrentFactor);
+            if (mbusParams.documentExpansionFactor != null)
+                builder.documentExpansionFactor(mbusParams.documentExpansionFactor);
+            if (mbusParams.containerCoreMemory != null)
+                builder.containerCoreMemory(mbusParams.containerCoreMemory);
+        }
+        if (getDocproc() != null)
+            getDocproc().getConfig(builder);
+    }
+
     public Optional<TlsSecrets> getTlsSecrets() {
         return tlsSecrets;
     }
@@ -163,4 +186,39 @@ public final class ApplicationContainerCluster extends ContainerCluster<Applicat
     public boolean enableGroupingSessionCache() {
         return enableGroupingSessionCache;
     }
+
+    public void setMbusParams(MbusParams mbusParams) {
+        this.mbusParams = mbusParams;
+    }
+
+    public final void setMessageBusEnabled(boolean messageBusEnabled) { this.messageBusEnabled = messageBusEnabled; }
+
+    protected boolean messageBusEnabled() { return messageBusEnabled; }
+
+    public void addMbusServer(ComponentId chainId) {
+        ComponentId serviceId = chainId.nestInNamespace(ComponentId.fromString("MbusServer"));
+
+        addComponent(
+                new Component<>(new ComponentModel(new BundleInstantiationSpecification(
+                        serviceId,
+                        ComponentSpecification.fromString(MbusServerProvider.class.getName()),
+                        null))));
+    }
+
+    public static class MbusParams {
+        // the amount of the maxpendingbytes to process concurrently, typically 0.2 (20%)
+        final Double maxConcurrentFactor;
+
+        // the amount that documents expand temporarily when processing them
+        final Double documentExpansionFactor;
+
+        // the space to reserve for container, docproc stuff (memory that cannot be used for processing documents), in MB
+        final Integer containerCoreMemory;
+
+        public MbusParams(Double maxConcurrentFactor, Double documentExpansionFactor, Integer containerCoreMemory) {
+            this.maxConcurrentFactor = maxConcurrentFactor;
+            this.documentExpansionFactor = documentExpansionFactor;
+            this.containerCoreMemory = containerCoreMemory;
+        }
+    }
 }
diff --git a/config-model/src/main/java/com/yahoo/vespa/model/container/ContainerCluster.java b/config-model/src/main/java/com/yahoo/vespa/model/container/ContainerCluster.java
index 47adac637ee..779d0eb028d 100755
--- a/config-model/src/main/java/com/yahoo/vespa/model/container/ContainerCluster.java
+++ b/config-model/src/main/java/com/yahoo/vespa/model/container/ContainerCluster.java
@@ -5,7 +5,6 @@ import com.yahoo.cloud.config.ClusterInfoConfig;
 import com.yahoo.cloud.config.ConfigserverConfig;
 import com.yahoo.cloud.config.RoutingProviderConfig;
 import com.yahoo.component.ComponentId;
-import com.yahoo.component.ComponentSpecification;
 import com.yahoo.config.application.api.ApplicationMetaData;
 import com.yahoo.config.docproc.DocprocConfig;
 import com.yahoo.config.docproc.SchemamappingConfig;
@@ -20,10 +19,8 @@ import com.yahoo.container.bundle.BundleInstantiationSpecification;
 import com.yahoo.container.core.ApplicationMetadataConfig;
 import com.yahoo.container.core.document.ContainerDocumentConfig;
 import com.yahoo.container.handler.ThreadPoolProvider;
-import com.yahoo.container.jdisc.ContainerMbusConfig;
 import com.yahoo.container.jdisc.JdiscBindingsConfig;
 import com.yahoo.container.jdisc.config.HealthMonitorConfig;
-import com.yahoo.container.jdisc.messagebus.MbusServerProvider;
 import com.yahoo.container.jdisc.state.StateHandler;
 import com.yahoo.container.logging.AccessLog;
 import com.yahoo.container.usability.BindingsOverviewHandler;
@@ -89,7 +86,6 @@ public abstract class ContainerCluster<CONTAINER extends Container>
         ComponentsConfig.Producer,
         JdiscBindingsConfig.Producer,
         DocumentmanagerConfig.Producer,
-        ContainerMbusConfig.Producer,
         ContainerDocumentConfig.Producer,
         HealthMonitorConfig.Producer,
         ApplicationMetadataConfig.Producer,
@@ -139,10 +135,6 @@ public abstract class ContainerCluster<CONTAINER extends Container>
     private ContainerDocumentApi containerDocumentApi;
     private SecretStore secretStore;
 
-    // TODO: move all message-bus related fields/methods to ApplicationContainerCluster. No need for mbus for other clusters.
-    private MbusParams mbusParams;
-    private boolean messageBusEnabled = true;
-
     private boolean rpcServerEnabled = true;
     private boolean httpServerEnabled = true;
 
@@ -297,16 +289,6 @@ public abstract class ContainerCluster<CONTAINER extends Container>
 
     protected abstract void doPrepare(DeployState deployState);
 
-    public void addMbusServer(ComponentId chainId) {
-        ComponentId serviceId = chainId.nestInNamespace(ComponentId.fromString("MbusServer"));
-
-        addComponent(
-                new Component<>(new ComponentModel(new BundleInstantiationSpecification(
-                        serviceId,
-                        ComponentSpecification.fromString(MbusServerProvider.class.getName()),
-                        null))));
-    }
-
     public String getName() {
         return name;
     }
@@ -551,24 +533,6 @@ public abstract class ContainerCluster<CONTAINER extends Container>
         if (containerSearch != null) containerSearch.getConfig(builder);
     }
 
-    @Override
-    public void getConfig(ContainerMbusConfig.Builder builder) {
-        if (mbusParams != null) {
-            if (mbusParams.maxConcurrentFactor != null)
-                builder.maxConcurrentFactor(mbusParams.maxConcurrentFactor);
-            if (mbusParams.documentExpansionFactor != null)
-                builder.documentExpansionFactor(mbusParams.documentExpansionFactor);
-            if (mbusParams.containerCoreMemory != null)
-                builder.containerCoreMemory(mbusParams.containerCoreMemory);
-        }
-        if (containerDocproc != null)
-            containerDocproc.getConfig(builder);
-    }
-
-    public void setMbusParams(MbusParams mbusParams) {
-        this.mbusParams = mbusParams;
-    }
-
     public void initialize(Map<String, AbstractSearchCluster> clusterMap) {
         if (containerSearch != null) containerSearch.connectSearchClusters(clusterMap);
     }
@@ -660,10 +624,6 @@ public abstract class ContainerCluster<CONTAINER extends Container>
      */
     public Optional<Integer> getMemoryPercentage() { return Optional.ofNullable(memoryPercentage); }
 
-    public final void setMessageBusEnabled(boolean messageBusEnabled) { this.messageBusEnabled = messageBusEnabled; }
-
-    boolean messageBusEnabled() { return messageBusEnabled; }
-
     public final void setRpcServerEnabled(boolean rpcServerEnabled) { this.rpcServerEnabled = rpcServerEnabled; }
 
     boolean rpcServerEnabled() { return rpcServerEnabled; }
@@ -677,21 +637,6 @@ public abstract class ContainerCluster<CONTAINER extends Container>
         return "container cluster '" + getName() + "'";
     }
 
-    public static class MbusParams {
-        // the amount of the maxpendingbytes to process concurrently, typically 0.2 (20%)
-        final Double maxConcurrentFactor;
-
-        // the amount that documents expand temporarily when processing them
-        final Double documentExpansionFactor;
-
-        // the space to reserve for container, docproc stuff (memory that cannot be used for processing documents), in MB
-        final Integer containerCoreMemory;
-
-        public MbusParams(Double maxConcurrentFactor, Double documentExpansionFactor, Integer containerCoreMemory) {
-            this.maxConcurrentFactor = maxConcurrentFactor;
-            this.documentExpansionFactor = documentExpansionFactor;
-            this.containerCoreMemory = containerCoreMemory;
-        }
-    }
+    protected abstract boolean messageBusEnabled();
 
 }
diff --git a/config-model/src/main/java/com/yahoo/vespa/model/container/docproc/DocprocChains.java b/config-model/src/main/java/com/yahoo/vespa/model/container/docproc/DocprocChains.java
index cafc65019ce..5d08a0a6998 100644
--- a/config-model/src/main/java/com/yahoo/vespa/model/container/docproc/DocprocChains.java
+++ b/config-model/src/main/java/com/yahoo/vespa/model/container/docproc/DocprocChains.java
@@ -4,14 +4,14 @@ package com.yahoo.vespa.model.container.docproc;
 import com.yahoo.component.ComponentId;
 import com.yahoo.config.model.producer.AbstractConfigProducer;
 import com.yahoo.container.jdisc.config.SessionConfig;
+import com.yahoo.vespa.model.container.ApplicationContainerCluster;
 import com.yahoo.vespa.model.container.ContainerCluster;
 import com.yahoo.vespa.model.container.component.Component;
 import com.yahoo.vespa.model.container.component.chain.Chains;
 import com.yahoo.vespa.model.container.component.chain.ProcessingHandler;
 
 /**
- * @author einarmr
- * @since 5.1.9
+ * @author Einar M R Rosenvinge
  */
 public class DocprocChains extends Chains<DocprocChain> {
     private final ProcessingHandler<DocprocChains> docprocHandler;
@@ -22,10 +22,6 @@ public class DocprocChains extends Chains<DocprocChain> {
         addComponent(docprocHandler);
     }
 
-    public ProcessingHandler<DocprocChains> getDocprocHandler() {
-        return docprocHandler;
-    }
-
     private void addComponent(Component component) {
         if (!(getParent() instanceof ContainerCluster)) {
             return;
@@ -35,13 +31,13 @@ public class DocprocChains extends Chains<DocprocChain> {
 
 
     public void addServersAndClientsForChains() {
-        if (getParent() instanceof ContainerCluster) {
+        if (getParent() instanceof ApplicationContainerCluster) {
             for (DocprocChain chain: getChainGroup().getComponents())
-                addServerAndClientForChain((ContainerCluster) getParent(), chain);
+                addServerAndClientForChain((ApplicationContainerCluster) getParent(), chain);
         }
     }
 
-    private void addServerAndClientForChain(ContainerCluster cluster, DocprocChain docprocChain) {
+    private void addServerAndClientForChain(ApplicationContainerCluster cluster, DocprocChain docprocChain) {
         docprocHandler.addServerBindings("mbus://*/" + docprocChain.getSessionName());
 
         cluster.addMbusServer(ComponentId.fromString(docprocChain.getSessionName()));