:fix: make events group id configurable by event type

Signed-off-by: riccardomodanese <[email protected]>
eclipse · Dec 11, 2024 · 916e4f1 · 916e4f1
1 parent e4837dc
commit 916e4f1
Show file tree

Hide file tree

Showing 24 changed files with 462 additions and 52 deletions.
diff --git a/broker/artemis/plugin/src/main/java/org/eclipse/kapua/broker/artemis/AppModule.java b/broker/artemis/plugin/src/main/java/org/eclipse/kapua/broker/artemis/AppModule.java
@@ -1,5 +1,5 @@
 /*******************************************************************************
- * Copyright (c) 2016, 2022 Eurotech and/or its affiliates and others
+ * Copyright (c) 2016, 2024 Eurotech and/or its affiliates and others
  *
  * This program and the accompanying materials are made
  * available under the terms of the Eclipse Public License 2.0
@@ -17,12 +17,13 @@
 
 import org.eclipse.kapua.KapuaException;
 import org.eclipse.kapua.broker.artemis.plugin.security.setting.BrokerSetting;
-import org.eclipse.kapua.broker.artemis.plugin.security.setting.BrokerSettingKey;
 import org.eclipse.kapua.broker.artemis.plugin.utils.BrokerHostResolver;
 import org.eclipse.kapua.broker.artemis.plugin.utils.BrokerIdResolver;
 import org.eclipse.kapua.broker.artemis.plugin.utils.BrokerIdentity;
 import org.eclipse.kapua.broker.artemis.plugin.utils.DefaultBrokerHostResolver;
 import org.eclipse.kapua.broker.artemis.plugin.utils.DefaultBrokerIdResolver;
+import org.eclipse.kapua.commons.ContainerIdResolver;
+import org.eclipse.kapua.commons.DefaultContainerIdResolver;
 import org.eclipse.kapua.commons.core.AbstractKapuaModule;
 import org.eclipse.kapua.commons.core.JaxbClassProvider;
 import org.eclipse.kapua.commons.liquibase.DatabaseCheckUpdate;
@@ -67,12 +68,6 @@ String metricModuleName() {
         return "broker-telemetry";
     }
 
-    @Provides
-    @Named("eventsModuleName")
-    String eventModuleName() {
-        return "telemetry";
-    }
-
     @Provides
     @Singleton
     @Named("brokerHost")
@@ -88,7 +83,49 @@ BrokerIdResolver brokerIdResolver(BrokerSetting brokerSettings) throws KapuaExce
 
     @Singleton
     @Provides
-    BrokerHostResolver brokerHostResolver(BrokerSetting brokerSettings) throws KapuaException {
-        return new DefaultBrokerHostResolver(brokerSettings.getString(BrokerSettingKey.BROKER_HOST));
+    ContainerIdResolver containerIdResolver(SystemSetting systemSetting) throws KapuaException {
+        return new DefaultContainerIdResolver(systemSetting.getString(SystemSettingKey.CONTAINER_ID));
+    }
+
+    @Singleton
+    @Provides
+    BrokerHostResolver brokerHostResolver(SystemSetting systemSetting) throws KapuaException {
+        return new DefaultBrokerHostResolver(systemSetting.getString(SystemSettingKey.BROKER_HOST));
+    }
+
+    @Provides
+    @Named("accountEvtSubscriptionGroupId")
+    String accountEvtSubscriptionGroupId(ContainerIdResolver containerIdResolver) {
+        return "brk-tel-" + containerIdResolver.getContainerId();
+    }
+
+    @Provides
+    @Named("authenticationEvtSubscriptionGroupId")
+    String authenticationEvtSubscriptionGroupId(ContainerIdResolver containerIdResolver) {
+        return "brk-tel-" + containerIdResolver.getContainerId();
+    }
+
+    @Provides
+    @Named("authorizationEvtSubscriptionGroupId")
+    String authorizationEvtSubscriptionGroupId(ContainerIdResolver containerIdResolver) {
+        return "brk-tel-" + containerIdResolver.getContainerId();
+    }
+
+    @Provides
+    @Named("deviceConnectionEvtSubscriptionGroupId")
+    String deviceConnectionEvtSubscriptionGroupId(ContainerIdResolver containerIdResolver) {
+        return "brk-tel-" + containerIdResolver.getContainerId();
+    }
+
+    @Provides
+    @Named("deviceRegistryEvtSubscriptionGroupId")
+    String deviceRegistryEvtSubscriptionGroupId(ContainerIdResolver containerIdResolver) {
+        return "brk-tel-" + containerIdResolver.getContainerId();
+    }
+
+    @Provides
+    @Named("userEvtSubscriptionGroupId")
+    String userEvtSubscriptionGroupId(ContainerIdResolver containerIdResolver) {
+        return "brk-tel-" + containerIdResolver.getContainerId();
     }
 }
diff --git a/...in/java/org/eclipse/kapua/broker/artemis/plugin/security/ArtemisSecurityModuleClient.java b/...in/java/org/eclipse/kapua/broker/artemis/plugin/security/ArtemisSecurityModuleClient.java
@@ -1,5 +1,5 @@
 /*******************************************************************************
- * Copyright (c) 2019, 2022 Eurotech and/or its affiliates and others
+ * Copyright (c) 2019, 2024 Eurotech and/or its affiliates and others
  *
  * This program and the accompanying materials are made
  * available under the terms of the Eclipse Public License 2.0
@@ -90,4 +90,5 @@ ServiceClient authServiceClient(
     public String authServiceRequestAddress() {
         return "$SYS/SVC/auth/request";
     }
+
 }
diff --git a/commons/src/main/java/org/eclipse/kapua/commons/ContainerIdResolver.java b/commons/src/main/java/org/eclipse/kapua/commons/ContainerIdResolver.java
@@ -0,0 +1,22 @@
+/*******************************************************************************
+ * Copyright (c) 2024 Eurotech and/or its affiliates and others
+ *
+ * This program and the accompanying materials are made
+ * available under the terms of the Eclipse Public License 2.0
+ * which is available at https://www.eclipse.org/legal/epl-2.0/
+ *
+ * SPDX-License-Identifier: EPL-2.0
+ *
+ * Contributors:
+ *     Eurotech - initial API and implementation
+ *******************************************************************************/
+package org.eclipse.kapua.commons;
+
+/**
+ * Resolves the container id
+ */
+public interface ContainerIdResolver {
+
+    String getContainerId();
+
+}
diff --git a/commons/src/main/java/org/eclipse/kapua/commons/DefaultContainerIdResolver.java b/commons/src/main/java/org/eclipse/kapua/commons/DefaultContainerIdResolver.java
@@ -0,0 +1,47 @@
+/*******************************************************************************
+ * Copyright (c) 2017, 2024 Eurotech and/or its affiliates and others
+ *
+ * This program and the accompanying materials are made
+ * available under the terms of the Eclipse Public License 2.0
+ * which is available at https://www.eclipse.org/legal/epl-2.0/
+ *
+ * SPDX-License-Identifier: EPL-2.0
+ *
+ * Contributors:
+ *     Eurotech - initial API and implementation
+ *******************************************************************************/
+package org.eclipse.kapua.commons;
+
+import com.google.inject.Inject;
+import org.apache.commons.lang3.StringUtils;
+import org.eclipse.kapua.KapuaErrorCodes;
+import org.eclipse.kapua.KapuaException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Lookup from the configuration file
+ *
+ * @since 1.0
+ */
+public class DefaultContainerIdResolver implements ContainerIdResolver {
+
+    protected static final Logger logger = LoggerFactory.getLogger(DefaultContainerIdResolver.class);
+
+    private static final String CANNOT_FIND_IP_ERROR_MSG = "Cannot resolve the container id. Please check the configuration!";
+    private final String containerId;
+
+    @Inject
+    public DefaultContainerIdResolver(String containerId) throws KapuaException {
+        this.containerId = containerId;
+        logger.info("Loaded container id: {}", this.containerId);
+        if (StringUtils.isEmpty(this.containerId)) {
+            throw new KapuaException(KapuaErrorCodes.INTERNAL_ERROR, CANNOT_FIND_IP_ERROR_MSG);
+        }
+    }
+
+    @Override
+    public String getContainerId() {
+        return containerId;
+    }
+}
diff --git a/commons/src/main/java/org/eclipse/kapua/commons/event/ServiceEventTransactionalModule.java b/commons/src/main/java/org/eclipse/kapua/commons/event/ServiceEventTransactionalModule.java
@@ -56,11 +56,11 @@ private String getSubscriptionName(String address, String subscriber) {
     public ServiceEventTransactionalModule(
             ServiceEventClientConfiguration[] serviceEventClientConfigurations,
             String internalAddress,
-            String uniqueClientId,
+            String subscriptionGroupId,
             ServiceEventHouseKeeperFactory serviceEventTransactionalHousekeeperFactory,
             ServiceEventBus serviceEventBus) {
         this.serviceEventBus = serviceEventBus;
-        this.serviceEventClientConfigurations = appendClientId(uniqueClientId, serviceEventClientConfigurations);
+        this.serviceEventClientConfigurations = appendClientId(subscriptionGroupId, serviceEventClientConfigurations);
         this.internalAddress = internalAddress;
         this.houseKeeperFactory = serviceEventTransactionalHousekeeperFactory;
     }

diff --git a/commons/src/main/java/org/eclipse/kapua/commons/setting/system/SystemSettingKey.java b/commons/src/main/java/org/eclipse/kapua/commons/setting/system/SystemSettingKey.java
@@ -23,6 +23,10 @@
  */
 public enum SystemSettingKey implements SettingKey {
 
+    /**
+     * Container id (used for example to set a unique identifier for event listeners so events could be received by all the instances of a certain type)
+    */
+    CONTAINER_ID("container.id"),
     /**
      * Cluster name
      */

diff --git a/commons/src/main/resources/kapua-environment-setting.properties b/commons/src/main/resources/kapua-environment-setting.properties
@@ -68,6 +68,8 @@ broker.connector.internal.amqp.name=amqp
 broker.connector.internal.username=internalUsername
 broker.connector.internal.password=internalPassword
 
+container.id=default
+
 character.encoding=UTF-8
 
 #

diff --git a/console/web/src/main/java/org/eclipse/kapua/app/console/AppModule.java b/console/web/src/main/java/org/eclipse/kapua/app/console/AppModule.java
@@ -15,6 +15,7 @@
 import javax.inject.Named;
 import javax.inject.Singleton;
 
+import org.eclipse.kapua.commons.ContainerIdResolver;
 import org.eclipse.kapua.commons.core.AbstractKapuaModule;
 import org.eclipse.kapua.commons.util.xml.JAXBContextProvider;
 
@@ -33,9 +34,46 @@ String metricModuleName() {
         return "web-console";
     }
 
+    @Singleton
     @Provides
-    @Named("eventsModuleName")
-    String eventModuleName() {
-        return "console";
+    ContainerIdResolver containerIdResolver(SystemSetting systemSetting) throws KapuaException {
+        return new DefaultContainerIdResolver(systemSetting.getString(SystemSettingKey.CONTAINER_ID));
     }
+
+    @Provides
+    @Named("accountEvtSubscriptionGroupId")
+    String accountEvtSubscriptionGroupId(ContainerIdResolver containerIdResolver) {
+        return "console-" + containerIdResolver.getContainerId();
+    }
+
+    @Provides
+    @Named("authenticationEvtSubscriptionGroupId")
+    String authenticationEvtSubscriptionGroupId(ContainerIdResolver containerIdResolver) {
+        return "console-" + containerIdResolver.getContainerId();
+    }
+
+    @Provides
+    @Named("authorizationEvtSubscriptionGroupId")
+    String authorizationEvtSubscriptionGroupId(ContainerIdResolver containerIdResolver) {
+        return "console-" + containerIdResolver.getContainerId();
+    }
+
+    @Provides
+    @Named("deviceConnectionEvtSubscriptionGroupId")
+    String deviceConnectionEvtSubscriptionGroupId(ContainerIdResolver containerIdResolver) {
+        return "brk-tel-" + containerIdResolver.getContainerId();
+    }
+
+    @Provides
+    @Named("deviceRegistryEvtSubscriptionGroupId")
+    String deviceRegistryEvtSubscriptionGroupId(ContainerIdResolver containerIdResolver) {
+        return "console-" + containerIdResolver.getContainerId();
+    }
+
+    @Provides
+    @Named("userEvtSubscriptionGroupId")
+    String userEvtSubscriptionGroupId(ContainerIdResolver containerIdResolver) {
+        return "console-" + containerIdResolver.getContainerId();
+    }
+
 }
diff --git a/consumer/lifecycle-app/src/main/java/org/eclipse/kapua/consumer/lifecycle/AppModule.java b/consumer/lifecycle-app/src/main/java/org/eclipse/kapua/consumer/lifecycle/AppModule.java
@@ -15,9 +15,14 @@
 import javax.inject.Named;
 import javax.inject.Singleton;
 
+import org.eclipse.kapua.KapuaException;
+import org.eclipse.kapua.commons.ContainerIdResolver;
+import org.eclipse.kapua.commons.DefaultContainerIdResolver;
 import org.eclipse.kapua.commons.core.AbstractKapuaModule;
 import org.eclipse.kapua.commons.core.JaxbClassProvider;
 import org.eclipse.kapua.commons.liquibase.DatabaseCheckUpdate;
+import org.eclipse.kapua.commons.setting.system.SystemSetting;
+import org.eclipse.kapua.commons.setting.system.SystemSettingKey;
 import org.eclipse.kapua.commons.util.xml.JAXBContextProvider;
 import org.eclipse.kapua.commons.util.xml.JAXBContextProviderImpl;
 import org.eclipse.kapua.commons.util.xml.XmlRootAnnotatedJaxbClassesScanner;
@@ -48,9 +53,53 @@ String metricModuleName() {
         return MetricsLifecycle.CONSUMER_LIFECYCLE;
     }
 
+    @Singleton
     @Provides
-    @Named("eventsModuleName")
-    String eventModuleName() {
-        return "lifecycle";
+    ContainerIdResolver containerIdResolver(SystemSetting systemSetting) throws KapuaException {
+        return new DefaultContainerIdResolver(systemSetting.getString(SystemSettingKey.CONTAINER_ID));
     }
+
+    @Provides
+    @Named("accountEvtSubscriptionGroupId")
+    String accountEvtSubscriptionGroupId(ContainerIdResolver containerIdResolver) {
+        return "con-lif-" + containerIdResolver.getContainerId();
+    }
+
+    @Provides
+    @Named("authenticationEvtSubscriptionGroupId")
+    String authenticationEvtSubscriptionGroupId(ContainerIdResolver containerIdResolver) {
+        return "con-lif-" + containerIdResolver.getContainerId();
+    }
+
+    @Provides
+    @Named("authorizationEvtSubscriptionGroupId")
+    String authorizationEvtSubscriptionGroupId(ContainerIdResolver containerIdResolver) {
+        return "con-lif-" + containerIdResolver.getContainerId();
+    }
+
+    @Provides
+    @Named("deviceConnectionEvtSubscriptionGroupId")
+    String deviceConnectionEvtSubscriptionGroupId(ContainerIdResolver containerIdResolver) {
+        return "brk-tel-" + containerIdResolver.getContainerId();
+    }
+
+    @Provides
+    @Named("deviceRegistryEvtSubscriptionGroupId")
+    String deviceRegistryEvtSubscriptionGroupId(ContainerIdResolver containerIdResolver) {
+        return "con-lif-" + containerIdResolver.getContainerId();
+    }
+
+    @Provides
+    @Named("userEvtSubscriptionGroupId")
+    String userEvtSubscriptionGroupId(ContainerIdResolver containerIdResolver) {
+        return "con-lif-" + containerIdResolver.getContainerId();
+    }
+
+    @Provides
+    @Singleton
+    @Named("clusterName")
+    String clusterName(SystemSetting systemSetting) {
+        return systemSetting.getString(SystemSettingKey.CLUSTER_NAME);
+    }
+
 }
diff --git a/consumer/telemetry-app/src/main/java/org/eclipse/kapua/consumer/telemetry/AppModule.java b/consumer/telemetry-app/src/main/java/org/eclipse/kapua/consumer/telemetry/AppModule.java
@@ -15,9 +15,14 @@
 import javax.inject.Named;
 import javax.inject.Singleton;
 
+import org.eclipse.kapua.KapuaException;
+import org.eclipse.kapua.commons.ContainerIdResolver;
+import org.eclipse.kapua.commons.DefaultContainerIdResolver;
 import org.eclipse.kapua.commons.core.AbstractKapuaModule;
 import org.eclipse.kapua.commons.core.JaxbClassProvider;
 import org.eclipse.kapua.commons.liquibase.DatabaseCheckUpdate;
+import org.eclipse.kapua.commons.setting.system.SystemSetting;
+import org.eclipse.kapua.commons.setting.system.SystemSettingKey;
 import org.eclipse.kapua.commons.util.xml.JAXBContextProvider;
 import org.eclipse.kapua.commons.util.xml.JAXBContextProviderImpl;
 import org.eclipse.kapua.commons.util.xml.XmlRootAnnotatedJaxbClassesScanner;
@@ -48,9 +53,45 @@ String metricModuleName() {
         return MetricsTelemetry.CONSUMER_TELEMETRY;
     }
 
+    @Singleton
     @Provides
-    @Named("eventsModuleName")
-    String eventModuleName() {
-        return "telemetry";
+    ContainerIdResolver containerIdResolver(SystemSetting systemSetting) throws KapuaException {
+        return new DefaultContainerIdResolver(systemSetting.getString(SystemSettingKey.CONTAINER_ID));
+    }
+
+    @Provides
+    @Named("accountEvtSubscriptionGroupId")
+    String accountEvtSubscriptionGroupId(ContainerIdResolver containerIdResolver) {
+        return "con-tel-" + containerIdResolver.getContainerId();
+    }
+
+    @Provides
+    @Named("authenticationEvtSubscriptionGroupId")
+    String authenticationEvtSubscriptionGroupId(ContainerIdResolver containerIdResolver) {
+        return "con-tel-" + containerIdResolver.getContainerId();
+    }
+
+    @Provides
+    @Named("authorizationEvtSubscriptionGroupId")
+    String authorizationEvtSubscriptionGroupId(ContainerIdResolver containerIdResolver) {
+        return "con-tel-" + containerIdResolver.getContainerId();
+    }
+
+    @Provides
+    @Named("deviceConnectionEvtSubscriptionGroupId")
+    String deviceConnectionEvtSubscriptionGroupId(ContainerIdResolver containerIdResolver) {
+        return "brk-tel-" + containerIdResolver.getContainerId();
+    }
+
+    @Provides
+    @Named("deviceRegistryEvtSubscriptionGroupId")
+    String deviceRegistryEvtSubscriptionGroupId(ContainerIdResolver containerIdResolver) {
+        return "con-tel-" + containerIdResolver.getContainerId();
+    }
+
+    @Provides
+    @Named("userEvtSubscriptionGroupId")
+    String userEvtSubscriptionGroupId(ContainerIdResolver containerIdResolver) {
+        return "con-tel-" + containerIdResolver.getContainerId();
     }
 }