Internal Handler Buffering

The Anypoint Runtime Manager agent provides a simple way of implementing internal handlers with buffering support. This document explains how to implement and configure buffers for the Runtime Manager agent internal handlers.

Implement a Buffered Internal Handler

BufferedHandler Abstract Class

To implement a buffered internal handler, you need to extend the BufferedHandler<T> abstract class. The methods to overwrite are:

boolean canHandle(T message)

Returns true if the message can be handled by the internal handler.

boolean flush(Collection<T> collectionOfMessages)

Provides the implementation to send the messages stored in the buffer to the external service.

Example

@Named("com.mulesoft.agent.mypublisher")
@Singleton
public class MyPublisher extends BufferedHandler<List<Metric>>
{
    @Configurable(value = "mule")
    String metricPrefix;

    @Inject
    public MyPublisher()
    {
        super();
    }

    public MyPublisher(OnOffSwitch enabledSwitch)
    {
        super();
        this.enabledSwitch = enabledSwitch;
    }

    @Override
    public boolean canHandle(@NotNull List<Metric> metrics)
    {
        // Evaluates if the message could be handled
    }

    @Override
    public boolean flush(@NotNull Collection<List<Metric>> listOfMetrics)
    {
        // Send messages to external service
    }
}

Configure a Buffering Internal Handler

Buffering Configuration Object

If the internal handler includes buffering support, you can configure the buffer in the mule-agent.yml file. The following table lists the available configuration parameters.

Field Description Default value

Field	Description	Default value
`type`	Buffer type. Valid values are `MEMORY` or `DISK`.
`retryCount`	Number of retries after an exception was thrown in the flush method.
`maximumCapacity`	Maximum number of events to keep in buffer.
`flushFrequency`	Frequency at which the flush method will run, in milliseconds.
`filePath`	Path to the buffer file. Only valid for buffers of type `DISK`.	`PATH/bufferFile`
`whenExhausted`	Strategy to follow when the buffer is exhausted. If not specified, the buffer will grow automatically. Valid values are `FAIL` and `FLUSH.`	N/A
`flushThreadsNumber`	Number of threads used by the FlushThreadExecutor when the buffer is exhausted and the strategy defined by `whenExhausted` is `FLUSH` (see row above).	`60`
`discardMessagesOnFlushFailure`	If true, messages will not be returned to the buffer if the flush is not successful.	`false`

type

Buffer type. Valid values are MEMORY or DISK.

retryCount

Number of retries after an exception was thrown in the flush method.

maximumCapacity

Maximum number of events to keep in buffer.

flushFrequency

Frequency at which the flush method will run, in milliseconds.

filePath

Path to the buffer file. Only valid for buffers of type DISK.

PATH/bufferFile

whenExhausted

Strategy to follow when the buffer is exhausted. If not specified, the buffer will grow automatically. Valid values are FAIL and FLUSH.

N/A

flushThreadsNumber

Number of threads used by the FlushThreadExecutor when the buffer is exhausted and the strategy defined by whenExhausted is FLUSH (see row above).

60

discardMessagesOnFlushFailure

If true, messages will not be returned to the buffer if the flush is not successful.

false

The maximum size capacity for the buffer is configured based on a number of events and not in bytes. Keep in mind that if the events are of a considerable size and you don’t set maximumCapacity to a low value, the heap might suffer stress and this might impact the application’s performance.

If you configure a DISK buffer, the agent will also instantiate an in-memory buffer in order to improve the performance of the I/O operations. This memory buffer has a default value of 10000 events and is flushed to disk every 500ms, and in batches of 10000 events each. These values are configurable through the system properties agent.disk.buffering.cache.size, agent.disk.buffering.cache.flush.frequency and agent.disk.buffering.max.flushing.size, respectively.

Flushing from disk to the external system in batches of 10000 events each and the agent.disk.buffering.max.flushing.size property are supported for Mule agent plugin versions 2.1.9 and 1.11.3 and later.

Sample `mule-agent.yml` Files

Configure Buffering in a Publisher

---
muleInstanceUniqueId: validId
organizationId: organizationId

transports:
    websocket.transport:
        security:
            keyStorePassword: mykeystorePassword
            keyStoreAlias: agent
            keyStoreAliasPassword: agentpassword

    rest.agent.transport:
        security:
            keyStorePassword: mykeystorePassword
            keyStoreAlias: agent
            keyStoreAliasPassword: agentpassword
        port: 9997

services:
    mule.agent.application.service:
        enabled: true

    mule.agent.domain.service:
        enabled: true

    mule.agent.jmx.publisher.service:
        enabled: true
        frequency: 15
        frequencyTimeUnit: MINUTES
        beans:
            -   beanQueryPattern: java.lang:type=Runtime
                attribute: Uptime
                monitorMessage: Monitoring memory up-time
            -   beanQueryPattern: java.lang:type=MemoryPool,*
                attribute: Usage.used
                monitorMessage" : Used Memory

internalHandlers:
    domaindeploymentnotification.internal.message.handler:
        enabled: true

    applicationdeploymentnotification.internal.message.handler:
        enabled: false

    com.mulesoft.agent.test.buffering.jmx.internal.handler:
        enabled: true
        buffer:
            type: DISK
            retryCount: 1
            flushFrequency: 10000
            maximumCapacity: 30
            filePath: publisher-buffer.log

externalHandlers:
    applications.request.handler:
        enabled: true

    domains.request.handler:
        enabled: true

Configure Buffering in Events Tracking

---
muleInstanceUniqueId: validId
organizationId: organizationId

transports:
    websocket.transport:
        security:
            keyStorePassword: mykeystorePassword
            keyStoreAlias: agent
            keyStoreAliasPassword: agentpassword

    rest.agent.transport:
        security:
            keyStorePassword: mykeystorePassword
            keyStoreAlias: agent
            keyStoreAliasPassword: agentpassword
        port: 9997

services:
    mule.agent.application.service:
        enabled: true

    mule.agent.domain.service:
        enabled: true

    mule.agent.jmx.publisher.service:
        enabled: true
        frequency: 15
        frequencyTimeUnit: MINUTES
        beans:
            -   beanQueryPattern: java.lang:type=Runtime
                attribute: Uptime
                monitorMessage: Monitoring memory up-time
            -   beanQueryPattern: java.lang:type=MemoryPool,*
                attribute: Usage.used
                monitorMessage" : Used Memory

internalHandlers:
    domaindeploymentnotification.internal.message.handler:
        enabled: true

    applicationdeploymentnotification.internal.message.handler:
        enabled: false

    tracking.notification.internal.message.handler:
        enabled: true
        buffer:
            type: MEMORY
            retryCount: 1
            flushFrequency: 10000
            maximumCapacity: 30

externalHandlers:
    applications.request.handler:
        enabled: true

    domains.request.handler:
        enabled: true