diff --git a/application/src/main/data/upgrade/3.4.1/schema_update.sql b/application/src/main/data/upgrade/3.4.1/schema_update.sql index 5891246b44..8dceb0115d 100644 --- a/application/src/main/data/upgrade/3.4.1/schema_update.sql +++ b/application/src/main/data/upgrade/3.4.1/schema_update.sql @@ -14,6 +14,7 @@ -- limitations under the License. -- +-- AUDIT LOGS MIGRATION START DO $$ DECLARE table_partition RECORD; @@ -73,3 +74,64 @@ BEGIN WHERE created_time >= start_time_ms AND created_time < end_time_ms; END; $$; +-- AUDIT LOGS MIGRATION END + + +-- EDGE EVENTS MIGRATION START +DO +$$ + DECLARE table_partition RECORD; + BEGIN + -- in case of running the upgrade script a second time: + IF NOT (SELECT exists(SELECT FROM pg_tables WHERE tablename = 'old_edge_event')) THEN + ALTER TABLE edge_event RENAME TO old_edge_event; + ALTER INDEX IF EXISTS idx_edge_event_tenant_id_and_created_time RENAME TO idx_old_edge_event_tenant_id_and_created_time; + + FOR table_partition IN SELECT tablename AS name, split_part(tablename, '_', 3) AS partition_ts + FROM pg_tables WHERE tablename LIKE 'edge_event_%' + LOOP + EXECUTE format('ALTER TABLE %s RENAME TO old_edge_event_%s', table_partition.name, table_partition.partition_ts); + END LOOP; + ELSE + RAISE NOTICE 'Table old_edge_event already exists, leaving as is'; + END IF; +END; +$$; + +CREATE TABLE IF NOT EXISTS edge_event ( + id uuid NOT NULL, + created_time bigint NOT NULL, + edge_id uuid, + edge_event_type varchar(255), + edge_event_uid varchar(255), + entity_id uuid, + edge_event_action varchar(255), + body varchar(10000000), + tenant_id uuid, + ts bigint NOT NULL + ) PARTITION BY RANGE (created_time); +CREATE INDEX IF NOT EXISTS idx_edge_event_tenant_id_and_created_time ON edge_event(tenant_id, created_time DESC); + +CREATE OR REPLACE PROCEDURE migrate_edge_event(IN start_time_ms BIGINT, IN end_time_ms BIGINT, IN partition_size_ms BIGINT) + LANGUAGE plpgsql AS +$$ +DECLARE + p RECORD; + partition_end_ts BIGINT; +BEGIN + FOR p IN SELECT DISTINCT (created_time - created_time % partition_size_ms) AS partition_ts FROM old_edge_event + WHERE created_time >= start_time_ms AND created_time < end_time_ms + LOOP + partition_end_ts = p.partition_ts + partition_size_ms; + RAISE NOTICE '[edge_event] Partition to create : [%-%]', p.partition_ts, partition_end_ts; + EXECUTE format('CREATE TABLE IF NOT EXISTS edge_event_%s PARTITION OF edge_event ' || + 'FOR VALUES FROM ( %s ) TO ( %s )', p.partition_ts, p.partition_ts, partition_end_ts); + END LOOP; + + INSERT INTO edge_event + SELECT id, created_time, edge_id, edge_event_type, edge_event_uid, entity_id, edge_event_action, body, tenant_id, ts + FROM old_edge_event + WHERE created_time >= start_time_ms AND created_time < end_time_ms; +END; +$$; +-- EDGE EVENTS MIGRATION END diff --git a/application/src/main/data/upgrade/3.4.1/schema_update_before.sql b/application/src/main/data/upgrade/3.4.1/schema_update_before.sql index 59566e42b5..27f772aba3 100644 --- a/application/src/main/data/upgrade/3.4.1/schema_update_before.sql +++ b/application/src/main/data/upgrade/3.4.1/schema_update_before.sql @@ -37,9 +37,10 @@ CREATE OR REPLACE PROCEDURE update_asset_profiles() LANGUAGE plpgsql AS $$ BEGIN -UPDATE asset as a SET asset_profile_id = p.id - FROM - (SELECT id, tenant_id, name from asset_profile) as p -WHERE a.asset_profile_id IS NULL AND p.tenant_id = a.tenant_id AND a.type = p.name; + UPDATE asset a SET asset_profile_id = COALESCE( + (SELECT id from asset_profile p WHERE p.tenant_id = a.tenant_id AND a.type = p.name), + (SELECT id from asset_profile p WHERE p.tenant_id = a.tenant_id AND p.name = 'default') + ) + WHERE a.asset_profile_id IS NULL; END; $$; diff --git a/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java b/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java index 2ffe2002f9..8a365b4e56 100644 --- a/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java +++ b/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java @@ -823,6 +823,9 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor { body.put("expirationTime", msg.getExpirationTime()); body.put("method", msg.getBody().getMethod()); body.put("params", msg.getBody().getParams()); + body.put("persisted", msg.isPersisted()); + body.put("retries", msg.getRetries()); + body.put("additionalInfo", msg.getAdditionalInfo()); EdgeEvent edgeEvent = EdgeUtils.constructEdgeEvent(tenantId, edgeId, EdgeEventType.DEVICE, EdgeEventActionType.RPC_CALL, deviceId, body); diff --git a/application/src/main/java/org/thingsboard/server/controller/AssetController.java b/application/src/main/java/org/thingsboard/server/controller/AssetController.java index 8450c67ea3..968829b65a 100644 --- a/application/src/main/java/org/thingsboard/server/controller/AssetController.java +++ b/application/src/main/java/org/thingsboard/server/controller/AssetController.java @@ -38,7 +38,6 @@ import org.thingsboard.server.common.data.asset.Asset; import org.thingsboard.server.common.data.asset.AssetInfo; import org.thingsboard.server.common.data.asset.AssetSearchQuery; import org.thingsboard.server.common.data.edge.Edge; -import org.thingsboard.server.common.data.exception.ThingsboardErrorCode; import org.thingsboard.server.common.data.exception.ThingsboardException; import org.thingsboard.server.common.data.id.AssetId; import org.thingsboard.server.common.data.id.AssetProfileId; @@ -86,7 +85,6 @@ import static org.thingsboard.server.controller.ControllerConstants.TENANT_AUTHO import static org.thingsboard.server.controller.ControllerConstants.TENANT_OR_CUSTOMER_AUTHORITY_PARAGRAPH; import static org.thingsboard.server.controller.ControllerConstants.UUID_WIKI_LINK; import static org.thingsboard.server.controller.EdgeController.EDGE_ID; -import static org.thingsboard.server.dao.asset.BaseAssetService.TB_SERVICE_QUEUE; @RestController @TbCoreComponent @@ -148,9 +146,6 @@ public class AssetController extends BaseController { @RequestMapping(value = "/asset", method = RequestMethod.POST) @ResponseBody public Asset saveAsset(@ApiParam(value = "A JSON value representing the asset.") @RequestBody Asset asset) throws Exception { - if (TB_SERVICE_QUEUE.equals(asset.getType())) { - throw new ThingsboardException("Unable to save asset with type " + TB_SERVICE_QUEUE, ThingsboardErrorCode.BAD_REQUEST_PARAMS); - } asset.setTenantId(getTenantId()); checkEntity(asset.getId(), asset, Resource.ASSET); return tbAssetService.save(asset, getCurrentUser()); diff --git a/application/src/main/java/org/thingsboard/server/install/ThingsboardInstallService.java b/application/src/main/java/org/thingsboard/server/install/ThingsboardInstallService.java index 33a6d4bdd1..d49cf2dccb 100644 --- a/application/src/main/java/org/thingsboard/server/install/ThingsboardInstallService.java +++ b/application/src/main/java/org/thingsboard/server/install/ThingsboardInstallService.java @@ -26,6 +26,7 @@ import org.thingsboard.server.service.component.ComponentDiscoveryService; import org.thingsboard.server.service.install.DatabaseEntitiesUpgradeService; import org.thingsboard.server.service.install.DatabaseTsUpgradeService; import org.thingsboard.server.service.install.EntityDatabaseSchemaService; +import org.thingsboard.server.service.install.NoSqlKeyspaceService; import org.thingsboard.server.service.install.SystemDataLoaderService; import org.thingsboard.server.service.install.TsDatabaseSchemaService; import org.thingsboard.server.service.install.TsLatestDatabaseSchemaService; @@ -51,6 +52,9 @@ public class ThingsboardInstallService { @Autowired private EntityDatabaseSchemaService entityDatabaseSchemaService; + @Autowired(required = false) + private NoSqlKeyspaceService noSqlKeyspaceService; + @Autowired private TsDatabaseSchemaService tsDatabaseSchemaService; @@ -252,6 +256,10 @@ public class ThingsboardInstallService { log.info("Installing DataBase schema for timeseries..."); + if (noSqlKeyspaceService != null) { + noSqlKeyspaceService.createDatabaseSchema(); + } + tsDatabaseSchemaService.createDatabaseSchema(); if (tsLatestDatabaseSchemaService != null) { 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 2342997c65..957c699e32 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 @@ -611,7 +611,7 @@ public final class EdgeGrpcSession implements Closeable { } if (uplinkMsg.getDeviceRpcCallMsgCount() > 0) { for (DeviceRpcCallMsg deviceRpcCallMsg : uplinkMsg.getDeviceRpcCallMsgList()) { - result.add(ctx.getDeviceProcessor().processDeviceRpcCallResponseFromEdge(edge.getTenantId(), deviceRpcCallMsg)); + result.add(ctx.getDeviceProcessor().processDeviceRpcCallFromEdge(edge.getTenantId(), edge, deviceRpcCallMsg)); } } if (uplinkMsg.getWidgetBundleTypesRequestMsgCount() > 0) { diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/DeviceMsgConstructor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/DeviceMsgConstructor.java index 511910dd8a..522bcd3a4e 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/DeviceMsgConstructor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/DeviceMsgConstructor.java @@ -21,13 +21,13 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.server.common.data.Device; -import org.thingsboard.server.common.data.id.CustomerId; import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.security.DeviceCredentials; import org.thingsboard.server.gen.edge.v1.DeviceCredentialsUpdateMsg; import org.thingsboard.server.gen.edge.v1.DeviceRpcCallMsg; import org.thingsboard.server.gen.edge.v1.DeviceUpdateMsg; import org.thingsboard.server.gen.edge.v1.RpcRequestMsg; +import org.thingsboard.server.gen.edge.v1.RpcResponseMsg; import org.thingsboard.server.gen.edge.v1.UpdateMsgType; import org.thingsboard.server.queue.util.DataDecodingEncodingService; import org.thingsboard.server.queue.util.TbCoreComponent; @@ -97,25 +97,55 @@ public class DeviceMsgConstructor { } public DeviceRpcCallMsg constructDeviceRpcCallMsg(UUID deviceId, JsonNode body) { - int requestId = body.get("requestId").asInt(); - boolean oneway = body.get("oneway").asBoolean(); - UUID requestUUID = UUID.fromString(body.get("requestUUID").asText()); - long expirationTime = body.get("expirationTime").asLong(); - String method = body.get("method").asText(); - String params = body.get("params").asText(); + DeviceRpcCallMsg.Builder builder = constructDeviceRpcMsg(deviceId, body); + if (body.has("error") || body.has("response")) { + RpcResponseMsg.Builder responseBuilder = RpcResponseMsg.newBuilder(); + if (body.has("error")) { + responseBuilder.setError(body.get("error").asText()); + } else { + responseBuilder.setResponse(body.get("response").asText()); + } + builder.setResponseMsg(responseBuilder.build()); + } else { + RpcRequestMsg.Builder requestBuilder = RpcRequestMsg.newBuilder(); + requestBuilder.setMethod(body.get("method").asText()); + requestBuilder.setParams(body.get("params").asText()); + builder.setRequestMsg(requestBuilder.build()); + } + return builder.build(); + } - RpcRequestMsg.Builder requestBuilder = RpcRequestMsg.newBuilder(); - requestBuilder.setMethod(method); - requestBuilder.setParams(params); + private DeviceRpcCallMsg.Builder constructDeviceRpcMsg(UUID deviceId, JsonNode body) { DeviceRpcCallMsg.Builder builder = DeviceRpcCallMsg.newBuilder() .setDeviceIdMSB(deviceId.getMostSignificantBits()) .setDeviceIdLSB(deviceId.getLeastSignificantBits()) - .setRequestUuidMSB(requestUUID.getMostSignificantBits()) - .setRequestUuidLSB(requestUUID.getLeastSignificantBits()) - .setRequestId(requestId) - .setExpirationTime(expirationTime) - .setOneway(oneway) - .setRequestMsg(requestBuilder.build()); - return builder.build(); + .setRequestId(body.get("requestId").asInt()); + if (body.get("oneway") != null) { + builder.setOneway(body.get("oneway").asBoolean()); + } + if (body.get("requestUUID") != null) { + UUID requestUUID = UUID.fromString(body.get("requestUUID").asText()); + builder.setRequestUuidMSB(requestUUID.getMostSignificantBits()) + .setRequestUuidLSB(requestUUID.getLeastSignificantBits()); + } + if (body.get("expirationTime") != null) { + builder.setExpirationTime(body.get("expirationTime").asLong()); + } + if (body.get("persisted") != null) { + builder.setPersisted(body.get("persisted").asBoolean()); + } + if (body.get("retries") != null) { + builder.setRetries(body.get("retries").asInt()); + } + if (body.get("additionalInfo") != null) { + builder.setAdditionalInfo(JacksonUtil.toString(body.get("additionalInfo"))); + } + if (body.get("serviceId") != null) { + builder.setServiceId(body.get("serviceId").asText()); + } + if (body.get("sessionId") != null) { + builder.setSessionId(body.get("sessionId").asText()); + } + return builder; } } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/DeviceEdgeProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/DeviceEdgeProcessor.java index e09e036953..81e7970611 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/DeviceEdgeProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/DeviceEdgeProcessor.java @@ -52,6 +52,7 @@ import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.TbMsgDataType; import org.thingsboard.server.common.msg.TbMsgMetaData; import org.thingsboard.server.common.msg.rpc.FromDeviceRpcResponse; +import org.thingsboard.server.common.msg.session.SessionMsgType; import org.thingsboard.server.dao.exception.DataValidationException; import org.thingsboard.server.gen.edge.v1.DeviceCredentialsRequestMsg; import org.thingsboard.server.gen.edge.v1.DeviceCredentialsUpdateMsg; @@ -325,8 +326,17 @@ public class DeviceEdgeProcessor extends BaseEdgeProcessor { return metaData; } - public ListenableFuture processDeviceRpcCallResponseFromEdge(TenantId tenantId, DeviceRpcCallMsg deviceRpcCallMsg) { - log.trace("[{}] processDeviceRpcCallResponseMsg [{}]", tenantId, deviceRpcCallMsg); + public ListenableFuture processDeviceRpcCallFromEdge(TenantId tenantId, Edge edge, DeviceRpcCallMsg deviceRpcCallMsg) { + log.trace("[{}] processDeviceRpcCallFromEdge [{}]", tenantId, deviceRpcCallMsg); + if (deviceRpcCallMsg.hasResponseMsg()) { + return processDeviceRpcResponseFromEdge(tenantId, deviceRpcCallMsg); + } else if (deviceRpcCallMsg.hasRequestMsg()) { + return processDeviceRpcRequestFromEdge(tenantId, edge, deviceRpcCallMsg); + } + return Futures.immediateFuture(null); + } + + private ListenableFuture processDeviceRpcResponseFromEdge(TenantId tenantId, DeviceRpcCallMsg deviceRpcCallMsg) { SettableFuture futureToSet = SettableFuture.create(); UUID requestUuid = new UUID(deviceRpcCallMsg.getRequestUuidMSB(), deviceRpcCallMsg.getRequestUuidLSB()); DeviceId deviceId = new DeviceId(new UUID(deviceRpcCallMsg.getDeviceIdMSB(), deviceRpcCallMsg.getDeviceIdLSB())); @@ -357,6 +367,46 @@ public class DeviceEdgeProcessor extends BaseEdgeProcessor { return futureToSet; } + private ListenableFuture processDeviceRpcRequestFromEdge(TenantId tenantId, Edge edge, DeviceRpcCallMsg deviceRpcCallMsg) { + DeviceId deviceId = new DeviceId(new UUID(deviceRpcCallMsg.getDeviceIdMSB(), deviceRpcCallMsg.getDeviceIdLSB())); + try { + TbMsgMetaData metaData = new TbMsgMetaData(); + String requestId = Integer.toString(deviceRpcCallMsg.getRequestId()); + metaData.putValue("requestId", requestId); + metaData.putValue("serviceId", deviceRpcCallMsg.getServiceId()); + metaData.putValue("sessionId", deviceRpcCallMsg.getSessionId()); + metaData.putValue(DataConstants.EDGE_ID, edge.getId().toString()); + Device device = deviceService.findDeviceById(tenantId, deviceId); + if (device != null) { + metaData.putValue("deviceName", device.getName()); + metaData.putValue("deviceType", device.getType()); + metaData.putValue(DataConstants.DEVICE_ID, deviceId.getId().toString()); + } + ObjectNode data = JacksonUtil.OBJECT_MAPPER.createObjectNode(); + data.put("method", deviceRpcCallMsg.getRequestMsg().getMethod()); + data.put("params", deviceRpcCallMsg.getRequestMsg().getParams()); + TbMsg tbMsg = TbMsg.newMsg(SessionMsgType.TO_SERVER_RPC_REQUEST.name(), deviceId, null, metaData, + TbMsgDataType.JSON, JacksonUtil.OBJECT_MAPPER.writeValueAsString(data)); + tbClusterService.pushMsgToRuleEngine(tenantId, deviceId, tbMsg, new TbQueueCallback() { + @Override + public void onSuccess(TbQueueMsgMetadata metadata) { + log.debug("Successfully send TO_SERVER_RPC_REQUEST to rule engine [{}], deviceRpcCallMsg {}", + device, deviceRpcCallMsg); + } + + @Override + public void onFailure(Throwable t) { + log.debug("Failed to send TO_SERVER_RPC_REQUEST to rule engine [{}], deviceRpcCallMsg {}", + device, deviceRpcCallMsg, t); + } + }); + } catch (JsonProcessingException | IllegalArgumentException e) { + log.warn("[{}] Failed to push TO_SERVER_RPC_REQUEST to rule engine. deviceRpcCallMsg {}", deviceId, deviceRpcCallMsg, e); + } + + return Futures.immediateFuture(null); + } + public DownlinkMsg convertDeviceEventToDownlink(EdgeEvent edgeEvent) { DeviceId deviceId = new DeviceId(edgeEvent.getEntityId()); DownlinkMsg downlinkMsg = null; @@ -413,11 +463,9 @@ public class DeviceEdgeProcessor extends BaseEdgeProcessor { private DownlinkMsg convertRpcCallEventToDownlink(EdgeEvent edgeEvent) { log.trace("Executing convertRpcCallEventToDownlink, edgeEvent [{}]", edgeEvent); - DeviceRpcCallMsg deviceRpcCallMsg = - deviceMsgConstructor.constructDeviceRpcCallMsg(edgeEvent.getEntityId(), edgeEvent.getBody()); return DownlinkMsg.newBuilder() .setDownlinkMsgId(EdgeUtils.nextPositiveInt()) - .addDeviceRpcCallMsg(deviceRpcCallMsg) + .addDeviceRpcCallMsg(deviceMsgConstructor.constructDeviceRpcCallMsg(edgeEvent.getEntityId(), edgeEvent.getBody())) .build(); } diff --git a/application/src/main/java/org/thingsboard/server/service/entitiy/asset/DefaultTbAssetService.java b/application/src/main/java/org/thingsboard/server/service/entitiy/asset/DefaultTbAssetService.java index 118dfe3b58..abaef626b5 100644 --- a/application/src/main/java/org/thingsboard/server/service/entitiy/asset/DefaultTbAssetService.java +++ b/application/src/main/java/org/thingsboard/server/service/entitiy/asset/DefaultTbAssetService.java @@ -22,8 +22,10 @@ import org.thingsboard.server.common.data.Customer; import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.User; import org.thingsboard.server.common.data.asset.Asset; +import org.thingsboard.server.common.data.asset.AssetProfile; import org.thingsboard.server.common.data.audit.ActionType; import org.thingsboard.server.common.data.edge.Edge; +import org.thingsboard.server.common.data.exception.ThingsboardErrorCode; import org.thingsboard.server.common.data.exception.ThingsboardException; import org.thingsboard.server.common.data.id.AssetId; import org.thingsboard.server.common.data.id.CustomerId; @@ -31,20 +33,32 @@ import org.thingsboard.server.common.data.id.EdgeId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.dao.asset.AssetService; import org.thingsboard.server.service.entitiy.AbstractTbEntityService; +import org.thingsboard.server.service.profile.TbAssetProfileCache; import java.util.List; +import static org.thingsboard.server.dao.asset.BaseAssetService.TB_SERVICE_QUEUE; + @Service @AllArgsConstructor public class DefaultTbAssetService extends AbstractTbEntityService implements TbAssetService { private final AssetService assetService; + private final TbAssetProfileCache assetProfileCache; @Override public Asset save(Asset asset, User user) throws Exception { ActionType actionType = asset.getId() == null ? ActionType.ADDED : ActionType.UPDATED; TenantId tenantId = asset.getTenantId(); try { + if (TB_SERVICE_QUEUE.equals(asset.getType())) { + throw new ThingsboardException("Unable to save asset with type " + TB_SERVICE_QUEUE, ThingsboardErrorCode.BAD_REQUEST_PARAMS); + } else if (asset.getAssetProfileId() != null) { + AssetProfile assetProfile = assetProfileCache.get(tenantId, asset.getAssetProfileId()); + if (assetProfile != null && TB_SERVICE_QUEUE.equals(assetProfile.getName())) { + throw new ThingsboardException("Unable to save asset with profile " + TB_SERVICE_QUEUE, ThingsboardErrorCode.BAD_REQUEST_PARAMS); + } + } Asset savedAsset = checkNotNull(assetService.saveAsset(asset)); autoCommit(user, savedAsset.getId()); notificationEntityService.notifyCreateOrUpdateEntity(tenantId, savedAsset.getId(), savedAsset, diff --git a/application/src/main/java/org/thingsboard/server/service/install/CassandraKeyspaceService.java b/application/src/main/java/org/thingsboard/server/service/install/CassandraKeyspaceService.java new file mode 100644 index 0000000000..dfdd05e829 --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/install/CassandraKeyspaceService.java @@ -0,0 +1,37 @@ +/** + * Copyright © 2016-2022 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.service.install; + +import org.springframework.context.annotation.Profile; +import org.springframework.stereotype.Service; +import org.thingsboard.server.dao.util.NoSqlAnyDaoNonCloud; + +/* +* Create keyspace for Cassandra NoSQL database for non-cloud deployment. +* For cloud service like Astra DBaas admin have to create keyspace manually on cloud UI. +* Then create tokens with database admin role and put it on Thingsboard parameters. +* Without this service cloud DB will end up with exception like +* UnauthorizedException: Missing correct permission on thingsboard +* */ +@Service +@NoSqlAnyDaoNonCloud +@Profile("install") +public class CassandraKeyspaceService extends CassandraAbstractDatabaseSchemaService + implements NoSqlKeyspaceService { + public CassandraKeyspaceService() { + super("schema-keyspace.cql"); + } +} diff --git a/application/src/main/java/org/thingsboard/server/service/install/DbUpgradeExecutorService.java b/application/src/main/java/org/thingsboard/server/service/install/DbUpgradeExecutorService.java new file mode 100644 index 0000000000..b3ce77ac7a --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/install/DbUpgradeExecutorService.java @@ -0,0 +1,26 @@ +/** + * Copyright © 2016-2022 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.service.install; + +import org.springframework.context.annotation.Profile; +import org.springframework.stereotype.Component; +import org.thingsboard.server.service.executors.DbCallbackExecutorService; + +@Component +@Profile("install") +public class DbUpgradeExecutorService extends DbCallbackExecutorService { + +} diff --git a/application/src/main/java/org/thingsboard/server/service/install/NoSqlKeyspaceService.java b/application/src/main/java/org/thingsboard/server/service/install/NoSqlKeyspaceService.java new file mode 100644 index 0000000000..5b8d772dca --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/install/NoSqlKeyspaceService.java @@ -0,0 +1,19 @@ +/** + * Copyright © 2016-2022 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.service.install; + +public interface NoSqlKeyspaceService extends DatabaseSchemaService { +} 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 092726d0c8..80665d544f 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 @@ -15,6 +15,8 @@ */ package org.thingsboard.server.service.install; +import com.google.common.util.concurrent.Futures; +import com.google.common.util.concurrent.ListenableFuture; import lombok.extern.slf4j.Slf4j; import org.apache.commons.collections.CollectionUtils; import org.springframework.beans.factory.annotation.Autowired; @@ -32,14 +34,14 @@ import org.thingsboard.server.common.data.queue.ProcessingStrategyType; import org.thingsboard.server.common.data.queue.Queue; import org.thingsboard.server.common.data.queue.SubmitStrategy; import org.thingsboard.server.common.data.queue.SubmitStrategyType; +import org.thingsboard.server.dao.asset.AssetDao; import org.thingsboard.server.dao.asset.AssetProfileService; -import org.thingsboard.server.dao.asset.AssetService; import org.thingsboard.server.dao.dashboard.DashboardService; import org.thingsboard.server.dao.device.DeviceProfileService; import org.thingsboard.server.dao.device.DeviceService; +import org.thingsboard.server.common.data.util.TbPair; import org.thingsboard.server.dao.queue.QueueService; -import org.thingsboard.server.dao.rule.RuleChainService; -import org.thingsboard.server.dao.tenant.TenantProfileService; +import org.thingsboard.server.dao.sql.tenant.TenantRepository; import org.thingsboard.server.dao.tenant.TenantService; import org.thingsboard.server.dao.usagerecord.ApiUsageStateService; import org.thingsboard.server.queue.settings.TbRuleEngineQueueConfiguration; @@ -56,7 +58,9 @@ import java.sql.SQLException; import java.sql.SQLSyntaxErrorException; import java.sql.SQLWarning; import java.sql.Statement; +import java.util.ArrayList; import java.util.List; +import java.util.UUID; import java.util.concurrent.TimeUnit; import static org.thingsboard.server.service.install.DatabaseHelper.ADDITIONAL_INFO; @@ -106,11 +110,14 @@ public class SqlDatabaseUpgradeService implements DatabaseEntitiesUpgradeService @Autowired private TenantService tenantService; + @Autowired + private TenantRepository tenantRepository; + @Autowired private DeviceService deviceService; @Autowired - private AssetService assetService; + private AssetDao assetDao; @Autowired private DeviceProfileService deviceProfileService; @@ -129,10 +136,7 @@ public class SqlDatabaseUpgradeService implements DatabaseEntitiesUpgradeService private TbRuleEngineQueueConfigService queueConfig; @Autowired - private RuleChainService ruleChainService; - - @Autowired - private TenantProfileService tenantProfileService; + private DbUpgradeExecutorService dbUpgradeExecutor; @Override public void upgradeDatabase(String fromVersion) throws Exception { @@ -620,26 +624,45 @@ public class SqlDatabaseUpgradeService implements DatabaseEntitiesUpgradeService schemaUpdateFile = Paths.get(installScripts.getDataDir(), "upgrade", "3.4.1", "schema_update_before.sql"); loadSql(schemaUpdateFile, conn); + conn.createStatement().execute("DELETE FROM asset a WHERE NOT exists(SELECT id FROM tenant WHERE id = a.tenant_id);"); + log.info("Creating default asset profiles..."); - PageLink pageLink = new PageLink(100); - PageData pageData; + + PageLink pageLink = new PageLink(1000); + PageData tenantIds; do { - pageData = tenantService.findTenants(pageLink); - for (Tenant tenant : pageData.getData()) { - List assetTypes = assetService.findAssetTypesByTenantId(tenant.getId()).get(); - try { - assetProfileService.createDefaultAssetProfile(tenant.getId()); - } catch (Exception e) { - } - for (EntitySubtype assetType : assetTypes) { + List> futures = new ArrayList<>(); + tenantIds = tenantService.findTenantsIds(pageLink); + for (TenantId tenantId : tenantIds.getData()) { + futures.add(dbUpgradeExecutor.submit(() -> { try { - assetProfileService.findOrCreateAssetProfile(tenant.getId(), assetType.getType()); - } catch (Exception e) { - } + assetProfileService.createDefaultAssetProfile(tenantId); + } catch (Exception e) {} + })); + } + Futures.allAsList(futures).get(); + pageLink = pageLink.nextPageLink(); + } while (tenantIds.hasNext()); + + pageLink = new PageLink(1000); + PageData> pairs; + do { + List> futures = new ArrayList<>(); + pairs = assetDao.getAllAssetTypes(pageLink); + for (TbPair pair : pairs.getData()) { + TenantId tenantId = new TenantId(pair.getFirst()); + String assetType = pair.getSecond(); + if (!"default".equals(assetType)) { + futures.add(dbUpgradeExecutor.submit(() -> { + try { + assetProfileService.findOrCreateAssetProfile(tenantId, assetType); + } catch (Exception e) {} + })); } } + Futures.allAsList(futures).get(); pageLink = pageLink.nextPageLink(); - } while (pageData.hasNext()); + } while (pairs.hasNext()); log.info("Updating asset profiles..."); conn.createStatement().execute("call update_asset_profiles()"); @@ -728,5 +751,4 @@ public class SqlDatabaseUpgradeService implements DatabaseEntitiesUpgradeService return queue; } - } diff --git a/application/src/main/java/org/thingsboard/server/service/install/update/DefaultCacheCleanupService.java b/application/src/main/java/org/thingsboard/server/service/install/update/DefaultCacheCleanupService.java index 9af49f4a7c..6e67876dc4 100644 --- a/application/src/main/java/org/thingsboard/server/service/install/update/DefaultCacheCleanupService.java +++ b/application/src/main/java/org/thingsboard/server/service/install/update/DefaultCacheCleanupService.java @@ -73,6 +73,11 @@ public class DefaultCacheCleanupService implements CacheCleanupService { log.info("Clear cache to upgrade from version 3.3.4 to 3.4.0 ..."); clearAll(); break; + case "3.4.1": + log.info("Clear cache to upgrade from version 3.4.1 to 3.4.2 ..."); + clearCacheByName("assets"); + clearCacheByName("repositorySettings"); + break; default: //Do nothing, since cache cleanup is optional. } diff --git a/application/src/main/java/org/thingsboard/server/service/install/update/DefaultDataUpdateService.java b/application/src/main/java/org/thingsboard/server/service/install/update/DefaultDataUpdateService.java index a8cf5374a2..93f83414ae 100644 --- a/application/src/main/java/org/thingsboard/server/service/install/update/DefaultDataUpdateService.java +++ b/application/src/main/java/org/thingsboard/server/service/install/update/DefaultDataUpdateService.java @@ -64,6 +64,7 @@ import org.thingsboard.server.common.data.tenant.profile.TenantProfileQueueConfi import org.thingsboard.server.dao.DaoUtil; import org.thingsboard.server.dao.alarm.AlarmDao; import org.thingsboard.server.dao.audit.AuditLogDao; +import org.thingsboard.server.dao.edge.EdgeEventDao; import org.thingsboard.server.dao.entity.EntityService; import org.thingsboard.server.dao.entityview.EntityViewService; import org.thingsboard.server.dao.event.EventService; @@ -142,6 +143,9 @@ public class DefaultDataUpdateService implements DataUpdateService { @Autowired private AuditLogDao auditLogDao; + @Autowired + private EdgeEventDao edgeEventDao; + @Override public void updateData(String fromVersion) throws Exception { switch (fromVersion) { @@ -181,14 +185,21 @@ public class DefaultDataUpdateService implements DataUpdateService { } break; case "3.4.1": + log.info("Updating data from version 3.4.1 to 3.4.2 ..."); boolean skipAuditLogsMigration = getEnv("TB_SKIP_AUDIT_LOGS_MIGRATION", false); if (!skipAuditLogsMigration) { - log.info("Updating data from version 3.4.1 to 3.4.2 ..."); log.info("Starting audit logs migration. Can be skipped with TB_SKIP_AUDIT_LOGS_MIGRATION env variable set to true"); auditLogDao.migrateAuditLogs(); } else { log.info("Skipping audit logs migration"); } + boolean skipEdgeEventsMigration = getEnv("TB_SKIP_EDGE_EVENTS_MIGRATION", false); + if (!skipEdgeEventsMigration) { + log.info("Starting edge events migration. Can be skipped with TB_SKIP_EDGE_EVENTS_MIGRATION env variable set to true"); + edgeEventDao.migrateEdgeEvents(); + } else { + log.info("Skipping edge events migration"); + } break; default: throw new RuntimeException("Unable to update data, unsupported fromVersion: " + fromVersion); diff --git a/application/src/main/java/org/thingsboard/server/service/state/DefaultDeviceStateService.java b/application/src/main/java/org/thingsboard/server/service/state/DefaultDeviceStateService.java index c9c54161dd..d52c712074 100644 --- a/application/src/main/java/org/thingsboard/server/service/state/DefaultDeviceStateService.java +++ b/application/src/main/java/org/thingsboard/server/service/state/DefaultDeviceStateService.java @@ -24,6 +24,7 @@ import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.ListeningExecutorService; import com.google.common.util.concurrent.MoreExecutors; import lombok.Getter; +import lombok.Setter; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; @@ -121,7 +122,8 @@ public class DefaultDeviceStateService extends AbstractPartitionBasedService PERSISTENT_ATTRIBUTE_KEYS = Arrays.asList( new EntityKey(EntityKeyType.SERVER_ATTRIBUTE, LAST_ACTIVITY_TIME), @@ -152,14 +154,21 @@ public class DefaultDeviceStateService extends AbstractPartitionBasedService transformInactivityTimeout(ListenableFuture future) { return Futures.transformAsync(future, deviceStateData -> { - if (!persistToTelemetry || deviceStateData.getState().getInactivityTimeout() != TimeUnit.SECONDS.toMillis(defaultInactivityTimeoutInSec)) { + if (!persistToTelemetry || deviceStateData.getState().getInactivityTimeout() != defaultInactivityTimeoutMs) { return future; //fail fast } var attributesFuture = attributesService.find(TenantId.SYS_TENANT_ID, deviceStateData.getDeviceId(), SERVER_SCOPE, INACTIVITY_TIMEOUT); @@ -563,7 +572,7 @@ public class DefaultDeviceStateService extends AbstractPartitionBasedService 0") public class EdgeEventsCleanUpService extends AbstractCleanUpService { public static final String RANDOM_DELAY_INTERVAL_MS_EXPRESSION = @@ -34,20 +41,29 @@ public class EdgeEventsCleanUpService extends AbstractCleanUpService { @Value("${sql.ttl.edge_events.edge_events_ttl}") private long ttl; - @Value("${sql.ttl.edge_events.enabled}") + @Value("${sql.edge_events.partition_size:168}") + private int partitionSizeInHours; + + @Value("${sql.ttl.edge_events.enabled:true}") private boolean ttlTaskExecutionEnabled; private final EdgeEventService edgeEventService; - public EdgeEventsCleanUpService(PartitionService partitionService, EdgeEventService edgeEventService) { + private final SqlPartitioningRepository partitioningRepository; + + public EdgeEventsCleanUpService(PartitionService partitionService, EdgeEventService edgeEventService, SqlPartitioningRepository partitioningRepository) { super(partitionService); this.edgeEventService = edgeEventService; + this.partitioningRepository = partitioningRepository; } @Scheduled(initialDelayString = RANDOM_DELAY_INTERVAL_MS_EXPRESSION, fixedDelayString = "${sql.ttl.edge_events.execution_interval_ms}") public void cleanUp() { + long edgeEventsExpTime = System.currentTimeMillis() - TimeUnit.SECONDS.toMillis(ttl); if (ttlTaskExecutionEnabled && isSystemTenantPartitionMine()) { - edgeEventService.cleanupEvents(ttl); + edgeEventService.cleanupEvents(edgeEventsExpTime); + } else { + partitioningRepository.cleanupPartitionsCache(EDGE_EVENT_COLUMN_FAMILY_NAME, edgeEventsExpTime, TimeUnit.HOURS.toMillis(partitionSizeInHours)); } } diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index 4fae947883..28149246c9 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -200,6 +200,14 @@ cassandra: username: "${CASSANDRA_USERNAME:}" # Specify your password password: "${CASSANDRA_PASSWORD:}" + # Astra DB connect https://astra.datastax.com/ + cloud: + # /etc/thingsboard/astra/secure-connect-thingsboard.zip + secure_connect_bundle_path: "${CASSANDRA_CLOUD_SECURE_BUNDLE_PATH:}" + # DucitQPHMzPCBOZqFYexAfKk + client_id: "${CASSANDRA_CLOUD_CLIENT_ID:}" + # ZnF7FpuHp43FP5BzM+KY8wGmSb4Ql6BhT4Z7sOU13ze+gXQ-n7OkFpNuB,oACUIQObQnK0g4bSPoZhK5ejkcF9F.j6f64j71Sr.tiRe0Fsq2hPS1ZCGSfAaIgg63IydG + client_secret: "${CASSANDRA_CLOUD_CLIENT_SECRET:}" # Cassandra cluster connection socket parameters # socket: @@ -265,6 +273,7 @@ sql: batch_size: "${SQL_EDGE_EVENTS_BATCH_SIZE:1000}" batch_max_delay: "${SQL_EDGE_EVENTS_BATCH_MAX_DELAY_MS:100}" stats_print_interval_ms: "${SQL_EDGE_EVENTS_BATCH_STATS_PRINT_MS:10000}" + partition_size: "${SQL_EDGE_EVENTS_PARTITION_SIZE_HOURS:168}" # Number of hours to partition the events. The current value corresponds to one week. audit_logs: partition_size: "${SQL_AUDIT_LOGS_PARTITION_SIZE_HOURS:168}" # Default value - 1 week # Specify whether to sort entities before batch update. Should be enabled for cluster mode to avoid deadlocks @@ -534,6 +543,7 @@ spring.servlet.multipart.max-file-size: "50MB" spring.servlet.multipart.max-request-size: "50MB" spring.jpa.properties.hibernate.jdbc.lob.non_contextual_creation: "true" +# Note: as for current Spring JPA version, custom NullHandling for the Sort.Order is ignored and this parameter is used spring.jpa.properties.hibernate.order_by.default_null_ordering: "${SPRING_JPA_PROPERTIES_HIBERNATE_ORDER_BY_DEFAULT_NULL_ORDERING:last}" # SQL DAO Configuration @@ -615,6 +625,7 @@ mvel: max_black_list_duration_sec: "${MVEL_MAX_BLACKLIST_DURATION_SEC:60}" # Specify thread pool size for javascript executor service thread_pool_size: "${MVEL_THREAD_POOL_SIZE:50}" + compiled_scripts_cache_size: "${MVEL_COMPILED_SCRIPTS_CACHE_SIZE:1000}" stats: enabled: "${TB_MVEL_STATS_ENABLED:false}" print_interval_ms: "${TB_MVEL_STATS_PRINT_INTERVAL_MS:10000}" diff --git a/application/src/test/java/org/thingsboard/server/controller/BaseEdgeEventControllerTest.java b/application/src/test/java/org/thingsboard/server/controller/BaseEdgeEventControllerTest.java index ef66270130..ccf218c132 100644 --- a/application/src/test/java/org/thingsboard/server/controller/BaseEdgeEventControllerTest.java +++ b/application/src/test/java/org/thingsboard/server/controller/BaseEdgeEventControllerTest.java @@ -22,6 +22,9 @@ import org.junit.After; import org.junit.Assert; import org.junit.Before; import org.junit.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.boot.test.mock.mockito.SpyBean; import org.springframework.test.context.TestPropertySource; import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.Tenant; @@ -29,16 +32,27 @@ import org.thingsboard.server.common.data.User; import org.thingsboard.server.common.data.asset.Asset; import org.thingsboard.server.common.data.edge.Edge; import org.thingsboard.server.common.data.edge.EdgeEvent; +import org.thingsboard.server.common.data.edge.EdgeEventActionType; import org.thingsboard.server.common.data.edge.EdgeEventType; import org.thingsboard.server.common.data.id.EdgeId; import org.thingsboard.server.common.data.page.PageData; import org.thingsboard.server.common.data.page.TimePageLink; import org.thingsboard.server.common.data.relation.EntityRelation; import org.thingsboard.server.common.data.security.Authority; +import org.thingsboard.server.dao.edge.EdgeEventDao; +import org.thingsboard.server.dao.sqlts.insert.sql.SqlPartitioningRepository; +import org.thingsboard.server.service.ttl.EdgeEventsCleanUpService; +import java.time.LocalDate; +import java.time.ZoneOffset; import java.util.List; +import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.reset; +import static org.mockito.Mockito.verify; +import static org.assertj.core.api.Assertions.assertThat; import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; @TestPropertySource(properties = { @@ -50,6 +64,18 @@ public abstract class BaseEdgeEventControllerTest extends AbstractControllerTest private Tenant savedTenant; private User tenantAdmin; + @Autowired + private EdgeEventDao edgeEventDao; + @SpyBean + private SqlPartitioningRepository partitioningRepository; + @Autowired + private EdgeEventsCleanUpService edgeEventsCleanUpService; + + @Value("#{${sql.edge_events.partition_size} * 60 * 60 * 1000}") + private long partitionDurationInMs; + @Value("${sql.ttl.edge_events.edge_event_ttl}") + private long edgeEventTtlInSec; + @Before public void beforeTest() throws Exception { loginSysAdmin(); @@ -114,6 +140,34 @@ public abstract class BaseEdgeEventControllerTest extends AbstractControllerTest Assert.assertTrue(edgeEvents.stream().anyMatch(ee -> EdgeEventType.RELATION.equals(ee.getType()))); } + @Test + public void saveEdgeEvent_thenCreatePartitionIfNotExist() { + reset(partitioningRepository); + EdgeEvent edgeEvent = createEdgeEvent(); + verify(partitioningRepository).createPartitionIfNotExists(eq("edge_event"), eq(edgeEvent.getCreatedTime()), eq(partitionDurationInMs)); + List partitions = partitioningRepository.fetchPartitions("edge_event"); + assertThat(partitions).singleElement().satisfies(partitionStartTs -> { + assertThat(partitionStartTs).isEqualTo(partitioningRepository.calculatePartitionStartTime(edgeEvent.getCreatedTime(), partitionDurationInMs)); + }); + } + + @Test + public void cleanUpEdgeEventByTtl_dropOldPartitions() { + long oldEdgeEventTs = LocalDate.of(2020, 10, 1).atStartOfDay().toInstant(ZoneOffset.UTC).toEpochMilli(); + long partitionStartTs = partitioningRepository.calculatePartitionStartTime(oldEdgeEventTs, partitionDurationInMs); + partitioningRepository.createPartitionIfNotExists("edge_event", oldEdgeEventTs, partitionDurationInMs); + List partitions = partitioningRepository.fetchPartitions("edge_event"); + assertThat(partitions).contains(partitionStartTs); + + edgeEventsCleanUpService.cleanUp(); + partitions = partitioningRepository.fetchPartitions("edge_event"); + assertThat(partitions).doesNotContain(partitionStartTs); + assertThat(partitions).allSatisfy(partitionsStart -> { + long partitionEndTs = partitionsStart + partitionDurationInMs; + assertThat(partitionEndTs).isGreaterThan(System.currentTimeMillis() - TimeUnit.SECONDS.toMillis(edgeEventTtlInSec)); + }); + } + private List findEdgeEvents(EdgeId edgeId) throws Exception { return doGetTypedWithTimePageLink("/api/edge/" + edgeId.toString() + "/events?", new TypeReference>() { @@ -134,4 +188,19 @@ public abstract class BaseEdgeEventControllerTest extends AbstractControllerTest return asset; } + private EdgeEvent createEdgeEvent() { + EdgeEvent edgeEvent = new EdgeEvent(); + edgeEvent.setCreatedTime(System.currentTimeMillis()); + edgeEvent.setTenantId(tenantId); + edgeEvent.setAction(EdgeEventActionType.ADDED); + edgeEvent.setEntityId(tenantAdmin.getUuidId()); + edgeEvent.setType(EdgeEventType.ALARM); + try { + edgeEventDao.saveAsync(edgeEvent).get(); + } catch (InterruptedException | ExecutionException e) { + throw new RuntimeException(e); + } + return edgeEvent; + } + } diff --git a/application/src/test/java/org/thingsboard/server/edge/BaseDeviceEdgeTest.java b/application/src/test/java/org/thingsboard/server/edge/BaseDeviceEdgeTest.java index 392a6814d4..ffe21ae961 100644 --- a/application/src/test/java/org/thingsboard/server/edge/BaseDeviceEdgeTest.java +++ b/application/src/test/java/org/thingsboard/server/edge/BaseDeviceEdgeTest.java @@ -511,8 +511,11 @@ abstract public class BaseDeviceEdgeTest extends AbstractEdgeTest { body.put("expirationTime", System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(10)); body.put("method", "test_method"); body.put("params", "{\"param1\":\"value1\"}"); + body.put("persisted", true); + body.put("retries", 2); - EdgeEvent edgeEvent = constructEdgeEvent(tenantId, edge.getId(), EdgeEventActionType.RPC_CALL, device.getId().getId(), EdgeEventType.DEVICE, body); + EdgeEvent edgeEvent = constructEdgeEvent(tenantId, edge.getId(), EdgeEventActionType.RPC_CALL, + device.getId().getId(), EdgeEventType.DEVICE, body); edgeImitator.expectMessageAmount(1); edgeEventService.saveAsync(edgeEvent).get(); clusterService.onEdgeEventUpdate(tenantId, edge.getId()); @@ -522,6 +525,8 @@ abstract public class BaseDeviceEdgeTest extends AbstractEdgeTest { Assert.assertTrue(latestMessage instanceof DeviceRpcCallMsg); DeviceRpcCallMsg latestDeviceRpcCallMsg = (DeviceRpcCallMsg) latestMessage; Assert.assertEquals("test_method", latestDeviceRpcCallMsg.getRequestMsg().getMethod()); + Assert.assertTrue(latestDeviceRpcCallMsg.getPersisted()); + Assert.assertEquals(2, latestDeviceRpcCallMsg.getRetries()); } private void sendAttributesRequestAndVerify(Device device, String scope, String attributesDataStr, String expectedKey, diff --git a/application/src/test/java/org/thingsboard/server/service/script/MvelInvokeServiceTest.java b/application/src/test/java/org/thingsboard/server/service/script/MvelInvokeServiceTest.java index a40a087514..1c3c134a82 100644 --- a/application/src/test/java/org/thingsboard/server/service/script/MvelInvokeServiceTest.java +++ b/application/src/test/java/org/thingsboard/server/service/script/MvelInvokeServiceTest.java @@ -16,6 +16,7 @@ package org.thingsboard.server.service.script; import com.fasterxml.jackson.databind.node.ObjectNode; +import com.github.benmanes.caffeine.cache.Cache; import org.junit.Assert; import org.junit.jupiter.api.Test; import org.springframework.beans.factory.annotation.Autowired; @@ -24,15 +25,22 @@ import org.springframework.test.context.TestPropertySource; import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.script.api.ScriptType; import org.thingsboard.script.api.mvel.MvelInvokeService; +import org.thingsboard.script.api.mvel.MvelScript; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.controller.AbstractControllerTest; import org.thingsboard.server.dao.service.DaoSqlTest; +import java.io.Serializable; +import java.lang.reflect.Field; +import java.util.ArrayList; +import java.util.List; import java.util.Map; import java.util.UUID; +import java.util.concurrent.ConcurrentMap; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; +import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; @DaoSqlTest @@ -41,6 +49,7 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy; "mvel.max_total_args_size=50", "mvel.max_result_size=50", "mvel.max_errors=2", + "mvel.compiled_scripts_cache_size=100" }) class MvelInvokeServiceTest extends AbstractControllerTest { @@ -110,6 +119,89 @@ class MvelInvokeServiceTest extends AbstractControllerTest { assertThatScriptIsBlocked(scriptId); } + @Test + void givenScriptsWithSameBody_thenCompileAndCacheOnlyOnce() throws Exception { + String script = "return msg.temperature > 20;"; + List scriptsIds = new ArrayList<>(); + for (int i = 0; i < 100; i++) { + UUID scriptId = evalScript(script); + scriptsIds.add(scriptId); + } + + Map scriptIdToHash = getFieldValue(invokeService, "scriptIdToHash"); + Map scriptMap = getFieldValue(invokeService, "scriptMap"); + Cache compiledScriptsCache = getFieldValue(invokeService, "compiledScriptsCache"); + + String scriptHash = scriptIdToHash.get(scriptsIds.get(0)); + + assertThat(scriptsIds.stream().map(scriptIdToHash::get)).containsOnly(scriptHash); + assertThat(scriptMap).containsKey(scriptHash); + assertThat(compiledScriptsCache.getIfPresent(scriptHash)).isNotNull(); + } + + @Test + public void whenReleasingScript_thenCheckForScriptHashUsages() throws Exception { + String script = "return msg.temperature > 20;"; + List scriptsIds = new ArrayList<>(); + for (int i = 0; i < 10; i++) { + UUID scriptId = evalScript(script); + scriptsIds.add(scriptId); + } + + Map scriptIdToHash = getFieldValue(invokeService, "scriptIdToHash"); + Map scriptMap = getFieldValue(invokeService, "scriptMap"); + Cache compiledScriptsCache = getFieldValue(invokeService, "compiledScriptsCache"); + + String scriptHash = scriptIdToHash.get(scriptsIds.get(0)); + for (int i = 0; i < 9; i++) { + UUID scriptId = scriptsIds.get(i); + assertThat(scriptIdToHash).containsKey(scriptId); + invokeService.release(scriptId); + assertThat(scriptIdToHash).doesNotContainKey(scriptId); + } + assertThat(scriptMap).containsKey(scriptHash); + assertThat(compiledScriptsCache.getIfPresent(scriptHash)).isNotNull(); + + invokeService.release(scriptsIds.get(9)); + assertThat(scriptMap).doesNotContainKey(scriptHash); + assertThat(compiledScriptsCache.getIfPresent(scriptHash)).isNull(); + } + + @Test + public void whenCompiledScriptsCacheIsTooBig_thenRemoveRarelyUsedScripts() throws Exception { + Map scriptIdToHash = getFieldValue(invokeService, "scriptIdToHash"); + Cache compiledScriptsCache = getFieldValue(invokeService, "compiledScriptsCache"); + + List scriptsIds = new ArrayList<>(); + for (int i = 0; i < 110; i++) { // mvel.compiled_scripts_cache_size = 100 + String script = "return msg.temperature > " + i; + UUID scriptId = evalScript(script); + scriptsIds.add(scriptId); + + for (int j = 0; j < i; j++) { + invokeScript(scriptId, "{ \"temperature\": 12 }"); // so that scriptsIds is ordered by number of invocations + } + } + + ConcurrentMap cache = compiledScriptsCache.asMap(); + + for (int i = 0; i < 10; i++) { // iterating rarely used scripts + UUID scriptId = scriptsIds.get(i); + String scriptHash = scriptIdToHash.get(scriptId); + assertThat(cache).doesNotContainKey(scriptHash); + } + for (int i = 10; i < 110; i++) { + UUID scriptId = scriptsIds.get(i); + String scriptHash = scriptIdToHash.get(scriptId); + assertThat(cache).containsKey(scriptHash); + } + + UUID scriptRemovedFromCache = scriptsIds.get(0); + assertThat(compiledScriptsCache.getIfPresent(scriptIdToHash.get(scriptRemovedFromCache))).isNull(); + invokeScript(scriptRemovedFromCache, "{ \"temperature\": 12 }"); + assertThat(compiledScriptsCache.getIfPresent(scriptIdToHash.get(scriptRemovedFromCache))).isNotNull(); + } + private void assertThatScriptIsBlocked(UUID scriptId) { assertThatThrownBy(() -> { invokeScript(scriptId, "{}"); @@ -125,4 +217,10 @@ class MvelInvokeServiceTest extends AbstractControllerTest { return invokeService.invokeScript(TenantId.SYS_TENANT_ID, null, scriptId, msg, "{}", "POST_TELEMETRY_REQUEST").get().toString(); } + private T getFieldValue(Object target, String fieldName) throws Exception { + Field field = target.getClass().getDeclaredField(fieldName); + field.setAccessible(true); + return (T) field.get(target); + } + } diff --git a/application/src/test/java/org/thingsboard/server/service/state/DefaultDeviceStateServiceTest.java b/application/src/test/java/org/thingsboard/server/service/state/DefaultDeviceStateServiceTest.java index d5d74f2512..868889a6cc 100644 --- a/application/src/test/java/org/thingsboard/server/service/state/DefaultDeviceStateServiceTest.java +++ b/application/src/test/java/org/thingsboard/server/service/state/DefaultDeviceStateServiceTest.java @@ -15,27 +15,37 @@ */ package org.thingsboard.server.service.state; +import org.junit.Assert; import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; import org.mockito.Mock; import org.mockito.Mockito; import org.mockito.junit.MockitoJUnitRunner; +import org.thingsboard.server.cluster.TbClusterService; +import org.thingsboard.server.common.data.DeviceIdInfo; import org.thingsboard.server.common.data.id.DeviceId; +import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.data.query.EntityData; +import org.thingsboard.server.common.data.query.EntityKeyType; +import org.thingsboard.server.common.data.query.TsValue; import org.thingsboard.server.dao.attributes.AttributesService; import org.thingsboard.server.dao.device.DeviceService; import org.thingsboard.server.dao.tenant.TenantService; import org.thingsboard.server.dao.timeseries.TimeseriesService; import org.thingsboard.server.queue.discovery.PartitionService; -import org.thingsboard.server.cluster.TbClusterService; import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; +import java.util.Map; +import java.util.UUID; + import static org.hamcrest.CoreMatchers.is; import static org.hamcrest.MatcherAssert.assertThat; import static org.mockito.BDDMockito.willReturn; import static org.mockito.Mockito.never; import static org.mockito.Mockito.spy; import static org.mockito.Mockito.times; +import static org.thingsboard.server.service.state.DefaultDeviceStateService.INACTIVITY_TIMEOUT; @RunWith(MockitoJUnitRunner.class) public class DefaultDeviceStateServiceTest { @@ -83,4 +93,40 @@ public class DefaultDeviceStateServiceTest { Mockito.verify(service, times(1)).fetchDeviceStateDataUsingEntityDataQuery(deviceId); } + @Test + public void givenPersistToTelemetryAndDefaultInactivityTimeoutFetched_whenTransformingToDeviceStateData_thenTryGetInactivityFromAttribute() { + var defaultInactivityTimeoutInSec = 60L; + var latest = + Map.of( + EntityKeyType.TIME_SERIES, Map.of(INACTIVITY_TIMEOUT, new TsValue(0, Long.toString(defaultInactivityTimeoutInSec * 1000))), + EntityKeyType.SERVER_ATTRIBUTE, Map.of(INACTIVITY_TIMEOUT, new TsValue(0, Long.toString(5000L))) + ); + + process(latest, defaultInactivityTimeoutInSec); + } + + @Test + public void givenPersistToTelemetryAndNoInactivityTimeoutFetchedFromTimeSeries_whenTransformingToDeviceStateData_thenTryGetInactivityFromAttribute() { + var defaultInactivityTimeoutInSec = 60L; + var latest = + Map.of( + EntityKeyType.SERVER_ATTRIBUTE, Map.of(INACTIVITY_TIMEOUT, new TsValue(0, Long.toString(5000L))) + ); + + process(latest, defaultInactivityTimeoutInSec); + } + + private void process(Map> latest, long defaultInactivityTimeoutInSec) { + service.setDefaultInactivityTimeoutInSec(defaultInactivityTimeoutInSec); + service.setDefaultInactivityTimeoutMs(defaultInactivityTimeoutInSec * 1000); + service.setPersistToTelemetry(true); + + var deviceUuid = UUID.randomUUID(); + var deviceId = new DeviceId(deviceUuid); + + DeviceStateData deviceStateData = service.toDeviceStateData(new EntityData(deviceId, latest, Map.of()), new DeviceIdInfo(TenantId.SYS_TENANT_ID.getId(), UUID.randomUUID(), deviceUuid)); + + Assert.assertEquals(5000L, deviceStateData.getState().getInactivityTimeout()); + } + } \ No newline at end of file diff --git a/application/src/test/java/org/thingsboard/server/transport/TransportNoSqlTestSuite.java b/application/src/test/java/org/thingsboard/server/transport/TransportNoSqlTestSuite.java index 255057cc2f..caaf5be6de 100644 --- a/application/src/test/java/org/thingsboard/server/transport/TransportNoSqlTestSuite.java +++ b/application/src/test/java/org/thingsboard/server/transport/TransportNoSqlTestSuite.java @@ -35,6 +35,7 @@ public class TransportNoSqlTestSuite { public static CustomCassandraCQLUnit cassandraUnit = new CustomCassandraCQLUnit( Arrays.asList( + new ClassPathCQLDataSet("cassandra/schema-keyspace.cql", false, false), new ClassPathCQLDataSet("cassandra/schema-ts.cql", false, false), new ClassPathCQLDataSet("cassandra/schema-ts-latest.cql", false, false) ), diff --git a/application/src/test/resources/application-test.properties b/application/src/test/resources/application-test.properties index eacd1733d3..e4a36c151d 100644 --- a/application/src/test/resources/application-test.properties +++ b/application/src/test/resources/application-test.properties @@ -59,4 +59,7 @@ queue.rule-engine.queues[2].processing-strategy.max-pause-between-retries=0 usage.stats.report.enabled=false sql.audit_logs.partition_size=24 -sql.ttl.audit_logs.ttl=2592000 \ No newline at end of file +sql.ttl.audit_logs.ttl=2592000 + +sql.edge_events.partition_size=168 +sql.ttl.edge_events.edge_event_ttl=2592000 diff --git a/application/src/test/resources/logback-test.xml b/application/src/test/resources/logback-test.xml index d3301bf660..3762c8aa7c 100644 --- a/application/src/test/resources/logback-test.xml +++ b/application/src/test/resources/logback-test.xml @@ -16,6 +16,8 @@ + + diff --git a/common/dao-api/src/main/java/org/thingsboard/server/dao/cassandra/AbstractCassandraCluster.java b/common/dao-api/src/main/java/org/thingsboard/server/dao/cassandra/AbstractCassandraCluster.java index 6586f42d23..f3f76aba9f 100644 --- a/common/dao-api/src/main/java/org/thingsboard/server/dao/cassandra/AbstractCassandraCluster.java +++ b/common/dao-api/src/main/java/org/thingsboard/server/dao/cassandra/AbstractCassandraCluster.java @@ -24,11 +24,13 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.core.env.Environment; import org.springframework.core.env.Profiles; +import org.thingsboard.server.common.data.StringUtils; import org.thingsboard.server.dao.cassandra.guava.GuavaSession; import org.thingsboard.server.dao.cassandra.guava.GuavaSessionBuilder; import org.thingsboard.server.dao.cassandra.guava.GuavaSessionUtils; import javax.annotation.PreDestroy; +import java.nio.file.Paths; @Slf4j public abstract class AbstractCassandraCluster { @@ -40,6 +42,13 @@ public abstract class AbstractCassandraCluster { @Value("${cassandra.local_datacenter:datacenter1}") private String localDatacenter; + @Value("${cassandra.cloud.secure_connect_bundle_path:}") + private String cloudSecureConnectBundlePath; + @Value("${cassandra.cloud.client_id:}") + private String cloudClientId; + @Value("${cassandra.cloud.client_secret:}") + private String cloudClientSecret; + @Autowired private CassandraDriverOptions driverOptions; @@ -86,7 +95,14 @@ public abstract class AbstractCassandraCluster { this.sessionBuilder.withKeyspace(this.keyspaceName); } this.sessionBuilder.withLocalDatacenter(localDatacenter); + + if (StringUtils.isNotBlank(cloudSecureConnectBundlePath)) { + this.sessionBuilder.withCloudSecureConnectBundle(Paths.get(cloudSecureConnectBundlePath)); + this.sessionBuilder.withAuthCredentials(cloudClientId, cloudClientSecret); + } + session = sessionBuilder.build(); + if (this.metrics && this.jmx) { MetricRegistry registry = session.getMetrics().orElseThrow( diff --git a/common/dao-api/src/main/java/org/thingsboard/server/dao/cassandra/guava/GuavaDriverContext.java b/common/dao-api/src/main/java/org/thingsboard/server/dao/cassandra/guava/GuavaDriverContext.java index b5472d78a2..f193b0a7fe 100644 --- a/common/dao-api/src/main/java/org/thingsboard/server/dao/cassandra/guava/GuavaDriverContext.java +++ b/common/dao-api/src/main/java/org/thingsboard/server/dao/cassandra/guava/GuavaDriverContext.java @@ -40,26 +40,8 @@ import java.util.function.Predicate; */ public class GuavaDriverContext extends DefaultDriverContext { - public GuavaDriverContext( - DriverConfigLoader configLoader, - List> typeCodecs, - NodeStateListener nodeStateListener, - SchemaChangeListener schemaChangeListener, - RequestTracker requestTracker, - Map localDatacenters, - Map> nodeFilters, - ClassLoader classLoader) { - super( - configLoader, - ProgrammaticArguments.builder() - .addTypeCodecs(typeCodecs.toArray(new TypeCodec[0])) - .withNodeStateListener(nodeStateListener) - .withSchemaChangeListener(schemaChangeListener) - .withRequestTracker(requestTracker) - .withLocalDatacenters(localDatacenters) - .withNodeFilters(nodeFilters) - .withClassLoader(classLoader) - .build()); + public GuavaDriverContext(DriverConfigLoader configLoader, ProgrammaticArguments programmaticArguments) { + super(configLoader, programmaticArguments); } @Override diff --git a/common/dao-api/src/main/java/org/thingsboard/server/dao/cassandra/guava/GuavaSessionBuilder.java b/common/dao-api/src/main/java/org/thingsboard/server/dao/cassandra/guava/GuavaSessionBuilder.java index c95c653d64..60ff71a8bd 100644 --- a/common/dao-api/src/main/java/org/thingsboard/server/dao/cassandra/guava/GuavaSessionBuilder.java +++ b/common/dao-api/src/main/java/org/thingsboard/server/dao/cassandra/guava/GuavaSessionBuilder.java @@ -25,18 +25,8 @@ import edu.umd.cs.findbugs.annotations.NonNull; public class GuavaSessionBuilder extends SessionBuilder { @Override - protected DriverContext buildContext( - DriverConfigLoader configLoader, - ProgrammaticArguments programmaticArguments) { - return new GuavaDriverContext( - configLoader, - programmaticArguments.getTypeCodecs(), - programmaticArguments.getNodeStateListener(), - programmaticArguments.getSchemaChangeListener(), - programmaticArguments.getRequestTracker(), - programmaticArguments.getLocalDatacenters(), - programmaticArguments.getNodeFilters(), - programmaticArguments.getClassLoader()); + protected DriverContext buildContext(DriverConfigLoader configLoader, ProgrammaticArguments programmaticArguments) { + return new GuavaDriverContext(configLoader, programmaticArguments); } @Override diff --git a/common/dao-api/src/main/java/org/thingsboard/server/dao/util/NoSqlAnyDaoNonCloud.java b/common/dao-api/src/main/java/org/thingsboard/server/dao/util/NoSqlAnyDaoNonCloud.java new file mode 100644 index 0000000000..0ec01c8b0a --- /dev/null +++ b/common/dao-api/src/main/java/org/thingsboard/server/dao/util/NoSqlAnyDaoNonCloud.java @@ -0,0 +1,27 @@ +/** + * Copyright © 2016-2022 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.dao.util; + +import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; + +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; + +@Retention(RetentionPolicy.RUNTIME) +@ConditionalOnExpression("('${database.ts.type}'=='cassandra' || '${database.ts_latest.type}'=='cassandra') " + + "&& ('${cassandra.cloud.secure_connect_bundle_path}' == null || '${cassandra.cloud.secure_connect_bundle_path}'.isBlank() )") +public @interface NoSqlAnyDaoNonCloud { +} diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/DataConstants.java b/common/data/src/main/java/org/thingsboard/server/common/data/DataConstants.java index 85b9e681a1..69c36697d3 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/DataConstants.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/DataConstants.java @@ -40,6 +40,8 @@ public class DataConstants { public static final String EXPIRATION_TIME = "expirationTime"; public static final String ADDITIONAL_INFO = "additionalInfo"; public static final String RETRIES = "retries"; + public static final String EDGE_ID = "edgeId"; + public static final String DEVICE_ID = "deviceId"; public static final String COAP_TRANSPORT_NAME = "COAP"; public static final String LWM2M_TRANSPORT_NAME = "LWM2M"; public static final String MQTT_TRANSPORT_NAME = "MQTT"; diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/util/TbPair.java b/common/data/src/main/java/org/thingsboard/server/common/data/util/TbPair.java new file mode 100644 index 0000000000..c7dba0c848 --- /dev/null +++ b/common/data/src/main/java/org/thingsboard/server/common/data/util/TbPair.java @@ -0,0 +1,26 @@ +/** + * Copyright © 2016-2022 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.common.data.util; + +import lombok.AllArgsConstructor; +import lombok.Data; + +@Data +@AllArgsConstructor +public class TbPair { + private S first; + private T second; +} diff --git a/common/edge-api/src/main/proto/edge.proto b/common/edge-api/src/main/proto/edge.proto index a2e3b5989b..249f251238 100644 --- a/common/edge-api/src/main/proto/edge.proto +++ b/common/edge-api/src/main/proto/edge.proto @@ -430,6 +430,11 @@ message DeviceRpcCallMsg { bool oneway = 7; RpcRequestMsg requestMsg = 8; RpcResponseMsg responseMsg = 9; + optional bool persisted = 10; + optional int32 retries = 11; + optional string additionalInfo = 12; + optional string serviceId = 13; + optional string sessionId = 14; } message RpcRequestMsg { diff --git a/common/script/script-api/pom.xml b/common/script/script-api/pom.xml index 62dc00ca97..b7ae4993f9 100644 --- a/common/script/script-api/pom.xml +++ b/common/script/script-api/pom.xml @@ -56,6 +56,10 @@ com.google.code.gson gson + + com.github.ben-manes.caffeine + caffeine + org.slf4j slf4j-api diff --git a/common/script/script-api/src/main/java/org/thingsboard/script/api/mvel/DefaultMvelInvokeService.java b/common/script/script-api/src/main/java/org/thingsboard/script/api/mvel/DefaultMvelInvokeService.java index 901a49180e..4fb029096b 100644 --- a/common/script/script-api/src/main/java/org/thingsboard/script/api/mvel/DefaultMvelInvokeService.java +++ b/common/script/script-api/src/main/java/org/thingsboard/script/api/mvel/DefaultMvelInvokeService.java @@ -15,6 +15,10 @@ */ package org.thingsboard.script.api.mvel; +import com.github.benmanes.caffeine.cache.Cache; +import com.github.benmanes.caffeine.cache.Caffeine; +import com.google.common.hash.Hasher; +import com.google.common.hash.Hashing; import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.ListeningExecutorService; import com.google.common.util.concurrent.MoreExecutors; @@ -42,12 +46,15 @@ import org.thingsboard.server.common.stats.TbApiUsageStateClient; import javax.annotation.PostConstruct; import javax.annotation.PreDestroy; import java.io.Serializable; +import java.nio.charset.StandardCharsets; import java.util.Collections; import java.util.Map; import java.util.Optional; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.Executor; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; import java.util.regex.Pattern; @Slf4j @@ -55,7 +62,10 @@ import java.util.regex.Pattern; @Service public class DefaultMvelInvokeService extends AbstractScriptInvokeService implements MvelInvokeService { - protected Map scriptMap = new ConcurrentHashMap<>(); + protected final Map scriptIdToHash = new ConcurrentHashMap<>(); + protected final Map scriptMap = new ConcurrentHashMap<>(); + protected Cache compiledScriptsCache; + private SandboxedParserConfiguration parserConfig; private static final Pattern NEW_KEYWORD_PATTERN = Pattern.compile("new\\s"); @@ -92,8 +102,13 @@ public class DefaultMvelInvokeService extends AbstractScriptInvokeService implem @Value("${mvel.max_memory_limit_mb:8}") private long maxMemoryLimitMb; + @Value("${mvel.compiled_scripts_cache_size:1000}") + private int compiledScriptsCacheSize; + private ListeningExecutorService executor; + private final Lock lock = new ReentrantLock(); + protected DefaultMvelInvokeService(Optional apiUsageStateClient, Optional apiUsageReportClient) { super(apiUsageStateClient, apiUsageReportClient); } @@ -115,11 +130,14 @@ public class DefaultMvelInvokeService extends AbstractScriptInvokeService implem executor = MoreExecutors.listeningDecorator(ThingsBoardExecutors.newWorkStealingPool(threadPoolSize, "mvel-executor")); try { // Special command to warm up MVEL engine - Serializable script = MVEL.compileExpression("var warmUp = {}; warmUp", new SandboxedParserContext(parserConfig)); + Serializable script = compileScript("var warmUp = {}; warmUp"); MVEL.executeTbExpression(script, new ExecutionContext(parserConfig), Collections.emptyMap()); } catch (Exception e) { // do nothing } + compiledScriptsCache = Caffeine.newBuilder() + .maximumSize(compiledScriptsCacheSize) + .build(); } @PreDestroy @@ -141,16 +159,26 @@ public class DefaultMvelInvokeService extends AbstractScriptInvokeService implem @Override protected boolean isScriptPresent(UUID scriptId) { - return scriptMap.containsKey(scriptId); + return scriptIdToHash.containsKey(scriptId); } @Override protected ListenableFuture doEvalScript(TenantId tenantId, ScriptType scriptType, String scriptBody, UUID scriptId, String[] argNames) { return executor.submit(() -> { try { - Serializable compiledScript = MVEL.compileExpression(scriptBody, new SandboxedParserContext(parserConfig)); - MvelScript script = new MvelScript(compiledScript, scriptBody, argNames); - scriptMap.put(scriptId, script); + String scriptHash = hash(scriptBody, argNames); + compiledScriptsCache.get(scriptHash, k -> { + return compileScript(scriptBody); + }); + lock.lock(); + try { + scriptIdToHash.put(scriptId, scriptHash); + scriptMap.computeIfAbsent(scriptHash, k -> { + return new MvelScript(scriptBody, argNames); + }); + } finally { + lock.unlock(); + } return scriptId; } catch (Exception e) { throw new TbScriptException(scriptId, TbScriptException.ErrorCode.COMPILATION, scriptBody, e); @@ -162,12 +190,16 @@ public class DefaultMvelInvokeService extends AbstractScriptInvokeService implem protected MvelScriptExecutionTask doInvokeFunction(UUID scriptId, Object[] args) { ExecutionContext executionContext = new ExecutionContext(this.parserConfig, maxMemoryLimitMb * 1024 * 1024); return new MvelScriptExecutionTask(executionContext, executor.submit(() -> { - MvelScript script = scriptMap.get(scriptId); - if (script == null) { + String scriptHash = scriptIdToHash.get(scriptId); + if (scriptHash == null) { throw new TbScriptException(scriptId, TbScriptException.ErrorCode.OTHER, null, new RuntimeException("Script not found!")); } + MvelScript script = scriptMap.get(scriptHash); + Serializable compiledScript = compiledScriptsCache.get(scriptHash, k -> { + return compileScript(script.getScriptBody()); + }); try { - return MVEL.executeTbExpression(script.getCompiledScript(), executionContext, script.createVars(args)); + return MVEL.executeTbExpression(compiledScript, executionContext, script.createVars(args)); } catch (ScriptMemoryOverflowException e) { throw new TbScriptException(scriptId, TbScriptException.ErrorCode.OTHER, script.getScriptBody(), new RuntimeException("Script memory overflow!")); } catch (Exception e) { @@ -178,6 +210,32 @@ public class DefaultMvelInvokeService extends AbstractScriptInvokeService implem @Override protected void doRelease(UUID scriptId) throws Exception { - scriptMap.remove(scriptId); + String scriptHash = scriptIdToHash.remove(scriptId); + if (scriptHash != null) { + lock.lock(); + try { + if (!scriptIdToHash.containsValue(scriptHash)) { + scriptMap.remove(scriptHash); + compiledScriptsCache.invalidate(scriptHash); + } + } finally { + lock.unlock(); + } + } } + + private Serializable compileScript(String scriptBody) { + return MVEL.compileExpression(scriptBody, new SandboxedParserContext(parserConfig)); + } + + @SuppressWarnings("UnstableApiUsage") + protected String hash(String scriptBody, String[] argNames) { + Hasher hasher = Hashing.murmur3_128().newHasher(); + hasher.putUnencodedChars(scriptBody); + for (String argName : argNames) { + hasher.putString(argName, StandardCharsets.UTF_8); + } + return hasher.hash().toString(); + } + } diff --git a/common/script/script-api/src/main/java/org/thingsboard/script/api/mvel/MvelScript.java b/common/script/script-api/src/main/java/org/thingsboard/script/api/mvel/MvelScript.java index 7a84c7b0af..bca5d8d546 100644 --- a/common/script/script-api/src/main/java/org/thingsboard/script/api/mvel/MvelScript.java +++ b/common/script/script-api/src/main/java/org/thingsboard/script/api/mvel/MvelScript.java @@ -24,7 +24,6 @@ import java.util.Map; @Data public class MvelScript { - private final Serializable compiledScript; private final String scriptBody; private final String[] argNames; diff --git a/dao/src/main/java/org/thingsboard/server/dao/asset/AssetDao.java b/dao/src/main/java/org/thingsboard/server/dao/asset/AssetDao.java index 4d67a794b6..46a1a1bac5 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/asset/AssetDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/asset/AssetDao.java @@ -26,6 +26,7 @@ import org.thingsboard.server.common.data.page.PageLink; import org.thingsboard.server.dao.Dao; import org.thingsboard.server.dao.ExportableEntityDao; import org.thingsboard.server.dao.TenantEntityDao; +import org.thingsboard.server.common.data.util.TbPair; import java.util.List; import java.util.Optional; @@ -222,4 +223,6 @@ public interface AssetDao extends Dao, TenantEntityDao, ExportableEntityD * @return the list of asset objects */ PageData findAssetsByTenantIdAndEdgeIdAndType(UUID tenantId, UUID edgeId, String type, PageLink pageLink); + + PageData> getAllAssetTypes(PageLink pageLink); } diff --git a/dao/src/main/java/org/thingsboard/server/dao/edge/BaseEdgeEventService.java b/dao/src/main/java/org/thingsboard/server/dao/edge/BaseEdgeEventService.java index f9e94af613..be78866e7a 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/edge/BaseEdgeEventService.java +++ b/dao/src/main/java/org/thingsboard/server/dao/edge/BaseEdgeEventService.java @@ -16,8 +16,8 @@ package org.thingsboard.server.dao.edge; import com.google.common.util.concurrent.ListenableFuture; +import lombok.AllArgsConstructor; import lombok.extern.slf4j.Slf4j; -import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; import org.thingsboard.server.common.data.edge.EdgeEvent; import org.thingsboard.server.common.data.id.EdgeId; @@ -28,13 +28,12 @@ import org.thingsboard.server.dao.service.DataValidator; @Service @Slf4j +@AllArgsConstructor public class BaseEdgeEventService implements EdgeEventService { - @Autowired - private EdgeEventDao edgeEventDao; + private final EdgeEventDao edgeEventDao; - @Autowired - private DataValidator edgeEventValidator; + private final DataValidator edgeEventValidator; @Override public ListenableFuture saveAsync(EdgeEvent edgeEvent) { diff --git a/dao/src/main/java/org/thingsboard/server/dao/edge/EdgeEventDao.java b/dao/src/main/java/org/thingsboard/server/dao/edge/EdgeEventDao.java index 7a43d6c065..cb31869213 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/edge/EdgeEventDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/edge/EdgeEventDao.java @@ -54,4 +54,6 @@ public interface EdgeEventDao extends Dao { */ void cleanupEvents(long ttl); + void migrateEdgeEvents(); + } diff --git a/dao/src/main/java/org/thingsboard/server/dao/rule/BaseRuleChainService.java b/dao/src/main/java/org/thingsboard/server/dao/rule/BaseRuleChainService.java index 420739e37c..1d11d34b2d 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/rule/BaseRuleChainService.java +++ b/dao/src/main/java/org/thingsboard/server/dao/rule/BaseRuleChainService.java @@ -731,6 +731,8 @@ public class BaseRuleChainService extends AbstractEntityService implements RuleC ConstraintViolationException e = extractConstraintViolationException(t).orElse(null); if (e != null && e.getConstraintName() != null && e.getConstraintName().equalsIgnoreCase("fk_default_rule_chain_device_profile")) { throw new DataValidationException("The rule chain referenced by the device profiles cannot be deleted!"); + } else if (e != null && e.getConstraintName() != null && e.getConstraintName().equalsIgnoreCase("fk_default_rule_chain_asset_profile")) { + throw new DataValidationException("The rule chain referenced by the asset profiles cannot be deleted!"); } else { throw t; } diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/asset/AssetRepository.java b/dao/src/main/java/org/thingsboard/server/dao/sql/asset/AssetRepository.java index eadf0d2b84..aec62f0021 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/asset/AssetRepository.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/asset/AssetRepository.java @@ -23,6 +23,7 @@ import org.springframework.data.repository.query.Param; import org.thingsboard.server.dao.ExportableEntityRepository; import org.thingsboard.server.dao.model.sql.AssetEntity; import org.thingsboard.server.dao.model.sql.AssetInfoEntity; +import org.thingsboard.server.common.data.util.TbPair; import java.util.List; import java.util.UUID; @@ -70,9 +71,9 @@ public interface AssetRepository extends JpaRepository, Expor "AND a.assetProfileId = :profileId " + "AND LOWER(a.searchText) LIKE LOWER(CONCAT('%', :searchText, '%'))") Page findByTenantIdAndProfileId(@Param("tenantId") UUID tenantId, - @Param("profileId") UUID profileId, - @Param("searchText") String searchText, - Pageable pageable); + @Param("profileId") UUID profileId, + @Param("searchText") String searchText, + Pageable pageable); @Query("SELECT new org.thingsboard.server.dao.model.sql.AssetInfoEntity(a, c.title, c.additionalInfo, p.name) " + "FROM AssetEntity a " + @@ -186,14 +187,17 @@ public interface AssetRepository extends JpaRepository, Expor "AND a.type = :type " + "AND LOWER(a.searchText) LIKE LOWER(CONCAT('%', :searchText, '%'))") Page findByTenantIdAndEdgeIdAndType(@Param("tenantId") UUID tenantId, - @Param("edgeId") UUID edgeId, - @Param("type") String type, - @Param("searchText") String searchText, - Pageable pageable); + @Param("edgeId") UUID edgeId, + @Param("type") String type, + @Param("searchText") String searchText, + Pageable pageable); Long countByTenantIdAndTypeIsNot(UUID tenantId, String type); @Query("SELECT externalId FROM AssetEntity WHERE id = :id") UUID getExternalIdById(@Param("id") UUID id); + @Query(value = "SELECT DISTINCT new org.thingsboard.server.common.data.util.TbPair(a.tenantId , a.type) FROM AssetEntity a") + Page> getAllAssetTypes(Pageable pageable); + } diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/asset/JpaAssetDao.java b/dao/src/main/java/org/thingsboard/server/dao/sql/asset/JpaAssetDao.java index 489e15502e..b0e1921e87 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/asset/JpaAssetDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/asset/JpaAssetDao.java @@ -28,14 +28,17 @@ import org.thingsboard.server.common.data.id.AssetId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.page.PageData; import org.thingsboard.server.common.data.page.PageLink; +import org.thingsboard.server.common.data.page.SortOrder; import org.thingsboard.server.dao.DaoUtil; import org.thingsboard.server.dao.asset.AssetDao; import org.thingsboard.server.dao.model.sql.AssetEntity; import org.thingsboard.server.dao.model.sql.AssetInfoEntity; +import org.thingsboard.server.common.data.util.TbPair; import org.thingsboard.server.dao.sql.JpaAbstractSearchTextDao; import org.thingsboard.server.dao.util.SqlDao; import java.util.ArrayList; +import java.util.Arrays; import java.util.Collections; import java.util.List; import java.util.Objects; @@ -243,6 +246,12 @@ public class JpaAssetDao extends JpaAbstractSearchTextDao im DaoUtil.toPageable(pageLink))); } + public PageData> getAllAssetTypes(PageLink pageLink) { + log.debug("Try to find all asset types and pageLink [{}]", pageLink); + return DaoUtil.pageToPageData(assetRepository.getAllAssetTypes( + DaoUtil.toPageable(pageLink, Arrays.asList(new SortOrder("tenantId"), new SortOrder("type"))))); + } + @Override public Long countByTenantId(TenantId tenantId) { return assetRepository.countByTenantIdAndTypeIsNot(tenantId.getId(), TB_SERVICE_QUEUE); diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/edge/JpaBaseEdgeEventDao.java b/dao/src/main/java/org/thingsboard/server/dao/sql/edge/JpaBaseEdgeEventDao.java index df182659bf..3ab21ae6ff 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/edge/JpaBaseEdgeEventDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/edge/JpaBaseEdgeEventDao.java @@ -17,10 +17,11 @@ package org.thingsboard.server.dao.sql.edge; import com.datastax.oss.driver.api.core.uuid.Uuids; import com.google.common.util.concurrent.ListenableFuture; +import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; -import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.data.jpa.repository.JpaRepository; +import org.springframework.jdbc.core.JdbcTemplate; import org.springframework.stereotype.Component; import org.thingsboard.server.common.data.StringUtils; import org.thingsboard.server.common.data.edge.EdgeEvent; @@ -31,19 +32,17 @@ import org.thingsboard.server.common.data.page.TimePageLink; import org.thingsboard.server.common.stats.StatsFactory; import org.thingsboard.server.dao.DaoUtil; import org.thingsboard.server.dao.edge.EdgeEventDao; +import org.thingsboard.server.dao.model.ModelConstants; import org.thingsboard.server.dao.model.sql.EdgeEventEntity; import org.thingsboard.server.dao.sql.JpaAbstractSearchTextDao; import org.thingsboard.server.dao.sql.ScheduledLogExecutorComponent; import org.thingsboard.server.dao.sql.TbSqlBlockingQueueParams; import org.thingsboard.server.dao.sql.TbSqlBlockingQueueWrapper; +import org.thingsboard.server.dao.sqlts.insert.sql.SqlPartitioningRepository; import org.thingsboard.server.dao.util.SqlDao; import javax.annotation.PostConstruct; import javax.annotation.PreDestroy; -import java.sql.Connection; -import java.sql.PreparedStatement; -import java.sql.ResultSet; -import java.sql.SQLException; import java.util.Comparator; import java.util.Objects; import java.util.UUID; @@ -52,18 +51,25 @@ import java.util.function.Function; import static org.thingsboard.server.dao.model.ModelConstants.NULL_UUID; -@Slf4j @Component @SqlDao +@RequiredArgsConstructor +@Slf4j public class JpaBaseEdgeEventDao extends JpaAbstractSearchTextDao implements EdgeEventDao { private final UUID systemTenantId = NULL_UUID; - @Autowired - ScheduledLogExecutorComponent logExecutor; + private final ScheduledLogExecutorComponent logExecutor; + + private final StatsFactory statsFactory; + + private final EdgeEventRepository edgeEventRepository; + + private final EdgeEventInsertRepository edgeEventInsertRepository; - @Autowired - private StatsFactory statsFactory; + private final SqlPartitioningRepository partitioningRepository; + + private final JdbcTemplate jdbcTemplate; @Value("${sql.edge_events.batch_size:1000}") private int batchSize; @@ -74,13 +80,15 @@ public class JpaBaseEdgeEventDao extends JpaAbstractSearchTextDao queue; + @Value("${sql.edge_events.partitions_size:168}") + private int partitionSizeInHours; - @Autowired - private EdgeEventRepository edgeEventRepository; + @Value("${sql.ttl.edge_events.edge_events_ttl:2628000}") + private long edge_events_ttl; - @Autowired - private EdgeEventInsertRepository edgeEventInsertRepository; + private static final String TABLE_NAME = ModelConstants.EDGE_EVENT_COLUMN_FAMILY_NAME; + + private TbSqlBlockingQueueWrapper queue; @Override protected Class getEntityClass() { @@ -140,6 +148,7 @@ public class JpaBaseEdgeEventDao extends JpaAbstractSearchTextDao 0 ? System.currentTimeMillis() - TimeUnit.SECONDS.toMillis(edge_events_ttl) : 1629158400000L; + + long currentTime = System.currentTimeMillis(); + var partitionStepInMs = TimeUnit.HOURS.toMillis(partitionSizeInHours); + long numberOfPartitions = (currentTime - startTime) / partitionStepInMs; + + if (numberOfPartitions > 1000) { + String error = "Please adjust your edge event partitioning configuration. Configuration with partition size " + + "of " + partitionSizeInHours + " hours and corresponding TTL will use " + numberOfPartitions + " " + + "(> 1000) partitions which is not recommended!"; + log.error(error); + throw new RuntimeException(error); } + + while (startTime < currentTime) { + var endTime = startTime + partitionStepInMs; + log.info("Migrating edge event for time period: {} - {}", startTime, endTime); + callMigrationFunction(startTime, endTime, partitionStepInMs); + startTime = endTime; + } + log.info("Event edge migration finished"); + jdbcTemplate.execute("DROP TABLE IF EXISTS old_edge_event"); + } + + private void callMigrationFunction(long startTime, long endTime, long partitionSIzeInMs) { + jdbcTemplate.update("CALL migrate_edge_event(?, ?, ?)", startTime, endTime, partitionSIzeInMs); } } diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/query/DefaultAlarmQueryRepository.java b/dao/src/main/java/org/thingsboard/server/dao/sql/query/DefaultAlarmQueryRepository.java index 69cb4a2d86..d91a8c5fc4 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/query/DefaultAlarmQueryRepository.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/query/DefaultAlarmQueryRepository.java @@ -140,6 +140,7 @@ public class DefaultAlarmQueryRepository implements AlarmQueryRepository { selectPart.append(" a.originator_id as entity_id "); } EntityDataSortOrder sortOrder = pageLink.getSortOrder(); + String textSearchQuery = buildTextSearchQuery(ctx, query.getAlarmFields(), pageLink.getTextSearch()); if (sortOrder != null && sortOrder.getKey().getType().equals(EntityKeyType.ALARM_FIELD)) { String sortOrderKey = sortOrder.getKey().getKey(); sortPart.append(alarmFieldColumnMap.getOrDefault(sortOrderKey, sortOrderKey)) @@ -166,7 +167,11 @@ public class DefaultAlarmQueryRepository implements AlarmQueryRepository { } joinPart.append(" as e(id, priority)) e "); if (pageLink.isSearchPropagatedAlarms()) { - joinPart.append("on ea.entity_id = e.id"); + if (textSearchQuery.isEmpty()) { + joinPart.append("on ea.entity_id = e.id"); + } else { + joinPart.append("on a.entity_id = e.id"); + } } else { joinPart.append("on a.originator_id = e.id"); } @@ -230,13 +235,11 @@ public class DefaultAlarmQueryRepository implements AlarmQueryRepository { } } - String textSearchQuery = buildTextSearchQuery(ctx, query.getAlarmFields(), pageLink.getTextSearch()); - String mainQuery; - if (!textSearchQuery.isEmpty()) { - mainQuery = selectPart.toString() + fromPart.toString() + wherePart.toString(); - mainQuery = String.format("select * from (%s) a %s WHERE %s", mainQuery, joinPart, textSearchQuery); + String mainQuery = String.format("%s%s", selectPart, fromPart); + if (textSearchQuery.isEmpty()) { + mainQuery = String.format("%s%s%s", mainQuery, joinPart, wherePart); } else { - mainQuery = selectPart.toString() + fromPart.toString() + joinPart.toString() + wherePart.toString(); + mainQuery = String.format("select * from (%s%s) a %s WHERE %s", mainQuery, wherePart, joinPart, textSearchQuery); } String countQuery = String.format("select count(*) from (%s) result", mainQuery); long queryTs = System.currentTimeMillis(); diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/tenant/TenantRepository.java b/dao/src/main/java/org/thingsboard/server/dao/sql/tenant/TenantRepository.java index 7042dcbb16..ac02d77bc8 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/tenant/TenantRepository.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/tenant/TenantRepository.java @@ -53,5 +53,4 @@ public interface TenantRepository extends JpaRepository { @Query("SELECT t.id FROM TenantEntity t where t.tenantProfileId = :tenantProfileId") List findTenantIdsByTenantProfileId(@Param("tenantProfileId") UUID tenantProfileId); - } diff --git a/dao/src/main/java/org/thingsboard/server/dao/sqlts/AbstractChunkedAggregationTimeseriesDao.java b/dao/src/main/java/org/thingsboard/server/dao/sqlts/AbstractChunkedAggregationTimeseriesDao.java index 22c4661fcd..cf726c32c4 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sqlts/AbstractChunkedAggregationTimeseriesDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sqlts/AbstractChunkedAggregationTimeseriesDao.java @@ -20,7 +20,7 @@ import com.google.common.util.concurrent.ListenableFuture; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.data.domain.PageRequest; -import org.springframework.data.domain.Sort; +import org.springframework.data.domain.Sort.Direction; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.kv.Aggregation; @@ -143,8 +143,7 @@ public abstract class AbstractChunkedAggregationTimeseriesDao extends AbstractSq keyId, query.getStartTs(), query.getEndTs(), - PageRequest.of(0, query.getLimit(), - Sort.by(new Sort.Order(Sort.Direction.fromString(query.getOrder()), "ts").nullsNative()))); + PageRequest.ofSize(query.getLimit()).withSort(Direction.fromString(query.getOrder()), "ts")); tsKvEntities.forEach(tsKvEntity -> tsKvEntity.setStrKey(query.getKey())); List tsKvEntries = DaoUtil.convertDataList(tsKvEntities); long lastTs = tsKvEntries.stream().map(TsKvEntry::getTs).max(Long::compare).orElse(query.getStartTs()); diff --git a/dao/src/main/java/org/thingsboard/server/dao/sqlts/timescale/TimescaleTimeseriesDao.java b/dao/src/main/java/org/thingsboard/server/dao/sqlts/timescale/TimescaleTimeseriesDao.java index 6e9f9e61b8..c9daf1ec6d 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sqlts/timescale/TimescaleTimeseriesDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sqlts/timescale/TimescaleTimeseriesDao.java @@ -175,9 +175,7 @@ public class TimescaleTimeseriesDao extends AbstractSqlTimeseriesDao implements keyId, query.getStartTs(), query.getEndTs(), - PageRequest.of(0, query.getLimit(), - Sort.by(new Sort.Order(Sort.Direction.fromString(query.getOrder()), "ts").nullsNative()))); - ; + PageRequest.ofSize(query.getLimit()).withSort(Sort.Direction.fromString(query.getOrder()), "ts")); timescaleTsKvEntities.forEach(tsKvEntity -> tsKvEntity.setStrKey(strKey)); var tsKvEntries = DaoUtil.convertDataList(timescaleTsKvEntities); long lastTs = tsKvEntries.stream().map(TsKvEntry::getTs).max(Long::compare).orElse(query.getStartTs()); diff --git a/dao/src/main/java/org/thingsboard/server/dao/sqlts/timescale/TsKvTimescaleRepository.java b/dao/src/main/java/org/thingsboard/server/dao/sqlts/timescale/TsKvTimescaleRepository.java index 075cffcbc7..821833cad0 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sqlts/timescale/TsKvTimescaleRepository.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sqlts/timescale/TsKvTimescaleRepository.java @@ -31,14 +31,13 @@ import java.util.UUID; @TimescaleDBTsOrTsLatestDao public interface TsKvTimescaleRepository extends JpaRepository { - @Query("SELECT tskv FROM TimescaleTsKvEntity tskv WHERE tskv.entityId = :entityId " + - "AND tskv.key = :entityKey " + - "AND tskv.ts >= :startTs AND tskv.ts < :endTs") - List findAllWithLimit( - @Param("entityId") UUID entityId, - @Param("entityKey") int key, - @Param("startTs") long startTs, - @Param("endTs") long endTs, Pageable pageable); + @Query(value = "SELECT * FROM ts_kv WHERE entity_id = :entityId " + + "AND key = :entityKey AND ts >= :startTs AND ts < :endTs", nativeQuery = true) + List findAllWithLimit(@Param("entityId") UUID entityId, + @Param("entityKey") int key, + @Param("startTs") long startTs, + @Param("endTs") long endTs, + Pageable pageable); @Transactional @Modifying diff --git a/dao/src/main/java/org/thingsboard/server/dao/sqlts/ts/TsKvRepository.java b/dao/src/main/java/org/thingsboard/server/dao/sqlts/ts/TsKvRepository.java index 1d9817a5bd..66d619d0ea 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sqlts/ts/TsKvRepository.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sqlts/ts/TsKvRepository.java @@ -29,8 +29,15 @@ import java.util.UUID; public interface TsKvRepository extends JpaRepository { - @Query("SELECT tskv FROM TsKvEntity tskv WHERE tskv.entityId = :entityId " + - "AND tskv.key = :entityKey AND tskv.ts >= :startTs AND tskv.ts < :endTs") + /* + * Using native query to avoid adding 'nulls first' or 'nulls last' (ignoring spring.jpa.properties.hibernate.order_by.default_null_ordering) + * to the order so that index scan is done instead of full scan. + * + * Note: even when setting custom NullHandling for the Sort.Order for non-native queries, + * it will be ignored and default_null_ordering will be used + * */ + @Query(value = "SELECT * FROM ts_kv WHERE entity_id = :entityId " + + "AND key = :entityKey AND ts >= :startTs AND ts < :endTs ", nativeQuery = true) List findAllWithLimit(@Param("entityId") UUID entityId, @Param("entityKey") int key, @Param("startTs") long startTs, diff --git a/dao/src/main/resources/cassandra/schema-keyspace.cql b/dao/src/main/resources/cassandra/schema-keyspace.cql new file mode 100644 index 0000000000..8323f149ce --- /dev/null +++ b/dao/src/main/resources/cassandra/schema-keyspace.cql @@ -0,0 +1,21 @@ +-- +-- Copyright © 2016-2022 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. +-- + +CREATE KEYSPACE IF NOT EXISTS thingsboard +WITH replication = { + 'class' : 'SimpleStrategy', + 'replication_factor' : 1 +}; diff --git a/dao/src/main/resources/cassandra/schema-ts-latest.cql b/dao/src/main/resources/cassandra/schema-ts-latest.cql index 0de4e88158..10b3edb84e 100644 --- a/dao/src/main/resources/cassandra/schema-ts-latest.cql +++ b/dao/src/main/resources/cassandra/schema-ts-latest.cql @@ -14,12 +14,6 @@ -- limitations under the License. -- -CREATE KEYSPACE IF NOT EXISTS thingsboard -WITH replication = { - 'class' : 'SimpleStrategy', - 'replication_factor' : 1 -}; - CREATE TABLE IF NOT EXISTS thingsboard.ts_kv_latest_cf ( entity_type text, // (DEVICE, CUSTOMER, TENANT) entity_id timeuuid, diff --git a/dao/src/main/resources/cassandra/schema-ts.cql b/dao/src/main/resources/cassandra/schema-ts.cql index 9cb609add2..6ab099f841 100644 --- a/dao/src/main/resources/cassandra/schema-ts.cql +++ b/dao/src/main/resources/cassandra/schema-ts.cql @@ -14,12 +14,6 @@ -- limitations under the License. -- -CREATE KEYSPACE IF NOT EXISTS thingsboard -WITH replication = { - 'class' : 'SimpleStrategy', - 'replication_factor' : 1 -}; - CREATE TABLE IF NOT EXISTS thingsboard.ts_kv_cf ( entity_type text, // (DEVICE, CUSTOMER, TENANT) entity_id timeuuid, diff --git a/dao/src/main/resources/sql/schema-entities-idx.sql b/dao/src/main/resources/sql/schema-entities-idx.sql index 34862e5af3..e7586b17a8 100644 --- a/dao/src/main/resources/sql/schema-entities-idx.sql +++ b/dao/src/main/resources/sql/schema-entities-idx.sql @@ -50,6 +50,8 @@ CREATE INDEX IF NOT EXISTS idx_attribute_kv_by_key_and_last_update_ts ON attribu CREATE INDEX IF NOT EXISTS idx_audit_log_tenant_id_and_created_time ON audit_log(tenant_id, created_time DESC); +CREATE INDEX IF NOT EXISTS idx_edge_event_tenant_id_and_created_time ON edge_event(tenant_id, created_time DESC); + CREATE INDEX IF NOT EXISTS idx_rpc_tenant_id_device_id ON rpc(tenant_id, device_id); CREATE INDEX IF NOT EXISTS idx_device_external_id ON device(tenant_id, external_id); diff --git a/dao/src/main/resources/sql/schema-entities.sql b/dao/src/main/resources/sql/schema-entities.sql index 51df863ae5..73039274a6 100644 --- a/dao/src/main/resources/sql/schema-entities.sql +++ b/dao/src/main/resources/sql/schema-entities.sql @@ -719,7 +719,7 @@ CREATE TABLE IF NOT EXISTS edge ( ); CREATE TABLE IF NOT EXISTS edge_event ( - id uuid NOT NULL CONSTRAINT edge_event_pkey PRIMARY KEY, + id uuid NOT NULL, created_time bigint NOT NULL, edge_id uuid, edge_event_type varchar(255), @@ -729,7 +729,7 @@ CREATE TABLE IF NOT EXISTS edge_event ( body varchar(10000000), tenant_id uuid, ts bigint NOT NULL -); +) PARTITION BY RANGE(created_time); CREATE TABLE IF NOT EXISTS rpc ( id uuid NOT NULL CONSTRAINT rpc_pkey PRIMARY KEY, diff --git a/dao/src/test/java/org/thingsboard/server/dao/NoSqlDaoServiceTestSuite.java b/dao/src/test/java/org/thingsboard/server/dao/NoSqlDaoServiceTestSuite.java index 838be51349..e601218845 100644 --- a/dao/src/test/java/org/thingsboard/server/dao/NoSqlDaoServiceTestSuite.java +++ b/dao/src/test/java/org/thingsboard/server/dao/NoSqlDaoServiceTestSuite.java @@ -33,6 +33,7 @@ public class NoSqlDaoServiceTestSuite { public static CustomCassandraCQLUnit cassandraUnit = new CustomCassandraCQLUnit( Arrays.asList( + new ClassPathCQLDataSet("cassandra/schema-keyspace.cql", false, false), new ClassPathCQLDataSet("cassandra/schema-ts.cql", false, false), new ClassPathCQLDataSet("cassandra/schema-ts-latest.cql", false, false) ), diff --git a/docker/tb-js-executor.env b/docker/tb-js-executor.env index e080906549..1938449d53 100644 --- a/docker/tb-js-executor.env +++ b/docker/tb-js-executor.env @@ -3,4 +3,5 @@ LOGGER_LEVEL=info LOG_FOLDER=logs LOGGER_FILENAME=tb-js-executor-%DATE%.log DOCKER_MODE=true -SCRIPT_BODY_TRACE_FREQUENCY=1000 \ No newline at end of file +SCRIPT_BODY_TRACE_FREQUENCY=1000 +NODE_OPTIONS="--max-old-space-size=200" diff --git a/msa/js-executor/api/jsExecutor.models.ts b/msa/js-executor/api/jsExecutor.models.ts index 7a6b53cd8a..17407f4d50 100644 --- a/msa/js-executor/api/jsExecutor.models.ts +++ b/msa/js-executor/api/jsExecutor.models.ts @@ -56,7 +56,7 @@ export interface JsCompileResponse extends TbMessage { export interface JsInvokeResponse { success: boolean; - result: string; + result?: string; errorCode?: number; errorDetails?: string; } diff --git a/msa/js-executor/api/jsInvokeMessageProcessor.ts b/msa/js-executor/api/jsInvokeMessageProcessor.ts index 668cd61f50..52a337c74a 100644 --- a/msa/js-executor/api/jsInvokeMessageProcessor.ts +++ b/msa/js-executor/api/jsInvokeMessageProcessor.ts @@ -39,6 +39,7 @@ const TIMEOUT_ERROR = 2; const NOT_FOUND_ERROR = 3; const statFrequency = Number(config.get('script.stat_print_frequency')); +const memoryUsageTraceFrequency = Number(config.get('script.memory_usage_trace_frequency')); const scriptBodyTraceFrequency = Number(config.get('script.script_body_trace_frequency')); const useSandbox = config.get('script.use_sandbox') === 'true'; const maxActiveScripts = Number(config.get('script.max_active_scripts')); @@ -167,11 +168,15 @@ export class JsInvokeMessageProcessor { if (this.executedScriptsCounter % scriptBodyTraceFrequency == 0) { this.logger.info('[%s] Executing script body: [%s]', scriptId, invokeRequest.scriptBody); } + if (this.executedScriptsCounter % memoryUsageTraceFrequency == 0) { + this.logger.info('Current memory usage: [%s]', process.memoryUsage()); + } + this.getOrCompileScript(scriptId, invokeRequest.scriptBody).then( (script) => { this.executor.executeScript(script, invokeRequest.args, invokeRequest.timeout).then( - (result) => { - if (result.length <= maxResultSize) { + (result: string | undefined) => { + if (!result || result.length <= maxResultSize) { const invokeResponse = JsInvokeMessageProcessor.createInvokeResponse(result, true); this.logger.debug('[%s] Sending success invoke response, scriptId: [%s]', requestId, scriptId); this.sendResponse(requestId, responseTopic, headers, scriptId, undefined, invokeResponse); @@ -323,7 +328,7 @@ export class JsInvokeMessageProcessor { } } - private static createInvokeResponse(result: string, success: boolean, errorCode?: number, err?: any): JsInvokeResponse { + private static createInvokeResponse(result: string | undefined, success: boolean, errorCode?: number, err?: any): JsInvokeResponse { return { errorCode: errorCode, success: success, diff --git a/msa/js-executor/config/custom-environment-variables.yml b/msa/js-executor/config/custom-environment-variables.yml index b9c24c8d8d..2ebea4ccc1 100644 --- a/msa/js-executor/config/custom-environment-variables.yml +++ b/msa/js-executor/config/custom-environment-variables.yml @@ -75,6 +75,7 @@ logger: script: use_sandbox: "SCRIPT_USE_SANDBOX" + memory_usage_trace_frequency: "MEMORY_USAGE_TRACE_FREQUENCY" stat_print_frequency: "SCRIPT_STAT_PRINT_FREQUENCY" script_body_trace_frequency: "SCRIPT_BODY_TRACE_FREQUENCY" max_active_scripts: "MAX_ACTIVE_SCRIPTS" diff --git a/msa/js-executor/config/default.yml b/msa/js-executor/config/default.yml index 64829ef792..96f3401da5 100644 --- a/msa/js-executor/config/default.yml +++ b/msa/js-executor/config/default.yml @@ -64,6 +64,7 @@ logger: script: use_sandbox: "true" + memory_usage_trace_frequency: "1000" script_body_trace_frequency: "10000" stat_print_frequency: "10000" max_active_scripts: "1000" diff --git a/msa/js-executor/docker/start-js-executor.sh b/msa/js-executor/docker/start-js-executor.sh index 575f93c389..d30b62c145 100755 --- a/msa/js-executor/docker/start-js-executor.sh +++ b/msa/js-executor/docker/start-js-executor.sh @@ -27,4 +27,4 @@ source "${CONF_FOLDER}/${configfile}" cd ${pkg.installFolder} # This will forward this PID 1 to the node.js and forward SIGTERM for graceful shutdown as well -exec node server.js +exec node --no-compilation-cache server.js diff --git a/pom.xml b/pom.xml index 60e8783bca..5f5aa3703b 100755 --- a/pom.xml +++ b/pom.xml @@ -77,7 +77,7 @@ 3.5.5 3.21.9 1.42.1 - 2.4.23TB + 2.4.25TB 1.18.18 1.2.4 4.1.75.Final diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/edge/AbstractTbMsgPushNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/edge/AbstractTbMsgPushNode.java index 28dc64c068..f120276df3 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/edge/AbstractTbMsgPushNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/edge/AbstractTbMsgPushNode.java @@ -142,8 +142,11 @@ public abstract class AbstractTbMsgPushNode future = ctx.getEdgeEventService().saveAsync(edgeEvent); + Futures.addCallback(future, new FutureCallback<>() { + @Override + public void onSuccess(Void result) { + ctx.onEdgeEventUpdate(ctx.getTenantId(), edgeId); + ctx.tellSuccess(msg); + } + + @Override + public void onFailure(Throwable t) { + ctx.tellFailure(msg, t); + } + }, ctx.getDbCallbackExecutor()); + } } diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/rpc/TbSendRPCReplyNodeTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/rpc/TbSendRPCReplyNodeTest.java new file mode 100644 index 0000000000..095505aff3 --- /dev/null +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/rpc/TbSendRPCReplyNodeTest.java @@ -0,0 +1,119 @@ +/** + * Copyright © 2016-2022 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.rule.engine.rpc; + +import com.google.common.util.concurrent.SettableFuture; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mock; +import org.mockito.Mockito; +import org.mockito.junit.MockitoJUnitRunner; +import org.thingsboard.common.util.JacksonUtil; +import org.thingsboard.common.util.ListeningExecutor; +import org.thingsboard.rule.engine.api.RuleEngineRpcService; +import org.thingsboard.rule.engine.api.TbContext; +import org.thingsboard.rule.engine.api.TbNodeConfiguration; +import org.thingsboard.rule.engine.api.TbNodeException; +import org.thingsboard.server.common.data.DataConstants; +import org.thingsboard.server.common.data.id.DeviceId; +import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.msg.TbMsg; +import org.thingsboard.server.common.msg.TbMsgDataType; +import org.thingsboard.server.common.msg.TbMsgMetaData; +import org.thingsboard.server.common.msg.session.SessionMsgType; +import org.thingsboard.server.dao.edge.EdgeEventService; + +import java.util.UUID; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; + +@RunWith(MockitoJUnitRunner.class) +public class TbSendRPCReplyNodeTest { + + private static final String DUMMY_SERVICE_ID = "testServiceId"; + private static final int DUMMY_REQUEST_ID = 0; + private static final UUID DUMMY_SESSION_ID = UUID.randomUUID(); + private static final String DUMMY_DATA = "{\"key\":\"value\"}"; + + TbSendRPCReplyNode node; + + private final TenantId tenantId = TenantId.fromUUID(UUID.randomUUID()); + private final DeviceId deviceId = new DeviceId(UUID.randomUUID()); + + @Mock + private TbContext ctx; + + @Mock + private RuleEngineRpcService rpcService; + + @Mock + private EdgeEventService edgeEventService; + + @Mock + private ListeningExecutor listeningExecutor; + + @Before + public void setUp() throws TbNodeException { + node = new TbSendRPCReplyNode(); + TbSendRpcReplyNodeConfiguration config = new TbSendRpcReplyNodeConfiguration().defaultConfiguration(); + node.init(ctx, new TbNodeConfiguration(JacksonUtil.valueToTree(config))); + } + + @Test + public void sendReplyToTransport() { + Mockito.when(ctx.getRpcService()).thenReturn(rpcService); + + + TbMsg msg = TbMsg.newMsg(SessionMsgType.POST_TELEMETRY_REQUEST.name(), deviceId, getDefaultMetadata(), + TbMsgDataType.JSON, DUMMY_DATA, null, null); + + node.onMsg(ctx, msg); + + verify(rpcService).sendRpcReplyToDevice(DUMMY_SERVICE_ID, DUMMY_SESSION_ID, DUMMY_REQUEST_ID, DUMMY_DATA); + verify(edgeEventService, never()).saveAsync(any()); + } + + @Test + public void sendReplyToEdgeQueue() { + Mockito.when(ctx.getTenantId()).thenReturn(tenantId); + Mockito.when(ctx.getEdgeEventService()).thenReturn(edgeEventService); + Mockito.when(edgeEventService.saveAsync(any())).thenReturn(SettableFuture.create()); + Mockito.when(ctx.getDbCallbackExecutor()).thenReturn(listeningExecutor); + + TbMsgMetaData defaultMetadata = getDefaultMetadata(); + defaultMetadata.putValue(DataConstants.EDGE_ID, UUID.randomUUID().toString()); + defaultMetadata.putValue(DataConstants.DEVICE_ID, UUID.randomUUID().toString()); + TbMsg msg = TbMsg.newMsg(SessionMsgType.POST_TELEMETRY_REQUEST.name(), deviceId, defaultMetadata, + TbMsgDataType.JSON, DUMMY_DATA, null, null); + + node.onMsg(ctx, msg); + + verify(edgeEventService).saveAsync(any()); + verify(rpcService, never()).sendRpcReplyToDevice(DUMMY_SERVICE_ID, DUMMY_SESSION_ID, DUMMY_REQUEST_ID, DUMMY_DATA); + } + + private TbMsgMetaData getDefaultMetadata() { + TbSendRpcReplyNodeConfiguration config = new TbSendRpcReplyNodeConfiguration().defaultConfiguration(); + TbMsgMetaData metadata = new TbMsgMetaData(); + metadata.putValue(config.getServiceIdMetaDataAttribute(), DUMMY_SERVICE_ID); + metadata.putValue(config.getSessionIdMetaDataAttribute(), DUMMY_SESSION_ID.toString()); + metadata.putValue(config.getRequestIdMetaDataAttribute(), Integer.toString(DUMMY_REQUEST_ID)); + return metadata; + } +} diff --git a/tools/src/main/java/org/thingsboard/client/tools/migrator/README.md b/tools/src/main/java/org/thingsboard/client/tools/migrator/README.md index a468cbd3af..6995632514 100644 --- a/tools/src/main/java/org/thingsboard/client/tools/migrator/README.md +++ b/tools/src/main/java/org/thingsboard/client/tools/migrator/README.md @@ -63,7 +63,7 @@ Tool execution time depends on DB size, CPU resources and Disk throughput * Note that this this part works only for single node Cassandra Cluster. If you have more nodes - it is better to use `sstableloader` tool. 1. [Optional] install Cassandra on the instance -2. [Optional] Using `cqlsh` create `thingsboard` keyspace and requred tables from this files `schema-ts.cql` and `schema-ts-latest.cql` using `source` command +2. [Optional] Using `cqlsh` create `thingsboard` keyspace and requred tables from this files `schema-keyspace.cql`, `schema-ts.cql` and `schema-ts-latest.cql` using `source` command 3. Stop Cassandra 4. Look at `/var/lib/cassandra/data/thingsboard` and check for names of data folders 5. Copy generated SSTable files into cassandra data dir using next command: diff --git a/ui-ngx/src/app/core/http/attribute.service.ts b/ui-ngx/src/app/core/http/attribute.service.ts index bad6547b5d..984b1e607b 100644 --- a/ui-ngx/src/app/core/http/attribute.service.ts +++ b/ui-ngx/src/app/core/http/attribute.service.ts @@ -43,7 +43,7 @@ export class AttributeService { public deleteEntityAttributes(entityId: EntityId, attributeScope: AttributeScope, attributes: Array, config?: RequestConfig): Observable { - const keys = attributes.map(attribute => encodeURI(attribute.key)).join(','); + const keys = attributes.map(attribute => encodeURIComponent(attribute.key)).join(','); return this.http.delete(`/api/plugins/telemetry/${entityId.entityType}/${entityId.id}/${attributeScope}` + `?keys=${keys}`, defaultHttpOptionsFromConfig(config)); @@ -51,7 +51,7 @@ export class AttributeService { public deleteEntityTimeseries(entityId: EntityId, timeseries: Array, deleteAllDataForKeys = false, startTs?: number, endTs?: number, config?: RequestConfig): Observable { - const keys = timeseries.map(attribute => encodeURI(attribute.key)).join(','); + const keys = timeseries.map(attribute => encodeURIComponent(attribute.key)).join(','); let url = `/api/plugins/telemetry/${entityId.entityType}/${entityId.id}/timeseries/delete` + `?keys=${keys}&deleteAllDataForKeys=${deleteAllDataForKeys}`; if (isDefinedAndNotNull(startTs)) { diff --git a/ui-ngx/src/app/modules/home/components/dashboard-page/layout/manage-dashboard-layouts-dialog.component.html b/ui-ngx/src/app/modules/home/components/dashboard-page/layout/manage-dashboard-layouts-dialog.component.html index 98c5226228..74e7db937b 100644 --- a/ui-ngx/src/app/modules/home/components/dashboard-page/layout/manage-dashboard-layouts-dialog.component.html +++ b/ui-ngx/src/app/modules/home/components/dashboard-page/layout/manage-dashboard-layouts-dialog.component.html @@ -65,12 +65,15 @@ mat-raised-button color="primary" class="tb-layout-button" - [matTooltip]="layoutButtonText('main')" - matTooltipPosition="above" - matTooltipClass="tb-layout-button-tooltip" + (mouseover)="mainLayoutTooltip.show()" + (mouseleave)="mainLayoutTooltip.hide()" (click)="setFixedLayout('main')" [ngClass]="layoutButtonClass('main', true)"> - {{ (layoutsFormGroup.value.right ? 'layout.left' : 'layout.main') | translate }} + + {{ (layoutsFormGroup.value.right ? 'layout.left' : 'layout.main') | translate }} +
- {{ 'layout.right' | translate }} + + {{ 'layout.right' | translate }} +
{ - elementToDisable.disabled = false; - }, 250); - } - - if (this.layoutsFormGroup.get('type').value === LayoutWidthType.FIXED) { + if (this.layoutsFormGroup.get('type').value === LayoutWidthType.FIXED && this.layoutsFormGroup.get('right').value) { this.layoutsFormGroup.get('fixedLayout').setValue(layout); } }