Cluster mode for Alarms API

@@ -21,17 +21,22 @@ import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Service;
 import org.thingsboard.rule.engine.api.RpcError;
 import org.thingsboard.server.actors.ActorSystemContext;
+import org.thingsboard.server.common.data.alarm.Alarm;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.msg.MsgType;
 import org.thingsboard.server.common.msg.TbActorMsg;
 import org.thingsboard.server.common.msg.queue.ServiceType;
 import org.thingsboard.server.common.msg.queue.TbCallback;
+import org.thingsboard.server.dao.util.mapping.JacksonUtil;
+import org.thingsboard.server.gen.transport.TransportProtos;
 import org.thingsboard.server.gen.transport.TransportProtos.DeviceStateServiceMsgProto;
 import org.thingsboard.server.gen.transport.TransportProtos.FromDeviceRPCResponseProto;
 import org.thingsboard.server.gen.transport.TransportProtos.LocalSubscriptionServiceMsgProto;
 import org.thingsboard.server.gen.transport.TransportProtos.SubscriptionMgrMsgProto;
 import org.thingsboard.server.gen.transport.TransportProtos.TbAttributeUpdateProto;
 import org.thingsboard.server.gen.transport.TransportProtos.TbAttributeDeleteProto;
+import org.thingsboard.server.gen.transport.TransportProtos.TbAlarmUpdateProto;
+import org.thingsboard.server.gen.transport.TransportProtos.TbAlarmDeleteProto;
 import org.thingsboard.server.gen.transport.TransportProtos.TbSubscriptionCloseProto;
 import org.thingsboard.server.gen.transport.TransportProtos.TbTimeSeriesUpdateProto;
 import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg;
@@ -236,6 +241,8 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore
     private void forwardToLocalSubMgrService(LocalSubscriptionServiceMsgProto msg, TbCallback callback) {
         if (msg.hasSubUpdate()) {
             localSubscriptionService.onSubscriptionUpdate(msg.getSubUpdate().getSessionId(), TbSubscriptionUtils.fromProto(msg.getSubUpdate()), callback);
+        } else if (msg.hasAlarmSubUpdate()) {
+            localSubscriptionService.onSubscriptionUpdate(msg.getAlarmSubUpdate().getSessionId(), TbSubscriptionUtils.fromProto(msg.getAlarmSubUpdate()), callback);
         } else {
             throwNotHandled(msg, callback);
         }
@@ -246,6 +253,8 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore
             subscriptionManagerService.addSubscription(TbSubscriptionUtils.fromProto(msg.getAttributeSub()), callback);
         } else if (msg.hasTelemetrySub()) {
             subscriptionManagerService.addSubscription(TbSubscriptionUtils.fromProto(msg.getTelemetrySub()), callback);
+        } else if (msg.hasAlarmSub()) {
+            subscriptionManagerService.addSubscription(TbSubscriptionUtils.fromProto(msg.getAlarmSub()), callback);
         } else if (msg.hasSubClose()) {
             TbSubscriptionCloseProto closeProto = msg.getSubClose();
             subscriptionManagerService.cancelSubscription(closeProto.getSessionId(), closeProto.getSubscriptionId(), callback);
@@ -267,6 +276,18 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore
                     new TenantId(new UUID(proto.getTenantIdMSB(), proto.getTenantIdLSB())),
                     TbSubscriptionUtils.toEntityId(proto.getEntityType(), proto.getEntityIdMSB(), proto.getEntityIdLSB()),
                     proto.getScope(), proto.getKeysList(), callback);
+        } else if (msg.hasAlarmUpdate()) {
+            TbAlarmUpdateProto proto = msg.getAlarmUpdate();
+            subscriptionManagerService.onAlarmUpdate(
+                    new TenantId(new UUID(proto.getTenantIdMSB(), proto.getTenantIdLSB())),
+                    TbSubscriptionUtils.toEntityId(proto.getEntityType(), proto.getEntityIdMSB(), proto.getEntityIdLSB()),
+                    JacksonUtil.fromString(proto.getAlarm(), Alarm.class), callback);
+        } else if (msg.hasAlarmDelete()) {
+            TbAlarmDeleteProto proto = msg.getAlarmDelete();
+            subscriptionManagerService.onAlarmDeleted(
+                    new TenantId(new UUID(proto.getTenantIdMSB(), proto.getTenantIdLSB())),
+                    TbSubscriptionUtils.toEntityId(proto.getEntityType(), proto.getEntityIdMSB(), proto.getEntityIdLSB()),
+                    JacksonUtil.fromString(proto.getAlarm(), Alarm.class), callback);
         } else {
             throwNotHandled(msg, callback);
         }

@@ -39,6 +39,7 @@ import org.thingsboard.server.common.msg.queue.TbCallback;
 import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
 import org.thingsboard.server.dao.attributes.AttributesService;
 import org.thingsboard.server.dao.timeseries.TimeseriesService;
+import org.thingsboard.server.dao.util.mapping.JacksonUtil;
 import org.thingsboard.server.gen.transport.TransportProtos.*;
 import org.thingsboard.server.gen.transport.TransportProtos.LocalSubscriptionServiceMsgProto;
 import org.thingsboard.server.gen.transport.TransportProtos.TbSubscriptionUpdateProto;
@@ -189,7 +190,11 @@ public class DefaultSubscriptionManagerService implements SubscriptionManagerSer
             removedPartitions.forEach(partition -> {
                 Set<TbSubscription> subs = partitionedSubscriptions.remove(partition);
                 if (subs != null) {
-                    subs.forEach(this::removeSubscriptionFromEntityMap);
+                    subs.forEach(sub -> {
+                        if (!serviceId.equals(sub.getServiceId())) {
+                            removeSubscriptionFromEntityMap(sub);
+                        }
+                    });
                 }
             });
         }
@@ -355,7 +360,7 @@ public class DefaultSubscriptionManagerService implements SubscriptionManagerSer
                         localSubscriptionService.onSubscriptionUpdate(s.getSessionId(), update, TbCallback.EMPTY);
                     } else {
                         TopicPartitionInfo tpi = partitionService.getNotificationsTopic(ServiceType.TB_CORE, s.getServiceId());
-                        toCoreNotificationsProducer.send(tpi, toProto(s, alarm), null);
+                        toCoreNotificationsProducer.send(tpi, toProto(s, alarm, deleted), null);
                     }
                 }
             });
@@ -471,19 +476,19 @@ public class DefaultSubscriptionManagerService implements SubscriptionManagerSer
         return new TbProtoQueueMsg<>(subscription.getEntityId().getId(), toCoreMsg);
     }
 
-    private TbProtoQueueMsg<ToCoreNotificationMsg> toProto(TbSubscription subscription, Alarm alarm) {
-        TbSubscriptionUpdateProto.Builder builder = TbSubscriptionUpdateProto.newBuilder();
+    private TbProtoQueueMsg<ToCoreNotificationMsg> toProto(TbSubscription subscription, Alarm alarm, boolean deleted) {
+        TbAlarmSubscriptionUpdateProto.Builder builder = TbAlarmSubscriptionUpdateProto.newBuilder();
 
         builder.setSessionId(subscription.getSessionId());
         builder.setSubscriptionId(subscription.getSubscriptionId());
+        builder.setAlarm(JacksonUtil.toString(alarm));
+        builder.setDeleted(deleted);
 
-        //TODO 3.1
-        throw new RuntimeException("Not implemented!");
-//
-//        ToCoreNotificationMsg toCoreMsg = ToCoreNotificationMsg.newBuilder().setToLocalSubscriptionServiceMsg(
-//                LocalSubscriptionServiceMsgProto.newBuilder().setSubUpdate(builder.build()).build())
-//                .build();
-//        return new TbProtoQueueMsg<>(subscription.getEntityId().getId(), toCoreMsg);
+        ToCoreNotificationMsg toCoreMsg = ToCoreNotificationMsg.newBuilder().setToLocalSubscriptionServiceMsg(
+                LocalSubscriptionServiceMsgProto.newBuilder()
+                        .setAlarmSubUpdate(builder.build()).build())
+                .build();
+        return new TbProtoQueueMsg<>(subscription.getEntityId().getId(), toCoreMsg);
     }
 
 }

@@ -251,9 +251,7 @@ public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubsc
         ctx.cancelTasks();
         ctx.clearEntitySubscriptions();
         if (entities.isEmpty()) {
-            AlarmDataUpdate update = new AlarmDataUpdate(cmd.getCmdId(),
-                    new PageData<>(Collections.emptyList(), 1, 0, false),
-                    null, false);
+            AlarmDataUpdate update = new AlarmDataUpdate(cmd.getCmdId(), new PageData<>(), null, false);
             wsService.sendWsMsg(ctx.getSessionId(), update);
         } else {
             ctx.fetchAlarms();

@@ -87,14 +87,19 @@ public class TbAlarmDataSubCtx extends TbAbstractDataSubCtx<AlarmDataQuery> {
     }
 
     public void fetchAlarms() {
-        long start = System.currentTimeMillis();
-        PageData<AlarmData> alarms = alarmService.findAlarmDataByQueryForEntities(getTenantId(), getCustomerId(),
-                query, getOrderedEntityIds());
-        long end = System.currentTimeMillis();
-        stats.getAlarmQueryInvocationCnt().incrementAndGet();
-        stats.getAlarmQueryTimeSpent().addAndGet(end - start);
-        alarms = setAndMergeAlarmsData(alarms);
-        AlarmDataUpdate update = new AlarmDataUpdate(cmdId, alarms, null, tooManyEntities);
+        AlarmDataUpdate update;
+        if(!entitiesMap.isEmpty()) {
+            long start = System.currentTimeMillis();
+            PageData<AlarmData> alarms = alarmService.findAlarmDataByQueryForEntities(getTenantId(), getCustomerId(),
+                    query, getOrderedEntityIds());
+            long end = System.currentTimeMillis();
+            stats.getAlarmQueryInvocationCnt().incrementAndGet();
+            stats.getAlarmQueryTimeSpent().addAndGet(end - start);
+            alarms = setAndMergeAlarmsData(alarms);
+            update = new AlarmDataUpdate(cmdId, alarms, null, tooManyEntities);
+        } else {
+            update = new AlarmDataUpdate(cmdId, new PageData<>(), null, false);
+        }
         wsService.sendWsMsg(getSessionId(), update);
     }
 
@@ -146,7 +151,6 @@ public class TbAlarmDataSubCtx extends TbAbstractDataSubCtx<AlarmDataQuery> {
         subToEntityIdMap.put(subIdx, entityData.getEntityId());
         log.trace("[{}][{}][{}] Creating alarms subscription for [{}] with query: {}", serviceId, cmdId, subIdx, entityData.getEntityId(), pageLink);
         TbAlarmsSubscription subscription = TbAlarmsSubscription.builder()
-                .type(TbSubscriptionType.ALARMS)
                 .serviceId(serviceId)
                 .sessionId(sessionRef.getSessionId())
                 .subscriptionId(subIdx)

@@ -33,8 +33,8 @@ public class TbAlarmsSubscription extends TbSubscription<AlarmSubscriptionUpdate
 
     @Builder
     public TbAlarmsSubscription(String serviceId, String sessionId, int subscriptionId, TenantId tenantId, EntityId entityId,
-                                TbSubscriptionType type, BiConsumer<String, AlarmSubscriptionUpdate> updateConsumer, long ts) {
-        super(serviceId, sessionId, subscriptionId, tenantId, entityId, type, updateConsumer);
+                                BiConsumer<String, AlarmSubscriptionUpdate> updateConsumer, long ts) {
+        super(serviceId, sessionId, subscriptionId, tenantId, entityId, TbSubscriptionType.ALARMS, updateConsumer);
         this.ts = ts;
     }
 

@@ -30,6 +30,8 @@ import org.thingsboard.server.common.data.kv.KvEntry;
 import org.thingsboard.server.common.data.kv.LongDataEntry;
 import org.thingsboard.server.common.data.kv.StringDataEntry;
 import org.thingsboard.server.common.data.kv.TsKvEntry;
+import org.thingsboard.server.dao.util.mapping.JacksonUtil;
+import org.thingsboard.server.gen.transport.TransportProtos;
 import org.thingsboard.server.gen.transport.TransportProtos.KeyValueProto;
 import org.thingsboard.server.gen.transport.TransportProtos.KeyValueType;
 import org.thingsboard.server.gen.transport.TransportProtos.SubscriptionMgrMsgProto;
@@ -44,6 +46,9 @@ import org.thingsboard.server.gen.transport.TransportProtos.TbTimeSeriesSubscrip
 import org.thingsboard.server.gen.transport.TransportProtos.TbTimeSeriesUpdateProto;
 import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg;
 import org.thingsboard.server.gen.transport.TransportProtos.TsKvProto;
+import org.thingsboard.server.gen.transport.TransportProtos.TbAlarmUpdateProto;
+import org.thingsboard.server.gen.transport.TransportProtos.TbAlarmDeleteProto;
+import org.thingsboard.server.service.telemetry.sub.AlarmSubscriptionUpdate;
 import org.thingsboard.server.service.telemetry.sub.SubscriptionErrorCode;
 import org.thingsboard.server.service.telemetry.sub.TelemetrySubscriptionUpdate;
 
@@ -90,6 +95,13 @@ public class TbSubscriptionUtils {
                         TbSubscriptionKetStateProto.newBuilder().setKey(key).setTs(value).build()));
                 msgBuilder.setAttributeSub(aSubProto.build());
                 break;
+            case ALARMS:
+                TbAlarmsSubscription alarmSub = (TbAlarmsSubscription) subscription;
+                TransportProtos.TbAlarmSubscriptionProto.Builder alarmSubProto = TransportProtos.TbAlarmSubscriptionProto.newBuilder()
+                        .setSub(subscriptionProto)
+                        .setTs(alarmSub.getTs());
+                msgBuilder.setAlarmSub(alarmSubProto.build());
+                break;
         }
         return ToCoreMsg.newBuilder().setToSubscriptionMgrMsg(msgBuilder.build()).build();
     }
@@ -138,6 +150,18 @@ public class TbSubscriptionUtils {
         return builder.build();
     }
 
+    public static TbSubscription fromProto(TransportProtos.TbAlarmSubscriptionProto alarmSub) {
+        TbSubscriptionProto subProto = alarmSub.getSub();
+        TbAlarmsSubscription.TbAlarmsSubscriptionBuilder builder = TbAlarmsSubscription.builder()
+                .serviceId(subProto.getServiceId())
+                .sessionId(subProto.getSessionId())
+                .subscriptionId(subProto.getSubscriptionId())
+                .entityId(EntityIdFactory.getByTypeAndUuid(subProto.getEntityType(), new UUID(subProto.getEntityIdMSB(), subProto.getEntityIdLSB())))
+                .tenantId(new TenantId(new UUID(subProto.getTenantIdMSB(), subProto.getTenantIdLSB())));
+        builder.ts(alarmSub.getTs());
+        return builder.build();
+    }
+
     public static TelemetrySubscriptionUpdate fromProto(TbSubscriptionUpdateProto proto) {
         if (proto.getErrorCode() > 0) {
             return new TelemetrySubscriptionUpdate(proto.getSubscriptionId(), SubscriptionErrorCode.forCode(proto.getErrorCode()), proto.getErrorMsg());
@@ -156,6 +180,16 @@ public class TbSubscriptionUtils {
         }
     }
 
+    public static AlarmSubscriptionUpdate fromProto(TransportProtos.TbAlarmSubscriptionUpdateProto proto) {
+        if (proto.getErrorCode() > 0) {
+            return new AlarmSubscriptionUpdate(proto.getSubscriptionId(), SubscriptionErrorCode.forCode(proto.getErrorCode()), proto.getErrorMsg());
+        } else {
+            Alarm alarm = JacksonUtil.fromString(proto.getAlarm(), Alarm.class);
+            return new AlarmSubscriptionUpdate(proto.getSubscriptionId(), alarm);
+        }
+    }
+
+
     public static ToCoreMsg toTimeseriesUpdateProto(TenantId tenantId, EntityId entityId, List<TsKvEntry> ts) {
         TbTimeSeriesUpdateProto.Builder builder = TbTimeSeriesUpdateProto.newBuilder();
         builder.setEntityType(entityId.getEntityType().name());
@@ -264,12 +298,28 @@ public class TbSubscriptionUtils {
     }
 
     public static ToCoreMsg toAlarmUpdateProto(TenantId tenantId, EntityId entityId, Alarm alarm) {
-//        TODO: 3.1
-        throw new RuntimeException("Not implemented!");
+        TbAlarmUpdateProto.Builder builder = TbAlarmUpdateProto.newBuilder();
+        builder.setEntityType(entityId.getEntityType().name());
+        builder.setEntityIdMSB(entityId.getId().getMostSignificantBits());
+        builder.setEntityIdLSB(entityId.getId().getLeastSignificantBits());
+        builder.setTenantIdMSB(tenantId.getId().getMostSignificantBits());
+        builder.setTenantIdLSB(tenantId.getId().getLeastSignificantBits());
+        builder.setAlarm(JacksonUtil.toString(alarm));
+        SubscriptionMgrMsgProto.Builder msgBuilder = SubscriptionMgrMsgProto.newBuilder();
+        msgBuilder.setAlarmUpdate(builder);
+        return ToCoreMsg.newBuilder().setToSubscriptionMgrMsg(msgBuilder.build()).build();
     }
 
     public static ToCoreMsg toAlarmDeletedProto(TenantId tenantId, EntityId entityId, Alarm alarm) {
-//        TODO: 3.1
-        throw new RuntimeException("Not implemented!");
+        TbAlarmDeleteProto.Builder builder = TbAlarmDeleteProto.newBuilder();
+        builder.setEntityType(entityId.getEntityType().name());
+        builder.setEntityIdMSB(entityId.getId().getMostSignificantBits());
+        builder.setEntityIdLSB(entityId.getId().getLeastSignificantBits());
+        builder.setTenantIdMSB(tenantId.getId().getMostSignificantBits());
+        builder.setTenantIdLSB(tenantId.getId().getLeastSignificantBits());
+        builder.setAlarm(JacksonUtil.toString(alarm));
+        SubscriptionMgrMsgProto.Builder msgBuilder = SubscriptionMgrMsgProto.newBuilder();
+        msgBuilder.setAlarmDelete(builder);
+        return ToCoreMsg.newBuilder().setToSubscriptionMgrMsg(msgBuilder.build()).build();
     }
 }

@@ -340,28 +340,28 @@ caffeine:
   specs:
     relations:
       timeToLiveInMinutes: 1440
-      maxSize: 100000
+      maxSize: 0
     deviceCredentials:
       timeToLiveInMinutes: 1440
-      maxSize: 100000
+      maxSize: 0
     devices:
       timeToLiveInMinutes: 1440
-      maxSize: 100000
+      maxSize: 0
     sessions:
       timeToLiveInMinutes: 1440
-      maxSize: 100000
+      maxSize: 0
     assets:
       timeToLiveInMinutes: 1440
-      maxSize: 100000
+      maxSize: 0
     entityViews:
       timeToLiveInMinutes: 1440
-      maxSize: 100000
+      maxSize: 0
     claimDevices:
       timeToLiveInMinutes: 1
-      maxSize: 100000
+      maxSize: 0
     securitySettings:
       timeToLiveInMinutes: 1440
-      maxSize: 1
+      maxSize: 0
 
 redis:
   # standalone or cluster

@@ -277,6 +277,11 @@ message TbAttributeSubscriptionProto {
   string scope = 4;
 }
 
+message TbAlarmSubscriptionProto {
+  TbSubscriptionProto sub = 1;
+  int64 ts = 2;
+}
+
 message TbSubscriptionUpdateProto {
   string sessionId = 1;
   int32 subscriptionId = 2;
@@ -285,6 +290,15 @@ message TbSubscriptionUpdateProto {
   repeated TbSubscriptionUpdateValueListProto data = 5;
 }
 
+message TbAlarmSubscriptionUpdateProto {
+  string sessionId = 1;
+  int32 subscriptionId = 2;
+  int32 errorCode = 3;
+  string errorMsg = 4;
+  string alarm = 5;
+  bool deleted = 6;
+}
+
 message TbAttributeUpdateProto {
   string entityType = 1;
   int64 entityIdMSB = 2;
@@ -295,6 +309,24 @@ message TbAttributeUpdateProto {
   repeated TsKvProto data = 7;
 }
 
+message TbAlarmUpdateProto {
+  string entityType = 1;
+  int64 entityIdMSB = 2;
+  int64 entityIdLSB = 3;
+  int64 tenantIdMSB = 4;
+  int64 tenantIdLSB = 5;
+  string alarm = 6;
+}
+
+message TbAlarmDeleteProto {
+  string entityType = 1;
+  int64 entityIdMSB = 2;
+  int64 entityIdLSB = 3;
+  int64 tenantIdMSB = 4;
+  int64 tenantIdLSB = 5;
+  string alarm = 6;
+}
+
 message TbAttributeDeleteProto {
   string entityType = 1;
   int64 entityIdMSB = 2;
@@ -351,10 +383,14 @@ message SubscriptionMgrMsgProto {
   TbTimeSeriesUpdateProto tsUpdate = 4;
   TbAttributeUpdateProto attrUpdate = 5;
   TbAttributeDeleteProto attrDelete = 6;
+  TbAlarmSubscriptionProto alarmSub = 7;
+  TbAlarmUpdateProto alarmUpdate = 8;
+  TbAlarmDeleteProto alarmDelete = 9;
 }
 
 message LocalSubscriptionServiceMsgProto {
   TbSubscriptionUpdateProto subUpdate = 1;
+  TbAlarmSubscriptionUpdateProto alarmSubUpdate = 2;
 }
 
 message FromDeviceRPCResponseProto {