[#OCESPER-11] Improve Channel throughput monitor so throughput is more informative in the case of pollable channels

OpenCredo Esper

Improve Channel throughput monitor so throughput is more informative in the case of pollable channels

Created: 17/Feb/10 11:56 AM Updated: 21/Feb/11 12:13 PM

Component/s:

EsperSpringIntegrationSupport

Affects Version/s:

2.0

Fix Version/s:

2.2

Time Tracking:

Original Estimate:

1 day

Remaining Estimate:

1 day

Time Spent:

Not Specified

Description

« Hide

Currently the reported throughput relates to messages sent to the channel. The throughput monitor should distinguish between pollable and subscribable channels.
In the case of pollable throughput should be measured as messages received and in addition average queue size should be made available. In the case of subscribable it should be messages sent as currently since there is no buffering and no receive call to intercept.

All

Comments

Work Log

Change History

Sort Order:

[ Permalink | « Hide ]

Tomas Lukosius added a comment - 19/Aug/10 10:27 AM

Changed assignee and fix version.

[ Permalink | « Hide ]

Tomas Lukosius added a comment - 21/Aug/10 10:58 AM

If throughput monitor looks after pollable channel then:

....
EsperWireTap wireTap = new EsperWireTap(template, this.sourceId);
wireTap.setSendContext(true);
// We always want to listen to event per message sent
wireTap.setPostSend(true);
wireTap.setPreSend(false);
wireTap.setPreReceive(false);
wireTap.setPostReceive(false);

if (channel instanceof PollableChannel) { // This channel is pollable - need to calculate messages received // and average queue size. // We want to know how many messages were received. wireTap.setPostReceive(true); StringBuilder sb = new StringBuilder(); sb.append("select count(PS), count(PR) from pattern [every PS="); sb.append(windowName).append("(operation=").append(IntegrationOperation.class.getName()).append(".") .append(IntegrationOperation.POST_SEND).append(") OR every PR="); sb.append(windowName).append("(operation=").append(IntegrationOperation.class.getName()).append(".") .append(IntegrationOperation.POST_RECEIVE).append(")]"); sb.append(".win:time_batch(").append(timeSample).append(")"); EsperStatement listenForSourceId = new EsperStatement(sb.toString()); listenForSourceId.setSubscriber(this); template.addStatement(listenForSourceId); } else { StringBuilder sb = new StringBuilder(); sb.append("select count(*) as throughput from ").append(windowName); sb.append(".win:time_batch(").append(this.timeSample).append(")"); sb.append(" where operation=").append(IntegrationOperation.class.getName()).append(".") .append(IntegrationOperation.POST_SEND); EsperStatement listenForSourceId = new EsperStatement(sb.toString()); listenForSourceId.setSubscriber(this); template.addStatement(listenForSourceId); }
.....

ELP in PollableChannel branch will contain count of sent messages (event from EsperWireTap.postSend) and received (event from EsperWireTap.postReceive) messages.

Question 1:
Should we create new Esper event every time pollable channel is polled (that means EsperWireTap.postReceive is invoked) and in case no messages received - EsperWireTap.postReceive is invoked with null as message parameter?

Note that in ELP has "win:time_batch(this.timeSample)" that means results are calculated for every time interval which is "this.timeSample" and does not include results from previous interval. For example assume we do have empty pollable channel and during the time "t1" (which is equals this.timeSample) 5 messages where sent to and 2 messages received from that channel. EsperChannelThroughputMonitor will print: Sent throughput of 5, received throughput of 2 on pollable channel - <messageChannel-name>. But for the next time interval "t2" no new messages were sent neither received from the channel - in that case EsperChannelThroughputMonitor will print: Sent throughput of 0, received throughput of 0 on pollable channel - <messageChannel-name>.

Question 2:
Do you expect to show activity which happened on channel during last configured interval or channel life time?

Requirement for PollableChannel is: "...throughput should be measured as messages received and in addition average queue size should be made available."
Question 3:
average-queue-size = (queue-size1+queue-size2+...queue-sizeN)/N. How big N should be?

[ Show » ]

Tomas Lukosius added a comment - 21/Aug/10 10:58 AM If throughput monitor looks after pollable channel then: .... EsperWireTap wireTap = new EsperWireTap(template, this.sourceId); wireTap.setSendContext(true); // We always want to listen to event per message sent wireTap.setPostSend(true); wireTap.setPreSend(false); wireTap.setPreReceive(false); wireTap.setPostReceive(false); if (channel instanceof PollableChannel) { // This channel is pollable - need to calculate messages received // and average queue size. // We want to know how many messages were received. wireTap.setPostReceive(true); StringBuilder sb = new StringBuilder(); sb.append("select count(PS), count(PR) from pattern [every PS="); sb.append(windowName).append("(operation=").append(IntegrationOperation.class.getName()).append(".") .append(IntegrationOperation.POST_SEND).append(") OR every PR="); sb.append(windowName).append("(operation=").append(IntegrationOperation.class.getName()).append(".") .append(IntegrationOperation.POST_RECEIVE).append(")]"); sb.append(".win:time_batch(").append(timeSample).append(")"); EsperStatement listenForSourceId = new EsperStatement(sb.toString()); listenForSourceId.setSubscriber(this); template.addStatement(listenForSourceId); } else { StringBuilder sb = new StringBuilder(); sb.append("select count(*) as throughput from ").append(windowName); sb.append(".win:time_batch(").append(this.timeSample).append(")"); sb.append(" where operation=").append(IntegrationOperation.class.getName()).append(".") .append(IntegrationOperation.POST_SEND); EsperStatement listenForSourceId = new EsperStatement(sb.toString()); listenForSourceId.setSubscriber(this); template.addStatement(listenForSourceId); } ..... ELP in PollableChannel branch will contain count of sent messages (event from EsperWireTap.postSend) and received (event from EsperWireTap.postReceive) messages. Question 1: Should we create new Esper event every time pollable channel is polled (that means EsperWireTap.postReceive is invoked) and in case no messages received - EsperWireTap.postReceive is invoked with null as message parameter? Note that in ELP has "win:time_batch(this.timeSample)" that means results are calculated for every time interval which is "this.timeSample" and does not include results from previous interval. For example assume we do have empty pollable channel and during the time "t1" (which is equals this.timeSample) 5 messages where sent to and 2 messages received from that channel. EsperChannelThroughputMonitor will print: Sent throughput of 5, received throughput of 2 on pollable channel - <messageChannel-name>. But for the next time interval "t2" no new messages were sent neither received from the channel - in that case EsperChannelThroughputMonitor will print: Sent throughput of 0, received throughput of 0 on pollable channel - <messageChannel-name>. Question 2: Do you expect to show activity which happened on channel during last configured interval or channel life time? Requirement for PollableChannel is: "...throughput should be measured as messages received and in addition average queue size should be made available." Question 3: average-queue-size = (queue-size1+queue-size2+...queue-sizeN)/N. How big N should be?