[rhq-metrics] Possible implementation idea for pluggable aggregators
by Heiko W.Rupp
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-se...
[2] http://management-platform.blogspot.de/2014/11/messaging-infrastructure-u...
[3] https://github.com/pilhuhn/rhq-metrics/blob/msg-integration/rest-servlet/...
[4] https://github.com/pilhuhn/rhq-metrics/blob/msg-integration/clients/metri...
[5] https://github.com/pilhuhn/rhq-metrics/blob/msg-integration/rest-servlet/...
[6] https://github.com/pilhuhn/rhq-metrics/blob/msg-integration/clients/metri...
9 years, 4 months