just for review

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

servers/zms/src/main/java/com/yahoo/athenz/zms/DBService.java

@@ -29,7 +29,9 @@
 import com.yahoo.athenz.common.server.log.AuditLogger;
 import com.yahoo.athenz.common.server.util.AuthzHelper;
 import com.yahoo.athenz.common.server.util.ResourceUtils;
+import com.yahoo.athenz.common.server.util.config.dynamic.DynamicConfigInteger;
 import com.yahoo.athenz.zms.config.MemberDueDays;
+import com.yahoo.athenz.zms.purge.PurgeMember;
 import com.yahoo.athenz.zms.store.*;
 import com.yahoo.athenz.zms.utils.ZMSUtils;
 import com.yahoo.rdl.JSON;
@@ -47,6 +49,8 @@
 import java.util.function.Function;
 import java.util.stream.Collectors;
 
+import static com.yahoo.athenz.common.server.util.config.ConfigManagerSingleton.CONFIG_MANAGER;
+
 public class DBService implements RolesProvider {
 
     ObjectStore store;
@@ -275,6 +279,36 @@ void saveChanges(ObjectStoreConnection con, String domainName) {
         cacheStore.invalidate(domainName);
     }
 
+    void purgeTaskSaveChanges(ResourceContext ctx, ObjectStoreConnection con, List<PurgeMember> purgeMemberList, String auditRef, String caller, DomainChangeMessage.ObjectType collectionType) {
+
+        // we're first going to commit our changes which will
+        // also set the connection in auto-commit mode.
+        con.commitChanges();
+
+        Set<String> domainsSet = new HashSet<>();
+        Set<String> collectionSet = new HashSet<>();
+
+        for (PurgeMember purgeMember: purgeMemberList) {
+            //we are going to change the domain timestamp in
+            // auto-commit mode so that we don't have a contention
+            if (!domainsSet.contains(purgeMember.getDomainName())) {
+                con.updateDomainModTimestamp(purgeMember.getDomainName());
+                cacheStore.invalidate(purgeMember.getDomainName());
+                domainsSet.add(purgeMember.getDomainName());
+            }
+
+            // audit log the request
+            auditLogRequest(ctx, purgeMember.getDomainName(), auditRef, caller, ZMSConsts.HTTP_DELETE,
+                    purgeMember.getCollectionName(), "{\"member\": \"" + purgeMember.getPrincipalName() + "\"}");
+
+            if (!collectionSet.contains(purgeMember.getCollectionName())) {
+                // add domain change event
+                addDomainChangeMessage(ctx, purgeMember.getDomainName(), purgeMember.getCollectionName(), collectionType);
+                collectionSet.add(purgeMember.getCollectionName());
+            }
+        }
+    }
+
     void auditLogRequest(ResourceContext ctx, String domainName, String auditRef,
             String caller, String operation, String entityName, String auditDetails) {
         auditLogger.log(getAuditLogMsgBuilder(ctx, domainName, auditRef, caller, operation, entityName, auditDetails));
@@ -1796,6 +1830,11 @@ void executePutEntity(ResourceContext ctx, String domainName, String entityName,
         }
     }
 
+//    void executeDeleteMembership(ResourceContext ctx, String domainName, String roleName,
+//                                 String normalizedMember, String auditRef, String caller) {
+//        executeDeleteMembership( ctx,  null, domainName, roleName, normalizedMember, auditRef, caller);
+//    }
+
     void executeDeleteMembership(ResourceContext ctx, String domainName, String roleName,
             String normalizedMember, String auditRef, String caller) {
 
@@ -7994,6 +8033,117 @@ public DomainList listDomainDependencies(String service) {
         }
     }
 
+    void executeDeleteExpiredMemberships(ResourceContext ctx, ObjectStoreConnection con, int roleId, String roleName,
+                                 int principalId, String normalizedMember, String auditRef, String caller) {
+        try {
+            final String principal = getPrincipalName(ctx);
+
+            // process our delete role member operation
+
+            if (!con.deleteExpiredRoleMember(roleId, principalId, normalizedMember, principal, auditRef)) {
+                throw ZMSUtils.notFoundError(caller + ": unable to delete role member: " +
+                        normalizedMember + " from role: " + roleName, caller);
+            }
+
+            return;
+
+        } catch (ResourceException ex) {
+            throw ex;
+        }
+    }
+
+    public void executeDeleteAllExpiredRoleMemberships(ResourceContext ctx, String auditRef, String caller) {
+        final int maxDbCallsPerRun = new DynamicConfigInteger(CONFIG_MANAGER, ZMSConsts.ZMS_PROP_PURGE_TASK_MAX_DB_CALLS_PER_RUN, ZMSConsts.PURGE_TASK_MAX_DB_CALLS_PER_RUN_DEF).get();
+        final int limitPerCall = new DynamicConfigInteger(CONFIG_MANAGER, ZMSConsts.ZMS_PROP_PURGE_TASK_LIMIT_PER_CALL, ZMSConsts.PURGE_TASK_LIMIT_PER_CALL_DEF).get();
+        int offset = 0;
+        List<PurgeMember> allExpiredRoleMembers = new ArrayList<>();
+        List<PurgeMember> changedList = new ArrayList<>();
+        ObjectStoreConnection con;
+        try {
+            con = store.getConnection(false, false);
+            // get all expired members from all roles
+            for (int i = 0; i < maxDbCallsPerRun; ++i) {
+                List<PurgeMember> expiredRoleMembers = con.getAllExpiredRoleMembers(limitPerCall, offset);
+                if (expiredRoleMembers == null || expiredRoleMembers.isEmpty()) {
+                    break;
+                }
+                allExpiredRoleMembers.addAll(expiredRoleMembers);
+                if (expiredRoleMembers.size() < limitPerCall) {
+                    break;
+                }
+                offset += limitPerCall;
+            }
+            // delete all expired role members one by one (required due auditLog)
+            for (PurgeMember purgeMember: allExpiredRoleMembers) {
+                try {
+                    executeDeleteExpiredMemberships(ctx, con, purgeMember.getCollectionId(), purgeMember.getCollectionName(),
+                            purgeMember.getPrincipalId(), purgeMember.getPrincipalName(), auditRef, caller);
+                    changedList.add(purgeMember);
+
+                } catch (RuntimeException e) {
+                    //TODO - write the response status to a log file
+                }
+            }
+
+            purgeTaskSaveChanges(ctx, con, changedList, auditRef, caller, DomainChangeMessage.ObjectType.ROLE);
+
+        } catch (ResourceException ex) {
+
+        }
+    }
+
+    void executeDeleteAllExpiredGroupMemberships(ResourceContext ctx, ObjectStoreConnection con, int groupId, String groupName,
+                                                 int principalId, String normalizedMember, String auditRef) {
+        final String principal = getPrincipalName(ctx);
+
+        // process our delete group member operation
+        if (!con.deleteExpiredGroupMember(groupId, principalId, normalizedMember, principal, auditRef)) {
+            throw ZMSUtils.notFoundError("unable to delete group member: " +
+                    normalizedMember + " from group: " + groupName, ctx.getApiName());
+        }
+    }
+
+    public void executeDeleteAllExpiredGroupMemberships(ResourceContext ctx, String auditRef, String caller) {
+        final int maxDbCallsPerRun = new DynamicConfigInteger(CONFIG_MANAGER, ZMSConsts.ZMS_PROP_PURGE_TASK_MAX_DB_CALLS_PER_RUN, ZMSConsts.PURGE_TASK_MAX_DB_CALLS_PER_RUN_DEF).get();
+        final int limitPerCall = new DynamicConfigInteger(CONFIG_MANAGER, ZMSConsts.ZMS_PROP_PURGE_TASK_LIMIT_PER_CALL, ZMSConsts.PURGE_TASK_LIMIT_PER_CALL_DEF).get();
+        int offset = 0;
+        List<PurgeMember> allExpiredGroupMembers = new ArrayList<>();
+        List<PurgeMember> changedList = new ArrayList<>();
+        ObjectStoreConnection con;
+        try {
+            con = store.getConnection(false, false);
+            // get all expired members from all groups
+            for (int i = 0; i < maxDbCallsPerRun; ++i) {
+                List<PurgeMember> expiredGroupMembers = con.getAllExpiredGroupMembers(limitPerCall, offset);
+                if (expiredGroupMembers == null || expiredGroupMembers.isEmpty()) {
+                    break;
+                }
+                allExpiredGroupMembers.addAll(expiredGroupMembers);
+                if (expiredGroupMembers.size() < limitPerCall) {
+                    break;
+                }
+                offset += limitPerCall;
+            }
+            // delete all expired group members one by one (required due auditLog)
+            for (PurgeMember purgeMember: allExpiredGroupMembers) {
+                try {
+                    executeDeleteAllExpiredGroupMemberships(ctx, con, purgeMember.getCollectionId(), purgeMember.getCollectionName(),
+                            purgeMember.getPrincipalId(), purgeMember.getPrincipalName(), auditRef);
+                    changedList.add(purgeMember);
+
+                } catch (RuntimeException e) {
+                    //TODO - write the response status to a log file
+                }
+            }
+
+            purgeTaskSaveChanges(ctx, con, changedList, auditRef, caller, DomainChangeMessage.ObjectType.GROUP);
+
+        } catch (ResourceException ex) {
+
+        }
+    }
+
+
     public AuthHistoryDependencies getAuthHistory(String domain) {
         if (authHistoryStore == null) {
             LOG.warn("Authentication History Store is disabled. getAuthHistory will return empty lists.");

servers/zms/src/main/java/com/yahoo/athenz/zms/ZMSConsts.java

@@ -368,4 +368,11 @@ public final class ZMSConsts {
     public static final String ZMS_PROP_AUTH_HISTORY_DYNAMODB_REGION = "athenz.zms.auth_history_dynamodb_region";
     public static final String ZMS_DYNAMODB_URI_DOMAIN_INDEX_NAME = "uriDomain-index";
     public static final String ZMS_DYNAMODB_PRINCIPAL_DOMAIN_INDEX_NAME = "principalDomain-index";
+
+    // purge task
+    public static final String ZMS_PROP_PURGE_TASK_MAX_DB_CALLS_PER_RUN = "purge_task_max_db_calls_per_run";
+    public static final Integer PURGE_TASK_MAX_DB_CALLS_PER_RUN_DEF = 20;
+    public static final String ZMS_PROP_PURGE_TASK_LIMIT_PER_CALL = "purge_task_limit_per_call";
+    public static final Integer PURGE_TASK_LIMIT_PER_CALL_DEF = 500;
+    public static final int DELAY_PURGE_EXPIRED_MEMBERS_DAYS_DEFAULT = 180;
 }

servers/zms/src/main/java/com/yahoo/athenz/zms/ZMSImpl.java

@@ -685,6 +685,7 @@ public ZMSImpl() {
         // load the DomainChangePublisher
 
         loadDomainChangePublisher();
+
     }
 
     void loadDomainChangePublisher() {

servers/zms/src/main/java/com/yahoo/athenz/zms/purge/PurgeMember.java

@@ -0,0 +1,57 @@
+package com.yahoo.athenz.zms.purge;
+
+public class PurgeMember {
+    private String domainName;
+    private int collectionId;
+    private String collectionName;
+    private int principalId;
+    private String principalName;
+
+    public PurgeMember() {
+    }
+
+    public String getDomainName() {
+        return domainName;
+    }
+
+    public PurgeMember setDomainName(String domainName) {
+        this.domainName = domainName;
+        return this;
+    }
+
+    public int getCollectionId() {
+        return collectionId;
+    }
+
+    public PurgeMember setCollectionId(int collectionId) {
+        this.collectionId = collectionId;
+        return this;
+    }
+
+    public String getCollectionName() {
+        return collectionName;
+    }
+
+    public PurgeMember setCollectionName(String collectionName) {
+        this.collectionName = collectionName;
+        return this;
+    }
+
+    public int getPrincipalId() {
+        return principalId;
+    }
+
+    public PurgeMember setPrincipalId(int principalId) {
+        this.principalId = principalId;
+        return this;
+    }
+
+    public String getPrincipalName() {
+        return principalName;
+    }
+
+    public PurgeMember setPrincipalName(String principalName) {
+        this.principalName = principalName;
+        return this;
+    }
+}

servers/zms/src/main/java/com/yahoo/athenz/zms/store/ObjectStoreConnection.java

@@ -21,6 +21,7 @@
 import java.util.Set;
 
 import com.yahoo.athenz.zms.*;
+import com.yahoo.athenz.zms.purge.PurgeMember;
 
 public interface ObjectStoreConnection extends Closeable {
 
@@ -227,4 +228,11 @@ public interface ObjectStoreConnection extends Closeable {
     boolean deleteDomainDependency(String domainName, String service);
     List<String> listServiceDependencies(String domainName);
     List<String> listDomainDependencies(String service);
+
+    // purge commands
+
+    List<PurgeMember> getAllExpiredRoleMembers(int limit, int offset);
+    List<PurgeMember> getAllExpiredGroupMembers(int limit, int offset);
+    boolean deleteExpiredRoleMember(int roleId, int principalId, String member, String principal, String auditRef);
+    boolean deleteExpiredGroupMember(int groupId, int principalId, String member, String principal, String auditRef);
 }