diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java index 14c1b5e30a..c0e76c32e4 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java @@ -306,7 +306,7 @@ public abstract class EdgeGrpcSession implements Closeable { if (isConnected() && !pageData.getData().isEmpty()) { if (fetcher instanceof GeneralEdgeEventFetcher) { long queueSize = pageData.getTotalElements() - ((long) pageLink.getPageSize() * pageLink.getPage()); - ctx.getStatsCounterService().ifPresent(statsCounterService -> statsCounterService.setDownlinkMsgsLag(edge.getTenantId(), edge.getId(), queueSize)); + ctx.getStatsCounterService().ifPresent(statsCounterService -> statsCounterService.recordEvent(EdgeStatsKey.DOWNLINK_MSGS_LAG, tenantId, edge.getId(), queueSize)); } log.trace("[{}][{}][{}] event(s) are going to be processed.", tenantId, edge.getId(), pageData.getData().size()); List downlinkMsgsPack = convertToDownlinkMsgsPack(pageData.getData()); diff --git a/application/src/main/java/org/thingsboard/server/service/edge/stats/EdgeStatsService.java b/application/src/main/java/org/thingsboard/server/service/edge/stats/EdgeStatsService.java index 48b2a47cfb..697f7b6216 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/stats/EdgeStatsService.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/stats/EdgeStatsService.java @@ -31,6 +31,7 @@ import org.thingsboard.server.common.data.kv.BasicTsKvEntry; import org.thingsboard.server.common.data.kv.LongDataEntry; import org.thingsboard.server.common.data.kv.TimeseriesSaveResult; import org.thingsboard.server.common.data.kv.TsKvEntry; +import org.thingsboard.server.dao.edge.stats.EdgeStats; import org.thingsboard.server.dao.edge.stats.EdgeStatsCounterService; import org.thingsboard.server.dao.edge.stats.MsgCounters; import org.thingsboard.server.dao.timeseries.TimeseriesService; @@ -80,13 +81,14 @@ public class EdgeStatsService { long now = System.currentTimeMillis(); long ts = now - (now % reportIntervalMillis); - Map countersByEdge = statsCounterService.getCounterByEdge(); - Map lagByEdgeId = kafkaAdmin.isPresent() ? getEdgeLagByEdgeId(countersByEdge) : Collections.emptyMap(); - Map countersByEdgeSnapshot = new HashMap<>(statsCounterService.getCounterByEdge()); - countersByEdgeSnapshot.forEach((edgeId, counters) -> { + Map statsByEdgeSnapshot = new HashMap<>(statsCounterService.getStatsByEdge()); + boolean isKafkaStats = kafkaAdmin.isPresent(); + Map lagByEdgeId = isKafkaStats ? getLagByEdgeId(statsByEdgeSnapshot) : Collections.emptyMap(); + statsByEdgeSnapshot.forEach((edgeId, edgeStats) -> { + MsgCounters counters = edgeStats.getMsgCounters(); TenantId tenantId = counters.getTenantId(); - if (kafkaAdmin.isPresent()) { + if (isKafkaStats) { counters.getMsgsLag().set(lagByEdgeId.getOrDefault(edgeId, 0L)); } List statsEntries = List.of( @@ -102,11 +104,11 @@ public class EdgeStatsService { }); } - private Map getEdgeLagByEdgeId(Map countersByEdge) { - Map edgeToTopicMap = countersByEdge.entrySet().stream() + private Map getLagByEdgeId(Map edgeStatsByEdge) { + Map edgeToTopicMap = edgeStatsByEdge.entrySet().stream() .collect(Collectors.toMap( Map.Entry::getKey, - e -> topicService.buildEdgeEventNotificationsTopicPartitionInfo(e.getValue().getTenantId(), e.getKey()).getTopic() + e -> topicService.buildEdgeEventNotificationsTopicPartitionInfo(e.getValue().getMsgCounters().getTenantId(), e.getKey()).getTopic() )); Map lagByTopic = kafkaAdmin.get().getTotalLagForGroupsBulk(new HashSet<>(edgeToTopicMap.values())); diff --git a/application/src/test/java/org/thingsboard/server/edge/EdgeStatsIntegrationTest.java b/application/src/test/java/org/thingsboard/server/edge/EdgeStatsIntegrationTest.java new file mode 100644 index 0000000000..b993e4f564 --- /dev/null +++ b/application/src/test/java/org/thingsboard/server/edge/EdgeStatsIntegrationTest.java @@ -0,0 +1,293 @@ +/** + * Copyright © 2016-2025 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.edge; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.node.ObjectNode; +import com.google.protobuf.AbstractMessage; +import lombok.extern.slf4j.Slf4j; +import org.junit.Assert; +import org.junit.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.thingsboard.common.util.JacksonUtil; +import org.thingsboard.server.common.data.Customer; +import org.thingsboard.server.common.data.Device; +import org.thingsboard.server.common.data.asset.Asset; +import org.thingsboard.server.common.data.edge.Edge; +import org.thingsboard.server.common.data.edge.EdgeEvent; +import org.thingsboard.server.common.data.edge.EdgeEventActionType; +import org.thingsboard.server.common.data.edge.EdgeEventType; +import org.thingsboard.server.common.data.kv.TsKvEntry; +import org.thingsboard.server.dao.edge.stats.EdgeStatsCounterService; +import org.thingsboard.server.dao.edge.stats.EdgeStatsKey; +import org.thingsboard.server.dao.edge.stats.MsgCounters; +import org.thingsboard.server.dao.service.DaoSqlTest; +import org.thingsboard.server.gen.edge.v1.EntityDataProto; +import org.thingsboard.server.gen.transport.TransportProtos; +import org.thingsboard.server.service.edge.stats.EdgeStatsService; + +import java.time.Duration; +import java.util.Arrays; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.function.Supplier; +import java.util.stream.Collectors; + +import static org.awaitility.Awaitility.await; +import static org.junit.jupiter.api.Assertions.assertAll; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.thingsboard.server.dao.edge.stats.EdgeStatsKey.DOWNLINK_MSGS_ADDED; +import static org.thingsboard.server.dao.edge.stats.EdgeStatsKey.DOWNLINK_MSGS_PERMANENTLY_FAILED; +import static org.thingsboard.server.dao.edge.stats.EdgeStatsKey.DOWNLINK_MSGS_PUSHED; +import static org.thingsboard.server.dao.edge.stats.EdgeStatsKey.DOWNLINK_MSGS_TMP_FAILED; + +@DaoSqlTest +@Slf4j +public class EdgeStatsIntegrationTest extends AbstractEdgeTest { + + private static final String STATISTICS_DEVICE_PROFILE = "STATISTICS"; + + private final Map EXPECTED_EDGE_STATS = Map.of( + DOWNLINK_MSGS_ADDED, 6L, + DOWNLINK_MSGS_PUSHED, 6L, + DOWNLINK_MSGS_PERMANENTLY_FAILED, 0L, + DOWNLINK_MSGS_TMP_FAILED, 0L + ); + + private final Map EXPECTED_EMPTY_EDGE_STATS = Map.of( + DOWNLINK_MSGS_ADDED, 0L, + DOWNLINK_MSGS_PUSHED, 0L, + DOWNLINK_MSGS_PERMANENTLY_FAILED, 0L, + DOWNLINK_MSGS_TMP_FAILED, 0L + ); + + @Autowired + private EdgeStatsService edgeStatsService; + @Autowired + private EdgeStatsCounterService statsCounterService; + + @Test + public void testFullEdgeStatsCycle() throws Exception { + // 1. Clear previous stats and prepare test data + prepareTestData(); + + // 2. Wait until Edge counters are updated + awaitEdgeCountersUpdated(); + + // 3. Report statistics + edgeStatsService.reportStats(); + + // 4. Wait until timeseries data is persisted + awaitStatsMatch(EXPECTED_EDGE_STATS); + + // 5. Send statistics to Edge + List latestStatsEntries = tsService.findLatest( + tenantId, + edge.getId(), + Arrays.stream(EdgeStatsKey.values()).map(EdgeStatsKey::getKey).toList() + ).get(); + sendStatsToEdge(latestStatsEntries); + + // 6. Wait until telemetry Proto contains the expected stats + await().atMost(10, TimeUnit.SECONDS).pollInterval(Duration.ofMillis(200)).untilAsserted(() -> { + EntityDataProto latestMsg = getLatestEntityDataMessage(); + Map actualStats = toMap(latestMsg.getPostTelemetryMsg().getTsKvList(0)); + assertAllStatsEqual(EXPECTED_EDGE_STATS, actualStats, "Proto stats"); + }); + } + + @Test + public void testNoMessagesFromEdge() throws ExecutionException, InterruptedException { + // 1. Clear stats counters for the Edge + statsCounterService.clear(edge.getId()); + + // 2. Report stats with no data from the Edge + edgeStatsService.reportStats(); + + // 3. Verify that persisted timeseries contains only empty stats + awaitStatsMatch(EXPECTED_EMPTY_EDGE_STATS); + + List actual = tsService.findLatest( + tenantId, + edge.getId(), + Arrays.stream(EdgeStatsKey.values()).map(EdgeStatsKey::getKey).toList() + ).get(); + + assertAllStatsEqual(EXPECTED_EMPTY_EDGE_STATS, toMap(actual), "Empty stats"); + } + + @Test + public void testRepeatedReportStatsDoesNotDuplicate() throws ExecutionException, InterruptedException { + // 1. Clear previous stats and prepare test data + prepareTestData(); + + // 2. Wait until Edge counters are updated + awaitEdgeCountersUpdated(); + + // 3. First report call + edgeStatsService.reportStats(); + + // 4. Verify that the persisted stats match expectations + awaitStatsMatch(EXPECTED_EDGE_STATS); + + // 5. Remove persisted stats to simulate a re-report scenario + tsService.removeLatest( + tenantId, + edge.getId(), + Arrays.stream(EdgeStatsKey.values()).map(EdgeStatsKey::getKey).toList() + ); + + // 6. Second report call without increments (counters already cleared) + edgeStatsService.reportStats(); + + // 7. Verify that the stats are empty after the second report + awaitStatsMatch(EXPECTED_EMPTY_EDGE_STATS); + } + + private void awaitStatsMatch(Map expected) { + await().atMost(10, TimeUnit.SECONDS).pollInterval(Duration.ofMillis(200)).untilAsserted(() -> { + Map actualStats = fetchLatestStats(); + assertAllStatsEqual(expected, actualStats, "Timeseries stats"); + }); + } + + private Map fetchLatestStats() throws ExecutionException, InterruptedException { + List latestStatsEntries = tsService.findLatest( + tenantId, + edge.getId(), + Arrays.stream(EdgeStatsKey.values()).map(EdgeStatsKey::getKey).toList() + ).get(); + return toMap(latestStatsEntries); + } + + private void prepareTestData() throws InterruptedException, ExecutionException { + statsCounterService.clear(edge.getId()); + // 2 stats message ADDED Device Profile, ASSIGN Device + edgeImitator.expectMessageAmount(4); + Device device = saveDevice("StatisticDevice", STATISTICS_DEVICE_PROFILE); + doPost("/api/edge/" + edge.getUuidId() + "/device/" + device.getUuidId(), Device.class); + edgeImitator.waitForMessages(); + // 1 stats message ASSIGN Asset + edgeImitator.expectMessageAmount(2); + Asset savedAsset = saveAsset("Edge Asset 2"); + doPost("/api/edge/" + edge.getUuidId() + + "/asset/" + savedAsset.getUuidId(), Asset.class); + Assert.assertTrue(edgeImitator.waitForMessages()); + + // 2 stats message ADDED Customer, ASSIGN Customer + edgeImitator.expectMessageAmount(1); + Customer customer = new Customer(); + customer.setTitle("Edge Customer"); + Customer savedCustomer = doPost("/api/customer", customer, Customer.class); + Assert.assertFalse(edgeImitator.waitForMessages(5)); + + // assign edge to customer + edgeImitator.expectMessageAmount(2); + doPost("/api/customer/" + savedCustomer.getUuidId() + + "/edge/" + edge.getUuidId(), Edge.class); + Assert.assertTrue(edgeImitator.waitForMessages()); + + //1 stats message Timeseries + edgeImitator.expectMessageAmount(1); + String timeseriesData = "{\"data\":{\"temperature\":25},\"ts\":" + System.currentTimeMillis() + "}"; + JsonNode timeseriesEntityData = JacksonUtil.toJsonNode(timeseriesData); + EdgeEvent edgeEvent = constructEdgeEvent(tenantId, edge.getId(), EdgeEventActionType.TIMESERIES_UPDATED, device.getId().getId(), EdgeEventType.DEVICE, timeseriesEntityData); + edgeEventService.saveAsync(edgeEvent).get(); + Assert.assertTrue(edgeImitator.waitForMessages()); + } + + private void assertAllStatsEqual(Map expected, Map actual, String context) { + assertAll(context, + expected.entrySet().stream() + .map(e -> () -> assertEquals(e.getValue(), actual.get(e.getKey().getKey()), "Mismatch for stat: " + e.getKey())) + ); + } + + private void awaitEdgeCountersUpdated() { + await().atMost(10, TimeUnit.SECONDS).pollInterval(Duration.ofMillis(200)).untilAsserted(() -> { + MsgCounters counters = statsCounterService.getStatsByEdge().get(edge.getId()).getMsgCounters(); + Map actualCounters = toMap( + Map.entry(DOWNLINK_MSGS_ADDED.getKey(), () -> counters.getMsgsAdded().get()), + Map.entry(DOWNLINK_MSGS_PUSHED.getKey(), () -> counters.getMsgsPushed().get()), + Map.entry(DOWNLINK_MSGS_PERMANENTLY_FAILED.getKey(), () -> counters.getMsgsPermanentlyFailed().get()), + Map.entry(DOWNLINK_MSGS_TMP_FAILED.getKey(), () -> counters.getMsgsTmpFailed().get()) + ); + assertAllStatsEqual(EXPECTED_EDGE_STATS, actualCounters, "Edge counters"); + }); + } + + @SafeVarargs + private final Map toMap(Map.Entry>... suppliers) { + return Arrays.stream(suppliers) + .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().get())); + } + + private Map toMap(List stats) { + return stats.stream() + .collect(Collectors.toMap(TsKvEntry::getKey, e -> e.getLongValue().orElse(0L))); + } + + private Map toMap(TransportProtos.TsKvListProto kvList) { + Map map = kvList.getKvList().stream() + .collect(Collectors.toMap( + TransportProtos.KeyValueProto::getKey, + TransportProtos.KeyValueProto::getLongV + )); + for (EdgeStatsKey key : EdgeStatsKey.values()) { + map.putIfAbsent(key.getKey(), 0L); + } + return map; + } + + private void sendStatsToEdge(List stats) throws Exception { + edgeImitator.expectMessageAmount(1); + EdgeEvent edgeEvent = constructEdgeEvent( + tenantId, + edge.getId(), + EdgeEventActionType.TIMESERIES_UPDATED, + edge.getId().getId(), + EdgeEventType.EDGE, + buildStatsJson(System.currentTimeMillis(), stats) + ); + edgeEventService.saveAsync(edgeEvent).get(); + assertTrue(edgeImitator.waitForMessages()); + } + + private EntityDataProto getLatestEntityDataMessage() { + AbstractMessage latestMessage = edgeImitator.getLatestMessage(); + assertInstanceOf(EntityDataProto.class, latestMessage); + EntityDataProto msg = (EntityDataProto) latestMessage; + assertEquals(edge.getUuidId().getMostSignificantBits(), msg.getEntityIdMSB()); + assertEquals(edge.getUuidId().getLeastSignificantBits(), msg.getEntityIdLSB()); + assertEquals(edge.getId().getEntityType().name(), msg.getEntityType()); + assertTrue(msg.hasPostTelemetryMsg()); + return msg; + } + + private ObjectNode buildStatsJson(long ts, List statsEntries) { + ObjectNode entityBody = JacksonUtil.newObjectNode(); + entityBody.put("ts", ts); + ObjectNode data = JacksonUtil.newObjectNode(); + statsEntries.forEach(entry -> data.put(entry.getKey(), entry.getValueAsString())); + entityBody.set("data", data); + return entityBody; + } + +} diff --git a/application/src/test/java/org/thingsboard/server/service/edge/EdgeStatsTest.java b/application/src/test/java/org/thingsboard/server/service/edge/EdgeStatsTest.java index 25ff0f1b5d..fb24cdb5bb 100644 --- a/application/src/test/java/org/thingsboard/server/service/edge/EdgeStatsTest.java +++ b/application/src/test/java/org/thingsboard/server/service/edge/EdgeStatsTest.java @@ -29,6 +29,7 @@ import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.kv.TimeseriesSaveResult; import org.thingsboard.server.common.data.kv.TsKvEntry; import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; +import org.thingsboard.server.dao.edge.stats.EdgeStats; import org.thingsboard.server.dao.edge.stats.EdgeStatsCounterService; import org.thingsboard.server.dao.edge.stats.MsgCounters; import org.thingsboard.server.dao.timeseries.TimeseriesService; @@ -44,9 +45,11 @@ import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.stream.Collectors; +import static org.mockito.ArgumentMatchers.anyList; import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import static org.thingsboard.server.dao.edge.stats.EdgeStatsKey.DOWNLINK_MSGS_ADDED; @@ -58,6 +61,9 @@ import static org.thingsboard.server.dao.edge.stats.EdgeStatsKey.DOWNLINK_MSGS_T @ExtendWith(MockitoExtension.class) public class EdgeStatsTest { + private static final int TTL_DAYS = 30; + private static final long REPORT_INTERVAL_MILLIS = 600_000L; + @Mock private TimeseriesService tsService; @Mock @@ -71,40 +77,45 @@ public class EdgeStatsTest { @BeforeEach void setUp() { - edgeStatsService = new EdgeStatsService( + edgeStatsService = createEdgeStatsService(Optional.empty()); + } + + private EdgeStatsService createEdgeStatsService(Optional kafkaAdmin) { + EdgeStatsService service = new EdgeStatsService( tsService, statsCounterService, topicService, - Optional.empty() + kafkaAdmin ); - - ReflectionTestUtils.setField(edgeStatsService, "edgesStatsTtlDays", 30); - ReflectionTestUtils.setField(edgeStatsService, "reportIntervalMillis", 600_000L); + ReflectionTestUtils.setField(service, "edgesStatsTtlDays", TTL_DAYS); + ReflectionTestUtils.setField(service, "reportIntervalMillis", REPORT_INTERVAL_MILLIS); + return service; } @Test public void testReportStatsSavesTelemetry() { - // given - MsgCounters counters = new MsgCounters(tenantId); + EdgeStats edgeStats = new EdgeStats(tenantId); + MsgCounters counters = edgeStats.getMsgCounters(); counters.getMsgsAdded().set(5); counters.getMsgsPushed().set(3); counters.getMsgsPermanentlyFailed().set(1); counters.getMsgsTmpFailed().set(0); counters.getMsgsLag().set(10); - ConcurrentHashMap countersByEdge = new ConcurrentHashMap<>(); - countersByEdge.put(edgeId, counters); + ConcurrentHashMap edgeStatsByEdge = new ConcurrentHashMap<>(); + edgeStatsByEdge.put(edgeId, edgeStats); - when(statsCounterService.getCounterByEdge()).thenReturn(countersByEdge); + when(statsCounterService.getStatsByEdge()).thenReturn(edgeStatsByEdge); - ArgumentCaptor> captor = ArgumentCaptor.forClass(List.class); + ArgumentCaptor> captor = ArgumentCaptor.forClass((Class) List.class); when(tsService.save(eq(tenantId), eq(edgeId), captor.capture(), anyLong())) .thenReturn(Futures.immediateFuture(mock(TimeseriesSaveResult.class))); - // when edgeStatsService.reportStats(); - // then + verify(tsService, times(1)).save(eq(tenantId), eq(edgeId), anyList(), anyLong()); + verify(statsCounterService, times(1)).clear(edgeId); + List entries = captor.getValue(); Assertions.assertEquals(5, entries.size()); @@ -116,26 +127,22 @@ public class EdgeStatsTest { Assertions.assertEquals(1L, valuesByKey.get(DOWNLINK_MSGS_PERMANENTLY_FAILED.getKey()).longValue()); Assertions.assertEquals(0L, valuesByKey.get(DOWNLINK_MSGS_TMP_FAILED.getKey()).longValue()); Assertions.assertEquals(10L, valuesByKey.get(DOWNLINK_MSGS_LAG.getKey()).longValue()); - - - verify(statsCounterService).clear(edgeId); } @Test public void testReportStatsWithKafkaLag() { - // given - MsgCounters counters = new MsgCounters(tenantId); + EdgeStats edgeStats = new EdgeStats(tenantId); + MsgCounters counters = edgeStats.getMsgCounters(); counters.getMsgsAdded().set(2); counters.getMsgsPushed().set(2); counters.getMsgsPermanentlyFailed().set(0); counters.getMsgsTmpFailed().set(1); counters.getMsgsLag().set(0); - ConcurrentHashMap countersByEdge = new ConcurrentHashMap<>(); - countersByEdge.put(edgeId, counters); + ConcurrentHashMap edgeStatsByEdge = new ConcurrentHashMap<>(); + edgeStatsByEdge.put(edgeId, edgeStats); - // mocks - when(statsCounterService.getCounterByEdge()).thenReturn(countersByEdge); + when(statsCounterService.getStatsByEdge()).thenReturn(edgeStatsByEdge); String topic = "edge-topic"; TopicPartitionInfo partitionInfo = new TopicPartitionInfo(topic, tenantId, 0, false); @@ -145,29 +152,22 @@ public class EdgeStatsTest { when(kafkaAdmin.getTotalLagForGroupsBulk(Set.of(topic))) .thenReturn(Map.of(topic, 15L)); - ArgumentCaptor> captor = ArgumentCaptor.forClass(List.class); + ArgumentCaptor> captor = ArgumentCaptor.forClass((Class) List.class); when(tsService.save(eq(tenantId), eq(edgeId), captor.capture(), anyLong())) .thenReturn(Futures.immediateFuture(mock(TimeseriesSaveResult.class))); - edgeStatsService = new EdgeStatsService( - tsService, - statsCounterService, - topicService, - Optional.of(kafkaAdmin) - ); - ReflectionTestUtils.setField(edgeStatsService, "edgesStatsTtlDays", 30); - ReflectionTestUtils.setField(edgeStatsService, "reportIntervalMillis", 600_000L); + edgeStatsService = createEdgeStatsService(Optional.of(kafkaAdmin)); - // when edgeStatsService.reportStats(); - // then + verify(tsService, times(1)).save(eq(tenantId), eq(edgeId), anyList(), anyLong()); + verify(statsCounterService, times(1)).clear(edgeId); + List entries = captor.getValue(); Map valuesByKey = entries.stream() .collect(Collectors.toMap(TsKvEntry::getKey, e -> e.getLongValue().orElse(-1L))); Assertions.assertEquals(15L, valuesByKey.get(DOWNLINK_MSGS_LAG.getKey())); - verify(statsCounterService).clear(edgeId); } } diff --git a/common/edge-api/src/main/java/org/thingsboard/edge/rpc/EdgeRpcClient.java b/common/edge-api/src/main/java/org/thingsboard/edge/rpc/EdgeRpcClient.java index 423c59251b..ed0d5b603e 100644 --- a/common/edge-api/src/main/java/org/thingsboard/edge/rpc/EdgeRpcClient.java +++ b/common/edge-api/src/main/java/org/thingsboard/edge/rpc/EdgeRpcClient.java @@ -41,4 +41,5 @@ public interface EdgeRpcClient { void sendDownlinkResponseMsg(DownlinkResponseMsg downlinkResponseMsg); int getServerMaxInboundMessageSize(); + } diff --git a/dao/src/main/java/org/thingsboard/server/dao/edge/stats/EdgeStats.java b/dao/src/main/java/org/thingsboard/server/dao/edge/stats/EdgeStats.java new file mode 100644 index 0000000000..36d87691cc --- /dev/null +++ b/dao/src/main/java/org/thingsboard/server/dao/edge/stats/EdgeStats.java @@ -0,0 +1,34 @@ +/** + * Copyright © 2016-2025 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.dao.edge.stats; + +import lombok.Data; +import org.thingsboard.server.common.data.id.TenantId; + +import java.util.Queue; +import java.util.concurrent.ConcurrentLinkedQueue; + +@Data +public class EdgeStats { + private final MsgCounters msgCounters; + private final Queue uplinkRate; + + public EdgeStats(TenantId tenantId) { + this.msgCounters = new MsgCounters(tenantId); + this.uplinkRate = new ConcurrentLinkedQueue<>(); + } + +} diff --git a/dao/src/main/java/org/thingsboard/server/dao/edge/stats/EdgeStatsCounterService.java b/dao/src/main/java/org/thingsboard/server/dao/edge/stats/EdgeStatsCounterService.java index 16111cf514..c537fb75be 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/edge/stats/EdgeStatsCounterService.java +++ b/dao/src/main/java/org/thingsboard/server/dao/edge/stats/EdgeStatsCounterService.java @@ -30,28 +30,26 @@ import java.util.concurrent.ConcurrentHashMap; @Getter public class EdgeStatsCounterService { - private final ConcurrentHashMap counterByEdge = new ConcurrentHashMap<>(); + private final ConcurrentHashMap statsByEdge = new ConcurrentHashMap<>(); public void recordEvent(EdgeStatsKey type, TenantId tenantId, EdgeId edgeId, long value) { - MsgCounters counters = getOrCreateCounters(tenantId, edgeId); + EdgeStats edgeStats = getOrCreateEdgeStats(tenantId, edgeId); + MsgCounters counters = edgeStats.getMsgCounters(); switch (type) { case DOWNLINK_MSGS_ADDED -> counters.getMsgsAdded().addAndGet(value); case DOWNLINK_MSGS_PUSHED -> counters.getMsgsPushed().addAndGet(value); case DOWNLINK_MSGS_PERMANENTLY_FAILED -> counters.getMsgsPermanentlyFailed().addAndGet(value); case DOWNLINK_MSGS_TMP_FAILED -> counters.getMsgsTmpFailed().addAndGet(value); + case DOWNLINK_MSGS_LAG -> counters.getMsgsLag().set(value); } } - public void setDownlinkMsgsLag(TenantId tenantId, EdgeId edgeId, long value) { - getOrCreateCounters(tenantId, edgeId).getMsgsLag().set(value); + public EdgeStats getOrCreateEdgeStats(TenantId tenantId, EdgeId edgeId) { + return statsByEdge.computeIfAbsent(edgeId, id -> new EdgeStats(tenantId)); } public void clear(EdgeId edgeId) { - counterByEdge.remove(edgeId); - } - - public MsgCounters getOrCreateCounters(TenantId tenantId, EdgeId edgeId) { - return counterByEdge.computeIfAbsent(edgeId, id -> new MsgCounters(tenantId)); + statsByEdge.remove(edgeId); } }