[rhq-metrics] Possible implementation idea for pluggable aggregators

Heiko W.Rupp hrupp at redhat.com
Wed Dec 3 20:56:45 UTC 2014


Hey,

so I have been thinking on how users can write their own aggregators and aggregation functions and deploy them in a pluggable way. 

As we are thinking about messaging for rhq.next anyway [1] I thought this could be done by forwarding data and/or requests to a message queue / topic and have listeners react on those messages.
Being old-school, I decided to use MessageDrivenBeans - especially as Mazz has already created the necessary bits for running an ActiveMQ broker as subsystem in WildFly and also a ResourceAdapter [2]

I've implemented two kinds of things so far (don't worry, all in a branch in my own repo :-):

1) forward all incoming data to a topic and have then MDBs work on them [3] - this could be used in alerting, where
the alert engine just picks up those messages and works on them.
Samples via MDB look like this [4]:



@MessageDriven(activationConfig = {
    @ActivationConfigProperty(propertyName = "destinationType", propertyValue = "javax.jms.Topic"),
    @ActivationConfigProperty(propertyName = "destination", propertyValue = "metrics")
}
)
@SuppressWarnings("unused")
public class MaxMdb extends AbstractMetricDrivenMDB {

    @Override
    void workOn(List<RawNumericMetric> metrics) {
/// do work here
  }
}

I guess the activation config could even go into the abstract superclass


2) Adhoc aggregation worker
Those are MDBs that are triggered by this code block in the MetricHandler [5]


  @GET
    @Path("/metrics/{agg}/{id}")
    public void getAggregate(@Suspended final AsyncResponse asyncResponse,
                                 @PathParam("agg")String name, @PathParam("id")String id) throws Exception {

        BasicMessage msg = new BasicMessage(id);
        Map<String,String> headers = new HashMap<>();
        headers.put("function",name);

        aggregationProcessor.sendAndListen(msg,
            new BasicMessageListener<BasicMessage>() {
                @Override
                protected void onBasicMessage(BasicMessage basicMessage) {
  /// Work on results - e.g  asyncResponse.resume();

Basically the name of the aggregation function is passed as path parameter and
added as a header field for messaging

The MDB code would then look like this [6]:


@MessageDriven(activationConfig = {
    @ActivationConfigProperty(propertyName = "destinationType", propertyValue = "javax.jms.Queue"),
    @ActivationConfigProperty(propertyName = "destination", propertyValue = "aggregationTask"),
    @ActivationConfigProperty(propertyName = "messageSelector", propertyValue = "function = 'invert'")
}
)
@SuppressWarnings("unused")
public class InvertAdHocAggregator extends AbstractAggregationWorkerMDB {

    @Override
    BasicMessage work(BasicMessage msg) {
 /// compute the result here
  }
}

The most interesting part here is 

@ActivationConfigProperty(propertyName = "messageSelector", propertyValue = "function = 'invert'")

where the propertyValue defines a selector that determines what messages can reach this MDB.

So with above MetricHandler code

GET /metrics/bla/123  would not trigger this MDB but
GET /metrics/invert/123 would do.

I am not saying we should adopt this way for user definable aggregations; the beauties are though
- each MDB can just be deployed into the server as its own jar file
- processing can easily be distributed onto many compute nodes because of the message bus.

[1] https://developer.jboss.org/en/rhq/blog/2014/08/14/thoughts-on-rhqnext-server-architecture
[2] http://management-platform.blogspot.de/2014/11/messaging-infrastructure-using-activemq.html
[3] https://github.com/pilhuhn/rhq-metrics/blob/msg-integration/rest-servlet/src/main/java/org/rhq/metrics/restServlet/MetricHandler.java#L123
[4] https://github.com/pilhuhn/rhq-metrics/blob/msg-integration/clients/metric-driven-mdb/src/main/java/org/rhq/metrics/clients/metricDrivenMDB/MaxMdb.java
[5] https://github.com/pilhuhn/rhq-metrics/blob/msg-integration/rest-servlet/src/main/java/org/rhq/metrics/restServlet/MetricHandler.java#L251
[6] https://github.com/pilhuhn/rhq-metrics/blob/msg-integration/clients/metric-driven-mdb/src/main/java/org/rhq/metrics/clients/metricDrivenMDB/adhoc/InvertAdHocAggregator.java


More information about the rhq-devel mailing list