Align rollouts and autoassign metrics (#2844)

* Refactor auto-assign locking and metrics * Align rollouts and autoassign metrics Signed-off-by: Avgustin Marinov <Avgustin.Marinov@bosch.com>
2025-12-03 12:34:40 +02:00
parent 977b3fe40c
commit 904c8b180d
12 changed files with 211 additions and 136 deletions
--- a/hawkbit-repository/hawkbit-repository-jpa/src/main/java/org/eclipse/hawkbit/repository/jpa/JpaRepositoryConfiguration.java
+++ b/hawkbit-repository/hawkbit-repository-jpa/src/main/java/org/eclipse/hawkbit/repository/jpa/JpaRepositoryConfiguration.java
@@ -28,7 +28,7 @@ import org.eclipse.hawkbit.ql.jpa.QLSupport;
 import org.eclipse.hawkbit.ql.jpa.QLSupport.NodeTransformer;
 import org.eclipse.hawkbit.ql.jpa.QLSupport.QueryParser;
 import org.eclipse.hawkbit.ql.rsql.RsqlParser;
-import org.eclipse.hawkbit.repository.AutoAssignExecutor;
+import org.eclipse.hawkbit.repository.AutoAssignHandler;
 import org.eclipse.hawkbit.repository.DeploymentManagement;
 import org.eclipse.hawkbit.repository.PropertiesQuotaManagement;
 import org.eclipse.hawkbit.repository.QuotaManagement;
@@ -341,10 +341,10 @@ public class JpaRepositoryConfiguration {
    @Profile("!test")
    @ConditionalOnProperty(prefix = "hawkbit.autoassign.scheduler", name = "enabled", matchIfMissing = true)
    AutoAssignScheduler autoAssignScheduler(
-            final SystemManagement systemManagement, final AutoAssignExecutor autoAssignExecutor,
+            final SystemManagement systemManagement, final AutoAssignHandler autoAssignHandler,
            @Value("${hawkbit.autoassign.executor.thread-pool.size:1}") final int threadPoolSize,
            final LockRegistry lockRegistry, final Optional<MeterRegistry> meterRegistry) {
-        return new AutoAssignScheduler(systemManagement, autoAssignExecutor, threadPoolSize, lockRegistry, meterRegistry);
+        return new AutoAssignScheduler(systemManagement, autoAssignHandler, threadPoolSize, meterRegistry);
    }

    /**
--- a/hawkbit-repository/hawkbit-repository-jpa/src/main/java/org/eclipse/hawkbit/repository/jpa/scheduler/AutoAssignScheduler.java
+++ b/hawkbit-repository/hawkbit-repository-jpa/src/main/java/org/eclipse/hawkbit/repository/jpa/scheduler/AutoAssignScheduler.java
@@ -14,14 +14,11 @@ import static org.eclipse.hawkbit.context.AccessContext.asSystemAsTenant;

 import java.util.Optional;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.locks.Lock;

 import io.micrometer.core.instrument.MeterRegistry;
 import lombok.extern.slf4j.Slf4j;
-import org.eclipse.hawkbit.repository.AutoAssignExecutor;
+import org.eclipse.hawkbit.repository.AutoAssignHandler;
 import org.eclipse.hawkbit.repository.SystemManagement;
-import org.eclipse.hawkbit.tenancy.DefaultTenantConfiguration;
-import org.springframework.integration.support.locks.LockRegistry;
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

@@ -34,18 +31,16 @@ public class AutoAssignScheduler {
    private static final String PROP_SCHEDULER_DELAY_PLACEHOLDER = "${hawkbit.autoassign.scheduler.fixedDelay:2000}";

    private final SystemManagement systemManagement;
-    private final AutoAssignExecutor autoAssignExecutor;
-    private final LockRegistry lockRegistry;
+    private final AutoAssignHandler autoAssignHandler;
    private final Optional<MeterRegistry> meterRegistry;
+
    private final ThreadPoolTaskExecutor autoAssignTaskExecutor;

    public AutoAssignScheduler(
-            final SystemManagement systemManagement, final AutoAssignExecutor autoAssignExecutor,
-            final int threadPoolSize,
-            final LockRegistry lockRegistry, final Optional<MeterRegistry> meterRegistry) {
+            final SystemManagement systemManagement, final AutoAssignHandler autoAssignHandler,
+            final int threadPoolSize, final Optional<MeterRegistry> meterRegistry) {
        this.systemManagement = systemManagement;
-        this.autoAssignExecutor = autoAssignExecutor;
-        this.lockRegistry = lockRegistry;
+        this.autoAssignHandler = autoAssignHandler;
        this.meterRegistry = meterRegistry;
        autoAssignTaskExecutor = SchedulerUtils.threadPoolTaskExecutor("auto-assign-exec-", threadPoolSize);
    }
@@ -58,7 +53,8 @@ public class AutoAssignScheduler {
    public void autoAssignScheduler() {
        // run this code in system code privileged to have the necessary permission to query and create entities.
        log.debug("Triggered auto-assign scheduler.");
-        final long startNano = java.lang.System.nanoTime();
+        final long startNano = System.nanoTime();
+
        asSystem(() ->
                systemManagement.forEachTenantAsSystem(tenant -> {
                    if (autoAssignTaskExecutor == null) {// sync
@@ -68,31 +64,19 @@ public class AutoAssignScheduler {
                    }
                })
        );
-        meterRegistry
-                .map(mReg -> mReg.timer("hawkbit.autoassign.scheduler.all"))
-                .ifPresent(timer -> timer.record(java.lang.System.nanoTime() - startNano, TimeUnit.NANOSECONDS));
+
+        meterRegistry // handle all tenants (some could be skipped if lock could not be obtained)
+                .map(mReg -> mReg.timer("hawkbit.autoassign.scheduler"))
+                .ifPresent(timer -> timer.record(System.nanoTime() - startNano, TimeUnit.NANOSECONDS));
        log.debug("Finished auto-assign scheduler run.");
    }

    private void handleAll(final String tenant) {
-        final Lock lock = lockRegistry.obtain(createAutoAssignmentLockKey(tenant));
-        if (!lock.tryLock()) {
-            return;
-        }
-        final long startNanoT = System.nanoTime();
+        log.trace("Handling auto-assignments for tenant: {}", tenant);
        try {
-            autoAssignExecutor.checkAllTargets();
-        } finally {
-            lock.unlock();
-            meterRegistry
-                    .map(mReg -> mReg.timer(
-                            "hawkbit.autoassign.scheduler",
-                            DefaultTenantConfiguration.TENANT_TAG, tenant))
-                    .ifPresent(timer -> timer.record(System.nanoTime() - startNanoT, TimeUnit.NANOSECONDS));
+            autoAssignHandler.handleAll();
+        } catch (final Exception e) {
+            log.error("Error auto-assignments rollout for tenant {}", tenant, e);
        }
    }
-
-    private static String createAutoAssignmentLockKey(final String tenant) {
-        return tenant + "-auto-assign";
-    }
 }
--- a/hawkbit-repository/hawkbit-repository-jpa/src/main/java/org/eclipse/hawkbit/repository/jpa/scheduler/JpaAutoAssignExecutor.java
+++ b/hawkbit-repository/hawkbit-repository-jpa/src/main/java/org/eclipse/hawkbit/repository/jpa/scheduler/JpaAutoAssignExecutor.java
@@ -11,16 +11,24 @@ package org.eclipse.hawkbit.repository.jpa.scheduler;

 import static org.eclipse.hawkbit.context.AccessContext.asActor;
 import static org.eclipse.hawkbit.context.AccessContext.withSecurityContext;
+import static org.eclipse.hawkbit.tenancy.DefaultTenantConfiguration.TENANT_TAG;

 import java.util.Collections;
 import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.Lock;
 import java.util.function.Consumer;

+import jakarta.persistence.LockTimeoutException;
 import jakarta.persistence.PersistenceException;

+import io.micrometer.core.instrument.MeterRegistry;
 import lombok.extern.slf4j.Slf4j;
+import org.eclipse.hawkbit.context.AccessContext;
 import org.eclipse.hawkbit.exception.AbstractServerRtException;
-import org.eclipse.hawkbit.repository.AutoAssignExecutor;
+import org.eclipse.hawkbit.repository.AutoAssignHandler;
 import org.eclipse.hawkbit.repository.DeploymentManagement;
 import org.eclipse.hawkbit.repository.TargetFilterQueryManagement;
 import org.eclipse.hawkbit.repository.TargetManagement;
@@ -33,6 +41,7 @@ import org.eclipse.hawkbit.repository.model.TargetFilterQuery;
 import org.springframework.data.domain.PageRequest;
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.domain.Slice;
+import org.springframework.integration.support.locks.LockRegistry;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.PlatformTransactionManager;
 import org.springframework.transaction.annotation.Isolation;
@@ -47,7 +56,7 @@ import org.springframework.util.StringUtils;
 */
@Slf4j
@Service
-public class JpaAutoAssignExecutor implements AutoAssignExecutor {
+public class JpaAutoAssignHandler implements AutoAssignHandler {

    /**
     * The message which is added to the action status when a distribution set is assigned to a target.
@@ -60,33 +69,98 @@ public class JpaAutoAssignExecutor implements AutoAssignExecutor {
     */
    private static final int PAGE_SIZE = 1000;

+    private static final LockTimeoutException LOCK_TIMEOUT_EXCEPTION = new LockTimeoutException("Could not obtain lock for auto assignment");
+
    private final TargetFilterQueryManagement<? extends TargetFilterQuery> targetFilterQueryManagement;
    private final TargetManagement<? extends Target> targetManagement;
    private final DeploymentManagement deploymentManagement;
    private final PlatformTransactionManager transactionManager;
+    private final LockRegistry lockRegistry;
+    private final Optional<MeterRegistry> meterRegistry;

-    public JpaAutoAssignExecutor(
+    public JpaAutoAssignHandler(
            final TargetFilterQueryManagement<? extends TargetFilterQuery> targetFilterQueryManagement,
            final TargetManagement<? extends Target> targetManagement, final DeploymentManagement deploymentManagement,
-            final PlatformTransactionManager transactionManager) {
+            final PlatformTransactionManager transactionManager, final LockRegistry lockRegistry, final Optional<MeterRegistry> meterRegistry) {
        this.targetFilterQueryManagement = targetFilterQueryManagement;
        this.targetManagement = targetManagement;
        this.deploymentManagement = deploymentManagement;
        this.transactionManager = transactionManager;
+        this.lockRegistry = lockRegistry;
+        this.meterRegistry = meterRegistry;
    }

    @Override
    @Transactional(propagation = Propagation.REQUIRES_NEW)
-    public void checkAllTargets() {
-        log.debug("Auto assign check call started");
-        forEachFilterWithAutoAssignDS(this::checkByTargetFilterQueryAndAssignDS);
-        log.debug("Auto assign check call finished");
+    public void handleAll() {
+        final long startNano = System.nanoTime();
+
+        final AtomicReference<Lock> lockRef = new AtomicReference<>();
+        try {
+            forEachFilterWithAutoAssignDS(targetFilterQuery -> {
+                if (lockRef.get() == null) {
+                    // only if there are targetFilterQueries to process we try to obtain the lock (on the first one)
+                    final Lock lock = lockRegistry.obtain(createAutoAssignmentLockKey(AccessContext.tenant()));
+                    if (!lock.tryLock()) {
+                        if (log.isTraceEnabled()) {
+                            log.trace("Could not obtain lock {}", lock);
+                        }
+                        throw LOCK_TIMEOUT_EXCEPTION;
+                    } else {
+                        lockRef.set(lock);
+
+                        log.debug("Start auto-assign handling");
+                    }
+                }
+
+                final long startNanoPartial = System.nanoTime();
+                try {
+                    checkByTargetFilterQueryAndAssignDS(targetFilterQuery);
+                } finally {
+                    meterRegistry // handle single targetFilterQuery
+                            .map(mReg -> mReg.timer(
+                                    "hawkbit.autoassign.handle", TENANT_TAG, AccessContext.tenant(), "targetFilterQuery",
+                                    String.valueOf(targetFilterQuery.getId())))
+                            .ifPresent(timer -> timer.record(System.nanoTime() - startNanoPartial, TimeUnit.NANOSECONDS));
+                }
+            });
+        } finally {
+            final Lock lock = lockRef.get();
+            if (lock != null) {
+                lock.unlock();
+
+                // only if there is at least one targetFilterQuery and lock has been obtained then will be measured (as in rollouts)
+                meterRegistry // handle single targetFilterQuery for single target
+                        .map(mReg -> mReg.timer(
+                                "hawkbit.autoassign.handle.all", TENANT_TAG, AccessContext.tenant()))
+                        .ifPresent(timer -> timer.record(System.nanoTime() - startNano, TimeUnit.NANOSECONDS));
+                log.debug("Auto assign check all targets finished");
+            }
+        }
    }

    @Override
-    public void checkSingleTarget(final String controllerId) {
+    public void handleSingleTarget(final String controllerId) {
        log.debug("Auto assign check call for device {} started", controllerId);
-        forEachFilterWithAutoAssignDS(filter -> checkForDevice(controllerId, filter));
+        final long startNano = System.nanoTime();
+
+        forEachFilterWithAutoAssignDS(targetFilterQuery -> {
+            final long startNanoPartial = System.nanoTime();
+            try {
+                checkForDevice(controllerId, targetFilterQuery);
+            } finally {
+                meterRegistry // handle single targetFilterQuery for single target
+                        .map(mReg -> mReg.timer(
+                                "hawkbit.autoassign.handle.single", TENANT_TAG, AccessContext.tenant(), "targetFilterQuery",
+                                String.valueOf(targetFilterQuery.getId())))
+                        .ifPresent(timer -> timer.record(System.nanoTime() - startNanoPartial, TimeUnit.NANOSECONDS));
+            }
+        });
+
+        meterRegistry // handle single targetFilterQuery for single target
+                .map(mReg -> mReg.timer(
+                        "hawkbit.autoassign.handle.single.all", TENANT_TAG, AccessContext.tenant()))
+                .ifPresent(timer -> timer.record(System.nanoTime() - startNano, TimeUnit.NANOSECONDS));
        log.debug("Auto assign check call for device {} finished", controllerId);
    }

@@ -134,25 +208,32 @@ public class JpaAutoAssignExecutor implements AutoAssignExecutor {
        do {
            filterQueries = targetFilterQueryManagement.findWithAutoAssignDS(query);

-            filterQueries.forEach(filterQuery -> {
-                try {
-                    filterQuery.getAccessControlContext().ifPresentOrElse(
-                            // has stored context - executes it with it
-                            context -> withSecurityContext(context, () -> consumer.accept(filterQuery)),
-                            // has no stored context - executes it in the tenant & user scope
-                            () -> asActor(getAutoAssignmentInitiatedBy(filterQuery), () -> consumer.accept(filterQuery))
-                    );
-                } catch (final RuntimeException ex) {
-                    if (log.isDebugEnabled()) {
-                        log.debug("Exception on forEachFilterWithAutoAssignDS execution for filter id {}. Continue with next filter query.",
-                                filterQuery.getId(), ex);
-                    } else {
-                        log.error(
-                                "Exception on forEachFilterWithAutoAssignDS execution for filter id {} and error message [{}]. Continue with next filter query.",
-                                filterQuery.getId(), ex.getMessage());
+            try {
+                filterQueries.forEach(filterQuery -> {
+                    try {
+                        filterQuery.getAccessControlContext().ifPresentOrElse(
+                                // has stored context - executes it with it
+                                context -> withSecurityContext(context, () -> consumer.accept(filterQuery)),
+                                // has no stored context - executes it in the tenant & user scope
+                                () -> asActor(getAutoAssignmentInitiatedBy(filterQuery), () -> consumer.accept(filterQuery)));
+                    } catch (final RuntimeException ex) {
+                        if (ex == LOCK_TIMEOUT_EXCEPTION) {
+                            // expected - just stop processing further
+                            throw ex; // throw in order to break
+                        }
+                        if (log.isDebugEnabled()) {
+                            log.debug("Exception on forEachFilterWithAutoAssignDS execution for filter id {}. Continue with next filter query.",
+                                    filterQuery.getId(), ex);
+                        } else {
+                            log.error(
+                                    "Exception on forEachFilterWithAutoAssignDS execution for filter id {} and error message [{}]. Continue with next filter query.",
+                                    filterQuery.getId(), ex.getMessage());
+                        }
                    }
-                }
-            });
+                });
+            } catch (final LockTimeoutException lte) {
+                break; // lock not found
+            }
        } while (filterQueries.hasNext() && (query = filterQueries.nextPageable()) != Pageable.unpaged());
    }

@@ -209,4 +290,8 @@ public class JpaAutoAssignExecutor implements AutoAssignExecutor {
        }
        log.debug("Auto assign check call for target filter query id {} for device {} finished", targetFilterQuery.getId(), controllerId);
    }
+
+    private static String createAutoAssignmentLockKey(final String tenant) {
+        return tenant + "-auto-assign";
+    }
 }
--- a/hawkbit-repository/hawkbit-repository-jpa/src/main/java/org/eclipse/hawkbit/repository/jpa/scheduler/JpaRolloutHandler.java
+++ b/hawkbit-repository/hawkbit-repository-jpa/src/main/java/org/eclipse/hawkbit/repository/jpa/scheduler/JpaRolloutHandler.java
@@ -9,6 +9,8 @@
 */
 package org.eclipse.hawkbit.repository.jpa.scheduler;

+import static org.eclipse.hawkbit.tenancy.DefaultTenantConfiguration.TENANT_TAG;
+
 import java.util.List;
 import java.util.Optional;
 import java.util.concurrent.TimeUnit;
@@ -21,7 +23,6 @@ import org.eclipse.hawkbit.repository.RolloutExecutor;
 import org.eclipse.hawkbit.repository.RolloutHandler;
 import org.eclipse.hawkbit.repository.RolloutManagement;
 import org.eclipse.hawkbit.repository.jpa.utils.DeploymentHelper;
-import org.eclipse.hawkbit.tenancy.DefaultTenantConfiguration;
 import org.springframework.integration.support.locks.LockRegistry;
 import org.springframework.transaction.PlatformTransactionManager;

@@ -57,6 +58,8 @@ public class JpaRolloutHandler implements RolloutHandler {

    @Override
    public void handleAll() {
+        final long startNano = System.nanoTime();
+
        final List<Long> rollouts = rolloutManagement.findActiveRollouts();
        if (rollouts.isEmpty()) {
            return;
@@ -66,15 +69,13 @@ public class JpaRolloutHandler implements RolloutHandler {
        final Lock lock = lockRegistry.obtain(handlerId);
        if (!lock.tryLock()) {
            if (log.isTraceEnabled()) {
-                log.trace("Could not perform lock {}", lock);
+                log.trace("Could not obtain lock {}", lock);
            }
            return;
        }

        try {
-            log.debug("Trigger handling {} rollouts.", rollouts.size());
-
-            final long startNano = System.nanoTime();
+            log.debug("Start handling {} rollouts.", rollouts.size());
            rollouts.forEach(rolloutId -> {
                try {
                    handleRolloutInNewTransaction(rolloutId, handlerId);
@@ -82,16 +83,16 @@ public class JpaRolloutHandler implements RolloutHandler {
                    log.error("Failed to process rollout with id {}", rolloutId, throwable);
                }
            });
-            meterRegistry
-                    .map(mReg -> mReg.timer("hawkbit.rollout.handler.all", DefaultTenantConfiguration.TENANT_TAG, AccessContext.tenant()))
-                    .ifPresent(timer -> timer.record(System.nanoTime() - startNano, TimeUnit.NANOSECONDS));
-
            log.debug("Finished handling of the rollouts.");
        } finally {
+            lock.unlock();
+
            if (log.isTraceEnabled()) {
                log.trace("Unlock lock {}", lock);
            }
-            lock.unlock();
+            meterRegistry // handle all rollouts of a tenant
+                    .map(mReg -> mReg.timer("hawkbit.rollout.handle.all", TENANT_TAG, AccessContext.tenant()))
+                    .ifPresent(timer -> timer.record(System.nanoTime() - startNano, TimeUnit.NANOSECONDS));
        }
    }

@@ -110,11 +111,9 @@ public class JpaRolloutHandler implements RolloutHandler {
            return 0L;
        });

-        meterRegistry
+        meterRegistry // handle single rollout
                .map(mReg -> mReg.timer(
-                        "hawkbit.rollout.handler",
-                        DefaultTenantConfiguration.TENANT_TAG, AccessContext.tenant(),
-                        "rollout", String.valueOf(rolloutId)))
+                        "hawkbit.rollout.handle", TENANT_TAG, AccessContext.tenant(), "rollout", String.valueOf(rolloutId)))
                .ifPresent(timer -> timer.record(System.nanoTime() - startNano, TimeUnit.NANOSECONDS));
    }
 }
--- a/hawkbit-repository/hawkbit-repository-jpa/src/main/java/org/eclipse/hawkbit/repository/jpa/scheduler/RolloutScheduler.java
+++ b/hawkbit-repository/hawkbit-repository-jpa/src/main/java/org/eclipse/hawkbit/repository/jpa/scheduler/RolloutScheduler.java
@@ -19,7 +19,6 @@ import io.micrometer.core.instrument.MeterRegistry;
 import lombok.extern.slf4j.Slf4j;
 import org.eclipse.hawkbit.repository.RolloutHandler;
 import org.eclipse.hawkbit.repository.SystemManagement;
-import org.eclipse.hawkbit.tenancy.DefaultTenantConfiguration;
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

@@ -54,8 +53,8 @@ public class RolloutScheduler {
     */
    @Scheduled(initialDelayString = PROP_SCHEDULER_DELAY_PLACEHOLDER, fixedDelayString = PROP_SCHEDULER_DELAY_PLACEHOLDER)
    public void runningRolloutScheduler() {
-        log.debug("rollout schedule checker has been triggered.");
-        final long startNano = java.lang.System.nanoTime();
+        log.debug("Rollout scheduler has been triggered.");
+        final long startNano = System.nanoTime();

        // run this code in system code privileged to have the necessary system code permission execute forEachTenant
        asSystem(() ->
@@ -66,28 +65,21 @@ public class RolloutScheduler {
                    if (rolloutTaskExecutor == null) {
                        handleAll(tenant);
                    } else {
-                        rolloutTaskExecutor.submit(() -> asSystemAsTenant(tenant, () -> handleAll(tenant)));
+                        rolloutTaskExecutor.execute(() -> asSystemAsTenant(tenant, () -> handleAll(tenant)));
                    }
                }));

-        meterRegistry
-                .map(mReg -> mReg.timer("hawkbit.rollout.scheduler.all"))
-                .ifPresent(timer -> timer.record(java.lang.System.nanoTime() - startNano, TimeUnit.NANOSECONDS));
+        meterRegistry // handle all tenants (some could be skipped if lock could not be obtained)
+                .map(mReg -> mReg.timer("hawkbit.rollout.scheduler"))
+                .ifPresent(timer -> timer.record(System.nanoTime() - startNano, TimeUnit.NANOSECONDS));
    }

    private void handleAll(final String tenant) {
-        log.trace("Handling rollout for tenant: {}", tenant);
-        final long startNano = java.lang.System.nanoTime();
-
+        log.trace("Handling rollouts for tenant: {}", tenant);
        try {
            rolloutHandler.handleAll();
        } catch (final Exception e) {
            log.error("Error processing rollout for tenant {}", tenant, e);
        }
-
-        meterRegistry
-                .map(mReg -> mReg.timer("hawkbit.rollout.scheduler", DefaultTenantConfiguration.TENANT_TAG, tenant))
-                .ifPresent(timer -> timer.record(java.lang.System.nanoTime() - startNano, TimeUnit.NANOSECONDS));
    }
-
 }