ConsistentlyFailingMonitor.java

package com.ricston.monitor;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.mule.construct.Flow;

public class ConsistentlyFailingMonitor {
  
	private long lastRecordedEventsReceived = 0;
	private long lastRecordedFailedEvents = 0;
	private float threshold = 0.5f;
	private float lastRecordedRatio;
	private boolean constantlyFailing = false;
	private String constantlyFailingString = Boolean.toString(constantlyFailing);
	
	private Flow flow;
	protected Log logger = LogFactory.getLog(getClass());
	private int count;
	
	private final String logLineLastRecorded = "lastRecordedEventsReceived: %s -- lastRecordedFailedEvents: %s";
	private final String logLineNewEvents = "newEventsReceived: %s -- newFailedEvents: %s";
	private final String logLineDiffEvents = "diffEventsReceived: %s -- diffFailedEvents: %s";
	private final String logLineRatioFailed = "Ratio: %s -- Failing: %s";

	public synchronized void monitor() {
		count++;
		
		// quick fix for spring task scheduler scheduling our tasks twice
		if (count % 2 == 0) {
			
			//retrieve values of events received and events failed
			long newEventsReceived = flow.getStatistics().getTotalEventsReceived();
			long newFailedEvents = flow.getStatistics().getExecutionErrors();
			
			//calculate difference between now and last recorded statistics
			long diffEventsReceived = newEventsReceived - lastRecordedEventsReceived;
			long diffFailedEvents = newFailedEvents - lastRecordedFailedEvents;
			
			//some logging
			logger.debug(String.format(logLineLastRecorded, lastRecordedEventsReceived, lastRecordedFailedEvents));
			logger.debug(String.format(logLineNewEvents, newEventsReceived, newFailedEvents));
			logger.debug(String.format(logLineDiffEvents, diffEventsReceived, diffFailedEvents));
			
			//store last recorded values
			lastRecordedEventsReceived = newEventsReceived;
			lastRecordedFailedEvents = newFailedEvents;
			
			//if we did not receive any good events since last time recorded, lets check if we received error events 
			if (diffEventsReceived == 0) {
				
				//if we received only error events, set failed to true
				if (diffFailedEvents > 0) {
					lastRecordedRatio = 1;
					updateConstantlyFailing(true);
				} 
				//if we received no events at all, then set failed to false
				else {
					lastRecordedRatio = 0;
					updateConstantlyFailing(false);
				}
			} 
			//if we received good events, lets calculate the ratio of bad events against good events
			//and update the failed status accordingly (checking with the threshold)
			else {
				lastRecordedRatio = (float) diffFailedEvents / diffEventsReceived;
				updateConstantlyFailing(lastRecordedRatio > threshold);
			}
			
			//some more logging
			logger.debug(String.format(logLineRatioFailed, lastRecordedRatio, constantlyFailingString));
		}
	}

	protected void updateConstantlyFailing(boolean value) {
		constantlyFailing = value;
		constantlyFailingString = Boolean.toString(constantlyFailing);
	}

	//all getters and setters
}

cxfproxy.xml

<?xml version="1.0" encoding="UTF-8"?>

<mule xmlns:http="http://www.mulesoft.org/schema/mule/http" xmlns:cxf="http://www.mulesoft.org/schema/mule/cxf" xmlns="http://www.mulesoft.org/schema/mule/core" xmlns:doc="http://www.mulesoft.org/schema/mule/documentation"
  xmlns:spring="http://www.springframework.org/schema/beans" version="EE-3.3.1"
	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xsi:schemaLocation="
http://www.mulesoft.org/schema/mule/http http://www.mulesoft.org/schema/mule/http/current/mule-http.xsd 
http://www.mulesoft.org/schema/mule/cxf http://www.mulesoft.org/schema/mule/cxf/current/mule-cxf.xsd 
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-current.xsd 
http://www.mulesoft.org/schema/mule/core http://www.mulesoft.org/schema/mule/core/current/mule.xsd ">
	
    <flow name="CxfProxyFlow" doc:name="CxfProxyFlow">
        <http:inbound-endpoint exchange-pattern="request-response" host="localhost" port="8081" doc:name="HTTP" path="proxy"/>
        <cxf:proxy-service doc:name="SOAP" payload="body"/>
        <cxf:proxy-client doc:name="SOAP" enableMuleSoapHeaders="true" payload="body"/>
        <http:outbound-endpoint exchange-pattern="request-response" host="localhost" port="8081" doc:name="HTTP" path="service"/>
    </flow>

</mule>

monitor-mbean.xml

<?xml version="1.0" encoding="UTF-8"?>

<mule xmlns:http="http://www.mulesoft.org/schema/mule/http"  xmlns:cxf="http://www.mulesoft.org/schema/mule/cxf" xmlns="http://www.mulesoft.org/schema/mule/core" xmlns:doc="http://www.mulesoft.org/schema/mule/documentation"
  xmlns:spring="http://www.springframework.org/schema/beans" version="EE-3.3.1"
	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xsi:schemaLocation="
http://www.mulesoft.org/schema/mule/http http://www.mulesoft.org/schema/mule/http/current/mule-http.xsd 
http://www.mulesoft.org/schema/mule/cxf http://www.mulesoft.org/schema/mule/cxf/current/mule-cxf.xsd 
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-current.xsd
http://www.mulesoft.org/schema/mule/core http://www.mulesoft.org/schema/mule/core/current/mule.xsd  ">

	<spring:beans>

		 <spring:bean id="CxfProxyFlowMonitorExporter" class="org.springframework.jmx.export.MBeanExporter">
			<spring:property name="beans">
				<spring:map>
					<spring:entry
						key="Mule.${app.name}:category=custom,name=CxfProxyFlowMonitor" value-ref="CxfProxyFlowMonitor" />
				</spring:map>
			</spring:property>
		</spring:bean> 

	</spring:beans>
	
</mule>

monitor-scheduler.xml

<?xml version="1.0" encoding="UTF-8"?>

<mule xmlns:http="http://www.mulesoft.org/schema/mule/http" xmlns:cxf="http://www.mulesoft.org/schema/mule/cxf" xmlns="http://www.mulesoft.org/schema/mule/core" xmlns:doc="http://www.mulesoft.org/schema/mule/documentation"
  xmlns:spring="http://www.springframework.org/schema/beans" xmlns:task="http://www.springframework.org/schema/task" version="EE-3.3.1"
	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xsi:schemaLocation="
http://www.mulesoft.org/schema/mule/http http://www.mulesoft.org/schema/mule/http/current/mule-http.xsd 
http://www.mulesoft.org/schema/mule/cxf http://www.mulesoft.org/schema/mule/cxf/current/mule-cxf.xsd 
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-current.xsd 
http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-3.1.xsd
http://www.mulesoft.org/schema/mule/core http://www.mulesoft.org/schema/mule/core/current/mule.xsd  ">

	<spring:beans>
		 <spring:bean id="CxfProxyFlowMonitor" class="com.ricston.monitor.ConsistentlyFailingMonitor"> 
		 	<spring:property name="flow" ref="CxfProxyFlow" /> 
		 </spring:bean> 

		<task:scheduler id="CxfProxyFlowMonitorScheduler" pool-size="1" />
		
		<task:scheduled-tasks scheduler="CxfProxyFlowMonitorScheduler">
			<task:scheduled ref="CxfProxyFlowMonitor" method="monitor" fixed-delay="10000" />
		</task:scheduled-tasks>
	</spring:beans>

</mule>