WSO2 MI: Envío y Recepción con ActiveMQ

Buenas, hoy veremos un ejemplo completo de como interacturar desde WSO2 con una cola ActiveMQ. Mostraremos como configurar el Micro Integrator, MI, escribir y leer de la cola y consultar cuantos registros tiene. 

Para el ejemplo utilizaremos las siguientes versiones:

• Micro Integrator 4.0.0
• Active MQ 5.8.14. 

    1. Docker Compose

Por un lado debemos incluir las librerías y dependencias de ActiveMQ en el docker compose base que tenemos:

• activemq-broker, activemq-client y activemq-openwire-legacy 5.14.0
• geronimo-j2ee-management_1_1_spec-1.0.1 y geronimo-jms_1.1_spec-1.1.1
• hawtbu-1.11

He incluir la instancia de ActiveMQ:

  activemq:
    image: rmohr/activemq:5.14.5
    mem_limit: 1G
    hostname: activemq
    container_name: activemq
    ports:
      - 8161:8161   #UI
      - 61616:61616 #JMS
    networks:
      wso2-net:
        ipv4_address: 172.16.238.20

Si no configuramos nada más, cuando intentemos conectarnos con ActiveMQ, recibiremos el siguiente error:

Unexpected error during sending message out org.apache.axis2.AxisFault: 
The system cannot infer the transport information from the jms:/.....

Por tanto deberemos modificar el fichero de configuración deployment.toml para añadir:

[[transport.jms.sender]]
name = "myQueueSender"
parameter.initial_naming_factory = "org.apache.activemq.jndi.ActiveMQInitialContextFactory"
parameter.provider_url = "tcp://activemq:61616"
parameter.connection_factory_name = "QueueConnectionFactory"
parameter.connection_factory_type = "queue"
parameter.cache_level = "producer"

Y para poder leer de una cola necesitamos configurar un JMS Listener también en el fichero de configuración deployment.toml

[[transport.jms.listener]]
name = "myQueueConnectionFactory"
parameter.initial_naming_factory = "org.apache.activemq.jndi.ActiveMQInitialContextFactory"
parameter.provider_url = "failover:tcp://activemq:61616"
parameter.connection_factory_name = "QueueConnectionFactory"
parameter.connection_factory_type = "queue"
parameter.cache_level = "consumer"

[[transport.jms.listener]]
name = "default"
parameter.initial_naming_factory = "org.apache.activemq.jndi.ActiveMQInitialContextFactory"
parameter.provider_url = "failover:tcp://activemq:61616"
parameter.connection_factory_name = "QueueConnectionFactory"
parameter.connection_factory_type = "queue"

En esta configuración es muy importante indicar la configuración de failover. Puesto que de lo contrario si se cayese el nodo de ActiveMQ, no se podría volver a conectar con el mismo a menos que se reiniciase el MI. 

    2. Envío de mensajes

Ahora procederemos a crear el primer recurso de una API, con el cual enviaremos información al activeMQ. Pero previamente transformaremos el mensaje, simplemente por añadir una pequeña lógica al ejemplo.

<resource methods="POST" uri-template="/activeMQ/">
    <inSequence>
        <payloadFactory media-type="xml">
            <format>
                <m:getQuote xmlns:m="http://services.samples">
                    <m:request>
                        <m:book>
                            <m:isbn>$1</m:isbn>
                            <m:release>$2</m:release>
                        </m:book>
                    </m:request>
                </m:getQuote>
            </format>
            <args>
                <arg expression="$.isbn" evaluator="json"/>
                <arg expression="$.year" evaluator="json"/>
            </args>
        </payloadFactory>
        <property name="OUT_ONLY" value="true"/>
        <call>
        <endpoint>
            <address uri="jms:/myqueue?transport.jms.ConnectionFactoryJNDIName=QueueConnectionFactory

                   &java.naming.provider.url=failover:(tcp://activemq:61616)

                   &java.naming.factory.initial=org.apache.activemq.jndi.ActiveMQInitialContextFactory

                   &transport.jms.DestinationType=queue&redeliveryPolicy.redeliveryDelay=30000

                   &maximumRedeliveries=10" />
        </endpoint>
        </call>
        <respond />
    </inSequence>
    <outSequence/>
    <faultSequence/>
</resource>

En este código hay dos apartados interesantes:

• La configuración de la propiedad OUT_ONLY que nos permite indicarle al MI que no se quede a la espera de una respuesta por parte del ActiveMQ. 
• La configuración de la invocación al ActiveMQ, donde aparte de los valores básicos para la configuración,  podemos incluir la política de reintentos a través de las propiedades de redelivery o aumentar la resiliencia con la configuración de failover, donde podemos incluir varios nodos de ActiveMQ.

Si hubiésemos configurado con valores más específicos, no necesitaríamos un address endpoint con tanto detalle. Solo necesitaríamos indicar la cola y el transport sender a utilizar. 

<address uri="jms:/myqueue?transport.jms.ConnectionFactory=myCustomSender" />

    3. Lectura de mensajes en la cola

Desde la versión 5.8 ActiveMQ incluye una API a través de la cual podemos enviar y consultar mensajes. Además con la ayuda de Jolokia pone a disposición de los usuarios una API de gestión. Esta  API será la que utilicemos para ver cuantos mensajes hay disponibles en la cola.   

La lógica de nuestro recurso será sencillo: 

    - Creamos una cabecera de autorización básica para la invocación de la API de activeMQ.

    - En la llamada a la API indicamos el broker y la cola de la cual deseamos los datos. Los cuales obtenemos desde nuestro propio recurso.

    - Leemos la respuesta y en caso de ser correcta en base al valor del campo status. (a tener en cuenta que siempre devuelve HTTP Status Code a 200 y Content-Type a text/plain). Leemos el campo value y lo devolvemos.  

<resource methods="GET" uri-template="/activeMQ/{broker}/{queue}/count">
    <inSequence>
        <property name="Authorization" expression="fn:concat('Basic ', base64Encode('admin:admin'))" scope="transport"/>  
        <send>
            <endpoint xmlns="http://ws.apache.org/ns/synapse" name="mock_v1_activemqCount_httpEp">
                <http method="GET" uri-template="http://activemq:8161/api/jolokia/read/org.apache.activemq:type=Broker,brokerName={uri.var.broker},destinationType=Queue,destinationName={uri.var.queue}/QueueSize" />
            </endpoint>
        </send>
    </inSequence>
    <outSequence>
        <property name="ContentType" scope="axis2" type="STRING" value="application/json"/>
        <property name="messageType" scope="axis2" type="STRING" value="application/json"/>
        <switch source="json-eval($.status)">
            <case regex="200">
                <payloadFactory media-type="json">
                    <format> { "count": $1 } </format>
                    <args> <arg evaluator="json" expression="json-eval($.value)"/> </args>
                </payloadFactory>
            </case>
            <default>
                <property name="HTTP_SC" value="500" scope="axis2" />
                <payloadFactory media-type="json">
                    <format>{ "msg": "Unexpected error"}</format>
                    <args />
                </payloadFactory>
            </default>
        </switch>
        <respond/>
    </outSequence>
    <faultSequence/>
</resource>

    4. Lectura de mensajes 

Por último crearemos un Proxy Service que nos permitirá la lectura de esa misma cola. En este caso es importante también devolverle una respuesta a la cola, puesto que de lo contrarío reenviaría el mensaje a la cola de Dead Letter Channel, ActiveMQ.DLQ. 

<proxy xmlns="http://ws.apache.org/ns/synapse" name="activemqExample_v1_consumer_pr" transports="jms" statistics="disable" trace="disable" startOnLoad="true">
    <target>
        <inSequence>
            <log level="full">
                <property name="MSG" value="Receiving messages from Queue" />
            </log>
            <property action="set" name="OUT_ONLY" value="true"/>
            <drop />
        </inSequence>
        <faultSequence />
    </target>
    <parameter name="redeliveryPolicy.maximumRedeliveries">1</parameter>
    <parameter name="transport.jms.DestinationType">queue</parameter>
    <parameter name="transport.jms.SessionTransacted">true</parameter>
    <parameter name="transport.jms.Destination">myqueue</parameter>
    <parameter name="redeliveryPolicy.redeliveryDelay">2000</parameter>
    <parameter name="transport.jms.CacheLevel">consumer</parameter>
    <description />
</proxy>

Ahora podremos realizar la siguiente invocación y comprobar como por un lado enviamos el mensaje y lo leemos por el otro. 

curl --request POST 'https://localhost:8253/mock/activeMQ' \
--header 'Content-Type: application/json' \
--data-raw '{
    "author":"Alfred Bester",
    "name":"the stars my destination",
    "isbn":"1-85798-814-0",
    "year": 1956
}'

Con esto hemos visto un completo ejemplo sobre que podemos hacer para leer y escribir en una cola de ActiveMQ con WSO2. Si nos adentramos un poco más en la configuración podremos realizar tareas más complejas. Como siempre, todo el código lo podemos ver aquí.

Apache Camel y Hystrix: Como mejorar la resiliencia en microservicios

Hoy vamos uno de los patrones para mejorar la resiliencia de los micro servicios. En este caso concreto, será el patrón Circuit Breaker. Y podremos verlo aplicado a un micro servicio con Apache Camel a través de Hystrix. 

Como siempre empezaremos con un poco de teoría. Vamos a explicar que es el patrón Circuit Breaker y cual es su funcionalidad. Este patrón se basa en la idea de que algunas intercomunicaciones dentro del micro servicio pueden fallar. Pero al seguir funcionando la lógica normal, se pierde tiempo y recursos en hacer llamadas que posiblemente sean infructuosas. La idea del patrón es monitorizar el servicio externo al que vamos a invocar, para en caso de cualquier error, redirigir el flujo a otro servicio o lógica de negocio que nos haga mantener a nuestro micro servicio operativo y con baja latencia. 

Este patrón basa su funcionamiento en el control de estados del propio circuito. Contando con tres estados diferentes:

• Cerrado: Circuito operativo, todo funciona correctamente.
• Abierto: Circuito defectuoso. Se redirige el flujo hacia otro lado, permitiendo la perdida de recursos y seguir ofreciendo servicio. 
• Semi-abierto: Estado intermedio que se encargará de comprobar si el servicio externo se encuentra o no disponible y si mantiene el circuito abierto o lo cierra. 

Resumiendo mucho, el patrón circuit breaker actuará como un proxy que nos permitirá comprobar si el servicio externo se encuentra funcionando antes de realizar la llamada a dicho sistema, para poder ahorrarnos recursos y tiempo. 

Dentro de Apache Camel, hay tres implementaciones para llevar a cabo dicho patrón. Estas son:

• Hystrix: Librería de Netflix sobre la que basaremos el ejemplo. Actualmente (05/2021) ya no se encuentra en desarrollo y si en estado de mantenimiento. 
• Resilience4j: Librería basada en Hystrix y orientada a la implementación de varios patrones para la implementación de la resiliencia.
• MicroProfile Fault Tolerance: Librería de la fundación Eclipse para la implementación de patrones de resiliencia. 

Por tanto el uso a nivel de código del patrón Circuit Breaker será igual independientemente de la librería que seleccionemos para su implementación. Lo único que cambiará será librería que indiquemos como dependencia y la configuración de la misma. 

<dependency>
	<groupId>org.apache.camel.springboot</groupId>
	<artifactId>camel-hystrix-starter</artifactId>
</dependency>

Para el caso de Hystrix hay hasta 31 opciones de configuración que podemos introducir en el fichero de configuración de la aplicación de Apache Camel: application.properties. 

Para nuestro ejemplo modificaremos algunas opciones. De esta forma no tendremos que hacer tantas invocaciones para apreciar el funcionamiento del patrón Circuit Breaker. Estas opciones son: 

• circuitBreakerErrorThresholdPercentage: Porcentaje de error a partir del cual considera el circuito abierto y no acepta invocaciones.
• circuitBreakerSleepWindowInMilliseconds: Tiempo a pasar antes de volver a aceptar peticiones.
• metricsRollingStatisticalWindowInMilliseconds: Tiempo de duracion de las métricas asociadas.
• circuitBreakerRequestVolumeThreshold: Número mínimo de request a existir durante la duración de las métricas. Si no se alcanza este número mínimo de invocaciones no se considerará el circuito abierto, independientemente del porcentaje de error.  

Como hemos hecho referencia en el punto anterior, hay otro punto a destacar. Este son las métricas asociadas a la ejecución de nuestro micro servicio. En función a estas métricas y los valores de configuración, se decidirá el estado del circuito y como se debe actuar. 

Si utilizamos Spring en nuestra aplicación de Apache Camel, podemos acceder a esas métricas fácilmente a través de la URL /hystrix.stream.

Ahora pasaremos al primer ejemplo. Para llevarlo a cabo nos basaremos en ejemplos que hemos realizado anteriormente y que puedes ver aquí. En este ejemplo, invocaremos a una BBDD (la cual levantaremos con docker-compose). En función de si la BBDD se encuentra operativa o no, devolveremos uno u otro resultado. En el caso de que no hay BBDD, simplemente devolveremos un conjunto vacío. 

A través del método circuitBreaker indicaremos que vamos a implementar dicho patrón. Seguidamente podemos indicar si vamos a utilizar alguna configuración, en caso de que no indiquemos ninguna será utilizada la de por defecto. Debemos indicar una ruta a la que redirigir la llamada, que será la de nuestro servicio externo, en este caso la BBDD. Y debemos indicar que vamos a hacer en caso de error, para este caso enviaremos siempre el mismo resultado, lo cual podemos indicarlo a través del método onFallback. Y por último indicar que se termina la lógica asociada al circuit breaker. 

@Component
public class BookHystrixRouter extends RouteBuilder {
    @Override
    public void configure() throws Exception {
	restConfiguration().component("servlet").bindingMode(RestBindingMode.json_xml);

	rest().get("hystrix/sql").produces(MediaType.APPLICATION_JSON_VALUE)

		.route().circuitBreaker().id("hystrix/sql")
		.to("sql:{{sql.selectAll}}").onFallback()

		.setBody(bodyAs(ArrayList.class)).endCircuitBreaker()
		.log("--- SQL select all books---");
    }
}

Si levantamos el docker-compose, la aplicación y realizamos invocaciones a la siguiente URL, podremos ver como obtenemos de forma correcta todos los libros de la BBDD.

curl --location --request GET 'http://localhost:9090/hystrix/sql'

Una vez arrancada la aplicación, podemos acceder al Portainer, a través de la URL http://localhost:9000, y parar el contenedor de MySQL. En ese momento podremos volver a invocar a la aplicación y ver como las llamadas devuelven un conjunto vacío de elementos. 

El método onFallback nos permite responder un mensaje de forma inmediata, pero si lo que queremos hacer es reenviar la llamada a otro servicio externo para ello deberemos utilizar el método onFallbackViaNetwork. 

En el siguiente ejemplo que invoca a un Wiremock podremos comprobar esta segunda invocación. Además también nos permitirá visualizar a través de la consola del Wiremock si cuando el circuito se encuentra abierto, se sigue invocando o no al primer servicio. Aspecto que nos era más difícil de verificar con el contenedor de MySQL. 

Por ello crearemos un mapping para Wiremock que responda al path /book con la misma respuesta que devuelve la BBDD. Y nuestro ejemplo intentará invocar al path /fail, que al no existir siempre fallará y posteriormente invocará a /book. 

rest().get("hystrix/fail").produces(MediaType.APPLICATION_JSON_VALUE).route()

	.circuitBreaker().id("hystrix/fail")

	.toD("http://localhost:57001/fail?bridgeEndpoint=true").onFallbackViaNetwork()

	.toD("http://localhost:57001/book?bridgeEndpoint=true").end().unmarshal().json()
	.log("--- Wiremock select all books---");

Si invocamos al servicio y a la misma vez consultamos la consola del Wiremock, podremos ver como primero llama a un servicio y luego al otro. Devolviéndonos nuestro microservicio un listado de libros, como hacía el método anterior. 

curl --location --request GET 'http://localhost:9090/hystrix/fail'

Y en base a la configuración anteriormente indicada, si realizamos un conjunto elevado de llamadas. Podremos visualizar en la consola de Wiremock, como cuando el circuito se encuentra abierto, no se realizan llamadas al primer servicio. 

Por último indicar, que en versiones anteriores de Spring existía la dependencia spring-cloud-starter-hystrix-dashboard que generaba un dashboard para la visualización de las métricas de Hystrix. Pero la cual no es compatible con las versiones actuales de Spring y utilizadas en este ejemplo. En vez de usar esa librería, utilizaremos una imagen de docker llamada mlabouardy/hystrix-dashboard que nos permitirá acceder a dicho dashboard.  

Para ello debemos arrancar la imagen en el puerto 8080 y acceder a la URL: http://localhost:8080/hystrix. El siguiente paso será indicar la URL donde Hystrix esta mostrando las métricas. Hay que tener en cuenta, que si la aplicación de Apache Camel no esta siendo lanzada a través de una imagen en la misma red que la del dashboard deberemos indicar la URL de la siguiente forma en sistemas Windows o Mac: http://host.docker.internal:9090/hystrix.stream. A continuación un ejemplo de como se pueden visualizar dichas métricas con la ayuda de esta imagen:

Esto es todo, amigos. Ya hemos aprendido un poco más sobre resiliencia orientado a micro servicios y como hacerlo funcionar fácilmente en nuestra aplicación de Apache Camel. Si quieres acceso al código, lo tienes aquí.