diff --git a/plugins/event-bus/rabbitmq/src/main/java/org/apache/cloudstack/mom/rabbitmq/RabbitMQEventBus.java b/plugins/event-bus/rabbitmq/src/main/java/org/apache/cloudstack/mom/rabbitmq/RabbitMQEventBus.java index b98147b5f031..d7742ee32266 100644 --- a/plugins/event-bus/rabbitmq/src/main/java/org/apache/cloudstack/mom/rabbitmq/RabbitMQEventBus.java +++ b/plugins/event-bus/rabbitmq/src/main/java/org/apache/cloudstack/mom/rabbitmq/RabbitMQEventBus.java @@ -23,6 +23,7 @@ import java.net.ConnectException; import java.security.KeyManagementException; import java.security.NoSuchAlgorithmException; +import java.util.ArrayDeque; import java.util.Map; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; @@ -85,11 +86,23 @@ public static void setUseSsl(String useSsl) { private static Integer retryInterval; + // max number of events kept in memory while the AMQP server is unreachable, 0 disables buffering + private static Integer bufferSize = 1000; + + // time in milliseconds after which a buffered event is dropped if it could not be sent + private static Integer bufferExpiry = 300000; + + // events (routing key, description, time of buffering) waiting to be sent, oldest first + private static final ArrayDeque> s_buffer = new ArrayDeque<>(); + + // events are buffered only after the AMQP server was reachable at least once + private static volatile boolean s_connectedOnce = false; + // hashmap to book keep the registered subscribers private static ConcurrentHashMap> s_subscribers; // connection to AMQP server, - private static Connection s_connection = null; + private static volatile Connection s_connection = null; // AMQP server should consider messages acknowledged once delivered if _autoAck is true private static boolean s_autoAck = true; @@ -177,6 +190,14 @@ public static void setRetryInterval(Integer retryInterval) { RabbitMQEventBus.retryInterval = retryInterval; } + public static void setBufferSize(Integer bufferSize) { + RabbitMQEventBus.bufferSize = bufferSize; + } + + public static void setBufferExpiry(Integer bufferExpiry) { + RabbitMQEventBus.bufferExpiry = bufferExpiry; + } + /** Call to subscribe to interested set of events * * @param topic defines category and type of the events being subscribed to @@ -273,20 +294,106 @@ public void publish(Event event) throws EventBusException { String routingKey = createRoutingKey(event); String eventDescription = event.getDescription(); + // keep the order: nothing is sent directly while older events are waiting or the connection is down + if (bufferEvent(routingKey, eventDescription, false)) { + return; + } + try { - Connection connection = getConnection(); - Channel channel = createChannel(connection); - createExchange(channel, amqpExchangeName); - publishEventToExchange(channel, amqpExchangeName, routingKey, eventDescription); - channel.close(); + sendEvent(getConnection(), routingKey, eventDescription); } catch (AlreadyClosedException e) { closeConnection(); + if (bufferEvent(routingKey, eventDescription, true)) { + return; + } throw new EventBusException("Failed to publish event to message broker as connection to AMQP broker in lost"); + } catch (IOException e) { + if (bufferEvent(routingKey, eventDescription, true)) { + return; + } + throw new EventBusException("Failed to publish event to message broker due to " + e.getMessage()); } catch (Exception e) { throw new EventBusException("Failed to publish event to message broker due to " + e.getMessage()); } } + private void sendEvent(Connection connection, String routingKey, String eventDescription) throws Exception { + Channel channel = createChannel(connection); + createExchange(channel, amqpExchangeName); + publishEventToExchange(channel, amqpExchangeName, routingKey, eventDescription); + channel.close(); + } + + /** adds the event to the buffer, dropping the oldest one if the buffer is full + * + * @param sendFailed false if the event was not tried yet and has to be buffered only to keep the order + * @return true if the event was buffered, false if the caller has to deal with it + */ + private boolean bufferEvent(String routingKey, String eventDescription, boolean sendFailed) { + if (!s_connectedOnce || bufferSize <= 0) { + return false; + } + synchronized (s_buffer) { + if (!sendFailed && s_buffer.isEmpty() && s_connection != null) { + return false; + } + if (s_buffer.isEmpty()) { + logger.warn("Unable to send events to AMQP server, buffering them until they can be sent"); + } + if (s_buffer.size() >= bufferSize) { + logger.warn("Event buffer is full, dropping the oldest event " + s_buffer.poll().first()); + } + s_buffer.add(new Ternary<>(routingKey, eventDescription, System.currentTimeMillis())); + } + return true; + } + + // drops the expired events and, if connected, sends the buffered ones in the order they were added + private void flushBuffer() { + int sent = 0; + while (true) { + Ternary bufferedEvent; + synchronized (s_buffer) { + bufferedEvent = s_buffer.peek(); + } + if (bufferedEvent == null) { + break; + } + + boolean expired = System.currentTimeMillis() - bufferedEvent.third() > bufferExpiry; + Connection connection = s_connection; + if (!expired && connection == null) { + break; + } + + if (expired) { + logger.warn("Dropping expired event " + bufferedEvent.first() + " as it could not be sent to AMQP server in time"); + } else { + try { + sendEvent(connection, bufferedEvent.first(), bufferedEvent.second()); + sent++; + } catch (AlreadyClosedException e) { + closeConnection(); + break; + } catch (IOException e) { + break; // retry with the next run + } catch (Exception e) { + logger.warn("Dropping buffered event " + bufferedEvent.first() + " due to " + e.getMessage()); + } + } + + // the event stays in the buffer while being sent, so that newer events queue up behind it + synchronized (s_buffer) { + if (s_buffer.peek() == bufferedEvent) { + s_buffer.poll(); + } + } + } + if (sent > 0) { + logger.info("Sent " + sent + " buffered events to AMQP server"); + } + } + /** creates a routing key from the event details. * created routing key will be used while publishing the message to exchange on AMQP server */ @@ -392,6 +499,7 @@ private synchronized Connection createConnection() throws KeyManagementException connection.addShutdownListener(disconnectHandler); connection.addBlockedListener(blockedConnectionHandler); s_connection = connection; + s_connectedOnce = true; return s_connection; } @@ -508,6 +616,7 @@ public synchronized boolean stop() { } closeConnection(); + executorService.shutdownNow(); return true; } @@ -540,35 +649,38 @@ public void shutdownCompleted(ShutdownSignalException shutdownSignalException) { } abortConnection(); // disconnected to AMQP server, so abort the connection and channels + // the ReconnectionTask started with the event bus takes care of the re-connect logger.warn("Connection has been shutdown by AMQP server. Attempting to reconnect."); - - // initiate re-connect process - ReconnectionTask reconnect = new ReconnectionTask(); - executorService.submit(reconnect); } } } - // retry logic to connect back to AMQP server after loss of connection + // runs for the lifetime of the event bus: connects back to AMQP server after any loss of connection + // and sends the events buffered in the meantime private class ReconnectionTask extends ManagedContextRunnable { - boolean connected = false; Connection connection = null; @Override protected void runInContext() { - while (!connected) { + while (true) { try { Thread.sleep(retryInterval); } catch (InterruptedException ie) { - // ignore timer interrupts + return; // event bus is being stopped } try { + flushBuffer(); try { - connection = createConnection(); - connected = true; + // check and connect under the same lock as getConnection(), so a connection is never replaced + synchronized (RabbitMQEventBus.this) { + if (s_connection != null) { + continue; + } + connection = createConnection(); + } } catch (IOException ie) { continue; // can't establish connection to AMQP server yet, so continue } @@ -617,11 +729,11 @@ public void handleDelivery(String queueName, Envelope envelope, AMQP.BasicProper subscriberDetails.second(channel); s_subscribers.put(subscriberId, subscriberDetails); } + flushBuffer(); } catch (Exception e) { logger.warn("Failed to recreate queues and binding for the subscribers due to " + e.getMessage()); } } - return; } } } diff --git a/plugins/event-bus/rabbitmq/src/test/java/org/apache/cloudstack/mom/rabbitmq/RabbitMQEventBusTest.java b/plugins/event-bus/rabbitmq/src/test/java/org/apache/cloudstack/mom/rabbitmq/RabbitMQEventBusTest.java new file mode 100644 index 000000000000..281e461575d8 --- /dev/null +++ b/plugins/event-bus/rabbitmq/src/test/java/org/apache/cloudstack/mom/rabbitmq/RabbitMQEventBusTest.java @@ -0,0 +1,167 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package org.apache.cloudstack.mom.rabbitmq; + +import java.io.IOException; +import java.lang.reflect.Field; +import java.lang.reflect.Method; +import java.util.ArrayDeque; + +import org.apache.cloudstack.framework.events.Event; +import org.apache.cloudstack.framework.events.EventBusException; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.mockito.ArgumentMatchers; +import org.mockito.InOrder; +import org.mockito.Mockito; + +import com.rabbitmq.client.Channel; +import com.rabbitmq.client.Connection; + +public class RabbitMQEventBusTest { + + private RabbitMQEventBus eventBus; + private Connection connection; + private Channel channel; + + @Before + public void setUp() throws Exception { + eventBus = new RabbitMQEventBus(); + RabbitMQEventBus.setExchange("cloudstack-events"); + RabbitMQEventBus.setBufferSize(1000); + RabbitMQEventBus.setBufferExpiry(300000); + + channel = Mockito.mock(Channel.class); + connection = Mockito.mock(Connection.class); + Mockito.when(connection.createChannel()).thenReturn(channel); + + ((ArrayDeque)getField("s_buffer").get(null)).clear(); + setConnection(null); + getField("s_connectedOnce").set(null, true); + } + + @After + public void tearDown() throws Exception { + setConnection(null); + getField("s_connectedOnce").set(null, false); + } + + private Field getField(String name) throws Exception { + Field field = RabbitMQEventBus.class.getDeclaredField(name); + field.setAccessible(true); + return field; + } + + private void setConnection(Connection connection) throws Exception { + getField("s_connection").set(null, connection); + } + + private void flushBuffer() throws Exception { + Method method = RabbitMQEventBus.class.getDeclaredMethod("flushBuffer"); + method.setAccessible(true); + method.invoke(eventBus); + } + + private Event event(String description) { + Event event = new Event("server", "ActionEvent", "VM.START", "VirtualMachine", "uuid"); + event.setDescription(description); + return event; + } + + private void verifyPublished(InOrder inOrder, String description) throws IOException { + inOrder.verify(channel).basicPublish(ArgumentMatchers.anyString(), ArgumentMatchers.anyString(), ArgumentMatchers.any(), + ArgumentMatchers.eq(description.getBytes())); + } + + private void verifyPublishedCount(int count) throws IOException { + Mockito.verify(channel, Mockito.times(count)).basicPublish(ArgumentMatchers.anyString(), ArgumentMatchers.anyString(), + ArgumentMatchers.any(), ArgumentMatchers.any(byte[].class)); + } + + @Test(expected = EventBusException.class) + public void publishFailsWhenBrokerWasNeverReachable() throws Exception { + getField("s_connectedOnce").set(null, false); + RabbitMQEventBus.setServer("localhost"); + RabbitMQEventBus.setPort(1); + RabbitMQEventBus.setUsername("guest"); + RabbitMQEventBus.setPassword("guest"); + + eventBus.publish(event("1")); + } + + @Test + public void eventsPublishedWhileDisconnectedAreSentInOrderAfterReconnect() throws Exception { + eventBus.publish(event("1")); + eventBus.publish(event("2")); + Mockito.verifyNoInteractions(channel); + + setConnection(connection); + eventBus.publish(event("3")); // must not overtake the buffered ones + Mockito.verifyNoInteractions(channel); + + flushBuffer(); + InOrder inOrder = Mockito.inOrder(channel); + verifyPublished(inOrder, "1"); + verifyPublished(inOrder, "2"); + verifyPublished(inOrder, "3"); + + eventBus.publish(event("4")); // buffer is empty again, sent directly + verifyPublished(inOrder, "4"); + verifyPublishedCount(4); + } + + @Test + public void eventIsBufferedWhenSendingFails() throws Exception { + setConnection(connection); + Mockito.doThrow(new IOException("broken pipe")).doNothing().when(channel) + .basicPublish(ArgumentMatchers.anyString(), ArgumentMatchers.anyString(), ArgumentMatchers.any(), ArgumentMatchers.any(byte[].class)); + + eventBus.publish(event("1")); + verifyPublishedCount(1); + + flushBuffer(); + verifyPublishedCount(2); + } + + @Test + public void oldestEventIsDroppedWhenBufferIsFull() throws Exception { + RabbitMQEventBus.setBufferSize(2); + eventBus.publish(event("1")); + eventBus.publish(event("2")); + eventBus.publish(event("3")); + + setConnection(connection); + flushBuffer(); + InOrder inOrder = Mockito.inOrder(channel); + verifyPublished(inOrder, "2"); + verifyPublished(inOrder, "3"); + verifyPublishedCount(2); + } + + @Test + public void expiredEventsAreDropped() throws Exception { + RabbitMQEventBus.setBufferExpiry(-1); + eventBus.publish(event("1")); + + flushBuffer(); // expires also without a connection + setConnection(connection); + flushBuffer(); + Mockito.verifyNoInteractions(channel); + } +}