DMF health check (#577)

* DMF support PING message for health checks. Signed-off-by: kaizimmerm <kai.zimmermann@bosch-si.com> * Device simulator supports PING. Signed-off-by: kaizimmerm <kai.zimmermann@bosch-si.com> * Code cleanup. Signed-off-by: kaizimmerm <kai.zimmermann@bosch-si.com> * Revert accidental checkin. Signed-off-by: kaizimmerm <kai.zimmermann@bosch-si.com> * Fix tests. Signed-off-by: kaizimmerm <kai.zimmermann@bosch-si.com> * Simplify API. Signed-off-by: kaizimmerm <kai.zimmermann@bosch-si.com> * Remove simulator dead letter. Signed-off-by: kaizimmerm <kai.zimmermann@bosch-si.com> * Remove dead code. Signed-off-by: kaizimmerm <kai.zimmermann@bosch-si.com> * Reduce code. Signed-off-by: kaizimmerm <kai.zimmermann@bosch-si.com> * Add message for one more error case. Signed-off-by: kaizimmerm <kai.zimmermann@bosch-si.com>
2017-09-15 12:50:27 +02:00
parent 2818c8505a
commit da13bd79d9
18 changed files with 245 additions and 133 deletions
--- a/hawkbit-dmf/hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/AmqpConfiguration.java
+++ b/hawkbit-dmf/hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/AmqpConfiguration.java
@@ -264,8 +264,8 @@ public class AmqpConfiguration {
     */
    @Bean
    @ConditionalOnMissingBean
-    public AmqpSenderService amqpSenderServiceBean() {
-        return new DefaultAmqpSenderService(rabbitTemplate());
+    public AmqpMessageSenderService amqpSenderServiceBean() {
+        return new DefaultAmqpMessageSenderService(rabbitTemplate());
    }

    @Bean
@@ -325,7 +325,7 @@ public class AmqpConfiguration {
    @Bean
    @ConditionalOnMissingBean(AmqpMessageDispatcherService.class)
    public AmqpMessageDispatcherService amqpMessageDispatcherService(final RabbitTemplate rabbitTemplate,
-            final AmqpSenderService amqpSenderService, final ArtifactUrlHandler artifactUrlHandler,
+            final AmqpMessageSenderService amqpSenderService, final ArtifactUrlHandler artifactUrlHandler,
            final SystemSecurityContext systemSecurityContext, final SystemManagement systemManagement,
            final TargetManagement targetManagement) {
        return new AmqpMessageDispatcherService(rabbitTemplate, amqpSenderService, artifactUrlHandler,
--- 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
@@ -39,6 +39,7 @@ import org.eclipse.hawkbit.util.IpUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.amqp.core.Message;
+import org.springframework.amqp.core.MessageBuilder;
 import org.springframework.amqp.core.MessageProperties;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.cloud.bus.ServiceMatcher;
@@ -47,7 +48,7 @@ import org.springframework.context.event.EventListener;

 /**
 * {@link AmqpMessageDispatcherService} create all outgoing AMQP messages and
- * delegate the messages to a {@link AmqpSenderService}.
+ * delegate the messages to a {@link AmqpMessageSenderService}.
 *
 * Additionally the dispatcher listener/subscribe for some target events e.g.
 * assignment.
@@ -58,7 +59,7 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
    private static final Logger LOG = LoggerFactory.getLogger(AmqpMessageDispatcherService.class);

    private final ArtifactUrlHandler artifactUrlHandler;
-    private final AmqpSenderService amqpSenderService;
+    private final AmqpMessageSenderService amqpSenderService;
    private final SystemSecurityContext systemSecurityContext;
    private final SystemManagement systemManagement;
    private final TargetManagement targetManagement;
@@ -83,10 +84,10 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
     *            to check in cluster case if the message is from the same
     *            cluster node
     */
-    public AmqpMessageDispatcherService(final RabbitTemplate rabbitTemplate, final AmqpSenderService amqpSenderService,
-            final ArtifactUrlHandler artifactUrlHandler, final SystemSecurityContext systemSecurityContext,
-            final SystemManagement systemManagement, final TargetManagement targetManagement,
-            final ServiceMatcher serviceMatcher) {
+    public AmqpMessageDispatcherService(final RabbitTemplate rabbitTemplate,
+            final AmqpMessageSenderService amqpSenderService, final ArtifactUrlHandler artifactUrlHandler,
+            final SystemSecurityContext systemSecurityContext, final SystemManagement systemManagement,
+            final TargetManagement targetManagement, final ServiceMatcher serviceMatcher) {
        super(rabbitTemplate);
        this.artifactUrlHandler = artifactUrlHandler;
        this.amqpSenderService = amqpSenderService;
@@ -143,6 +144,16 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
        amqpSenderService.sendMessage(message, targetAdress);
    }

+    void sendPingReponseToDmfReceiver(final Message ping, final String tenant) {
+        final Message message = MessageBuilder.withBody(String.valueOf(System.currentTimeMillis()).getBytes())
+                .setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN)
+                .setCorrelationId(ping.getMessageProperties().getCorrelationId())
+                .setHeader(MessageHeaderKey.TYPE, MessageType.PING_RESPONSE).setHeader(MessageHeaderKey.TENANT, tenant)
+                .build();
+
+        amqpSenderService.sendMessage(message, ping.getMessageProperties().getReplyTo());
+    }
+
    /**
     * Method to send a message to a RabbitMQ Exchange after the assignment of
     * the Distribution set to a Target has been canceled.
--- 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
@@ -115,20 +115,25 @@ public class AmqpMessageHandlerService extends BaseAmqpService {
     * @return the rpc message back to supplier.
     */
    public Message onMessage(final Message message, final String type, final String tenant, final String virtualHost) {
-        checkContentTypeJson(message);
+
        final SecurityContext oldContext = SecurityContextHolder.getContext();
        try {
            final MessageType messageType = MessageType.valueOf(type);
            switch (messageType) {
            case THING_CREATED:
+                checkContentTypeJson(message);
                setTenantSecurityContext(tenant);
                registerTarget(message, virtualHost);
                break;
            case EVENT:
+                checkContentTypeJson(message);
                setTenantSecurityContext(tenant);
-                final String topicValue = getStringHeaderKey(message, MessageHeaderKey.TOPIC, "EventTopic is null");
-                final EventTopic eventTopic = EventTopic.valueOf(topicValue);
-                handleIncomingEvent(message, eventTopic);
+                handleIncomingEvent(message);
+                break;
+            case PING:
+                if (isCorrelationIdNotEmpty(message)) {
+                    amqpMessageDispatcherService.sendPingReponseToDmfReceiver(message, tenant);
+                }
                break;
            default:
                logAndThrowMessageError(message, "No handle method was found for the given message type.");
@@ -206,8 +211,8 @@ public class AmqpMessageHandlerService extends BaseAmqpService {
     * @param topic
     *            the topic of the event.
     */
-    private void handleIncomingEvent(final Message message, final EventTopic topic) {
-        switch (topic) {
+    private void handleIncomingEvent(final Message message) {
+        switch (EventTopic.valueOf(getStringHeaderKey(message, MessageHeaderKey.TOPIC, "EventTopic is null"))) {
        case UPDATE_ACTION_STATUS:
            updateActionStatus(message);
            break;
--- a/hawkbit-dmf/hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/AmqpMessageSenderService.java
+++ b/hawkbit-dmf/hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/AmqpMessageSenderService.java
@@ -12,13 +12,14 @@ import java.net.URI;

 import javax.validation.constraints.NotNull;

+import org.eclipse.hawkbit.util.IpUtil;
 import org.springframework.amqp.core.Message;

 /**
 * Interface to send a amqp message.
 */
@FunctionalInterface
-public interface AmqpSenderService {
+public interface AmqpMessageSenderService {

    /**
     * Send the given message to the given uri. The uri contains the (virtual)
@@ -29,18 +30,24 @@ public interface AmqpSenderService {
     * @param replyTo
     *            the reply to uri
     */
-    void sendMessage(@NotNull Message message, @NotNull URI replyTo);
+    default void sendMessage(@NotNull final Message message, @NotNull final URI replyTo) {
+        if (!IpUtil.isAmqpUri(replyTo)) {
+            return;
+        }

-    /**
-     * Extract the exchange from the uri. Default implementation removes the
-     * first /.
-     * 
-     * @param amqpUri
-     *            the amqp uri
-     * @return the exchange.
-     */
-    default String extractExchange(final URI amqpUri) {
-        return amqpUri.getPath().substring(1);
+        sendMessage(message, replyTo.getPath().substring(1));
    }

+    /**
+     * Send the given message to the given host and exchange.
+     * 
+     * @param message
+     *            the amqp message
+     * @param exchange
+     *            to send to
+     * @param virtualHost
+     *            to send to
+     */
+    void sendMessage(@NotNull final Message message, @NotNull final String exchange);
+
 }
--- a/hawkbit-dmf/hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/DefaultAmqpMessageSenderService.java
+++ b/hawkbit-dmf/hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/DefaultAmqpMessageSenderService.java
@@ -8,11 +8,9 @@
 */
 package org.eclipse.hawkbit.amqp;

-import java.net.URI;
 import java.nio.charset.StandardCharsets;
 import java.util.UUID;

-import org.eclipse.hawkbit.util.IpUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.amqp.core.Message;
@@ -24,30 +22,27 @@ import org.springframework.amqp.rabbit.support.CorrelationData;
 * message to the configured spring rabbitmq connections. The exchange is
 * extracted from the uri.
 */
-public class DefaultAmqpSenderService implements AmqpSenderService {
-    private static final Logger LOGGER = LoggerFactory.getLogger(DefaultAmqpSenderService.class);
-
-    private final RabbitTemplate internalAmqpTemplate;
+public class DefaultAmqpMessageSenderService extends BaseAmqpService implements AmqpMessageSenderService {
+    private static final Logger LOGGER = LoggerFactory.getLogger(DefaultAmqpMessageSenderService.class);

    /**
     * Constructor.
     * 
-     * @param internalAmqpTemplate
-     *            the amqp template
+     * @param rabbitTemplate
+     *            the AMQP template
     */
-    public DefaultAmqpSenderService(final RabbitTemplate internalAmqpTemplate) {
-        this.internalAmqpTemplate = internalAmqpTemplate;
+    public DefaultAmqpMessageSenderService(final RabbitTemplate rabbitTemplate) {
+        super(rabbitTemplate);
    }

    @Override
-    public void sendMessage(final Message message, final URI replyTo) {
-        if (!IpUtil.isAmqpUri(replyTo)) {
-            return;
-        }
+    public void sendMessage(final Message message, final String exchange) {

        final String correlationId = UUID.randomUUID().toString();
-        final String exchange = extractExchange(replyTo);
-        message.getMessageProperties().setCorrelationId(correlationId.getBytes(StandardCharsets.UTF_8));
+
+        if (isCorrelationIdEmpty(message)) {
+            message.getMessageProperties().setCorrelationId(correlationId.getBytes(StandardCharsets.UTF_8));
+        }

        if (LOGGER.isTraceEnabled()) {
            LOGGER.trace("Sending message {} to exchange {} with correlationId {}", message, exchange, correlationId);
@@ -55,7 +50,12 @@ public class DefaultAmqpSenderService implements AmqpSenderService {
            LOGGER.debug("Sending message to exchange {} with correlationId {}", exchange, correlationId);
        }

-        internalAmqpTemplate.send(exchange, null, message, new CorrelationData(correlationId));
+        getRabbitTemplate().send(exchange, null, message, new CorrelationData(correlationId));
+    }
+
+    protected static boolean isCorrelationIdEmpty(final Message message) {
+        return message.getMessageProperties().getCorrelationId() == null
+                || message.getMessageProperties().getCorrelationId().length <= 0;
    }

 }
--- a/hawkbit-dmf/hawkbit-dmf-amqp/src/test/java/org/eclipse/hawkbit/amqp/AmqpMessageDispatcherServiceTest.java
+++ b/hawkbit-dmf/hawkbit-dmf-amqp/src/test/java/org/eclipse/hawkbit/amqp/AmqpMessageDispatcherServiceTest.java
@@ -81,7 +81,7 @@ public class AmqpMessageDispatcherServiceTest extends AbstractIntegrationTest {

    private RabbitTemplate rabbitTemplate;

-    private DefaultAmqpSenderService senderService;
+    private DefaultAmqpMessageSenderService senderService;

    private Target testTarget;

@@ -94,7 +94,7 @@ public class AmqpMessageDispatcherServiceTest extends AbstractIntegrationTest {
        this.rabbitTemplate = Mockito.mock(RabbitTemplate.class);
        when(rabbitTemplate.getMessageConverter()).thenReturn(new Jackson2JsonMessageConverter());

-        senderService = Mockito.mock(DefaultAmqpSenderService.class);
+        senderService = Mockito.mock(DefaultAmqpMessageSenderService.class);

        final ArtifactUrlHandler artifactUrlHandlerMock = Mockito.mock(ArtifactUrlHandler.class);
        when(artifactUrlHandlerMock.getUrls(anyObject(), anyObject()))
--- a/hawkbit-dmf/hawkbit-dmf-amqp/src/test/java/org/eclipse/hawkbit/integration/AmqpMessageHandlerServiceIntegrationTest.java
+++ b/hawkbit-dmf/hawkbit-dmf-amqp/src/test/java/org/eclipse/hawkbit/integration/AmqpMessageHandlerServiceIntegrationTest.java
@@ -59,6 +59,17 @@ public class AmqpMessageHandlerServiceIntegrationTest extends AmqpServiceIntegra
    @Autowired
    private AmqpProperties amqpProperties;

+    @Test
+    @Description("Tests DMF PING request and expected reponse.")
+    public void pingDmfInterface() {
+        final Message pingMessage = createPingMessage(CORRELATION_ID, TENANT_EXIST);
+        getDmfClient().send(pingMessage);
+
+        assertPingReplyMessage(CORRELATION_ID);
+
+        Mockito.verifyZeroInteractions(getDeadletterListener());
+    }
+
    @Test
    @Description("Tests register target")
    @ExpectEvents({ @Expect(type = TargetCreatedEvent.class, count = 2),
@@ -450,7 +461,7 @@ public class AmqpMessageHandlerServiceIntegrationTest extends AmqpServiceIntegra
        final String controllerId = TARGET_PREFIX + "receiveDownLoadAndInstallMessageAfterAssignment";

        // setup
-        controllerManagement.findOrRegisterTargetIfItDoesNotexist(controllerId, TEST_URI);
+        createAndSendTarget(controllerId, TENANT_EXIST);
        final DistributionSet distributionSet = testdataFactory.createDistributionSet(UUID.randomUUID().toString());
        assignDistributionSet(distributionSet.getId(), controllerId);

@@ -476,7 +487,7 @@ public class AmqpMessageHandlerServiceIntegrationTest extends AmqpServiceIntegra
        final String controllerId = TARGET_PREFIX + "receiveCancelUpdateMessageAfterAssignmentWasCanceled";

        // Setup
-        controllerManagement.findOrRegisterTargetIfItDoesNotexist(controllerId, TEST_URI);
+        createAndSendTarget(controllerId, TENANT_EXIST);
        final DistributionSet distributionSet = testdataFactory.createDistributionSet(UUID.randomUUID().toString());
        final DistributionSetAssignmentResult distributionSetAssignmentResult = assignDistributionSet(
                distributionSet.getId(), controllerId);
--- a/hawkbit-dmf/hawkbit-dmf-amqp/src/test/java/org/eclipse/hawkbit/integration/AmqpServiceIntegrationTest.java
+++ b/hawkbit-dmf/hawkbit-dmf-amqp/src/test/java/org/eclipse/hawkbit/integration/AmqpServiceIntegrationTest.java
@@ -10,6 +10,7 @@ package org.eclipse.hawkbit.integration;

 import static org.assertj.core.api.Assertions.assertThat;

+import java.nio.charset.StandardCharsets;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
@@ -138,6 +139,22 @@ public abstract class AmqpServiceIntegrationTest extends AbstractAmqpIntegration
        assertThat(headers.get(MessageHeaderKey.TYPE)).isEqualTo(MessageType.THING_DELETED.toString());
    }

+    protected void assertPingReplyMessage(final String correlationId) {
+
+        verifyReplyToListener();
+        final Message replyMessage = replyToListener.getPingResponseMessages().get(correlationId);
+
+        final Map<String, Object> headers = replyMessage.getMessageProperties().getHeaders();
+
+        assertThat(headers.get(MessageHeaderKey.TENANT)).isEqualTo(TENANT_EXIST);
+        assertThat(correlationId)
+                .isEqualTo(new String(replyMessage.getMessageProperties().getCorrelationId(), StandardCharsets.UTF_8));
+        assertThat(headers.get(MessageHeaderKey.TYPE)).isEqualTo(MessageType.PING_RESPONSE.toString());
+        assertThat(Long.valueOf(new String(replyMessage.getBody(), StandardCharsets.UTF_8)))
+                .isLessThanOrEqualTo(System.currentTimeMillis());
+
+    }
+
    protected void assertDownloadAndInstallMessage(final Set<SoftwareModule> dsModules, final String controllerId) {
        final Message replyMessage = assertReplyMessageHeader(EventTopic.DOWNLOAD_AND_INSTALL, controllerId);
        assertAllTargetsCount(1);
@@ -158,6 +175,12 @@ public abstract class AmqpServiceIntegrationTest extends AbstractAmqpIntegration
        getDmfClient().send(message);
    }

+    protected Message createAndSendPingMessage(final String correlationId, final String tenant) {
+        final Message message = createPingMessage(correlationId, tenant);
+        getDmfClient().send(message);
+        return message;
+    }
+
    protected void verifyReplyToListener() {
        createConditionFactory().until(() -> {
            Mockito.verify(replyToListener, Mockito.atLeast(1)).handleMessage(Mockito.any());
@@ -244,6 +267,15 @@ public abstract class AmqpServiceIntegrationTest extends AbstractAmqpIntegration
        return createMessage(null, messageProperties);
    }

+    protected Message createPingMessage(final String correlationId, final String tenant) {
+        final MessageProperties messageProperties = createMessagePropertiesWithTenant(tenant);
+        messageProperties.getHeaders().put(MessageHeaderKey.TYPE, MessageType.PING.toString());
+        messageProperties.setCorrelationId(correlationId.getBytes());
+        messageProperties.setReplyTo(DmfTestConfiguration.REPLY_TO_EXCHANGE);
+
+        return createMessage(null, messageProperties);
+    }
+
    protected MessageProperties createMessagePropertiesWithTenant(final String tenant) {
        final MessageProperties messageProperties = new MessageProperties();
        messageProperties.getHeaders().put(MessageHeaderKey.TENANT, tenant);
--- a/hawkbit-dmf/hawkbit-dmf-amqp/src/test/java/org/eclipse/hawkbit/integration/listener/ReplyToListener.java
+++ b/hawkbit-dmf/hawkbit-dmf-amqp/src/test/java/org/eclipse/hawkbit/integration/listener/ReplyToListener.java
@@ -10,6 +10,8 @@ package org.eclipse.hawkbit.integration.listener;

 import static org.junit.Assert.fail;

+import java.nio.charset.StandardCharsets;
+import java.util.EnumMap;
 import java.util.HashMap;
 import java.util.Map;

@@ -25,8 +27,9 @@ public class ReplyToListener implements TestRabbitListener {
    public static final String LISTENER_ID = "replyto";
    public static final String REPLY_TO_QUEUE = "reply_queue";

-    private final Map<EventTopic, Message> eventTopicMessages = new HashMap<>();
+    private final Map<EventTopic, Message> eventTopicMessages = new EnumMap<>(EventTopic.class);
    private final Map<String, Message> deleteMessages = new HashMap<>();
+    private final Map<String, Message> pingResponseMessages = new HashMap<>();

    @Override
    @RabbitListener(id = LISTENER_ID, queues = REPLY_TO_QUEUE)
@@ -49,6 +52,13 @@ public class ReplyToListener implements TestRabbitListener {
            return;
        }

+        if (messageType == MessageType.PING_RESPONSE) {
+            final String correlationId = new String(message.getMessageProperties().getCorrelationId(),
+                    StandardCharsets.UTF_8);
+            pingResponseMessages.put(correlationId, message);
+            return;
+        }
+
        // if message type is not EVENT or THING_DELETED something unexpected
        // happened
        fail("Unexpected message type");
@@ -63,4 +73,8 @@ public class ReplyToListener implements TestRabbitListener {
        return deleteMessages;
    }

+    public Map<String, Message> getPingResponseMessages() {
+        return pingResponseMessages;
+    }
+
 }
--- a/hawkbit-dmf/hawkbit-dmf-api/src/main/java/org/eclipse/hawkbit/dmf/amqp/api/MessageType.java
+++ b/hawkbit-dmf/hawkbit-dmf-api/src/main/java/org/eclipse/hawkbit/dmf/amqp/api/MessageType.java
@@ -15,7 +15,7 @@ package org.eclipse.hawkbit.dmf.amqp.api;
 public enum MessageType {

    /**
-     * The event type.
+     * The event type related to interaction with a thing.
     */
    EVENT,

@@ -29,4 +29,14 @@ public enum MessageType {
     */
    THING_DELETED,

+    /**
+     * DMF receiver health check type.
+     */
+    PING,
+
+    /**
+     * DMF receiver health check reponse type.
+     */
+    PING_RESPONSE;
+
 }
--- a/hawkbit-dmf/hawkbit-dmf-rabbitmq-test/src/main/java/org/eclipse/hawkbit/rabbitmq/test/AbstractAmqpIntegrationTest.java
+++ b/hawkbit-dmf/hawkbit-dmf-rabbitmq-test/src/main/java/org/eclipse/hawkbit/rabbitmq/test/AbstractAmqpIntegrationTest.java
@@ -8,12 +8,10 @@
 */
 package org.eclipse.hawkbit.rabbitmq.test;

-import java.net.URI;
 import java.util.concurrent.TimeUnit;

 import org.eclipse.hawkbit.repository.jpa.RepositoryApplicationConfiguration;
 import org.eclipse.hawkbit.repository.test.util.AbstractIntegrationTest;
-import org.eclipse.hawkbit.util.IpUtil;
 import org.junit.Before;
 import org.junit.Rule;
 import org.springframework.amqp.core.Message;
@@ -37,8 +35,6 @@ import com.jayway.awaitility.core.ConditionFactory;
@DirtiesContext(classMode = ClassMode.AFTER_CLASS)
 public abstract class AbstractAmqpIntegrationTest extends AbstractIntegrationTest {

-    protected static final URI TEST_URI = IpUtil.createAmqpUri("testHost", "testExcange");
-
    @Rule
    @Autowired
    public BrokerRunning brokerRunning;