Introduce Stop of a rollout (#2595)
* Stop of a rollout feature Signed-off-by: strailov <Stanislav.Trailov@bosch.io> * remove some test comments Signed-off-by: strailov <Stanislav.Trailov@bosch.io> * make stop transactional Signed-off-by: strailov <Stanislav.Trailov@bosch.io> * attempt to fix hibernate failed tests Signed-off-by: strailov <Stanislav.Trailov@bosch.io> * fix some sonar issues Signed-off-by: strailov <Stanislav.Trailov@bosch.io> * changes after review Signed-off-by: strailov <Stanislav.Trailov@bosch.io> * fix build Signed-off-by: strailov <Stanislav.Trailov@bosch.io> * fixes after review Signed-off-by: strailov <Stanislav.Trailov@bosch.io> * finish all rollout groups on deletion of rollout Signed-off-by: strailov <Stanislav.Trailov@bosch.io> * refactor finishing groups Signed-off-by: strailov <Stanislav.Trailov@bosch.io> * fix RolloutManagementTest Signed-off-by: strailov <Stanislav.Trailov@bosch.io> --------- Signed-off-by: strailov <Stanislav.Trailov@bosch.io>
This commit is contained in:
Stanislav Trailov
committed by
GitHub
GitHub
parent
4566702030
commit
45cd012532
@@ -54,7 +54,9 @@ import org.eclipse.hawkbit.repository.jpa.utils.QuotaHelper;
|
||||
import org.eclipse.hawkbit.repository.model.Action;
|
||||
import org.eclipse.hawkbit.repository.model.Action.ActionType;
|
||||
import org.eclipse.hawkbit.repository.model.Action.Status;
|
||||
import org.eclipse.hawkbit.repository.model.ActionCancellationType;
|
||||
import org.eclipse.hawkbit.repository.model.DistributionSet;
|
||||
import org.eclipse.hawkbit.repository.model.DistributionSetInvalidation;
|
||||
import org.eclipse.hawkbit.repository.model.Rollout;
|
||||
import org.eclipse.hawkbit.repository.model.Rollout.RolloutStatus;
|
||||
import org.eclipse.hawkbit.repository.model.RolloutGroup;
|
||||
@@ -288,10 +290,19 @@ public class JpaRolloutExecutor implements RolloutExecutor {
|
||||
return;
|
||||
}
|
||||
|
||||
// set soft delete
|
||||
rollout.setStatus(RolloutStatus.DELETED);
|
||||
rollout.setDeleted(true);
|
||||
rolloutRepository.save(rollout);
|
||||
finishRolloutGroups(rollout);
|
||||
|
||||
rolloutManagement.cancelActiveActionsForRollouts(rollout, ActionCancellationType.FORCE);
|
||||
entityManager.flush();
|
||||
|
||||
boolean hasActiveActionsLeft = actionRepository.countByRolloutIdAndActive(rollout.getId(), true) > 0;
|
||||
log.trace("rollout {} has active actions left : {} ", rollout.getId(), hasActiveActionsLeft);
|
||||
if (!hasActiveActionsLeft) {
|
||||
// set soft delete
|
||||
rollout.setStatus(RolloutStatus.DELETED);
|
||||
rollout.setDeleted(true);
|
||||
rolloutRepository.save(rollout);
|
||||
}
|
||||
}
|
||||
|
||||
private void handleStopRollout(final JpaRollout rollout) {
|
||||
@@ -309,18 +320,20 @@ public class JpaRolloutExecutor implements RolloutExecutor {
|
||||
return;
|
||||
}
|
||||
|
||||
rolloutGroupRepository.findByRolloutAndStatusNotIn(rollout, List.of(RolloutGroupStatus.FINISHED, RolloutGroupStatus.ERROR))
|
||||
.forEach(rolloutGroup -> {
|
||||
rolloutGroup.setStatus(RolloutGroupStatus.FINISHED);
|
||||
rolloutGroupRepository.save(rolloutGroup);
|
||||
});
|
||||
finishRolloutGroups(rollout);
|
||||
|
||||
rollout.setStatus(RolloutStatus.FINISHED);
|
||||
rolloutRepository.save(rollout);
|
||||
// Soft cancel all active rollouts actions
|
||||
rolloutManagement.cancelActiveActionsForRollouts(rollout, ActionCancellationType.SOFT);
|
||||
// check if all actions are non-active and then finish or finish once all are processed.
|
||||
boolean hasActiveActions = actionRepository.countByRolloutIdAndActiveAndStatusNot(rollout.getId(), true, Status.CANCELING) > 0;
|
||||
if (!hasActiveActions) {
|
||||
rollout.setStatus(RolloutStatus.STOPPED);
|
||||
rolloutRepository.save(rollout);
|
||||
|
||||
final List<Long> groupIds = rollout.getRolloutGroups().stream().map(RolloutGroup::getId).toList();
|
||||
afterCommit.afterCommit(() -> EventPublisherHolder.getInstance().getEventPublisher().publishEvent(new RolloutStoppedEvent(
|
||||
tenantAware.getCurrentTenant(), rollout.getId(), groupIds)));
|
||||
final List<Long> groupIds = rollout.getRolloutGroups().stream().map(RolloutGroup::getId).toList();
|
||||
afterCommit.afterCommit(() -> EventPublisherHolder.getInstance().getEventPublisher().publishEvent(new RolloutStoppedEvent(
|
||||
tenantAware.getCurrentTenant(), rollout.getId(), groupIds)));
|
||||
}
|
||||
}
|
||||
|
||||
private void handleReadyRollout(final Rollout rollout) {
|
||||
@@ -392,6 +405,14 @@ public class JpaRolloutExecutor implements RolloutExecutor {
|
||||
}
|
||||
}
|
||||
|
||||
private void finishRolloutGroups(final JpaRollout rollout) {
|
||||
rolloutGroupRepository.findByRolloutAndStatusNotIn(rollout, List.of(RolloutGroupStatus.FINISHED, RolloutGroupStatus.ERROR))
|
||||
.forEach(rolloutGroup -> {
|
||||
rolloutGroup.setStatus(RolloutGroupStatus.FINISHED);
|
||||
rolloutGroupRepository.save(rolloutGroup);
|
||||
});
|
||||
}
|
||||
|
||||
private Slice<JpaAction> findScheduledActionsByRollout(final JpaRollout rollout) {
|
||||
return actionRepository.findByRolloutIdAndStatus(PageRequest.of(0, TRANSACTION_ACTIONS), rollout.getId(),
|
||||
Status.SCHEDULED);
|
||||
|
||||
@@ -74,11 +74,11 @@ import org.eclipse.hawkbit.repository.jpa.utils.WeightValidationHelper;
|
||||
import org.eclipse.hawkbit.repository.model.Action;
|
||||
import org.eclipse.hawkbit.repository.model.Action.ActionType;
|
||||
import org.eclipse.hawkbit.repository.model.Action.Status;
|
||||
import org.eclipse.hawkbit.repository.model.ActionCancellationType;
|
||||
import org.eclipse.hawkbit.repository.model.ActionStatus;
|
||||
import org.eclipse.hawkbit.repository.model.DeploymentRequest;
|
||||
import org.eclipse.hawkbit.repository.model.DistributionSet;
|
||||
import org.eclipse.hawkbit.repository.model.DistributionSetAssignmentResult;
|
||||
import org.eclipse.hawkbit.repository.model.DistributionSetInvalidation.CancelationType;
|
||||
import org.eclipse.hawkbit.repository.model.DistributionSetType;
|
||||
import org.eclipse.hawkbit.repository.model.SoftwareModuleType;
|
||||
import org.eclipse.hawkbit.repository.model.Target;
|
||||
@@ -239,7 +239,7 @@ public class JpaDeploymentManagement extends JpaActionManagement implements Depl
|
||||
RepositoryConstants.SERVER_MESSAGE_PREFIX + "manual cancelation requested"));
|
||||
final Action saveAction = actionRepository.save(action);
|
||||
|
||||
onlineDsAssignmentStrategy.cancelAssignment(action);
|
||||
onlineDsAssignmentStrategy.sendCancellationMessage(action);
|
||||
|
||||
return saveAction;
|
||||
} else {
|
||||
@@ -516,7 +516,7 @@ public class JpaDeploymentManagement extends JpaActionManagement implements Depl
|
||||
|
||||
@Override
|
||||
@Transactional
|
||||
public void cancelActionsForDistributionSet(final CancelationType cancelationType, final DistributionSet distributionSet) {
|
||||
public void cancelActionsForDistributionSet(final ActionCancellationType cancelationType, final DistributionSet distributionSet) {
|
||||
actionRepository.findAll(ActionSpecifications.byDistributionSetIdAndActiveAndStatusIsNot(distributionSet.getId(), Status.CANCELING))
|
||||
.forEach(action -> {
|
||||
try {
|
||||
@@ -529,7 +529,7 @@ public class JpaDeploymentManagement extends JpaActionManagement implements Depl
|
||||
log.trace("Could not cancel action {} due to entity not found exception.", action.getId(), e);
|
||||
}
|
||||
});
|
||||
if (cancelationType == CancelationType.FORCE) {
|
||||
if (cancelationType == ActionCancellationType.FORCE) {
|
||||
actionRepository.findAll(ActionSpecifications.byDistributionSetIdAndActive(distributionSet.getId())).forEach(action -> {
|
||||
try {
|
||||
assertTargetUpdateAllowed(action);
|
||||
@@ -544,10 +544,6 @@ public class JpaDeploymentManagement extends JpaActionManagement implements Depl
|
||||
}
|
||||
}
|
||||
|
||||
protected ActionRepository getActionRepository() {
|
||||
return actionRepository;
|
||||
}
|
||||
|
||||
protected boolean isActionsAutocloseEnabled() {
|
||||
return getConfigValue(REPOSITORY_ACTIONS_AUTOCLOSE_ENABLED, Boolean.class);
|
||||
}
|
||||
@@ -1001,4 +997,9 @@ public class JpaDeploymentManagement extends JpaActionManagement implements Depl
|
||||
throw new EntityNotFoundException(Action.class, actionId);
|
||||
}
|
||||
}
|
||||
|
||||
private Page<JpaAction> findActiveActionsForRollout(long rolloutId, Pageable pageable) {
|
||||
return actionRepository
|
||||
.findAll(ActionSpecifications.byRolloutIdAndActive(rolloutId), pageable);
|
||||
}
|
||||
}
|
||||
@@ -25,9 +25,9 @@ import org.eclipse.hawkbit.repository.exception.StopRolloutException;
|
||||
import org.eclipse.hawkbit.repository.jpa.repository.ActionRepository;
|
||||
import org.eclipse.hawkbit.repository.jpa.utils.DeploymentHelper;
|
||||
import org.eclipse.hawkbit.repository.model.Action.Status;
|
||||
import org.eclipse.hawkbit.repository.model.ActionCancellationType;
|
||||
import org.eclipse.hawkbit.repository.model.DistributionSet;
|
||||
import org.eclipse.hawkbit.repository.model.DistributionSetInvalidation;
|
||||
import org.eclipse.hawkbit.repository.model.DistributionSetInvalidation.CancelationType;
|
||||
import org.eclipse.hawkbit.repository.model.DistributionSetInvalidationCount;
|
||||
import org.eclipse.hawkbit.repository.model.TargetFilterQuery;
|
||||
import org.eclipse.hawkbit.security.SystemSecurityContext;
|
||||
@@ -80,7 +80,7 @@ public class JpaDistributionSetInvalidationManagement implements DistributionSet
|
||||
public void invalidateDistributionSet(final DistributionSetInvalidation distributionSetInvalidation) {
|
||||
log.debug("Invalidate distribution sets {}", distributionSetInvalidation.getDistributionSetIds());
|
||||
final String tenant = tenantAware.getCurrentTenant();
|
||||
if (shouldRolloutsBeCanceled(distributionSetInvalidation.getCancelationType(), distributionSetInvalidation.isCancelRollouts())) {
|
||||
if (shouldRolloutsBeCanceled(distributionSetInvalidation.getActionCancellationType())) {
|
||||
final String handlerId = JpaRolloutManagement.createRolloutLockKey(tenant);
|
||||
final Lock lock = lockRegistry.obtain(handlerId);
|
||||
try {
|
||||
@@ -108,30 +108,30 @@ public class JpaDistributionSetInvalidationManagement implements DistributionSet
|
||||
final DistributionSetInvalidation distributionSetInvalidation) {
|
||||
return systemSecurityContext.runAsSystem(() -> {
|
||||
final Collection<Long> setIds = distributionSetInvalidation.getDistributionSetIds();
|
||||
final long rolloutsCount = shouldRolloutsBeCanceled(distributionSetInvalidation.getCancelationType(),
|
||||
distributionSetInvalidation.isCancelRollouts()) ? countRolloutsForInvalidation(setIds) : 0;
|
||||
final long rolloutsCount = shouldRolloutsBeCanceled(distributionSetInvalidation.getActionCancellationType()) ? countRolloutsForInvalidation(setIds) : 0;
|
||||
final long autoAssignmentsCount = countAutoAssignmentsForInvalidation(setIds);
|
||||
final long actionsCount = countActionsForInvalidation(setIds,
|
||||
distributionSetInvalidation.getCancelationType());
|
||||
distributionSetInvalidation.getActionCancellationType());
|
||||
|
||||
return new DistributionSetInvalidationCount(rolloutsCount, autoAssignmentsCount, actionsCount);
|
||||
});
|
||||
}
|
||||
|
||||
private static boolean shouldRolloutsBeCanceled(final CancelationType cancelationType,
|
||||
final boolean cancelRollouts) {
|
||||
return cancelationType != CancelationType.NONE || cancelRollouts;
|
||||
private static boolean shouldRolloutsBeCanceled(final ActionCancellationType cancelationType) {
|
||||
return cancelationType != ActionCancellationType.NONE;
|
||||
}
|
||||
|
||||
private void invalidateDistributionSetsInTransaction(final DistributionSetInvalidation distributionSetInvalidation, final String tenant) {
|
||||
private void invalidateDistributionSetsInTransaction(final DistributionSetInvalidation distributionSetInvalidation,
|
||||
final String tenant) {
|
||||
DeploymentHelper.runInNewTransaction(txManager, tenant + "-invalidateDS", status -> {
|
||||
distributionSetInvalidation.getDistributionSetIds().forEach(setId -> invalidateDistributionSet(
|
||||
setId, distributionSetInvalidation.getCancelationType(), distributionSetInvalidation.isCancelRollouts()));
|
||||
distributionSetInvalidation.getDistributionSetIds().forEach(setId -> invalidateDistributionSet(setId,
|
||||
distributionSetInvalidation.getActionCancellationType()));
|
||||
|
||||
return 0;
|
||||
});
|
||||
}
|
||||
|
||||
private void invalidateDistributionSet(final long setId, final CancelationType cancelationType, final boolean cancelRollouts) {
|
||||
private void invalidateDistributionSet(final long setId, final ActionCancellationType cancelationType) {
|
||||
final DistributionSet distributionSet = distributionSetManagement.getOrElseThrowException(setId);
|
||||
if (!distributionSet.isComplete()) {
|
||||
throw new IncompleteDistributionSetException(
|
||||
@@ -141,18 +141,18 @@ public class JpaDistributionSetInvalidationManagement implements DistributionSet
|
||||
log.debug("Distribution set {} marked as invalid.", setId);
|
||||
|
||||
// rollout cancellation should only be permitted with UPDATE_ROLLOUT permission
|
||||
if (shouldRolloutsBeCanceled(cancelationType, cancelRollouts)) {
|
||||
if (shouldRolloutsBeCanceled(cancelationType)) {
|
||||
log.debug("Cancel rollouts after ds invalidation. ID: {}", setId);
|
||||
rolloutManagement.cancelRolloutsForDistributionSet(distributionSet);
|
||||
rolloutManagement.cancelRolloutsForDistributionSet(distributionSet, cancelationType);
|
||||
}
|
||||
|
||||
if (cancelationType != ActionCancellationType.NONE) {
|
||||
log.debug("Cancel actions after ds invalidation. ID: {}", setId);
|
||||
deploymentManagement.cancelActionsForDistributionSet(cancelationType, distributionSet);
|
||||
}
|
||||
|
||||
// Do run as system to ensure all actions (even invisible) are canceled due to invalidation.
|
||||
systemSecurityContext.runAsSystem(() -> {
|
||||
if (cancelationType != CancelationType.NONE) {
|
||||
log.debug("Cancel actions after ds invalidation. ID: {}", setId);
|
||||
deploymentManagement.cancelActionsForDistributionSet(cancelationType, distributionSet);
|
||||
}
|
||||
|
||||
log.debug("Cancel auto assignments after ds invalidation. ID: {}", setId);
|
||||
targetFilterQueryManagement.cancelAutoAssignmentForDistributionSet(setId);
|
||||
return null;
|
||||
@@ -167,11 +167,11 @@ public class JpaDistributionSetInvalidationManagement implements DistributionSet
|
||||
return setIds.stream().mapToLong(targetFilterQueryManagement::countByAutoAssignDistributionSetId).sum();
|
||||
}
|
||||
|
||||
private long countActionsForInvalidation(final Collection<Long> setIds, final CancelationType cancelationType) {
|
||||
private long countActionsForInvalidation(final Collection<Long> setIds, final ActionCancellationType cancelationType) {
|
||||
long affectedActionsByDSInvalidation = 0;
|
||||
if (cancelationType == CancelationType.FORCE) {
|
||||
if (cancelationType == ActionCancellationType.FORCE) {
|
||||
affectedActionsByDSInvalidation = countActionsForForcedInvalidation(setIds);
|
||||
} else if (cancelationType == CancelationType.SOFT) {
|
||||
} else if (cancelationType == ActionCancellationType.SOFT) {
|
||||
affectedActionsByDSInvalidation = countActionsForSoftInvalidation(setIds);
|
||||
}
|
||||
return affectedActionsByDSInvalidation;
|
||||
|
||||
@@ -19,6 +19,8 @@ import java.util.function.Function;
|
||||
import java.util.function.UnaryOperator;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
import jakarta.persistence.EntityManager;
|
||||
import jakarta.persistence.Query;
|
||||
import jakarta.validation.ConstraintDeclarationException;
|
||||
import jakarta.validation.Valid;
|
||||
import jakarta.validation.ValidationException;
|
||||
@@ -30,6 +32,7 @@ import org.eclipse.hawkbit.im.authentication.SpPermission;
|
||||
import org.eclipse.hawkbit.im.authentication.SpRole;
|
||||
import org.eclipse.hawkbit.repository.DistributionSetManagement;
|
||||
import org.eclipse.hawkbit.repository.QuotaManagement;
|
||||
import org.eclipse.hawkbit.repository.RepositoryConstants;
|
||||
import org.eclipse.hawkbit.repository.RepositoryProperties;
|
||||
import org.eclipse.hawkbit.repository.RolloutApprovalStrategy;
|
||||
import org.eclipse.hawkbit.repository.RolloutFields;
|
||||
@@ -46,22 +49,30 @@ import org.eclipse.hawkbit.repository.exception.IncompleteDistributionSetExcepti
|
||||
import org.eclipse.hawkbit.repository.exception.InsufficientPermissionException;
|
||||
import org.eclipse.hawkbit.repository.exception.InvalidDistributionSetException;
|
||||
import org.eclipse.hawkbit.repository.exception.RolloutIllegalStateException;
|
||||
import org.eclipse.hawkbit.repository.jpa.Jpa;
|
||||
import org.eclipse.hawkbit.repository.jpa.JpaManagementHelper;
|
||||
import org.eclipse.hawkbit.repository.jpa.configuration.Constants;
|
||||
import org.eclipse.hawkbit.repository.jpa.executor.AfterTransactionCommitExecutor;
|
||||
import org.eclipse.hawkbit.repository.jpa.model.AbstractJpaBaseEntity_;
|
||||
import org.eclipse.hawkbit.repository.jpa.model.JpaAction;
|
||||
import org.eclipse.hawkbit.repository.jpa.model.JpaActionStatus;
|
||||
import org.eclipse.hawkbit.repository.jpa.model.JpaDistributionSet;
|
||||
import org.eclipse.hawkbit.repository.jpa.model.JpaRollout;
|
||||
import org.eclipse.hawkbit.repository.jpa.model.JpaRolloutGroup;
|
||||
import org.eclipse.hawkbit.repository.jpa.model.JpaRollout_;
|
||||
import org.eclipse.hawkbit.repository.jpa.repository.ActionRepository;
|
||||
import org.eclipse.hawkbit.repository.jpa.repository.ActionStatusRepository;
|
||||
import org.eclipse.hawkbit.repository.jpa.repository.RolloutGroupRepository;
|
||||
import org.eclipse.hawkbit.repository.jpa.repository.RolloutRepository;
|
||||
import org.eclipse.hawkbit.repository.jpa.repository.TargetRepository;
|
||||
import org.eclipse.hawkbit.repository.jpa.rollout.condition.StartNextGroupRolloutGroupSuccessAction;
|
||||
import org.eclipse.hawkbit.repository.jpa.rsql.RsqlUtility;
|
||||
import org.eclipse.hawkbit.repository.jpa.specifications.ActionSpecifications;
|
||||
import org.eclipse.hawkbit.repository.jpa.specifications.RolloutSpecification;
|
||||
import org.eclipse.hawkbit.repository.jpa.utils.QuotaHelper;
|
||||
import org.eclipse.hawkbit.repository.jpa.utils.WeightValidationHelper;
|
||||
import org.eclipse.hawkbit.repository.model.Action;
|
||||
import org.eclipse.hawkbit.repository.model.ActionCancellationType;
|
||||
import org.eclipse.hawkbit.repository.model.DistributionSet;
|
||||
import org.eclipse.hawkbit.repository.model.DistributionSetType;
|
||||
import org.eclipse.hawkbit.repository.model.Rollout;
|
||||
@@ -75,6 +86,9 @@ import org.eclipse.hawkbit.repository.model.TotalTargetCountActionStatus;
|
||||
import org.eclipse.hawkbit.repository.model.TotalTargetCountStatus;
|
||||
import org.eclipse.hawkbit.security.SystemSecurityContext;
|
||||
import org.eclipse.hawkbit.utils.ObjectCopyUtil;
|
||||
import org.eclipse.hawkbit.tenancy.TenantAware;
|
||||
import org.eclipse.hawkbit.utils.TenantConfigHelper;
|
||||
import org.springframework.beans.factory.annotation.Value;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnBooleanProperty;
|
||||
import org.springframework.dao.ConcurrencyFailureException;
|
||||
import org.springframework.data.domain.Page;
|
||||
@@ -106,6 +120,10 @@ public class JpaRolloutManagement implements RolloutManagement {
|
||||
RolloutStatus.CREATING, RolloutStatus.READY, RolloutStatus.WAITING_FOR_APPROVAL, RolloutStatus.STARTING, RolloutStatus.RUNNING,
|
||||
RolloutStatus.PAUSED, RolloutStatus.APPROVAL_DENIED);
|
||||
|
||||
@Value("${org.eclipse.hawkbit.repository.jpa.management.rollout.max.actions.per.transaction:5000}")
|
||||
private int MAX_ACTIONS;
|
||||
|
||||
private final EntityManager entityManager;
|
||||
private final RolloutRepository rolloutRepository;
|
||||
private final RolloutGroupRepository rolloutGroupRepository;
|
||||
private final RolloutApprovalStrategy rolloutApprovalStrategy;
|
||||
@@ -113,42 +131,55 @@ public class JpaRolloutManagement implements RolloutManagement {
|
||||
private final RolloutStatusCache rolloutStatusCache;
|
||||
private final ActionRepository actionRepository;
|
||||
private final TargetManagement<? extends Target> targetManagement;
|
||||
private final ActionStatusRepository actionStatusRepository;
|
||||
private final DistributionSetManagement<? extends DistributionSet> distributionSetManagement;
|
||||
private final TenantAware tenantAware;
|
||||
private final TenantConfigurationManagement tenantConfigurationManagement;
|
||||
private final QuotaManagement quotaManagement;
|
||||
private final AfterTransactionCommitExecutor afterCommit;
|
||||
private final SystemSecurityContext systemSecurityContext;
|
||||
private final ContextAware contextAware;
|
||||
private final RepositoryProperties repositoryProperties;
|
||||
private final OnlineDsAssignmentStrategy onlineDsAssignmentStrategy;
|
||||
|
||||
protected JpaRolloutManagement(
|
||||
final EntityManager entityManager,
|
||||
final RolloutRepository rolloutRepository,
|
||||
final RolloutGroupRepository rolloutGroupRepository,
|
||||
final RolloutApprovalStrategy rolloutApprovalStrategy,
|
||||
final StartNextGroupRolloutGroupSuccessAction startNextRolloutGroupAction,
|
||||
final RolloutStatusCache rolloutStatusCache,
|
||||
final ActionRepository actionRepository,
|
||||
final ActionStatusRepository actionStatusRepository,
|
||||
final TargetRepository targetRepository,
|
||||
final TargetManagement<? extends Target> targetManagement,
|
||||
final DistributionSetManagement<? extends DistributionSet> distributionSetManagement,
|
||||
final TenantAware tenantAware,
|
||||
final TenantConfigurationManagement tenantConfigurationManagement,
|
||||
final QuotaManagement quotaManagement,
|
||||
final AfterTransactionCommitExecutor afterCommit,
|
||||
final SystemSecurityContext systemSecurityContext, final ContextAware contextAware,
|
||||
final RepositoryProperties repositoryProperties) {
|
||||
this.entityManager = entityManager;
|
||||
this.rolloutRepository = rolloutRepository;
|
||||
this.rolloutGroupRepository = rolloutGroupRepository;
|
||||
this.rolloutApprovalStrategy = rolloutApprovalStrategy;
|
||||
this.startNextRolloutGroupAction = startNextRolloutGroupAction;
|
||||
this.rolloutStatusCache = rolloutStatusCache;
|
||||
this.actionRepository = actionRepository;
|
||||
this.actionStatusRepository = actionStatusRepository;
|
||||
this.targetManagement = targetManagement;
|
||||
this.distributionSetManagement = distributionSetManagement;
|
||||
this.tenantAware = tenantAware;
|
||||
this.tenantConfigurationManagement = tenantConfigurationManagement;
|
||||
this.quotaManagement = quotaManagement;
|
||||
this.afterCommit = afterCommit;
|
||||
this.systemSecurityContext = systemSecurityContext;
|
||||
this.contextAware = contextAware;
|
||||
this.repositoryProperties = repositoryProperties;
|
||||
|
||||
this.onlineDsAssignmentStrategy = new OnlineDsAssignmentStrategy(targetRepository, afterCommit, actionRepository, actionStatusRepository,
|
||||
quotaManagement, this::isMultiAssignmentsEnabled, this::isConfirmationFlowEnabled, repositoryProperties);
|
||||
}
|
||||
|
||||
public static String createRolloutLockKey(final String tenant) {
|
||||
@@ -409,27 +440,49 @@ public class JpaRolloutManagement implements RolloutManagement {
|
||||
@Transactional
|
||||
@Retryable(retryFor = { ConcurrencyFailureException.class }, maxAttempts = Constants.TX_RT_MAX,
|
||||
backoff = @Backoff(delay = Constants.TX_RT_DELAY))
|
||||
public void delete(final long rolloutId) {
|
||||
public Rollout stop(long rolloutId) {
|
||||
final JpaRollout jpaRollout = rolloutRepository.findById(rolloutId)
|
||||
.orElseThrow(() -> new EntityNotFoundException(Rollout.class, rolloutId));
|
||||
|
||||
if (RolloutStatus.DELETING == jpaRollout.getStatus()) {
|
||||
return;
|
||||
if (!ROLLOUT_STATUS_STOPPABLE.contains(jpaRollout.getStatus())) {
|
||||
log.debug("Failed to stop rollout {} because it is in {} status.", rolloutId, jpaRollout.getStatus());
|
||||
throw new RolloutIllegalStateException("Rollout can only be stopped into the following statuses " + ROLLOUT_STATUS_STOPPABLE);
|
||||
}
|
||||
jpaRollout.setStatus(RolloutStatus.DELETING);
|
||||
rolloutRepository.save(jpaRollout);
|
||||
|
||||
log.debug("Stopping Rollout {}", jpaRollout.getId());
|
||||
jpaRollout.setStatus(RolloutStatus.STOPPING);
|
||||
return rolloutRepository.save(jpaRollout);
|
||||
}
|
||||
|
||||
@Override
|
||||
@Transactional
|
||||
public void cancelRolloutsForDistributionSet(final DistributionSet set) {
|
||||
@Retryable(retryFor = { ConcurrencyFailureException.class }, maxAttempts = Constants.TX_RT_MAX,
|
||||
backoff = @Backoff(delay = Constants.TX_RT_DELAY))
|
||||
public void delete(final long rolloutId) {
|
||||
final JpaRollout jpaRollout = rolloutRepository.findById(rolloutId)
|
||||
.orElseThrow(() -> new EntityNotFoundException(Rollout.class, rolloutId));
|
||||
this.delete0(jpaRollout);
|
||||
}
|
||||
|
||||
@Override
|
||||
@Transactional
|
||||
public void cancelRolloutsForDistributionSet(final DistributionSet set, final ActionCancellationType cancelationType) {
|
||||
// stop all rollouts for this distribution set
|
||||
rolloutRepository.findByDistributionSetAndStatusIn(set, ROLLOUT_STATUS_STOPPABLE).forEach(rollout -> {
|
||||
final JpaRollout jpaRollout = (JpaRollout) rollout;
|
||||
jpaRollout.setStatus(RolloutStatus.STOPPING);
|
||||
rolloutRepository.save(jpaRollout);
|
||||
log.debug("Rollout {} stopped", jpaRollout.getId());
|
||||
});
|
||||
if (cancelationType.equals(ActionCancellationType.SOFT)) {
|
||||
rolloutRepository.findByDistributionSetAndStatusIn(set, ROLLOUT_STATUS_STOPPABLE).forEach(rollout -> {
|
||||
final JpaRollout jpaRollout = (JpaRollout) rollout;
|
||||
jpaRollout.setStatus(RolloutStatus.STOPPING);
|
||||
rolloutRepository.save(jpaRollout);
|
||||
log.debug("Rollout {} stopping", jpaRollout.getId());
|
||||
});
|
||||
} else if (cancelationType.equals(ActionCancellationType.FORCE)) {
|
||||
// Use same status filter here like in the soft case ? Seems they make sense
|
||||
rolloutRepository.findByDistributionSetAndStatusIn(set, ROLLOUT_STATUS_STOPPABLE).forEach(rollout -> {
|
||||
final JpaRollout jpaRollout = (JpaRollout) rollout;
|
||||
this.delete0(jpaRollout);
|
||||
log.debug("Rollout {} deleting", jpaRollout.getId());
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
@@ -459,6 +512,124 @@ public class JpaRolloutManagement implements RolloutManagement {
|
||||
startNextRolloutGroupAction.exec(rollout, latestRunning);
|
||||
}
|
||||
|
||||
@Override
|
||||
@Transactional
|
||||
public void cancelActiveActionsForRollouts(Rollout rollout, ActionCancellationType cancelationType) {
|
||||
// check cancellation type
|
||||
if (ActionCancellationType.FORCE.equals(cancelationType)) {
|
||||
forceQuitActionsOfRollout(rollout);
|
||||
} else if (ActionCancellationType.SOFT.equals(cancelationType)) {
|
||||
softCancelActionsOfRollout(rollout);
|
||||
}
|
||||
}
|
||||
|
||||
private void softCancelActionsOfRollout(final Rollout rollout) {
|
||||
final List<JpaAction> actions = actionRepository.findAll(
|
||||
ActionSpecifications
|
||||
.byRolloutIdAndActiveAndStatusIsNot(rollout.getId(),
|
||||
List.of(Action.Status.CANCELING)), // avoid cancelling state here, because it is count as still active
|
||||
Pageable.ofSize(MAX_ACTIONS))
|
||||
.getContent();
|
||||
log.info("Found {} active actions for rollout {}, performing soft cancel.", actions.size(), rollout.getId());
|
||||
|
||||
storeActionsAndStatuses(actions, Action.Status.CANCELING);
|
||||
|
||||
// send cancellation messages to event publisher
|
||||
onlineDsAssignmentStrategy.sendCancellationMessages(actions, tenantAware.getCurrentTenant());
|
||||
}
|
||||
|
||||
private void forceQuitActionsOfRollout(final Rollout rollout) {
|
||||
final List<JpaAction> actions = findActiveActionsForRollout(rollout.getId(), Pageable.ofSize(MAX_ACTIONS))
|
||||
.getContent();
|
||||
log.info("Found {} active actions for rollout {}", actions.size(), rollout.getId());
|
||||
|
||||
storeActionsAndStatuses(actions, Action.Status.CANCELED);
|
||||
|
||||
// find next active actions - filter by targetId list and isActive
|
||||
final List<Long> targetIds = actions.stream()
|
||||
.map(action -> action.getTarget().getId())
|
||||
.toList();
|
||||
entityManager.flush();
|
||||
|
||||
int modifiedRows = updateTargetAssignedDsWithFirstActiveAction(targetIds);
|
||||
log.debug("Updated {} targets with their previously active action", modifiedRows);
|
||||
|
||||
// if no active actions
|
||||
// set assignedDs to previously installedDs and status to IN_SYNC
|
||||
// otherwise set assigned ds to the active action ...
|
||||
modifiedRows = updateTargetAssignedDsWithInstalledIfNoActiveActions(targetIds);
|
||||
log.debug("Updated assignDs to previously installed to {} number of targets.", modifiedRows);
|
||||
}
|
||||
|
||||
private void storeActionsAndStatuses(List<JpaAction> actions, Action.Status status) {
|
||||
final List<JpaActionStatus> cancellingStatuses = new ArrayList<>(actions.size());
|
||||
final long currentTimestamp = System.currentTimeMillis();
|
||||
final boolean active = Action.Status.CANCELING.equals(status);
|
||||
final String typeOfCancellation = active ? "cancellation" : "force quit";
|
||||
|
||||
actions.forEach(action -> {
|
||||
action.setStatus(status);
|
||||
action.setActive(active);
|
||||
|
||||
JpaActionStatus actionStatus = new JpaActionStatus();
|
||||
actionStatus.setAction(action);
|
||||
actionStatus.setStatus(status);
|
||||
actionStatus.setOccurredAt(currentTimestamp);
|
||||
actionStatus.addMessage(RepositoryConstants.SERVER_MESSAGE_PREFIX + "A " + typeOfCancellation + " has been performed by server.");
|
||||
cancellingStatuses.add(actionStatus);
|
||||
});
|
||||
|
||||
actionStatusRepository.saveAll(cancellingStatuses);
|
||||
actionRepository.saveAll(actions);
|
||||
}
|
||||
|
||||
private int updateTargetAssignedDsWithFirstActiveAction(List<Long> targetIds) {
|
||||
final Query updateQuery = entityManager.createNativeQuery(
|
||||
"UPDATE sp_target t " +
|
||||
"SET t.assigned_distribution_set = ( " +
|
||||
"SELECT a.distribution_set" +
|
||||
" FROM sp_action a" +
|
||||
" WHERE a.target = t.id AND a.active = 1" +
|
||||
" ORDER BY a.id ASC" +
|
||||
" LIMIT 1" +
|
||||
") " +
|
||||
"WHERE t.id IN (" + Jpa.formatNativeQueryInClause("tid", targetIds) + ")"
|
||||
);
|
||||
Jpa.setNativeQueryInParameter(updateQuery, "tid", targetIds);
|
||||
final int updated = updateQuery.executeUpdate();
|
||||
log.info("{} of target assigned distribution values updated for tenant {}",
|
||||
updated, tenantAware.getCurrentTenant());
|
||||
return updated;
|
||||
}
|
||||
|
||||
private int updateTargetAssignedDsWithInstalledIfNoActiveActions(List<Long> targetIds) {
|
||||
final Query updateQuery = entityManager.createNativeQuery(
|
||||
"UPDATE sp_target t " +
|
||||
"SET t.assigned_distribution_set = t.installed_distribution_set, t.update_status = 1 " +
|
||||
"WHERE t.id IN (" + Jpa.formatNativeQueryInClause("tid", targetIds) + ") " +
|
||||
" AND (SELECT count(*) FROM sp_action a " +
|
||||
" WHERE a.target=t.id and a.active=1) = 0"
|
||||
);
|
||||
Jpa.setNativeQueryInParameter(updateQuery, "tid", targetIds);
|
||||
final int updated = updateQuery.executeUpdate();
|
||||
log.info("{} of target assigned distribution set to previously installed distribution value for tenant {}",
|
||||
updated, tenantAware.getCurrentTenant());
|
||||
return updated;
|
||||
}
|
||||
|
||||
private Page<JpaAction> findActiveActionsForRollout(long rolloutId, Pageable pageable) {
|
||||
return actionRepository
|
||||
.findAll(ActionSpecifications.byRolloutIdAndActive(rolloutId), pageable);
|
||||
}
|
||||
|
||||
private void delete0(final JpaRollout jpaRollout) {
|
||||
if (RolloutStatus.DELETING == jpaRollout.getStatus()) {
|
||||
return;
|
||||
}
|
||||
jpaRollout.setStatus(RolloutStatus.DELETING);
|
||||
rolloutRepository.save(jpaRollout);
|
||||
}
|
||||
|
||||
private Page<Rollout> appendStatusDetails(final Page<Rollout> rollouts) {
|
||||
final List<Long> rolloutIds = rollouts.getContent().stream().map(Rollout::getId).toList();
|
||||
final Map<Long, List<TotalTargetCountActionStatus>> allStatesForRollout = getStatusCountItemForRollout(rolloutIds);
|
||||
@@ -851,5 +1022,13 @@ public class JpaRolloutManagement implements RolloutManagement {
|
||||
return new TargetCount(totalTargets, baseFilter);
|
||||
}
|
||||
|
||||
private boolean isMultiAssignmentsEnabled() {
|
||||
return TenantConfigHelper.usingContext(systemSecurityContext, tenantConfigurationManagement).isMultiAssignmentsEnabled();
|
||||
}
|
||||
|
||||
private boolean isConfirmationFlowEnabled() {
|
||||
return TenantConfigHelper.usingContext(systemSecurityContext, tenantConfigurationManagement).isConfirmationFlowEnabled();
|
||||
}
|
||||
|
||||
private record TargetCount(long total, String filter) {}
|
||||
}
|
||||
@@ -176,7 +176,7 @@ class OnlineDsAssignmentStrategy extends AbstractDsAssignmentStrategy {
|
||||
}
|
||||
}
|
||||
|
||||
void cancelAssignment(final JpaAction action) {
|
||||
void sendCancellationMessage(final JpaAction action) {
|
||||
if (isMultiAssignmentsEnabled()) {
|
||||
sendMultiActionCancelEvent(action);
|
||||
} else {
|
||||
@@ -184,6 +184,14 @@ class OnlineDsAssignmentStrategy extends AbstractDsAssignmentStrategy {
|
||||
}
|
||||
}
|
||||
|
||||
void sendCancellationMessages(final List<JpaAction> actions, final String tenant) {
|
||||
if(isMultiAssignmentsEnabled()) {
|
||||
sendMultiActionCancelEvent(tenant, Collections.unmodifiableList(actions));
|
||||
} else {
|
||||
actions.forEach(this::cancelAssignDistributionSetEvent);
|
||||
}
|
||||
}
|
||||
|
||||
private static Stream<Action> filterCancellations(final List<Action> actions) {
|
||||
return actions.stream().filter(action -> {
|
||||
final Status actionStatus = action.getStatus();
|
||||
|
||||
@@ -183,6 +183,25 @@ public interface ActionRepository extends BaseEntityRepository<JpaAction> {
|
||||
*/
|
||||
Long countByRolloutIdAndStatus(Long rolloutId, Action.Status status);
|
||||
|
||||
/**
|
||||
* Returns the number of active/non-active actions for a rollout
|
||||
*
|
||||
* @param rolloutId - the ID of the rollout the actions belong to
|
||||
* @param active - wether the actions should be active or not
|
||||
* @return number of actions which match the criteria
|
||||
*/
|
||||
Long countByRolloutIdAndActive(Long rolloutId, boolean active);
|
||||
|
||||
/**
|
||||
* Returns the number of active/non-active actions for a rollout which status is not the provided one.
|
||||
*
|
||||
* @param rolloutId - the ID of the rollout the actions belong to
|
||||
* @param active - wether the actions should be active or not
|
||||
* @param status - the status that matched actions should not be.
|
||||
* @return number of actions which match the criteria
|
||||
*/
|
||||
Long countByRolloutIdAndActiveAndStatusNot(Long rolloutId, boolean active, Action.Status status);
|
||||
|
||||
/**
|
||||
* Returns {@code true} if actions for the given rollout exists, otherwise {@code false}
|
||||
* <p/>
|
||||
|
||||
@@ -28,6 +28,8 @@ import org.eclipse.hawkbit.repository.jpa.model.JpaTarget_;
|
||||
import org.eclipse.hawkbit.repository.model.Action;
|
||||
import org.springframework.data.jpa.domain.Specification;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* Utility class for {@link Action}s {@link Specification}s. The class provides Spring Data JPQL Specifications.
|
||||
*/
|
||||
@@ -84,6 +86,21 @@ public final class ActionSpecifications {
|
||||
cb.equal(root.get(JpaAction_.active), true));
|
||||
}
|
||||
|
||||
public static Specification<JpaAction> byRolloutIdAndActive(final Long rolloutId) {
|
||||
return (root, query, cb) -> cb.and(
|
||||
cb.equal(root.get(JpaAction_.rollout).get(AbstractJpaBaseEntity_.id), rolloutId),
|
||||
cb.equal(root.get(JpaAction_.active), true)
|
||||
);
|
||||
}
|
||||
|
||||
public static Specification<JpaAction> byRolloutIdAndActiveAndStatusIsNot(final Long rolloutId, final List<Action.Status> statuses) {
|
||||
return (root, query, cb) -> cb.and(
|
||||
cb.equal(root.get(JpaAction_.rollout).get(AbstractJpaBaseEntity_.id), rolloutId),
|
||||
cb.equal(root.get(JpaAction_.active), true),
|
||||
cb.not(root.get(JpaAction_.status).in(statuses))
|
||||
);
|
||||
}
|
||||
|
||||
public static Specification<JpaAction> byDistributionSetIdAndActiveAndStatusIsNot(final Long distributionSetId, final Action.Status status) {
|
||||
return (root, query, cb) -> cb.and(
|
||||
cb.equal(root.get(JpaAction_.distributionSet).get(AbstractJpaBaseEntity_.id), distributionSetId),
|
||||
|
||||
Reference in New Issue
Block a user