Optimised locking on large scale tenant scenarios. (#477)

* Optimized locking on large scale tenant scenarios.

Signed-off-by: kaizimmerm <kai.zimmermann@bosch-si.com>

* Ensure that a problem in a tenant does not block the others.

Signed-off-by: kaizimmerm <kai.zimmermann@bosch-si.com>

* Memory optimisation.

Signed-off-by: kaizimmerm <kai.zimmermann@bosch-si.com>

hawkbit-repository/hawkbit-repository-jpa/src/main/java/org/eclipse/hawkbit/repository/jpa/JpaRolloutManagement.java

@@ -8,6 +8,7 @@
  */
 package org.eclipse.hawkbit.repository.jpa;
 
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.List;
 import java.util.Map;
@@ -109,6 +110,9 @@ public class JpaRolloutManagement extends AbstractRolloutManagement {
      */
     private static final int TRANSACTION_ACTIONS = 5_000;
 
+    private static final List<RolloutStatus> ACTIVE_ROLLOUTS = Arrays.asList(RolloutStatus.CREATING,
+            RolloutStatus.DELETING, RolloutStatus.STARTING, RolloutStatus.READY, RolloutStatus.RUNNING);
+
     @Autowired
     private RolloutRepository rolloutRepository;
 
@@ -716,31 +720,30 @@ public class JpaRolloutManagement extends AbstractRolloutManagement {
     // No transaction, will be created per handled rollout
     @Transactional(propagation = Propagation.NEVER)
     public void handleRollouts() {
-        rolloutRepository
+        final List<Long> rollouts = rolloutRepository.findByStatusIn(ACTIVE_ROLLOUTS);
-                .findByStatusIn(Lists.newArrayList(RolloutStatus.CREATING, RolloutStatus.DELETING,
-                        RolloutStatus.STARTING, RolloutStatus.READY, RolloutStatus.RUNNING))
-                .forEach(this::handleRollout);
-    }
 
-    private void handleRollout(final Long rolloutId) {
+        if (rollouts.isEmpty()) {
-        LOGGER.debug("handleRollout called for rollout {}", rolloutId);
+            return;
+        }
 
         final String tenant = tenantAware.getCurrentTenant();
 
-        final String handlerId = tenant + "-rollout-" + rolloutId;
+        final String handlerId = tenant + "-rollout";
         final Lock lock = lockRegistry.obtain(handlerId);
         if (!lock.tryLock()) {
             return;
         }
 
         try {
-            runInNewTransaction(handlerId, status -> executeFittingHandler(rolloutId));
+            rollouts.forEach(rolloutId -> runInNewTransaction(handlerId + "-" + rolloutId,
+                    status -> executeFittingHandler(rolloutId)));
         } finally {
             lock.unlock();
         }
     }
 
     private int executeFittingHandler(final Long rolloutId) {
+        LOGGER.debug("handle rollout {}", rolloutId);
         final JpaRollout rollout = rolloutRepository.findOne(rolloutId);
 
         switch (rollout.getStatus()) {

hawkbit-repository/hawkbit-repository-jpa/src/main/java/org/eclipse/hawkbit/repository/jpa/RepositoryApplicationConfiguration.java

@@ -9,7 +9,6 @@
 package org.eclipse.hawkbit.repository.jpa;
 
 import java.util.Map;
-import java.util.concurrent.Executor;
 
 import javax.persistence.EntityManager;
 import javax.sql.DataSource;
@@ -70,7 +69,6 @@ import org.eclipse.hawkbit.tenancy.TenantAware;
 import org.eclipse.hawkbit.tenancy.configuration.TenantConfigurationProperties;
 import org.springframework.beans.factory.ObjectProvider;
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.boot.autoconfigure.orm.jpa.JpaBaseConfiguration;
@@ -545,7 +543,7 @@ public class RepositoryApplicationConfiguration extends JpaBaseConfiguration {
      * @param autoAssignChecker
      *            to run a check as tenant
      * @param lockRegistry
-     *            to lock the tenant for auto assigment
+     *            to lock the tenant for auto assignment
      * @return a new {@link AutoAssignChecker}
      */
     @Bean
@@ -575,8 +573,6 @@ public class RepositoryApplicationConfiguration extends JpaBaseConfiguration {
      *            to run the rollout handler
      * @param systemSecurityContext
      *            to run as system
-     * @param threadPoolExecutor
-     *            to execute the handlers in parallel
      * @return a new {@link RolloutScheduler} bean.
      */
     @Bean
@@ -584,9 +580,7 @@ public class RepositoryApplicationConfiguration extends JpaBaseConfiguration {
     @Profile("!test")
     @ConditionalOnProperty(prefix = "hawkbit.rollout.scheduler", name = "enabled", matchIfMissing = true)
     RolloutScheduler rolloutScheduler(final TenantAware tenantAware, final SystemManagement systemManagement,
-            final RolloutManagement rolloutManagement, final SystemSecurityContext systemSecurityContext,
+            final RolloutManagement rolloutManagement, final SystemSecurityContext systemSecurityContext) {
-            @Qualifier("asyncExecutor") final Executor threadPoolExecutor) {
+        return new RolloutScheduler(tenantAware, systemManagement, rolloutManagement, systemSecurityContext);
-        return new RolloutScheduler(tenantAware, systemManagement, rolloutManagement, systemSecurityContext,
-                threadPoolExecutor);
     }
 }

hawkbit-repository/hawkbit-repository-jpa/src/main/java/org/eclipse/hawkbit/repository/jpa/autoassign/AutoAssignScheduler.java

@@ -84,21 +84,22 @@ public class AutoAssignScheduler {
         // each tenant separately.
         final List<String> tenants = systemManagement.findTenants();
         LOGGER.info("Checking target filter queries for tenants: {}", tenants.size());
-        for (final String tenant : tenants) {
 
-            final Lock lock = lockRegistry.obtain(tenant + "-autoassign");
+        final Lock lock = lockRegistry.obtain("autoassign");
         if (!lock.tryLock()) {
             return null;
         }
+
         try {
+            for (final String tenant : tenants) {
                 tenantAware.runAsTenant(tenant, () -> {
                     autoAssignChecker.check();
                     return null;
                 });
+            }
         } finally {
             lock.unlock();
         }
-        }
         return null;
     }
 }

hawkbit-repository/hawkbit-repository-jpa/src/main/java/org/eclipse/hawkbit/repository/jpa/rollout/RolloutScheduler.java

@@ -9,9 +9,6 @@
 package org.eclipse.hawkbit.repository.jpa.rollout;
 
 import java.util.List;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.Executor;
-import java.util.concurrent.ExecutorCompletionService;
 
 import org.eclipse.hawkbit.repository.RolloutManagement;
 import org.eclipse.hawkbit.repository.SystemManagement;
@@ -21,8 +18,6 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.scheduling.annotation.Scheduled;
 
-import com.google.common.base.Throwables;
-
 /**
  * Scheduler to schedule the {@link RolloutManagement#handleRollouts()}. The
  * delay between the checks be be configured using the property from
@@ -42,8 +37,6 @@ public class RolloutScheduler {
 
     private final SystemSecurityContext systemSecurityContext;
 
-    private final ExecutorCompletionService<Void> completionService;
-
     /**
      * Constructor.
      * 
@@ -55,17 +48,13 @@ public class RolloutScheduler {
      *            to run the rollout handler
      * @param systemSecurityContext
      *            to run as system
-     * @param threadPoolExecutor
-     *            to execute the handlers in parallel
      */
     public RolloutScheduler(final TenantAware tenantAware, final SystemManagement systemManagement,
-            final RolloutManagement rolloutManagement, final SystemSecurityContext systemSecurityContext,
+            final RolloutManagement rolloutManagement, final SystemSecurityContext systemSecurityContext) {
-            final Executor threadPoolExecutor) {
         this.tenantAware = tenantAware;
         this.systemManagement = systemManagement;
         this.rolloutManagement = rolloutManagement;
         this.systemSecurityContext = systemSecurityContext;
-        completionService = new ExecutorCompletionService<>(threadPoolExecutor);
     }
 
     /**
@@ -80,7 +69,7 @@ public class RolloutScheduler {
 
         // run this code in system code privileged to have the necessary
         // permission to query and create entities.
-        final int tasks = systemSecurityContext.runAsSystem(() -> {
+        systemSecurityContext.runAsSystem(() -> {
             // workaround eclipselink that is currently not possible to
             // execute a query without multitenancy if MultiTenant
             // annotation is used.
@@ -90,25 +79,21 @@ public class RolloutScheduler {
             final List<String> tenants = systemManagement.findTenants();
             LOGGER.info("Checking rollouts for {} tenants", tenants.size());
             for (final String tenant : tenants) {
-                completionService.submit(() -> tenantAware.runAsTenant(tenant, () -> {
+                tenantAware.runAsTenant(tenant, () -> {
-                    rolloutManagement.handleRollouts();
-                    return null;
-                }));
-            }
-            return tenants.size();
-        });
-
-        waitUntilHandlersAreComplete(tasks);
-    }
-
-    private void waitUntilHandlersAreComplete(final int tasks) {
                     try {
-            for (int i = 0; i < tasks; i++) {
+                        rolloutManagement.handleRollouts();
-                completionService.take().get();
+                        // We catch all potential runtime exceptions here to
+                        // ensure that not all tenants are blocked if we have a
+                        // problem with a rollout.
+                    } catch (@SuppressWarnings("squid:S1166") final RuntimeException e) {
+                        LOGGER.error("Failed to handle rollouts for tenant {}. I will move on to next tenant.", tenant,
+                                e);
                     }
-        } catch (InterruptedException | ExecutionException e) {
+                    return null;
-            Thread.currentThread().interrupt();
+                });
-            throw Throwables.propagate(e);
             }
+            return null;
+        });
     }
+
 }