KEYCLOAK-2412 Added ClusterProvider. Avoid concurrent federation sync execution by more cluster nodes at the same time.

Clustering - more progress

services/src/main/java/org/keycloak/protocol/oidc/TokenManager.java

@@ -17,6 +17,7 @@
 
 package org.keycloak.protocol.oidc;
 
+import org.keycloak.cluster.ClusterProvider;
 import org.keycloak.common.ClientConnection;
 import org.keycloak.OAuth2Constants;
 import org.keycloak.OAuthErrorException;
@@ -193,7 +194,7 @@ public class TokenManager {
         int currentTime = Time.currentTime();
 
         if (realm.isRevokeRefreshToken()) {
-            int clusterStartupTime = session.sessions().getClusterStartupTime();
+            int clusterStartupTime = session.getProvider(ClusterProvider.class).getClusterStartupTime();
 
             if (refreshToken.getIssuedAt() < validation.clientSession.getTimestamp() && (clusterStartupTime != validation.clientSession.getTimestamp())) {
                 throw new OAuthErrorException(OAuthErrorException.INVALID_GRANT, "Stale token");

services/src/main/java/org/keycloak/services/managers/RealmManager.java

@@ -231,7 +231,7 @@ public class RealmManager implements RealmImporter {
             // Remove all periodic syncs for configured federation providers
             UsersSyncManager usersSyncManager = new UsersSyncManager();
             for (final UserFederationProviderModel fedProvider : federationProviders) {
-                usersSyncManager.removePeriodicSyncForProvider(session.getProvider(TimerProvider.class), fedProvider);
+                usersSyncManager.notifyToRefreshPeriodicSync(session, realm, fedProvider, true);
             }
         }
         return removed;
@@ -434,7 +434,7 @@ public class RealmManager implements RealmImporter {
         List<UserFederationProviderModel> federationProviders = realm.getUserFederationProviders();
         UsersSyncManager usersSyncManager = new UsersSyncManager();
         for (final UserFederationProviderModel fedProvider : federationProviders) {
-            usersSyncManager.refreshPeriodicSyncForProvider(session.getKeycloakSessionFactory(), session.getProvider(TimerProvider.class), fedProvider, realm.getId());
+            usersSyncManager.notifyToRefreshPeriodicSync(session, realm, fedProvider, false);
         }
         return realm;
     }

services/src/main/java/org/keycloak/services/managers/UsersSyncManager.java

@@ -16,6 +16,10 @@
  */
 package org.keycloak.services.managers;
 
+import org.keycloak.cluster.ClusterEvent;
+import org.keycloak.cluster.ClusterListener;
+import org.keycloak.cluster.ClusterProvider;
+import org.keycloak.cluster.ExecutionResult;
 import org.keycloak.common.util.Time;
 import org.keycloak.models.KeycloakSession;
 import org.keycloak.models.KeycloakSessionFactory;
@@ -30,13 +34,17 @@ import org.keycloak.services.ServicesLogger;
 import org.keycloak.timer.TimerProvider;
 
 
+import java.io.Serializable;
 import java.util.List;
+import java.util.concurrent.Callable;
 
 /**
  * @author <a href="mailto:mposolda@redhat.com">Marek Posolda</a>
  */
 public class UsersSyncManager {
 
+    private static final String FEDERATION_TASK_KEY = "federation";
+
     protected static final ServicesLogger logger = ServicesLogger.ROOT_LOGGER;
 
     /**
@@ -57,26 +65,106 @@ public class UsersSyncManager {
                         refreshPeriodicSyncForProvider(sessionFactory, timer, fedProvider, realm.getId());
                     }
                 }
+
+                ClusterProvider clusterProvider = session.getProvider(ClusterProvider.class);
+                clusterProvider.registerListener(FEDERATION_TASK_KEY, new UserFederationClusterListener(sessionFactory));
             }
         });
     }
 
-    public UserFederationSyncResult syncAllUsers(final KeycloakSessionFactory sessionFactory, String realmId, final UserFederationProviderModel fedProvider) {
-        final UserFederationProviderFactory fedProviderFactory = (UserFederationProviderFactory) sessionFactory.getProviderFactory(UserFederationProvider.class, fedProvider.getProviderName());
-        updateLastSyncInterval(sessionFactory, fedProvider, realmId);
-        return fedProviderFactory.syncAllUsers(sessionFactory, realmId, fedProvider);
+
+    private class Holder {
+        ExecutionResult<UserFederationSyncResult> result;
     }
 
-    public UserFederationSyncResult syncChangedUsers(final KeycloakSessionFactory sessionFactory, String realmId, final UserFederationProviderModel fedProvider) {
-        final UserFederationProviderFactory fedProviderFactory = (UserFederationProviderFactory) sessionFactory.getProviderFactory(UserFederationProvider.class, fedProvider.getProviderName());
+    public UserFederationSyncResult syncAllUsers(final KeycloakSessionFactory sessionFactory, final String realmId, final UserFederationProviderModel fedProvider) {
+        final Holder holder = new Holder();
 
-        // See when we did last sync.
-        int oldLastSync = fedProvider.getLastSync();
-        updateLastSyncInterval(sessionFactory, fedProvider, realmId);
-        return fedProviderFactory.syncChangedUsers(sessionFactory, realmId, fedProvider, Time.toDate(oldLastSync));
+        // Ensure not executed concurrently on this or any other cluster node
+        KeycloakModelUtils.runJobInTransaction(sessionFactory, new KeycloakSessionTask() {
+
+            @Override
+            public void run(KeycloakSession session) {
+                ClusterProvider clusterProvider = session.getProvider(ClusterProvider.class);
+                // shared key for "full" and "changed" . Improve if needed
+                String taskKey = fedProvider.getId() + "::sync";
+
+                // 30 seconds minimal timeout for now
+                int timeout = Math.max(30, fedProvider.getFullSyncPeriod());
+                holder.result = clusterProvider.executeIfNotExecuted(taskKey, timeout, new Callable<UserFederationSyncResult>() {
+
+                    @Override
+                    public UserFederationSyncResult call() throws Exception {
+                        final UserFederationProviderFactory fedProviderFactory = (UserFederationProviderFactory) sessionFactory.getProviderFactory(UserFederationProvider.class, fedProvider.getProviderName());
+                        updateLastSyncInterval(sessionFactory, fedProvider, realmId);
+                        return fedProviderFactory.syncAllUsers(sessionFactory, realmId, fedProvider);
+                    }
+
+                });
+            }
+
+        });
+
+        if (holder.result == null || !holder.result.isExecuted()) {
+            logger.infof("syncAllUsers for federation provider %s was ignored as it's already in progress", fedProvider.getDisplayName());
+            return UserFederationSyncResult.ignored();
+        } else {
+            return holder.result.getResult();
+        }
     }
 
-    public void refreshPeriodicSyncForProvider(final KeycloakSessionFactory sessionFactory, TimerProvider timer, final UserFederationProviderModel fedProvider, final String realmId) {
+    public UserFederationSyncResult syncChangedUsers(final KeycloakSessionFactory sessionFactory, final String realmId, final UserFederationProviderModel fedProvider) {
+        final Holder holder = new Holder();
+
+        // Ensure not executed concurrently on this or any other cluster node
+        KeycloakModelUtils.runJobInTransaction(sessionFactory, new KeycloakSessionTask() {
+
+            @Override
+            public void run(KeycloakSession session) {
+                ClusterProvider clusterProvider = session.getProvider(ClusterProvider.class);
+                // shared key for "full" and "changed" . Improve if needed
+                String taskKey = fedProvider.getId() + "::sync";
+
+                // 30 seconds minimal timeout for now
+                int timeout = Math.max(30, fedProvider.getChangedSyncPeriod());
+                holder.result = clusterProvider.executeIfNotExecuted(taskKey, timeout, new Callable<UserFederationSyncResult>() {
+
+                    @Override
+                    public UserFederationSyncResult call() throws Exception {
+                        final UserFederationProviderFactory fedProviderFactory = (UserFederationProviderFactory) sessionFactory.getProviderFactory(UserFederationProvider.class, fedProvider.getProviderName());
+
+                        // See when we did last sync.
+                        int oldLastSync = fedProvider.getLastSync();
+                        updateLastSyncInterval(sessionFactory, fedProvider, realmId);
+                        return fedProviderFactory.syncChangedUsers(sessionFactory, realmId, fedProvider, Time.toDate(oldLastSync));
+                    }
+
+                });
+            }
+
+        });
+
+        if (holder.result == null || !holder.result.isExecuted()) {
+            logger.infof("syncChangedUsers for federation provider %s was ignored as it's already in progress", fedProvider.getDisplayName());
+            return UserFederationSyncResult.ignored();
+        } else {
+            return holder.result.getResult();
+        }
+    }
+
+
+    // Ensure all cluster nodes are notified
+    public void notifyToRefreshPeriodicSync(KeycloakSession session, RealmModel realm, UserFederationProviderModel federationProvider, boolean removed) {
+        FederationProviderClusterEvent event = FederationProviderClusterEvent.createEvent(removed, realm.getId(), federationProvider);
+        session.getProvider(ClusterProvider.class).notify(FEDERATION_TASK_KEY, event);
+    }
+
+
+    // Executed once it receives notification that some UserFederationProvider was created or updated
+    protected void refreshPeriodicSyncForProvider(final KeycloakSessionFactory sessionFactory, TimerProvider timer, final UserFederationProviderModel fedProvider, final String realmId) {
+        logger.infof("Going to refresh periodic sync for provider '%s' . Full sync period: %d , changed users sync period: %d",
+                fedProvider.getDisplayName(), fedProvider.getFullSyncPeriod(), fedProvider.getChangedSyncPeriod());
+
         if (fedProvider.getFullSyncPeriod() > 0) {
             // We want periodic full sync for this provider
             timer.schedule(new Runnable() {
@@ -84,7 +172,12 @@ public class UsersSyncManager {
                 @Override
                 public void run() {
                     try {
-                        syncAllUsers(sessionFactory, realmId, fedProvider);
+                        boolean shouldPerformSync = shouldPerformNewPeriodicSync(fedProvider.getLastSync(), fedProvider.getChangedSyncPeriod());
+                        if (shouldPerformSync) {
+                            syncAllUsers(sessionFactory, realmId, fedProvider);
+                        } else {
+                            logger.infof("Ignored periodic full sync with federation provider %s due small time since last sync", fedProvider.getDisplayName());
+                        }
                     } catch (Throwable t) {
                         logger.errorDuringFullUserSync(t);
                     }
@@ -102,7 +195,12 @@ public class UsersSyncManager {
                 @Override
                 public void run() {
                     try {
-                        syncChangedUsers(sessionFactory, realmId, fedProvider);
+                        boolean shouldPerformSync = shouldPerformNewPeriodicSync(fedProvider.getLastSync(), fedProvider.getChangedSyncPeriod());
+                        if (shouldPerformSync) {
+                            syncChangedUsers(sessionFactory, realmId, fedProvider);
+                        } else {
+                            logger.infof("Ignored periodic changed-users sync with federation provider %s due small time since last sync", fedProvider.getDisplayName());
+                        }
                     } catch (Throwable t) {
                         logger.errorDuringChangedUserSync(t);
                     }
@@ -115,7 +213,21 @@ public class UsersSyncManager {
         }
     }
 
-    public void removePeriodicSyncForProvider(TimerProvider timer, final UserFederationProviderModel fedProvider) {
+    // Skip syncing if there is short time since last sync time.
+    private boolean shouldPerformNewPeriodicSync(int lastSyncTime, int period) {
+        if (lastSyncTime <= 0) {
+            return true;
+        }
+
+        int currentTime = Time.currentTime();
+        int timeSinceLastSync = currentTime - lastSyncTime;
+
+        return (timeSinceLastSync * 2 > period);
+    }
+
+    // Executed once it receives notification that some UserFederationProvider was removed
+    protected void removePeriodicSyncForProvider(TimerProvider timer, UserFederationProviderModel fedProvider) {
+        logger.infof("Removing periodic sync for provider %s", fedProvider.getDisplayName());
         timer.cancelTask(fedProvider.getId() + "-FULL");
         timer.cancelTask(fedProvider.getId() + "-CHANGED");
     }
@@ -144,4 +256,73 @@ public class UsersSyncManager {
         });
     }
 
+
+    private class UserFederationClusterListener implements ClusterListener {
+
+        private final KeycloakSessionFactory sessionFactory;
+
+        public UserFederationClusterListener(KeycloakSessionFactory sessionFactory) {
+            this.sessionFactory = sessionFactory;
+        }
+
+        @Override
+        public void run(ClusterEvent event) {
+            final FederationProviderClusterEvent fedEvent = (FederationProviderClusterEvent) event;
+            KeycloakModelUtils.runJobInTransaction(sessionFactory, new KeycloakSessionTask() {
+
+                @Override
+                public void run(KeycloakSession session) {
+                    TimerProvider timer = session.getProvider(TimerProvider.class);
+                    if (fedEvent.isRemoved()) {
+                        removePeriodicSyncForProvider(timer, fedEvent.getFederationProvider());
+                    } else {
+                        refreshPeriodicSyncForProvider(sessionFactory, timer, fedEvent.getFederationProvider(), fedEvent.getRealmId());
+                    }
+                }
+
+            });
+        }
+    }
+
+
+    // Send to cluster during each update or remove of federationProvider, so all nodes can update sync periods
+    public static class FederationProviderClusterEvent implements ClusterEvent {
+
+        private boolean removed;
+        private String realmId;
+        private UserFederationProviderModel federationProvider;
+
+        public boolean isRemoved() {
+            return removed;
+        }
+
+        public void setRemoved(boolean removed) {
+            this.removed = removed;
+        }
+
+        public String getRealmId() {
+            return realmId;
+        }
+
+        public void setRealmId(String realmId) {
+            this.realmId = realmId;
+        }
+
+        public UserFederationProviderModel getFederationProvider() {
+            return federationProvider;
+        }
+
+        public void setFederationProvider(UserFederationProviderModel federationProvider) {
+            this.federationProvider = federationProvider;
+        }
+
+        public static FederationProviderClusterEvent createEvent(boolean removed, String realmId, UserFederationProviderModel fedProvider) {
+            FederationProviderClusterEvent notification = new FederationProviderClusterEvent();
+            notification.setRemoved(removed);
+            notification.setRealmId(realmId);
+            notification.setFederationProvider(fedProvider);
+            return notification;
+        }
+    }
+
 }

services/src/main/java/org/keycloak/services/resources/KeycloakApplication.java

@@ -38,6 +38,7 @@ import org.keycloak.services.managers.UsersSyncManager;
 import org.keycloak.services.resources.admin.AdminRoot;
 import org.keycloak.services.scheduled.ClearExpiredEvents;
 import org.keycloak.services.scheduled.ClearExpiredUserSessions;
+import org.keycloak.services.scheduled.ClusterAwareScheduledTaskRunner;
 import org.keycloak.services.scheduled.ScheduledTaskRunner;
 import org.keycloak.services.util.JsonConfigProvider;
 import org.keycloak.services.util.ObjectMapperResolver;
@@ -217,8 +218,8 @@ public class KeycloakApplication extends Application {
         KeycloakSession session = sessionFactory.create();
         try {
             TimerProvider timer = session.getProvider(TimerProvider.class);
-            timer.schedule(new ScheduledTaskRunner(sessionFactory, new ClearExpiredEvents()), interval, "ClearExpiredEvents");
-            timer.schedule(new ScheduledTaskRunner(sessionFactory, new ClearExpiredUserSessions()), interval, "ClearExpiredUserSessions");
+            timer.schedule(new ClusterAwareScheduledTaskRunner(sessionFactory, new ClearExpiredEvents(), interval), interval, "ClearExpiredEvents");
+            timer.schedule(new ClusterAwareScheduledTaskRunner(sessionFactory, new ClearExpiredUserSessions(), interval), interval, "ClearExpiredUserSessions");
             new UsersSyncManager().bootstrapPeriodic(sessionFactory, timer);
         } finally {
             session.close();

services/src/main/java/org/keycloak/services/resources/admin/RealmAdminResource.java

@@ -242,7 +242,7 @@ public class RealmAdminResource {
             List<UserFederationProviderModel> federationProviders = realm.getUserFederationProviders();
             UsersSyncManager usersSyncManager = new UsersSyncManager();
             for (final UserFederationProviderModel fedProvider : federationProviders) {
-                usersSyncManager.refreshPeriodicSyncForProvider(session.getKeycloakSessionFactory(), session.getProvider(TimerProvider.class), fedProvider, realm.getId());
+                usersSyncManager.notifyToRefreshPeriodicSync(session, realm, fedProvider, false);
             }
 
             adminEvent.operation(OperationType.UPDATE).representation(rep).success();

services/src/main/java/org/keycloak/services/resources/admin/UserFederationProviderResource.java

@@ -106,7 +106,7 @@ public class UserFederationProviderResource {
         UserFederationProviderModel model = new UserFederationProviderModel(rep.getId(), rep.getProviderName(), rep.getConfig(), rep.getPriority(), displayName,
                 rep.getFullSyncPeriod(), rep.getChangedSyncPeriod(), rep.getLastSync());
         realm.updateUserFederationProvider(model);
-        new UsersSyncManager().refreshPeriodicSyncForProvider(session.getKeycloakSessionFactory(), session.getProvider(TimerProvider.class), model, realm.getId());
+        new UsersSyncManager().notifyToRefreshPeriodicSync(session, realm, model, false);
         boolean kerberosCredsAdded = UserFederationProvidersResource.checkKerberosCredential(session, realm, model);
         if (kerberosCredsAdded) {
             logger.addedKerberosToRealmCredentials();
@@ -138,7 +138,7 @@ public class UserFederationProviderResource {
         auth.requireManage();
 
         realm.removeUserFederationProvider(this.federationProviderModel);
-        new UsersSyncManager().removePeriodicSyncForProvider(session.getProvider(TimerProvider.class), this.federationProviderModel);
+        new UsersSyncManager().notifyToRefreshPeriodicSync(session, realm, this.federationProviderModel, true);
 
         adminEvent.operation(OperationType.DELETE).resourcePath(uriInfo).success();
 

services/src/main/java/org/keycloak/services/resources/admin/UserFederationProvidersResource.java

@@ -178,7 +178,7 @@ public class UserFederationProvidersResource {
         }
         UserFederationProviderModel model = realm.addUserFederationProvider(rep.getProviderName(), rep.getConfig(), rep.getPriority(), displayName,
                 rep.getFullSyncPeriod(), rep.getChangedSyncPeriod(), rep.getLastSync());
-        new UsersSyncManager().refreshPeriodicSyncForProvider(session.getKeycloakSessionFactory(), session.getProvider(TimerProvider.class), model, realm.getId());
+        new UsersSyncManager().notifyToRefreshPeriodicSync(session, realm, model, false);
         boolean kerberosCredsAdded = checkKerberosCredential(session, realm, model);
         if (kerberosCredsAdded) {
             logger.addedKerberosToRealmCredentials();

services/src/main/java/org/keycloak/services/scheduled/ClusterAwareScheduledTaskRunner.java

@@ -0,0 +1,68 @@
+/*
+ * Copyright 2016 Red Hat, Inc. and/or its affiliates
+ * and other contributors as indicated by the @author tags.
+ *
+ * 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.keycloak.services.scheduled;
+
+import java.util.concurrent.Callable;
+
+import org.keycloak.cluster.ClusterProvider;
+import org.keycloak.cluster.ExecutionResult;
+import org.keycloak.models.KeycloakSession;
+import org.keycloak.models.KeycloakSessionFactory;
+
+/**
+ * Ensures that there are not concurrent executions of same task (either on this host or any other cluster host)
+ *
+ * @author <a href="mailto:mposolda@redhat.com">Marek Posolda</a>
+ */
+public class ClusterAwareScheduledTaskRunner extends ScheduledTaskRunner {
+
+    private final int intervalSecs;
+
+    public ClusterAwareScheduledTaskRunner(KeycloakSessionFactory sessionFactory, ScheduledTask task, long intervalMillis) {
+        super(sessionFactory, task);
+        this.intervalSecs = (int) (intervalMillis / 1000);
+    }
+
+    @Override
+    protected void runTask(final KeycloakSession session) {
+        session.getTransaction().begin();
+
+        ClusterProvider clusterProvider = session.getProvider(ClusterProvider.class);
+        String taskKey = task.getClass().getSimpleName();
+
+        ExecutionResult<Void> result = clusterProvider.executeIfNotExecuted(taskKey, intervalSecs, new Callable<Void>() {
+
+            @Override
+            public Void call() throws Exception {
+                task.run(session);
+                return null;
+            }
+
+        });
+
+        session.getTransaction().commit();
+
+        if (result.isExecuted()) {
+            logger.debugf("Executed scheduled task %s", taskKey);
+        } else {
+            logger.debugf("Skipped execution of task %s as other cluster node is executing it", taskKey);
+        }
+    }
+
+
+}

services/src/main/java/org/keycloak/services/scheduled/ScheduledTaskRunner.java

@@ -17,6 +17,10 @@
 
 package org.keycloak.services.scheduled;
 
+import java.util.concurrent.Callable;
+
+import org.keycloak.cluster.ClusterProvider;
+import org.keycloak.cluster.ExecutionResult;
 import org.keycloak.models.KeycloakSession;
 import org.keycloak.models.KeycloakSessionFactory;
 import org.keycloak.services.ServicesLogger;
@@ -26,10 +30,10 @@ import org.keycloak.services.ServicesLogger;
  */
 public class ScheduledTaskRunner implements Runnable {
 
-    private static final ServicesLogger logger = ServicesLogger.ROOT_LOGGER;
+    protected static final ServicesLogger logger = ServicesLogger.ROOT_LOGGER;
 
-    private final KeycloakSessionFactory sessionFactory;
-    private final ScheduledTask task;
+    protected final KeycloakSessionFactory sessionFactory;
+    protected final ScheduledTask task;
 
     public ScheduledTaskRunner(KeycloakSessionFactory sessionFactory, ScheduledTask task) {
         this.sessionFactory = sessionFactory;
@@ -40,11 +44,7 @@ public class ScheduledTaskRunner implements Runnable {
     public void run() {
         KeycloakSession session = sessionFactory.create();
         try {
-            session.getTransaction().begin();
-            task.run(session);
-            session.getTransaction().commit();
-
-            logger.debug("Executed scheduled task " + task.getClass().getSimpleName());
+            runTask(session);
         } catch (Throwable t) {
             logger.failedToRunScheduledTask(t, task.getClass().getSimpleName());
 
@@ -58,4 +58,12 @@ public class ScheduledTaskRunner implements Runnable {
         }
     }
 
+    protected void runTask(KeycloakSession session) {
+        session.getTransaction().begin();
+        task.run(session);
+        session.getTransaction().commit();
+
+        logger.debug("Executed scheduled task " + task.getClass().getSimpleName());
+    }
+
 }