Merge pull request #13130 from volodymyr-babak/public-edge-customer

Avoid Creation of Unnecessary Public Customers During Edge Connection

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

@@ -65,8 +65,10 @@ public class EdgeSyncCursor {
             fetchers.add(new AdminSettingsEdgeEventFetcher(ctx.getAdminSettingsService()));
             fetchers.add(new TenantAdminUsersEdgeEventFetcher(ctx.getUserService()));
         }
-        Customer publicCustomer = ctx.getCustomerService().findOrCreatePublicCustomer(edge.getTenantId());
+        Customer publicCustomer = ctx.getCustomerService().findPublicCustomer(edge.getTenantId());
+        if (publicCustomer != null) {
             fetchers.add(new CustomerEdgeEventFetcher(publicCustomer.getId()));
+        }
         if (edge.getCustomerId() != null && !EntityId.NULL_UUID.equals(edge.getCustomerId().getId())) {
             fetchers.add(new CustomerEdgeEventFetcher(edge.getCustomerId()));
             fetchers.add(new CustomerUsersEdgeEventFetcher(ctx.getUserService(), edge.getCustomerId()));

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

@@ -81,13 +81,16 @@ public class CustomerEdgeProcessor extends BaseEdgeProcessor {
         UUID uuid = new UUID(edgeNotificationMsg.getEntityIdMSB(), edgeNotificationMsg.getEntityIdLSB());
         CustomerId customerId = new CustomerId(EntityIdFactory.getByEdgeEventTypeAndUuid(type, uuid).getId());
         switch (actionType) {
-            case UPDATED:
+            case ADDED:
-                List<ListenableFuture<Void>> futures = new ArrayList<>();
+                Customer customerById = edgeCtx.getCustomerService().findCustomerById(tenantId, customerId);
-                PageDataIterable<Edge> edges = new PageDataIterable<>(link -> edgeCtx.getEdgeService().findEdgesByTenantIdAndCustomerId(tenantId, customerId, link), 1024);
+                if (customerById != null && customerById.isPublic()) {
-                for (Edge edge : edges) {
+                    return findEdgesAndSaveEdgeEvents(link -> edgeCtx.getEdgeService().findEdgesByTenantId(tenantId, link),
-                    futures.add(saveEdgeEvent(tenantId, edge.getId(), type, actionType, customerId, null));
+                            tenantId, type, actionType, customerId);
                 }
-                return Futures.transform(Futures.allAsList(futures), voids -> null, dbCallbackExecutorService);
+                return Futures.immediateFuture(null);
+            case UPDATED:
+                return findEdgesAndSaveEdgeEvents(link -> edgeCtx.getEdgeService().findEdgesByTenantIdAndCustomerId(tenantId, customerId, link),
+                        tenantId, type, actionType, customerId);
             case DELETED:
                 EdgeId edgeId = new EdgeId(new UUID(edgeNotificationMsg.getEdgeIdMSB(), edgeNotificationMsg.getEdgeIdLSB()));
                 return saveEdgeEvent(tenantId, edgeId, type, actionType, customerId, null);
@@ -96,6 +99,16 @@ public class CustomerEdgeProcessor extends BaseEdgeProcessor {
         }
     }
 
+    public ListenableFuture<Void> findEdgesAndSaveEdgeEvents(PageDataIterable.FetchFunction<Edge> edgeFetcher, TenantId tenantId,
+                                                             EdgeEventType type, EdgeEventActionType actionType, CustomerId customerId) {
+        List<ListenableFuture<Void>> futures = new ArrayList<>();
+        PageDataIterable<Edge> edges = new PageDataIterable<>(edgeFetcher, 1024);
+        for (Edge edge : edges) {
+            futures.add(saveEdgeEvent(tenantId, edge.getId(), type, actionType, customerId, null));
+        }
+        return Futures.transform(Futures.allAsList(futures), voids -> null, dbCallbackExecutorService);
+    }
+
     @Override
     public EdgeEventType getEdgeEventType() {
         return EdgeEventType.CUSTOMER;

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

@@ -551,11 +551,15 @@ public class DefaultTbClusterService implements TbClusterService {
     }
 
     private void processEdgeNotification(EdgeId edgeId, ToEdgeNotificationMsg toEdgeNotificationMsg) {
+        if (edgesEnabled) {
             var serviceIdOpt = Optional.ofNullable(edgeIdServiceIdCache.get(edgeId));
             serviceIdOpt.ifPresentOrElse(
                     serviceId -> pushMsgToEdgeNotification(toEdgeNotificationMsg, serviceId.get()),
                     () -> broadcastEdgeNotification(edgeId, toEdgeNotificationMsg)
             );
+        } else {
+            log.trace("Edges disabled. Ignoring edge notification {} for edgeId: {}", toEdgeNotificationMsg, edgeId);
+        }
     }
 
     private void pushMsgToEdgeNotification(ToEdgeNotificationMsg toEdgeNotificationMsg, String serviceId) {

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

@@ -51,6 +51,7 @@ import org.thingsboard.server.queue.provider.TbCoreQueueFactory;
 import org.thingsboard.server.queue.util.TbCoreComponent;
 import org.thingsboard.server.service.edge.EdgeContextComponent;
 import org.thingsboard.server.queue.common.consumer.MainQueueConsumerManager;
+import org.thingsboard.server.service.edge.rpc.EdgeRpcService;
 import org.thingsboard.server.service.queue.processing.AbstractConsumerService;
 import org.thingsboard.server.service.queue.processing.IdMsgPair;
 
@@ -195,36 +196,42 @@ public class DefaultTbEdgeConsumerService extends AbstractConsumerService<ToEdge
     protected void handleNotification(UUID id, TbProtoQueueMsg<ToEdgeNotificationMsg> msg, TbCallback callback) {
         ToEdgeNotificationMsg toEdgeNotificationMsg = msg.getValue();
         try {
+            EdgeRpcService edgeRpcService = edgeCtx.getEdgeRpcService();
+            if (edgeRpcService == null) {
+                log.debug("No EdgeRpcService available (edge functionality disabled), ignoring msg: {}", toEdgeNotificationMsg);
+                callback.onSuccess();
+                return;
+            }
             if (toEdgeNotificationMsg.hasEdgeHighPriority()) {
                 EdgeSessionMsg edgeSessionMsg = ProtoUtils.fromProto(toEdgeNotificationMsg.getEdgeHighPriority());
-                edgeCtx.getEdgeRpcService().onToEdgeSessionMsg(edgeSessionMsg.getTenantId(), edgeSessionMsg);
+                edgeRpcService.onToEdgeSessionMsg(edgeSessionMsg.getTenantId(), edgeSessionMsg);
                 callback.onSuccess();
             } else if (toEdgeNotificationMsg.hasEdgeEventUpdate()) {
                 EdgeSessionMsg edgeSessionMsg = ProtoUtils.fromProto(toEdgeNotificationMsg.getEdgeEventUpdate());
-                edgeCtx.getEdgeRpcService().onToEdgeSessionMsg(edgeSessionMsg.getTenantId(), edgeSessionMsg);
+                edgeRpcService.onToEdgeSessionMsg(edgeSessionMsg.getTenantId(), edgeSessionMsg);
                 callback.onSuccess();
             } else if (toEdgeNotificationMsg.hasToEdgeSyncRequest()) {
                 EdgeSessionMsg edgeSessionMsg = ProtoUtils.fromProto(toEdgeNotificationMsg.getToEdgeSyncRequest());
-                edgeCtx.getEdgeRpcService().onToEdgeSessionMsg(edgeSessionMsg.getTenantId(), edgeSessionMsg);
+                edgeRpcService.onToEdgeSessionMsg(edgeSessionMsg.getTenantId(), edgeSessionMsg);
                 callback.onSuccess();
             } else if (toEdgeNotificationMsg.hasFromEdgeSyncResponse()) {
                 EdgeSessionMsg edgeSessionMsg = ProtoUtils.fromProto(toEdgeNotificationMsg.getFromEdgeSyncResponse());
-                edgeCtx.getEdgeRpcService().onToEdgeSessionMsg(edgeSessionMsg.getTenantId(), edgeSessionMsg);
+                edgeRpcService.onToEdgeSessionMsg(edgeSessionMsg.getTenantId(), edgeSessionMsg);
                 callback.onSuccess();
             } else if (toEdgeNotificationMsg.hasComponentLifecycle()) {
                 ComponentLifecycleMsg componentLifecycle = ProtoUtils.fromProto(toEdgeNotificationMsg.getComponentLifecycle());
                 TenantId tenantId = componentLifecycle.getTenantId();
                 EdgeId edgeId = new EdgeId(componentLifecycle.getEntityId().getId());
                 if (ComponentLifecycleEvent.DELETED.equals(componentLifecycle.getEvent())) {
-                    edgeCtx.getEdgeRpcService().deleteEdge(tenantId, edgeId);
+                    edgeRpcService.deleteEdge(tenantId, edgeId);
                 } else if (ComponentLifecycleEvent.UPDATED.equals(componentLifecycle.getEvent())) {
                     Edge edge = edgeCtx.getEdgeService().findEdgeById(tenantId, edgeId);
-                    edgeCtx.getEdgeRpcService().updateEdge(tenantId, edge);
+                    edgeRpcService.updateEdge(tenantId, edge);
                 }
                 callback.onSuccess();
             }
         } catch (Exception e) {
-            log.error("Error processing edge notification message", e);
+            log.error("Error processing edge notification message {}", toEdgeNotificationMsg, e);
             callback.onFailure(e);
         }
 

application/src/test/java/org/thingsboard/server/controller/EdgeControllerTest.java

@@ -885,6 +885,10 @@ public class EdgeControllerTest extends AbstractControllerTest {
         device.setType("default");
         Device savedDevice = doPost("/api/device", device, Device.class);
 
+        // create public customer
+        doPost("/api/customer/public/device/" + savedDevice.getId().getId(), Device.class);
+        doDelete("/api/customer/device/" + savedDevice.getId().getId(), Device.class);
+
         simulateEdgeActivation(edge);
 
         doPost("/api/edge/" + edge.getId().getId().toString()

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

@@ -168,6 +168,10 @@ abstract public class AbstractEdgeTest extends AbstractControllerTest {
 
         Device savedDevice = saveDevice("Edge Device 1", THERMOSTAT_DEVICE_PROFILE_NAME);
 
+        // create public customer
+        doPost("/api/customer/public/device/" + savedDevice.getId().getId(), Device.class);
+        doDelete("/api/customer/device/" + savedDevice.getId().getId(), Device.class);
+
         Asset savedAsset = saveAsset("Edge Asset 1");
 
         updateRootRuleChainMetadata();

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

@@ -44,9 +44,8 @@ public class TenantProfileEdgeTest extends AbstractEdgeTest {
     @Test
     public void testTenantProfiles() throws Exception {
         loginSysAdmin();
-
+        TenantProfile originalTenantProfile = doGet("/api/tenantProfile/" + tenantProfileId.getId(), TenantProfile.class);
-        // save current values into tmp to revert after test
+        TenantProfile edgeTenantProfile = new TenantProfile(originalTenantProfile);
-        TenantProfile edgeTenantProfile = doGet("/api/tenantProfile/" + tenantProfileId.getId(), TenantProfile.class);
 
         // updated edge tenant profile
         edgeTenantProfile.setName("Tenant Profile Edge Test");
@@ -64,14 +63,15 @@ public class TenantProfileEdgeTest extends AbstractEdgeTest {
         Assert.assertEquals("Updated tenant profile Edge Test", tenantProfileMsg.getDescription());
         Assert.assertEquals("Tenant Profile Edge Test", tenantProfileMsg.getName());
 
+        doPost("/api/tenantProfile", originalTenantProfile, TenantProfile.class);
         loginTenantAdmin();
     }
 
     @Test
     public void testIsolatedTenantProfile() throws Exception {
         loginSysAdmin();
-
+        TenantProfile originalTenantProfile = doGet("/api/tenantProfile/" + tenantProfileId.getId(), TenantProfile.class);
-        TenantProfile edgeTenantProfile = doGet("/api/tenantProfile/" + tenantProfileId.getId(), TenantProfile.class);
+        TenantProfile edgeTenantProfile = new TenantProfile(originalTenantProfile);
 
         // set tenant profile isolated and add 2 queues - main and isolated
         edgeTenantProfile.setIsolatedTbRuleEngine(true);
@@ -110,6 +110,10 @@ public class TenantProfileEdgeTest extends AbstractEdgeTest {
             Assert.assertNotNull(queue);
             Assert.assertEquals(tenantId, queue.getTenantId());
         }
+
+        loginSysAdmin();
+        doPost("/api/tenantProfile", originalTenantProfile, TenantProfile.class);
+        loginTenantAdmin();
     }
 
     private TenantProfileQueueConfiguration createQueueConfig(String queueName, String queueTopic) {

common/dao-api/src/main/java/org/thingsboard/server/dao/customer/CustomerService.java

@@ -41,6 +41,8 @@ public interface CustomerService extends EntityDaoService {
 
     Customer findOrCreatePublicCustomer(TenantId tenantId);
 
+    Customer findPublicCustomer(TenantId tenantId);
+
     PageData<Customer> findCustomersByTenantId(TenantId tenantId, PageLink pageLink);
 
     void deleteCustomersByTenantId(TenantId tenantId);

dao/src/main/java/org/thingsboard/server/dao/customer/CustomerServiceImpl.java

@@ -213,12 +213,11 @@ public class CustomerServiceImpl extends AbstractCachedEntityService<CustomerCac
     @Override
     public Customer findOrCreatePublicCustomer(TenantId tenantId) {
         log.trace("Executing findOrCreatePublicCustomer, tenantId [{}]", tenantId);
-        Validator.validateId(tenantId, id -> INCORRECT_TENANT_ID + id);
+        var publicCustomer = findPublicCustomer(tenantId);
-        Optional<Customer> publicCustomerOpt = customerDao.findPublicCustomerByTenantId(tenantId.getId());
+        if (publicCustomer != null) {
-        if (publicCustomerOpt.isPresent()) {
+            return publicCustomer;
-            return publicCustomerOpt.get();
         }
-        var publicCustomer = new Customer();
+        publicCustomer = new Customer();
         publicCustomer.setTenantId(tenantId);
         publicCustomer.setTitle(PUBLIC_CUSTOMER_TITLE);
         try {
@@ -230,7 +229,7 @@ public class CustomerServiceImpl extends AbstractCachedEntityService<CustomerCac
             return saveCustomer(publicCustomer, false);
         } catch (DataValidationException e) {
             if (CUSTOMER_UNIQUE_TITLE_EX_MSG.equals(e.getMessage())) {
-                publicCustomerOpt = customerDao.findPublicCustomerByTenantId(tenantId.getId());
+                Optional<Customer> publicCustomerOpt = customerDao.findPublicCustomerByTenantId(tenantId.getId());
                 if (publicCustomerOpt.isPresent()) {
                     return publicCustomerOpt.get();
                 }
@@ -239,6 +238,14 @@ public class CustomerServiceImpl extends AbstractCachedEntityService<CustomerCac
         }
     }
 
+    @Override
+    public Customer findPublicCustomer(TenantId tenantId) {
+        log.trace("Executing findPublicCustomer, tenantId [{}]", tenantId);
+        Validator.validateId(tenantId, id -> INCORRECT_TENANT_ID + id);
+        Optional<Customer> publicCustomerOpt = customerDao.findPublicCustomerByTenantId(tenantId.getId());
+        return publicCustomerOpt.orElse(null);
+    }
+
     @Override
     public PageData<Customer> findCustomersByTenantId(TenantId tenantId, PageLink pageLink) {
         log.trace("Executing findCustomersByTenantId, tenantId [{}], pageLink [{}]", tenantId, pageLink);