Changed to allow setting the maximum number of records to be deleted …

…and the interval between deletions (#2837)

Signed-off-by: takumats <[email protected]>

libs/java/server_aws_common/src/main/java/io/athenz/server/aws/common/cert/impl/DynamoDBCertRecordStoreConnection.java

@@ -282,7 +282,7 @@ public boolean deleteX509CertRecord(String provider, String instanceId, String s
     }
 
     @Override
-    public int deleteExpiredX509CertRecords(int expiryTimeMins) {
+    public int deleteExpiredX509CertRecords(int expiryTimeMins, int limit) {
 
         // with dynamo db there is no need to manually expunge expired
         // record since we have the TTL option enabled for our table,

libs/java/server_aws_common/src/main/java/io/athenz/server/aws/common/cert/impl/DynamoDBSSHRecordStoreConnection.java

@@ -178,7 +178,7 @@ public boolean deleteSSHCertRecord(String instanceId, String service) {
     }
 
     @Override
-    public int deleteExpiredSSHCertRecords(int expiryTimeMins) {
+    public int deleteExpiredSSHCertRecords(int expiryTimeMins, int limit) {
 
         // with dynamo db there is no need to manually expunge expired
         // record since we have the TTL option enabled for our table,

libs/java/server_aws_common/src/test/java/io/athenz/server/aws/common/cert/impl/DynamoDBCertRecordStoreConnectionTest.java

@@ -482,8 +482,8 @@ public void testDeleteX509RecordException() {
     @Test
     public void testdeleteExpiredX509CertRecords() {
         DynamoDBCertRecordStoreConnection dbConn = getDBConnection();
-        assertEquals(dbConn.deleteExpiredX509CertRecords(100), 0);
-        assertEquals(dbConn.deleteExpiredX509CertRecords(100000), 0);
+        assertEquals(dbConn.deleteExpiredX509CertRecords(100, 0), 0);
+        assertEquals(dbConn.deleteExpiredX509CertRecords(100000, 0), 0);
         dbConn.close();
     }
 

libs/java/server_aws_common/src/test/java/io/athenz/server/aws/common/cert/impl/DynamoDBSSHRecordStoreConnectionTest.java

@@ -258,8 +258,8 @@ public void testDeleteSSHRecordException() {
     @Test
     public void testDeleteExpiredSSHCertRecords() {
         DynamoDBSSHRecordStoreConnection dbConn = new DynamoDBSSHRecordStoreConnection(dynamoDB, tableName);
-        assertEquals(dbConn.deleteExpiredSSHCertRecords(100), 0);
-        assertEquals(dbConn.deleteExpiredSSHCertRecords(100000), 0);
+        assertEquals(dbConn.deleteExpiredSSHCertRecords(100, 0), 0);
+        assertEquals(dbConn.deleteExpiredSSHCertRecords(100000, 0), 0);
         dbConn.close();
     }
 }

libs/java/server_common/src/main/java/com/yahoo/athenz/common/server/cert/CertRecordStoreConnection.java

@@ -70,9 +70,10 @@ public interface CertRecordStoreConnection extends Closeable {
      * considered expired if it hasn't been updated within the
      * specified number of minutes
      * @param expiryTimeMins expiry time in minutes
+     * @param limit delete limit
      * @return number of records deleted
      */
-    int deleteExpiredX509CertRecords(int expiryTimeMins) throws ServerResourceException;
+    int deleteExpiredX509CertRecords(int expiryTimeMins, int limit) throws ServerResourceException;
 
     /**
      * Return all certificate records that failed to refresh after updating them with the current notification time and server.

libs/java/server_common/src/main/java/com/yahoo/athenz/common/server/cert/impl/JDBCCertRecordStoreConnection.java

@@ -51,6 +51,8 @@ public class JDBCCertRecordStoreConnection implements CertRecordStoreConnection
             "WHERE provider=? AND instanceId=? AND service=?;";
     private static final String SQL_DELETE_EXPIRED_X509_RECORDS = "DELETE FROM certificates " +
             "WHERE currentTime < ADDDATE(NOW(), INTERVAL -? MINUTE);";
+    private static final String SQL_DELETE_EXPIRED_X509_RECORDS_WITH_LIMIT = "DELETE FROM certificates " +
+            "WHERE currentTime < ADDDATE(NOW(), INTERVAL -? MINUTE) LIMIT ?;";
 
     // Get all records that didn't refresh and update notification time.
     // Query explanation:
@@ -277,7 +279,7 @@ public boolean deleteX509CertRecord(String provider, String instanceId, String s
     }
 
     @Override
-    public int deleteExpiredX509CertRecords(int expiryTimeMins) throws ServerResourceException {
+    public int deleteExpiredX509CertRecords(int expiryTimeMins, int limit) throws ServerResourceException {
 
         int affectedRows;
         final String caller = "deleteExpiredX509CertRecords";
@@ -287,9 +289,16 @@ public int deleteExpiredX509CertRecords(int expiryTimeMins) throws ServerResourc
         if (expiryTimeMins <= 0) {
             return 0;
         }
-
-        try (PreparedStatement ps = con.prepareStatement(SQL_DELETE_EXPIRED_X509_RECORDS)) {
+
+        String sql = SQL_DELETE_EXPIRED_X509_RECORDS;
+        if (limit > 0) {
+            sql = SQL_DELETE_EXPIRED_X509_RECORDS_WITH_LIMIT;
+        }
+        try (PreparedStatement ps = con.prepareStatement(sql)) {
             ps.setInt(1, expiryTimeMins);
+            if (limit > 0) {
+                ps.setInt(2, limit);
+            }
             affectedRows = executeUpdate(ps, caller);
         } catch (SQLException ex) {
             throw sqlError(ex, caller);

libs/java/server_common/src/main/java/com/yahoo/athenz/common/server/cert/impl/JDBCSSHRecordStoreConnection.java

@@ -40,6 +40,8 @@ public class JDBCSSHRecordStoreConnection implements SSHRecordStoreConnection {
             "WHERE instanceId=? AND service=?;";
     private static final String SQL_DELETE_EXPIRED_X509_RECORDS = "DELETE FROM ssh_certificates " +
             "WHERE issueTime < ADDDATE(NOW(), INTERVAL -? MINUTE);";
+    private static final String SQL_DELETE_EXPIRED_X509_RECORDS_WITH_LIMIT = "DELETE FROM ssh_certificates " +
+            "WHERE issueTime < ADDDATE(NOW(), INTERVAL -? MINUTE) LIMIT ?;";
 
     public static final String DB_COLUMN_SERVICE        = "service";
     public static final String DB_COLUMN_CLIENT_IP      = "clientIP";
@@ -190,7 +192,7 @@ public boolean deleteSSHCertRecord(String instanceId, String service) throws Ser
     }
 
     @Override
-    public int deleteExpiredSSHCertRecords(int expiryTimeMins) throws ServerResourceException {
+    public int deleteExpiredSSHCertRecords(int expiryTimeMins, int limit) throws ServerResourceException {
 
         int affectedRows;
         final String caller = "deleteExpiredSSHCertRecords";
@@ -200,9 +202,16 @@ public int deleteExpiredSSHCertRecords(int expiryTimeMins) throws ServerResource
         if (expiryTimeMins <= 0) {
             return 0;
         }
-
-        try (PreparedStatement ps = con.prepareStatement(SQL_DELETE_EXPIRED_X509_RECORDS)) {
+
+        String sql = SQL_DELETE_EXPIRED_X509_RECORDS;
+        if (limit > 0) {
+            sql = SQL_DELETE_EXPIRED_X509_RECORDS_WITH_LIMIT;
+        }
+        try (PreparedStatement ps = con.prepareStatement(sql)) {
             ps.setInt(1, expiryTimeMins);
+            if (limit > 0) {
+                ps.setInt(2, limit);
+            }
             affectedRows = executeUpdate(ps, caller);
         } catch (SQLException ex) {
             throw sqlError(ex, caller);

libs/java/server_common/src/main/java/com/yahoo/athenz/common/server/ssh/SSHRecordStoreConnection.java

@@ -67,7 +67,8 @@ public interface SSHRecordStoreConnection extends Closeable {
      * considered expired if it hasn't been updated within the
      * specified number of minutes
      * @param expiryTimeMins expiry time in minutes
+     * @param limit maximum number of records to delete
      * @return number of records deleted
      */
-    int deleteExpiredSSHCertRecords(int expiryTimeMins) throws ServerResourceException;
+    int deleteExpiredSSHCertRecords(int expiryTimeMins, int limit) throws ServerResourceException;
 }

libs/java/server_common/src/test/java/com/yahoo/athenz/common/server/cert/impl/JDBCCertRecordStoreConnectionTest.java

@@ -418,23 +418,42 @@ public void testConnectionCloseException() throws SQLException {
 
     @Test
     public void testdeleteExpiredX509CertRecords() throws Exception {
+        final String SQL_NO_LIMIT = "DELETE FROM certificates " +
+                "WHERE currentTime < ADDDATE(NOW(), INTERVAL -? MINUTE);";
 
         JDBCCertRecordStoreConnection jdbcConn = new JDBCCertRecordStoreConnection(mockConn);
 
         Mockito.doReturn(1).when(mockPrepStmt).executeUpdate();
-        jdbcConn.deleteExpiredX509CertRecords(360);
+        jdbcConn.deleteExpiredX509CertRecords(360, 0);
 
+        Mockito.verify(mockConn, times(1)).prepareStatement(SQL_NO_LIMIT);
         Mockito.verify(mockPrepStmt, times(1)).setInt(1, 360);
         jdbcConn.close();
     }
+
+    @Test
+    public void testdeleteExpiredX509CertRecordsWithLimit() throws Exception {
+        final String SQL_WITH_LIMIT = "DELETE FROM certificates " +
+                "WHERE currentTime < ADDDATE(NOW(), INTERVAL -? MINUTE) LIMIT ?;";
+
+        JDBCCertRecordStoreConnection jdbcConn = new JDBCCertRecordStoreConnection(mockConn);
+
+        Mockito.doReturn(1000).when(mockPrepStmt).executeUpdate();
+        jdbcConn.deleteExpiredX509CertRecords(360, 1000);
+
+        Mockito.verify(mockConn, times(1)).prepareStatement(SQL_WITH_LIMIT);
+        Mockito.verify(mockPrepStmt, times(1)).setInt(1, 360);
+        Mockito.verify(mockPrepStmt, times(1)).setInt(2, 1000);
+        jdbcConn.close();
+    }
 
     @Test
     public void testdeleteExpiredX509CertRecordsInvalidValue() throws Exception {
 
         JDBCCertRecordStoreConnection jdbcConn = new JDBCCertRecordStoreConnection(mockConn);
 
         Mockito.doReturn(1).when(mockPrepStmt).executeUpdate();
-        jdbcConn.deleteExpiredX509CertRecords(0);
+        jdbcConn.deleteExpiredX509CertRecords(0, 0);
 
         Mockito.verify(mockPrepStmt, times(0)).setInt(1, 0);
         jdbcConn.close();
@@ -447,7 +466,7 @@ public void testdeleteExpiredX509CertRecordsException() throws SQLException {
 
         Mockito.when(mockPrepStmt.executeUpdate()).thenThrow(new SQLException("exc", "exc", 101));
         try {
-            jdbcConn.deleteExpiredX509CertRecords(360);
+            jdbcConn.deleteExpiredX509CertRecords(360, 0);
             fail();
         } catch (ServerResourceException ex) {
             Assert.assertEquals(ex.getCode(), ServerResourceException.INTERNAL_SERVER_ERROR);

libs/java/server_common/src/test/java/com/yahoo/athenz/common/server/cert/impl/JDBCSSHRecordStoreConnectionTest.java

@@ -339,23 +339,43 @@ public void testConnectionCloseException() throws SQLException {
 
     @Test
     public void testdeleteExpiredSSHCertRecords() throws Exception {
+        final String SQL_NO_LIMIT = "DELETE FROM ssh_certificates " +
+                "WHERE issueTime < ADDDATE(NOW(), INTERVAL -? MINUTE);";
 
         JDBCSSHRecordStoreConnection jdbcConn = new JDBCSSHRecordStoreConnection(mockConn);
 
         Mockito.doReturn(1).when(mockPrepStmt).executeUpdate();
-        jdbcConn.deleteExpiredSSHCertRecords(360);
+        jdbcConn.deleteExpiredSSHCertRecords(360, 0);
 
+        Mockito.verify(mockConn, times(1)).prepareStatement(SQL_NO_LIMIT);
         Mockito.verify(mockPrepStmt, times(1)).setInt(1, 360);
         jdbcConn.close();
     }
+
+    @Test
+    public void testdeleteExpiredSSHCertRecordsWithLimit() throws Exception {
+
+        final String SQL_WITH_LIMIT = "DELETE FROM ssh_certificates " +
+                "WHERE issueTime < ADDDATE(NOW(), INTERVAL -? MINUTE) LIMIT ?;";
+
+        JDBCSSHRecordStoreConnection jdbcConn = new JDBCSSHRecordStoreConnection(mockConn);
+
+        Mockito.doReturn(1).when(mockPrepStmt).executeUpdate();
+        jdbcConn.deleteExpiredSSHCertRecords(360, 1000);
+
+        Mockito.verify(mockConn, times(1)).prepareStatement(SQL_WITH_LIMIT);
+        Mockito.verify(mockPrepStmt, times(1)).setInt(1, 360);
+        Mockito.verify(mockPrepStmt, times(1)).setInt(2, 1000);
+        jdbcConn.close();
+    }
 
     @Test
     public void testdeleteExpiredSSHCertRecordsInvalidValue() throws Exception {
 
         JDBCSSHRecordStoreConnection jdbcConn = new JDBCSSHRecordStoreConnection(mockConn);
 
         Mockito.doReturn(1).when(mockPrepStmt).executeUpdate();
-        jdbcConn.deleteExpiredSSHCertRecords(0);
+        jdbcConn.deleteExpiredSSHCertRecords(0, 0);
 
         Mockito.verify(mockPrepStmt, times(0)).setInt(1, 0);
         jdbcConn.close();
@@ -368,7 +388,7 @@ public void testdeleteExpiredSSHCertRecordsException() throws SQLException {
 
         Mockito.when(mockPrepStmt.executeUpdate()).thenThrow(new SQLException("exc", "exc", 101));
         try {
-            jdbcConn.deleteExpiredSSHCertRecords(360);
+            jdbcConn.deleteExpiredSSHCertRecords(360, 0);
             fail();
         } catch (ServerResourceException ex) {
             Assert.assertEquals(ex.getCode(), ServerResourceException.INTERNAL_SERVER_ERROR);

servers/zts/conf/zts.properties

@@ -797,3 +797,16 @@ athenz.zts.k8s_provider_distribution_validator_factory_class=com.yahoo.athenz.in
 # the server allows the certificate signer module to impose the limit. The certificate
 # signer will either honor that value or lower it based on its own configuration.
 #athenz.zts.service_cert_default_expiry_mins=0
+
+# This property specifies the maximum number of records that the CertRecordCleaner
+# can delete in a single execution. If set to 0, there is no limit on the number of records
+# that can be deleted.
+#athenz.zts.cert_record_cleaner_limit=0
+
+# These properties define the interval at which the CertRecordCleaner is executed.
+# The cert_record_cleaner_duration property specifies the duration as an integer value,
+# and the cert_record_cleaner_timeunit property determines the unit of time.
+# The supported time units are: second, minute, hour, or day. The interval is calculated
+# as duration * timeunit, and the CertRecordCleaner will run at this defined interval.
+#athenz.zts.cert_record_cleaner_duration=1
+#athenz.zts.cert_record_cleaner_timeunit=day

servers/zts/src/main/java/com/yahoo/athenz/zts/ZTSConsts.java

@@ -137,6 +137,9 @@ public final class ZTSConsts {
     public static final String ZTS_PROP_SSH_USER_CA_CERT_KEYID_FNAME = "athenz.zts.ssh_user_ca_cert_keyid_fname";
     public static final String ZTS_PROP_RESP_X509_SIGNER_CERTS       = "athenz.zts.resp_x509_signer_certs";
     public static final String ZTS_PROP_RESP_SSH_SIGNER_CERTS        = "athenz.zts.resp_ssh_signer_certs";
+    public static final String ZTS_PROP_CERT_RECORD_CLEANER_LIMIT    = "athenz.zts.cert_record_cleaner_limit";
+    public static final String ZTS_PROP_CERT_RECORD_CLEANER_DURATION = "athenz.zts.cert_record_cleaner_duration";
+    public static final String ZTS_PROP_CERT_RECORD_CLEANER_TIMEUNIT = "athenz.zts.cert_record_cleaner_timeunit";
 
     public static final String DB_PROP_USER               = "user";
     public static final String DB_PROP_PASSWORD           = "password";

servers/zts/src/main/java/com/yahoo/athenz/zts/cert/InstanceCertManager.java

@@ -150,26 +150,44 @@ public InstanceCertManager(final PrivateKeyStore keyStore, Authorizer authorizer
         // start our thread to delete expired cert records once a day
         // unless we're running in read-only mode thus no modifications
         // to the database
+        final int limit = Integer.parseInt(System.getProperty(ZTSConsts.ZTS_PROP_CERT_RECORD_CLEANER_LIMIT, "0"));
+        final int duration = Integer.parseInt(System.getProperty(ZTSConsts.ZTS_PROP_CERT_RECORD_CLEANER_DURATION, "1"));
+        final TimeUnit timeUnit = parseTimeUnit(System.getProperty(ZTSConsts.ZTS_PROP_CERT_RECORD_CLEANER_TIMEUNIT, "day"));
 
         if (certStore != null && certSigner != null) {
             certScheduledExecutor = Executors.newScheduledThreadPool(1);
             certScheduledExecutor.scheduleAtFixedRate(
-                    new ExpiredX509CertRecordCleaner(certStore, certSigner.getMaxCertExpiryTimeMins(), readOnlyMode),
-                    0, 1, TimeUnit.DAYS);
+                    new ExpiredX509CertRecordCleaner(certStore, certSigner.getMaxCertExpiryTimeMins(), limit, readOnlyMode),
+                    0, duration, timeUnit);
         }
 
         if (sshStore != null) {
             int expiryTimeMins = (int) TimeUnit.MINUTES.convert(30, TimeUnit.DAYS);
             sshScheduledExecutor = Executors.newScheduledThreadPool(1);
             sshScheduledExecutor.scheduleAtFixedRate(
-                    new ExpiredSSHCertRecordCleaner(sshStore, expiryTimeMins, readOnlyMode),
-                    0, 1, TimeUnit.DAYS);
+                    new ExpiredSSHCertRecordCleaner(sshStore, expiryTimeMins, limit, readOnlyMode),
+                    0, duration, timeUnit);
         }
 
         // check to see if we have it configured to validate IP addresses
 
         validateIPAddress = new DynamicConfigBoolean(CONFIG_MANAGER, ZTSConsts.ZTS_PROP_SSH_CERT_VALIDATE_IP, false);
     }
+
+    static TimeUnit parseTimeUnit(String timeUnitStr) {
+        switch (timeUnitStr) {
+            case "second":
+                return TimeUnit.SECONDS;
+            case "minute":
+                return TimeUnit.MINUTES;
+            case "hour":
+                return TimeUnit.HOURS;
+            case "day":
+                return TimeUnit.DAYS;
+            default:
+                return TimeUnit.DAYS;
+        }
+    }
 
     void shutdown() {
         if (certScheduledExecutor != null) {
@@ -1417,11 +1435,13 @@ static class ExpiredX509CertRecordCleaner implements Runnable {
 
         private final CertRecordStore store;
         private final int expiryTimeMins;
+        private final int limit;
         private final DynamicConfigBoolean readOnlyMode;
 
-        public ExpiredX509CertRecordCleaner(CertRecordStore store, int expiryTimeMins, DynamicConfigBoolean readOnlyMode) {
+        public ExpiredX509CertRecordCleaner(CertRecordStore store, int expiryTimeMins, int limit, DynamicConfigBoolean readOnlyMode) {
             this.store = store;
             this.expiryTimeMins = expiryTimeMins;
+            this.limit = limit;
             this.readOnlyMode = readOnlyMode;
         }
 
@@ -1452,7 +1472,7 @@ int cleanupExpiredX509CertRecords() {
 
             int deletedRecords;
             try (CertRecordStoreConnection storeConnection = store.getConnection()) {
-                deletedRecords = storeConnection.deleteExpiredX509CertRecords(expiryTimeMins);
+                deletedRecords = storeConnection.deleteExpiredX509CertRecords(expiryTimeMins, limit);
             } catch (ServerResourceException ex) {
                 throw ZTSUtils.error(ex);
             }
@@ -1464,11 +1484,13 @@ static class ExpiredSSHCertRecordCleaner implements Runnable {
 
         private final SSHRecordStore store;
         private final int expiryTimeMins;
+        private final int limit;
         private final DynamicConfigBoolean readOnlyMode;
 
-        public ExpiredSSHCertRecordCleaner(SSHRecordStore store, int expiryTimeMins, DynamicConfigBoolean readOnlyMode) {
+        public ExpiredSSHCertRecordCleaner(SSHRecordStore store, int expiryTimeMins, int limit, DynamicConfigBoolean readOnlyMode) {
             this.store = store;
             this.expiryTimeMins = expiryTimeMins;
+            this.limit = limit;
             this.readOnlyMode = readOnlyMode;
         }
 
@@ -1499,7 +1521,7 @@ int cleanupExpiredSSHCertRecords() {
 
             int deletedRecords;
             try (SSHRecordStoreConnection storeConnection = store.getConnection()) {
-                deletedRecords = storeConnection.deleteExpiredSSHCertRecords(expiryTimeMins);
+                deletedRecords = storeConnection.deleteExpiredSSHCertRecords(expiryTimeMins, limit);
             } catch (ServerResourceException ex) {
                 throw ZTSUtils.error(ex);
             }

servers/zts/src/main/java/com/yahoo/athenz/zts/cert/impl/FileCertRecordStoreConnection.java

@@ -79,7 +79,7 @@ public boolean deleteX509CertRecord(String provider, String instanceId, String s
     }
 
     @Override
-    public int deleteExpiredX509CertRecords(int expiryTimeMins) {
+    public int deleteExpiredX509CertRecords(int expiryTimeMins, int limit) {
         String[] fnames = rootDir.list();
         if (fnames == null) {
             return 0;
@@ -98,6 +98,9 @@ public int deleteExpiredX509CertRecords(int expiryTimeMins) {
             //noinspection ResultOfMethodCallIgnored
             file.delete();
             count += 1;
+            if (limit == count) {
+                break;
+            }
         }
         return count;
     }