Notifications deduplication

application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java

@@ -90,7 +90,7 @@ import org.thingsboard.server.dao.widget.WidgetsBundleService;
 import org.thingsboard.server.queue.discovery.DiscoveryService;
 import org.thingsboard.server.queue.discovery.PartitionService;
 import org.thingsboard.server.queue.discovery.TbServiceInfoProvider;
-import org.thingsboard.server.queue.notification.NotificationRuleProcessor;
+import org.thingsboard.server.common.msg.notification.NotificationRuleProcessor;
 import org.thingsboard.server.queue.util.DataDecodingEncodingService;
 import org.thingsboard.server.service.apiusage.TbApiUsageStateService;
 import org.thingsboard.server.service.component.ComponentDiscoveryService;

application/src/main/java/org/thingsboard/server/service/notification/rule/DefaultNotificationRuleProcessor.java

@@ -66,6 +66,7 @@ import java.util.stream.Collectors;
 
 @Service
 @RequiredArgsConstructor
+@ConfigurationProperties(prefix = "notification-system.rules")
 @Slf4j
 @SuppressWarnings({"rawtypes", "unchecked"})
 public class DefaultNotificationRuleProcessor implements NotificationRuleProcessor {
@@ -79,6 +80,8 @@ public class DefaultNotificationRuleProcessor implements NotificationRuleProcess
     private final NotificationExecutorService notificationExecutor;
     private final CacheManager cacheManager;
     private Cache sentNotifications;
+    @Setter
+    private Map<NotificationRuleTriggerType, TriggerTypeConfig> triggerTypesConfigs;
 
     private final Map<NotificationRuleTriggerType, NotificationRuleTriggerProcessor> triggerProcessors = new EnumMap<>(NotificationRuleTriggerType.class);
 
@@ -142,9 +145,6 @@ public class DefaultNotificationRuleProcessor implements NotificationRuleProcess
                 log.debug("[{}] Rate limit for notification requests per rule was exceeded (rule '{}')", rule.getTenantId(), rule.getName());
                 return;
             }
-            if (trigger.getType().isDeduplicate() && alreadySent(rule.getId(), trigger)) {
-                return;
-            }
 
             NotificationInfo notificationInfo = constructNotificationInfo(trigger, triggerConfig);
             rule.getRecipientsConfig().getTargetsTable().forEach((delay, targets) -> {
@@ -194,23 +194,34 @@ public class DefaultNotificationRuleProcessor implements NotificationRuleProcess
         return triggerProcessors.get(triggerConfig.getTriggerType()).constructNotificationInfo(trigger);
     }
 
-    private boolean alreadySent(NotificationRuleId ruleId, NotificationRuleTrigger trigger) {
-        String key = ruleId + "_" + trigger.getOriginatorEntityId();
-        SentNotification sent = sentNotifications.get(key, SentNotification.class);
-        boolean alreadySent;
-        if (sent != null && sent.getTrigger().equals(trigger)) {
-            alreadySent = true;
-            log.debug("Notification for {} trigger was already sent, ignoring", trigger.getType());
-            // updating cache anyway so that the value is not removed by ttl
-        } else {
-            alreadySent = false;
-            sent = new SentNotification(trigger);
+    private boolean alreadySent(NotificationRule rule, NotificationRuleTrigger trigger) {
+        String deduplicationKey = getDeduplicationKey(trigger, rule);
+
+        boolean alreadySent = false;
+        Long lastSentTs = sentNotifications.get(deduplicationKey, Long.class);
+        if (lastSentTs != null) {
+            long deduplicationDuration = Optional.ofNullable(triggerTypesConfigs)
+                    .map(triggerTypes -> triggerTypes.get(trigger.getType()))
+                    .map(TriggerTypeConfig::getDeduplicationDuration)
+                    .orElseGet(trigger::getDefaultDeduplicationDuration);
+            long passed = System.currentTimeMillis() - lastSentTs;
+            log.trace("Deduplicating trigger {} for rule '{}' by key '{}'. Deduplication duration: {} ms, passed: {} ms",
+                    trigger.getType(), rule.getName(), deduplicationKey, deduplicationDuration, passed);
+            if (deduplicationDuration == 0 || passed <= deduplicationDuration) {
+                alreadySent = true;
+            }
         }
-        log.trace("[{}] Putting to sentNotifications cache: {}", ruleId, trigger);
-        sentNotifications.put(key, sent);
+        if (!alreadySent) {
+            lastSentTs = System.currentTimeMillis();
+        }
+        sentNotifications.put(deduplicationKey, lastSentTs);
         return alreadySent;
     }
 
+    public static String getDeduplicationKey(NotificationRuleTrigger trigger, NotificationRule rule) {
+        return String.join("_", trigger.getDeduplicationKey(), rule.getDeduplicationKey());
+    }
+
     @EventListener(ComponentLifecycleMsg.class)
     public void onNotificationRuleDeleted(ComponentLifecycleMsg componentLifecycleMsg) {
         if (componentLifecycleMsg.getEvent() != ComponentLifecycleEvent.DELETED ||

application/src/main/resources/thingsboard.yml

@@ -1261,6 +1261,11 @@ vc:
 
 notification_system:
   thread_pool_size: "${TB_NOTIFICATION_SYSTEM_THREAD_POOL_SIZE:10}"
+  rules:
+    trigger_types_configs:
+      RATE_LIMITS:
+        # In milliseconds, 4 hours by default
+        deduplication_duration: "${RATE_LIMITS_NOTIFICATION_RULE_DEDUPLICATION_DURATION:14400000}"
 
 management:
   endpoints:

application/src/test/java/org/thingsboard/server/service/notification/NotificationRuleApiTest.java

@@ -444,7 +444,7 @@ public class NotificationRuleApiTest extends AbstractNotificationApiTest {
     }
 
     @Test
-    public void testNotificationsDeduplication() throws Exception {
+    public void testNotificationsDeduplication_newPlatformVersion() throws Exception {
         loginSysAdmin();
         NewPlatformVersionNotificationRuleTriggerConfig triggerConfig = new NewPlatformVersionNotificationRuleTriggerConfig();
         createNotificationRule(triggerConfig, "Test", "Test", createNotificationTarget(tenantAdminUserId).getId());

common/data/src/main/java/org/thingsboard/server/common/data/notification/rule/NotificationRule.java

@@ -36,6 +36,8 @@ import javax.validation.constraints.AssertTrue;
 import javax.validation.constraints.NotBlank;
 import javax.validation.constraints.NotNull;
 import java.io.Serializable;
+import java.util.List;
+import java.util.stream.Collectors;
 
 @Data
 @NoArgsConstructor
@@ -84,4 +86,12 @@ public class NotificationRule extends BaseData<NotificationRuleId> implements Ha
                 triggerType == recipientsConfig.getTriggerType();
     }
 
+    @JsonIgnore
+    public String getDeduplicationKey() {
+        String targets = recipientsConfig.getTargetsTable().values().stream()
+                .flatMap(List::stream).sorted().map(Object::toString)
+                .collect(Collectors.joining(","));
+        return String.join(":", targets, triggerConfig.getDeduplicationKey());
+    }
+
 }

common/data/src/main/java/org/thingsboard/server/common/data/notification/rule/trigger/NotificationRuleTriggerConfig.java

@@ -15,6 +15,7 @@
  */
 package org.thingsboard.server.common.data.notification.rule.trigger;
 
+import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
 import com.fasterxml.jackson.annotation.JsonSubTypes;
 import com.fasterxml.jackson.annotation.JsonSubTypes.Type;
@@ -39,4 +40,9 @@ public interface NotificationRuleTriggerConfig extends Serializable {
 
     NotificationRuleTriggerType getTriggerType();
 
+    @JsonIgnore
+    default String getDeduplicationKey() {
+        return "#";
+    }
+
 }

common/data/src/main/java/org/thingsboard/server/common/data/notification/rule/trigger/NotificationRuleTriggerType.java

@@ -16,10 +16,8 @@
 package org.thingsboard.server.common.data.notification.rule.trigger;
 
 import lombok.Getter;
-import lombok.RequiredArgsConstructor;
 
 @Getter
-@RequiredArgsConstructor
 public enum NotificationRuleTriggerType {
 
     ENTITY_ACTION,
@@ -28,15 +26,18 @@ public enum NotificationRuleTriggerType {
     ALARM_ASSIGNMENT,
     DEVICE_ACTIVITY,
     RULE_ENGINE_COMPONENT_LIFECYCLE_EVENT,
-    NEW_PLATFORM_VERSION(false, true),
-    ENTITIES_LIMIT(false, false),
-    API_USAGE_LIMIT(false, false);
+    NEW_PLATFORM_VERSION(false),
+    ENTITIES_LIMIT(false),
+    API_USAGE_LIMIT(false);
 
     private final boolean tenantLevel;
-    private final boolean deduplicate;
 
     NotificationRuleTriggerType() {
-        this(true, false);
+        this(true);
+    }
+
+    NotificationRuleTriggerType(boolean tenantLevel) {
+        this.tenantLevel = tenantLevel;
     }
 
 }

application/src/main/java/org/thingsboard/server/service/notification/rule/trigger/RuleEngineMsgNotificationRuleTriggerProcessor.java → common/data/src/main/java/org/thingsboard/server/common/data/notification/settings/TriggerTypeConfig.java

@@ -13,15 +13,11 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.thingsboard.server.service.notification.rule.trigger;
+package org.thingsboard.server.common.data.notification.settings;
 
-import org.thingsboard.server.common.data.notification.rule.trigger.NotificationRuleTriggerConfig;
-import org.thingsboard.server.common.msg.notification.trigger.RuleEngineMsgTrigger;
-
-import java.util.Set;
-
-public interface RuleEngineMsgNotificationRuleTriggerProcessor<C extends NotificationRuleTriggerConfig> extends NotificationRuleTriggerProcessor<RuleEngineMsgTrigger, C> {
-
-    Set<String> getSupportedMsgTypes();
+import lombok.Data;
 
+@Data
+public class TriggerTypeConfig {
+    private long deduplicationDuration;
 }

common/queue/src/main/java/org/thingsboard/server/queue/notification/NotificationRuleProcessor.java → common/message/src/main/java/org/thingsboard/server/common/msg/notification/NotificationRuleProcessor.java

@@ -13,7 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.thingsboard.server.queue.notification;
+package org.thingsboard.server.common.msg.notification;
 
 import org.thingsboard.server.common.msg.notification.trigger.NotificationRuleTrigger;
 

common/message/src/main/java/org/thingsboard/server/common/msg/notification/trigger/NewPlatformVersionTrigger.java

@@ -43,4 +43,21 @@ public class NewPlatformVersionTrigger implements NotificationRuleTrigger {
         return TenantId.SYS_TENANT_ID;
     }
 
+
+    @Override
+    public boolean deduplicate() {
+        return true;
+    }
+
+    @Override
+    public String getDeduplicationKey() {
+        return String.join(":", NotificationRuleTrigger.super.getDeduplicationKey(),
+                updateInfo.getCurrentVersion(), updateInfo.getLatestVersion());
+    }
+
+    @Override
+    public long getDefaultDeduplicationDuration() {
+        return 0;
+    }
+
 }

common/message/src/main/java/org/thingsboard/server/common/msg/notification/trigger/NotificationRuleTrigger.java

@@ -29,4 +29,18 @@ public interface NotificationRuleTrigger extends Serializable {
 
     EntityId getOriginatorEntityId();
 
+
+    default boolean deduplicate() {
+        return false;
+    }
+
+    default String getDeduplicationKey() {
+        EntityId originatorEntityId = getOriginatorEntityId();
+        return String.join(":", getType().toString(), originatorEntityId.getEntityType().toString(), originatorEntityId.getId().toString());
+    }
+
+    default long getDefaultDeduplicationDuration() {
+        return 0;
+    }
+
 }

common/queue/src/main/java/org/thingsboard/server/queue/notification/RemoteNotificationRuleProcessor.java

@@ -19,7 +19,12 @@ import com.google.protobuf.ByteString;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.stereotype.Service;
+import org.springframework.util.ConcurrentReferenceHashMap;
+import org.thingsboard.server.common.data.notification.rule.trigger.NotificationRuleTriggerType;
+import org.thingsboard.server.common.data.notification.settings.TriggerTypeConfig;
+import org.thingsboard.server.common.msg.notification.NotificationRuleProcessor;
 import org.thingsboard.server.common.msg.notification.trigger.NotificationRuleTrigger;
 import org.thingsboard.server.common.msg.queue.ServiceType;
 import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
@@ -30,10 +35,17 @@ import org.thingsboard.server.queue.discovery.PartitionService;
 import org.thingsboard.server.queue.provider.TbQueueProducerProvider;
 import org.thingsboard.server.queue.util.DataDecodingEncodingService;
 
+import java.util.EnumMap;
+import java.util.Map;
 import java.util.UUID;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.springframework.util.ConcurrentReferenceHashMap.ReferenceType.SOFT;
 
 @Service
 @ConditionalOnMissingBean(value = NotificationRuleProcessor.class, ignored = RemoteNotificationRuleProcessor.class)
+@ConfigurationProperties(prefix = "notification-system.rules")
 @RequiredArgsConstructor
 @Slf4j
 public class RemoteNotificationRuleProcessor implements NotificationRuleProcessor {
@@ -43,10 +55,16 @@ public class RemoteNotificationRuleProcessor implements NotificationRuleProcesso
     private final PartitionService partitionService;
     private final DataDecodingEncodingService encodingService;
 
+    private Map<NotificationRuleTriggerType, TriggerTypeConfig> triggerTypesConfigs;
+    private final ConcurrentMap<String, Long> submittedTriggers = new ConcurrentReferenceHashMap<>(16, SOFT);
+
     @Override
     public void process(NotificationRuleTrigger trigger) {
+        if (trigger.deduplicate() && alreadySubmitted(trigger)) {
+            return;
+        }
         try {
-            log.trace("Submitting notification rule trigger: {}", trigger);
+            log.debug("Submitting notification rule trigger: {}", trigger);
             TransportProtos.NotificationRuleProcessorMsg.Builder msg = TransportProtos.NotificationRuleProcessorMsg.newBuilder()
                     .setTrigger(ByteString.copyFrom(encodingService.encode(trigger)));
 
@@ -57,9 +75,52 @@ public class RemoteNotificationRuleProcessor implements NotificationRuleProcesso
                                 .setNotificationRuleProcessorMsg(msg)
                                 .build()), null);
             });
-        } catch (Exception e) {
+        } catch (Throwable e) {
             log.error("Failed to submit notification rule trigger: {}", trigger, e);
         }
     }
 
+    private boolean alreadySubmitted(NotificationRuleTrigger trigger) {
+        String deduplicationKey = trigger.getDeduplicationKey();
+
+        AtomicBoolean alreadySubmitted = new AtomicBoolean(false);
+        submittedTriggers.compute(deduplicationKey, (key, lastSubmittedTs) -> {
+            long currentTs = System.currentTimeMillis();
+            if (lastSubmittedTs == null) {
+                return currentTs;
+            } else {
+                long deduplicationDuration = getDeduplicationDuration(trigger);
+                long passed = currentTs - lastSubmittedTs;
+                if (deduplicationDuration == 0 || passed <= deduplicationDuration) {
+                    log.trace("Notification rule trigger {} was already submitted {} ms ago, deduplication duration is {} ms. Key: '{}'",
+                            trigger.getType(), passed, deduplicationDuration, deduplicationKey);
+                    alreadySubmitted.set(true);
+                    return lastSubmittedTs;
+                } else {
+                    return currentTs;
+                }
+            }
+        });
+        return alreadySubmitted.get();
+    }
+
+    private long getDeduplicationDuration(NotificationRuleTrigger trigger) {
+        if (triggerTypesConfigs == null) {
+            triggerTypesConfigs = new EnumMap<>(NotificationRuleTriggerType.class);
+        }
+        TriggerTypeConfig triggerTypeConfig = triggerTypesConfigs.computeIfAbsent(trigger.getType(), triggerType -> {
+            TriggerTypeConfig config = new TriggerTypeConfig();
+            config.setDeduplicationDuration(trigger.getDefaultDeduplicationDuration());
+            return config;
+        });
+        return triggerTypeConfig.getDeduplicationDuration();
+    }
+
+    // set from ConfigurationProperties
+    public void setTriggerTypesConfigs(Map<NotificationRuleTriggerType, TriggerTypeConfig> triggerTypesConfigs) {
+        if (triggerTypesConfigs != null) {
+            this.triggerTypesConfigs = new EnumMap<>(triggerTypesConfigs);
+        }
+    }
+
 }

msa/vc-executor/src/main/resources/tb-vc-executor.yml

@@ -202,4 +202,11 @@ management:
 service:
   type: "${TB_SERVICE_TYPE:tb-vc-executor}"
   # Unique id for this service (autogenerated if empty)
-  id: "${TB_SERVICE_ID:}"
+  id: "${TB_SERVICE_ID:}"
+
+notification_system:
+  rules:
+    trigger_types_configs:
+      RATE_LIMITS:
+        # In milliseconds, 4 hours by default
+        deduplication_duration: "${RATE_LIMITS_NOTIFICATION_RULE_DEDUPLICATION_DURATION:14400000}"

transport/coap/src/main/resources/tb-coap-transport.yml

@@ -302,3 +302,10 @@ management:
       exposure:
         # Expose metrics endpoint (use value 'prometheus' to enable prometheus metrics).
         include: '${METRICS_ENDPOINTS_EXPOSE:info}'
+
+notification_system:
+  rules:
+    trigger_types_configs:
+      RATE_LIMITS:
+        # In milliseconds, 4 hours by default
+        deduplication_duration: "${RATE_LIMITS_NOTIFICATION_RULE_DEDUPLICATION_DURATION:14400000}"

transport/http/src/main/resources/tb-http-transport.yml

@@ -287,3 +287,10 @@ management:
       exposure:
         # Expose metrics endpoint (use value 'prometheus' to enable prometheus metrics).
         include: '${METRICS_ENDPOINTS_EXPOSE:info}'
+
+notification_system:
+  rules:
+    trigger_types_configs:
+      RATE_LIMITS:
+        # In milliseconds, 4 hours by default
+        deduplication_duration: "${RATE_LIMITS_NOTIFICATION_RULE_DEDUPLICATION_DURATION:14400000}"

transport/lwm2m/src/main/resources/tb-lwm2m-transport.yml

@@ -369,3 +369,10 @@ management:
       exposure:
         # Expose metrics endpoint (use value 'prometheus' to enable prometheus metrics).
         include: '${METRICS_ENDPOINTS_EXPOSE:info}'
+
+notification_system:
+  rules:
+    trigger_types_configs:
+      RATE_LIMITS:
+        # In milliseconds, 4 hours by default
+        deduplication_duration: "${RATE_LIMITS_NOTIFICATION_RULE_DEDUPLICATION_DURATION:14400000}"

transport/mqtt/src/main/resources/tb-mqtt-transport.yml

@@ -317,3 +317,10 @@ management:
       exposure:
         # Expose metrics endpoint (use value 'prometheus' to enable prometheus metrics).
         include: '${METRICS_ENDPOINTS_EXPOSE:info}'
+
+notification_system:
+  rules:
+    trigger_types_configs:
+      RATE_LIMITS:
+        # In milliseconds, 4 hours by default
+        deduplication_duration: "${RATE_LIMITS_NOTIFICATION_RULE_DEDUPLICATION_DURATION:14400000}"

transport/snmp/src/main/resources/tb-snmp-transport.yml

@@ -267,3 +267,10 @@ management:
       exposure:
         # Expose metrics endpoint (use value 'prometheus' to enable prometheus metrics).
         include: '${METRICS_ENDPOINTS_EXPOSE:info}'
+
+notification_system:
+  rules:
+    trigger_types_configs:
+      RATE_LIMITS:
+        # In milliseconds, 4 hours by default
+        deduplication_duration: "${RATE_LIMITS_NOTIFICATION_RULE_DEDUPLICATION_DURATION:14400000}"