WIP CachedRedisSqlTimeseriesLatestDao.java

application/src/main/resources/thingsboard.yml

@@ -489,6 +489,10 @@ cache:
   attributes:
     # make sure that if cache.type is 'redis' and cache.attributes.enabled is 'true' if you change 'maxmemory-policy' Redis config property to 'allkeys-lru', 'allkeys-lfu' or 'allkeys-random'
     enabled: "${CACHE_ATTRIBUTES_ENABLED:true}"
+  ts_latest:
+    # Will enable cache-aside strategy for SQL timeseries latest DAO.
+    # make sure that if cache.type is 'redis' and cache.ts_latest.enabled is 'true' if you change 'maxmemory-policy' Redis config property to 'allkeys-lru', 'allkeys-lfu' or 'allkeys-random'
+    enabled: "${CACHE_TS_LATEST_ENABLED:true}"
   specs:
     relations:
       timeToLiveInMinutes: "${CACHE_SPECS_RELATIONS_TTL:1440}" # Relations cache TTL
@@ -539,6 +543,9 @@ cache:
     attributes:
       timeToLiveInMinutes: "${CACHE_SPECS_ATTRIBUTES_TTL:1440}" # Attributes cache TTL
       maxSize: "${CACHE_SPECS_ATTRIBUTES_MAX_SIZE:100000}" # 0 means the cache is disabled
+    tsLatest:
+      timeToLiveInMinutes: "${CACHE_SPECS_TS_LATEST_TTL:1440}" # Timeseries latest cache TTL
+      maxSize: "${CACHE_SPECS_TS_LATEST_MAX_SIZE:100000}" # 0 means the cache is disabled
     userSessionsInvalidation:
       # The value of this TTL is ignored and replaced by the JWT refresh token expiration time
       timeToLiveInMinutes: "0"

build.sh

@@ -0,0 +1,57 @@
+#!/bin/bash
+#
+# Copyright © 2016-2024 The Thingsboard Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+set -e # exit on any error
+
+#PROJECTS="msa/tb-node,msa/web-ui,rule-engine-pe/rule-node-twilio-sms"
+PROJECTS=""
+
+if [ "$1" ]; then
+  PROJECTS="--projects $1"
+fi
+
+echo "Building and pushing [amd64,arm64] projects '$PROJECTS' ..."
+echo "HELP: usage ./build.sh [projects]"
+echo "HELP: example ./build.sh msa/web-ui,msa/web-report"
+java -version
+#echo "Cleaning ui-ngx/node_modules" && rm -rf ui-ngx/node_modules
+
+MAVEN_OPTS="-Xmx1024m" NODE_OPTIONS="--max_old_space_size=4096" DOCKER_CLI_EXPERIMENTAL=enabled DOCKER_BUILDKIT=0 \
+mvn -T2 license:format clean install -DskipTests \
+  $PROJECTS --also-make
+#   \
+#  -Dpush-docker-amd-arm-images
+#  -Ddockerfile.skip=false -Dpush-docker-image=true
+#  --offline
+#  --projects '!msa/web-report' --also-make
+
+# push all
+# mvn -T 1C license:format clean install -DskipTests -Ddockerfile.skip=false -Dpush-docker-image=true
+
+
+## Build and push AMD and ARM docker images using docker buildx
+## Reference to article how to setup docker miltiplatform build environment: https://medium.com/@artur.klauser/building-multi-architecture-docker-images-with-buildx-27d80f7e2408
+## install docker-ce from docker repo https://docs.docker.com/engine/install/ubuntu/
+# sudo apt install -y qemu-user-static binfmt-support
+# export DOCKER_CLI_EXPERIMENTAL=enabled
+# docker version
+# docker run --rm --privileged multiarch/qemu-user-static --reset -p yes
+# docker buildx create --name mybuilder
+# docker buildx use mybuilder
+# docker buildx inspect --bootstrap
+# docker buildx ls
+# mvn clean install -P push-docker-amd-arm-images

common/cache/src/main/java/org/thingsboard/server/cache/TbJavaRedisSerializer.java

@@ -0,0 +1,35 @@
+/**
+ * Copyright © 2016-2024 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.server.cache;
+
+import org.springframework.data.redis.serializer.RedisSerializer;
+import org.springframework.data.redis.serializer.SerializationException;
+
+public class TbJavaRedisSerializer<K, V> implements TbRedisSerializer<K, V> {
+
+    final RedisSerializer<Object> serializer = RedisSerializer.java();
+
+    @Override
+    public byte[] serialize(V value) throws SerializationException {
+        return serializer.serialize(value);
+    }
+
+    @Override
+    public V deserialize(K key, byte[] bytes) throws SerializationException {
+        return (V) serializer.deserialize(bytes);
+    }
+
+}

common/dao-api/src/main/java/org/thingsboard/server/dao/util/SqlTsLatestAnyDaoCachedRedis.java

@@ -0,0 +1,26 @@
+/**
+ * Copyright © 2016-2024 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.server.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_latest.type}'=='sql' || '${database.ts_latest.type}'=='timescale') && '${cache.ts_latest.enabled:false}'=='true' && '${cache.type:caffeine}'=='redis' ")
+public @interface SqlTsLatestAnyDaoCachedRedis {
+}

common/data/src/main/java/org/thingsboard/server/common/data/CacheConstants.java

@@ -34,6 +34,7 @@ public class CacheConstants {
 
     public static final String ASSET_PROFILE_CACHE = "assetProfiles";
     public static final String ATTRIBUTES_CACHE = "attributes";
+    public static final String TS_LATEST_CACHE = "tsLatest";
     public static final String USERS_SESSION_INVALIDATION_CACHE = "userSessionsInvalidation";
     public static final String OTA_PACKAGE_CACHE = "otaPackages";
     public static final String OTA_PACKAGE_DATA_CACHE = "otaPackagesData";

dao/src/main/java/org/thingsboard/server/dao/sqlts/CachedRedisSqlTimeseriesLatestDao.java

@@ -0,0 +1,104 @@
+/**
+ * Copyright © 2016-2024 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.server.dao.sqlts;
+
+import com.google.common.util.concurrent.ListenableFuture;
+import jakarta.annotation.PostConstruct;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.context.annotation.Primary;
+import org.springframework.stereotype.Component;
+import org.thingsboard.server.cache.TbTransactionalCache;
+import org.thingsboard.server.common.data.id.DeviceProfileId;
+import org.thingsboard.server.common.data.id.EntityId;
+import org.thingsboard.server.common.data.id.TenantId;
+import org.thingsboard.server.common.data.kv.DeleteTsKvQuery;
+import org.thingsboard.server.common.data.kv.TsKvEntry;
+import org.thingsboard.server.common.data.kv.TsKvLatestRemovingResult;
+import org.thingsboard.server.common.stats.DefaultCounter;
+import org.thingsboard.server.common.stats.StatsFactory;
+import org.thingsboard.server.dao.cache.CacheExecutorService;
+import org.thingsboard.server.dao.timeseries.TimeseriesLatestDao;
+import org.thingsboard.server.dao.timeseries.TsLatestCacheKey;
+import org.thingsboard.server.dao.util.SqlTsLatestAnyDaoCachedRedis;
+
+import java.util.List;
+import java.util.Optional;
+
+@Slf4j
+@Component
+@SqlTsLatestAnyDaoCachedRedis
+@RequiredArgsConstructor
+@Primary
+public class CachedRedisSqlTimeseriesLatestDao extends BaseAbstractSqlTimeseriesDao implements TimeseriesLatestDao {
+    public static final String STATS_NAME = "ts_latest.cache";
+
+    DefaultCounter hitCounter;
+    DefaultCounter missCounter;
+
+    final CacheExecutorService cacheExecutorService;
+    final SqlTimeseriesLatestDao sqlDao;
+    final StatsFactory statsFactory;
+    final TbTransactionalCache<TsLatestCacheKey, TsKvEntry> cache;
+
+    @PostConstruct
+    public void init() {
+        log.info("Init Redis cache-aside SQL Timeseries Latest DAO");
+        this.hitCounter = statsFactory.createDefaultCounter(STATS_NAME, "result", "hit");
+        this.missCounter = statsFactory.createDefaultCounter(STATS_NAME, "result", "miss");
+    }
+
+    @Override
+    public ListenableFuture<Void> saveLatest(TenantId tenantId, EntityId entityId, TsKvEntry tsKvEntry) {
+        return sqlDao.saveLatest(tenantId, entityId, tsKvEntry);
+    }
+
+    @Override
+    public ListenableFuture<TsKvLatestRemovingResult> removeLatest(TenantId tenantId, EntityId entityId, DeleteTsKvQuery query) {
+        return sqlDao.removeLatest(tenantId, entityId, query);
+    }
+
+    @Override
+    public ListenableFuture<Optional<TsKvEntry>> findLatestOpt(TenantId tenantId, EntityId entityId, String key) {
+        return sqlDao.findLatestOpt(tenantId, entityId, key);
+    }
+
+    @Override
+    public ListenableFuture<TsKvEntry> findLatest(TenantId tenantId, EntityId entityId, String key) {
+        return sqlDao.findLatest(tenantId, entityId, key);
+    }
+
+    @Override
+    public TsKvEntry findLatestSync(TenantId tenantId, EntityId entityId, String key) {
+        return sqlDao.findLatestSync(tenantId, entityId, key);
+    }
+
+    @Override
+    public ListenableFuture<List<TsKvEntry>> findAllLatest(TenantId tenantId, EntityId entityId) {
+        return sqlDao.findAllLatest(tenantId, entityId);
+    }
+
+    @Override
+    public List<String> findAllKeysByDeviceProfileId(TenantId tenantId, DeviceProfileId deviceProfileId) {
+        return sqlDao.findAllKeysByDeviceProfileId(tenantId, deviceProfileId);
+    }
+
+    @Override
+    public List<String> findAllKeysByEntityIds(TenantId tenantId, List<EntityId> entityIds) {
+        return sqlDao.findAllKeysByEntityIds(tenantId, entityIds);
+    }
+
+}

dao/src/main/java/org/thingsboard/server/dao/timeseries/TsLatestCacheKey.java

@@ -0,0 +1,40 @@
+/**
+ * Copyright © 2016-2024 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.server.dao.timeseries;
+
+import lombok.AllArgsConstructor;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import org.thingsboard.server.common.data.AttributeScope;
+import org.thingsboard.server.common.data.id.EntityId;
+
+import java.io.Serial;
+import java.io.Serializable;
+
+@EqualsAndHashCode
+@Getter
+@AllArgsConstructor
+public class TsLatestCacheKey implements Serializable {
+    private static final long serialVersionUID = 2024369077925351881L;
+
+    private final EntityId entityId;
+    private final String key;
+
+    @Override
+    public String toString() {
+        return "{" + entityId + "}" + key;
+    }
+}

dao/src/main/java/org/thingsboard/server/dao/timeseries/TsLatestRedisCache.java

@@ -0,0 +1,35 @@
+/**
+ * Copyright © 2016-2024 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.server.dao.timeseries;
+
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.stereotype.Service;
+import org.thingsboard.server.cache.CacheSpecsMap;
+import org.thingsboard.server.cache.RedisTbTransactionalCache;
+import org.thingsboard.server.cache.TBRedisCacheConfiguration;
+import org.thingsboard.server.cache.TbJavaRedisSerializer;
+import org.thingsboard.server.common.data.CacheConstants;
+import org.thingsboard.server.common.data.kv.TsKvEntry;
+
+@ConditionalOnProperty(prefix = "cache", value = "type", havingValue = "redis")
+@Service("TsLatestCache")
+public class TsLatestRedisCache extends RedisTbTransactionalCache<TsLatestCacheKey, TsKvEntry> {
+
+    public TsLatestRedisCache(TBRedisCacheConfiguration configuration, CacheSpecsMap cacheSpecsMap, RedisConnectionFactory connectionFactory) {
+        super(CacheConstants.TS_LATEST_CACHE, cacheSpecsMap, connectionFactory, configuration, new TbJavaRedisSerializer<>());    }
+
+}

dao/src/test/resources/application-test.properties

@@ -10,6 +10,7 @@ audit-log.sink.type=none
 #cache.type=caffeine # will be injected redis by RedisContainer or will be default (caffeine)
 cache.maximumPoolSize=16
 cache.attributes.enabled=true
+cache.ts_latest.enabled=true
 
 cache.specs.relations.timeToLiveInMinutes=1440
 cache.specs.relations.maxSize=100000
@@ -53,6 +54,9 @@ cache.specs.assetProfiles.maxSize=100000
 cache.specs.attributes.timeToLiveInMinutes=1440
 cache.specs.attributes.maxSize=100000
 
+cache.specs.tsLatest.timeToLiveInMinutes=1440
+cache.specs.tsLatest.maxSize=100000
+
 cache.specs.tokensOutdatageTime.timeToLiveInMinutes=1440
 cache.specs.tokensOutdatageTime.maxSize=100000
 

dao/src/test/resources/logback.xml

@@ -9,6 +9,7 @@
 
     <logger name="org.thingsboard.server.dao" level="WARN"/>
     <logger name="org.testcontainers" level="INFO" />
+    <logger name="org.thingsboard.server.dao.sqlts" level="INFO" />
 
     <!-- Log Hibernate SQL queries -->
     <!-- <logger name="org.hibernate.SQL" level="DEBUG"/> -->