Add batch deletion support to removeRawUsageRecords (apache#12522)

* Add batch deletion support to `removeRawUsageRecords`

* Remove ORDER BY from batch expunge

Author: winterhazel

engine/schema/src/main/java/com/cloud/usage/dao/UsageDao.java

@@ -62,7 +62,7 @@ public interface UsageDao extends GenericDao<UsageVO, Long> {
 
     void saveUsageRecords(List<UsageVO> usageRecords);
 
-    void removeOldUsageRecords(int days);
+    void expungeAllOlderThan(int days, long limitPerQuery);
 
     UsageVO persistUsage(final UsageVO usage);
 

engine/schema/src/main/java/com/cloud/usage/dao/UsageDaoImpl.java

@@ -26,15 +26,16 @@
 import com.cloud.utils.db.Filter;
 import com.cloud.utils.db.GenericDaoBase;
 import com.cloud.utils.db.QueryBuilder;
+import com.cloud.utils.db.SearchBuilder;
 import com.cloud.utils.db.SearchCriteria;
 import com.cloud.utils.db.Transaction;
 import com.cloud.utils.db.TransactionCallback;
-import com.cloud.utils.db.TransactionCallbackNoReturn;
 import com.cloud.utils.db.TransactionLegacy;
 import com.cloud.utils.db.TransactionStatus;
 import com.cloud.utils.exception.CloudRuntimeException;
 
 import org.apache.cloudstack.acl.RoleType;
+import org.apache.commons.lang3.time.DateUtils;
 import org.springframework.stereotype.Component;
 
 import java.sql.PreparedStatement;
@@ -51,7 +52,6 @@
 public class UsageDaoImpl extends GenericDaoBase<UsageVO, Long> implements UsageDao {
     private static final String DELETE_ALL = "DELETE FROM cloud_usage";
     private static final String DELETE_ALL_BY_ACCOUNTID = "DELETE FROM cloud_usage WHERE account_id = ?";
-    private static final String DELETE_ALL_BY_INTERVAL = "DELETE FROM cloud_usage WHERE end_date < DATE_SUB(CURRENT_DATE(), INTERVAL ? DAY)";
     private static final String INSERT_ACCOUNT = "INSERT INTO cloud_usage.account (id, account_name, uuid, type, role_id, domain_id, removed, cleanup_needed) VALUES (?,?,?,?,?,?,?,?)";
     private static final String INSERT_USER_STATS = "INSERT INTO cloud_usage.user_statistics (id, data_center_id, account_id, public_ip_address, device_id, device_type, network_id, net_bytes_received,"
             + " net_bytes_sent, current_bytes_received, current_bytes_sent, agg_bytes_received, agg_bytes_sent) VALUES (?,?,?,?,?,?,?,?,?,?, ?, ?, ?)";
@@ -88,8 +88,12 @@ public class UsageDaoImpl extends GenericDaoBase<UsageVO, Long> implements Usage
 
     private static final String UPDATE_BUCKET_STATS = "UPDATE cloud_usage.bucket_statistics SET size=? WHERE id=?";
 
+    protected SearchBuilder<UsageVO> endDateLessThanSearch;
 
     public UsageDaoImpl() {
+        endDateLessThanSearch = createSearchBuilder();
+        endDateLessThanSearch.and("endDate", endDateLessThanSearch.entity().getEndDate(), SearchCriteria.Op.LT);
+        endDateLessThanSearch.done();
     }
 
     @Override
@@ -539,21 +543,20 @@ public void saveUsageRecords(List<UsageVO> usageRecords) {
     }
 
     @Override
-    public void removeOldUsageRecords(int days) {
-        Transaction.execute(TransactionLegacy.USAGE_DB, new TransactionCallbackNoReturn() {
-            @Override
-            public void doInTransactionWithoutResult(TransactionStatus status) {
-                TransactionLegacy txn = TransactionLegacy.currentTxn();
-                PreparedStatement pstmt = null;
-                try {
-                    pstmt = txn.prepareAutoCloseStatement(DELETE_ALL_BY_INTERVAL);
-                    pstmt.setLong(1, days);
-                    pstmt.executeUpdate();
-                } catch (Exception ex) {
-                    logger.error("error removing old cloud_usage records for interval: " + days);
-                }
-            }
-        });
+    public void expungeAllOlderThan(int days, long limitPerQuery) {
+        SearchCriteria<UsageVO> sc = endDateLessThanSearch.create();
+
+        Date limit = DateUtils.addDays(new Date(), -days);
+        sc.setParameters("endDate", limit);
+
+        TransactionLegacy txn = TransactionLegacy.open(TransactionLegacy.USAGE_DB);
+        try {
+            logger.debug("Removing all cloud_usage records older than [{}].", limit);
+            int totalExpunged = batchExpunge(sc, limitPerQuery);
+            logger.info("Removed a total of [{}] cloud_usage records older than [{}].", totalExpunged, limit);
+        } finally {
+            txn.close();
+        }
     }
 
     public UsageVO persistUsage(final UsageVO usage) {

framework/db/src/main/java/com/cloud/utils/db/Filter.java

@@ -57,7 +57,8 @@ public Filter(Class<?> clazz, String field, boolean ascending) {
     }
 
     public Filter(long limit) {
-        _orderBy = " ORDER BY RAND() LIMIT " + limit;
+        _orderBy = " ORDER BY RAND()";
+        _limit = limit;
     }
 
     public Filter(Long offset, Long limit) {

framework/db/src/main/java/com/cloud/utils/db/GenericDaoBase.java

@@ -1161,6 +1161,8 @@ protected void addFilter(final StringBuilder sql, final Filter filter) {
                 if (filter.getLimit() != null) {
                     sql.append(", ").append(filter.getLimit());
                 }
+            } else if (filter.getLimit() != null) {
+                sql.append(" LIMIT ").append(filter.getLimit());
             }
         }
     }
@@ -1322,7 +1324,7 @@ public int batchExpunge(final SearchCriteria<T> sc, final Long batchSize) {
         Filter filter = null;
         final long batchSizeFinal = ObjectUtils.defaultIfNull(batchSize, 0L);
         if (batchSizeFinal > 0) {
-            filter = new Filter(batchSizeFinal);
+            filter = new Filter(null, batchSizeFinal);
         }
         int expunged = 0;
         int currentExpunged = 0;

server/src/main/java/com/cloud/configuration/ConfigurationManagerImpl.java

@@ -527,7 +527,7 @@ public class ConfigurationManagerImpl extends ManagerBase implements Configurati
     public static final ConfigKey<Long> DELETE_QUERY_BATCH_SIZE = new ConfigKey<>("Advanced", Long.class, "delete.query.batch.size", "0",
             "Indicates the limit applied while deleting entries in bulk. With this, the delete query will apply the limit as many times as necessary," +
                     " to delete all the entries. This is advised when retaining several days of records, which can lead to slowness. <= 0 means that no limit will " +
-                    "be applied. Default value is 0. For now, this is used for deletion of vm & volume stats only.", true);
+                    "be applied. Default value is 0. For now, this is used for deletion of VM stats, volume stats, and usage records.", true);
 
     private static final String IOPS_READ_RATE = "IOPS Read";
     private static final String IOPS_WRITE_RATE = "IOPS Write";

server/src/main/java/com/cloud/usage/UsageServiceImpl.java

@@ -26,6 +26,7 @@
 import javax.inject.Inject;
 import javax.naming.ConfigurationException;
 
+import com.cloud.configuration.ConfigurationManagerImpl;
 import org.apache.cloudstack.api.command.admin.usage.GenerateUsageRecordsCmd;
 import org.apache.cloudstack.api.command.admin.usage.ListUsageRecordsCmd;
 import org.apache.cloudstack.api.command.admin.usage.RemoveRawUsageRecordsCmd;
@@ -489,7 +490,7 @@ public boolean removeRawUsageRecords(RemoveRawUsageRecordsCmd cmd) throws Invali
                     }
                 }
             }
-            _usageDao.removeOldUsageRecords(interval);
+            _usageDao.expungeAllOlderThan(interval, ConfigurationManagerImpl.DELETE_QUERY_BATCH_SIZE.value());
         } else {
             throw new InvalidParameterValueException("Invalid interval value. Interval to remove cloud_usage records should be greater than 0");
         }