ActiveMQ in a master-slave setup

In this guide, we will go over the process of configuring two ActiveMQ instances in a master-slave setup. This type of set-up is typically done to improve performance (by distributing loads) and high-availability by hooking a failover instance, which are both important when running TORO Integrate in a production environment.

TORO Integrate uses the configured broker when indexing data with Tracker and Monitor; in fact, a message is created, published, received, and resolved per transaction. By enhancing ActiveMQ's performance, it should be safe to assume that TORO Integrate's overall performance would also improve, especially if your organization uses search indices a lot.

Prerequisites

To implement this setup, you must have two instances of ActiveMQ running; one will be the master instance, and the other, the slave. It is recommended to host these instances in separate machines, but if this arrangement is not an option, then it is still possible to deploy them on the same machine. An NFS server will also be required in order to share data between these instances.

Configuration + authentication and authorization

Before diving onto this guide, it is recommended to read the previous pages on general remote ActiveMQ instance configuration, as well as setting up remote ActiveMQ instances with authentication and authorization.

Procedures

Configuring the NFS server & NFS clients

A network file system (NFS) is a setup that involves a server and a client. An NFS enables you to mount directories to share and access files directly within a network.

Below are the steps required in order to configure NFS servers and NFS clients for CentOS 7. Of course, it is possible to use a different operating system as long as you are able to correctly configure the NFS server its clients.

Configuring the NFS server

1. Determine which instance or machine will act as the NFS server.

2. Install NFS packages using yum:

   yum install nfs-utils
   

   nfs-utils package

   The nfs-utils package contains the tools for both the server and client which are necessary for the kernel's NFS abilities.

3. Determine and create the directory which will be shared for the NFS setup.

   TORO recommends creating and sharing the /datastore/activemq-data directory, but it is also possible to choose a different directory which will satisfy your preferences and/or needs better.

   mkdir -p /datastore/activemq-data
   

4. Set the NFS directory's permissions.

   Once you have chosen and have created the NFS directory in the NFS server, you can now set the permissions of the folder using the chmod and chown commands.

   chmod -R 7777 /datastore/activemq-data && chown $user:$group /datastore/activemq-data
   

5. Start and enable services.

   systemctl enable rpcbind
   systemctl enable nfs-server
   systemctl enable nfs-lock
   systemctl enable nfs-idmap
   systemctl start rpcbind
   systemctl start nfs-server
   systemctl start nfs-lock
   systemctl start nfs-idmap
   

6. Share the NFS directory to enable client access.

   1. First, open the /etc/exports file and edit it with your preferred text editor.

      vi /etc/exports
      

   2. Next, input the following:

      /datastore/activemq-data    <nfs-client-ip>(rw,sync,no_root_squash,no_all_squash)
      

   A connection must be establishable

   The server and client must be able to ping each other.

7. Start the NFS service using the command:

   systemctl restart nfs-server
   

8. Configure the firewall for CentOS 7.

   firewall-cmd --permanent --zone=public --add-service=nfs
   firewall-cmd --permanent --zone=public --add-service=mountd
   firewall-cmd --permanent --zone=public --add-service=rpc-bind   
   firewall-cmd --reload
   

Configuring the NFS client

1. Install NFS packages using yum:

   yum install nfs-utils
   

2. Create the NFS directory mount point(s).

   mkdir /datastore/activemq-data
   

3. Mount the directory.

   mount -t nfs <nfs-server-ip>:/datastore/activemq-data /datastore/activemq-data
   

4. Verify the NFS mount.

   df -kh
   

   After executing this command, you should see a list of file systems and belonging in that list should be the NFS configured earlier.

Setting up a permanent NFS mount

By default, you will have to remount all your NFS directories after every reboot. To make sure it is available upon reboot, you may follow the steps below:

1. Open the fstab file using the text editor of your choice.

   vi /etc/fstab
   

2. Edit the file and add the following lines to the file to have the mount points automatically configured after reboot:

   <nfs-server-ip>:/datastore/activemq-data    /datastore/activemq-data    nfs   defaults 0 0
   

Configuring TORO Integrate

To tell TORO Integrate that there will be two instances of ActiveMQ, edit the jms.url application property in your application .properties file. This property contains the list of broker connection URLs, separated by commas if there is more than one broker, which is the case for our setup. For example:

jms.url=failover:tcp://<activemq1-ip-address>:61616,tcp://<activemq2-ip-address>:61616?CloseAsync=false

Embedded ActiveMQ

You can select an embedded ActiveMQ instance as the fail-over option by doing something like:

jms.url=failover:tcp://<activemq1-ip-address>:61616,tcp://<activemq2-ip-address>:61616,tcp://0.0.0.0:61616?CloseAsync=false

... wherein tcp://0.0.0.0:61616 is the URL for the embedded ActiveMQ instance.

Duplicate JMS client IDs

No two instances should be using the same jms.clientId or the connection will fail.

ActiveMQ configuration

Remote ActiveMQ instances can be configured via XML or via the command line. TORO and ActiveMQ both recommend configuring your instance via XML, as this type of configuration allows for advanced configuration options and flexibility. This is also the type of configuration the following sections will be referencing, as we discuss which components of the ActiveMQ instance should be configured for a master-slave setup.

Configuring the broker bean element

The broker bean element is used to configure ActiveMQ instance-wide properties. TORO recommends the following setup for your remote ActiveMQ instance:

<broker xmlns="http://activemq.apache.org/schema/core"
    brokerName="activemq-broker1"
    dataDirectory="${activemq.data}"
    useJmx="true" advisorySupport="true"
    persistent="true"
    deleteAllMessagesOnStartup="false"
    useShutdownHook="false"
    schedulerSupport="true"
    start="false">
<!-- ... -->
</broker>

In fact, the configuration above is used for remote ActiveMQ instances deployed for TORO Integrate instances on TORO Cloud.

Configuring destination policies

According to the ActiveMQ documentation on per destination policies:

"Multiple different policies can be applied per destination (queue or topic), or using wildcard notation to apply to a hierarchy of queues or topics, making it possible, therefore, to configure how different regions of the JMS destination space are handled."

When using XML configuration, destination policies are defined using the <destinationPolicy> tag, which is under the <broker> XML element. For example, the following configuration sets the behavior of topics:

<destinationPolicy>
<policyMap>
    <policyEntries>
        <policyEntry topic=">"
            producerFlowControl="false"
            gcInactiveDestinations="true"
            inactiveTimeoutBeforeGC="5000"
            memoryLimit="5mb"
            expireMessagesPeriod="0"
            advisoryWhenFull="true"
            maxDestinations="200" >
            <pendingMessageLimitStrategy>
                <constantPendingMessageLimitStrategy limit="20"/> <!-- 20 seconds worth -->
            </pendingMessageLimitStrategy>
            <dispatchPolicy>
                <strictOrderDispatchPolicy />
            </dispatchPolicy>
            <subscriptionRecoveryPolicy>
                <timedSubscriptionRecoveryPolicy recoverDuration="60000" /> <!-- 1 minute worth -->
            </subscriptionRecoveryPolicy>
        </policyEntry>
    </policyEntries>
</policyMap>
</destinationPolicy>

Changes

Changes are immediately applied once saved. Check your logs for confirmation.

Configuring the management context

This is not necessary but if you'd like to be able to control the behavior of the broker via JMX MBeans, then you must configure the ActiveMQ instance's management context. Doing so will allow you to connect, monitor, and control the instance remotely via JConsole.

<managementContext>
    <managementContext createConnector="true" rmiServerPort="1098" connectorPort="1099"/>
</managementContext>

With this, you may now connect to the service via JConsole using the URL service:jmx:rmi:///jndi/rmi://localhost:1099/jmxrmi.

Configuring the KahaDB directory

According to the ActiveMQ website, "KahaDB is a file-based persistence database that is local to the message broker that is using it." We must set KahaDB's directory to /datastore/activemq-data created earlier, like so:

<persistenceAdapter>
    <kahaDB directory="/datastore/activemq-data" indexWriteBatchSize="1000" enableIndexWriteAsync="true"/>
</persistenceAdapter>

Configuring transport connectors

Since our master-slave setup features a fail-over, we must use the openwire transport connector. This transport connector allows brokers to talk to each other over the network. The configuration below shows the transport connector for our master broker instance:

<transportConnectors>
    <transportConnector name="openwire" uri="tcp://0.0.0.0:61616?jms.useAsyncSend=true&amp;randomize=false&amp;initialReconnectDelay=100&amp;maximumConnections=1000&amp;nested.wireFormat.maxInactivityDuration=1000" updateClusterClients="true"/>
    <transportConnector name="amqp" uri="amqp://0.0.0.0:5672?jms.useAsyncSend=true&amp;randomize=false&amp;initialReconnectDelay=100&amp;maximumConnections=1000&amp;nested.wireFormat.maxInactivityDuration=1000" updateClusterClients="true"/>
    <!-- Other transport connectors... -->
</transportConnectors>

Default ports

• 61616 is the default TCP port
• 8161 is the default web console port
• 5672 is the default AMQP port

Summary

After successfully following through this guide, you should now have an ActiveMQ master-slave setup for TORO Integrate. Your final activemq.xml file should look roughly something like:

<!--
    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.
-->
<!-- START SNIPPET: example -->
<beans xmlns="http://www.springframework.org/schema/beans" xmlns:amq="http://activemq.apache.org/schema/core" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://www.springframework.org/schema/beans                             http://www.springframework.org/schema/beans/spring-beans.xsd                            http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core.xsd">
    <!-- Allows us to use system properties as variables in this configuration file -->
    <bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">
        <property name="locations">
            <value>file:${activemq.conf}/credentials.properties</value>
        </property>
    </bean>
    <!-- Allows access to the server logs -->
    <bean id="logQuery" class="io.fabric8.insight.log.log4j.Log4jLogQuery" lazy-init="false" scope="singleton" init-method="start" destroy-method="stop" />
    <!-- The <broker> element is used to configure the ActiveMQ broker. -->
    <broker xmlns="http://activemq.apache.org/schema/core" brokerName="activemq-broker1" dataDirectory="${activemq.data}" useJmx="true" advisorySupport="true" persistent="true" deleteAllMessagesOnStartup="false" useShutdownHook="false" schedulerSupport="true" start="false">
        <!--
            For better performances use VM cursor and small memory limit.
            For more information, see:

            http://activemq.apache.org/message-cursors.html

            Also, if your producer is "hanging", it's probably due to producer flow control.
            For more information, see:
            http://activemq.apache.org/producer-flow-control.html
        -->
        <destinationPolicy>
            <policyMap>
                <policyEntries>
                    <policyEntry topic="&gt;" producerFlowControl="false" gcInactiveDestinations="true" inactiveTimeoutBeforeGC="5000" memoryLimit="5mb" expireMessagesPeriod="0" advisoryWhenFull="true" maxDestinations="200">
                        <!--
                            The constantPendingMessageLimitStrategy is used to prevent
                            slow topic consumers to block producers and affect other consumers
                            by limiting the number of messages that are retained
                            For more information, see:

                            http://activemq.apache.org/slow-consumer-handling.html
                        -->
                        <pendingMessageLimitStrategy>
                            <constantPendingMessageLimitStrategy limit="20" /> <!-- 20 seconds worth -->
                        </pendingMessageLimitStrategy>
                        <dispatchPolicy>
                            <strictOrderDispatchPolicy />
                        </dispatchPolicy>
                        <subscriptionRecoveryPolicy>
                            <timedSubscriptionRecoveryPolicy recoverDuration="60000" /> <!--  1 minute worth -->
                        </subscriptionRecoveryPolicy>
                    </policyEntry>
                </policyEntries>
            </policyMap>
        </destinationPolicy>
        <!--
            The managementContext is used to configure how ActiveMQ is exposed in
            JMX. By default, ActiveMQ uses the MBean server that is started by
            the JVM. For more information, see:

            http://activemq.apache.org/jmx.html
        -->
        <managementContext>
            <managementContext createConnector="true" rmiServerPort="1098" connectorPort="1099" />
        </managementContext>
        <plugins>
            <jaasAuthenticationPlugin configuration="activemq-domain" />
            <runtimeConfigurationPlugin checkPeriod="1000" />
            <authorizationPlugin>
                <map>
                    <authorizationMap>
                        <authorizationEntries>
                            <!-- To make security on every topic, the application needs be to configured to have their unique identifier on their topic name -->
                            <authorizationEntry topic="jmsPrefix.statistics.&gt;" read="admins" write="admins" admin="admins" />
                            <authorizationEntry topic="jmsPrefix.io.toro.integrate.&gt;" read="admins" write="admins" admin="admins" />
                            <authorizationEntry queue="jmsPrefix.io.toro.integrate.&gt;" read="admins" write="admins" admin="admins" />
                            <authorizationEntry topic="ActiveMQ.Advisory.&gt;" read="admins" write="admins" admin="admins" />
                        </authorizationEntries>
                    </authorizationMap>
                </map>
            </authorizationPlugin>
        </plugins>
        <!--
            Configure message persistence for the broker. The default persistence
            mechanism is the KahaDB store (identified by the kahaDB tag).
            For more information, see:

            http://activemq.apache.org/persistence.html
        -->
        <persistenceAdapter>
            <kahaDB directory="/datastore/activemq-data" indexWriteBatchSize="1000" enableIndexWriteAsync="true" />
        </persistenceAdapter>
        <!--
            The systemUsage controls the maximum amount of space the broker will
            use before disabling caching and/or slowing down producers. For more information, see:
            http://activemq.apache.org/producer-flow-control.html
            If using ActiveMQ embedded - the following limits could safely be used:

        <systemUsage><systemUsage><memoryUsage><memoryUsage limit="20 mb"/></memoryUsage><storeUsage><storeUsage limit="1 gb"/></storeUsage><tempUsage><tempUsage limit="100 mb"/></tempUsage></systemUsage></systemUsage>
        -->
        <systemUsage>
            <systemUsage>
                <memoryUsage>
                    <memoryUsage percentOfJvmHeap="70" />
                </memoryUsage>
                <storeUsage>
                    <storeUsage limit="100 gb" />
                </storeUsage>
                <tempUsage>
                    <tempUsage limit="50 gb" />
                </tempUsage>
            </systemUsage>
        </systemUsage>
        <!--
            The transport connectors expose ActiveMQ over a given protocol to
            clients and other brokers. For more information, see:

            http://activemq.apache.org/configuring-transports.html
        -->
        <transportConnectors>
            <!-- DOS protection, limit concurrent connections to 1000 and frame size to 100MB -->
            <transportConnector name="openwire" uri="tcp://0.0.0.0:61616?jms.useAsyncSend=true&amp;randomize=false&amp;initialReconnectDelay=100&amp;maximumConnections=1000&amp;nested.wireFormat.maxInactivityDuration=1000" updateClusterClients="true" />
            <transportConnector name="amqp" uri="amqp://0.0.0.0:5672?jms.useAsyncSend=true&amp;randomize=false&amp;initialReconnectDelay=100&amp;maximumConnections=1000&amp;nested.wireFormat.maxInactivityDuration=1000" updateClusterClients="true" />
            <transportConnector name="stomp" uri="stomp://0.0.0.0:61613?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600" />
            <transportConnector name="mqtt" uri="mqtt://0.0.0.0:1883?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600" />
            <transportConnector name="ws" uri="ws://0.0.0.0:61614?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600" />
        </transportConnectors>
        <!-- destroy the spring context on shutdown to stop jetty -->
        <shutdownHooks>
            <bean xmlns="http://www.springframework.org/schema/beans" class="org.apache.activemq.hooks.SpringContextHook" />
        </shutdownHooks>
    </broker>
    <!--
        Enable web consoles, REST and Ajax APIs and demos
        The web console requires by default login; you can disable this in the jetty.xml file
        Take a look at ${ACTIVEMQ_HOME}/conf/jetty.xml for more details
    -->
    <import resource="jetty.xml" />
</beans>
<!-- END SNIPPET: example -->

Jetty settings

The activemq.xml file will often either contain Jetty settings or import them from another file. For example:

<import resource="jetty.xml"/>

MCollective doesn't use this. If you're not using it to manage ActiveMQ, leaving it enabled may be a security risk.