lingwei.kong/hawkbit
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

Clustering: Add distributed lock (#2333)

Browse Source 
To support sync of activities in cluster setups

Signed-off-by: Avgustin Marinov <Avgustin.Marinov@bosch.com>
This commit is contained in:
Avgustin Marinov
2025-04-02 12:25:08 +03:00
committed by GitHub
parent 32990ab2ea
commit 2af5439b39
12 changed files with 484 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -13,6 +13,8 @@ import java.util.List;
import java.util.Optional;
import java.util.concurrent.ScheduledExecutorService;
import javax.sql.DataSource;
import jakarta.persistence.EntityManager;
import jakarta.validation.Validation;
@@ -78,6 +80,8 @@ import org.eclipse.hawkbit.repository.jpa.builder.JpaSoftwareModuleMetadataBuild
import org.eclipse.hawkbit.repository.jpa.builder.JpaTargetBuilder;
import org.eclipse.hawkbit.repository.jpa.builder.JpaTargetFilterQueryBuilder;
import org.eclipse.hawkbit.repository.jpa.builder.JpaTargetTypeBuilder;
import org.eclipse.hawkbit.repository.jpa.cluster.LockProperties;
import org.eclipse.hawkbit.repository.jpa.cluster.DistributedLockRepository;
import org.eclipse.hawkbit.repository.jpa.event.JpaEventEntityManager;
import org.eclipse.hawkbit.repository.jpa.executor.AfterTransactionCommitDefaultServiceExecutor;
import org.eclipse.hawkbit.repository.jpa.executor.AfterTransactionCommitExecutor;
@@ -186,6 +190,9 @@ import org.springframework.context.annotation.PropertySource;
import org.springframework.data.domain.AuditorAware;
import org.springframework.data.jpa.repository.config.EnableJpaAuditing;
import org.springframework.data.jpa.repository.config.EnableJpaRepositories;
import org.springframework.integration.jdbc.lock.DefaultLockRepository;
import org.springframework.integration.jdbc.lock.LockRepository;
import org.springframework.integration.support.locks.DefaultLockRegistry;
import org.springframework.integration.support.locks.LockRegistry;
import org.springframework.lang.NonNull;
import org.springframework.retry.annotation.EnableRetry;
@@ -206,7 +213,7 @@ import org.springframework.validation.beanvalidation.MethodValidationPostProcess
@EnableRetry
@EntityScan("org.eclipse.hawkbit.repository.jpa.model")
@PropertySource("classpath:/hawkbit-jpa-defaults.properties")
@Import({ JpaConfiguration.class, RepositoryDefaultConfiguration.class, DataSourceAutoConfiguration.class, SystemManagementCacheKeyGenerator.class })
@Import({ JpaConfiguration.class, RepositoryDefaultConfiguration.class, LockProperties.class, DataSourceAutoConfiguration.class, SystemManagementCacheKeyGenerator.class })
@AutoConfigureAfter(DataSourceAutoConfiguration.class)
public class RepositoryApplicationConfiguration {
@@ -265,6 +272,21 @@ public class RepositoryApplicationConfiguration {
};
}
@Bean
@ConditionalOnMissingBean
LockRepository lockRepository(
final DataSource dataSource, final LockProperties lockProperties, final PlatformTransactionManager txManager) {
final DefaultLockRepository repository = new DistributedLockRepository(dataSource, lockProperties, txManager);
repository.setPrefix("SP_");
return repository;
}
@Bean
@ConditionalOnMissingBean
public LockRegistry lockRegistry() {
return new DefaultLockRegistry();
}
@Bean
@ConditionalOnMissingBean
PauseRolloutGroupAction pauseRolloutGroupAction(final RolloutManagement rolloutManagement,

165
hawkbit-repository/hawkbit-repository-jpa/src/main/java/org/eclipse/hawkbit/repository/jpa/cluster/DistributedLockRepository.java Normal file
View File

@@ -0,0 +1,165 @@
/**
* Copyright (c) 2025 Contributors to the Eclipse Foundation
*
* This program and the accompanying materials are made
* available under the terms of the Eclipse Public License 2.0
* which is available at https://www.eclipse.org/legal/epl-2.0/
*
* SPDX-License-Identifier: EPL-2.0
*/
package org.eclipse.hawkbit.repository.jpa.cluster;
import java.time.Instant;
import java.time.temporal.ChronoUnit;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import javax.sql.DataSource;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.hawkbit.repository.jpa.utils.DeploymentHelper;
import org.springframework.dao.DataIntegrityViolationException;
import org.springframework.dao.DeadlockLoserDataAccessException;
import org.springframework.dao.PessimisticLockingFailureException;
import org.springframework.dao.QueryTimeoutException;
import org.springframework.integration.jdbc.lock.DefaultLockRepository;
import org.springframework.integration.jdbc.lock.JdbcLockRegistry;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionTimedOutException;
import org.springframework.transaction.annotation.Isolation;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
/**
* Repository for {@link JdbcLockRegistry}. This class is not thread safe. Adds support for keeping lock longer then ttl
* if really used by the instance.
*/
@Slf4j
public class DistributedLockRepository extends DefaultLockRepository {
private static final int MAX_DELETE_RETRY = 10;
// period between successive refresh tics
private static final String TIC_PERIOD_MS = "${hawkbit.repository.cluster.lock.ticPeriodMS:2000}";
private final PlatformTransactionManager txManager;
private final int refreshOnRemainMS;
private final int refreshOnRemainPercent;
// if empty refresh is effectively disabled (when both REFRESH_ON_REMAINS_MS and REFRESH_ON_REMAINS_PERCENT are non-positive)
// otherwise, a refresh is triggered at refreshAfterMillis after lock acquisition or last refresh
private Optional<Integer> refreshAfterMillis;
// lock <-> next refresh time
private final Map<String, Instant> lockToRefreshTime = new ConcurrentHashMap<>();
/**
* @param dataSource to use for managing the locks
*/
public DistributedLockRepository(final DataSource dataSource, final LockProperties lockProperties, final PlatformTransactionManager txManager) {
super(dataSource);
this.txManager = txManager;
this.refreshOnRemainMS = lockProperties.getRefreshOnRemainMS();
this.refreshOnRemainPercent = lockProperties.getRefreshOnRemainPercent();
setTimeToLive(lockProperties.getTtl());
}
// interceptor that handles refreshAfterMillis update when time to live is changed
@Override
public void setTimeToLive(final int timeToLive) {
super.setTimeToLive(timeToLive);
refreshAfterMillis = refreshAfterMillis(timeToLive);
}
@Transactional(propagation = Propagation.REQUIRES_NEW)
@Override
public boolean delete(final String lock) {
synchronized (this) {
lockToRefreshTime.remove(lock);
}
return delete(lock, 0);
}
private boolean delete(final String lock, final int count) {
try {
return super.delete(lock);
} catch (final PessimisticLockingFailureException e) {
if (count < MAX_DELETE_RETRY) {
log.debug("Failed to delete cluster lock {}. We try again.", lock, e);
return delete(lock, count + 1);
} else {
log.warn("Failed to delete cluster lock {}!", lock, e);
return false;
}
}
}
@Transactional(propagation=Propagation.NOT_SUPPORTED)
@Override
public boolean acquire(final String lock) {
try {
// real acquisition (made by super.acquire) is made in a new transaction
// because we need to know real (after transaction commit) result Ïto know if it is really successful.
// otherwise the super.acquire will return result before been committed and could be false positive
final boolean acquired = DeploymentHelper.runInNewTransaction(
txManager, "lock-acquire", Isolation.READ_COMMITTED.value(), status -> super.acquire(lock));
if (acquired) {
// update next refresh time
refreshAfterMillis.ifPresent(
afterMillis -> lockToRefreshTime.put(lock, Instant.now().plus(afterMillis, ChronoUnit.MILLIS)));
}
return acquired;
} catch (final DataIntegrityViolationException | DeadlockLoserDataAccessException e) {
log.debug("Could not acquire cluster lock {}. I guess another node has it.", lock, e);
return false;
} catch (final QueryTimeoutException e) {
log.debug("Query timed out for lock {}.", lock, e);
throw new TransactionTimedOutException("DB query timed out for lock " + lock, e);
}
}
@SuppressWarnings({"java:S1066"})
@Scheduled(initialDelayString = TIC_PERIOD_MS, fixedDelayString = TIC_PERIOD_MS)
public void refresh() {
refreshAfterMillis.ifPresentOrElse(afterMillis -> {
final Instant now = Instant.now();
lockToRefreshTime.forEach((lock, refreshTime) -> {
if (now.isAfter(refreshTime)) {
synchronized (this) {
// if delete is called while iterating we must skip record update
// otherwise, the lock will be unavailable for everyone until expiration
if (lockToRefreshTime.containsKey(lock)) {
if (!acquire(lock)) { // try to update record in lock table
log.warn("Failed to refresh cluster lock {}!", lock);
}
}
}
}
});
}, lockToRefreshTime::clear);
}
private Optional<Integer> refreshAfterMillis(final int timeToLive) {
final int triggerOnRemainMS = Math.max(refreshOnRemainMS, timeToLive * refreshOnRemainPercent / 100);
final int refreshAfterMS = timeToLive - triggerOnRemainMS;
return refreshAfterMS <= 0 ? Optional.empty() : Optional.of(refreshAfterMS);
}
// * May be required if the super class doesn't execute in new transactions
// * See https://github.com/spring-projects/spring-integration/issues/3683
// * may be not needed anymore
// @Transactional(propagation = Propagation.REQUIRES_NEW, readOnly = true)
// @Override
// public boolean isAcquired(final String lock) {
// return super.isAcquired(lock);
// }
//
// @Transactional(propagation = Propagation.REQUIRES_NEW)
// @Override
// public void close() {
// super.close();
// }
}

27
hawkbit-repository/hawkbit-repository-jpa/src/main/java/org/eclipse/hawkbit/repository/jpa/cluster/LockProperties.java Normal file
View File

@@ -0,0 +1,27 @@
/**
* Copyright (c) 2025 Contributors to the Eclipse Foundation
*
* This program and the accompanying materials are made
* available under the terms of the Eclipse Public License 2.0
* which is available at https://www.eclipse.org/legal/epl-2.0/
*
* SPDX-License-Identifier: EPL-2.0
*/
package org.eclipse.hawkbit.repository.jpa.cluster;
import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.validation.annotation.Validated;
@Data
@ConfigurationProperties(prefix = "hawkbit.repository.cluster.lock")
@Validated
public class LockProperties {
private int ttl = 5 * 60_000; // 5 minutes
// when less than that time (in milliseconds) remains to lock expiration a refresh is triggered
private int refreshOnRemainMS = 4 * 60_000; // refresh after a minute, 4 minutes before expiration
// when less than that time (in percent of expiration) remains to lock expiration a refresh is triggered
private int refreshOnRemainPercent = 80; // refresh after a minute, 4 minutes before expiration
}

3
hawkbit-repository/hawkbit-repository-jpa/src/main/resources/hawkbit-jpa-defaults.properties
View File

@@ -22,3 +22,6 @@ spring.jpa.properties.eclipselink.logging.level=off
spring.jpa.properties.eclipselink.query-results-cache=false
spring.jpa.properties.eclipselink.cache.shared.default=false
### JPA / Datasource - END
# disables RolloutsLockRepository warnings when other instance has already obtained the lock
logging.level.org.mariadb.jdbc.message.server.ErrorPacket=ERROR