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

Improve artifact binary cleanup - only after commit (#2134)

Browse Source 
Signed-off-by: Avgustin Marinov <Avgustin.Marinov@bosch.com>
This commit is contained in:
Avgustin Marinov
2024-12-10 13:41:35 +02:00
committed by GitHub
parent d0c952f236
commit d3eeb71826
8 changed files with 128 additions and 120 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
hawkbit-repository/hawkbit-repository-jpa/src/main/java/org/eclipse/hawkbit/repository/jpa/JpaRolloutExecutor.java
View File

@@ -476,9 +476,8 @@ public class JpaRolloutExecutor implements RolloutExecutor {
}
private void updateTotalTargetCount(final JpaRolloutGroup rolloutGroup, final long countTargetsOfRolloutGroup) {
final JpaRollout jpaRollout = (JpaRollout) rolloutGroup.getRollout();
final long updatedTargetCount = jpaRollout.getTotalTargets()
- (rolloutGroup.getTotalTargets() - countTargetsOfRolloutGroup);
final JpaRollout jpaRollout = rolloutGroup.getRollout();
final long updatedTargetCount = jpaRollout.getTotalTargets() - (rolloutGroup.getTotalTargets() - countTargetsOfRolloutGroup);
jpaRollout.setTotalTargets(updatedTargetCount);
rolloutGroup.setTotalTargets((int) countTargetsOfRolloutGroup);
rolloutRepository.save(jpaRollout);
@@ -574,12 +573,8 @@ public class JpaRolloutExecutor implements RolloutExecutor {
private boolean ensureAllGroupsAreScheduled(final Rollout rollout) {
final JpaRollout jpaRollout = (JpaRollout) rollout;
final List<JpaRolloutGroup> groupsToBeScheduled = rolloutGroupRepository.findByRolloutAndStatus(rollout,
RolloutGroupStatus.READY);
final long scheduledGroups = groupsToBeScheduled.stream()
.filter(group -> scheduleRolloutGroup(jpaRollout, group)).count();
final List<JpaRolloutGroup> groupsToBeScheduled = rolloutGroupRepository.findByRolloutAndStatus(rollout, RolloutGroupStatus.READY);
final long scheduledGroups = groupsToBeScheduled.stream().filter(group -> scheduleRolloutGroup(jpaRollout, group)).count();
return scheduledGroups == groupsToBeScheduled.size();
}
@@ -630,8 +625,8 @@ public class JpaRolloutExecutor implements RolloutExecutor {
do {
// Add up to TRANSACTION_TARGETS of the left targets
// In case a TransactionException is thrown this loop aborts
final long assigned = assignTargetsToGroupInNewTransaction(rollout, group, groupTargetFilter,
Math.min(TRANSACTION_TARGETS, targetsLeftToAdd));
final long assigned = assignTargetsToGroupInNewTransaction(
rollout, group, groupTargetFilter, Math.min(TRANSACTION_TARGETS, targetsLeftToAdd));
if (assigned == 0) {
break; // percent > 100 or some could have disappeared
} else {

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

@@ -357,8 +357,8 @@ public class RepositoryApplicationConfiguration extends JpaBaseConfiguration {
final List<RolloutGroupConditionEvaluator<RolloutGroup.RolloutGroupSuccessCondition>> successConditionEvaluators,
final List<RolloutGroupActionEvaluator<RolloutGroup.RolloutGroupErrorAction>> errorActionEvaluators,
final List<RolloutGroupActionEvaluator<RolloutGroup.RolloutGroupSuccessAction>> successActionEvaluators) {
return new RolloutGroupEvaluationManager(errorConditionEvaluators, successConditionEvaluators,
errorActionEvaluators, successActionEvaluators);
return new RolloutGroupEvaluationManager(
errorConditionEvaluators, successConditionEvaluators, errorActionEvaluators, successActionEvaluators);
}
@Bean
@@ -696,8 +696,8 @@ public class RepositoryApplicationConfiguration extends JpaBaseConfiguration {
final DistributionSetTagRepository distributionSetTagRepository,
final DistributionSetRepository distributionSetRepository,
final VirtualPropertyReplacer virtualPropertyReplacer, final JpaProperties properties) {
return new JpaDistributionSetTagManagement(distributionSetTagRepository, distributionSetRepository,
virtualPropertyReplacer, properties.getDatabase());
return new JpaDistributionSetTagManagement(
distributionSetTagRepository, distributionSetRepository, virtualPropertyReplacer, properties.getDatabase());
}
/**
@@ -735,8 +735,7 @@ public class RepositoryApplicationConfiguration extends JpaBaseConfiguration {
final SoftwareModuleRepository softwareModuleRepository,
final JpaProperties properties) {
return new JpaSoftwareModuleTypeManagement(distributionSetTypeRepository, softwareModuleTypeRepository,
virtualPropertyReplacer, softwareModuleRepository,
properties.getDatabase());
virtualPropertyReplacer, softwareModuleRepository, properties.getDatabase());
}
@Bean
@@ -744,8 +743,7 @@ public class RepositoryApplicationConfiguration extends JpaBaseConfiguration {
RolloutHandler rolloutHandler(final TenantAware tenantAware, final RolloutManagement rolloutManagement,
final RolloutExecutor rolloutExecutor, final LockRegistry lockRegistry,
final PlatformTransactionManager txManager, final ContextAware contextAware) {
return new JpaRolloutHandler(tenantAware, rolloutManagement, rolloutExecutor, lockRegistry, txManager,
contextAware);
return new JpaRolloutHandler(tenantAware, rolloutManagement, rolloutExecutor, lockRegistry, txManager, contextAware);
}
@Bean
@@ -777,8 +775,7 @@ public class RepositoryApplicationConfiguration extends JpaBaseConfiguration {
final ContextAware contextAware) {
return new JpaRolloutManagement(targetManagement, distributionSetManagement, eventPublisherHolder,
virtualPropertyReplacer, properties.getDatabase(), rolloutApprovalStrategy,
tenantConfigurationManagement, systemSecurityContext,
contextAware);
tenantConfigurationManagement, systemSecurityContext, contextAware);
}
/**
@@ -860,10 +857,12 @@ public class RepositoryApplicationConfiguration extends JpaBaseConfiguration {
@Bean
@ConditionalOnMissingBean
ArtifactManagement artifactManagement(
final EntityManager entityManager, final LocalArtifactRepository localArtifactRepository,
final SoftwareModuleRepository softwareModuleRepository, final Optional<ArtifactRepository> artifactRepository,
final EntityManager entityManager, final PlatformTransactionManager txManager,
final LocalArtifactRepository localArtifactRepository, final SoftwareModuleRepository softwareModuleRepository,
final Optional<ArtifactRepository> artifactRepository,
final QuotaManagement quotaManagement, final TenantAware tenantAware) {
return new JpaArtifactManagement(entityManager, localArtifactRepository, softwareModuleRepository, artifactRepository.orElse(null),
return new JpaArtifactManagement(
entityManager, txManager, localArtifactRepository, softwareModuleRepository, artifactRepository.orElse(null),
quotaManagement, tenantAware);
}

19
hawkbit-repository/hawkbit-repository-jpa/src/main/java/org/eclipse/hawkbit/repository/jpa/executor/AfterTransactionCommitDefaultServiceExecutor.java
View File

@@ -13,15 +13,15 @@ import java.util.ArrayList;
import java.util.List;
import lombok.extern.slf4j.Slf4j;
import org.springframework.transaction.support.TransactionSynchronizationAdapter;
import org.springframework.transaction.support.TransactionSynchronization;
import org.springframework.transaction.support.TransactionSynchronizationManager;
/**
* A Service which calls register runnable. This runnables will executed after a
* successful spring transaction commit.The class is thread safe.
* A Service which calls register runnable. This runnables will be executed after a successful spring transaction commit.
* The class is thread safe.
*/
@Slf4j
public class AfterTransactionCommitDefaultServiceExecutor extends TransactionSynchronizationAdapter implements AfterTransactionCommitExecutor {
public class AfterTransactionCommitDefaultServiceExecutor implements TransactionSynchronization, AfterTransactionCommitExecutor {
private static final ThreadLocal<List<Runnable>> THREAD_LOCAL_RUNNABLES = new ThreadLocal<>();
@@ -30,6 +30,14 @@ public class AfterTransactionCommitDefaultServiceExecutor extends TransactionSyn
@SuppressWarnings({ "squid:S1217" })
public void afterCommit() {
final List<Runnable> afterCommitRunnables = THREAD_LOCAL_RUNNABLES.get();
if (afterCommitRunnables == null) {
log.trace("Transaction successfully committed, runnables is null");
return;
}
// removes the runnables that will process, so they would be able to start new transactions and
// inserting new after commit hooks
THREAD_LOCAL_RUNNABLES.remove();
log.debug("Transaction successfully committed, executing {} runnables", afterCommitRunnables.size());
for (final Runnable afterCommitRunnable : afterCommitRunnables) {
log.debug("Executing runnable {}", afterCommitRunnable);
@@ -44,8 +52,7 @@ public class AfterTransactionCommitDefaultServiceExecutor extends TransactionSyn
@Override
@SuppressWarnings({ "squid:S1217" })
public void afterCompletion(final int status) {
final String transactionStatus = status == STATUS_COMMITTED ? "COMMITTED" : "ROLLEDBACK";
log.debug("Transaction completed after commit with status {}", transactionStatus);
log.debug("Transaction completed after commit with status {}", status == STATUS_COMMITTED ? "COMMITTED" : "ROLLEDBACK");
THREAD_LOCAL_RUNNABLES.remove();
}

58
hawkbit-repository/hawkbit-repository-jpa/src/main/java/org/eclipse/hawkbit/repository/jpa/management/JpaArtifactManagement.java
View File

@@ -36,26 +36,31 @@ import org.eclipse.hawkbit.repository.exception.InvalidMD5HashException;
import org.eclipse.hawkbit.repository.exception.InvalidSHA1HashException;
import org.eclipse.hawkbit.repository.exception.InvalidSHA256HashException;
import org.eclipse.hawkbit.repository.jpa.EncryptionAwareDbArtifact;
import org.eclipse.hawkbit.repository.jpa.Jpa;
import org.eclipse.hawkbit.repository.jpa.JpaManagementHelper;
import org.eclipse.hawkbit.repository.jpa.acm.AccessController;
import org.eclipse.hawkbit.repository.jpa.configuration.Constants;
import org.eclipse.hawkbit.repository.jpa.model.JpaArtifact;
import org.eclipse.hawkbit.repository.jpa.model.JpaSoftwareModule;
import org.eclipse.hawkbit.repository.jpa.model.helper.AfterTransactionCommitExecutorHolder;
import org.eclipse.hawkbit.repository.jpa.repository.LocalArtifactRepository;
import org.eclipse.hawkbit.repository.jpa.repository.SoftwareModuleRepository;
import org.eclipse.hawkbit.repository.jpa.specifications.ArtifactSpecifications;
import org.eclipse.hawkbit.repository.jpa.utils.DeploymentHelper;
import org.eclipse.hawkbit.repository.jpa.utils.FileSizeAndStorageQuotaCheckingInputStream;
import org.eclipse.hawkbit.repository.jpa.utils.QuotaHelper;
import org.eclipse.hawkbit.repository.model.Artifact;
import org.eclipse.hawkbit.repository.model.ArtifactUpload;
import org.eclipse.hawkbit.repository.model.SoftwareModule;
import org.eclipse.hawkbit.tenancy.TenantAware;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.ConcurrencyFailureException;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.Pageable;
import org.springframework.retry.annotation.Backoff;
import org.springframework.retry.annotation.Retryable;
import org.springframework.security.access.prepost.PreAuthorize;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.support.TransactionSynchronization;
import org.springframework.transaction.support.TransactionSynchronizationManager;
@@ -70,6 +75,7 @@ import org.springframework.validation.annotation.Validated;
public class JpaArtifactManagement implements ArtifactManagement {
private final EntityManager entityManager;
private final PlatformTransactionManager txManager;
private final LocalArtifactRepository localArtifactRepository;
private final SoftwareModuleRepository softwareModuleRepository;
@Nullable
@@ -79,10 +85,13 @@ public class JpaArtifactManagement implements ArtifactManagement {
public JpaArtifactManagement(
final EntityManager entityManager,
final PlatformTransactionManager txManager,
final LocalArtifactRepository localArtifactRepository,
final SoftwareModuleRepository softwareModuleRepository, @Nullable final ArtifactRepository artifactRepository,
final QuotaManagement quotaManagement, final TenantAware tenantAware) {
final QuotaManagement quotaManagement,
final TenantAware tenantAware) {
this.entityManager = entityManager;
this.txManager = txManager;
this.localArtifactRepository = localArtifactRepository;
this.softwareModuleRepository = softwareModuleRepository;
this.artifactRepository = artifactRepository;
@@ -137,18 +146,19 @@ public class JpaArtifactManagement implements ArtifactManagement {
@Retryable(retryFor = { ConcurrencyFailureException.class }, maxAttempts = Constants.TX_RT_MAX,
backoff = @Backoff(delay = Constants.TX_RT_DELAY))
public void delete(final long id) {
final JpaArtifact toDelete = (JpaArtifact) get(id)
.orElseThrow(() -> new EntityNotFoundException(Artifact.class, id));
final JpaArtifact toDelete = (JpaArtifact) get(id).orElseThrow(() -> new EntityNotFoundException(Artifact.class, id));
final JpaSoftwareModule softwareModule = toDelete.getSoftwareModule();
// clearArtifactBinary checks (unconditionally) software module UPDATE access
softwareModuleRepository.getAccessController().ifPresent(accessController ->
accessController.assertOperationAllowed(AccessController.Operation.UPDATE,
(JpaSoftwareModule) toDelete.getSoftwareModule()));
((JpaSoftwareModule) toDelete.getSoftwareModule()).removeArtifact(toDelete);
softwareModuleRepository.save((JpaSoftwareModule) toDelete.getSoftwareModule());
accessController.assertOperationAllowed(AccessController.Operation.UPDATE, softwareModule));
softwareModule.removeArtifact(toDelete);
softwareModuleRepository.save(softwareModule);
localArtifactRepository.deleteById(id);
clearArtifactBinary(toDelete.getSha1Hash());
final String sha1Hash = toDelete.getSha1Hash();
AfterTransactionCommitExecutorHolder.getInstance().getAfterCommit().afterCommit(() -> clearArtifactBinary(sha1Hash));
}
@Override
@@ -211,39 +221,35 @@ public class JpaArtifactManagement implements ArtifactManagement {
}
/**
* Garbage collects artifact binaries if only referenced by given
* {@link SoftwareModule#getId()} or {@link SoftwareModule}'s that are
* Garbage collects artifact binaries if only referenced by given {@link SoftwareModule#getId()} or {@link SoftwareModule}'s that are
* marked as deleted.
* <p/>
* Software module related UPDATE permission shall be checked by the callers!
* <p/>
* Note: Internal method. Shall be called ONLY if @PreAuthorize(SpPermission.SpringEvalExpressions.HAS_AUTH_DELETE_REPOSITORY)
* has already been checked
*
* @param sha1Hash no longer needed
*/
@PreAuthorize(SpPermission.SpringEvalExpressions.HAS_AUTH_DELETE_REPOSITORY)
void clearArtifactBinary(final String sha1Hash) {
assertArtifactRepositoryAvailable();
// countBySha1HashAndTenantAndSoftwareModuleDeletedIsFalse will skip ACM checks and
// will return total count as it should be
final long count = localArtifactRepository.countBySha1HashAndTenantAndSoftwareModuleDeletedIsFalse(
sha1Hash,
tenantAware.getCurrentTenant());
if (count <= 1) { // 1 artifact is the one being deleted!
// removes the real artifact ONLY AFTER the delete of artifact or software module
// in local history has passed successfully (caller has permission and no errors)
TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
@Override
public void afterCommit() {
DeploymentHelper.runInNewTransaction(txManager, "clearArtifactBinary", status -> {
// countBySha1HashAndTenantAndSoftwareModuleDeletedIsFalse will skip ACM checks and will return total count as it should be
if (localArtifactRepository.countBySha1HashAndTenantAndSoftwareModuleDeletedIsFalse(sha1Hash, tenantAware.getCurrentTenant()) <= 0) { // 1 artifact is the one being deleted!
// removes the real artifact ONLY AFTER the delete of artifact or software module
// in local history has passed successfully (caller has permission and no errors)
AfterTransactionCommitExecutorHolder.getInstance().getAfterCommit().afterCommit(() -> {
try {
log.debug("deleting artifact from repository {}", sha1Hash);
artifactRepository.deleteBySha1(tenantAware.getCurrentTenant(), sha1Hash);
} catch (final ArtifactStoreException e) {
throw new ArtifactDeleteFailedException(e);
}
}
});
} // else there are still other artifacts that need the binary
});
} // else there are still other artifacts that need the binary
return null;
});
}
private AbstractDbArtifact storeArtifact(final ArtifactUpload artifactUpload, final boolean isSmEncrypted) {

14
hawkbit-repository/hawkbit-repository-jpa/src/main/java/org/eclipse/hawkbit/repository/jpa/management/JpaSoftwareModuleManagement.java
View File

@@ -48,6 +48,7 @@ import org.eclipse.hawkbit.repository.jpa.model.JpaSoftwareModuleMetadata;
import org.eclipse.hawkbit.repository.jpa.model.JpaSoftwareModuleMetadata_;
import org.eclipse.hawkbit.repository.jpa.model.JpaSoftwareModule_;
import org.eclipse.hawkbit.repository.jpa.model.SwMetadataCompositeKey;
import org.eclipse.hawkbit.repository.jpa.model.helper.AfterTransactionCommitExecutorHolder;
import org.eclipse.hawkbit.repository.jpa.repository.DistributionSetRepository;
import org.eclipse.hawkbit.repository.jpa.repository.SoftwareModuleMetadataRepository;
import org.eclipse.hawkbit.repository.jpa.repository.SoftwareModuleRepository;
@@ -199,10 +200,6 @@ public class JpaSoftwareModuleManagement implements SoftwareModuleManagement {
final Set<Long> assignedModuleIds = new HashSet<>();
swModulesToDelete.forEach(swModule -> {
// delete binary data of artifacts
deleteGridFsArtifacts(swModule);
// execute this count operation without access limitations since we have to
// ensure it's not assigned when deleting it.
if (distributionSetRepository.countByModulesId(swModule.getId()) <= 0) {
@@ -210,6 +207,8 @@ public class JpaSoftwareModuleManagement implements SoftwareModuleManagement {
} else {
assignedModuleIds.add(swModule.getId());
}
// schedule delete binary data of artifacts
deleteGridFsArtifacts(swModule);
});
if (!assignedModuleIds.isEmpty()) {
@@ -507,10 +506,9 @@ public class JpaSoftwareModuleManagement implements SoftwareModuleManagement {
private void deleteGridFsArtifacts(final JpaSoftwareModule swModule) {
softwareModuleRepository.getAccessController().ifPresent(accessController ->
accessController.assertOperationAllowed(AccessController.Operation.DELETE, swModule));
for (final Artifact localArtifact : swModule.getArtifacts()) {
((JpaArtifactManagement) artifactManagement)
.clearArtifactBinary(localArtifact.getSha1Hash());
}
final Set<String> sha1Hashes = swModule.getArtifacts().stream().map(Artifact::getSha1Hash).collect(Collectors.toSet());
AfterTransactionCommitExecutorHolder.getInstance().getAfterCommit().afterCommit(() ->
sha1Hashes.forEach(sha1Hash -> ((JpaArtifactManagement) artifactManagement).clearArtifactBinary(sha1Hash)));
}
private Specification<JpaSoftwareModule> buildSmSearchQuerySpec(final String searchText) {

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

@@ -60,6 +60,7 @@ import org.eclipse.hawkbit.repository.jpa.model.JpaTargetTag;
import org.eclipse.hawkbit.repository.jpa.model.JpaTargetType;
import org.eclipse.hawkbit.repository.jpa.model.JpaTarget_;
import org.eclipse.hawkbit.repository.jpa.model.TargetMetadataCompositeKey;
import org.eclipse.hawkbit.repository.jpa.model.helper.AfterTransactionCommitExecutorHolder;
import org.eclipse.hawkbit.repository.jpa.repository.RolloutGroupRepository;
import org.eclipse.hawkbit.repository.jpa.repository.TargetFilterQueryRepository;
import org.eclipse.hawkbit.repository.jpa.repository.TargetMetadataRepository;
@@ -648,15 +649,20 @@ public class JpaTargetManagement implements TargetManagement {
}
@Override
@Transactional
@Retryable(retryFor = { ConcurrencyFailureException.class }, maxAttempts = Constants.TX_RT_MAX,
backoff = @Backoff(delay = Constants.TX_RT_DELAY))
public void requestControllerAttributes(final String controllerId) {
final JpaTarget target = getByControllerIdAndThrowIfNotFound(controllerId);
targetRepository.getAccessController()
.ifPresent(acm -> acm.assertOperationAllowed(AccessController.Operation.UPDATE, target));
target.setRequestControllerAttributes(true);
eventPublisherHolder.getEventPublisher()
.publishEvent(new TargetAttributesRequestedEvent(tenantAware.getCurrentTenant(), target.getId(),
target.getControllerId(), target.getAddress() != null ? target.getAddress().toString() : null,
JpaTarget.class, eventPublisherHolder.getApplicationId()));
AfterTransactionCommitExecutorHolder.getInstance().getAfterCommit().afterCommit(() ->
eventPublisherHolder.getEventPublisher()
.publishEvent(new TargetAttributesRequestedEvent(
tenantAware.getCurrentTenant(), target.getId(), target.getControllerId(),
target.getAddress() != null ? target.getAddress().toString() : null,
JpaTarget.class, eventPublisherHolder.getApplicationId())));
}
@Override
@@ -718,14 +724,12 @@ public class JpaTargetManagement implements TargetManagement {
final JpaTarget updatedTarget = JpaManagementHelper.touch(entityManager, targetRepository, target);
final List<TargetMetadata> createdMetadata = md.stream()
.map(meta -> targetMetadataRepository
.save(new JpaTargetMetadata(meta.getKey(), meta.getValue(), updatedTarget)))
.map(meta -> targetMetadataRepository.save(new JpaTargetMetadata(meta.getKey(), meta.getValue(), updatedTarget)))
.collect(Collectors.toUnmodifiableList());
// TargetUpdatedEvent is not sent within the touch() method due to the
// "lastModifiedAt" field being ignored in JpaTarget
eventPublisherHolder.getEventPublisher()
.publishEvent(new TargetUpdatedEvent(updatedTarget, eventPublisherHolder.getApplicationId()));
eventPublisherHolder.getEventPublisher().publishEvent(new TargetUpdatedEvent(updatedTarget, eventPublisherHolder.getApplicationId()));
return createdMetadata;
}
@@ -738,8 +742,8 @@ public class JpaTargetManagement implements TargetManagement {
final JpaTargetMetadata metadata = (JpaTargetMetadata) getMetaDataByControllerId(controllerId, key)
.orElseThrow(() -> new EntityNotFoundException(TargetMetadata.class, controllerId, key));
final JpaTarget target = JpaManagementHelper.touch(entityManager, targetRepository,
getByControllerIdAndThrowIfNotFound(controllerId));
final JpaTarget target = JpaManagementHelper.touch(
entityManager, targetRepository, getByControllerIdAndThrowIfNotFound(controllerId));
targetRepository.getAccessController()
.ifPresent(acm -> acm.assertOperationAllowed(AccessController.Operation.UPDATE, target));