Merge pull request #6684 from volodymyr-babak/bug/fix-active-flag-on-cloud

[3.4] Report device activity from edge service - set active flag to true on cloud

application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/BaseEdgeProcessor.java

@@ -21,6 +21,7 @@ import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Lazy;
 import org.thingsboard.server.cluster.TbClusterService;
 import org.thingsboard.server.common.data.Device;
 import org.thingsboard.server.common.data.EdgeUtils;
@@ -53,6 +54,8 @@ import org.thingsboard.server.dao.service.DataValidator;
 import org.thingsboard.server.dao.user.UserService;
 import org.thingsboard.server.dao.widget.WidgetTypeService;
 import org.thingsboard.server.dao.widget.WidgetsBundleService;
+import org.thingsboard.server.queue.discovery.PartitionService;
+import org.thingsboard.server.queue.provider.TbQueueProducerProvider;
 import org.thingsboard.server.service.edge.rpc.constructor.AdminSettingsMsgConstructor;
 import org.thingsboard.server.service.edge.rpc.constructor.AlarmMsgConstructor;
 import org.thingsboard.server.service.edge.rpc.constructor.AssetMsgConstructor;
@@ -142,6 +145,13 @@ public abstract class BaseEdgeProcessor {
     @Autowired
     protected OtaPackageService otaPackageService;
 
+    @Autowired
+    protected PartitionService partitionService;
+
+    @Autowired
+    @Lazy
+    protected TbQueueProducerProvider producerProvider;
+
     @Autowired
     protected DataValidator<Device> deviceValidator;
 

application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/TelemetryEdgeProcessor.java

@@ -52,6 +52,8 @@ import org.thingsboard.server.common.data.kv.AttributeKey;
 import org.thingsboard.server.common.data.kv.AttributeKvEntry;
 import org.thingsboard.server.common.msg.TbMsg;
 import org.thingsboard.server.common.msg.TbMsgMetaData;
+import org.thingsboard.server.common.msg.queue.ServiceType;
+import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
 import org.thingsboard.server.common.msg.session.SessionMsgType;
 import org.thingsboard.server.common.transport.adaptor.JsonConverter;
 import org.thingsboard.server.common.transport.util.JsonUtils;
@@ -61,9 +63,12 @@ import org.thingsboard.server.gen.edge.v1.EntityDataProto;
 import org.thingsboard.server.gen.transport.TransportProtos;
 import org.thingsboard.server.queue.TbQueueCallback;
 import org.thingsboard.server.queue.TbQueueMsgMetadata;
+import org.thingsboard.server.queue.TbQueueProducer;
+import org.thingsboard.server.queue.common.TbProtoQueueMsg;
 import org.thingsboard.server.queue.util.TbCoreComponent;
 
 import javax.annotation.Nullable;
+import javax.annotation.PostConstruct;
 import java.util.ArrayList;
 import java.util.HashSet;
 import java.util.List;
@@ -77,14 +82,19 @@ public class TelemetryEdgeProcessor extends BaseEdgeProcessor {
 
     private final Gson gson = new Gson();
 
+    private TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToCoreMsg>> tbCoreMsgProducer;
+
+    @PostConstruct
+    public void init() {
+        tbCoreMsgProducer = producerProvider.getTbCoreMsgProducer();
+    }
+
     public List<ListenableFuture<Void>> processTelemetryFromEdge(TenantId tenantId, CustomerId customerId, EntityDataProto entityData) {
         log.trace("[{}] onTelemetryUpdate [{}]", tenantId, entityData);
         List<ListenableFuture<Void>> result = new ArrayList<>();
         EntityId entityId = constructEntityId(entityData);
         if ((entityData.hasPostAttributesMsg() || entityData.hasPostTelemetryMsg() || entityData.hasAttributesUpdatedMsg()) && entityId != null) {
-            // @voba - in terms of performance we should not fetch device from DB by id
+            TbMsgMetaData metaData = constructBaseMsgMetadata(tenantId, entityId);
-            // TbMsgMetaData metaData = constructBaseMsgMetadata(tenantId, entityId);
-            TbMsgMetaData metaData = new TbMsgMetaData();
             metaData.putValue(DataConstants.MSG_SOURCE_KEY, DataConstants.EDGE_MSG_SOURCE);
             if (entityData.hasPostAttributesMsg()) {
                 result.add(processPostAttributes(tenantId, customerId, entityId, entityData.getPostAttributesMsg(), metaData));
@@ -96,6 +106,20 @@ public class TelemetryEdgeProcessor extends BaseEdgeProcessor {
             if (entityData.hasPostTelemetryMsg()) {
                 result.add(processPostTelemetry(tenantId, customerId, entityId, entityData.getPostTelemetryMsg(), metaData));
             }
+            if (EntityType.DEVICE.equals(entityId.getEntityType())) {
+                DeviceId deviceId = new DeviceId(entityId.getId());
+
+                TransportProtos.DeviceActivityProto deviceActivityMsg = TransportProtos.DeviceActivityProto.newBuilder()
+                        .setTenantIdMSB(tenantId.getId().getMostSignificantBits())
+                        .setTenantIdLSB(tenantId.getId().getLeastSignificantBits())
+                        .setDeviceIdMSB(deviceId.getId().getMostSignificantBits())
+                        .setDeviceIdLSB(deviceId.getId().getLeastSignificantBits())
+                        .setLastActivityTime(System.currentTimeMillis()).build();
+
+                TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_CORE, tenantId, deviceId);
+                tbCoreMsgProducer.send(tpi, new TbProtoQueueMsg<>(deviceId.getId(),
+                        TransportProtos.ToCoreMsg.newBuilder().setDeviceActivityMsg(deviceActivityMsg).build()), null);
+            }
         }
         if (entityData.hasAttributeDeleteMsg()) {
             result.add(processAttributeDeleteMsg(tenantId, entityId, entityData.getAttributeDeleteMsg(), entityData.getEntityType()));

application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java

@@ -28,6 +28,7 @@ import org.thingsboard.common.util.JacksonUtil;
 import org.thingsboard.common.util.ThingsBoardThreadFactory;
 import org.thingsboard.server.actors.ActorSystemContext;
 import org.thingsboard.server.common.data.alarm.Alarm;
+import org.thingsboard.server.common.data.id.DeviceId;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.rpc.RpcError;
 import org.thingsboard.server.common.msg.MsgType;
@@ -236,6 +237,9 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore
                                 } else if (toCoreMsg.hasEdgeNotificationMsg()) {
                                     log.trace("[{}] Forwarding message to edge service {}", id, toCoreMsg.getEdgeNotificationMsg());
                                     forwardToEdgeNotificationService(toCoreMsg.getEdgeNotificationMsg(), callback);
+                                } else if (toCoreMsg.hasDeviceActivityMsg()) {
+                                    log.trace("[{}] Forwarding message to device state service {}", id, toCoreMsg.getDeviceActivityMsg());
+                                    forwardToStateService(toCoreMsg.getDeviceActivityMsg(), callback);
                                 } else if (!toCoreMsg.getToDeviceActorNotificationMsg().isEmpty()) {
                                     Optional<TbActorMsg> actorMsg = encodingService.decode(toCoreMsg.getToDeviceActorNotificationMsg().toByteArray());
                                     if (actorMsg.isPresent()) {
@@ -520,6 +524,20 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore
         stateService.onQueueMsg(deviceStateServiceMsg, callback);
     }
 
+    private void forwardToStateService(TransportProtos.DeviceActivityProto deviceActivityMsg, TbCallback callback) {
+        if (statsEnabled) {
+            stats.log(deviceActivityMsg);
+        }
+        TenantId tenantId = TenantId.fromUUID(new UUID(deviceActivityMsg.getTenantIdMSB(), deviceActivityMsg.getTenantIdLSB()));
+        DeviceId deviceId = new DeviceId(new UUID(deviceActivityMsg.getDeviceIdMSB(), deviceActivityMsg.getDeviceIdLSB()));
+        try {
+            stateService.onDeviceActivity(tenantId, deviceId, deviceActivityMsg.getLastActivityTime());
+            callback.onSuccess();
+        } catch (Exception e) {
+            callback.onFailure(new RuntimeException("Failed update device activity for device [" + deviceId.getId() + "]!", e));
+        }
+    }
+
     private void forwardToEdgeNotificationService(EdgeNotificationMsgProto edgeNotificationMsg, TbCallback callback) {
         if (statsEnabled) {
             stats.log(edgeNotificationMsg);

application/src/main/java/org/thingsboard/server/service/queue/TbCoreConsumerStats.java

@@ -38,6 +38,7 @@ public class TbCoreConsumerStats {
     public static final String SUBSCRIPTION_MSGS = "subMsgs";
     public static final String TO_CORE_NOTIFICATIONS = "coreNfs";
     public static final String EDGE_NOTIFICATIONS = "edgeNfs";
+    public static final String DEVICE_ACTIVITIES = "deviceActivity";
 
     private final StatsCounter totalCounter;
     private final StatsCounter sessionEventCounter;
@@ -52,6 +53,7 @@ public class TbCoreConsumerStats {
     private final StatsCounter subscriptionMsgCounter;
     private final StatsCounter toCoreNotificationsCounter;
     private final StatsCounter edgeNotificationsCounter;
+    private final StatsCounter deviceActivitiesCounter;
 
     private final List<StatsCounter> counters = new ArrayList<>();
 
@@ -70,6 +72,7 @@ public class TbCoreConsumerStats {
         this.subscriptionMsgCounter = register(statsFactory.createStatsCounter(statsKey, SUBSCRIPTION_MSGS));
         this.toCoreNotificationsCounter = register(statsFactory.createStatsCounter(statsKey, TO_CORE_NOTIFICATIONS));
         this.edgeNotificationsCounter = register(statsFactory.createStatsCounter(statsKey, EDGE_NOTIFICATIONS));
+        this.deviceActivitiesCounter = register(statsFactory.createStatsCounter(statsKey, DEVICE_ACTIVITIES));
     }
 
     private StatsCounter register(StatsCounter counter){
@@ -112,6 +115,11 @@ public class TbCoreConsumerStats {
         edgeNotificationsCounter.increment();
     }
 
+    public void log(TransportProtos.DeviceActivityProto msg) {
+        totalCounter.increment();
+        deviceActivitiesCounter.increment();
+    }
+
     public void log(TransportProtos.SubscriptionMgrMsgProto msg) {
         totalCounter.increment();
         subscriptionMsgCounter.increment();

application/src/test/java/org/thingsboard/server/edge/BaseEdgeTest.java

@@ -1352,17 +1352,17 @@ abstract public class BaseEdgeTest extends AbstractControllerTest {
         String timeseriesKey = "key";
         String timeseriesValue = "25";
         data.addProperty(timeseriesKey, timeseriesValue);
-        UplinkMsg.Builder uplinkMsgBuilder1 = UplinkMsg.newBuilder();
+        UplinkMsg.Builder uplinkMsgBuilder = UplinkMsg.newBuilder();
         EntityDataProto.Builder entityDataBuilder = EntityDataProto.newBuilder();
         entityDataBuilder.setPostTelemetryMsg(JsonConverter.convertToTelemetryProto(data, System.currentTimeMillis()));
         entityDataBuilder.setEntityType(device.getId().getEntityType().name());
         entityDataBuilder.setEntityIdMSB(device.getUuidId().getMostSignificantBits());
         entityDataBuilder.setEntityIdLSB(device.getUuidId().getLeastSignificantBits());
         testAutoGeneratedCodeByProtobuf(entityDataBuilder);
-        uplinkMsgBuilder1.addEntityData(entityDataBuilder.build());
+        uplinkMsgBuilder.addEntityData(entityDataBuilder.build());
 
-        testAutoGeneratedCodeByProtobuf(uplinkMsgBuilder1);
+        testAutoGeneratedCodeByProtobuf(uplinkMsgBuilder);
-        edgeImitator.sendUplinkMsg(uplinkMsgBuilder1.build());
+        edgeImitator.sendUplinkMsg(uplinkMsgBuilder.build());
 
         JsonObject attributesData = new JsonObject();
         String attributesKey = "test_attr";
@@ -1394,9 +1394,24 @@ abstract public class BaseEdgeTest extends AbstractControllerTest {
 
         String attributeValuesUrl = "/api/plugins/telemetry/DEVICE/" + device.getId() + "/values/attributes/" + DataConstants.SERVER_SCOPE;
         List<Map<String, String>> attributes = doGetAsyncTyped(attributeValuesUrl, new TypeReference<>() {});
-        Assert.assertEquals(2, attributes.size());
+
-        var result = attributes.stream().filter(kv -> kv.get("key").equals(attributesKey)).filter(kv -> kv.get("value").equals(attributesValue)).findFirst();
+        Assert.assertEquals(3, attributes.size());
-        Assert.assertTrue(result.isPresent());
+
+        Optional<Map<String, String>> activeAttributeOpt = getAttributeByKey("active", attributes);
+        Assert.assertTrue(activeAttributeOpt.isPresent());
+        Map<String, String> activeAttribute = activeAttributeOpt.get();
+        Assert.assertEquals("true", activeAttribute.get("value"));
+
+        Optional<Map<String, String>> customAttributeOpt = getAttributeByKey(attributesKey, attributes);
+        Assert.assertTrue(customAttributeOpt.isPresent());
+        Map<String, String> customAttribute = customAttributeOpt.get();
+        Assert.assertEquals(attributesValue, customAttribute.get("value"));
+
+        doDelete("/api/plugins/telemetry/DEVICE/" + device.getId().getId() + "/SERVER_SCOPE?keys=" + attributesKey, String.class);
+    }
+
+    private Optional<Map<String, String>> getAttributeByKey(String key, List<Map<String, String>> attributes) {
+        return attributes.stream().filter(kv -> kv.get("key").equals(key)).findFirst();
     }
 
     private Map<String, List<Map<String, String>>> loadDeviceTimeseries(Device device, String timeseriesKey) throws Exception {

common/cluster-api/src/main/proto/queue.proto

@@ -454,6 +454,14 @@ message GetOtaPackageResponseMsg {
   string fileName = 8;
 }
 
+message DeviceActivityProto {
+  int64 tenantIdMSB = 1;
+  int64 tenantIdLSB = 2;
+  int64 deviceIdMSB = 3;
+  int64 deviceIdLSB = 4;
+  int64 lastActivityTime = 5;
+}
+
 //Used to report session state to tb-Service and persist this state in the cache on the tb-Service level.
 message SubscriptionInfoProto {
   int64 lastActivityTime = 1;
@@ -902,6 +910,7 @@ message ToCoreMsg {
   SubscriptionMgrMsgProto toSubscriptionMgrMsg = 3;
   bytes toDeviceActorNotificationMsg = 4;
   EdgeNotificationMsgProto edgeNotificationMsg = 5;
+  DeviceActivityProto deviceActivityMsg = 6;
 }
 
 /* High priority messages with low latency are handled by ThingsBoard Core Service separately */