Implement alarms removal by TTL

+/**
+ * Copyright © 2016-2021 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.ttl.alarms;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Service;
+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.tenant.profile.DefaultTenantProfileConfiguration;
+import org.thingsboard.server.common.msg.queue.ServiceType;
+import org.thingsboard.server.dao.alarm.AlarmDao;
+import org.thingsboard.server.dao.tenant.TbTenantProfileCache;
+import org.thingsboard.server.dao.tenant.TenantDao;
+import org.thingsboard.server.dao.util.PsqlDao;
+import org.thingsboard.server.queue.discovery.PartitionService;
+
+import java.util.Optional;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+
+@PsqlDao
+@Service
+@Slf4j
+@RequiredArgsConstructor
+public class AlarmsCleanUpService {
+    @Value("${sql.ttl.alarms.removal_batch_size}")
+    private Integer removalBatchSize;
+
+    private final AlarmDao alarmDao;
+    private final TenantDao tenantDao;
+    private final PartitionService partitionService;
+    private final TbTenantProfileCache tenantProfileCache;
+
+    @Scheduled(initialDelayString = "${sql.ttl.alarms.checking_interval}", fixedDelayString = "${sql.ttl.alarms.checking_interval}")
+    public void cleanUp() {
+        if (!partitionService.resolve(ServiceType.TB_CORE, TenantId.SYS_TENANT_ID, TenantId.SYS_TENANT_ID).isMyPartition()) {
+            return;
+        }
+
+        PageLink tenantsBatchRequest = new PageLink(65536, 0);
+        PageLink alarmsRemovalBatchRequest = new PageLink(removalBatchSize, 0);
+        long currentTime = System.currentTimeMillis();
+
+        PageData<TenantId> tenantsIds;
+        do {
+            tenantsIds = tenantDao.findTenantsIds(tenantsBatchRequest);
+            tenantsIds.getData().forEach(tenantId -> {
+                Optional<DefaultTenantProfileConfiguration> tenantProfileConfiguration = tenantProfileCache.get(tenantId).getProfileConfiguration();
+                if (tenantProfileConfiguration.isEmpty() || tenantProfileConfiguration.get().getAlarmsTtlDays() == 0) {
+                    return;
+                }
+
+                PageData<UUID> toRemove;
+                long outdatageTime = currentTime - TimeUnit.DAYS.toMillis(tenantProfileConfiguration.get().getAlarmsTtlDays());
+                log.info("Cleaning up outdated alarms for tenant {}", tenantId);
+                do {
+                    toRemove = alarmDao.findAlarmsIdsByEndTsBeforeAndTenantId(outdatageTime, tenantId, alarmsRemovalBatchRequest);
+                    alarmDao.removeAllByIds(toRemove.getData());
+                } while (toRemove.hasNext());
+            });
+
+            tenantsBatchRequest = tenantsBatchRequest.nextPageLink();
+        } while (tenantsIds.hasNext());
+    }
+
+}

@@ -273,6 +273,9 @@ sql:
         enabled: "${SQL_TTL_EDGE_EVENTS_ENABLED:true}"
         execution_interval_ms: "${SQL_TTL_EDGE_EVENTS_EXECUTION_INTERVAL:86400000}" # Number of milliseconds. The current value corresponds to one day
         edge_events_ttl: "${SQL_TTL_EDGE_EVENTS_TTL:2628000}" # Number of seconds. The current value corresponds to one month
+      alarms:
+        checking_interval: "${SQL_ALARMS_TTL_CHECKING_INTERVAL:7200000}" # Number of milliseconds. The current value corresponds to two hours
+        removal_batch_size: "${SQL_ALARMS_TTL_REMOVAL_BATCH_SIZE:200}" # To delete outdated alarms not all at once but in batches
 
 # Actor system parameters
 actors:

@@ -27,6 +27,7 @@ import org.thingsboard.server.common.data.validation.NoXss;
 
 import java.io.ByteArrayInputStream;
 import java.io.IOException;
+import java.util.Optional;
 
 import static org.thingsboard.server.common.data.SearchTextBasedWithAdditionalInfo.mapper;
 
@@ -92,6 +93,12 @@ public class TenantProfile extends SearchTextBased<TenantProfileId> implements H
         }
     }
 
+    public Optional<DefaultTenantProfileConfiguration> getProfileConfiguration() {
+        return Optional.ofNullable(getProfileData().getConfiguration())
+                .filter(profileConfiguration -> profileConfiguration instanceof DefaultTenantProfileConfiguration)
+                .map(profileConfiguration -> (DefaultTenantProfileConfiguration) profileConfiguration);
+    }
+
     public TenantProfileData createDefaultTenantProfileData() {
         TenantProfileData tpd = new TenantProfileData();
         tpd.setConfiguration(new DefaultTenantProfileConfiguration());

@@ -17,10 +17,11 @@ package org.thingsboard.server.common.data.page;
 
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
-import org.thingsboard.server.common.data.BaseData;
 
 import java.util.Collections;
 import java.util.List;
+import java.util.function.Function;
+import java.util.stream.Collectors;
 
 public class PageData<T> {
 
@@ -61,4 +62,8 @@ public class PageData<T> {
         return hasNext;
     }
 
+    public <D> PageData<D> mapData(Function<T, D> mapper) {
+        return new PageData<>(getData().stream().map(mapper).collect(Collectors.toList()), getTotalPages(), getTotalElements(), hasNext());
+    }
+
 }

@@ -53,6 +53,7 @@ public class DefaultTenantProfileConfiguration implements TenantProfileConfigura
     private long maxCreatedAlarms;
 
     private int defaultStorageTtlDays;
+    private int alarmsTtlDays;
 
     private double warnThreshold;
 

@@ -18,6 +18,7 @@ package org.thingsboard.server.dao;
 import com.google.common.util.concurrent.ListenableFuture;
 import org.thingsboard.server.common.data.id.TenantId;
 
+import java.util.Collection;
 import java.util.List;
 import java.util.UUID;
 
@@ -33,4 +34,6 @@ public interface Dao<T> {
 
     boolean removeById(TenantId tenantId, UUID id);
 
+    void removeAllByIds(Collection<UUID> ids);
+
 }

@@ -25,6 +25,7 @@ import org.thingsboard.server.common.data.id.CustomerId;
 import org.thingsboard.server.common.data.id.EntityId;
 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.query.AlarmData;
 import org.thingsboard.server.common.data.query.AlarmDataQuery;
 import org.thingsboard.server.dao.Dao;
@@ -54,4 +55,7 @@ public interface AlarmDao extends Dao<Alarm> {
                                                         AlarmDataQuery query, Collection<EntityId> orderedEntityIds);
 
     Set<AlarmSeverity> findAlarmSeverities(TenantId tenantId, EntityId entityId, Set<AlarmStatus> status);
+
+    PageData<UUID> findAlarmsIdsByEndTsBeforeAndTenantId(Long time, TenantId tenantId, PageLink pageLink);
+
 }

@@ -26,6 +26,7 @@ import org.thingsboard.server.dao.Dao;
 import org.thingsboard.server.dao.DaoUtil;
 import org.thingsboard.server.dao.model.BaseEntity;
 
+import java.util.Collection;
 import java.util.List;
 import java.util.Optional;
 import java.util.UUID;
@@ -87,6 +88,12 @@ public abstract class JpaAbstractDao<E extends BaseEntity<D>, D>
         return !getCrudRepository().existsById(id);
     }
 
+    @Transactional
+    public void removeAllByIds(Collection<UUID> ids) {
+        CrudRepository<E, UUID> repository = getCrudRepository();
+        ids.forEach(repository::deleteById);
+    }
+
     @Override
     public List<D> find(TenantId tenantId) {
         List<E> entities = Lists.newArrayList(getCrudRepository().findAll());

@@ -17,6 +17,7 @@ package org.thingsboard.server.dao.sql.alarm;
 
 import org.springframework.data.domain.Page;
 import org.springframework.data.domain.Pageable;
+import org.springframework.data.jpa.repository.Modifying;
 import org.springframework.data.jpa.repository.Query;
 import org.springframework.data.repository.CrudRepository;
 import org.springframework.data.repository.query.Param;
@@ -159,4 +160,8 @@ public interface AlarmRepository extends CrudRepository<AlarmEntity, UUID> {
                                            @Param("affectedEntityId") UUID affectedEntityId,
                                            @Param("affectedEntityType") String affectedEntityType,
                                            @Param("alarmStatuses") Set<AlarmStatus> alarmStatuses);
+
+    @Query("SELECT a.id FROM AlarmEntity a WHERE a.createdTime < :time AND a.endTs < :time")
+    Page<UUID> findAlarmsIdsByEndTsBefore(@Param("time") Long time, Pageable pageable);
+
 }

@@ -30,6 +30,7 @@ import org.thingsboard.server.common.data.id.CustomerId;
 import org.thingsboard.server.common.data.id.EntityId;
 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.query.AlarmData;
 import org.thingsboard.server.common.data.query.AlarmDataQuery;
 import org.thingsboard.server.dao.DaoUtil;
@@ -161,4 +162,9 @@ public class JpaAlarmDao extends JpaAbstractDao<AlarmEntity, Alarm> implements A
     public Set<AlarmSeverity> findAlarmSeverities(TenantId tenantId, EntityId entityId, Set<AlarmStatus> statuses) {
         return alarmRepository.findAlarmSeverities(tenantId.getId(), entityId.getId(), entityId.getEntityType().name(), statuses);
     }
+
+    @Override
+    public PageData<UUID> findAlarmsIdsByEndTsBeforeAndTenantId(Long time, TenantId tenantId, PageLink pageLink) {
+        return DaoUtil.pageToPageData(alarmRepository.findAlarmsIdsByEndTsBefore(time, DaoUtil.toPageable(pageLink)));
+    }
 }

@@ -74,4 +74,10 @@ public class JpaTenantDao extends JpaAbstractSearchTextDao<TenantEntity, Tenant>
                         Objects.toString(pageLink.getTextSearch(), ""),
                         DaoUtil.toPageable(pageLink, TenantInfoEntity.tenantInfoColumnMap)));
     }
+
+    @Override
+    public PageData<TenantId> findTenantsIds(PageLink pageLink) {
+        return DaoUtil.pageToPageData(tenantRepository.findTenantsIds(DaoUtil.toPageable(pageLink))).mapData(TenantId::new);
+    }
+
 }

@@ -50,4 +50,8 @@ public interface TenantRepository extends PagingAndSortingRepository<TenantEntit
     Page<TenantInfoEntity> findTenantInfoByRegionNextPage(@Param("region") String region,
                                                           @Param("textSearch") String textSearch,
                                                           Pageable pageable);
+
+    @Query("SELECT t.id FROM TenantEntity t")
+    Page<UUID> findTenantsIds(Pageable pageable);
+
 }

@@ -46,5 +46,7 @@ public interface TenantDao extends Dao<Tenant> {
     PageData<Tenant> findTenantsByRegion(TenantId tenantId, String region, PageLink pageLink);
 
     PageData<TenantInfo> findTenantInfosByRegion(TenantId tenantId, String region, PageLink pageLink);
-    
+
+    PageData<TenantId> findTenantsIds(PageLink pageLink);
+
 }