From d4c4bf56562b4a6dad43a02cc0b68654d95bb6ba Mon Sep 17 00:00:00 2001 From: YevhenBondarenko Date: Mon, 9 Sep 2024 22:11:22 +0200 Subject: [PATCH 01/19] implemented new API for gateway latency --- .../src/main/resources/thingsboard.yml | 2 + ...AbstractMqttTimeseriesIntegrationTest.java | 98 ++++++++++++++ .../data/device/profile/MqttTopics.java | 1 + .../transport/mqtt/MqttTransportContext.java | 8 +- .../transport/mqtt/MqttTransportHandler.java | 10 +- .../transport/mqtt/MqttTransportService.java | 2 +- .../mqtt/TbMqttTransportComponent.java | 26 ++++ .../mqtt/gateway/GatewayLatencyService.java | 123 ++++++++++++++++++ .../gateway/latency/GatewayLatencyData.java | 19 +++ .../gateway/latency/GatewayLatencyState.java | 119 +++++++++++++++++ .../mqtt/session/GatewaySessionHandler.java | 49 ++++++- 11 files changed, 451 insertions(+), 6 deletions(-) create mode 100644 common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/TbMqttTransportComponent.java create mode 100644 common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/GatewayLatencyService.java create mode 100644 common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/latency/GatewayLatencyData.java create mode 100644 common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/latency/GatewayLatencyState.java diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index 785f124583..56dfe0ab46 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -1020,6 +1020,8 @@ transport: # MQTT disconnect timeout in milliseconds. The time to wait for the client to disconnect after the server sends a disconnect message. disconnect_timeout: "${MQTT_DISCONNECT_TIMEOUT:1000}" msg_queue_size_per_device_limit: "${MQTT_MSG_QUEUE_SIZE_PER_DEVICE_LIMIT:100}" # messages await in the queue before the device connected state. This limit works on the low level before TenantProfileLimits mechanism + # Interval of periodic report of the gateway latency + gateway_latency_report_interval_sec: "${MQTT_GATEWAY_LATENCY_REPORT_INTERVAL_SEC:3600}" netty: # Netty leak detector level leak_detector_level: "${NETTY_LEAK_DETECTOR_LVL:DISABLED}" diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/telemetry/timeseries/AbstractMqttTimeseriesIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/telemetry/timeseries/AbstractMqttTimeseriesIntegrationTest.java index 7c8cdb5d6e..6deceb6fb9 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/telemetry/timeseries/AbstractMqttTimeseriesIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/telemetry/timeseries/AbstractMqttTimeseriesIntegrationTest.java @@ -18,30 +18,45 @@ package org.thingsboard.server.transport.mqtt.mqttv3.telemetry.timeseries; import com.fasterxml.jackson.core.type.TypeReference; import io.netty.handler.codec.mqtt.MqttQoS; import lombok.extern.slf4j.Slf4j; +import org.awaitility.Awaitility; import org.junit.Before; import org.junit.Test; +import org.mockito.ArgumentCaptor; +import org.springframework.boot.test.mock.mockito.SpyBean; +import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.transport.mqtt.AbstractMqttIntegrationTest; import org.thingsboard.server.transport.mqtt.MqttTestConfigProperties; +import org.thingsboard.server.transport.mqtt.gateway.GatewayLatencyService; +import org.thingsboard.server.transport.mqtt.gateway.latency.GatewayLatencyData; +import org.thingsboard.server.transport.mqtt.gateway.latency.GatewayLatencyState; import org.thingsboard.server.transport.mqtt.mqttv3.MqttTestCallback; import org.thingsboard.server.transport.mqtt.mqttv3.MqttTestClient; +import java.util.ArrayList; import java.util.Arrays; +import java.util.HashMap; import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.OptionalLong; +import java.util.Random; import java.util.Set; import java.util.concurrent.TimeUnit; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; +import static org.mockito.Mockito.any; +import static org.mockito.Mockito.eq; +import static org.mockito.Mockito.verify; import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; import static org.thingsboard.server.common.data.device.profile.MqttTopics.DEVICE_ATTRIBUTES_TOPIC; import static org.thingsboard.server.common.data.device.profile.MqttTopics.DEVICE_TELEMETRY_SHORT_JSON_TOPIC; import static org.thingsboard.server.common.data.device.profile.MqttTopics.DEVICE_TELEMETRY_SHORT_TOPIC; import static org.thingsboard.server.common.data.device.profile.MqttTopics.DEVICE_TELEMETRY_TOPIC; import static org.thingsboard.server.common.data.device.profile.MqttTopics.GATEWAY_CONNECT_TOPIC; +import static org.thingsboard.server.common.data.device.profile.MqttTopics.GATEWAY_LATENCY_TOPIC; import static org.thingsboard.server.common.data.device.profile.MqttTopics.GATEWAY_TELEMETRY_TOPIC; @Slf4j @@ -53,6 +68,9 @@ public abstract class AbstractMqttTimeseriesIntegrationTest extends AbstractMqtt protected static final String MALFORMED_JSON_PAYLOAD = "{\"key1\":, \"key2\":true, \"key3\": 3.0, \"key4\": 4," + " \"key5\": {\"someNumber\": 42, \"someArray\": [1,2,3], \"someNestedObject\": {\"key\": \"value\"}}}"; + @SpyBean + GatewayLatencyService gatewayLatencyService; + @Before public void beforeTest() throws Exception { MqttTestConfigProperties configProperties = MqttTestConfigProperties.builder() @@ -113,6 +131,86 @@ public abstract class AbstractMqttTimeseriesIntegrationTest extends AbstractMqtt client.disconnect(); } + @Test + public void testPushLatencyGateway() throws Exception { + MqttTestClient client = new MqttTestClient(); + client.connectAndWait(gatewayAccessToken); + + Map> gwLatencies = new HashMap<>(); + List transportLatencies = new ArrayList<>(); + + publishLatency(client, gwLatencies, transportLatencies, 5); + + gatewayLatencyService.reportLatency(); + + List actualKeys = getActualKeysList(savedGateway.getId(), List.of("latencyCheck")); + assertEquals("latencyCheck", actualKeys.get(0)); + + String telemetryUrl = String.format("/api/plugins/telemetry/DEVICE/%s/values/timeseries?startTs=%d&endTs=%d&keys=latencyCheck", savedGateway.getId(), 0, System.currentTimeMillis()); + + Map>> gatewayTelemetry = doGetAsyncTyped(telemetryUrl, new TypeReference<>() {}); + Map latencyCheckTelemetry = gatewayTelemetry.get("latencyCheck").get(0); + Map latencyCheckValue = JacksonUtil.fromString((String) latencyCheckTelemetry.get("value"), new TypeReference<>() {}); + assertNotNull(latencyCheckValue); + + long avgTransportLatency = (long) transportLatencies.stream().mapToLong(Long::longValue).average().getAsDouble(); + long minTransportLatency = transportLatencies.stream().mapToLong(Long::longValue).min().getAsLong(); + long maxTransportLatency = transportLatencies.stream().mapToLong(Long::longValue).max().getAsLong(); + + + gwLatencies.forEach((connectorName, gwLatencyList) -> { + long avgGwLatency = (long) gwLatencyList.stream().mapToLong(Long::longValue).average().getAsDouble(); + long minGwLatency = gwLatencyList.stream().mapToLong(Long::longValue).min().getAsLong(); + long maxGwLatency = gwLatencyList.stream().mapToLong(Long::longValue).max().getAsLong(); + + GatewayLatencyState.ConnectorLatencyResult connectorLatencyResult = latencyCheckValue.get(connectorName); + assertNotNull(connectorLatencyResult); + checkConnectorLatencyResult(connectorLatencyResult, avgGwLatency, minGwLatency, maxGwLatency, avgTransportLatency, minTransportLatency, maxTransportLatency); + }); + + client.disconnect(); + + Awaitility.await() + .atMost(5, TimeUnit.SECONDS) + .untilAsserted(() -> verify(gatewayLatencyService).onDeviceDisconnect(savedGateway.getId())); + } + + private void publishLatency(MqttTestClient client, Map> gwLatencies, List transportLatencies, int n) throws Exception { + Random random = new Random(); + for (int i = 0; i < n; i++) { + long publishedTs = System.currentTimeMillis(); + long gatewayLatencyA = random.nextLong(1000, 5000); + long gatewayLatencyB = random.nextLong(1200, 4500); + long transportReceiveTs = publishLatencyAndGetTransportReceiveTs(client, publishedTs, gatewayLatencyA, gatewayLatencyB); + gwLatencies.computeIfAbsent("connectorA", key -> new ArrayList<>()).add(gatewayLatencyA); + gwLatencies.computeIfAbsent("connectorB", key -> new ArrayList<>()).add(gatewayLatencyB); + transportLatencies.add(transportReceiveTs - publishedTs); + Thread.sleep(1); + } + } + + private long publishLatencyAndGetTransportReceiveTs(MqttTestClient client, long publishedTs, long gatewayLatencyA, long gatewayLatencyB) throws Exception { + List data = new ArrayList<>(); + data.add(new GatewayLatencyData("connectorA", publishedTs - gatewayLatencyA, publishedTs)); + data.add(new GatewayLatencyData("connectorB", publishedTs - gatewayLatencyB, publishedTs)); + + client.publishAndWait(GATEWAY_LATENCY_TOPIC, JacksonUtil.writeValueAsBytes(data)); + ArgumentCaptor transportReceiveTsCaptor = ArgumentCaptor.forClass(Long.class); + verify(gatewayLatencyService).process(any(), eq(savedGateway.getId()), eq(data), transportReceiveTsCaptor.capture()); + return transportReceiveTsCaptor.getValue(); + } + + private void checkConnectorLatencyResult(GatewayLatencyState.ConnectorLatencyResult result, long avgGwLatency, long minGwLatency, long maxGwLatency, + long avgTransportLatency, long minTransportLatency, long maxTransportLatency) { + assertNotNull(result); + assertEquals(avgGwLatency, result.avgGwLatency()); + assertEquals(minGwLatency, result.minGwLatency()); + assertEquals(maxGwLatency, result.maxGwLatency()); + assertEquals(avgTransportLatency, result.transportLatencyAvg()); + assertEquals(minTransportLatency, result.minTransportLatency()); + assertEquals(maxTransportLatency, result.maxTransportLatency()); + } + protected void processJsonPayloadTelemetryTest(String topic, List expectedKeys, byte[] payload, boolean withTs) throws Exception { processTelemetryTest(topic, expectedKeys, payload, withTs, false); } diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/device/profile/MqttTopics.java b/common/data/src/main/java/org/thingsboard/server/common/data/device/profile/MqttTopics.java index cabdea145c..8945f2d084 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/device/profile/MqttTopics.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/device/profile/MqttTopics.java @@ -75,6 +75,7 @@ public class MqttTopics { public static final String GATEWAY_RPC_TOPIC = BASE_GATEWAY_API_TOPIC + RPC; public static final String GATEWAY_ATTRIBUTES_REQUEST_TOPIC = BASE_GATEWAY_API_TOPIC + ATTRIBUTES_REQUEST; public static final String GATEWAY_ATTRIBUTES_RESPONSE_TOPIC = BASE_GATEWAY_API_TOPIC + ATTRIBUTES_RESPONSE; + public static final String GATEWAY_LATENCY_TOPIC = BASE_GATEWAY_API_TOPIC + "/latency"; // v2 topics public static final String BASE_DEVICE_API_TOPIC_V2 = "v2"; public static final String REQUEST_ID_PATTERN = "(?\\d+)"; diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportContext.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportContext.java index 16d72ff89f..555d7de868 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportContext.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportContext.java @@ -22,11 +22,11 @@ import lombok.Setter; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; -import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; import org.springframework.stereotype.Component; import org.thingsboard.server.common.transport.TransportContext; import org.thingsboard.server.transport.mqtt.adaptors.JsonMqttAdaptor; import org.thingsboard.server.transport.mqtt.adaptors.ProtoMqttAdaptor; +import org.thingsboard.server.transport.mqtt.gateway.GatewayLatencyService; import java.net.InetSocketAddress; import java.util.concurrent.atomic.AtomicInteger; @@ -36,7 +36,7 @@ import java.util.concurrent.atomic.AtomicInteger; */ @Slf4j @Component -@ConditionalOnExpression("'${service.type:null}'=='tb-transport' || ('${service.type:null}'=='monolith' && '${transport.api_enabled:true}'=='true' && '${transport.mqtt.enabled}'=='true')") +@TbMqttTransportComponent public class MqttTransportContext extends TransportContext { @Getter @@ -51,6 +51,10 @@ public class MqttTransportContext extends TransportContext { @Autowired private ProtoMqttAdaptor protoMqttAdaptor; + @Getter + @Autowired + private GatewayLatencyService gatewayLatencyService; + @Getter @Value("${transport.mqtt.netty.max_payload_size}") private Integer maxPayloadSize; diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java index eeeb287dc8..ef63fb5902 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java @@ -419,6 +419,9 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement case MqttTopics.GATEWAY_DISCONNECT_TOPIC: gatewaySessionHandler.onDeviceDisconnect(mqttMsg); break; + case MqttTopics.GATEWAY_LATENCY_TOPIC: + gatewaySessionHandler.onGatewayLatency(mqttMsg); + break; default: ack(ctx, msgId, MqttReasonCodes.PubAck.TOPIC_NAME_INVALID); } @@ -1187,7 +1190,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement transportService.process(deviceSessionCtx.getSessionInfo(), SESSION_EVENT_MSG_CLOSED, null); transportService.deregisterSession(deviceSessionCtx.getSessionInfo()); if (gatewaySessionHandler != null) { - gatewaySessionHandler.onDevicesDisconnect(); + gatewaySessionHandler.onGatewayDisconnect(); } if (sparkplugSessionHandler != null) { // add Msg Telemetry node: key STATE type: String value: OFFLINE ts: sparkplugBProto.getTimestamp() @@ -1418,7 +1421,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement public void onDeviceUpdate(TransportProtos.SessionInfoProto sessionInfo, Device device, Optional deviceProfileOpt) { deviceSessionCtx.onDeviceUpdate(sessionInfo, device, deviceProfileOpt); if (gatewaySessionHandler != null) { - gatewaySessionHandler.onDeviceUpdate(sessionInfo, device, deviceProfileOpt); + gatewaySessionHandler.onGatewayUpdate(sessionInfo, device, deviceProfileOpt); } } @@ -1427,6 +1430,9 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement context.onAuthFailure(address); ChannelHandlerContext ctx = deviceSessionCtx.getChannel(); closeCtx(ctx, MqttReasonCodes.Disconnect.ADMINISTRATIVE_ACTION); + if (gatewaySessionHandler != null) { + gatewaySessionHandler.onGatewayDelete(deviceId); + } } public void sendErrorRpcResponse(TransportProtos.SessionInfoProto sessionInfo, int requestId, ThingsboardErrorCode result, String errorMsg) { diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportService.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportService.java index ffc7e9e561..b8d1dbe582 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportService.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportService.java @@ -39,7 +39,7 @@ import java.net.InetSocketAddress; * @author Andrew Shvayka */ @Service("MqttTransportService") -@ConditionalOnExpression("'${service.type:null}'=='tb-transport' || ('${service.type:null}'=='monolith' && '${transport.api_enabled:true}'=='true' && '${transport.mqtt.enabled}'=='true')") +@TbMqttTransportComponent @Slf4j public class MqttTransportService implements TbTransportService { diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/TbMqttTransportComponent.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/TbMqttTransportComponent.java new file mode 100644 index 0000000000..da0047cc7e --- /dev/null +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/TbMqttTransportComponent.java @@ -0,0 +1,26 @@ +/** + * Copyright © 2016-2024 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.transport.mqtt; + +import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; + +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; + +@Retention(RetentionPolicy.RUNTIME) +@ConditionalOnExpression("'${service.type:null}'=='tb-transport' || ('${service.type:null}'=='monolith' && '${transport.api_enabled:true}'=='true' && '${transport.mqtt.enabled}'=='true')") +public @interface TbMqttTransportComponent { +} diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/GatewayLatencyService.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/GatewayLatencyService.java new file mode 100644 index 0000000000..c6e9002877 --- /dev/null +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/GatewayLatencyService.java @@ -0,0 +1,123 @@ +/** + * Copyright © 2016-2024 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.transport.mqtt.gateway; + +import jakarta.annotation.PostConstruct; +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.stereotype.Service; +import org.thingsboard.common.util.JacksonUtil; +import org.thingsboard.server.common.data.id.DeviceId; +import org.thingsboard.server.common.transport.TransportService; +import org.thingsboard.server.common.transport.TransportServiceCallback; +import org.thingsboard.server.gen.transport.TransportProtos; +import org.thingsboard.server.queue.scheduler.SchedulerComponent; +import org.thingsboard.server.transport.mqtt.TbMqttTransportComponent; +import org.thingsboard.server.transport.mqtt.gateway.latency.GatewayLatencyData; +import org.thingsboard.server.transport.mqtt.gateway.latency.GatewayLatencyState; + +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.TimeUnit; + +@Slf4j +@Service +@TbMqttTransportComponent +public class GatewayLatencyService { + + @Value("${transport.mqtt.gateway_latency_report_interval_sec:3600}") + private int latencyReportIntervalSec; + + @Autowired + private SchedulerComponent scheduler; + + @Autowired + private TransportService transportService; + + private Map states = new ConcurrentHashMap<>(); + + @PostConstruct + private void init() { + scheduler.scheduleAtFixedRate(this::reportLatency, latencyReportIntervalSec, latencyReportIntervalSec, TimeUnit.SECONDS); + } + + public void process(TransportProtos.SessionInfoProto sessionInfo, DeviceId gatewayId, List data, long ts) { + states.computeIfAbsent(gatewayId, k -> new GatewayLatencyState(sessionInfo)).update(ts, data); + } + + public void onDeviceUpdate(TransportProtos.SessionInfoProto sessionInfo, DeviceId gatewayId) { + var state = states.get(gatewayId); + if (state != null) { + state.updateSessionInfo(sessionInfo); + } + } + + public void onDeviceDelete(DeviceId deviceId) { + var state = states.remove(deviceId); + if (state != null) { + state.clear(); + } + } + + public void onDeviceDisconnect(DeviceId deviceId) { + GatewayLatencyState state = states.remove(deviceId); + if (state != null) { + reportLatency(state, System.currentTimeMillis()); + } + } + + public void reportLatency() { + if (states.isEmpty()) { + return; + } + Map oldStates = states; + states = new ConcurrentHashMap<>(); + + long ts = System.currentTimeMillis(); + + oldStates.forEach((gatewayId, state) -> { + reportLatency(state, ts); + }); + oldStates.clear(); + } + + private void reportLatency(GatewayLatencyState state, long ts) { + if (state.isEmpty()) { + return; + } + var result = state.getLatencyStateResult(); + state.clear(); + var kvProto = TransportProtos.KeyValueProto.newBuilder() + .setKey("latencyCheck") + .setType(TransportProtos.KeyValueType.JSON_V) + .setJsonV(JacksonUtil.toString(result)) + .build(); + + TransportProtos.TsKvListProto tsKvList = TransportProtos.TsKvListProto.newBuilder() + .setTs(ts) + .addKv(kvProto) + .build(); + + TransportProtos.PostTelemetryMsg telemetryMsg = TransportProtos.PostTelemetryMsg.newBuilder() + .addTsKvList(tsKvList) + .build(); + + transportService.process(state.getSessionInfo(), telemetryMsg, TransportServiceCallback.EMPTY); + } + +} diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/latency/GatewayLatencyData.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/latency/GatewayLatencyData.java new file mode 100644 index 0000000000..6260c8356e --- /dev/null +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/latency/GatewayLatencyData.java @@ -0,0 +1,19 @@ +/** + * Copyright © 2016-2024 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.transport.mqtt.gateway.latency; + +public record GatewayLatencyData(String connectorName, long receivedTs, long publishedTs) { +} diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/latency/GatewayLatencyState.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/latency/GatewayLatencyState.java new file mode 100644 index 0000000000..3ed902f81d --- /dev/null +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/latency/GatewayLatencyState.java @@ -0,0 +1,119 @@ +/** + * Copyright © 2016-2024 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.transport.mqtt.gateway.latency; + +import lombok.Getter; +import org.thingsboard.server.gen.transport.TransportProtos; + +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; + +public class GatewayLatencyState { + + private final Map connectors; + private final Lock updateLock; + + @Getter + private volatile TransportProtos.SessionInfoProto sessionInfo; + + public GatewayLatencyState(TransportProtos.SessionInfoProto sessionInfo) { + this.connectors = new HashMap<>(); + this.updateLock = new ReentrantLock(); + this.sessionInfo = sessionInfo; + } + + public void updateSessionInfo(TransportProtos.SessionInfoProto sessionInfo) { + this.sessionInfo = sessionInfo; + } + + public void update(long ts, List latencyData) { + updateLock.lock(); + try { + latencyData.forEach(data -> { + connectors.computeIfAbsent(data.connectorName(), k -> new ConnectorLatencyState()).update(ts, data); + }); + } finally { + updateLock.unlock(); + } + } + + public void clear() { + connectors.clear(); + } + + public Map getLatencyStateResult() { + Map result = new HashMap<>(); + connectors.forEach((name, state) -> result.put(name, state.getResult())); + return result; + } + + public boolean isEmpty() { + return connectors.isEmpty(); + } + + private static class ConnectorLatencyState { + private final AtomicInteger count; + private final AtomicLong gwLatencySum; + private final AtomicLong transportLatencySum; + private volatile long minGwLatency; + private volatile long maxGwLatency; + private volatile long minTransportLatency; + private volatile long maxTransportLatency; + + private ConnectorLatencyState() { + this.count = new AtomicInteger(0); + this.gwLatencySum = new AtomicLong(0); + this.transportLatencySum = new AtomicLong(0); + } + + private void update(long serverReceiveTs, GatewayLatencyData latencyData) { + long gwLatency = latencyData.publishedTs() - latencyData.receivedTs(); + long transportLatency = serverReceiveTs - latencyData.publishedTs(); + count.incrementAndGet(); + gwLatencySum.addAndGet(gwLatency); + transportLatencySum.addAndGet(transportLatency); + if (minGwLatency == 0 || minGwLatency > gwLatency) { + minGwLatency = gwLatency; + } + if (maxGwLatency < gwLatency) { + maxGwLatency = gwLatency; + } + if (minTransportLatency == 0 || minTransportLatency > transportLatency) { + minTransportLatency = transportLatency; + } + if (maxTransportLatency < transportLatency) { + maxTransportLatency = transportLatency; + } + } + + private ConnectorLatencyResult getResult() { + long count = this.count.get(); + long avgGwLatency = gwLatencySum.get() / count; + long transportLatencyAvg = transportLatencySum.get() / count; + return new ConnectorLatencyResult(avgGwLatency, minGwLatency, maxGwLatency, transportLatencyAvg, minTransportLatency, maxTransportLatency); + } + } + + public record ConnectorLatencyResult(long avgGwLatency, long minGwLatency, long maxGwLatency, + long transportLatencyAvg, long minTransportLatency, long maxTransportLatency) { + } + +} diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewaySessionHandler.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewaySessionHandler.java index 9aa228afff..684f6b3114 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewaySessionHandler.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewaySessionHandler.java @@ -15,20 +15,36 @@ */ package org.thingsboard.server.transport.mqtt.session; +import com.fasterxml.jackson.core.type.TypeReference; import io.netty.buffer.ByteBuf; import io.netty.handler.codec.mqtt.MqttPublishMessage; +import io.netty.handler.codec.mqtt.MqttReasonCodes; +import lombok.extern.slf4j.Slf4j; +import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.server.common.adaptor.AdaptorException; +import org.thingsboard.server.common.data.Device; +import org.thingsboard.server.common.data.DeviceProfile; +import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.transport.auth.GetOrCreateDeviceFromGatewayResponse; +import org.thingsboard.server.gen.transport.TransportProtos; +import org.thingsboard.server.transport.mqtt.gateway.GatewayLatencyService; +import java.util.Optional; import java.util.UUID; +import static com.amazonaws.util.StringUtils.UTF8; + /** * Created by nickAS21 on 26.12.22 */ +@Slf4j public class GatewaySessionHandler extends AbstractGatewaySessionHandler { + private final GatewayLatencyService latencyService; + public GatewaySessionHandler(DeviceSessionCtx deviceSessionCtx, UUID sessionId, boolean overwriteDevicesActivity) { super(deviceSessionCtx, sessionId, overwriteDevicesActivity); + this.latencyService = deviceSessionCtx.getContext().getGatewayLatencyService(); } public void onDeviceConnect(MqttPublishMessage mqttMsg) throws AdaptorException { @@ -51,7 +67,38 @@ public class GatewaySessionHandler extends AbstractGatewaySessionHandler deviceProfileOpt) { + this.onDeviceUpdate(sessionInfo, device, deviceProfileOpt); + latencyService.onDeviceUpdate(sessionInfo, gateway.getDeviceId()); + } + + public void onGatewayDelete(DeviceId deviceId) { + latencyService.onDeviceDelete(deviceId); + } + + public void onGatewayLatency(MqttPublishMessage mqttMsg) throws AdaptorException { + int msgId = getMsgId(mqttMsg); + ByteBuf payloadData = mqttMsg.payload(); + String payload = payloadData.toString(UTF8); + if (payload == null) { + log.debug("[{}][{}][{}] Payload is empty!", gateway.getTenantId(), gateway.getDeviceId(), sessionId); + throw new AdaptorException(new IllegalArgumentException("Payload is empty!")); + } + long ts = System.currentTimeMillis(); + try { + latencyService.process(deviceSessionCtx.getSessionInfo(), gateway.getDeviceId(), JacksonUtil.fromString(payload, new TypeReference<>() {}), ts); + ack(msgId, MqttReasonCodes.PubAck.SUCCESS); + } catch (IllegalArgumentException e) { + throw new AdaptorException(e); + } } } From 06f30e1de534b6946f4dc08206024dca5de2e9e0 Mon Sep 17 00:00:00 2001 From: YevhenBondarenko Date: Mon, 9 Sep 2024 23:00:29 +0200 Subject: [PATCH 02/19] minor refactoring --- ...AbstractMqttTimeseriesIntegrationTest.java | 55 ++++++++++--------- .../mqtt/gateway/GatewayLatencyService.java | 3 +- .../gateway/latency/GatewayLatencyData.java | 2 +- .../gateway/latency/GatewayLatencyState.java | 7 +-- 4 files changed, 34 insertions(+), 33 deletions(-) diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/telemetry/timeseries/AbstractMqttTimeseriesIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/telemetry/timeseries/AbstractMqttTimeseriesIntegrationTest.java index 6deceb6fb9..2aeecfb663 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/telemetry/timeseries/AbstractMqttTimeseriesIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/telemetry/timeseries/AbstractMqttTimeseriesIntegrationTest.java @@ -40,7 +40,6 @@ import java.util.HashMap; import java.util.HashSet; import java.util.List; import java.util.Map; -import java.util.OptionalLong; import java.util.Random; import java.util.Set; import java.util.concurrent.TimeUnit; @@ -137,7 +136,7 @@ public abstract class AbstractMqttTimeseriesIntegrationTest extends AbstractMqtt client.connectAndWait(gatewayAccessToken); Map> gwLatencies = new HashMap<>(); - List transportLatencies = new ArrayList<>(); + Map> transportLatencies = new HashMap<>(); publishLatency(client, gwLatencies, transportLatencies, 5); @@ -153,16 +152,18 @@ public abstract class AbstractMqttTimeseriesIntegrationTest extends AbstractMqtt Map latencyCheckValue = JacksonUtil.fromString((String) latencyCheckTelemetry.get("value"), new TypeReference<>() {}); assertNotNull(latencyCheckValue); - long avgTransportLatency = (long) transportLatencies.stream().mapToLong(Long::longValue).average().getAsDouble(); - long minTransportLatency = transportLatencies.stream().mapToLong(Long::longValue).min().getAsLong(); - long maxTransportLatency = transportLatencies.stream().mapToLong(Long::longValue).max().getAsLong(); - - gwLatencies.forEach((connectorName, gwLatencyList) -> { long avgGwLatency = (long) gwLatencyList.stream().mapToLong(Long::longValue).average().getAsDouble(); long minGwLatency = gwLatencyList.stream().mapToLong(Long::longValue).min().getAsLong(); long maxGwLatency = gwLatencyList.stream().mapToLong(Long::longValue).max().getAsLong(); + List transportLatencyList = transportLatencies.get(connectorName); + assertNotNull(transportLatencyList); + + long avgTransportLatency = (long) transportLatencyList.stream().mapToLong(Long::longValue).average().getAsDouble(); + long minTransportLatency = transportLatencyList.stream().mapToLong(Long::longValue).min().getAsLong(); + long maxTransportLatency = transportLatencyList.stream().mapToLong(Long::longValue).max().getAsLong(); + GatewayLatencyState.ConnectorLatencyResult connectorLatencyResult = latencyCheckValue.get(connectorName); assertNotNull(connectorLatencyResult); checkConnectorLatencyResult(connectorLatencyResult, avgGwLatency, minGwLatency, maxGwLatency, avgTransportLatency, minTransportLatency, maxTransportLatency); @@ -175,29 +176,31 @@ public abstract class AbstractMqttTimeseriesIntegrationTest extends AbstractMqtt .untilAsserted(() -> verify(gatewayLatencyService).onDeviceDisconnect(savedGateway.getId())); } - private void publishLatency(MqttTestClient client, Map> gwLatencies, List transportLatencies, int n) throws Exception { + private void publishLatency(MqttTestClient client, Map> gwLatencies, Map> transportLatencies, int n) throws Exception { Random random = new Random(); for (int i = 0; i < n; i++) { - long publishedTs = System.currentTimeMillis(); - long gatewayLatencyA = random.nextLong(1000, 5000); - long gatewayLatencyB = random.nextLong(1200, 4500); - long transportReceiveTs = publishLatencyAndGetTransportReceiveTs(client, publishedTs, gatewayLatencyA, gatewayLatencyB); + Map data = new HashMap<>(); + long publishedTs = System.currentTimeMillis() - 10; + long gatewayLatencyA = random.nextLong(100, 500); + data.put("connectorA", new GatewayLatencyData(publishedTs - gatewayLatencyA, publishedTs)); gwLatencies.computeIfAbsent("connectorA", key -> new ArrayList<>()).add(gatewayLatencyA); - gwLatencies.computeIfAbsent("connectorB", key -> new ArrayList<>()).add(gatewayLatencyB); - transportLatencies.add(transportReceiveTs - publishedTs); - Thread.sleep(1); - } - } - - private long publishLatencyAndGetTransportReceiveTs(MqttTestClient client, long publishedTs, long gatewayLatencyA, long gatewayLatencyB) throws Exception { - List data = new ArrayList<>(); - data.add(new GatewayLatencyData("connectorA", publishedTs - gatewayLatencyA, publishedTs)); - data.add(new GatewayLatencyData("connectorB", publishedTs - gatewayLatencyB, publishedTs)); + boolean sendB = i % 2 == 0; + if (sendB) { + long gatewayLatencyB = random.nextLong(120, 450); + data.put("connectorB", new GatewayLatencyData(publishedTs - gatewayLatencyB, publishedTs)); + gwLatencies.computeIfAbsent("connectorB", key -> new ArrayList<>()).add(gatewayLatencyB); + } - client.publishAndWait(GATEWAY_LATENCY_TOPIC, JacksonUtil.writeValueAsBytes(data)); - ArgumentCaptor transportReceiveTsCaptor = ArgumentCaptor.forClass(Long.class); - verify(gatewayLatencyService).process(any(), eq(savedGateway.getId()), eq(data), transportReceiveTsCaptor.capture()); - return transportReceiveTsCaptor.getValue(); + client.publishAndWait(GATEWAY_LATENCY_TOPIC, JacksonUtil.writeValueAsBytes(data)); + ArgumentCaptor transportReceiveTsCaptor = ArgumentCaptor.forClass(Long.class); + verify(gatewayLatencyService).process(any(), eq(savedGateway.getId()), eq(data), transportReceiveTsCaptor.capture()); + Long transportReceiveTs = transportReceiveTsCaptor.getValue(); + Long transportLatency = transportReceiveTs - publishedTs; + transportLatencies.computeIfAbsent("connectorA", key -> new ArrayList<>()).add(transportLatency); + if (sendB) { + transportLatencies.computeIfAbsent("connectorB", key -> new ArrayList<>()).add(transportLatency); + } + } } private void checkConnectorLatencyResult(GatewayLatencyState.ConnectorLatencyResult result, long avgGwLatency, long minGwLatency, long maxGwLatency, diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/GatewayLatencyService.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/GatewayLatencyService.java index c6e9002877..148a8819ee 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/GatewayLatencyService.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/GatewayLatencyService.java @@ -30,7 +30,6 @@ import org.thingsboard.server.transport.mqtt.TbMqttTransportComponent; import org.thingsboard.server.transport.mqtt.gateway.latency.GatewayLatencyData; import org.thingsboard.server.transport.mqtt.gateway.latency.GatewayLatencyState; -import java.util.List; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.TimeUnit; @@ -56,7 +55,7 @@ public class GatewayLatencyService { scheduler.scheduleAtFixedRate(this::reportLatency, latencyReportIntervalSec, latencyReportIntervalSec, TimeUnit.SECONDS); } - public void process(TransportProtos.SessionInfoProto sessionInfo, DeviceId gatewayId, List data, long ts) { + public void process(TransportProtos.SessionInfoProto sessionInfo, DeviceId gatewayId, Map data, long ts) { states.computeIfAbsent(gatewayId, k -> new GatewayLatencyState(sessionInfo)).update(ts, data); } diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/latency/GatewayLatencyData.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/latency/GatewayLatencyData.java index 6260c8356e..2cae428a02 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/latency/GatewayLatencyData.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/latency/GatewayLatencyData.java @@ -15,5 +15,5 @@ */ package org.thingsboard.server.transport.mqtt.gateway.latency; -public record GatewayLatencyData(String connectorName, long receivedTs, long publishedTs) { +public record GatewayLatencyData(long receivedTs, long publishedTs) { } diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/latency/GatewayLatencyState.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/latency/GatewayLatencyState.java index 3ed902f81d..c139bb3f19 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/latency/GatewayLatencyState.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/latency/GatewayLatencyState.java @@ -19,7 +19,6 @@ import lombok.Getter; import org.thingsboard.server.gen.transport.TransportProtos; import java.util.HashMap; -import java.util.List; import java.util.Map; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; @@ -44,11 +43,11 @@ public class GatewayLatencyState { this.sessionInfo = sessionInfo; } - public void update(long ts, List latencyData) { + public void update(long ts, Map latencyData) { updateLock.lock(); try { - latencyData.forEach(data -> { - connectors.computeIfAbsent(data.connectorName(), k -> new ConnectorLatencyState()).update(ts, data); + latencyData.forEach((connectorName, data) -> { + connectors.computeIfAbsent(connectorName, k -> new ConnectorLatencyState()).update(ts, data); }); } finally { updateLock.unlock(); From db88ed1aab4c4d300830ad26dcaeea632b5d59eb Mon Sep 17 00:00:00 2001 From: Ikko Eltociear Ashimine Date: Wed, 11 Sep 2024 01:08:41 +0900 Subject: [PATCH 03/19] chore: update EdgeGrpcSession.java attemps -> attempts --- .../thingsboard/server/service/edge/rpc/EdgeGrpcSession.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 c397807c5f..d0d10f0174 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 @@ -110,7 +110,7 @@ public final class EdgeGrpcSession implements Closeable { private static final ReentrantLock downlinkMsgLock = new ReentrantLock(); private static final ConcurrentLinkedQueue highPriorityQueue = new ConcurrentLinkedQueue<>(); - private static final int MAX_DOWNLINK_ATTEMPTS = 10; // max number of attemps to send downlink message if edge connected + private static final int MAX_DOWNLINK_ATTEMPTS = 10; // max number of attempts to send downlink message if edge connected private static final String QUEUE_START_TS_ATTR_KEY = "queueStartTs"; private static final String QUEUE_START_SEQ_ID_ATTR_KEY = "queueStartSeqId"; From f64ec5ca56f9c884936bd22a40969d539394594c Mon Sep 17 00:00:00 2001 From: YevhenBondarenko Date: Wed, 11 Sep 2024 10:25:24 +0200 Subject: [PATCH 04/19] test fixes --- .../timeseries/AbstractMqttTimeseriesIntegrationTest.java | 7 +++++++ .../transport/mqtt/session/GatewaySessionHandler.java | 4 ++-- 2 files changed, 9 insertions(+), 2 deletions(-) diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/telemetry/timeseries/AbstractMqttTimeseriesIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/telemetry/timeseries/AbstractMqttTimeseriesIntegrationTest.java index 2aeecfb663..886cf1e309 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/telemetry/timeseries/AbstractMqttTimeseriesIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/telemetry/timeseries/AbstractMqttTimeseriesIntegrationTest.java @@ -25,6 +25,7 @@ import org.mockito.ArgumentCaptor; import org.springframework.boot.test.mock.mockito.SpyBean; import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.server.common.data.Device; +import org.thingsboard.server.common.data.TransportPayloadType; import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.transport.mqtt.AbstractMqttIntegrationTest; import org.thingsboard.server.transport.mqtt.MqttTestConfigProperties; @@ -57,6 +58,7 @@ import static org.thingsboard.server.common.data.device.profile.MqttTopics.DEVIC import static org.thingsboard.server.common.data.device.profile.MqttTopics.GATEWAY_CONNECT_TOPIC; import static org.thingsboard.server.common.data.device.profile.MqttTopics.GATEWAY_LATENCY_TOPIC; import static org.thingsboard.server.common.data.device.profile.MqttTopics.GATEWAY_TELEMETRY_TOPIC; +import static org.thingsboard.server.transport.mqtt.mqttv3.credentials.BasicMqttCredentialsTest.CLIENT_ID; @Slf4j public abstract class AbstractMqttTimeseriesIntegrationTest extends AbstractMqttIntegrationTest { @@ -132,6 +134,11 @@ public abstract class AbstractMqttTimeseriesIntegrationTest extends AbstractMqtt @Test public void testPushLatencyGateway() throws Exception { + MqttTestConfigProperties configProperties = MqttTestConfigProperties.builder() + .gatewayName("Test latency gateway") + .build(); + processBeforeTest(configProperties); + MqttTestClient client = new MqttTestClient(); client.connectAndWait(gatewayAccessToken); diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewaySessionHandler.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewaySessionHandler.java index 684f6b3114..a7634593ce 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewaySessionHandler.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewaySessionHandler.java @@ -96,8 +96,8 @@ public class GatewaySessionHandler extends AbstractGatewaySessionHandler() {}), ts); ack(msgId, MqttReasonCodes.PubAck.SUCCESS); - } catch (IllegalArgumentException e) { - throw new AdaptorException(e); + } catch (Throwable t) { + ackOrClose(msgId); } } From dfb633b8e8a31a19b890c3158ed2067f172f0458 Mon Sep 17 00:00:00 2001 From: YevhenBondarenko Date: Wed, 11 Sep 2024 11:26:04 +0200 Subject: [PATCH 05/19] improvements --- .../mqtt/gateway/GatewayLatencyService.java | 6 +----- .../mqtt/gateway/latency/GatewayLatencyState.java | 13 ++++++++----- 2 files changed, 9 insertions(+), 10 deletions(-) diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/GatewayLatencyService.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/GatewayLatencyService.java index 148a8819ee..acf10a03b8 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/GatewayLatencyService.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/GatewayLatencyService.java @@ -67,10 +67,7 @@ public class GatewayLatencyService { } public void onDeviceDelete(DeviceId deviceId) { - var state = states.remove(deviceId); - if (state != null) { - state.clear(); - } + states.remove(deviceId); } public void onDeviceDisconnect(DeviceId deviceId) { @@ -100,7 +97,6 @@ public class GatewayLatencyService { return; } var result = state.getLatencyStateResult(); - state.clear(); var kvProto = TransportProtos.KeyValueProto.newBuilder() .setKey("latencyCheck") .setType(TransportProtos.KeyValueType.JSON_V) diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/latency/GatewayLatencyState.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/latency/GatewayLatencyState.java index c139bb3f19..3ad75fee75 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/latency/GatewayLatencyState.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/latency/GatewayLatencyState.java @@ -54,13 +54,16 @@ public class GatewayLatencyState { } } - public void clear() { - connectors.clear(); - } - public Map getLatencyStateResult() { Map result = new HashMap<>(); - connectors.forEach((name, state) -> result.put(name, state.getResult())); + updateLock.lock(); + try { + connectors.forEach((name, state) -> result.put(name, state.getResult())); + connectors.clear(); + } finally { + updateLock.unlock(); + } + return result; } From 914d4b8ca378957f522a9f0d551b22fd388ab333 Mon Sep 17 00:00:00 2001 From: YevhenBondarenko Date: Thu, 12 Sep 2024 14:00:40 +0200 Subject: [PATCH 06/19] ranemed latency to metrics --- .../src/main/resources/thingsboard.yml | 4 +- ...AbstractMqttTimeseriesIntegrationTest.java | 47 +++++++++---------- .../data/device/profile/MqttTopics.java | 2 +- .../transport/mqtt/MqttTransportContext.java | 4 +- .../transport/mqtt/MqttTransportHandler.java | 2 +- ...ervice.java => GatewayMetricsService.java} | 36 +++++++------- .../GatewayMetricsData.java} | 4 +- .../GatewayMetricsState.java} | 38 +++++++-------- .../mqtt/session/GatewaySessionHandler.java | 14 +++--- .../src/main/resources/tb-mqtt-transport.yml | 2 + 10 files changed, 78 insertions(+), 75 deletions(-) rename common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/{GatewayLatencyService.java => GatewayMetricsService.java} (73%) rename common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/{latency/GatewayLatencyData.java => metrics/GatewayMetricsData.java} (83%) rename common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/{latency/GatewayLatencyState.java => metrics/GatewayMetricsState.java} (71%) diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index 2fc35d915a..6b61838918 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -1020,8 +1020,8 @@ transport: # MQTT disconnect timeout in milliseconds. The time to wait for the client to disconnect after the server sends a disconnect message. disconnect_timeout: "${MQTT_DISCONNECT_TIMEOUT:1000}" msg_queue_size_per_device_limit: "${MQTT_MSG_QUEUE_SIZE_PER_DEVICE_LIMIT:100}" # messages await in the queue before the device connected state. This limit works on the low level before TenantProfileLimits mechanism - # Interval of periodic report of the gateway latency - gateway_latency_report_interval_sec: "${MQTT_GATEWAY_LATENCY_REPORT_INTERVAL_SEC:3600}" + # Interval of periodic report of the gateway metrics + gateway_metrics_report_interval_sec: "${MQTT_GATEWAY_METRICS_REPORT_INTERVAL_SEC:3600}" netty: # Netty leak detector level leak_detector_level: "${NETTY_LEAK_DETECTOR_LVL:DISABLED}" diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/telemetry/timeseries/AbstractMqttTimeseriesIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/telemetry/timeseries/AbstractMqttTimeseriesIntegrationTest.java index 886cf1e309..4f89a8018e 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/telemetry/timeseries/AbstractMqttTimeseriesIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/telemetry/timeseries/AbstractMqttTimeseriesIntegrationTest.java @@ -25,13 +25,12 @@ import org.mockito.ArgumentCaptor; import org.springframework.boot.test.mock.mockito.SpyBean; import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.server.common.data.Device; -import org.thingsboard.server.common.data.TransportPayloadType; import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.transport.mqtt.AbstractMqttIntegrationTest; import org.thingsboard.server.transport.mqtt.MqttTestConfigProperties; -import org.thingsboard.server.transport.mqtt.gateway.GatewayLatencyService; -import org.thingsboard.server.transport.mqtt.gateway.latency.GatewayLatencyData; -import org.thingsboard.server.transport.mqtt.gateway.latency.GatewayLatencyState; +import org.thingsboard.server.transport.mqtt.gateway.GatewayMetricsService; +import org.thingsboard.server.transport.mqtt.gateway.metrics.GatewayMetricsData; +import org.thingsboard.server.transport.mqtt.gateway.metrics.GatewayMetricsState; import org.thingsboard.server.transport.mqtt.mqttv3.MqttTestCallback; import org.thingsboard.server.transport.mqtt.mqttv3.MqttTestClient; @@ -56,9 +55,9 @@ import static org.thingsboard.server.common.data.device.profile.MqttTopics.DEVIC import static org.thingsboard.server.common.data.device.profile.MqttTopics.DEVICE_TELEMETRY_SHORT_TOPIC; import static org.thingsboard.server.common.data.device.profile.MqttTopics.DEVICE_TELEMETRY_TOPIC; import static org.thingsboard.server.common.data.device.profile.MqttTopics.GATEWAY_CONNECT_TOPIC; -import static org.thingsboard.server.common.data.device.profile.MqttTopics.GATEWAY_LATENCY_TOPIC; +import static org.thingsboard.server.common.data.device.profile.MqttTopics.GATEWAY_METRICS_TOPIC; import static org.thingsboard.server.common.data.device.profile.MqttTopics.GATEWAY_TELEMETRY_TOPIC; -import static org.thingsboard.server.transport.mqtt.mqttv3.credentials.BasicMqttCredentialsTest.CLIENT_ID; +import static org.thingsboard.server.transport.mqtt.gateway.GatewayMetricsService.METRICS_CHECK; @Slf4j public abstract class AbstractMqttTimeseriesIntegrationTest extends AbstractMqttIntegrationTest { @@ -70,7 +69,7 @@ public abstract class AbstractMqttTimeseriesIntegrationTest extends AbstractMqtt " \"key5\": {\"someNumber\": 42, \"someArray\": [1,2,3], \"someNestedObject\": {\"key\": \"value\"}}}"; @SpyBean - GatewayLatencyService gatewayLatencyService; + GatewayMetricsService gatewayMetricsService; @Before public void beforeTest() throws Exception { @@ -133,9 +132,9 @@ public abstract class AbstractMqttTimeseriesIntegrationTest extends AbstractMqtt } @Test - public void testPushLatencyGateway() throws Exception { + public void testPushMetricsGateway() throws Exception { MqttTestConfigProperties configProperties = MqttTestConfigProperties.builder() - .gatewayName("Test latency gateway") + .gatewayName("Test metrics gateway") .build(); processBeforeTest(configProperties); @@ -147,16 +146,16 @@ public abstract class AbstractMqttTimeseriesIntegrationTest extends AbstractMqtt publishLatency(client, gwLatencies, transportLatencies, 5); - gatewayLatencyService.reportLatency(); + gatewayMetricsService.reportMetrics(); - List actualKeys = getActualKeysList(savedGateway.getId(), List.of("latencyCheck")); - assertEquals("latencyCheck", actualKeys.get(0)); + List actualKeys = getActualKeysList(savedGateway.getId(), List.of(METRICS_CHECK)); + assertEquals(METRICS_CHECK, actualKeys.get(0)); - String telemetryUrl = String.format("/api/plugins/telemetry/DEVICE/%s/values/timeseries?startTs=%d&endTs=%d&keys=latencyCheck", savedGateway.getId(), 0, System.currentTimeMillis()); + String telemetryUrl = String.format("/api/plugins/telemetry/DEVICE/%s/values/timeseries?startTs=%d&endTs=%d&keys=%s", savedGateway.getId(), 0, System.currentTimeMillis(), METRICS_CHECK); Map>> gatewayTelemetry = doGetAsyncTyped(telemetryUrl, new TypeReference<>() {}); - Map latencyCheckTelemetry = gatewayTelemetry.get("latencyCheck").get(0); - Map latencyCheckValue = JacksonUtil.fromString((String) latencyCheckTelemetry.get("value"), new TypeReference<>() {}); + Map latencyCheckTelemetry = gatewayTelemetry.get(METRICS_CHECK).get(0); + Map latencyCheckValue = JacksonUtil.fromString((String) latencyCheckTelemetry.get("value"), new TypeReference<>() {}); assertNotNull(latencyCheckValue); gwLatencies.forEach((connectorName, gwLatencyList) -> { @@ -171,7 +170,7 @@ public abstract class AbstractMqttTimeseriesIntegrationTest extends AbstractMqtt long minTransportLatency = transportLatencyList.stream().mapToLong(Long::longValue).min().getAsLong(); long maxTransportLatency = transportLatencyList.stream().mapToLong(Long::longValue).max().getAsLong(); - GatewayLatencyState.ConnectorLatencyResult connectorLatencyResult = latencyCheckValue.get(connectorName); + GatewayMetricsState.ConnectorMetricsResult connectorLatencyResult = latencyCheckValue.get(connectorName); assertNotNull(connectorLatencyResult); checkConnectorLatencyResult(connectorLatencyResult, avgGwLatency, minGwLatency, maxGwLatency, avgTransportLatency, minTransportLatency, maxTransportLatency); }); @@ -180,27 +179,27 @@ public abstract class AbstractMqttTimeseriesIntegrationTest extends AbstractMqtt Awaitility.await() .atMost(5, TimeUnit.SECONDS) - .untilAsserted(() -> verify(gatewayLatencyService).onDeviceDisconnect(savedGateway.getId())); + .untilAsserted(() -> verify(gatewayMetricsService).onDeviceDisconnect(savedGateway.getId())); } private void publishLatency(MqttTestClient client, Map> gwLatencies, Map> transportLatencies, int n) throws Exception { Random random = new Random(); for (int i = 0; i < n; i++) { - Map data = new HashMap<>(); + Map data = new HashMap<>(); long publishedTs = System.currentTimeMillis() - 10; long gatewayLatencyA = random.nextLong(100, 500); - data.put("connectorA", new GatewayLatencyData(publishedTs - gatewayLatencyA, publishedTs)); + data.put("connectorA", new GatewayMetricsData(publishedTs - gatewayLatencyA, publishedTs)); gwLatencies.computeIfAbsent("connectorA", key -> new ArrayList<>()).add(gatewayLatencyA); boolean sendB = i % 2 == 0; if (sendB) { long gatewayLatencyB = random.nextLong(120, 450); - data.put("connectorB", new GatewayLatencyData(publishedTs - gatewayLatencyB, publishedTs)); + data.put("connectorB", new GatewayMetricsData(publishedTs - gatewayLatencyB, publishedTs)); gwLatencies.computeIfAbsent("connectorB", key -> new ArrayList<>()).add(gatewayLatencyB); } - client.publishAndWait(GATEWAY_LATENCY_TOPIC, JacksonUtil.writeValueAsBytes(data)); + client.publishAndWait(GATEWAY_METRICS_TOPIC, JacksonUtil.writeValueAsBytes(data)); ArgumentCaptor transportReceiveTsCaptor = ArgumentCaptor.forClass(Long.class); - verify(gatewayLatencyService).process(any(), eq(savedGateway.getId()), eq(data), transportReceiveTsCaptor.capture()); + verify(gatewayMetricsService).process(any(), eq(savedGateway.getId()), eq(data), transportReceiveTsCaptor.capture()); Long transportReceiveTs = transportReceiveTsCaptor.getValue(); Long transportLatency = transportReceiveTs - publishedTs; transportLatencies.computeIfAbsent("connectorA", key -> new ArrayList<>()).add(transportLatency); @@ -210,13 +209,13 @@ public abstract class AbstractMqttTimeseriesIntegrationTest extends AbstractMqtt } } - private void checkConnectorLatencyResult(GatewayLatencyState.ConnectorLatencyResult result, long avgGwLatency, long minGwLatency, long maxGwLatency, + private void checkConnectorLatencyResult(GatewayMetricsState.ConnectorMetricsResult result, long avgGwLatency, long minGwLatency, long maxGwLatency, long avgTransportLatency, long minTransportLatency, long maxTransportLatency) { assertNotNull(result); assertEquals(avgGwLatency, result.avgGwLatency()); assertEquals(minGwLatency, result.minGwLatency()); assertEquals(maxGwLatency, result.maxGwLatency()); - assertEquals(avgTransportLatency, result.transportLatencyAvg()); + assertEquals(avgTransportLatency, result.avgTransportLatency()); assertEquals(minTransportLatency, result.minTransportLatency()); assertEquals(maxTransportLatency, result.maxTransportLatency()); } diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/device/profile/MqttTopics.java b/common/data/src/main/java/org/thingsboard/server/common/data/device/profile/MqttTopics.java index 8945f2d084..0590d57034 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/device/profile/MqttTopics.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/device/profile/MqttTopics.java @@ -75,7 +75,7 @@ public class MqttTopics { public static final String GATEWAY_RPC_TOPIC = BASE_GATEWAY_API_TOPIC + RPC; public static final String GATEWAY_ATTRIBUTES_REQUEST_TOPIC = BASE_GATEWAY_API_TOPIC + ATTRIBUTES_REQUEST; public static final String GATEWAY_ATTRIBUTES_RESPONSE_TOPIC = BASE_GATEWAY_API_TOPIC + ATTRIBUTES_RESPONSE; - public static final String GATEWAY_LATENCY_TOPIC = BASE_GATEWAY_API_TOPIC + "/latency"; + public static final String GATEWAY_METRICS_TOPIC = BASE_GATEWAY_API_TOPIC + "/metrics"; // v2 topics public static final String BASE_DEVICE_API_TOPIC_V2 = "v2"; public static final String REQUEST_ID_PATTERN = "(?\\d+)"; diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportContext.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportContext.java index dc61a47d78..d78a4afbc2 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportContext.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportContext.java @@ -27,7 +27,7 @@ import org.thingsboard.server.common.transport.TransportContext; import org.thingsboard.server.common.transport.TransportTenantProfileCache; import org.thingsboard.server.transport.mqtt.adaptors.JsonMqttAdaptor; import org.thingsboard.server.transport.mqtt.adaptors.ProtoMqttAdaptor; -import org.thingsboard.server.transport.mqtt.gateway.GatewayLatencyService; +import org.thingsboard.server.transport.mqtt.gateway.GatewayMetricsService; import java.net.InetSocketAddress; import java.util.concurrent.atomic.AtomicInteger; @@ -58,7 +58,7 @@ public class MqttTransportContext extends TransportContext { @Getter @Autowired - private GatewayLatencyService gatewayLatencyService; + private GatewayMetricsService gatewayMetricsService; @Getter @Value("${transport.mqtt.netty.max_payload_size}") diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java index 4ae0d54a6b..1eee03f722 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java @@ -423,7 +423,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement case MqttTopics.GATEWAY_DISCONNECT_TOPIC: gatewaySessionHandler.onDeviceDisconnect(mqttMsg); break; - case MqttTopics.GATEWAY_LATENCY_TOPIC: + case MqttTopics.GATEWAY_METRICS_TOPIC: gatewaySessionHandler.onGatewayLatency(mqttMsg); break; default: diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/GatewayLatencyService.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/GatewayMetricsService.java similarity index 73% rename from common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/GatewayLatencyService.java rename to common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/GatewayMetricsService.java index acf10a03b8..263888f4c2 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/GatewayLatencyService.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/GatewayMetricsService.java @@ -27,8 +27,8 @@ import org.thingsboard.server.common.transport.TransportServiceCallback; import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.queue.scheduler.SchedulerComponent; import org.thingsboard.server.transport.mqtt.TbMqttTransportComponent; -import org.thingsboard.server.transport.mqtt.gateway.latency.GatewayLatencyData; -import org.thingsboard.server.transport.mqtt.gateway.latency.GatewayLatencyState; +import org.thingsboard.server.transport.mqtt.gateway.metrics.GatewayMetricsData; +import org.thingsboard.server.transport.mqtt.gateway.metrics.GatewayMetricsState; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; @@ -37,10 +37,12 @@ import java.util.concurrent.TimeUnit; @Slf4j @Service @TbMqttTransportComponent -public class GatewayLatencyService { +public class GatewayMetricsService { - @Value("${transport.mqtt.gateway_latency_report_interval_sec:3600}") - private int latencyReportIntervalSec; + public static final String METRICS_CHECK = "metricsCheck"; + + @Value("${transport.mqtt.gateway_metrics_report_interval_sec:3600}") + private int metricsReportIntervalSec; @Autowired private SchedulerComponent scheduler; @@ -48,15 +50,15 @@ public class GatewayLatencyService { @Autowired private TransportService transportService; - private Map states = new ConcurrentHashMap<>(); + private Map states = new ConcurrentHashMap<>(); @PostConstruct private void init() { - scheduler.scheduleAtFixedRate(this::reportLatency, latencyReportIntervalSec, latencyReportIntervalSec, TimeUnit.SECONDS); + scheduler.scheduleAtFixedRate(this::reportMetrics, metricsReportIntervalSec, metricsReportIntervalSec, TimeUnit.SECONDS); } - public void process(TransportProtos.SessionInfoProto sessionInfo, DeviceId gatewayId, Map data, long ts) { - states.computeIfAbsent(gatewayId, k -> new GatewayLatencyState(sessionInfo)).update(ts, data); + public void process(TransportProtos.SessionInfoProto sessionInfo, DeviceId gatewayId, Map data, long ts) { + states.computeIfAbsent(gatewayId, k -> new GatewayMetricsState(sessionInfo)).update(ts, data); } public void onDeviceUpdate(TransportProtos.SessionInfoProto sessionInfo, DeviceId gatewayId) { @@ -71,34 +73,34 @@ public class GatewayLatencyService { } public void onDeviceDisconnect(DeviceId deviceId) { - GatewayLatencyState state = states.remove(deviceId); + GatewayMetricsState state = states.remove(deviceId); if (state != null) { - reportLatency(state, System.currentTimeMillis()); + reportMetrics(state, System.currentTimeMillis()); } } - public void reportLatency() { + public void reportMetrics() { if (states.isEmpty()) { return; } - Map oldStates = states; + Map oldStates = states; states = new ConcurrentHashMap<>(); long ts = System.currentTimeMillis(); oldStates.forEach((gatewayId, state) -> { - reportLatency(state, ts); + reportMetrics(state, ts); }); oldStates.clear(); } - private void reportLatency(GatewayLatencyState state, long ts) { + private void reportMetrics(GatewayMetricsState state, long ts) { if (state.isEmpty()) { return; } - var result = state.getLatencyStateResult(); + var result = state.getStateResult(); var kvProto = TransportProtos.KeyValueProto.newBuilder() - .setKey("latencyCheck") + .setKey(METRICS_CHECK) .setType(TransportProtos.KeyValueType.JSON_V) .setJsonV(JacksonUtil.toString(result)) .build(); diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/latency/GatewayLatencyData.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/metrics/GatewayMetricsData.java similarity index 83% rename from common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/latency/GatewayLatencyData.java rename to common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/metrics/GatewayMetricsData.java index 2cae428a02..23cc730b86 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/latency/GatewayLatencyData.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/metrics/GatewayMetricsData.java @@ -13,7 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.thingsboard.server.transport.mqtt.gateway.latency; +package org.thingsboard.server.transport.mqtt.gateway.metrics; -public record GatewayLatencyData(long receivedTs, long publishedTs) { +public record GatewayMetricsData(long receivedTs, long publishedTs) { } diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/latency/GatewayLatencyState.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/metrics/GatewayMetricsState.java similarity index 71% rename from common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/latency/GatewayLatencyState.java rename to common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/metrics/GatewayMetricsState.java index 3ad75fee75..45f945151b 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/latency/GatewayLatencyState.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/metrics/GatewayMetricsState.java @@ -13,7 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.thingsboard.server.transport.mqtt.gateway.latency; +package org.thingsboard.server.transport.mqtt.gateway.metrics; import lombok.Getter; import org.thingsboard.server.gen.transport.TransportProtos; @@ -25,15 +25,15 @@ import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; -public class GatewayLatencyState { +public class GatewayMetricsState { - private final Map connectors; + private final Map connectors; private final Lock updateLock; @Getter private volatile TransportProtos.SessionInfoProto sessionInfo; - public GatewayLatencyState(TransportProtos.SessionInfoProto sessionInfo) { + public GatewayMetricsState(TransportProtos.SessionInfoProto sessionInfo) { this.connectors = new HashMap<>(); this.updateLock = new ReentrantLock(); this.sessionInfo = sessionInfo; @@ -43,19 +43,19 @@ public class GatewayLatencyState { this.sessionInfo = sessionInfo; } - public void update(long ts, Map latencyData) { + public void update(long ts, Map metricsData) { updateLock.lock(); try { - latencyData.forEach((connectorName, data) -> { - connectors.computeIfAbsent(connectorName, k -> new ConnectorLatencyState()).update(ts, data); + metricsData.forEach((connectorName, data) -> { + connectors.computeIfAbsent(connectorName, k -> new ConnectorMetricsState()).update(ts, data); }); } finally { updateLock.unlock(); } } - public Map getLatencyStateResult() { - Map result = new HashMap<>(); + public Map getStateResult() { + Map result = new HashMap<>(); updateLock.lock(); try { connectors.forEach((name, state) -> result.put(name, state.getResult())); @@ -71,7 +71,7 @@ public class GatewayLatencyState { return connectors.isEmpty(); } - private static class ConnectorLatencyState { + private static class ConnectorMetricsState { private final AtomicInteger count; private final AtomicLong gwLatencySum; private final AtomicLong transportLatencySum; @@ -80,15 +80,15 @@ public class GatewayLatencyState { private volatile long minTransportLatency; private volatile long maxTransportLatency; - private ConnectorLatencyState() { + private ConnectorMetricsState() { this.count = new AtomicInteger(0); this.gwLatencySum = new AtomicLong(0); this.transportLatencySum = new AtomicLong(0); } - private void update(long serverReceiveTs, GatewayLatencyData latencyData) { - long gwLatency = latencyData.publishedTs() - latencyData.receivedTs(); - long transportLatency = serverReceiveTs - latencyData.publishedTs(); + private void update(long serverReceiveTs, GatewayMetricsData metricsData) { + long gwLatency = metricsData.publishedTs() - metricsData.receivedTs(); + long transportLatency = serverReceiveTs - metricsData.publishedTs(); count.incrementAndGet(); gwLatencySum.addAndGet(gwLatency); transportLatencySum.addAndGet(transportLatency); @@ -106,16 +106,16 @@ public class GatewayLatencyState { } } - private ConnectorLatencyResult getResult() { + private ConnectorMetricsResult getResult() { long count = this.count.get(); long avgGwLatency = gwLatencySum.get() / count; - long transportLatencyAvg = transportLatencySum.get() / count; - return new ConnectorLatencyResult(avgGwLatency, minGwLatency, maxGwLatency, transportLatencyAvg, minTransportLatency, maxTransportLatency); + long avgTransportLatency = transportLatencySum.get() / count; + return new ConnectorMetricsResult(avgGwLatency, minGwLatency, maxGwLatency, avgTransportLatency, minTransportLatency, maxTransportLatency); } } - public record ConnectorLatencyResult(long avgGwLatency, long minGwLatency, long maxGwLatency, - long transportLatencyAvg, long minTransportLatency, long maxTransportLatency) { + public record ConnectorMetricsResult(long avgGwLatency, long minGwLatency, long maxGwLatency, + long avgTransportLatency, long minTransportLatency, long maxTransportLatency) { } } diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewaySessionHandler.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewaySessionHandler.java index a7634593ce..32eaf6e0d7 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewaySessionHandler.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewaySessionHandler.java @@ -27,7 +27,7 @@ import org.thingsboard.server.common.data.DeviceProfile; import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.transport.auth.GetOrCreateDeviceFromGatewayResponse; import org.thingsboard.server.gen.transport.TransportProtos; -import org.thingsboard.server.transport.mqtt.gateway.GatewayLatencyService; +import org.thingsboard.server.transport.mqtt.gateway.GatewayMetricsService; import java.util.Optional; import java.util.UUID; @@ -40,11 +40,11 @@ import static com.amazonaws.util.StringUtils.UTF8; @Slf4j public class GatewaySessionHandler extends AbstractGatewaySessionHandler { - private final GatewayLatencyService latencyService; + private final GatewayMetricsService gatewayMetricsService; public GatewaySessionHandler(DeviceSessionCtx deviceSessionCtx, UUID sessionId, boolean overwriteDevicesActivity) { super(deviceSessionCtx, sessionId, overwriteDevicesActivity); - this.latencyService = deviceSessionCtx.getContext().getGatewayLatencyService(); + this.gatewayMetricsService = deviceSessionCtx.getContext().getGatewayMetricsService(); } public void onDeviceConnect(MqttPublishMessage mqttMsg) throws AdaptorException { @@ -72,16 +72,16 @@ public class GatewaySessionHandler extends AbstractGatewaySessionHandler deviceProfileOpt) { this.onDeviceUpdate(sessionInfo, device, deviceProfileOpt); - latencyService.onDeviceUpdate(sessionInfo, gateway.getDeviceId()); + gatewayMetricsService.onDeviceUpdate(sessionInfo, gateway.getDeviceId()); } public void onGatewayDelete(DeviceId deviceId) { - latencyService.onDeviceDelete(deviceId); + gatewayMetricsService.onDeviceDelete(deviceId); } public void onGatewayLatency(MqttPublishMessage mqttMsg) throws AdaptorException { @@ -94,7 +94,7 @@ public class GatewaySessionHandler extends AbstractGatewaySessionHandler() {}), ts); + gatewayMetricsService.process(deviceSessionCtx.getSessionInfo(), gateway.getDeviceId(), JacksonUtil.fromString(payload, new TypeReference<>() {}), ts); ack(msgId, MqttReasonCodes.PubAck.SUCCESS); } catch (Throwable t) { ackOrClose(msgId); diff --git a/transport/mqtt/src/main/resources/tb-mqtt-transport.yml b/transport/mqtt/src/main/resources/tb-mqtt-transport.yml index 140839ccc3..d2ac105623 100644 --- a/transport/mqtt/src/main/resources/tb-mqtt-transport.yml +++ b/transport/mqtt/src/main/resources/tb-mqtt-transport.yml @@ -146,6 +146,8 @@ transport: # MQTT disconnect timeout in milliseconds. The time to wait for the client to disconnect after the server sends a disconnect message. disconnect_timeout: "${MQTT_DISCONNECT_TIMEOUT:1000}" msg_queue_size_per_device_limit: "${MQTT_MSG_QUEUE_SIZE_PER_DEVICE_LIMIT:100}" # messages await in the queue before device connected state. This limit works on low level before TenantProfileLimits mechanism + # Interval of periodic report of the gateway metrics + gateway_metrics_report_interval_sec: "${MQTT_GATEWAY_METRICS_REPORT_INTERVAL_SEC:3600}" netty: # Netty leak detector level leak_detector_level: "${NETTY_LEAK_DETECTOR_LVL:DISABLED}" From e65fc9d55e881f5852fff1e6032d72926584a97b Mon Sep 17 00:00:00 2001 From: YevhenBondarenko Date: Thu, 12 Sep 2024 15:16:32 +0200 Subject: [PATCH 07/19] minor refactoring due to comments --- .../AbstractMqttTimeseriesIntegrationTest.java | 10 +++++----- .../server/transport/mqtt/MqttTransportHandler.java | 2 +- .../transport/mqtt/gateway/GatewayMetricsService.java | 4 ++-- .../transport/mqtt/session/GatewaySessionHandler.java | 2 +- 4 files changed, 9 insertions(+), 9 deletions(-) diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/telemetry/timeseries/AbstractMqttTimeseriesIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/telemetry/timeseries/AbstractMqttTimeseriesIntegrationTest.java index 4f89a8018e..32854a8735 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/telemetry/timeseries/AbstractMqttTimeseriesIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/telemetry/timeseries/AbstractMqttTimeseriesIntegrationTest.java @@ -57,7 +57,7 @@ import static org.thingsboard.server.common.data.device.profile.MqttTopics.DEVIC import static org.thingsboard.server.common.data.device.profile.MqttTopics.GATEWAY_CONNECT_TOPIC; import static org.thingsboard.server.common.data.device.profile.MqttTopics.GATEWAY_METRICS_TOPIC; import static org.thingsboard.server.common.data.device.profile.MqttTopics.GATEWAY_TELEMETRY_TOPIC; -import static org.thingsboard.server.transport.mqtt.gateway.GatewayMetricsService.METRICS_CHECK; +import static org.thingsboard.server.transport.mqtt.gateway.GatewayMetricsService.GATEWAY_METRICS; @Slf4j public abstract class AbstractMqttTimeseriesIntegrationTest extends AbstractMqttIntegrationTest { @@ -148,13 +148,13 @@ public abstract class AbstractMqttTimeseriesIntegrationTest extends AbstractMqtt gatewayMetricsService.reportMetrics(); - List actualKeys = getActualKeysList(savedGateway.getId(), List.of(METRICS_CHECK)); - assertEquals(METRICS_CHECK, actualKeys.get(0)); + List actualKeys = getActualKeysList(savedGateway.getId(), List.of(GATEWAY_METRICS)); + assertEquals(GATEWAY_METRICS, actualKeys.get(0)); - String telemetryUrl = String.format("/api/plugins/telemetry/DEVICE/%s/values/timeseries?startTs=%d&endTs=%d&keys=%s", savedGateway.getId(), 0, System.currentTimeMillis(), METRICS_CHECK); + String telemetryUrl = String.format("/api/plugins/telemetry/DEVICE/%s/values/timeseries?startTs=%d&endTs=%d&keys=%s", savedGateway.getId(), 0, System.currentTimeMillis(), GATEWAY_METRICS); Map>> gatewayTelemetry = doGetAsyncTyped(telemetryUrl, new TypeReference<>() {}); - Map latencyCheckTelemetry = gatewayTelemetry.get(METRICS_CHECK).get(0); + Map latencyCheckTelemetry = gatewayTelemetry.get(GATEWAY_METRICS).get(0); Map latencyCheckValue = JacksonUtil.fromString((String) latencyCheckTelemetry.get("value"), new TypeReference<>() {}); assertNotNull(latencyCheckValue); diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java index 1eee03f722..7fad363c6b 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java @@ -424,7 +424,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement gatewaySessionHandler.onDeviceDisconnect(mqttMsg); break; case MqttTopics.GATEWAY_METRICS_TOPIC: - gatewaySessionHandler.onGatewayLatency(mqttMsg); + gatewaySessionHandler.onGatewayMetrics(mqttMsg); break; default: ack(ctx, msgId, MqttReasonCodes.PubAck.TOPIC_NAME_INVALID); diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/GatewayMetricsService.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/GatewayMetricsService.java index 263888f4c2..c044c89e15 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/GatewayMetricsService.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/GatewayMetricsService.java @@ -39,7 +39,7 @@ import java.util.concurrent.TimeUnit; @TbMqttTransportComponent public class GatewayMetricsService { - public static final String METRICS_CHECK = "metricsCheck"; + public static final String GATEWAY_METRICS = "gatewayMetrics"; @Value("${transport.mqtt.gateway_metrics_report_interval_sec:3600}") private int metricsReportIntervalSec; @@ -100,7 +100,7 @@ public class GatewayMetricsService { } var result = state.getStateResult(); var kvProto = TransportProtos.KeyValueProto.newBuilder() - .setKey(METRICS_CHECK) + .setKey(GATEWAY_METRICS) .setType(TransportProtos.KeyValueType.JSON_V) .setJsonV(JacksonUtil.toString(result)) .build(); diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewaySessionHandler.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewaySessionHandler.java index 32eaf6e0d7..9444787193 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewaySessionHandler.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewaySessionHandler.java @@ -84,7 +84,7 @@ public class GatewaySessionHandler extends AbstractGatewaySessionHandler Date: Thu, 12 Sep 2024 22:28:10 +0200 Subject: [PATCH 08/19] implemented telemetry metadata instead of separate topic for metrics --- .../src/main/resources/thingsboard.yml | 2 +- ...AbstractMqttTimeseriesIntegrationTest.java | 75 ++++++++++--------- .../data/device/profile/MqttTopics.java | 1 - .../server/common/adaptor/JsonConverter.java | 20 ++++- .../transport/mqtt/MqttTransportHandler.java | 5 +- .../mqtt/gateway/GatewayMetricsService.java | 19 ++--- .../gateway/metrics/GatewayMetricsData.java | 2 +- .../gateway/metrics/GatewayMetricsState.java | 9 ++- .../AbstractGatewaySessionHandler.java | 22 +++++- .../mqtt/session/GatewaySessionHandler.java | 31 -------- .../src/main/resources/tb-mqtt-transport.yml | 2 +- 11 files changed, 92 insertions(+), 96 deletions(-) diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index 6b61838918..718c4823c4 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -1021,7 +1021,7 @@ transport: disconnect_timeout: "${MQTT_DISCONNECT_TIMEOUT:1000}" msg_queue_size_per_device_limit: "${MQTT_MSG_QUEUE_SIZE_PER_DEVICE_LIMIT:100}" # messages await in the queue before the device connected state. This limit works on the low level before TenantProfileLimits mechanism # Interval of periodic report of the gateway metrics - gateway_metrics_report_interval_sec: "${MQTT_GATEWAY_METRICS_REPORT_INTERVAL_SEC:3600}" + gateway_metrics_report_interval_sec: "${MQTT_GATEWAY_METRICS_REPORT_INTERVAL_SEC:60}" netty: # Netty leak detector level leak_detector_level: "${NETTY_LEAK_DETECTOR_LVL:DISABLED}" diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/telemetry/timeseries/AbstractMqttTimeseriesIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/telemetry/timeseries/AbstractMqttTimeseriesIntegrationTest.java index 32854a8735..f5e21e9e18 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/telemetry/timeseries/AbstractMqttTimeseriesIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/telemetry/timeseries/AbstractMqttTimeseriesIntegrationTest.java @@ -18,7 +18,6 @@ package org.thingsboard.server.transport.mqtt.mqttv3.telemetry.timeseries; import com.fasterxml.jackson.core.type.TypeReference; import io.netty.handler.codec.mqtt.MqttQoS; import lombok.extern.slf4j.Slf4j; -import org.awaitility.Awaitility; import org.junit.Before; import org.junit.Test; import org.mockito.ArgumentCaptor; @@ -55,7 +54,6 @@ import static org.thingsboard.server.common.data.device.profile.MqttTopics.DEVIC import static org.thingsboard.server.common.data.device.profile.MqttTopics.DEVICE_TELEMETRY_SHORT_TOPIC; import static org.thingsboard.server.common.data.device.profile.MqttTopics.DEVICE_TELEMETRY_TOPIC; import static org.thingsboard.server.common.data.device.profile.MqttTopics.GATEWAY_CONNECT_TOPIC; -import static org.thingsboard.server.common.data.device.profile.MqttTopics.GATEWAY_METRICS_TOPIC; import static org.thingsboard.server.common.data.device.profile.MqttTopics.GATEWAY_TELEMETRY_TOPIC; import static org.thingsboard.server.transport.mqtt.gateway.GatewayMetricsService.GATEWAY_METRICS; @@ -124,8 +122,8 @@ public abstract class AbstractMqttTimeseriesIntegrationTest extends AbstractMqtt String deviceName = "Device A"; Device device = doExecuteWithRetriesAndInterval(() -> doGet("/api/tenant/devices?deviceName=" + deviceName, Device.class), - 20, - 100); + 20, + 100); assertNotNull(device); client.disconnect(); @@ -138,13 +136,10 @@ public abstract class AbstractMqttTimeseriesIntegrationTest extends AbstractMqtt .build(); processBeforeTest(configProperties); - MqttTestClient client = new MqttTestClient(); - client.connectAndWait(gatewayAccessToken); - Map> gwLatencies = new HashMap<>(); Map> transportLatencies = new HashMap<>(); - publishLatency(client, gwLatencies, transportLatencies, 5); + publishLatency(gwLatencies, transportLatencies, 5); gatewayMetricsService.reportMetrics(); @@ -174,38 +169,37 @@ public abstract class AbstractMqttTimeseriesIntegrationTest extends AbstractMqtt assertNotNull(connectorLatencyResult); checkConnectorLatencyResult(connectorLatencyResult, avgGwLatency, minGwLatency, maxGwLatency, avgTransportLatency, minTransportLatency, maxTransportLatency); }); - - client.disconnect(); - - Awaitility.await() - .atMost(5, TimeUnit.SECONDS) - .untilAsserted(() -> verify(gatewayMetricsService).onDeviceDisconnect(savedGateway.getId())); } - private void publishLatency(MqttTestClient client, Map> gwLatencies, Map> transportLatencies, int n) throws Exception { + private void publishLatency(Map> gwLatencies, Map> transportLatencies, int n) throws Exception { Random random = new Random(); for (int i = 0; i < n; i++) { - Map data = new HashMap<>(); long publishedTs = System.currentTimeMillis() - 10; long gatewayLatencyA = random.nextLong(100, 500); - data.put("connectorA", new GatewayMetricsData(publishedTs - gatewayLatencyA, publishedTs)); + var firstData = new GatewayMetricsData("connectorA", publishedTs - gatewayLatencyA, publishedTs); gwLatencies.computeIfAbsent("connectorA", key -> new ArrayList<>()).add(gatewayLatencyA); - boolean sendB = i % 2 == 0; - if (sendB) { - long gatewayLatencyB = random.nextLong(120, 450); - data.put("connectorB", new GatewayMetricsData(publishedTs - gatewayLatencyB, publishedTs)); - gwLatencies.computeIfAbsent("connectorB", key -> new ArrayList<>()).add(gatewayLatencyB); - } - - client.publishAndWait(GATEWAY_METRICS_TOPIC, JacksonUtil.writeValueAsBytes(data)); - ArgumentCaptor transportReceiveTsCaptor = ArgumentCaptor.forClass(Long.class); - verify(gatewayMetricsService).process(any(), eq(savedGateway.getId()), eq(data), transportReceiveTsCaptor.capture()); - Long transportReceiveTs = transportReceiveTsCaptor.getValue(); - Long transportLatency = transportReceiveTs - publishedTs; - transportLatencies.computeIfAbsent("connectorA", key -> new ArrayList<>()).add(transportLatency); - if (sendB) { - transportLatencies.computeIfAbsent("connectorB", key -> new ArrayList<>()).add(transportLatency); - } + long gatewayLatencyB = random.nextLong(120, 450); + var secondData = new GatewayMetricsData("connectorB", publishedTs - gatewayLatencyB, publishedTs); + gwLatencies.computeIfAbsent("connectorB", key -> new ArrayList<>()).add(gatewayLatencyB); + + List expectedKeys = Arrays.asList("key1", "key2", "key3", "key4", "key5"); + String deviceName1 = "Device A"; + String deviceName2 = "Device B"; + String firstMetadata = JacksonUtil.writeValueAsString(firstData); + String secondMetadata = JacksonUtil.writeValueAsString(secondData); + String payload = getGatewayTelemetryJsonPayloadWithMetadata(deviceName1, deviceName2, "10000", "20000", firstMetadata, secondMetadata); + processGatewayTelemetryTest(GATEWAY_TELEMETRY_TOPIC, expectedKeys, payload.getBytes(), deviceName1, deviceName2); + + ArgumentCaptor transportReceiveTsCaptorA = ArgumentCaptor.forClass(Long.class); + ArgumentCaptor transportReceiveTsCaptorB = ArgumentCaptor.forClass(Long.class); + verify(gatewayMetricsService).process(any(), eq(savedGateway.getId()), eq(List.of(firstData)), transportReceiveTsCaptorA.capture()); + verify(gatewayMetricsService).process(any(), eq(savedGateway.getId()), eq(List.of(secondData)), transportReceiveTsCaptorB.capture()); + Long transportReceiveTsA = transportReceiveTsCaptorA.getValue(); + Long transportReceiveTsB = transportReceiveTsCaptorB.getValue(); + Long transportLatencyA = transportReceiveTsA - publishedTs; + Long transportLatencyB = transportReceiveTsB - publishedTs; + transportLatencies.computeIfAbsent("connectorA", key -> new ArrayList<>()).add(transportLatencyA); + transportLatencies.computeIfAbsent("connectorB", key -> new ArrayList<>()).add(transportLatencyB); } } @@ -250,7 +244,8 @@ public abstract class AbstractMqttTimeseriesIntegrationTest extends AbstractMqtt long end = System.currentTimeMillis() + 5000; Map>> values = null; while (start <= end) { - values = doGetAsyncTyped(getTelemetryValuesUrl, new TypeReference<>() {}); + values = doGetAsyncTyped(getTelemetryValuesUrl, new TypeReference<>() { + }); boolean valid = values.size() == expectedKeys.size(); if (valid) { for (String key : expectedKeys) { @@ -289,7 +284,7 @@ public abstract class AbstractMqttTimeseriesIntegrationTest extends AbstractMqtt MqttTestClient client = new MqttTestClient(); client.connectAndWait(gatewayAccessToken); client.publishAndWait(topic, payload); - client.disconnect(); + client.disconnectAndWait(); Device firstDevice = doExecuteWithRetriesAndInterval(() -> doGet("/api/tenant/devices?deviceName=" + firstDeviceName, Device.class), 20, @@ -346,6 +341,16 @@ public abstract class AbstractMqttTimeseriesIntegrationTest extends AbstractMqtt return "{\"" + deviceA + "\": " + payload + ", \"" + deviceB + "\": " + payload + "}"; } + protected String getGatewayTelemetryJsonPayloadWithMetadata(String deviceA, String deviceB, String firstTsValue, String secondTsValue, String firstMetadata, String secondMetadata) { + String payloadA = "[{\"ts\": " + firstTsValue + ", \"values\": " + PAYLOAD_VALUES_STR + ", \"metadata\":" + firstMetadata + "}, " + + "{\"ts\": " + secondTsValue + ", \"values\": " + PAYLOAD_VALUES_STR + "}]"; + + String payloadB = "[{\"ts\": " + firstTsValue + ", \"values\": " + PAYLOAD_VALUES_STR + ", \"metadata\":" + secondMetadata + "}, " + + "{\"ts\": " + secondTsValue + ", \"values\": " + PAYLOAD_VALUES_STR + "}]"; + + return "{\"" + deviceA + "\": " + payloadA + ", \"" + deviceB + "\": " + payloadB + "}"; + } + private String getTelemetryValuesUrl(DeviceId deviceId, Set actualKeySet) { return "/api/plugins/telemetry/DEVICE/" + deviceId + "/values/timeseries?startTs=0&endTs=25000&keys=" + String.join(",", actualKeySet); } diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/device/profile/MqttTopics.java b/common/data/src/main/java/org/thingsboard/server/common/data/device/profile/MqttTopics.java index 0590d57034..cabdea145c 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/device/profile/MqttTopics.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/device/profile/MqttTopics.java @@ -75,7 +75,6 @@ public class MqttTopics { public static final String GATEWAY_RPC_TOPIC = BASE_GATEWAY_API_TOPIC + RPC; public static final String GATEWAY_ATTRIBUTES_REQUEST_TOPIC = BASE_GATEWAY_API_TOPIC + ATTRIBUTES_REQUEST; public static final String GATEWAY_ATTRIBUTES_RESPONSE_TOPIC = BASE_GATEWAY_API_TOPIC + ATTRIBUTES_RESPONSE; - public static final String GATEWAY_METRICS_TOPIC = BASE_GATEWAY_API_TOPIC + "/metrics"; // v2 topics public static final String BASE_DEVICE_API_TOPIC_V2 = "v2"; public static final String REQUEST_ID_PATTERN = "(?\\d+)"; diff --git a/common/proto/src/main/java/org/thingsboard/server/common/adaptor/JsonConverter.java b/common/proto/src/main/java/org/thingsboard/server/common/adaptor/JsonConverter.java index f5c7bbb901..6a61531fec 100644 --- a/common/proto/src/main/java/org/thingsboard/server/common/adaptor/JsonConverter.java +++ b/common/proto/src/main/java/org/thingsboard/server/common/adaptor/JsonConverter.java @@ -74,22 +74,34 @@ public class JsonConverter { private static int maxStringValueLength = 0; public static PostTelemetryMsg convertToTelemetryProto(JsonElement jsonElement, long ts) throws JsonSyntaxException { + return convertToTelemetryProto(jsonElement, ts, null); + } + + public static PostTelemetryMsg convertToTelemetryProto(JsonElement jsonElement, long ts, List metadataResult) throws JsonSyntaxException { PostTelemetryMsg.Builder builder = PostTelemetryMsg.newBuilder(); - convertToTelemetry(jsonElement, ts, null, builder); + convertToTelemetry(jsonElement, ts, null, builder, metadataResult); return builder.build(); } + public static PostTelemetryMsg convertToTelemetryProto(JsonElement jsonElement, List metadataResult) throws JsonSyntaxException { + return convertToTelemetryProto(jsonElement, System.currentTimeMillis(), metadataResult); + } + public static PostTelemetryMsg convertToTelemetryProto(JsonElement jsonElement) throws JsonSyntaxException { return convertToTelemetryProto(jsonElement, System.currentTimeMillis()); } - private static void convertToTelemetry(JsonElement jsonElement, long systemTs, Map> result, PostTelemetryMsg.Builder builder) { + private static void convertToTelemetry(JsonElement jsonElement, long systemTs, Map> result, PostTelemetryMsg.Builder builder, List metadataResult) { if (jsonElement.isJsonObject()) { parseObject(systemTs, result, builder, jsonElement.getAsJsonObject()); } else if (jsonElement.isJsonArray()) { jsonElement.getAsJsonArray().forEach(je -> { if (je.isJsonObject()) { - parseObject(systemTs, result, builder, je.getAsJsonObject()); + JsonObject jo = je.getAsJsonObject(); + if (metadataResult != null && jo.has("metadata")) { + metadataResult.add(jo.get("metadata")); + } + parseObject(systemTs, result, builder, jo); } else { throw new JsonSyntaxException(CAN_T_PARSE_VALUE + je); } @@ -550,7 +562,7 @@ public class JsonConverter { public static Map> convertToTelemetry(JsonElement jsonElement, long systemTs, boolean sorted) throws JsonSyntaxException { Map> result = sorted ? new TreeMap<>() : new HashMap<>(); - convertToTelemetry(jsonElement, systemTs, result, null); + convertToTelemetry(jsonElement, systemTs, result, null, null); return result; } diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java index 7fad363c6b..d7f64e58ee 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java @@ -423,9 +423,6 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement case MqttTopics.GATEWAY_DISCONNECT_TOPIC: gatewaySessionHandler.onDeviceDisconnect(mqttMsg); break; - case MqttTopics.GATEWAY_METRICS_TOPIC: - gatewaySessionHandler.onGatewayMetrics(mqttMsg); - break; default: ack(ctx, msgId, MqttReasonCodes.PubAck.TOPIC_NAME_INVALID); } @@ -1199,7 +1196,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement transportService.process(deviceSessionCtx.getSessionInfo(), SESSION_EVENT_MSG_CLOSED, null); transportService.deregisterSession(deviceSessionCtx.getSessionInfo()); if (gatewaySessionHandler != null) { - gatewaySessionHandler.onGatewayDisconnect(); + gatewaySessionHandler.onDevicesDisconnect(); } if (sparkplugSessionHandler != null) { // add Msg Telemetry node: key STATE type: String value: OFFLINE ts: sparkplugBProto.getTimestamp() diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/GatewayMetricsService.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/GatewayMetricsService.java index c044c89e15..18f0047288 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/GatewayMetricsService.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/GatewayMetricsService.java @@ -30,6 +30,7 @@ import org.thingsboard.server.transport.mqtt.TbMqttTransportComponent; import org.thingsboard.server.transport.mqtt.gateway.metrics.GatewayMetricsData; import org.thingsboard.server.transport.mqtt.gateway.metrics.GatewayMetricsState; +import java.util.List; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.TimeUnit; @@ -41,7 +42,7 @@ public class GatewayMetricsService { public static final String GATEWAY_METRICS = "gatewayMetrics"; - @Value("${transport.mqtt.gateway_metrics_report_interval_sec:3600}") + @Value("${transport.mqtt.gateway_metrics_report_interval_sec:60}") private int metricsReportIntervalSec; @Autowired @@ -57,8 +58,8 @@ public class GatewayMetricsService { scheduler.scheduleAtFixedRate(this::reportMetrics, metricsReportIntervalSec, metricsReportIntervalSec, TimeUnit.SECONDS); } - public void process(TransportProtos.SessionInfoProto sessionInfo, DeviceId gatewayId, Map data, long ts) { - states.computeIfAbsent(gatewayId, k -> new GatewayMetricsState(sessionInfo)).update(ts, data); + public void process(TransportProtos.SessionInfoProto sessionInfo, DeviceId gatewayId, List data, long serverReceiveTs) { + states.computeIfAbsent(gatewayId, k -> new GatewayMetricsState(sessionInfo)).update(data, serverReceiveTs); } public void onDeviceUpdate(TransportProtos.SessionInfoProto sessionInfo, DeviceId gatewayId) { @@ -72,26 +73,18 @@ public class GatewayMetricsService { states.remove(deviceId); } - public void onDeviceDisconnect(DeviceId deviceId) { - GatewayMetricsState state = states.remove(deviceId); - if (state != null) { - reportMetrics(state, System.currentTimeMillis()); - } - } - public void reportMetrics() { if (states.isEmpty()) { return; } - Map oldStates = states; + Map statesToReport = states; states = new ConcurrentHashMap<>(); long ts = System.currentTimeMillis(); - oldStates.forEach((gatewayId, state) -> { + statesToReport.forEach((gatewayId, state) -> { reportMetrics(state, ts); }); - oldStates.clear(); } private void reportMetrics(GatewayMetricsState state, long ts) { diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/metrics/GatewayMetricsData.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/metrics/GatewayMetricsData.java index 23cc730b86..9829a1d373 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/metrics/GatewayMetricsData.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/metrics/GatewayMetricsData.java @@ -15,5 +15,5 @@ */ package org.thingsboard.server.transport.mqtt.gateway.metrics; -public record GatewayMetricsData(long receivedTs, long publishedTs) { +public record GatewayMetricsData(String connector, long receivedTs, long publishedTs) { } diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/metrics/GatewayMetricsState.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/metrics/GatewayMetricsState.java index 45f945151b..7bec5da6a7 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/metrics/GatewayMetricsState.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/metrics/GatewayMetricsState.java @@ -19,6 +19,7 @@ import lombok.Getter; import org.thingsboard.server.gen.transport.TransportProtos; import java.util.HashMap; +import java.util.List; import java.util.Map; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; @@ -43,11 +44,11 @@ public class GatewayMetricsState { this.sessionInfo = sessionInfo; } - public void update(long ts, Map metricsData) { + public void update(List metricsData, long serverReceiveTs) { updateLock.lock(); try { - metricsData.forEach((connectorName, data) -> { - connectors.computeIfAbsent(connectorName, k -> new ConnectorMetricsState()).update(ts, data); + metricsData.forEach(data -> { + connectors.computeIfAbsent(data.connector(), k -> new ConnectorMetricsState()).update(data, serverReceiveTs); }); } finally { updateLock.unlock(); @@ -86,7 +87,7 @@ public class GatewayMetricsState { this.transportLatencySum = new AtomicLong(0); } - private void update(long serverReceiveTs, GatewayMetricsData metricsData) { + private void update(GatewayMetricsData metricsData, long serverReceiveTs) { long gwLatency = metricsData.publishedTs() - metricsData.receivedTs(); long transportLatency = serverReceiveTs - metricsData.publishedTs(); count.incrementAndGet(); diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java index 3b3aaef983..f1728a2eac 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java @@ -62,6 +62,8 @@ import org.thingsboard.server.transport.mqtt.MqttTransportHandler; import org.thingsboard.server.transport.mqtt.adaptors.JsonMqttAdaptor; import org.thingsboard.server.transport.mqtt.adaptors.MqttTransportAdaptor; import org.thingsboard.server.transport.mqtt.adaptors.ProtoMqttAdaptor; +import org.thingsboard.server.transport.mqtt.gateway.GatewayMetricsService; +import org.thingsboard.server.transport.mqtt.gateway.metrics.GatewayMetricsData; import org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugConnectionState; import java.util.ArrayList; @@ -114,6 +116,7 @@ public abstract class AbstractGatewaySessionHandler mqttQoSMap; protected final ChannelHandlerContext channel; protected final DeviceSessionCtx deviceSessionCtx; + protected final GatewayMetricsService gatewayMetricsService; @Getter @Setter @@ -131,6 +134,7 @@ public abstract class AbstractGatewaySessionHandler createWeakMap() { @@ -380,7 +384,9 @@ public abstract class AbstractGatewaySessionHandler metadata = new ArrayList<>(); + TransportProtos.PostTelemetryMsg postTelemetryMsg = JsonConverter.convertToTelemetryProto(msg.getAsJsonArray(), metadata); + processTelemetryMetadataMsg(metadata); transportService.process(deviceCtx.getSessionInfo(), postTelemetryMsg, getPubAckCallback(channel, deviceName, msgId, postTelemetryMsg)); } catch (Throwable e) { log.warn("[{}][{}][{}] Failed to convert telemetry: [{}]", gateway.getTenantId(), gateway.getDeviceId(), deviceName, msg, e); @@ -388,6 +394,20 @@ public abstract class AbstractGatewaySessionHandler metadata) { + var serverReceiveTs = System.currentTimeMillis(); + var metricsData = metadata.stream() + .filter(JsonElement::isJsonObject) + .map(je -> { + var jo = je.getAsJsonObject(); + var connector = jo.get("connector").getAsString(); + var receivedTs = jo.get("receivedTs").getAsLong(); + var publishedTs = jo.get("publishedTs").getAsLong(); + return new GatewayMetricsData(connector, receivedTs, publishedTs); + }).toList(); + gatewayMetricsService.process(deviceSessionCtx.getSessionInfo(), gateway.getDeviceId(), metricsData, serverReceiveTs); + } + protected void onDeviceTelemetryProto(int msgId, ByteBuf payload) throws AdaptorException { try { TransportApiProtos.GatewayTelemetryMsg telemetryMsgProto = TransportApiProtos.GatewayTelemetryMsg.parseFrom(getBytes(payload)); diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewaySessionHandler.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewaySessionHandler.java index 9444787193..5dce8e8893 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewaySessionHandler.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewaySessionHandler.java @@ -15,36 +15,27 @@ */ package org.thingsboard.server.transport.mqtt.session; -import com.fasterxml.jackson.core.type.TypeReference; import io.netty.buffer.ByteBuf; import io.netty.handler.codec.mqtt.MqttPublishMessage; -import io.netty.handler.codec.mqtt.MqttReasonCodes; import lombok.extern.slf4j.Slf4j; -import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.server.common.adaptor.AdaptorException; import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.DeviceProfile; import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.transport.auth.GetOrCreateDeviceFromGatewayResponse; import org.thingsboard.server.gen.transport.TransportProtos; -import org.thingsboard.server.transport.mqtt.gateway.GatewayMetricsService; import java.util.Optional; import java.util.UUID; -import static com.amazonaws.util.StringUtils.UTF8; - /** * Created by nickAS21 on 26.12.22 */ @Slf4j public class GatewaySessionHandler extends AbstractGatewaySessionHandler { - private final GatewayMetricsService gatewayMetricsService; - public GatewaySessionHandler(DeviceSessionCtx deviceSessionCtx, UUID sessionId, boolean overwriteDevicesActivity) { super(deviceSessionCtx, sessionId, overwriteDevicesActivity); - this.gatewayMetricsService = deviceSessionCtx.getContext().getGatewayMetricsService(); } public void onDeviceConnect(MqttPublishMessage mqttMsg) throws AdaptorException { @@ -70,11 +61,6 @@ public class GatewaySessionHandler extends AbstractGatewaySessionHandler deviceProfileOpt) { this.onDeviceUpdate(sessionInfo, device, deviceProfileOpt); gatewayMetricsService.onDeviceUpdate(sessionInfo, gateway.getDeviceId()); @@ -84,21 +70,4 @@ public class GatewaySessionHandler extends AbstractGatewaySessionHandler() {}), ts); - ack(msgId, MqttReasonCodes.PubAck.SUCCESS); - } catch (Throwable t) { - ackOrClose(msgId); - } - } - } diff --git a/transport/mqtt/src/main/resources/tb-mqtt-transport.yml b/transport/mqtt/src/main/resources/tb-mqtt-transport.yml index d2ac105623..c9fd10a99d 100644 --- a/transport/mqtt/src/main/resources/tb-mqtt-transport.yml +++ b/transport/mqtt/src/main/resources/tb-mqtt-transport.yml @@ -147,7 +147,7 @@ transport: disconnect_timeout: "${MQTT_DISCONNECT_TIMEOUT:1000}" msg_queue_size_per_device_limit: "${MQTT_MSG_QUEUE_SIZE_PER_DEVICE_LIMIT:100}" # messages await in the queue before device connected state. This limit works on low level before TenantProfileLimits mechanism # Interval of periodic report of the gateway metrics - gateway_metrics_report_interval_sec: "${MQTT_GATEWAY_METRICS_REPORT_INTERVAL_SEC:3600}" + gateway_metrics_report_interval_sec: "${MQTT_GATEWAY_METRICS_REPORT_INTERVAL_SEC:60}" netty: # Netty leak detector level leak_detector_level: "${NETTY_LEAK_DETECTOR_LVL:DISABLED}" From 55b6b8c5645e3fd9748ecae83f9b4800dc59baf3 Mon Sep 17 00:00:00 2001 From: YevhenBondarenko Date: Fri, 13 Sep 2024 10:02:06 +0200 Subject: [PATCH 09/19] added one more filtering for metrics data --- .../mqtt/session/AbstractGatewaySessionHandler.java | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java index f1728a2eac..f963a78135 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java @@ -398,14 +398,19 @@ public abstract class AbstractGatewaySessionHandler { - var jo = je.getAsJsonObject(); + .map(JsonElement::getAsJsonObject) + .filter(jo -> jo.has("connector") + && jo.has("receivedTs") + && jo.has("publishedTs")) + .map(jo -> { var connector = jo.get("connector").getAsString(); var receivedTs = jo.get("receivedTs").getAsLong(); var publishedTs = jo.get("publishedTs").getAsLong(); return new GatewayMetricsData(connector, receivedTs, publishedTs); }).toList(); - gatewayMetricsService.process(deviceSessionCtx.getSessionInfo(), gateway.getDeviceId(), metricsData, serverReceiveTs); + if (!metadata.isEmpty()) { + gatewayMetricsService.process(deviceSessionCtx.getSessionInfo(), gateway.getDeviceId(), metricsData, serverReceiveTs); + } } protected void onDeviceTelemetryProto(int msgId, ByteBuf payload) throws AdaptorException { From 4440e559feff9b2ea440eb13bc3da9b8e64e3120 Mon Sep 17 00:00:00 2001 From: YevhenBondarenko Date: Fri, 13 Sep 2024 10:50:21 +0200 Subject: [PATCH 10/19] minor fix --- .../transport/mqtt/session/AbstractGatewaySessionHandler.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java index f963a78135..483522f85c 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java @@ -408,7 +408,7 @@ public abstract class AbstractGatewaySessionHandler Date: Fri, 13 Sep 2024 15:30:37 +0200 Subject: [PATCH 11/19] created separate converter methow for gateway telemetry --- ...AbstractMqttTimeseriesIntegrationTest.java | 6 +- .../msg/gateway/metrics/GatewayMetadata.java} | 4 +- .../server/common/adaptor/JsonConverter.java | 62 ++++++++++++++----- .../mqtt/gateway/GatewayMetricsService.java | 4 +- .../gateway/metrics/GatewayMetricsState.java | 5 +- .../AbstractGatewaySessionHandler.java | 32 +++------- 6 files changed, 67 insertions(+), 46 deletions(-) rename common/{transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/metrics/GatewayMetricsData.java => message/src/main/java/org/thingsboard/server/common/msg/gateway/metrics/GatewayMetadata.java} (80%) diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/telemetry/timeseries/AbstractMqttTimeseriesIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/telemetry/timeseries/AbstractMqttTimeseriesIntegrationTest.java index f5e21e9e18..5fd68417dd 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/telemetry/timeseries/AbstractMqttTimeseriesIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/telemetry/timeseries/AbstractMqttTimeseriesIntegrationTest.java @@ -28,7 +28,7 @@ import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.transport.mqtt.AbstractMqttIntegrationTest; import org.thingsboard.server.transport.mqtt.MqttTestConfigProperties; import org.thingsboard.server.transport.mqtt.gateway.GatewayMetricsService; -import org.thingsboard.server.transport.mqtt.gateway.metrics.GatewayMetricsData; +import org.thingsboard.server.common.msg.gateway.metrics.GatewayMetadata; import org.thingsboard.server.transport.mqtt.gateway.metrics.GatewayMetricsState; import org.thingsboard.server.transport.mqtt.mqttv3.MqttTestCallback; import org.thingsboard.server.transport.mqtt.mqttv3.MqttTestClient; @@ -176,10 +176,10 @@ public abstract class AbstractMqttTimeseriesIntegrationTest extends AbstractMqtt for (int i = 0; i < n; i++) { long publishedTs = System.currentTimeMillis() - 10; long gatewayLatencyA = random.nextLong(100, 500); - var firstData = new GatewayMetricsData("connectorA", publishedTs - gatewayLatencyA, publishedTs); + var firstData = new GatewayMetadata("connectorA", publishedTs - gatewayLatencyA, publishedTs); gwLatencies.computeIfAbsent("connectorA", key -> new ArrayList<>()).add(gatewayLatencyA); long gatewayLatencyB = random.nextLong(120, 450); - var secondData = new GatewayMetricsData("connectorB", publishedTs - gatewayLatencyB, publishedTs); + var secondData = new GatewayMetadata("connectorB", publishedTs - gatewayLatencyB, publishedTs); gwLatencies.computeIfAbsent("connectorB", key -> new ArrayList<>()).add(gatewayLatencyB); List expectedKeys = Arrays.asList("key1", "key2", "key3", "key4", "key5"); diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/metrics/GatewayMetricsData.java b/common/message/src/main/java/org/thingsboard/server/common/msg/gateway/metrics/GatewayMetadata.java similarity index 80% rename from common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/metrics/GatewayMetricsData.java rename to common/message/src/main/java/org/thingsboard/server/common/msg/gateway/metrics/GatewayMetadata.java index 9829a1d373..cc905f9383 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/metrics/GatewayMetricsData.java +++ b/common/message/src/main/java/org/thingsboard/server/common/msg/gateway/metrics/GatewayMetadata.java @@ -13,7 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.thingsboard.server.transport.mqtt.gateway.metrics; +package org.thingsboard.server.common.msg.gateway.metrics; -public record GatewayMetricsData(String connector, long receivedTs, long publishedTs) { +public record GatewayMetadata(String connector, long receivedTs, long publishedTs) { } diff --git a/common/proto/src/main/java/org/thingsboard/server/common/adaptor/JsonConverter.java b/common/proto/src/main/java/org/thingsboard/server/common/adaptor/JsonConverter.java index 6a61531fec..bffe59fa83 100644 --- a/common/proto/src/main/java/org/thingsboard/server/common/adaptor/JsonConverter.java +++ b/common/proto/src/main/java/org/thingsboard/server/common/adaptor/JsonConverter.java @@ -34,6 +34,8 @@ import org.thingsboard.server.common.data.kv.JsonDataEntry; 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.util.TbPair; +import org.thingsboard.server.common.msg.gateway.metrics.GatewayMetadata; import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.gen.transport.TransportProtos.AttributeUpdateNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ClaimDeviceMsg; @@ -74,33 +76,65 @@ public class JsonConverter { private static int maxStringValueLength = 0; public static PostTelemetryMsg convertToTelemetryProto(JsonElement jsonElement, long ts) throws JsonSyntaxException { - return convertToTelemetryProto(jsonElement, ts, null); - } - - public static PostTelemetryMsg convertToTelemetryProto(JsonElement jsonElement, long ts, List metadataResult) throws JsonSyntaxException { PostTelemetryMsg.Builder builder = PostTelemetryMsg.newBuilder(); - convertToTelemetry(jsonElement, ts, null, builder, metadataResult); + convertToTelemetry(jsonElement, ts, null, builder); return builder.build(); } - public static PostTelemetryMsg convertToTelemetryProto(JsonElement jsonElement, List metadataResult) throws JsonSyntaxException { - return convertToTelemetryProto(jsonElement, System.currentTimeMillis(), metadataResult); - } - public static PostTelemetryMsg convertToTelemetryProto(JsonElement jsonElement) throws JsonSyntaxException { return convertToTelemetryProto(jsonElement, System.currentTimeMillis()); } - private static void convertToTelemetry(JsonElement jsonElement, long systemTs, Map> result, PostTelemetryMsg.Builder builder, List metadataResult) { + public static TbPair> convertToGatewayTelemetry(JsonElement jsonElement, long systemTs) { + List metadataResult = null; + PostTelemetryMsg.Builder builder = PostTelemetryMsg.newBuilder(); + if (jsonElement.isJsonArray()) { + var ja = jsonElement.getAsJsonArray(); + for (int i = 0; i < ja.size(); i++) { + var je = ja.get(i); + if (je.isJsonObject()) { + JsonObject jo = je.getAsJsonObject(); + JsonElement metadataElem = jo.remove("metadata"); + if (metadataElem != null) { + if (metadataResult == null) { + metadataResult = new ArrayList<>(); + } + if (metadataElem.isJsonObject()) { + JsonObject metadataObj = metadataElem.getAsJsonObject(); + var connector = getAndValidateMetadataElement(metadataObj, "connector").getAsString(); + var receivedTs = getAndValidateMetadataElement(metadataObj, "receivedTs").getAsLong(); + var publishedTs = getAndValidateMetadataElement(metadataObj, "publishedTs").getAsLong(); + metadataResult.add(new GatewayMetadata(connector, receivedTs, publishedTs)); + } else { + throw new JsonSyntaxException("Can't parse gateway metadata: " + metadataElem); + } + } + parseObject(systemTs, null, builder, jo); + } else { + throw new JsonSyntaxException(CAN_T_PARSE_VALUE + je); + } + } + } else { + throw new JsonSyntaxException(CAN_T_PARSE_VALUE + jsonElement); + } + return TbPair.of(builder.build(), metadataResult); + } + + private static JsonElement getAndValidateMetadataElement(JsonObject metadata, String elementName) { + var element = metadata.get(elementName); + if (element == null || element.isJsonNull()) { + throw new JsonSyntaxException(String.format("Can't parse gateway element in metadata: [%s][%s]", metadata, elementName)); + } + return element; + } + + private static void convertToTelemetry(JsonElement jsonElement, long systemTs, Map> result, PostTelemetryMsg.Builder builder) { if (jsonElement.isJsonObject()) { parseObject(systemTs, result, builder, jsonElement.getAsJsonObject()); } else if (jsonElement.isJsonArray()) { jsonElement.getAsJsonArray().forEach(je -> { if (je.isJsonObject()) { JsonObject jo = je.getAsJsonObject(); - if (metadataResult != null && jo.has("metadata")) { - metadataResult.add(jo.get("metadata")); - } parseObject(systemTs, result, builder, jo); } else { throw new JsonSyntaxException(CAN_T_PARSE_VALUE + je); @@ -562,7 +596,7 @@ public class JsonConverter { public static Map> convertToTelemetry(JsonElement jsonElement, long systemTs, boolean sorted) throws JsonSyntaxException { Map> result = sorted ? new TreeMap<>() : new HashMap<>(); - convertToTelemetry(jsonElement, systemTs, result, null, null); + convertToTelemetry(jsonElement, systemTs, result, null); return result; } diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/GatewayMetricsService.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/GatewayMetricsService.java index 18f0047288..4f67866d06 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/GatewayMetricsService.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/GatewayMetricsService.java @@ -27,7 +27,7 @@ import org.thingsboard.server.common.transport.TransportServiceCallback; import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.queue.scheduler.SchedulerComponent; import org.thingsboard.server.transport.mqtt.TbMqttTransportComponent; -import org.thingsboard.server.transport.mqtt.gateway.metrics.GatewayMetricsData; +import org.thingsboard.server.common.msg.gateway.metrics.GatewayMetadata; import org.thingsboard.server.transport.mqtt.gateway.metrics.GatewayMetricsState; import java.util.List; @@ -58,7 +58,7 @@ public class GatewayMetricsService { scheduler.scheduleAtFixedRate(this::reportMetrics, metricsReportIntervalSec, metricsReportIntervalSec, TimeUnit.SECONDS); } - public void process(TransportProtos.SessionInfoProto sessionInfo, DeviceId gatewayId, List data, long serverReceiveTs) { + public void process(TransportProtos.SessionInfoProto sessionInfo, DeviceId gatewayId, List data, long serverReceiveTs) { states.computeIfAbsent(gatewayId, k -> new GatewayMetricsState(sessionInfo)).update(data, serverReceiveTs); } diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/metrics/GatewayMetricsState.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/metrics/GatewayMetricsState.java index 7bec5da6a7..a800c78e28 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/metrics/GatewayMetricsState.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/metrics/GatewayMetricsState.java @@ -16,6 +16,7 @@ package org.thingsboard.server.transport.mqtt.gateway.metrics; import lombok.Getter; +import org.thingsboard.server.common.msg.gateway.metrics.GatewayMetadata; import org.thingsboard.server.gen.transport.TransportProtos; import java.util.HashMap; @@ -44,7 +45,7 @@ public class GatewayMetricsState { this.sessionInfo = sessionInfo; } - public void update(List metricsData, long serverReceiveTs) { + public void update(List metricsData, long serverReceiveTs) { updateLock.lock(); try { metricsData.forEach(data -> { @@ -87,7 +88,7 @@ public class GatewayMetricsState { this.transportLatencySum = new AtomicLong(0); } - private void update(GatewayMetricsData metricsData, long serverReceiveTs) { + private void update(GatewayMetadata metricsData, long serverReceiveTs) { long gwLatency = metricsData.publishedTs() - metricsData.receivedTs(); long transportLatency = serverReceiveTs - metricsData.publishedTs(); count.incrementAndGet(); diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java index 483522f85c..2e205ff840 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java @@ -48,6 +48,8 @@ import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.DeviceProfile; import org.thingsboard.server.common.data.StringUtils; import org.thingsboard.server.common.data.id.DeviceId; +import org.thingsboard.server.common.data.util.TbPair; +import org.thingsboard.server.common.msg.gateway.metrics.GatewayMetadata; import org.thingsboard.server.common.msg.tools.TbRateLimitsException; import org.thingsboard.server.common.transport.TransportService; import org.thingsboard.server.common.transport.TransportServiceCallback; @@ -63,7 +65,6 @@ import org.thingsboard.server.transport.mqtt.adaptors.JsonMqttAdaptor; import org.thingsboard.server.transport.mqtt.adaptors.MqttTransportAdaptor; import org.thingsboard.server.transport.mqtt.adaptors.ProtoMqttAdaptor; import org.thingsboard.server.transport.mqtt.gateway.GatewayMetricsService; -import org.thingsboard.server.transport.mqtt.gateway.metrics.GatewayMetricsData; import org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugConnectionState; import java.util.ArrayList; @@ -384,9 +385,13 @@ public abstract class AbstractGatewaySessionHandler metadata = new ArrayList<>(); - TransportProtos.PostTelemetryMsg postTelemetryMsg = JsonConverter.convertToTelemetryProto(msg.getAsJsonArray(), metadata); - processTelemetryMetadataMsg(metadata); + long systemTs = System.currentTimeMillis(); + TbPair> gatewayPayloadPair = JsonConverter.convertToGatewayTelemetry(msg.getAsJsonArray(), systemTs); + TransportProtos.PostTelemetryMsg postTelemetryMsg = gatewayPayloadPair.getFirst(); + List metadata = gatewayPayloadPair.getSecond(); + if (!CollectionUtils.isEmpty(metadata)) { + gatewayMetricsService.process(deviceSessionCtx.getSessionInfo(), gateway.getDeviceId(), metadata, systemTs); + } transportService.process(deviceCtx.getSessionInfo(), postTelemetryMsg, getPubAckCallback(channel, deviceName, msgId, postTelemetryMsg)); } catch (Throwable e) { log.warn("[{}][{}][{}] Failed to convert telemetry: [{}]", gateway.getTenantId(), gateway.getDeviceId(), deviceName, msg, e); @@ -394,25 +399,6 @@ public abstract class AbstractGatewaySessionHandler metadata) { - var serverReceiveTs = System.currentTimeMillis(); - var metricsData = metadata.stream() - .filter(JsonElement::isJsonObject) - .map(JsonElement::getAsJsonObject) - .filter(jo -> jo.has("connector") - && jo.has("receivedTs") - && jo.has("publishedTs")) - .map(jo -> { - var connector = jo.get("connector").getAsString(); - var receivedTs = jo.get("receivedTs").getAsLong(); - var publishedTs = jo.get("publishedTs").getAsLong(); - return new GatewayMetricsData(connector, receivedTs, publishedTs); - }).toList(); - if (!metricsData.isEmpty()) { - gatewayMetricsService.process(deviceSessionCtx.getSessionInfo(), gateway.getDeviceId(), metricsData, serverReceiveTs); - } - } - protected void onDeviceTelemetryProto(int msgId, ByteBuf payload) throws AdaptorException { try { TransportApiProtos.GatewayTelemetryMsg telemetryMsgProto = TransportApiProtos.GatewayTelemetryMsg.parseFrom(getBytes(payload)); From b886bfe4f700f9bf72faeaa2425385678634d682 Mon Sep 17 00:00:00 2001 From: YevhenBondarenko Date: Fri, 13 Sep 2024 15:43:16 +0200 Subject: [PATCH 12/19] refactoring --- .../org/thingsboard/server/common/adaptor/JsonConverter.java | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/common/proto/src/main/java/org/thingsboard/server/common/adaptor/JsonConverter.java b/common/proto/src/main/java/org/thingsboard/server/common/adaptor/JsonConverter.java index bffe59fa83..8dc021510f 100644 --- a/common/proto/src/main/java/org/thingsboard/server/common/adaptor/JsonConverter.java +++ b/common/proto/src/main/java/org/thingsboard/server/common/adaptor/JsonConverter.java @@ -134,8 +134,7 @@ public class JsonConverter { } else if (jsonElement.isJsonArray()) { jsonElement.getAsJsonArray().forEach(je -> { if (je.isJsonObject()) { - JsonObject jo = je.getAsJsonObject(); - parseObject(systemTs, result, builder, jo); + parseObject(systemTs, result, builder, je.getAsJsonObject()); } else { throw new JsonSyntaxException(CAN_T_PARSE_VALUE + je); } From 118b2b7e10182e28e731b9232e64ab4b3ddc0a5d Mon Sep 17 00:00:00 2001 From: IrynaMatveieva Date: Fri, 13 Sep 2024 17:17:23 +0300 Subject: [PATCH 13/19] added upgrade script for device profile node --- .../service/install/SqlDatabaseUpgradeService.java | 14 +++++++++++++- 1 file changed, 13 insertions(+), 1 deletion(-) diff --git a/application/src/main/java/org/thingsboard/server/service/install/SqlDatabaseUpgradeService.java b/application/src/main/java/org/thingsboard/server/service/install/SqlDatabaseUpgradeService.java index 2d5ac5a99b..bfe389d6ff 100644 --- a/application/src/main/java/org/thingsboard/server/service/install/SqlDatabaseUpgradeService.java +++ b/application/src/main/java/org/thingsboard/server/service/install/SqlDatabaseUpgradeService.java @@ -122,7 +122,19 @@ public class SqlDatabaseUpgradeService implements DatabaseEntitiesUpgradeService updateSchema("3.6.4", 3006004, "3.7.0", 3007000, null); break; case "3.7.0": - updateSchema("3.7.0", 3007000, "3.7.1", 3007001, null); + updateSchema("3.7.0", 3007000, "3.7.1", 3007001, connection -> { + try { + connection.createStatement().execute("UPDATE rule_node SET " + + "configuration = CASE " + + " WHEN (configuration::jsonb ->> 'persistAlarmRulesState')::boolean = false THEN jsonb_set(configuration::jsonb, '{fetchAlarmRulesStateOnStart}', 'false'::jsonb)::varchar " + + " ELSE configuration " + + "END, " + + "configuration_version = 1 " + + "WHERE type = 'org.thingsboard.rule.engine.profile.TbDeviceProfileNode' " + + "AND configuration_version < 1;"); + } catch (Exception e) { + } + }); break; default: throw new RuntimeException("Unable to upgrade SQL database, unsupported fromVersion: " + fromVersion); From 788b33007d8d6e7648c3aa6c3c963fb22d0976f9 Mon Sep 17 00:00:00 2001 From: YevhenBondarenko Date: Fri, 13 Sep 2024 16:26:26 +0200 Subject: [PATCH 14/19] fixed NPE in RuleNOdeActorMessageProcessor --- .../actors/ruleChain/RuleNodeActorMessageProcessor.java | 7 ++++--- .../java/org/thingsboard/server/actors/TbActorMailbox.java | 2 +- 2 files changed, 5 insertions(+), 4 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleNodeActorMessageProcessor.java b/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleNodeActorMessageProcessor.java index e7b0113dbe..46ffb6e894 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleNodeActorMessageProcessor.java +++ b/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleNodeActorMessageProcessor.java @@ -43,6 +43,7 @@ import org.thingsboard.server.gen.transport.TransportProtos; @Slf4j public class RuleNodeActorMessageProcessor extends ComponentMsgProcessor { + private static final String UNKNOWN_NAME = "Unknown"; private final String ruleChainName; private final TbApiUsageReportClient apiUsageClient; private final DefaultTbContext defaultCtx; @@ -57,7 +58,7 @@ public class RuleNodeActorMessageProcessor extends ComponentMsgProcessor msg.onTbActorStopped(stopReason)); normalPriorityMsgs.forEach(msg -> msg.onTbActorStopped(stopReason)); } catch (Throwable t) { - log.warn("[{}] Failed to destroy actor: {}", selfId, t); + log.warn("[{}] Failed to destroy actor: ", selfId, t); } }); } From 5be11f439be68dd17c38ba31e2310fef6aba1a14 Mon Sep 17 00:00:00 2001 From: Igor Kulikov Date: Fri, 13 Sep 2024 20:27:10 +0300 Subject: [PATCH 15/19] UI: Add css animations to SCADA symbol API. --- .../widget/lib/scada/scada-symbol.models.ts | 304 +++++++++++++++++- 1 file changed, 302 insertions(+), 2 deletions(-) diff --git a/ui-ngx/src/app/modules/home/components/widget/lib/scada/scada-symbol.models.ts b/ui-ngx/src/app/modules/home/components/widget/lib/scada/scada-symbol.models.ts index 837e3a44ee..0418665d69 100644 --- a/ui-ngx/src/app/modules/home/components/widget/lib/scada/scada-symbol.models.ts +++ b/ui-ngx/src/app/modules/home/components/widget/lib/scada/scada-symbol.models.ts @@ -15,7 +15,7 @@ /// import { ValueType } from '@shared/models/constants'; -import { Box, Element, Runner, SVG, Svg, Text, Timeline } from '@svgdotjs/svg.js'; +import { Box, Element, Runner, Style, SVG, Svg, Text, Timeline } from '@svgdotjs/svg.js'; import '@svgdotjs/svg.panzoom.js'; import { DataToValueType, @@ -58,6 +58,7 @@ export interface ScadaSymbolApi { enable: (element: Element | Element[]) => void; callAction: (event: Event, behaviorId: string, value?: any, observer?: Partial>) => void; setValue: (valueId: string, value: any) => void; + cssAnimate: CssScadaSymbolAnimations; } export interface ScadaSymbolContext { @@ -621,7 +622,8 @@ export class ScadaSymbolObject { disable: this.disableElement.bind(this), enable: this.enableElement.bind(this), callAction: this.callAction.bind(this), - setValue: this.setValue.bind(this) + setValue: this.setValue.bind(this), + cssAnimate: new CssScadaSymbolAnimations(this.svgShape) }, tags: {}, properties: {}, @@ -1008,3 +1010,301 @@ export class ScadaSymbolObject { } } } + +const scadaSymbolAnimationId = 'scadaSymbolAnimation'; + +class CssScadaSymbolAnimations { + constructor(private svgShape: Svg) {} + + public rotate(element: Element, rotation: number, duration = 1000, loop = false): CssScadaSymbolAnimation { + this.checkOldAnimation(element); + return this.setupAnimation(element, + new RotateCssScadaSymbolAnimation(this.svgShape, element, loop, rotation || 0)).duration(duration); + } + + public move(element: Element, deltaX: number, deltaY: number, duration = 1000, loop = false): CssScadaSymbolAnimation { + this.checkOldAnimation(element); + return this.setupAnimation(element, + new MoveCssScadaSymbolAnimation(this.svgShape, element, loop, deltaX, deltaY).duration(duration)); + } + + public attr(element: Element, attrName: string, value: any, duration = 1000, loop = false): CssScadaSymbolAnimation { + this.checkOldAnimation(element); + return this.setupAnimation(element, + new AttrsCssScadaSymbolAnimation(this.svgShape, element, loop, {[attrName]: value}).duration(duration)); + } + + public attrs(element: Element, attr: any, duration = 1000, loop = false): CssScadaSymbolAnimation { + this.checkOldAnimation(element); + return this.setupAnimation(element, + new AttrsCssScadaSymbolAnimation(this.svgShape, element, loop, attr).duration(duration)); + } + + public animation(element: Element): CssScadaSymbolAnimation | undefined { + return element.remember(scadaSymbolAnimationId); + } + + private checkOldAnimation(element: Element) { + const previousAnimation: CssScadaSymbolAnimation = element.remember(scadaSymbolAnimationId); + if (previousAnimation) { + previousAnimation.destroy(); + } + } + + private setupAnimation(element: Element, animation: CssScadaSymbolAnimation): CssScadaSymbolAnimation { + animation.init(); + element.remember(scadaSymbolAnimationId, animation); + return animation; + } +} + +interface ScadaSymbolAnimationKeyframe { + stop: string; + style: any; +} + +abstract class CssScadaSymbolAnimation { + + private _animationName: string; + private _animationStyle: Style; + + private _running = true; + private _speed = 1; + private _duration = 1000; + private _easing = 'linear'; + + protected constructor(protected svgShape: Svg, + protected element: Element, + protected loop: boolean) { + } + + public init() { + this.prepareAnimation(); + } + + public start() { + if (!this._running) { + this.updateAnimationStyle('animation-play-state', 'running'); + this._running = true; + } + } + + public pause() { + if (this._running) { + this.updateAnimationStyle('animation-play-state', 'paused'); + this._running = false; + } + } + + public running(): boolean { + return this._running; + } + + public destroy() { + if (this._animationStyle) { + this._animationStyle.remove(); + this.element.removeClass(this._animationName); + this._animationStyle = null; + this._animationName = null; + } + } + + public duration(duration: number): CssScadaSymbolAnimation { + this._duration = duration; + this.updateAnimationStyle(this.loop ? 'animation-duration' : 'transition-duration', + Math.round(this._duration / this._speed) + 'ms'); + return this; + } + + public speed(speed: number): CssScadaSymbolAnimation { + this._speed = speed; + this.updateAnimationStyle(this.loop ? 'animation-duration' : 'transition-duration', + Math.round(this._duration / this._speed) + 'ms'); + return this; + } + + public easing(easing: string): CssScadaSymbolAnimation { + this._easing = easing; + this.updateAnimationStyle(this.loop ? 'animation-timing-function' : 'transition-timing-function', this._easing); + return this; + } + + private prepareAnimation() { + this._animationName = 'animation_' + generateElementId(); + this._animationStyle = this.svgShape.style(); + let styles: any; + if (this.loop) { + styles = { + 'animation-name': this._animationName, + 'animation-duration': this._duration + 'ms', + 'animation-timing-function': this._easing, + 'animation-iteration-count': 'infinite', + 'animation-fill-mode': 'forwards', + 'animation-play-state': 'running', + ...this.animationStyles() + }; + } else { + styles = { + 'transition-property': this.transitionProperties(), + 'transition-duration': this._duration + 'ms', + 'transition-timing-function': this._easing, + ...this.animationStyles() + }; + } + this._animationStyle.rule('.' + this._animationName, styles); + + if (this.loop) { + const keyframes = this.animationKeyframes(); + let keyframesCss = `\n@keyframes ${this._animationName} {\n`; + for (const keyframe of keyframes) { + let keyframeCss = ` ${keyframe.stop} {\n`; + for (const i of Object.keys(keyframe.style)) { + keyframeCss += ' ' + i + ':' + keyframe.style[i] + ';\n'; + } + keyframeCss += ' }\n'; + keyframesCss += keyframeCss; + } + keyframesCss += '}'; + this._animationStyle.addText(keyframesCss); + } + this.element.addClass(this._animationName); + if (!this.loop) { + this.doTransform(); + } + } + + private updateAnimationStyle(attrName: string, value: any) { + if (this._animationStyle) { + const styleText = this._animationStyle.node.innerHTML; + const attrValueRegex = new RegExp(`${attrName}:([^;]+);`); + this._animationStyle.node.innerHTML = styleText.replace(attrValueRegex, `${attrName}:${value};`); + } + } + + protected animationStyles(): any { + return {}; + } + + protected abstract animationKeyframes(): ScadaSymbolAnimationKeyframe[]; + + protected abstract transitionProperties(): string; + + protected doTransform() { + } + +} + +class RotateCssScadaSymbolAnimation extends CssScadaSymbolAnimation { + + constructor(protected svgShape: Svg, + protected element: Element, + protected loop: boolean, + private rotation: number) { + super(svgShape, element, loop); + } + + protected animationStyles(): any { + return { + 'transform-origin': `${this.element.cx()}px ${this.element.cy()}px` + }; + } + + protected animationKeyframes(): ScadaSymbolAnimationKeyframe[] { + const transform = this.element.transform(); + return [ + { + stop: '0%', + style: { + transform: `translate(${transform.translateX}px, ${transform.translateY}px) rotate(${transform.rotate}deg)` + } + }, + { + stop: '100%', + style: { + transform: `translate(${transform.translateX}px, ${transform.translateY}px) rotate(${this.rotation}deg)` + } + } + ]; + } + + protected transitionProperties(): string { + return 'transform'; + } + + protected doTransform() { + const transform = this.element.transform(); + this.element.attr({transform: `translate(${transform.translateX} ${transform.translateY}) rotate(${this.rotation})`}); + } + +} + +class MoveCssScadaSymbolAnimation extends CssScadaSymbolAnimation { + + constructor(protected svgShape: Svg, + protected element: Element, + protected loop: boolean, + private deltaX: number, + private deltaY: number) { + super(svgShape, element, loop); + } + + protected animationStyles(): any { + return { + 'transform-origin': `${this.element.cx()}px ${this.element.cy()}px` + }; + } + + protected animationKeyframes(): ScadaSymbolAnimationKeyframe[] { + const transform = this.element.transform(); + return [ + { + stop: '0%', + style: { + transform: `translate(${transform.translateX}px, ${transform.translateY}px)` + } + }, + { + stop: '100%', + style: { + transform: `translate(${transform.translateX+this.deltaX}px, ${transform.translateY+this.deltaY}px)` + } + } + ]; + } + + protected transitionProperties(): string { + return 'transform'; + } + + protected doTransform() { + this.element.relative(this.deltaX, this.deltaY); + } + +} + +class AttrsCssScadaSymbolAnimation extends CssScadaSymbolAnimation { + + constructor(protected svgShape: Svg, + protected element: Element, + protected loop: boolean, + private attr: any) { + super(svgShape, element, loop); + } + + protected animationStyles(): any { + return {}; + } + + protected animationKeyframes(): ScadaSymbolAnimationKeyframe[] { + return []; + } + + protected transitionProperties(): string { + return Object.keys(this.attr).join(' '); + } + + protected doTransform() { + this.element.attr(this.attr); + } + +} From 0c4e06a0f948c46eaa4d2014a09ad2accc5d2283 Mon Sep 17 00:00:00 2001 From: YevhenBondarenko Date: Mon, 16 Sep 2024 10:31:12 +0200 Subject: [PATCH 16/19] minor refactoring due to comments --- .../actors/ruleChain/RuleNodeActorMessageProcessor.java | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleNodeActorMessageProcessor.java b/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleNodeActorMessageProcessor.java index 46ffb6e894..75499b9a8d 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleNodeActorMessageProcessor.java +++ b/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleNodeActorMessageProcessor.java @@ -58,7 +58,7 @@ public class RuleNodeActorMessageProcessor extends ComponentMsgProcessor Date: Mon, 16 Sep 2024 11:53:12 +0200 Subject: [PATCH 17/19] fixed NPE in subscription service --- .../subscription/DefaultTbLocalSubscriptionService.java | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbLocalSubscriptionService.java b/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbLocalSubscriptionService.java index c713bece6b..4043b519da 100644 --- a/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbLocalSubscriptionService.java +++ b/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbLocalSubscriptionService.java @@ -334,7 +334,7 @@ public class DefaultTbLocalSubscriptionService implements TbLocalSubscriptionSer } private void onTimeSeriesUpdate(UUID entityId, List data, TbCallback callback) { - entityUpdates.get(entityId).timeSeriesUpdateTs = System.currentTimeMillis(); + getEntityUpdatesInfo(entityId).timeSeriesUpdateTs = System.currentTimeMillis(); processSubscriptionData(entityId, sub -> TbSubscriptionType.TIMESERIES.equals(sub.getType()), s -> { @@ -371,7 +371,7 @@ public class DefaultTbLocalSubscriptionService implements TbLocalSubscriptionSer } private void onAttributesUpdate(UUID entityId, String scope, List data, TbCallback callback) { - entityUpdates.get(entityId).attributesUpdateTs = System.currentTimeMillis(); + getEntityUpdatesInfo(entityId).attributesUpdateTs = System.currentTimeMillis(); processSubscriptionData(entityId, sub -> TbSubscriptionType.ATTRIBUTES.equals(sub.getType()), s -> { @@ -639,4 +639,8 @@ public class DefaultTbLocalSubscriptionService implements TbLocalSubscriptionSer throw new TbRateLimitsException(message); } + private TbEntityUpdatesInfo getEntityUpdatesInfo(UUID entityId) { + return entityUpdates.computeIfAbsent(entityId, id -> new TbEntityUpdatesInfo(0)); + } + } From af0ed296443a5341f732fd6589fef17f8d8cae26 Mon Sep 17 00:00:00 2001 From: IrynaMatveieva Date: Mon, 16 Sep 2024 14:30:00 +0300 Subject: [PATCH 18/19] fixed upgrade script --- .../server/service/install/SqlDatabaseUpgradeService.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/application/src/main/java/org/thingsboard/server/service/install/SqlDatabaseUpgradeService.java b/application/src/main/java/org/thingsboard/server/service/install/SqlDatabaseUpgradeService.java index bfe389d6ff..32833a63ba 100644 --- a/application/src/main/java/org/thingsboard/server/service/install/SqlDatabaseUpgradeService.java +++ b/application/src/main/java/org/thingsboard/server/service/install/SqlDatabaseUpgradeService.java @@ -126,7 +126,8 @@ public class SqlDatabaseUpgradeService implements DatabaseEntitiesUpgradeService try { connection.createStatement().execute("UPDATE rule_node SET " + "configuration = CASE " + - " WHEN (configuration::jsonb ->> 'persistAlarmRulesState')::boolean = false THEN jsonb_set(configuration::jsonb, '{fetchAlarmRulesStateOnStart}', 'false'::jsonb)::varchar " + + " WHEN (configuration::jsonb ->> 'persistAlarmRulesState') = 'false'" + + " THEN (configuration::jsonb || '{\"fetchAlarmRulesStateOnStart\": \"false\"}'::jsonb)::varchar " + " ELSE configuration " + "END, " + "configuration_version = 1 " + From 7e70e79f2fe3424c1ac4a63161c842973fdd8a30 Mon Sep 17 00:00:00 2001 From: IrynaMatveieva Date: Mon, 16 Sep 2024 15:39:06 +0300 Subject: [PATCH 19/19] added log --- .../server/service/install/SqlDatabaseUpgradeService.java | 1 + 1 file changed, 1 insertion(+) diff --git a/application/src/main/java/org/thingsboard/server/service/install/SqlDatabaseUpgradeService.java b/application/src/main/java/org/thingsboard/server/service/install/SqlDatabaseUpgradeService.java index 32833a63ba..b3d20ff5e9 100644 --- a/application/src/main/java/org/thingsboard/server/service/install/SqlDatabaseUpgradeService.java +++ b/application/src/main/java/org/thingsboard/server/service/install/SqlDatabaseUpgradeService.java @@ -134,6 +134,7 @@ public class SqlDatabaseUpgradeService implements DatabaseEntitiesUpgradeService "WHERE type = 'org.thingsboard.rule.engine.profile.TbDeviceProfileNode' " + "AND configuration_version < 1;"); } catch (Exception e) { + log.warn("Failed to execute update script for device profile rule nodes due to: ", e); } }); break;