Support for Multi-Assignments (#833)
* First draft of new assignment logic Signed-off-by: Stefan Behl <stefan.behl@bosch-si.com> * Enhancements of System Configuration view Signed-off-by: Stefan Behl <stefan.behl@bosch-si.com> * Drag&drop enhancements for multiple distribution set assignments Signed-off-by: Stefan Behl <stefan.behl@bosch-si.com> * Misc fixes Signed-off-by: Stefan Behl <stefan.behl@bosch-si.com> * test that previous assignments are not canceled Signed-off-by: Stefan Klotz <stefan.klotz@bosch-si.com> * add description and expected events to test Signed-off-by: Stefan Klotz <stefan.klotz@bosch-si.com> * extend TenantConfigurationManagement by NullPointerException proof getConfigurationValue method Signed-off-by: Stefan Klotz <stefan.klotz@bosch-si.com> * Hide "Required Migration Step" if Multi Assignments is enabled Signed-off-by: Stefan Klotz <stefan.klotz@bosch-si.com> * Make fields transient Signed-off-by: Stefan Klotz <stefan.klotz@bosch-si.com> * Add IDs for Required Migration Step elements Signed-off-by: Stefan Klotz <stefan.klotz@bosch-si.com> * Save work in progress Signed-off-by: Stefan Behl <stefan.behl@bosch-si.com> * Added new DMF message DmfMultiActionRequest Signed-off-by: Stefan Behl <stefan.behl@bosch-si.com> * DMF enhancements to send out MultiActionRequest messages Signed-off-by: Stefan Behl <stefan.behl@bosch-si.com> * Minor changes Signed-off-by: Stefan Behl <stefan.behl@bosch-si.com> * Multi Assignment support for cancellations Signed-off-by: Stefan Behl <stefan.behl@bosch-si.com> * fix permission problems and immutable lists Signed-off-by: Stefan Klotz <stefan.klotz@bosch-si.com> * add message dispatcher tests for outgoing multiassignment messages, fix old tests Signed-off-by: Stefan Klotz <stefan.klotz@bosch-si.com> * Implement Multi-Assignment support for rollout groups Signed-off-by: Stefan Behl <stefan.behl@bosch-si.com> * Minor changes Signed-off-by: Stefan Behl <stefan.behl@bosch-si.com> * Refactoring Signed-off-by: Stefan Behl <stefan.behl@bosch-si.com> * Register new deployment event with protobuff framework Signed-off-by: Stefan Behl <stefan.behl@bosch-si.com> * Allow same DS to be assigned multiple times Signed-off-by: Stefan Behl <stefan.behl@bosch-si.com> * Fix assignment with pending cancellations Signed-off-by: Stefan Behl <stefan.behl@bosch-si.com> * Reduce repository /DB calls Signed-off-by: Stefan Behl <stefan.behl@bosch-si.com> * Revert latest perf fix (causing a regression) Signed-off-by: Stefan Behl <stefan.behl@bosch-si.com> * test if a rollout sends multiaction messages in multiassignment mode Signed-off-by: Stefan Klotz <stefan.klotz@bosch-si.com> * Minor changes Signed-off-by: Stefan Behl <stefan.behl@bosch-si.com> * Do not close new action if DS is already assigned (if multi-assign on) Signed-off-by: Stefan Behl <stefan.behl@bosch-si.com> * add test that starts and finishes multiple rollouts in multiassignt mode Signed-off-by: Stefan Klotz <stefan.klotz@bosch-si.com> * add javadoc to test method Signed-off-by: Stefan Klotz <stefan.klotz@bosch-si.com> * Prevent Multi-Assignments from being disabled via Repo Config UI Signed-off-by: Stefan Behl <stefan.behl@bosch-si.com> * Add link to Provisioning State Machine Signed-off-by: Stefan Behl <stefan.behl@bosch-si.com> * test that Multiassignment can not be disabled via mgmt-api Signed-off-by: Stefan Klotz <stefan.klotz@bosch-si.com> * refactor AmqpMessageHandlerService code Signed-off-by: Stefan Klotz <stefan.klotz@bosch-si.com> * Prevent Multi-Assignments feature from being disabled via Mgmt REST API Signed-off-by: Stefan Behl <stefan.behl@bosch-si.com> * add license header, remove unused instance variables Signed-off-by: Stefan Klotz <stefan.klotz@bosch-si.com> * fix tenantConfigurationManagement mock in test Signed-off-by: Stefan Klotz <stefan.klotz@bosch-si.com> * return empty list instead of null Signed-off-by: Stefan Klotz <stefan.klotz@bosch-si.com> * add ddi test for multiassignment, fix old test Signed-off-by: Stefan Klotz <stefan.klotz@bosch-si.com> * Prevent autoclose from being modified if Multi-Assignments is enabled # WARNING: head commit changed in the meantime Signed-off-by: Stefan Behl <stefan.behl@bosch-si.com> * Add test for autoClose /multiAssign Signed-off-by: Stefan Behl <stefan.behl@bosch-si.com> * Javadoc improvements Signed-off-by: Stefan Behl <stefan.behl@bosch-si.com> * change test method that waits for dmf messages to be dispatched Signed-off-by: Stefan Klotz <stefan.klotz@bosch-si.com> * clean up code Signed-off-by: Stefan Klotz <stefan.klotz@bosch-si.com> * Fix UI-related PR review findings Signed-off-by: Stefan Behl <stefan.behl@bosch-si.com> * Fix PR review findings Signed-off-by: Stefan Behl <stefan.behl@bosch-si.com> * Fix PR review findings Signed-off-by: Stefan Behl <stefan.behl@bosch-si.com> * Fix PR review findings Signed-off-by: Stefan Behl <stefan.behl@bosch-si.com> * Fix PR review findings, Sonar issues, and test failures Signed-off-by: Stefan Behl <stefan.behl@bosch-si.com> * Fix PR review findings Signed-off-by: Stefan Behl <stefan.behl@bosch-si.com> * Fix PR review findings Signed-off-by: Stefan Behl <stefan.behl@bosch-si.com> * Fix PR review findings Signed-off-by: Stefan Behl <stefan.behl@bosch-si.com> * Fix Sonar findings Signed-off-by: Stefan Behl <stefan.behl@bosch-si.com> * Fix PR review findings Signed-off-by: Stefan Behl <stefan.behl@bosch-si.com> * Fix PR review findings Signed-off-by: Stefan Behl <stefan.behl@bosch-si.com> * Fix PR review findings Signed-off-by: Stefan Behl <stefan.behl@bosch-si.com>
This commit is contained in:
Stefan Behl
committed by
Dominic Schabel
Dominic Schabel
parent
f16f986eed
commit
1ac40a7a95
@@ -17,6 +17,7 @@ import org.eclipse.hawkbit.cache.DownloadIdCache;
|
||||
import org.eclipse.hawkbit.dmf.amqp.api.AmqpSettings;
|
||||
import org.eclipse.hawkbit.repository.ArtifactManagement;
|
||||
import org.eclipse.hawkbit.repository.ControllerManagement;
|
||||
import org.eclipse.hawkbit.repository.DeploymentManagement;
|
||||
import org.eclipse.hawkbit.repository.DistributionSetManagement;
|
||||
import org.eclipse.hawkbit.repository.EntityFactory;
|
||||
import org.eclipse.hawkbit.repository.SoftwareModuleManagement;
|
||||
@@ -240,9 +241,11 @@ public class AmqpConfiguration {
|
||||
@Bean
|
||||
public AmqpMessageHandlerService amqpMessageHandlerService(final RabbitTemplate rabbitTemplate,
|
||||
final AmqpMessageDispatcherService amqpMessageDispatcherService,
|
||||
final ControllerManagement controllerManagement, final EntityFactory entityFactory) {
|
||||
final ControllerManagement controllerManagement, final EntityFactory entityFactory,
|
||||
final SystemSecurityContext systemSecurityContext,
|
||||
final TenantConfigurationManagement tenantConfigurationManagement) {
|
||||
return new AmqpMessageHandlerService(rabbitTemplate, amqpMessageDispatcherService, controllerManagement,
|
||||
entityFactory);
|
||||
entityFactory, systemSecurityContext, tenantConfigurationManagement);
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -319,10 +322,10 @@ public class AmqpConfiguration {
|
||||
final AmqpMessageSenderService amqpSenderService, final ArtifactUrlHandler artifactUrlHandler,
|
||||
final SystemSecurityContext systemSecurityContext, final SystemManagement systemManagement,
|
||||
final TargetManagement targetManagement, final DistributionSetManagement distributionSetManagement,
|
||||
final SoftwareModuleManagement softwareModuleManagement) {
|
||||
final SoftwareModuleManagement softwareModuleManagement, final DeploymentManagement deploymentManagement) {
|
||||
return new AmqpMessageDispatcherService(rabbitTemplate, amqpSenderService, artifactUrlHandler,
|
||||
systemSecurityContext, systemManagement, targetManagement, serviceMatcher, distributionSetManagement,
|
||||
softwareModuleManagement);
|
||||
softwareModuleManagement, deploymentManagement);
|
||||
}
|
||||
|
||||
private static Map<String, Object> getTTLMaxArgsAuthenticationQueue() {
|
||||
|
||||
@@ -8,8 +8,11 @@
|
||||
*/
|
||||
package org.eclipse.hawkbit.amqp;
|
||||
|
||||
import static org.eclipse.hawkbit.repository.RepositoryConstants.MAX_ACTION_COUNT;
|
||||
|
||||
import java.net.URI;
|
||||
import java.util.Collections;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Map.Entry;
|
||||
@@ -28,12 +31,15 @@ import org.eclipse.hawkbit.dmf.json.model.DmfArtifact;
|
||||
import org.eclipse.hawkbit.dmf.json.model.DmfArtifactHash;
|
||||
import org.eclipse.hawkbit.dmf.json.model.DmfDownloadAndUpdateRequest;
|
||||
import org.eclipse.hawkbit.dmf.json.model.DmfMetadata;
|
||||
import org.eclipse.hawkbit.dmf.json.model.DmfMultiActionRequest;
|
||||
import org.eclipse.hawkbit.dmf.json.model.DmfSoftwareModule;
|
||||
import org.eclipse.hawkbit.repository.DeploymentManagement;
|
||||
import org.eclipse.hawkbit.repository.DistributionSetManagement;
|
||||
import org.eclipse.hawkbit.repository.RepositoryConstants;
|
||||
import org.eclipse.hawkbit.repository.SoftwareModuleManagement;
|
||||
import org.eclipse.hawkbit.repository.SystemManagement;
|
||||
import org.eclipse.hawkbit.repository.TargetManagement;
|
||||
import org.eclipse.hawkbit.repository.event.remote.MultiActionEvent;
|
||||
import org.eclipse.hawkbit.repository.event.remote.TargetAssignDistributionSetEvent;
|
||||
import org.eclipse.hawkbit.repository.event.remote.TargetAttributesRequestedEvent;
|
||||
import org.eclipse.hawkbit.repository.event.remote.TargetDeletedEvent;
|
||||
@@ -41,6 +47,7 @@ import org.eclipse.hawkbit.repository.event.remote.entity.CancelTargetAssignment
|
||||
import org.eclipse.hawkbit.repository.model.Action;
|
||||
import org.eclipse.hawkbit.repository.model.ActionProperties;
|
||||
import org.eclipse.hawkbit.repository.model.Artifact;
|
||||
import org.eclipse.hawkbit.repository.model.DistributionSet;
|
||||
import org.eclipse.hawkbit.repository.model.SoftwareModule;
|
||||
import org.eclipse.hawkbit.repository.model.SoftwareModuleMetadata;
|
||||
import org.eclipse.hawkbit.repository.model.Target;
|
||||
@@ -79,6 +86,7 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
|
||||
private final TargetManagement targetManagement;
|
||||
private final ServiceMatcher serviceMatcher;
|
||||
private final DistributionSetManagement distributionSetManagement;
|
||||
private final DeploymentManagement deploymentManagement;
|
||||
private final SoftwareModuleManagement softwareModuleManagement;
|
||||
|
||||
/**
|
||||
@@ -107,7 +115,7 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
|
||||
final SystemSecurityContext systemSecurityContext, final SystemManagement systemManagement,
|
||||
final TargetManagement targetManagement, final ServiceMatcher serviceMatcher,
|
||||
final DistributionSetManagement distributionSetManagement,
|
||||
final SoftwareModuleManagement softwareModuleManagement) {
|
||||
final SoftwareModuleManagement softwareModuleManagement, final DeploymentManagement deploymentManagement) {
|
||||
super(rabbitTemplate);
|
||||
this.artifactUrlHandler = artifactUrlHandler;
|
||||
this.amqpSenderService = amqpSenderService;
|
||||
@@ -117,6 +125,7 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
|
||||
this.serviceMatcher = serviceMatcher;
|
||||
this.distributionSetManagement = distributionSetManagement;
|
||||
this.softwareModuleManagement = softwareModuleManagement;
|
||||
this.deploymentManagement = deploymentManagement;
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -134,30 +143,112 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
|
||||
|
||||
LOG.debug("targetAssignDistributionSet retrieved. I will forward it to DMF broker.");
|
||||
|
||||
distributionSetManagement.get(assignedEvent.getDistributionSetId()).ifPresent(set -> {
|
||||
distributionSetManagement.get(assignedEvent.getDistributionSetId()).ifPresent(ds -> {
|
||||
|
||||
final Map<SoftwareModule, List<SoftwareModuleMetadata>> modules = Maps
|
||||
.newHashMapWithExpectedSize(set.getModules().size());
|
||||
set.getModules()
|
||||
.forEach(
|
||||
module -> modules.put(module,
|
||||
softwareModuleManagement.findMetaDataBySoftwareModuleIdAndTargetVisible(
|
||||
PageRequest.of(0, RepositoryConstants.MAX_META_DATA_COUNT), module.getId())
|
||||
.getContent()));
|
||||
final Map<SoftwareModule, List<SoftwareModuleMetadata>> softwareModules = getSoftwareModulesWithMetadata(
|
||||
ds);
|
||||
|
||||
targetManagement.getByControllerID(assignedEvent.getActions().keySet()).forEach(
|
||||
target -> sendUpdateMessageToTarget(assignedEvent.getActions().get(target.getControllerId()),
|
||||
target, modules));
|
||||
target, softwareModules));
|
||||
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Listener for Multi-Action events.
|
||||
*
|
||||
* @param e
|
||||
* the Multi-Action event to be processed
|
||||
*/
|
||||
@EventListener(classes = MultiActionEvent.class)
|
||||
protected void onMultiAction(final MultiActionEvent e) {
|
||||
if (isNotFromSelf(e)) {
|
||||
return;
|
||||
}
|
||||
LOG.debug("MultiActionEvent received for {}", e.getControllerIds());
|
||||
sendMultiActionRequestMessages(e.getTenant(), e.getControllerIds());
|
||||
}
|
||||
|
||||
protected void sendMultiActionRequestMessages(final String tenant, final List<String> controllerIds) {
|
||||
|
||||
final Map<Long, Map<SoftwareModule, List<SoftwareModuleMetadata>>> softwareModuleMetadata = new HashMap<>();
|
||||
targetManagement.getByControllerID(controllerIds).stream()
|
||||
.filter(target -> IpUtil.isAmqpUri(target.getAddress())).forEach(target -> {
|
||||
|
||||
final List<Action> activeActions = deploymentManagement
|
||||
.findActiveActionsByTarget(PageRequest.of(0, MAX_ACTION_COUNT), target.getControllerId())
|
||||
.getContent();
|
||||
|
||||
activeActions.forEach(action -> {
|
||||
final DistributionSet distributionSet = action.getDistributionSet();
|
||||
softwareModuleMetadata.computeIfAbsent(distributionSet.getId(),
|
||||
id -> getSoftwareModulesWithMetadata(distributionSet));
|
||||
});
|
||||
|
||||
if (!activeActions.isEmpty()) {
|
||||
sendMultiActionRequestToTarget(tenant, target, activeActions, softwareModuleMetadata);
|
||||
}
|
||||
|
||||
});
|
||||
|
||||
}
|
||||
|
||||
protected void sendMultiActionRequestToTarget(final String tenant, final Target target, final List<Action> actions,
|
||||
final Map<Long, Map<SoftwareModule, List<SoftwareModuleMetadata>>> softwareModulesPerDistributionSet) {
|
||||
|
||||
final URI targetAdress = target.getAddress();
|
||||
if (!IpUtil.isAmqpUri(targetAdress) || CollectionUtils.isEmpty(actions)) {
|
||||
return;
|
||||
}
|
||||
|
||||
final DmfMultiActionRequest multiActionRequest = new DmfMultiActionRequest();
|
||||
actions.forEach(action -> {
|
||||
final DmfActionRequest actionRequest = createDmfActionRequest(target, action,
|
||||
softwareModulesPerDistributionSet.get(action.getDistributionSet().getId()));
|
||||
multiActionRequest.addElement(getEventTypeForAction(action), actionRequest);
|
||||
});
|
||||
|
||||
final Message message = getMessageConverter().toMessage(multiActionRequest,
|
||||
createConnectorMessagePropertiesEvent(tenant, target.getControllerId(), EventTopic.MULTI_ACTION));
|
||||
amqpSenderService.sendMessage(message, targetAdress);
|
||||
|
||||
}
|
||||
|
||||
private DmfActionRequest createDmfActionRequest(final Target target, final Action action,
|
||||
final Map<SoftwareModule, List<SoftwareModuleMetadata>> softwareModules) {
|
||||
if (action.isCancelingOrCanceled()) {
|
||||
return createPlainActionRequest(action);
|
||||
}
|
||||
return createDownloadAndUpdateRequest(target, action, softwareModules);
|
||||
|
||||
}
|
||||
|
||||
private static DmfActionRequest createPlainActionRequest(final Action action) {
|
||||
final DmfActionRequest actionRequest = new DmfActionRequest();
|
||||
actionRequest.setActionId(action.getId());
|
||||
return actionRequest;
|
||||
}
|
||||
|
||||
private DmfDownloadAndUpdateRequest createDownloadAndUpdateRequest(final Target target, final Action action,
|
||||
final Map<SoftwareModule, List<SoftwareModuleMetadata>> softwareModules) {
|
||||
final DmfDownloadAndUpdateRequest request = new DmfDownloadAndUpdateRequest();
|
||||
request.setActionId(action.getId());
|
||||
request.setTargetSecurityToken(systemSecurityContext.runAsSystem(target::getSecurityToken));
|
||||
if (softwareModules != null) {
|
||||
softwareModules.entrySet()
|
||||
.forEach(entry -> request.addSoftwareModule(convertToAmqpSoftwareModule(target, entry)));
|
||||
}
|
||||
return request;
|
||||
}
|
||||
|
||||
/**
|
||||
* Method to get the type of event depending on whether the action is a
|
||||
* DOWNLOAD_ONLY action or if it has a valid maintenance window available
|
||||
* or not based on defined maintenance schedule. In case of no maintenance
|
||||
* schedule or if there is a valid window available, the topic {@link EventTopic#DOWNLOAD_AND_INSTALL} is
|
||||
* returned else {@link EventTopic#DOWNLOAD} is returned.
|
||||
* DOWNLOAD_ONLY action or if it has a valid maintenance window available or
|
||||
* not based on defined maintenance schedule. In case of no maintenance
|
||||
* schedule or if there is a valid window available, the topic
|
||||
* {@link EventTopic#DOWNLOAD_AND_INSTALL} is returned else
|
||||
* {@link EventTopic#DOWNLOAD} is returned.
|
||||
*
|
||||
* @param action
|
||||
* current action properties.
|
||||
@@ -165,8 +256,24 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
|
||||
* @return {@link EventTopic} to use for message.
|
||||
*/
|
||||
private static EventTopic getEventTypeForTarget(final ActionProperties action) {
|
||||
return (Action.ActionType.DOWNLOAD_ONLY.equals(action.getActionType()) ||
|
||||
!action.isMaintenanceWindowAvailable()) ? EventTopic.DOWNLOAD : EventTopic.DOWNLOAD_AND_INSTALL;
|
||||
return (Action.ActionType.DOWNLOAD_ONLY.equals(action.getActionType())
|
||||
|| !action.isMaintenanceWindowAvailable()) ? EventTopic.DOWNLOAD : EventTopic.DOWNLOAD_AND_INSTALL;
|
||||
}
|
||||
|
||||
/**
|
||||
* Determines the {@link EventTopic} for the given {@link Action}, depending
|
||||
* on its action type.
|
||||
*
|
||||
* @param action
|
||||
* to obtain the corresponding {@link EventTopic} for
|
||||
*
|
||||
* @return the {@link EventTopic} for this action
|
||||
*/
|
||||
private static EventTopic getEventTypeForAction(final Action action) {
|
||||
if (action.isCancelingOrCanceled()) {
|
||||
return EventTopic.CANCEL_DOWNLOAD;
|
||||
}
|
||||
return getEventTypeForTarget(new ActionProperties(action));
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -174,7 +281,7 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
|
||||
* the Distribution set to a Target has been canceled.
|
||||
*
|
||||
* @param cancelEvent
|
||||
* the object to be send.
|
||||
* that is to be converted to a DMF message
|
||||
*/
|
||||
@EventListener(classes = CancelTargetAssignmentEvent.class)
|
||||
protected void targetCancelAssignmentToDistributionSet(final CancelTargetAssignmentEvent cancelEvent) {
|
||||
@@ -211,7 +318,7 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
|
||||
protected void sendUpdateMessageToTarget(final ActionProperties action, final Target target,
|
||||
final Map<SoftwareModule, List<SoftwareModuleMetadata>> modules) {
|
||||
|
||||
String tenant = action.getTenant();
|
||||
final String tenant = action.getTenant();
|
||||
|
||||
final URI targetAdress = target.getAddress();
|
||||
if (!IpUtil.isAmqpUri(targetAdress)) {
|
||||
@@ -361,4 +468,17 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
|
||||
return artifact;
|
||||
}
|
||||
|
||||
}
|
||||
private Map<SoftwareModule, List<SoftwareModuleMetadata>> getSoftwareModulesWithMetadata(
|
||||
final DistributionSet distributionSet) {
|
||||
final Map<SoftwareModule, List<SoftwareModuleMetadata>> moduleMetadata = Maps
|
||||
.newHashMapWithExpectedSize(distributionSet.getModules().size());
|
||||
distributionSet.getModules()
|
||||
.forEach(
|
||||
module -> moduleMetadata.put(module,
|
||||
softwareModuleManagement.findMetaDataBySoftwareModuleIdAndTargetVisible(
|
||||
PageRequest.of(0, RepositoryConstants.MAX_META_DATA_COUNT), module.getId())
|
||||
.getContent()));
|
||||
return moduleMetadata;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -8,11 +8,16 @@
|
||||
*/
|
||||
package org.eclipse.hawkbit.amqp;
|
||||
|
||||
import static org.eclipse.hawkbit.repository.RepositoryConstants.MAX_ACTION_COUNT;
|
||||
import static org.eclipse.hawkbit.tenancy.configuration.TenantConfigurationProperties.TenantConfigurationKey.MULTI_ASSIGNMENTS_ENABLED;
|
||||
|
||||
import java.io.Serializable;
|
||||
import java.net.URI;
|
||||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Optional;
|
||||
import java.util.Set;
|
||||
import java.util.UUID;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
@@ -27,14 +32,17 @@ import org.eclipse.hawkbit.im.authentication.TenantAwareAuthenticationDetails;
|
||||
import org.eclipse.hawkbit.repository.ControllerManagement;
|
||||
import org.eclipse.hawkbit.repository.EntityFactory;
|
||||
import org.eclipse.hawkbit.repository.RepositoryConstants;
|
||||
import org.eclipse.hawkbit.repository.TenantConfigurationManagement;
|
||||
import org.eclipse.hawkbit.repository.UpdateMode;
|
||||
import org.eclipse.hawkbit.repository.builder.ActionStatusCreate;
|
||||
import org.eclipse.hawkbit.repository.model.Action;
|
||||
import org.eclipse.hawkbit.repository.model.Action.Status;
|
||||
import org.eclipse.hawkbit.repository.model.ActionProperties;
|
||||
import org.eclipse.hawkbit.repository.model.DistributionSet;
|
||||
import org.eclipse.hawkbit.repository.model.SoftwareModule;
|
||||
import org.eclipse.hawkbit.repository.model.SoftwareModuleMetadata;
|
||||
import org.eclipse.hawkbit.repository.model.Target;
|
||||
import org.eclipse.hawkbit.security.SystemSecurityContext;
|
||||
import org.eclipse.hawkbit.util.IpUtil;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
@@ -42,6 +50,7 @@ import org.springframework.amqp.AmqpRejectAndDontRequeueException;
|
||||
import org.springframework.amqp.core.Message;
|
||||
import org.springframework.amqp.rabbit.annotation.RabbitListener;
|
||||
import org.springframework.amqp.rabbit.core.RabbitTemplate;
|
||||
import org.springframework.data.domain.PageRequest;
|
||||
import org.springframework.messaging.handler.annotation.Header;
|
||||
import org.springframework.security.authentication.AnonymousAuthenticationToken;
|
||||
import org.springframework.security.core.Authentication;
|
||||
@@ -51,8 +60,6 @@ import org.springframework.security.core.context.SecurityContextHolder;
|
||||
import org.springframework.security.core.context.SecurityContextImpl;
|
||||
import org.springframework.util.StringUtils;
|
||||
|
||||
import com.google.common.collect.Maps;
|
||||
|
||||
/**
|
||||
*
|
||||
* {@link AmqpMessageHandlerService} handles all incoming target interaction
|
||||
@@ -70,6 +77,10 @@ public class AmqpMessageHandlerService extends BaseAmqpService {
|
||||
|
||||
private final EntityFactory entityFactory;
|
||||
|
||||
private final TenantConfigurationManagement tenantConfigurationManagement;
|
||||
|
||||
private final SystemSecurityContext systemSecurityContext;
|
||||
|
||||
/**
|
||||
* Constructor.
|
||||
*
|
||||
@@ -84,11 +95,15 @@ public class AmqpMessageHandlerService extends BaseAmqpService {
|
||||
*/
|
||||
public AmqpMessageHandlerService(final RabbitTemplate rabbitTemplate,
|
||||
final AmqpMessageDispatcherService amqpMessageDispatcherService,
|
||||
final ControllerManagement controllerManagement, final EntityFactory entityFactory) {
|
||||
final ControllerManagement controllerManagement, final EntityFactory entityFactory,
|
||||
final SystemSecurityContext systemSecurityContext,
|
||||
final TenantConfigurationManagement tenantConfigurationManagement) {
|
||||
super(rabbitTemplate);
|
||||
this.amqpMessageDispatcherService = amqpMessageDispatcherService;
|
||||
this.controllerManagement = controllerManagement;
|
||||
this.entityFactory = entityFactory;
|
||||
this.systemSecurityContext = systemSecurityContext;
|
||||
this.tenantConfigurationManagement = tenantConfigurationManagement;
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -191,11 +206,31 @@ public class AmqpMessageHandlerService extends BaseAmqpService {
|
||||
final Target target = controllerManagement.findOrRegisterTargetIfItDoesNotExist(thingId, amqpUri);
|
||||
LOG.debug("Target {} reported online state.", thingId);
|
||||
|
||||
lookIfUpdateAvailable(target);
|
||||
sendUpdateCommandToTarget(target);
|
||||
}
|
||||
|
||||
private void lookIfUpdateAvailable(final Target target) {
|
||||
private void sendUpdateCommandToTarget(final Target target) {
|
||||
if (isMultiAssignmentsEnabled()) {
|
||||
sendCurrentActionsAsMultiActionToTarget(target);
|
||||
} else {
|
||||
sendOldestActionToTarget(target);
|
||||
}
|
||||
}
|
||||
|
||||
private void sendCurrentActionsAsMultiActionToTarget(final Target target) {
|
||||
final List<Action> actions = controllerManagement
|
||||
.findActiveActionsByTarget(PageRequest.of(0, MAX_ACTION_COUNT), target.getControllerId()).getContent();
|
||||
|
||||
final Set<DistributionSet> distributionSets = actions.stream().map(Action::getDistributionSet)
|
||||
.collect(Collectors.toSet());
|
||||
final Map<Long, Map<SoftwareModule, List<SoftwareModuleMetadata>>> softwareModulesPerDistributionSet = distributionSets
|
||||
.stream().collect(Collectors.toMap(DistributionSet::getId, this::getSoftwareModulesWithMetadata));
|
||||
|
||||
amqpMessageDispatcherService.sendMultiActionRequestToTarget(target.getTenant(), target, actions,
|
||||
softwareModulesPerDistributionSet);
|
||||
}
|
||||
|
||||
private void sendOldestActionToTarget(final Target target) {
|
||||
final Optional<Action> actionOptional = controllerManagement
|
||||
.findOldestActiveActionByTarget(target.getControllerId());
|
||||
|
||||
@@ -207,20 +242,23 @@ public class AmqpMessageHandlerService extends BaseAmqpService {
|
||||
if (action.isCancelingOrCanceled()) {
|
||||
amqpMessageDispatcherService.sendCancelMessageToTarget(target.getTenant(), target.getControllerId(),
|
||||
action.getId(), target.getAddress());
|
||||
return;
|
||||
} else {
|
||||
amqpMessageDispatcherService.sendUpdateMessageToTarget(new ActionProperties(action), action.getTarget(),
|
||||
getSoftwareModulesWithMetadata(action.getDistributionSet()));
|
||||
}
|
||||
}
|
||||
|
||||
final Map<SoftwareModule, List<SoftwareModuleMetadata>> modules = Maps
|
||||
.newHashMapWithExpectedSize(action.getDistributionSet().getModules().size());
|
||||
private Map<SoftwareModule, List<SoftwareModuleMetadata>> getSoftwareModulesWithMetadata(
|
||||
final DistributionSet distributionSet) {
|
||||
final List<Long> smIds = distributionSet.getModules().stream().map(SoftwareModule::getId)
|
||||
.collect(Collectors.toList());
|
||||
|
||||
final Map<Long, List<SoftwareModuleMetadata>> metadata = controllerManagement
|
||||
.findTargetVisibleMetaDataBySoftwareModuleId(action.getDistributionSet().getModules().stream()
|
||||
.map(SoftwareModule::getId).collect(Collectors.toList()));
|
||||
.findTargetVisibleMetaDataBySoftwareModuleId(smIds);
|
||||
|
||||
action.getDistributionSet().getModules().forEach(module -> modules.put(module, metadata.get(module.getId())));
|
||||
return distributionSet.getModules().stream()
|
||||
.collect(Collectors.toMap(sm -> sm, sm -> metadata.getOrDefault(sm.getId(), Collections.emptyList())));
|
||||
|
||||
amqpMessageDispatcherService.sendUpdateMessageToTarget(new ActionProperties(action), action.getTarget(),
|
||||
modules);
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -275,14 +313,19 @@ public class AmqpMessageHandlerService extends BaseAmqpService {
|
||||
final ActionStatusCreate actionStatus = entityFactory.actionStatus().create(action.getId()).status(status)
|
||||
.messages(messages);
|
||||
|
||||
final Action addUpdateActionStatus = getUpdateActionStatus(status, actionStatus);
|
||||
final Action updatedAction = Status.CANCELED.equals(status)
|
||||
? controllerManagement.addCancelActionStatus(actionStatus)
|
||||
: controllerManagement.addUpdateActionStatus(actionStatus);
|
||||
|
||||
if (!addUpdateActionStatus.isActive() || (addUpdateActionStatus.hasMaintenanceSchedule()
|
||||
&& addUpdateActionStatus.isMaintenanceWindowAvailable())) {
|
||||
lookIfUpdateAvailable(action.getTarget());
|
||||
if (shouldTargetProceed(updatedAction)) {
|
||||
sendUpdateCommandToTarget(action.getTarget());
|
||||
}
|
||||
}
|
||||
|
||||
private static boolean shouldTargetProceed(final Action action) {
|
||||
return !action.isActive() || (action.hasMaintenanceSchedule() && action.isMaintenanceWindowAvailable());
|
||||
}
|
||||
|
||||
private static boolean isCorrelationIdNotEmpty(final Message message) {
|
||||
return StringUtils.hasLength(message.getMessageProperties().getCorrelationId());
|
||||
}
|
||||
@@ -337,13 +380,6 @@ public class AmqpMessageHandlerService extends BaseAmqpService {
|
||||
return null;
|
||||
}
|
||||
|
||||
private Action getUpdateActionStatus(final Status status, final ActionStatusCreate actionStatus) {
|
||||
if (Status.CANCELED.equals(status)) {
|
||||
return controllerManagement.addCancelActionStatus(actionStatus);
|
||||
}
|
||||
return controllerManagement.addUpdateActionStatus(actionStatus);
|
||||
}
|
||||
|
||||
// Exception squid:S3655 - logAndThrowMessageError throws exception, i.e.
|
||||
// get will not be called
|
||||
@SuppressWarnings("squid:S3655")
|
||||
@@ -373,4 +409,13 @@ public class AmqpMessageHandlerService extends BaseAmqpService {
|
||||
return null;
|
||||
}
|
||||
|
||||
}
|
||||
private boolean isMultiAssignmentsEnabled() {
|
||||
return getConfigValue(MULTI_ASSIGNMENTS_ENABLED, Boolean.class);
|
||||
}
|
||||
|
||||
private <T extends Serializable> T getConfigValue(final String key, final Class<T> valueType) {
|
||||
return systemSecurityContext
|
||||
.runAsSystem(() -> tenantConfigurationManagement.getConfigurationValue(key, valueType).getValue());
|
||||
}
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user