Viacheslav Klimov
1 year ago
committed by
GitHub
GitHub
17 changed files with 646 additions and 7 deletions
@ -0,0 +1,150 @@ |
|||
/** |
|||
* 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.service.edge.stats; |
|||
|
|||
import com.google.common.util.concurrent.FutureCallback; |
|||
import com.google.common.util.concurrent.Futures; |
|||
import com.google.common.util.concurrent.ListenableFuture; |
|||
import com.google.common.util.concurrent.MoreExecutors; |
|||
import lombok.RequiredArgsConstructor; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.beans.factory.annotation.Value; |
|||
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; |
|||
import org.springframework.scheduling.annotation.Scheduled; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.server.common.data.id.EdgeId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
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.EdgeStatsCounterService; |
|||
import org.thingsboard.server.dao.edge.stats.MsgCounters; |
|||
import org.thingsboard.server.dao.timeseries.TimeseriesService; |
|||
import org.thingsboard.server.queue.discovery.TopicService; |
|||
import org.thingsboard.server.queue.kafka.TbKafkaAdmin; |
|||
import org.thingsboard.server.queue.util.TbCoreComponent; |
|||
|
|||
import java.util.Collections; |
|||
import java.util.HashMap; |
|||
import java.util.HashSet; |
|||
import java.util.List; |
|||
import java.util.Map; |
|||
import java.util.Optional; |
|||
import java.util.concurrent.TimeUnit; |
|||
import java.util.stream.Collectors; |
|||
|
|||
import static org.thingsboard.server.dao.edge.stats.EdgeStatsKey.DOWNLINK_MSGS_ADDED; |
|||
import static org.thingsboard.server.dao.edge.stats.EdgeStatsKey.DOWNLINK_MSGS_LAG; |
|||
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; |
|||
|
|||
@TbCoreComponent |
|||
@ConditionalOnProperty(prefix = "edges.stats", name = "enabled", havingValue = "true", matchIfMissing = false) |
|||
@RequiredArgsConstructor |
|||
@Service |
|||
@Slf4j |
|||
public class EdgeStatsService { |
|||
|
|||
private final TimeseriesService tsService; |
|||
private final EdgeStatsCounterService statsCounterService; |
|||
private final TopicService topicService; |
|||
private final Optional<TbKafkaAdmin> tbKafkaAdmin; |
|||
|
|||
@Value("${edges.stats.ttl:30}") |
|||
private int edgesStatsTtlDays; |
|||
@Value("${edges.stats.report-interval-millis:600000}") |
|||
private long reportIntervalMillis; |
|||
|
|||
|
|||
@Scheduled( |
|||
fixedDelayString = "${edges.stats.report-interval-millis:600000}", |
|||
initialDelayString = "${edges.stats.report-interval-millis:600000}" |
|||
) |
|||
public void reportStats() { |
|||
log.debug("Reporting Edge communication stats..."); |
|||
long now = System.currentTimeMillis(); |
|||
long ts = now - (now % reportIntervalMillis); |
|||
|
|||
Map<EdgeId, MsgCounters> countersByEdge = statsCounterService.getCounterByEdge(); |
|||
Map<EdgeId, Long> lagByEdgeId = tbKafkaAdmin.isPresent() ? getEdgeLagByEdgeId(countersByEdge) : Collections.emptyMap(); |
|||
Map<EdgeId, MsgCounters> countersByEdgeSnapshot = new HashMap<>(statsCounterService.getCounterByEdge()); |
|||
countersByEdgeSnapshot.forEach((edgeId, counters) -> { |
|||
TenantId tenantId = counters.getTenantId(); |
|||
|
|||
if (tbKafkaAdmin.isPresent()) { |
|||
counters.getMsgsLag().set(lagByEdgeId.getOrDefault(edgeId, 0L)); |
|||
} |
|||
List<TsKvEntry> statsEntries = List.of( |
|||
entry(ts, DOWNLINK_MSGS_ADDED.getKey(), counters.getMsgsAdded().get()), |
|||
entry(ts, DOWNLINK_MSGS_PUSHED.getKey(), counters.getMsgsPushed().get()), |
|||
entry(ts, DOWNLINK_MSGS_PERMANENTLY_FAILED.getKey(), counters.getMsgsPermanentlyFailed().get()), |
|||
entry(ts, DOWNLINK_MSGS_TMP_FAILED.getKey(), counters.getMsgsTmpFailed().get()), |
|||
entry(ts, DOWNLINK_MSGS_LAG.getKey(), counters.getMsgsLag().get()) |
|||
); |
|||
|
|||
log.trace("Reported Edge communication stats: {} tenantId - {}, edgeId - {}", statsEntries, tenantId, edgeId); |
|||
saveTs(tenantId, edgeId, statsEntries); |
|||
}); |
|||
} |
|||
|
|||
private Map<EdgeId, Long> getEdgeLagByEdgeId(Map<EdgeId, MsgCounters> countersByEdge) { |
|||
Map<EdgeId, String> edgeToTopicMap = countersByEdge.entrySet().stream() |
|||
.collect(Collectors.toMap( |
|||
Map.Entry::getKey, |
|||
e -> topicService.buildEdgeEventNotificationsTopicPartitionInfo(e.getValue().getTenantId(), e.getKey()).getTopic() |
|||
)); |
|||
|
|||
Map<String, Long> lagByTopic = tbKafkaAdmin.get().getTotalLagForGroupsBulk(new HashSet<>(edgeToTopicMap.values())); |
|||
|
|||
return edgeToTopicMap.entrySet().stream() |
|||
.collect(Collectors.toMap( |
|||
Map.Entry::getKey, |
|||
e -> lagByTopic.getOrDefault(e.getValue(), 0L) |
|||
)); |
|||
} |
|||
|
|||
private void saveTs(TenantId tenantId, EdgeId edgeId, List<TsKvEntry> statsEntries) { |
|||
try { |
|||
ListenableFuture<TimeseriesSaveResult> future = tsService.save( |
|||
tenantId, |
|||
edgeId, |
|||
statsEntries, |
|||
TimeUnit.DAYS.toSeconds(edgesStatsTtlDays) |
|||
); |
|||
|
|||
Futures.addCallback(future, new FutureCallback<>() { |
|||
@Override |
|||
public void onSuccess(TimeseriesSaveResult result) { |
|||
log.debug("Successfully saved edge time-series stats: {} for edge: {}", statsEntries, edgeId); |
|||
} |
|||
|
|||
@Override |
|||
public void onFailure(Throwable t) { |
|||
log.warn("Failed to save edge time-series stats for edge: {}", edgeId, t); |
|||
} |
|||
}, MoreExecutors.directExecutor()); |
|||
} finally { |
|||
statsCounterService.clear(edgeId); |
|||
} |
|||
} |
|||
|
|||
private BasicTsKvEntry entry(long ts, String key, long value) { |
|||
return new BasicTsKvEntry(ts, new LongDataEntry(key, value)); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,173 @@ |
|||
/** |
|||
* 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.service.edge; |
|||
|
|||
import com.google.common.util.concurrent.Futures; |
|||
import org.junit.jupiter.api.Assertions; |
|||
import org.junit.jupiter.api.BeforeEach; |
|||
import org.junit.jupiter.api.Test; |
|||
import org.junit.jupiter.api.extension.ExtendWith; |
|||
import org.mockito.ArgumentCaptor; |
|||
import org.mockito.Mock; |
|||
import org.mockito.junit.jupiter.MockitoExtension; |
|||
import org.springframework.test.util.ReflectionTestUtils; |
|||
import org.thingsboard.server.common.data.id.EdgeId; |
|||
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.EdgeStatsCounterService; |
|||
import org.thingsboard.server.dao.edge.stats.MsgCounters; |
|||
import org.thingsboard.server.dao.timeseries.TimeseriesService; |
|||
import org.thingsboard.server.queue.discovery.TopicService; |
|||
import org.thingsboard.server.queue.kafka.TbKafkaAdmin; |
|||
import org.thingsboard.server.service.edge.stats.EdgeStatsService; |
|||
|
|||
import java.util.List; |
|||
import java.util.Map; |
|||
import java.util.Optional; |
|||
import java.util.Set; |
|||
import java.util.UUID; |
|||
import java.util.concurrent.ConcurrentHashMap; |
|||
import java.util.stream.Collectors; |
|||
|
|||
import static org.mockito.ArgumentMatchers.anyLong; |
|||
import static org.mockito.ArgumentMatchers.eq; |
|||
import static org.mockito.Mockito.mock; |
|||
import static org.mockito.Mockito.verify; |
|||
import static org.mockito.Mockito.when; |
|||
import static org.thingsboard.server.dao.edge.stats.EdgeStatsKey.DOWNLINK_MSGS_ADDED; |
|||
import static org.thingsboard.server.dao.edge.stats.EdgeStatsKey.DOWNLINK_MSGS_LAG; |
|||
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; |
|||
|
|||
@ExtendWith(MockitoExtension.class) |
|||
public class EdgeStatsTest { |
|||
|
|||
@Mock |
|||
private TimeseriesService tsService; |
|||
@Mock |
|||
private TopicService topicService; |
|||
@Mock |
|||
private EdgeStatsCounterService statsCounterService; |
|||
private EdgeStatsService edgeStatsService; |
|||
|
|||
private final TenantId tenantId = TenantId.fromUUID(UUID.randomUUID()); |
|||
private final EdgeId edgeId = new EdgeId(UUID.randomUUID()); |
|||
|
|||
@BeforeEach |
|||
void setUp() { |
|||
edgeStatsService = new EdgeStatsService( |
|||
tsService, |
|||
statsCounterService, |
|||
topicService, |
|||
Optional.empty() |
|||
); |
|||
|
|||
ReflectionTestUtils.setField(edgeStatsService, "edgesStatsTtlDays", 30); |
|||
ReflectionTestUtils.setField(edgeStatsService, "reportIntervalMillis", 600_000L); |
|||
} |
|||
|
|||
@Test |
|||
public void testReportStatsSavesTelemetry() { |
|||
// given
|
|||
MsgCounters counters = new MsgCounters(tenantId); |
|||
counters.getMsgsAdded().set(5); |
|||
counters.getMsgsPushed().set(3); |
|||
counters.getMsgsPermanentlyFailed().set(1); |
|||
counters.getMsgsTmpFailed().set(0); |
|||
counters.getMsgsLag().set(10); |
|||
|
|||
ConcurrentHashMap<EdgeId, MsgCounters> countersByEdge = new ConcurrentHashMap<>(); |
|||
countersByEdge.put(edgeId, counters); |
|||
|
|||
when(statsCounterService.getCounterByEdge()).thenReturn(countersByEdge); |
|||
|
|||
ArgumentCaptor<List<TsKvEntry>> captor = ArgumentCaptor.forClass(List.class); |
|||
when(tsService.save(eq(tenantId), eq(edgeId), captor.capture(), anyLong())) |
|||
.thenReturn(Futures.immediateFuture(mock(TimeseriesSaveResult.class))); |
|||
|
|||
// when
|
|||
edgeStatsService.reportStats(); |
|||
|
|||
// then
|
|||
List<TsKvEntry> entries = captor.getValue(); |
|||
Assertions.assertEquals(5, entries.size()); |
|||
|
|||
Map<String, Long> valuesByKey = entries.stream() |
|||
.collect(Collectors.toMap(TsKvEntry::getKey, e -> e.getLongValue().orElse(-1L))); |
|||
|
|||
Assertions.assertEquals(5L, valuesByKey.get(DOWNLINK_MSGS_ADDED.getKey()).longValue()); |
|||
Assertions.assertEquals(3L, valuesByKey.get(DOWNLINK_MSGS_PUSHED.getKey()).longValue()); |
|||
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); |
|||
counters.getMsgsAdded().set(2); |
|||
counters.getMsgsPushed().set(2); |
|||
counters.getMsgsPermanentlyFailed().set(0); |
|||
counters.getMsgsTmpFailed().set(1); |
|||
counters.getMsgsLag().set(0); |
|||
|
|||
ConcurrentHashMap<EdgeId, MsgCounters> countersByEdge = new ConcurrentHashMap<>(); |
|||
countersByEdge.put(edgeId, counters); |
|||
|
|||
// mocks
|
|||
when(statsCounterService.getCounterByEdge()).thenReturn(countersByEdge); |
|||
|
|||
String topic = "edge-topic"; |
|||
TopicPartitionInfo partitionInfo = new TopicPartitionInfo(topic, tenantId, 0, false); |
|||
when(topicService.buildEdgeEventNotificationsTopicPartitionInfo(tenantId, edgeId)).thenReturn(partitionInfo); |
|||
|
|||
TbKafkaAdmin kafkaAdmin = mock(TbKafkaAdmin.class); |
|||
when(kafkaAdmin.getTotalLagForGroupsBulk(Set.of(topic))) |
|||
.thenReturn(Map.of(topic, 15L)); |
|||
|
|||
ArgumentCaptor<List<TsKvEntry>> captor = ArgumentCaptor.forClass(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); |
|||
|
|||
// when
|
|||
edgeStatsService.reportStats(); |
|||
|
|||
// then
|
|||
List<TsKvEntry> entries = captor.getValue(); |
|||
Map<String, Long> 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); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,57 @@ |
|||
/** |
|||
* 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.Getter; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.server.common.data.id.EdgeId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
|
|||
import java.util.concurrent.ConcurrentHashMap; |
|||
|
|||
@ConditionalOnProperty(prefix = "edges.stats", name = "enabled", havingValue = "true", matchIfMissing = false) |
|||
@Service |
|||
@Slf4j |
|||
@Getter |
|||
public class EdgeStatsCounterService { |
|||
|
|||
private final ConcurrentHashMap<EdgeId, MsgCounters> counterByEdge = new ConcurrentHashMap<>(); |
|||
|
|||
public void recordEvent(EdgeStatsKey type, TenantId tenantId, EdgeId edgeId, long value) { |
|||
MsgCounters counters = getOrCreateCounters(tenantId, edgeId); |
|||
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); |
|||
} |
|||
} |
|||
|
|||
public void setDownlinkMsgsLag(TenantId tenantId, EdgeId edgeId, long value) { |
|||
getOrCreateCounters(tenantId, edgeId).getMsgsLag().set(value); |
|||
} |
|||
|
|||
public void clear(EdgeId edgeId) { |
|||
counterByEdge.remove(edgeId); |
|||
} |
|||
|
|||
public MsgCounters getOrCreateCounters(TenantId tenantId, EdgeId edgeId) { |
|||
return counterByEdge.computeIfAbsent(edgeId, id -> new MsgCounters(tenantId)); |
|||
} |
|||
|
|||
} |
|||
@ -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.Getter; |
|||
|
|||
@Getter |
|||
public enum EdgeStatsKey { |
|||
DOWNLINK_MSGS_ADDED("downlinkMsgsAdded"), |
|||
DOWNLINK_MSGS_PUSHED("downlinkMsgsPushed"), |
|||
DOWNLINK_MSGS_PERMANENTLY_FAILED("downlinkMsgsPermanentlyFailed"), |
|||
DOWNLINK_MSGS_TMP_FAILED("downlinkMsgsTmpFailed"), |
|||
DOWNLINK_MSGS_LAG("downlinkMsgsLag"); |
|||
|
|||
private final String key; |
|||
|
|||
EdgeStatsKey(String key) { |
|||
this.key = key; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,41 @@ |
|||
/** |
|||
* 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.concurrent.atomic.AtomicLong; |
|||
|
|||
@Data |
|||
public class MsgCounters { |
|||
|
|||
private final TenantId tenantId; |
|||
private final AtomicLong msgsAdded = new AtomicLong(); |
|||
private final AtomicLong msgsPushed = new AtomicLong(); |
|||
private final AtomicLong msgsPermanentlyFailed = new AtomicLong(); |
|||
private final AtomicLong msgsTmpFailed = new AtomicLong(); |
|||
private final AtomicLong msgsLag = new AtomicLong(); |
|||
|
|||
public void clear() { |
|||
msgsAdded.set(0); |
|||
msgsPushed.set(0); |
|||
msgsPermanentlyFailed.set(0); |
|||
msgsTmpFailed.set(0); |
|||
msgsLag.set(0); |
|||
} |
|||
|
|||
} |
|||
Loading…
Reference in new issue