Performance optimizations for Multi-Assignments (#858)

* Add remote event test for the new MultiActionEvent * Improve test descriptions * Improve sendMultiActionRequestMessages * Moved action filtering to the database query level (#12) * Use @ExpectedEvents instead of EventHandlerStubs * Removed @Param from 'existsByTargetControllerIdAndStatusAndActiveIsTrue' * Reverted metadata initialization * Fix hawkBit bot findings Signed-off-by: Stefan Behl <stefan.behl@bosch-si.com> Signed-off-by: Ahmed Sayed <ahmed.sayed@bosch-si.com>
2019-07-19 14:47:29 +02:00
parent 749218098f
commit 4e9308a949
15 changed files with 133 additions and 243 deletions
--- a/hawkbit-dmf/hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/AmqpMessageDispatcherService.java
+++ b/hawkbit-dmf/hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/AmqpMessageDispatcherService.java
@@ -16,6 +16,7 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
+import java.util.function.Function;
 import java.util.stream.Collectors;

 import org.eclipse.hawkbit.api.ApiType;
@@ -65,8 +66,6 @@ import org.springframework.context.event.EventListener;
 import org.springframework.data.domain.PageRequest;
 import org.springframework.util.CollectionUtils;

-import com.google.common.collect.Maps;
-
 /**
 * {@link AmqpMessageDispatcherService} create all outgoing AMQP messages and
 * delegate the messages to a {@link AmqpMessageSenderService}.
@@ -172,7 +171,7 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {

    protected void sendMultiActionRequestMessages(final String tenant, final List<String> controllerIds) {

-        final Map<Long, Map<SoftwareModule, List<SoftwareModuleMetadata>>> softwareModuleMetadata = new HashMap<>();
+        final Map<SoftwareModule, List<SoftwareModuleMetadata>> softwareModuleMetadata = new HashMap<>();
        targetManagement.getByControllerID(controllerIds).stream()
                .filter(target -> IpUtil.isAmqpUri(target.getAddress())).forEach(target -> {

@@ -180,14 +179,13 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
                            .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));
-                    });
+                    activeActions.forEach(action -> action.getDistributionSet().getModules().forEach(
+                            module -> softwareModuleMetadata.computeIfAbsent(module, this::getSoftwareModuleMetadata)));

                    if (!activeActions.isEmpty()) {
-                        sendMultiActionRequestToTarget(tenant, target, activeActions, softwareModuleMetadata);
+                        sendMultiActionRequestToTarget(tenant, target, activeActions,
+                                action -> action.getDistributionSet().getModules().stream()
+                                        .collect(Collectors.toMap(m -> m, softwareModuleMetadata::get)));
                    }

                });
@@ -195,7 +193,7 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
    }

    protected void sendMultiActionRequestToTarget(final String tenant, final Target target, final List<Action> actions,
-            final Map<Long, Map<SoftwareModule, List<SoftwareModuleMetadata>>> softwareModulesPerDistributionSet) {
+            final Function<Action, Map<SoftwareModule, List<SoftwareModuleMetadata>>> getSoftwareModuleMetaData) {

        final URI targetAdress = target.getAddress();
        if (!IpUtil.isAmqpUri(targetAdress) || CollectionUtils.isEmpty(actions)) {
@@ -205,7 +203,7 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
        final DmfMultiActionRequest multiActionRequest = new DmfMultiActionRequest();
        actions.forEach(action -> {
            final DmfActionRequest actionRequest = createDmfActionRequest(target, action,
-                    softwareModulesPerDistributionSet.get(action.getDistributionSet().getId()));
+                    getSoftwareModuleMetaData.apply(action));
            multiActionRequest.addElement(getEventTypeForAction(action), actionRequest);
        });

@@ -471,15 +469,12 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {

    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;
+        return distributionSet.getModules().stream().collect(Collectors.toMap(m -> m, this::getSoftwareModuleMetadata));
+    }
+
+    private List<SoftwareModuleMetadata> getSoftwareModuleMetadata(final SoftwareModule module) {
+        return softwareModuleManagement.findMetaDataBySoftwareModuleIdAndTargetVisible(
+                PageRequest.of(0, RepositoryConstants.MAX_META_DATA_COUNT), module.getId()).getContent();
    }

 }
--- a/hawkbit-dmf/hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/AmqpMessageHandlerService.java
+++ b/hawkbit-dmf/hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/AmqpMessageHandlerService.java
@@ -227,7 +227,7 @@ public class AmqpMessageHandlerService extends BaseAmqpService {
                .stream().collect(Collectors.toMap(DistributionSet::getId, this::getSoftwareModulesWithMetadata));

        amqpMessageDispatcherService.sendMultiActionRequestToTarget(target.getTenant(), target, actions,
-                softwareModulesPerDistributionSet);
+                action -> softwareModulesPerDistributionSet.get(action.getDistributionSet().getId()));
    }

    private void sendOldestActionToTarget(final Target target) {
--- a/hawkbit-dmf/hawkbit-dmf-amqp/src/test/java/org/eclipse/hawkbit/integration/AmqpMessageDispatcherServiceIntegrationTest.java
+++ b/hawkbit-dmf/hawkbit-dmf-amqp/src/test/java/org/eclipse/hawkbit/integration/AmqpMessageDispatcherServiceIntegrationTest.java
@@ -135,21 +135,21 @@ public class AmqpMessageDispatcherServiceIntegrationTest extends AbstractAmqpSer
    @Test
    @Description("Verify that a distribution assignment multiple times send cancel and assign events with right softwaremodules")
    @ExpectEvents({ @Expect(type = TargetCreatedEvent.class, count = 1),
-            @Expect(type = TargetAssignDistributionSetEvent.class, count = 1),
+            @Expect(type = TargetAssignDistributionSetEvent.class, count = 2),
            @Expect(type = CancelTargetAssignmentEvent.class, count = 1),
            @Expect(type = ActionCreatedEvent.class, count = 2), @Expect(type = ActionUpdatedEvent.class, count = 1),
            @Expect(type = SoftwareModuleCreatedEvent.class, count = 6),
-            @Expect(type = SoftwareModuleUpdatedEvent.class, count = 6),
+            @Expect(type = SoftwareModuleUpdatedEvent.class, count = 12),
            @Expect(type = DistributionSetCreatedEvent.class, count = 2),
-            @Expect(type = TargetUpdatedEvent.class, count = 2), @Expect(type = TargetPollEvent.class, count = 3) })
+            @Expect(type = TargetUpdatedEvent.class, count = 2), @Expect(type = TargetPollEvent.class, count = 2) })
    public void assignDistributionSetMultipleTimes() {
        final String controllerId = TARGET_PREFIX + "assignDistributionSetMultipleTimes";

        final DistributionSetAssignmentResult assignmentResult = registerTargetAndAssignDistributionSet(controllerId);
-
        final DistributionSet distributionSet2 = testdataFactory.createDistributionSet(UUID.randomUUID().toString());
-        registerTargetAndAssignDistributionSet(distributionSet2.getId(), TargetUpdateStatus.PENDING,
-                getDistributionSet().getModules(), controllerId);
+        testdataFactory.addSoftwareModuleMetadata(distributionSet2);
+        assignDistributionSet(distributionSet2.getId(), controllerId);
+        assertDownloadAndInstallMessage(distributionSet2.getModules(), controllerId);
        assertCancelActionMessage(assignmentResult.getActionIds().get(0), controllerId);

        createAndSendThingCreated(controllerId, TENANT_EXIST);
--- a/hawkbit-dmf/hawkbit-dmf-amqp/src/test/java/org/eclipse/hawkbit/matcher/SoftwareModuleJsonMatcher.java
+++ b/hawkbit-dmf/hawkbit-dmf-amqp/src/test/java/org/eclipse/hawkbit/matcher/SoftwareModuleJsonMatcher.java
@@ -57,50 +57,27 @@ public final class SoftwareModuleJsonMatcher {
            this.expectedModules = expectedModules;
        }

-        static boolean containsExactly(final Object actual, final List<DmfSoftwareModule> expected) {
+        boolean containsExactly(final Object actual) {
            if (actual == null) {
-                return expected == null;
+                return expectedModules == null;
            }

            @SuppressWarnings("unchecked")
            final Collection<SoftwareModule> modules = (Collection<SoftwareModule>) actual;

-            if (modules.size() != expected.size()) {
-                return false;
-            }
+            return expectedModules.stream().allMatch(e -> existsIn(e, modules));
+        }

-            for (final SoftwareModule repoSoftwareModule : modules) {
-                boolean containsElement = false;
-
-                for (final DmfSoftwareModule jsonSoftwareModule : expected) {
-                    if (!jsonSoftwareModule.getModuleId().equals(repoSoftwareModule.getId())) {
-                        continue;
-                    }
-                    containsElement = true;
-
-                    if (!jsonSoftwareModule.getModuleType().equals(repoSoftwareModule.getType().getKey())) {
-                        return false;
-                    }
-                    if (!jsonSoftwareModule.getModuleVersion().equals(repoSoftwareModule.getVersion())) {
-                        return false;
-                    }
-                    if (jsonSoftwareModule.getArtifacts().size() != repoSoftwareModule.getArtifacts().size()) {
-                        return false;
-                    }
-                }
-
-                if (!containsElement) {
-                    return false;
-                }
-
-            }
-
-            return true;
+        private static boolean existsIn(final DmfSoftwareModule module, final Collection<SoftwareModule> actual) {
+            return actual.stream()
+                    .anyMatch(e -> module.getModuleType().equals(e.getType().getKey())
+                            && module.getModuleVersion().equals(e.getVersion())
+                            && module.getArtifacts().size() == e.getArtifacts().size());
        }

        @Override
        public boolean matches(final Object actualValue) {
-            return containsExactly(actualValue, expectedModules);
+            return containsExactly(actualValue);
        }

        @Override