diff --git a/application/src/main/data/upgrade/3.3.3/schema_event_ttl_procedure.sql b/application/src/main/data/upgrade/3.3.3/schema_event_ttl_procedure.sql new file mode 100644 index 0000000000..69fa207885 --- /dev/null +++ b/application/src/main/data/upgrade/3.3.3/schema_event_ttl_procedure.sql @@ -0,0 +1,50 @@ +-- +-- 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. +-- + + +DROP PROCEDURE IF EXISTS public.cleanup_events_by_ttl(bigint, bigint, bigint); + +CREATE OR REPLACE PROCEDURE cleanup_events_by_ttl( + IN regular_events_start_ts bigint, + IN regular_events_end_ts bigint, + IN debug_events_start_ts bigint, + IN debug_events_end_ts bigint, + INOUT deleted bigint) + LANGUAGE plpgsql AS +$$ +DECLARE + ttl_deleted_count bigint DEFAULT 0; + debug_ttl_deleted_count bigint DEFAULT 0; +BEGIN + IF regular_events_start_ts > 0 AND regular_events_end_ts > 0 THEN + EXECUTE format( + 'WITH deleted AS (DELETE FROM event WHERE id in (SELECT id from event WHERE ts > %L::bigint AND ts < %L::bigint AND ' || + '(event_type != %L::varchar AND event_type != %L::varchar AND event_type != %L::varchar AND event_type != %L::varchar)) RETURNING *) ' || + 'SELECT count(*) FROM deleted', regular_events_start_ts, regular_events_end_ts, + 'DEBUG_RULE_NODE', 'DEBUG_RULE_CHAIN', 'DEBUG_CONVERTER', 'DEBUG_INTEGRATION') into ttl_deleted_count; + END IF; + IF debug_events_start_ts > 0 AND debug_events_end_ts > 0 THEN + EXECUTE format( + 'WITH deleted AS (DELETE FROM event WHERE id in (SELECT id from event WHERE ts > %L::bigint AND ts < %L::bigint AND ' || + '(event_type = %L::varchar OR event_type = %L::varchar OR event_type = %L::varchar OR event_type = %L::varchar)) RETURNING *) ' || + 'SELECT count(*) FROM deleted', debug_events_start_ts, debug_events_end_ts, + 'DEBUG_RULE_NODE', 'DEBUG_RULE_CHAIN', 'DEBUG_CONVERTER', 'DEBUG_INTEGRATION') into debug_ttl_deleted_count; + END IF; + RAISE NOTICE 'Events removed by ttl: %', ttl_deleted_count; + RAISE NOTICE 'Debug Events removed by ttl: %', debug_ttl_deleted_count; + deleted := ttl_deleted_count + debug_ttl_deleted_count; +END +$$; 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 22e0869bd1..e70f5df850 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 @@ -521,6 +521,10 @@ public class SqlDatabaseUpgradeService implements DatabaseEntitiesUpgradeService } catch (Exception ignored) { } + log.info("Updating TTL cleanup procedure for the event table..."); + schemaUpdateFile = Paths.get(installScripts.getDataDir(), "upgrade", "3.3.3", "schema_event_ttl_procedure.sql"); + loadSql(schemaUpdateFile, conn); + log.info("Updating schema settings..."); conn.createStatement().execute("UPDATE tb_schema_settings SET schema_version = 3003004;"); log.info("Schema updated."); diff --git a/application/src/main/java/org/thingsboard/server/service/ttl/EventsCleanUpService.java b/application/src/main/java/org/thingsboard/server/service/ttl/EventsCleanUpService.java index 2c2f82bc50..1fd3892f39 100644 --- a/application/src/main/java/org/thingsboard/server/service/ttl/EventsCleanUpService.java +++ b/application/src/main/java/org/thingsboard/server/service/ttl/EventsCleanUpService.java @@ -22,7 +22,8 @@ import org.springframework.stereotype.Service; import org.thingsboard.server.dao.event.EventService; import org.thingsboard.server.queue.discovery.PartitionService; import org.thingsboard.server.queue.util.TbCoreComponent; -import org.thingsboard.server.service.ttl.AbstractCleanUpService; + +import java.util.concurrent.TimeUnit; @TbCoreComponent @Slf4j @@ -33,10 +34,13 @@ public class EventsCleanUpService extends AbstractCleanUpService { "#{T(org.apache.commons.lang3.RandomUtils).nextLong(0, ${sql.ttl.events.execution_interval_ms})}"; @Value("${sql.ttl.events.events_ttl}") - private long ttl; + private long ttlInSec; @Value("${sql.ttl.events.debug_events_ttl}") - private long debugTtl; + private long debugTtlInSec; + + @Value("${sql.ttl.events.execution_interval_ms}") + private long executionIntervalInMs; @Value("${sql.ttl.events.enabled}") private boolean ttlTaskExecutionEnabled; @@ -51,7 +55,27 @@ public class EventsCleanUpService extends AbstractCleanUpService { @Scheduled(initialDelayString = RANDOM_DELAY_INTERVAL_MS_EXPRESSION, fixedDelayString = "${sql.ttl.events.execution_interval_ms}") public void cleanUp() { if (ttlTaskExecutionEnabled && isSystemTenantPartitionMine()) { - eventService.cleanupEvents(ttl, debugTtl); + long ts = System.currentTimeMillis(); + long regularEventStartTs; + long regularEventEndTs; + long debugEventStartTs; + long debugEventEndTs; + + if (ttlInSec > 0) { + regularEventEndTs = ts - TimeUnit.SECONDS.toMillis(ttlInSec); + regularEventStartTs = regularEventEndTs - 2 * executionIntervalInMs; + } else { + regularEventStartTs = regularEventEndTs = 0; + } + + if (debugTtlInSec > 0) { + debugEventEndTs = ts - TimeUnit.SECONDS.toMillis(debugTtlInSec); + debugEventStartTs = debugEventEndTs - 2 * executionIntervalInMs; + } else { + debugEventStartTs = debugEventEndTs = 0; + } + + eventService.cleanupEvents(regularEventStartTs, regularEventEndTs, debugEventStartTs, debugEventEndTs); } } diff --git a/application/src/test/java/org/thingsboard/server/service/ttl/EventsCleanUpServiceTest.java b/application/src/test/java/org/thingsboard/server/service/ttl/EventsCleanUpServiceTest.java index acfe122d17..0d115ded7d 100644 --- a/application/src/test/java/org/thingsboard/server/service/ttl/EventsCleanUpServiceTest.java +++ b/application/src/test/java/org/thingsboard/server/service/ttl/EventsCleanUpServiceTest.java @@ -42,7 +42,6 @@ public class EventsCleanUpServiceTest { public void givenInterval_whenRandomDelay_ThenDelayInInterval() { log.info("randomDelay {}", randomDelayMs); log.info("executionIntervalMs {}", executionIntervalMs); - assertThat(executionIntervalMs, is(2220000L)); assertThat(randomDelayMs, greaterThanOrEqualTo(0L)); assertThat(randomDelayMs, lessThanOrEqualTo(executionIntervalMs)); } diff --git a/common/dao-api/src/main/java/org/thingsboard/server/dao/event/EventService.java b/common/dao-api/src/main/java/org/thingsboard/server/dao/event/EventService.java index 7deb9cf2e6..f7ddfbc8de 100644 --- a/common/dao-api/src/main/java/org/thingsboard/server/dao/event/EventService.java +++ b/common/dao-api/src/main/java/org/thingsboard/server/dao/event/EventService.java @@ -48,6 +48,6 @@ public interface EventService { void removeEvents(TenantId tenantId, EntityId entityId, EventFilter eventFilter, Long startTime, Long endTime); - void cleanupEvents(long ttl, long debugTtl); + void cleanupEvents(long regularEventStartTs, long regularEventEndTs, long debugEventStartTs, long debugEventEndTs); } diff --git a/dao/src/main/java/org/thingsboard/server/dao/event/BaseEventService.java b/dao/src/main/java/org/thingsboard/server/dao/event/BaseEventService.java index a94b71405a..b80de363f1 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/event/BaseEventService.java +++ b/dao/src/main/java/org/thingsboard/server/dao/event/BaseEventService.java @@ -141,8 +141,8 @@ public class BaseEventService implements EventService { } @Override - public void cleanupEvents(long ttl, long debugTtl) { - eventDao.cleanupEvents(ttl, debugTtl); + public void cleanupEvents(long regularEventStartTs, long regularEventEndTs, long debugEventStartTs, long debugEventEndTs) { + eventDao.cleanupEvents(regularEventStartTs, regularEventEndTs, debugEventStartTs, debugEventEndTs); } private DataValidator eventValidator = diff --git a/dao/src/main/java/org/thingsboard/server/dao/event/EventDao.java b/dao/src/main/java/org/thingsboard/server/dao/event/EventDao.java index bb143fd248..e0f1160db1 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/event/EventDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/event/EventDao.java @@ -104,8 +104,10 @@ public interface EventDao extends Dao { /** * Executes stored procedure to cleanup old events. Uses separate ttl for debug and other events. - * @param otherEventsTtl the ttl for events in seconds - * @param debugEventsTtl the ttl for debug events in seconds + * @param regularEventStartTs the start time of the interval to use to delete non debug events + * @param regularEventEndTs the end time of the interval to use to delete non debug events + * @param debugEventStartTs the start time of the interval to use to delete debug events + * @param debugEventEndTs the end time of the interval to use to delete debug events */ - void cleanupEvents(long otherEventsTtl, long debugEventsTtl); + void cleanupEvents(long regularEventStartTs, long regularEventEndTs, long debugEventStartTs, long debugEventEndTs); } diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/event/EventCleanupRepository.java b/dao/src/main/java/org/thingsboard/server/dao/sql/event/EventCleanupRepository.java index 5562443474..757d69f794 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/event/EventCleanupRepository.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/event/EventCleanupRepository.java @@ -17,6 +17,6 @@ package org.thingsboard.server.dao.sql.event; public interface EventCleanupRepository { - void cleanupEvents(long otherEventsTtl, long debugEventsTtl); + void cleanupEvents(long regularEventStartTs, long regularEventEndTs, long debugEventStartTs, long debugEventEndTs); } diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/event/HsqlEventCleanupRepository.java b/dao/src/main/java/org/thingsboard/server/dao/sql/event/HsqlEventCleanupRepository.java deleted file mode 100644 index f7acc8af77..0000000000 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/event/HsqlEventCleanupRepository.java +++ /dev/null @@ -1,54 +0,0 @@ -/** - * 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.sql.event; - -import lombok.extern.slf4j.Slf4j; -import org.springframework.stereotype.Repository; -import org.thingsboard.server.dao.sql.JpaAbstractDaoListeningExecutorService; -import org.thingsboard.server.dao.util.HsqlDao; - -import java.sql.Connection; -import java.sql.PreparedStatement; -import java.sql.SQLException; -import java.util.concurrent.TimeUnit; - -@Slf4j -@HsqlDao -@Repository -public class HsqlEventCleanupRepository extends JpaAbstractDaoListeningExecutorService implements EventCleanupRepository { - - @Override - public void cleanupEvents(long otherEventsTtl, long debugEventsTtl) { - long otherExpirationTime = System.currentTimeMillis() - otherEventsTtl * 1000; - try (Connection connection = dataSource.getConnection(); - PreparedStatement stmt = connection.prepareStatement("DELETE FROM event WHERE ts < ? AND event_type != 'DEBUG_RULE_NODE' AND event_type != 'DEBUG_RULE_CHAIN'")) { - stmt.setLong(1, otherExpirationTime); - stmt.setQueryTimeout((int) TimeUnit.HOURS.toSeconds(1)); - stmt.execute(); - } catch (SQLException e) { - log.error("SQLException occurred during events TTL task execution ", e); - } - long debugExpirationTime = System.currentTimeMillis() - debugEventsTtl * 1000; - try (Connection connection = dataSource.getConnection(); - PreparedStatement stmt = connection.prepareStatement("DELETE FROM event WHERE ts < ? AND (event_type = 'DEBUG_RULE_NODE' OR event_type = 'DEBUG_RULE_CHAIN')")) { - stmt.setLong(1, debugExpirationTime); - stmt.setQueryTimeout((int) TimeUnit.HOURS.toSeconds(1)); - stmt.execute(); - } catch (SQLException e) { - log.error("SQLException occurred during events TTL task execution ", e); - } - } -} diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/event/JpaBaseEventDao.java b/dao/src/main/java/org/thingsboard/server/dao/sql/event/JpaBaseEventDao.java index d86a80233e..8398e54dde 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/event/JpaBaseEventDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/event/JpaBaseEventDao.java @@ -259,9 +259,9 @@ public class JpaBaseEventDao extends JpaAbstractDao implemen } @Override - public void cleanupEvents(long otherEventsTtl, long debugEventsTtl) { - log.info("Going to cleanup old events using debug events ttl: {}s and other events ttl: {}s", debugEventsTtl, otherEventsTtl); - eventCleanupRepository.cleanupEvents(otherEventsTtl, debugEventsTtl); + public void cleanupEvents(long regularEventStartTs, long regularEventEndTs, long debugEventStartTs, long debugEventEndTs) { + log.info("Going to cleanup old events. Interval for regular events: [{}:{}], for debug events: [{}:{}]", regularEventStartTs, regularEventEndTs, debugEventStartTs, debugEventEndTs); + eventCleanupRepository.cleanupEvents(regularEventStartTs, regularEventEndTs, debugEventStartTs, debugEventEndTs); } public Optional save(EventEntity entity, boolean ifNotExists) { diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/event/PsqlEventCleanupRepository.java b/dao/src/main/java/org/thingsboard/server/dao/sql/event/PsqlEventCleanupRepository.java index a50a2e7864..664a148948 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/event/PsqlEventCleanupRepository.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/event/PsqlEventCleanupRepository.java @@ -32,12 +32,14 @@ import java.util.concurrent.TimeUnit; public class PsqlEventCleanupRepository extends JpaAbstractDaoListeningExecutorService implements EventCleanupRepository { @Override - public void cleanupEvents(long otherEventsTtl, long debugEventsTtl) { + public void cleanupEvents(long regularEventStartTs, long regularEventEndTs, long debugEventStartTs, long debugEventEndTs) { try (Connection connection = dataSource.getConnection(); - PreparedStatement stmt = connection.prepareStatement("call cleanup_events_by_ttl(?,?,?)")) { - stmt.setLong(1, otherEventsTtl); - stmt.setLong(2, debugEventsTtl); - stmt.setLong(3, 0); + PreparedStatement stmt = connection.prepareStatement("call cleanup_events_by_ttl(?,?,?,?,?)")) { + stmt.setLong(1, regularEventStartTs); + stmt.setLong(2, regularEventEndTs); + stmt.setLong(3, debugEventStartTs); + stmt.setLong(4, debugEventEndTs); + stmt.setLong(5, 0); stmt.setQueryTimeout((int) TimeUnit.HOURS.toSeconds(1)); stmt.execute(); printWarnings(stmt); diff --git a/dao/src/main/resources/sql/schema-entities.sql b/dao/src/main/resources/sql/schema-entities.sql index cdae84fe15..2b17f4f21e 100644 --- a/dao/src/main/resources/sql/schema-entities.sql +++ b/dao/src/main/resources/sql/schema-entities.sql @@ -632,24 +632,31 @@ CREATE TABLE IF NOT EXISTS rpc ( status varchar(255) NOT NULL ); -CREATE OR REPLACE PROCEDURE cleanup_events_by_ttl(IN ttl bigint, IN debug_ttl bigint, INOUT deleted bigint) +CREATE OR REPLACE PROCEDURE cleanup_events_by_ttl( + IN regular_events_start_ts bigint, + IN regular_events_end_ts bigint, + IN debug_events_start_ts bigint, + IN debug_events_end_ts bigint, + INOUT deleted bigint) LANGUAGE plpgsql AS $$ DECLARE - ttl_ts bigint; - debug_ttl_ts bigint; ttl_deleted_count bigint DEFAULT 0; debug_ttl_deleted_count bigint DEFAULT 0; BEGIN - IF ttl > 0 THEN - ttl_ts := (EXTRACT(EPOCH FROM current_timestamp) * 1000 - ttl::bigint * 1000)::bigint; + IF regular_events_start_ts > 0 AND regular_events_end_ts > 0 THEN EXECUTE format( - 'WITH deleted AS (DELETE FROM event WHERE ts < %L::bigint AND (event_type != %L::varchar AND event_type != %L::varchar) RETURNING *) SELECT count(*) FROM deleted', ttl_ts, 'DEBUG_RULE_NODE', 'DEBUG_RULE_CHAIN') into ttl_deleted_count; + 'WITH deleted AS (DELETE FROM event WHERE id in (SELECT id from event WHERE ts > %L::bigint AND ts < %L::bigint AND ' || + '(event_type != %L::varchar AND event_type != %L::varchar AND event_type != %L::varchar AND event_type != %L::varchar)) RETURNING *) ' || + 'SELECT count(*) FROM deleted', regular_events_start_ts, regular_events_end_ts, + 'DEBUG_RULE_NODE', 'DEBUG_RULE_CHAIN', 'DEBUG_CONVERTER', 'DEBUG_INTEGRATION') into ttl_deleted_count; END IF; - IF debug_ttl > 0 THEN - debug_ttl_ts := (EXTRACT(EPOCH FROM current_timestamp) * 1000 - debug_ttl::bigint * 1000)::bigint; + IF debug_events_start_ts > 0 AND debug_events_end_ts > 0 THEN EXECUTE format( - 'WITH deleted AS (DELETE FROM event WHERE ts < %L::bigint AND (event_type = %L::varchar OR event_type = %L::varchar) RETURNING *) SELECT count(*) FROM deleted', debug_ttl_ts, 'DEBUG_RULE_NODE', 'DEBUG_RULE_CHAIN') into debug_ttl_deleted_count; + 'WITH deleted AS (DELETE FROM event WHERE id in (SELECT id from event WHERE ts > %L::bigint AND ts < %L::bigint AND ' || + '(event_type = %L::varchar OR event_type = %L::varchar OR event_type = %L::varchar OR event_type = %L::varchar)) RETURNING *) ' || + 'SELECT count(*) FROM deleted', debug_events_start_ts, debug_events_end_ts, + 'DEBUG_RULE_NODE', 'DEBUG_RULE_CHAIN', 'DEBUG_CONVERTER', 'DEBUG_INTEGRATION') into debug_ttl_deleted_count; END IF; RAISE NOTICE 'Events removed by ttl: %', ttl_deleted_count; RAISE NOTICE 'Debug Events removed by ttl: %', debug_ttl_deleted_count;