Updated Branches: refs/heads/master f3e0efc4f -> dc11e3176
Add partition_id to fault_message Project: http://git-wip-us.apache.org/repos/asf/incubator-stratos/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-stratos/commit/7df3d431 Tree: http://git-wip-us.apache.org/repos/asf/incubator-stratos/tree/7df3d431 Diff: http://git-wip-us.apache.org/repos/asf/incubator-stratos/diff/7df3d431 Branch: refs/heads/master Commit: 7df3d43103f78f13dec6e74f810de55481042929 Parents: 32499e8 Author: Manula Thantriwatte <[email protected]> Authored: Tue Dec 17 12:19:46 2013 +0530 Committer: Manula Thantriwatte <[email protected]> Committed: Tue Dec 17 12:19:46 2013 +0530 ---------------------------------------------------------------------- .../event-formatters/FaultMessageEventFormatter.xml | 2 +- .../cep/artifacts/execution-plans/AverageHeathRequest.xml | 6 +++--- .../artifacts/execution-plans/GradientOfHealthRequest.xml | 7 +++++-- .../artifacts/stream-definitions/stream-manager-config.xml | 6 ++++-- .../modules/health-stats/src/main/bin/health-publisher.sh | 2 +- .../cartridge/agent/health/publisher/HealthPublisher.java | 8 +++++++- 6 files changed, 21 insertions(+), 10 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/7df3d431/extensions/cep/artifacts/event-formatters/FaultMessageEventFormatter.xml ---------------------------------------------------------------------- diff --git a/extensions/cep/artifacts/event-formatters/FaultMessageEventFormatter.xml b/extensions/cep/artifacts/event-formatters/FaultMessageEventFormatter.xml index 9d91cef..a46241a 100644 --- a/extensions/cep/artifacts/event-formatters/FaultMessageEventFormatter.xml +++ b/extensions/cep/artifacts/event-formatters/FaultMessageEventFormatter.xml @@ -3,7 +3,7 @@ statistics="disable" trace="enable" xmlns="http://wso2.org/carbon/eventformatter";> <from streamName="fault_message" version="1.0.0"/> <mapping customMapping="enable" type="json"> - <inline>{"member_fault":{"cluster_id":"{{cluster_id}}","partition_id":"{{partition_id}}","member_id":"{{member_id}}"}}</inline> + <inline>{"member_fault":{"cluster_id":"{{cluster_id}}","network_partition_id":"{{network_partition_id}}","member_id":"{{member_id}}","partition_id":"{{partition_id}}"}}</inline> </mapping> <to eventAdaptorName="JMSOutputAdaptor" eventAdaptorType="jms"> <property name="transport.jms.Destination">summarized-health-stats</property> http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/7df3d431/extensions/cep/artifacts/execution-plans/AverageHeathRequest.xml ---------------------------------------------------------------------- diff --git a/extensions/cep/artifacts/execution-plans/AverageHeathRequest.xml b/extensions/cep/artifacts/execution-plans/AverageHeathRequest.xml index 8849385..c220a0d 100644 --- a/extensions/cep/artifacts/execution-plans/AverageHeathRequest.xml +++ b/extensions/cep/artifacts/execution-plans/AverageHeathRequest.xml @@ -9,9 +9,9 @@ <importedStreams> <stream as="healthStats1" name="cartridge_agent_health_stats" version="1.0.0"/> </importedStreams> - <queryExpressions><![CDATA[from healthStats1 [health_description == 'port_not_open'] select cluster_id,network_partition_id,member_id insert into fault_message; - from healthStats1 [health_description == 'load_average'] #window.timeBatch(1 min) select cluster_id,network_partition_id,avg(value) as average_load_average group by cluster_id,network_partition_id insert into average_load_average_stats; - from healthStats1 [health_description == 'memory_consumption'] #window.timeBatch(1 min) select cluster_id,network_partition_id,avg(value) as average_memory_consumption group by cluster_id,network_partition_id insert into average_memory_consumption_stats;]]></queryExpressions> + <queryExpressions><![CDATA[from healthStats1 [health_description == 'port_not_open'] select cluster_id, network_partition_id, member_id, partition_id insert into fault_message; + from healthStats1 [health_description == 'load_average'] #window.timeBatch(1 min) select cluster_id, network_partition_id, avg(value) as average_load_average group by cluster_id, network_partition_id insert into average_load_average_stats; + from healthStats1 [health_description == 'memory_consumption'] #window.timeBatch(1 min) select cluster_id, network_partition_id, avg(value) as average_memory_consumption group by cluster_id, network_partition_id insert into average_memory_consumption_stats;]]></queryExpressions> <exportedStreams> <stream name="average_load_average_stats" passthroughFlow="disable" valueOf="average_load_average_stats" version="1.0.0"/> http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/7df3d431/extensions/cep/artifacts/execution-plans/GradientOfHealthRequest.xml ---------------------------------------------------------------------- diff --git a/extensions/cep/artifacts/execution-plans/GradientOfHealthRequest.xml b/extensions/cep/artifacts/execution-plans/GradientOfHealthRequest.xml index c5bc672..2d79ebb 100644 --- a/extensions/cep/artifacts/execution-plans/GradientOfHealthRequest.xml +++ b/extensions/cep/artifacts/execution-plans/GradientOfHealthRequest.xml @@ -10,12 +10,15 @@ <stream as="healthStats2" name="cartridge_agent_health_stats" version="1.0.0"/> </importedStreams> <queryExpressions><![CDATA[ - from healthStats2 [health_description == 'load_average'] #window.stratos:gradient(1 min, value) select cluster_id,network_partition_id,value as gradient_load_average group by cluster_id,network_partition_id insert into gradient_load_average_stats; - from healthStats2 [health_description == 'memory_consumption'] #window.stratos:gradient(1 min, value) select cluster_id,network_partition_id,value as gradient_memory_consumption group by cluster_id,network_partition_id insert into gradient_memory_consumption_stats;]]></queryExpressions> + from healthStats2 #window.stratos:faultHandling(1min, member_id) select cluster_id, network_partition_id, member_id, partition_id group by cluster_id, network_partition_id insert into fault_message; + from healthStats2 [health_description == 'load_average'] #window.stratos:gradient(1 min, value) select cluster_id, network_partition_id, value as gradient_load_average group by cluster_id, network_partition_id insert into gradient_load_average_stats; + from healthStats2 [health_description == 'memory_consumption'] #window.stratos:gradient(1 min, value) select cluster_id, network_partition_id, value as gradient_memory_consumption group by cluster_id, network_partition_id insert into gradient_memory_consumption_stats;]]></queryExpressions> <exportedStreams> <stream name="gradient_load_average_stats" valueOf="gradient_load_average_stats" version="1.0.0"/> <stream name="gradient_memory_consumption_stats" valueOf="gradient_memory_consumption_stats" version="1.0.0"/> + <stream name="fault_message" passthroughFlow="disable" + valueOf="fault_message" version="1.0.0"/> </exportedStreams> </executionPlan> http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/7df3d431/extensions/cep/artifacts/stream-definitions/stream-manager-config.xml ---------------------------------------------------------------------- diff --git a/extensions/cep/artifacts/stream-definitions/stream-manager-config.xml b/extensions/cep/artifacts/stream-definitions/stream-manager-config.xml index 858da00..8089507 100755 --- a/extensions/cep/artifacts/stream-definitions/stream-manager-config.xml +++ b/extensions/cep/artifacts/stream-definitions/stream-manager-config.xml @@ -66,9 +66,10 @@ <correlationData> </correlationData> <payloadData> - <property name="cluster_id" type="String" /> + <property name="cluster_id" type="String" /> <property name="network_partition_id" type="String"/> <property name="member_id" type="String" /> + <property name="partition_id" type="String" /> <property name="health_description" type="String"/> <property name="value" type="double"/> </payloadData> @@ -167,8 +168,9 @@ </correlationData> <payloadData> <property name="cluster_id" type="String"/> - <property name="partition_id" type="String"/> + <property name="network_partition_id" type="String"/> <property name="member_id" type="String"/> + <property name="partition_id" type="String"/> </payloadData> </streamDefinition> <!-- cartridge agent health stats stream definitions end --> http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/7df3d431/products/cartridge-agent/modules/health-stats/src/main/bin/health-publisher.sh ---------------------------------------------------------------------- diff --git a/products/cartridge-agent/modules/health-stats/src/main/bin/health-publisher.sh b/products/cartridge-agent/modules/health-stats/src/main/bin/health-publisher.sh index ab115e6..2db8d62 100755 --- a/products/cartridge-agent/modules/health-stats/src/main/bin/health-publisher.sh +++ b/products/cartridge-agent/modules/health-stats/src/main/bin/health-publisher.sh @@ -28,7 +28,7 @@ class_path=${lib_path}andes-client-0.13.wso2v8.jar:${lib_path}ant-1.7.0.jar:${li current_path=`pwd` -java -cp $class_path -Dmember.id=$1 -Dkey.file.path=$current_path/../security/client-truststore.jks -Dthrift.receiver.ip=$2 -Dthrift.receiver.port=$3 -Dopen.ports=$4 -Dcluster.id=$5 -Dnetwork.partition.id=$6 org.apache.stratos.cartridge.agent.health.publisher.Main $* +java -cp $class_path -Dmember.id=$1 -Dkey.file.path=$current_path/../security/client-truststore.jks -Dthrift.receiver.ip=$2 -Dthrift.receiver.port=$3 -Dopen.ports=$4 -Dcluster.id=$5 -Dnetwork.partition.id=$6 -Dpartition.id=$7 org.apache.stratos.cartridge.agent.health.publisher.Main $* echo "Health publisher completed" http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/7df3d431/products/cartridge-agent/modules/health-stats/src/main/java/org/apache/stratos/cartridge/agent/health/publisher/HealthPublisher.java ---------------------------------------------------------------------- diff --git a/products/cartridge-agent/modules/health-stats/src/main/java/org/apache/stratos/cartridge/agent/health/publisher/HealthPublisher.java b/products/cartridge-agent/modules/health-stats/src/main/java/org/apache/stratos/cartridge/agent/health/publisher/HealthPublisher.java index fa4c519..bc34bcd 100644 --- a/products/cartridge-agent/modules/health-stats/src/main/java/org/apache/stratos/cartridge/agent/health/publisher/HealthPublisher.java +++ b/products/cartridge-agent/modules/health-stats/src/main/java/org/apache/stratos/cartridge/agent/health/publisher/HealthPublisher.java @@ -66,6 +66,7 @@ public class HealthPublisher implements Observer { " {'name':'cluster_id','type':'STRING'}," + " {'name':'network_partition_id','type':'STRING'}," + " {'name':'member_id','type':'STRING'}," + + " {'name':'partition_id','type':'STRING'}," + " {'name':'health_description','type':'STRING'}," + " {'name':'value','type':'DOUBLE'}" + " ]" + @@ -107,10 +108,15 @@ public class HealthPublisher implements Observer { if(networkPartitionId == null) { throw new RuntimeException("network.partition.id system property was not found"); } + String partitionId = System.getProperty("partition.id"); + if (partitionId == null){ + throw new RuntimeException("partition.id system property was not found"); + } for (Map.Entry<String, Double> entry : stats.entrySet()) { - Object[] payload = new Object[]{clusterID,networkPartitionId,memberID,entry.getKey(), entry.getValue()}; + Object[] payload = new Object[]{clusterID, networkPartitionId, memberID, partitionId, entry.getKey(), + entry.getValue()}; Event event = eventObject(null, null, payload, new HashMap<String, String>()); try { asyncDataPublisher.publish(DATA_STREAM_NAME, VERSION, event);
