From ce0bebf9e8f7760193e0e3587f30093b38a25d30 Mon Sep 17 00:00:00 2001 From: Andrii Shvaika Date: Fri, 10 Jul 2020 15:35:49 +0300 Subject: [PATCH] Cluster mode for Alarms API --- .../queue/DefaultTbCoreConsumerService.java | 21 +++++++ .../DefaultSubscriptionManagerService.java | 27 +++++---- ...efaultTbEntityDataSubscriptionService.java | 4 +- .../subscription/TbAlarmDataSubCtx.java | 22 ++++--- .../subscription/TbAlarmsSubscription.java | 4 +- .../subscription/TbSubscriptionUtils.java | 58 +++++++++++++++++-- .../src/main/resources/thingsboard.yml | 16 ++--- common/queue/src/main/proto/queue.proto | 36 ++++++++++++ 8 files changed, 151 insertions(+), 37 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java index 1d1ba3a3d5..e15ee4bcec 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java @@ -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 { Set 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 toProto(TbSubscription subscription, Alarm alarm) { - TbSubscriptionUpdateProto.Builder builder = TbSubscriptionUpdateProto.newBuilder(); + private TbProtoQueueMsg 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); } } diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbEntityDataSubscriptionService.java b/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbEntityDataSubscriptionService.java index d1b31b70ca..c43999aaf3 100644 --- a/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbEntityDataSubscriptionService.java +++ b/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbEntityDataSubscriptionService.java @@ -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(); diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/TbAlarmDataSubCtx.java b/application/src/main/java/org/thingsboard/server/service/subscription/TbAlarmDataSubCtx.java index cb425daf79..d9824a4655 100644 --- a/application/src/main/java/org/thingsboard/server/service/subscription/TbAlarmDataSubCtx.java +++ b/application/src/main/java/org/thingsboard/server/service/subscription/TbAlarmDataSubCtx.java @@ -87,14 +87,19 @@ public class TbAlarmDataSubCtx extends TbAbstractDataSubCtx { } public void fetchAlarms() { - long start = System.currentTimeMillis(); - PageData 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 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 { 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) diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/TbAlarmsSubscription.java b/application/src/main/java/org/thingsboard/server/service/subscription/TbAlarmsSubscription.java index 664467459a..3f570cffaa 100644 --- a/application/src/main/java/org/thingsboard/server/service/subscription/TbAlarmsSubscription.java +++ b/application/src/main/java/org/thingsboard/server/service/subscription/TbAlarmsSubscription.java @@ -33,8 +33,8 @@ public class TbAlarmsSubscription extends TbSubscription updateConsumer, long ts) { - super(serviceId, sessionId, subscriptionId, tenantId, entityId, type, updateConsumer); + BiConsumer updateConsumer, long ts) { + super(serviceId, sessionId, subscriptionId, tenantId, entityId, TbSubscriptionType.ALARMS, updateConsumer); this.ts = ts; } diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/TbSubscriptionUtils.java b/application/src/main/java/org/thingsboard/server/service/subscription/TbSubscriptionUtils.java index fbb9957969..9b9581754c 100644 --- a/application/src/main/java/org/thingsboard/server/service/subscription/TbSubscriptionUtils.java +++ b/application/src/main/java/org/thingsboard/server/service/subscription/TbSubscriptionUtils.java @@ -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 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(); } } diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index ec624a34fe..9b6065038d 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -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 diff --git a/common/queue/src/main/proto/queue.proto b/common/queue/src/main/proto/queue.proto index ac0ce48203..f5e7328131 100644 --- a/common/queue/src/main/proto/queue.proto +++ b/common/queue/src/main/proto/queue.proto @@ -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 {