You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@stratos.apache.org by im...@apache.org on 2015/02/08 12:51:07 UTC
stratos git commit: Introducing statistics publishing interfaces and
factory classes
Repository: stratos
Updated Branches:
refs/heads/master 3bf78f647 -> dc863184d
Introducing statistics publishing interfaces and factory classes
Project: http://git-wip-us.apache.org/repos/asf/stratos/repo
Commit: http://git-wip-us.apache.org/repos/asf/stratos/commit/dc863184
Tree: http://git-wip-us.apache.org/repos/asf/stratos/tree/dc863184
Diff: http://git-wip-us.apache.org/repos/asf/stratos/diff/dc863184
Branch: refs/heads/master
Commit: dc863184d6e70acf4deb31c311f246cccc3e363f
Parents: 3bf78f6
Author: Imesh Gunaratne <im...@apache.org>
Authored: Sun Feb 8 17:20:55 2015 +0530
Committer: Imesh Gunaratne <im...@apache.org>
Committed: Sun Feb 8 17:20:55 2015 +0530
----------------------------------------------------------------------
.../publisher/HealthStatisticsNotifier.java | 10 +-
.../publisher/HealthStatisticsPublisher.java | 96 -------------------
.../publisher/HealthStatisticsPublisher.java | 39 ++++++++
.../HealthStatisticsPublisherFactory.java | 36 +++++++
.../publisher/InFlightRequestPublisher.java | 36 +++++++
.../InFlightRequestPublisherFactory.java | 36 +++++++
.../publisher/StatisticsPublisherType.java | 27 ++++++
.../cep/WSO2CEPHealthStatisticsPublisher.java | 98 ++++++++++++++++++++
.../cep/WSO2CEPInFlightRequestPublisher.java | 86 +++++++++++++++++
.../LoadBalancerStatisticsNotifier.java | 9 +-
.../publisher/WSO2CEPFaultyMemberPublisher.java | 75 ---------------
.../WSO2CEPInFlightRequestPublisher.java | 82 ----------------
.../publisher/MockHealthStatisticsNotifier.java | 10 +-
.../MockHealthStatisticsPublisher.java | 93 -------------------
14 files changed, 376 insertions(+), 357 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/stratos/blob/dc863184/components/org.apache.stratos.cartridge.agent/src/main/java/org/apache/stratos/cartridge/agent/statistics/publisher/HealthStatisticsNotifier.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.cartridge.agent/src/main/java/org/apache/stratos/cartridge/agent/statistics/publisher/HealthStatisticsNotifier.java b/components/org.apache.stratos.cartridge.agent/src/main/java/org/apache/stratos/cartridge/agent/statistics/publisher/HealthStatisticsNotifier.java
index 6e8ebe3..04af668 100644
--- a/components/org.apache.stratos.cartridge.agent/src/main/java/org/apache/stratos/cartridge/agent/statistics/publisher/HealthStatisticsNotifier.java
+++ b/components/org.apache.stratos.cartridge.agent/src/main/java/org/apache/stratos/cartridge/agent/statistics/publisher/HealthStatisticsNotifier.java
@@ -19,14 +19,13 @@
package org.apache.stratos.cartridge.agent.statistics.publisher;
-import org.apache.stratos.cartridge.agent.statistics.publisher.*;
-
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.stratos.cartridge.agent.config.CartridgeAgentConfiguration;
-import org.apache.stratos.cartridge.agent.event.publisher.CartridgeAgentEventPublisher;
import org.apache.stratos.cartridge.agent.util.CartridgeAgentConstants;
-import org.apache.stratos.cartridge.agent.util.CartridgeAgentUtils;
+import org.apache.stratos.common.statistics.publisher.HealthStatisticsPublisher;
+import org.apache.stratos.common.statistics.publisher.HealthStatisticsPublisherFactory;
+import org.apache.stratos.common.statistics.publisher.StatisticsPublisherType;
import java.io.File;
import java.util.List;
@@ -43,7 +42,8 @@ public class HealthStatisticsNotifier implements Runnable {
private boolean terminated;
public HealthStatisticsNotifier() {
- this.statsPublisher = new HealthStatisticsPublisher();
+ this.statsPublisher = HealthStatisticsPublisherFactory.createHealthStatisticsPublisher(
+ StatisticsPublisherType.WSO2CEP);
/* Find all jars in the current working directory */
String pluginFileName = System.getProperty("health.stats.reader.plugin");
http://git-wip-us.apache.org/repos/asf/stratos/blob/dc863184/components/org.apache.stratos.cartridge.agent/src/main/java/org/apache/stratos/cartridge/agent/statistics/publisher/HealthStatisticsPublisher.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.cartridge.agent/src/main/java/org/apache/stratos/cartridge/agent/statistics/publisher/HealthStatisticsPublisher.java b/components/org.apache.stratos.cartridge.agent/src/main/java/org/apache/stratos/cartridge/agent/statistics/publisher/HealthStatisticsPublisher.java
deleted file mode 100644
index 14f7493..0000000
--- a/components/org.apache.stratos.cartridge.agent/src/main/java/org/apache/stratos/cartridge/agent/statistics/publisher/HealthStatisticsPublisher.java
+++ /dev/null
@@ -1,96 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.stratos.cartridge.agent.statistics.publisher;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.stratos.common.statistics.publisher.wso2.cep.WSO2CEPStatisticsPublisher;
-import org.wso2.carbon.databridge.commons.Attribute;
-import org.wso2.carbon.databridge.commons.AttributeType;
-import org.wso2.carbon.databridge.commons.StreamDefinition;
-
-import java.util.ArrayList;
-import java.util.List;
-
-/**
- * Health statistics publisher for publishing statistics to CEP.
- */
-public class HealthStatisticsPublisher extends WSO2CEPStatisticsPublisher {
- private static final Log log = LogFactory.getLog(HealthStatisticsPublisher.class);
-
- private static final String DATA_STREAM_NAME = "cartridge_agent_health_stats";
- private static final String VERSION = "1.0.0";
-
- private static StreamDefinition createStreamDefinition() {
- try {
- // Create stream definition
- StreamDefinition streamDefinition = new StreamDefinition(DATA_STREAM_NAME, VERSION);
- streamDefinition.setNickName("agent health stats");
- streamDefinition.setDescription("agent health stats");
-
- // Payload definition
- List<Attribute> payloadData = new ArrayList<Attribute>();
- payloadData.add(new Attribute("cluster_id", AttributeType.STRING));
- payloadData.add(new Attribute("cluster_instance_id", AttributeType.STRING));
- payloadData.add(new Attribute("network_partition_id", AttributeType.STRING));
- payloadData.add(new Attribute("member_id", AttributeType.STRING));
- payloadData.add(new Attribute("partition_id", AttributeType.STRING));
- payloadData.add(new Attribute("health_description", AttributeType.STRING));
- payloadData.add(new Attribute("value", AttributeType.DOUBLE));
-
- streamDefinition.setPayloadData(payloadData);
- return streamDefinition;
- } catch (Exception e) {
- throw new RuntimeException("Could not create stream definition", e);
- }
- }
-
- public HealthStatisticsPublisher() {
- super(createStreamDefinition());
- }
-
- /**
- * Publish health statistics to cep.
- * @param clusterId
- * @param clusterInstanceId
- * @param networkPartitionId
- * @param memberId
- * @param partitionId
- * @param health
- * @param value
- */
- public void publish(String clusterId, String clusterInstanceId, String networkPartitionId, String memberId, String partitionId, String health, double value) {
- if(log.isDebugEnabled()) {
- log.debug(String.format("Publishing health statistics: [cluster] %s [network-partition] %s [partition] %s [member] %s [health] %s [value] %f",
- clusterId, networkPartitionId, partitionId, memberId, health, value));
- }
- // Set payload values
- List<Object> payload = new ArrayList<Object>();
- payload.add(clusterId);
- payload.add(clusterInstanceId);
- payload.add(networkPartitionId);
- payload.add(memberId);
- payload.add(partitionId);
- payload.add(health);
- payload.add(value);
-
- super.publish(payload.toArray());
- }
-}
http://git-wip-us.apache.org/repos/asf/stratos/blob/dc863184/components/org.apache.stratos.common/src/main/java/org/apache/stratos/common/statistics/publisher/HealthStatisticsPublisher.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.common/src/main/java/org/apache/stratos/common/statistics/publisher/HealthStatisticsPublisher.java b/components/org.apache.stratos.common/src/main/java/org/apache/stratos/common/statistics/publisher/HealthStatisticsPublisher.java
new file mode 100644
index 0000000..41c98cb
--- /dev/null
+++ b/components/org.apache.stratos.common/src/main/java/org/apache/stratos/common/statistics/publisher/HealthStatisticsPublisher.java
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.stratos.common.statistics.publisher;
+
+/**
+ * Health statistics publisher interface.
+ */
+public interface HealthStatisticsPublisher extends StatisticsPublisher {
+
+ /**
+ * Publish health statistics to complex event processor.
+ * @param clusterId Cluster id of the member
+ * @param clusterInstanceId Cluster instance id of the member
+ * @param networkPartitionId Network partition id of the member
+ * @param memberId Member id
+ * @param partitionId Partition id of the member
+ * @param health Health type: memory_consumption | load_average
+ * @param value Health type value
+ */
+ void publish(String clusterId, String clusterInstanceId, String networkPartitionId,
+ String memberId, String partitionId, String health, double value);
+}
http://git-wip-us.apache.org/repos/asf/stratos/blob/dc863184/components/org.apache.stratos.common/src/main/java/org/apache/stratos/common/statistics/publisher/HealthStatisticsPublisherFactory.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.common/src/main/java/org/apache/stratos/common/statistics/publisher/HealthStatisticsPublisherFactory.java b/components/org.apache.stratos.common/src/main/java/org/apache/stratos/common/statistics/publisher/HealthStatisticsPublisherFactory.java
new file mode 100644
index 0000000..c0795a7
--- /dev/null
+++ b/components/org.apache.stratos.common/src/main/java/org/apache/stratos/common/statistics/publisher/HealthStatisticsPublisherFactory.java
@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.stratos.common.statistics.publisher;
+
+import org.apache.stratos.common.statistics.publisher.wso2.cep.WSO2CEPHealthStatisticsPublisher;
+
+/**
+ * Health statistics publisher factory.
+ */
+public class HealthStatisticsPublisherFactory {
+
+ public static HealthStatisticsPublisher createHealthStatisticsPublisher(StatisticsPublisherType type) {
+ if(type == StatisticsPublisherType.WSO2CEP) {
+ return new WSO2CEPHealthStatisticsPublisher();
+ } else {
+ throw new RuntimeException("Unknown statistics publisher type");
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/stratos/blob/dc863184/components/org.apache.stratos.common/src/main/java/org/apache/stratos/common/statistics/publisher/InFlightRequestPublisher.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.common/src/main/java/org/apache/stratos/common/statistics/publisher/InFlightRequestPublisher.java b/components/org.apache.stratos.common/src/main/java/org/apache/stratos/common/statistics/publisher/InFlightRequestPublisher.java
new file mode 100644
index 0000000..561173f
--- /dev/null
+++ b/components/org.apache.stratos.common/src/main/java/org/apache/stratos/common/statistics/publisher/InFlightRequestPublisher.java
@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.stratos.common.statistics.publisher;
+
+/**
+ * In-flight request publisher interface.
+ */
+public interface InFlightRequestPublisher extends StatisticsPublisher {
+
+ /**
+ * Publish in-flight request count.
+ * @param clusterId Cluster id
+ * @param networkPartitionId Network partition id of the cluster
+ * @param activeInstancesCount Active member instance count of the cluster
+ * @param inFlightRequestCount In-flight request count of the cluster
+ * @param servedRequestCount Served request count of the cluster
+ */
+ void publish(String clusterId, String networkPartitionId, int activeInstancesCount, int inFlightRequestCount, int servedRequestCount);
+}
http://git-wip-us.apache.org/repos/asf/stratos/blob/dc863184/components/org.apache.stratos.common/src/main/java/org/apache/stratos/common/statistics/publisher/InFlightRequestPublisherFactory.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.common/src/main/java/org/apache/stratos/common/statistics/publisher/InFlightRequestPublisherFactory.java b/components/org.apache.stratos.common/src/main/java/org/apache/stratos/common/statistics/publisher/InFlightRequestPublisherFactory.java
new file mode 100644
index 0000000..10b265a
--- /dev/null
+++ b/components/org.apache.stratos.common/src/main/java/org/apache/stratos/common/statistics/publisher/InFlightRequestPublisherFactory.java
@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.stratos.common.statistics.publisher;
+
+import org.apache.stratos.common.statistics.publisher.wso2.cep.WSO2CEPInFlightRequestPublisher;
+
+/**
+ * In-flight request publisher factory.
+ */
+public class InFlightRequestPublisherFactory {
+
+ public static InFlightRequestPublisher createInFlightRequestPublisher(StatisticsPublisherType type) {
+ if(type == StatisticsPublisherType.WSO2CEP) {
+ return new WSO2CEPInFlightRequestPublisher();
+ } else {
+ throw new RuntimeException("Unknown statistics publisher type");
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/stratos/blob/dc863184/components/org.apache.stratos.common/src/main/java/org/apache/stratos/common/statistics/publisher/StatisticsPublisherType.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.common/src/main/java/org/apache/stratos/common/statistics/publisher/StatisticsPublisherType.java b/components/org.apache.stratos.common/src/main/java/org/apache/stratos/common/statistics/publisher/StatisticsPublisherType.java
new file mode 100644
index 0000000..77c5b78
--- /dev/null
+++ b/components/org.apache.stratos.common/src/main/java/org/apache/stratos/common/statistics/publisher/StatisticsPublisherType.java
@@ -0,0 +1,27 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.stratos.common.statistics.publisher;
+
+/**
+ * Statistics publisher type enumneration.
+ */
+public enum StatisticsPublisherType {
+ WSO2CEP
+}
http://git-wip-us.apache.org/repos/asf/stratos/blob/dc863184/components/org.apache.stratos.common/src/main/java/org/apache/stratos/common/statistics/publisher/wso2/cep/WSO2CEPHealthStatisticsPublisher.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.common/src/main/java/org/apache/stratos/common/statistics/publisher/wso2/cep/WSO2CEPHealthStatisticsPublisher.java b/components/org.apache.stratos.common/src/main/java/org/apache/stratos/common/statistics/publisher/wso2/cep/WSO2CEPHealthStatisticsPublisher.java
new file mode 100644
index 0000000..85319d7
--- /dev/null
+++ b/components/org.apache.stratos.common/src/main/java/org/apache/stratos/common/statistics/publisher/wso2/cep/WSO2CEPHealthStatisticsPublisher.java
@@ -0,0 +1,98 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.stratos.common.statistics.publisher.wso2.cep;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.stratos.common.statistics.publisher.HealthStatisticsPublisher;
+import org.wso2.carbon.databridge.commons.Attribute;
+import org.wso2.carbon.databridge.commons.AttributeType;
+import org.wso2.carbon.databridge.commons.StreamDefinition;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Health statistics publisher for publishing statistics to WSO2 CEP.
+ */
+public class WSO2CEPHealthStatisticsPublisher extends WSO2CEPStatisticsPublisher implements HealthStatisticsPublisher {
+
+ private static final Log log = LogFactory.getLog(WSO2CEPHealthStatisticsPublisher.class);
+
+ private static final String DATA_STREAM_NAME = "cartridge_agent_health_stats";
+ private static final String VERSION = "1.0.0";
+
+ public WSO2CEPHealthStatisticsPublisher() {
+ super(createStreamDefinition());
+ }
+
+ private static StreamDefinition createStreamDefinition() {
+ try {
+ // Create stream definition
+ StreamDefinition streamDefinition = new StreamDefinition(DATA_STREAM_NAME, VERSION);
+ streamDefinition.setNickName("agent health stats");
+ streamDefinition.setDescription("agent health stats");
+
+ // Set payload definition
+ List<Attribute> payloadData = new ArrayList<Attribute>();
+ payloadData.add(new Attribute("cluster_id", AttributeType.STRING));
+ payloadData.add(new Attribute("cluster_instance_id", AttributeType.STRING));
+ payloadData.add(new Attribute("network_partition_id", AttributeType.STRING));
+ payloadData.add(new Attribute("member_id", AttributeType.STRING));
+ payloadData.add(new Attribute("partition_id", AttributeType.STRING));
+ payloadData.add(new Attribute("health_description", AttributeType.STRING));
+ payloadData.add(new Attribute("value", AttributeType.DOUBLE));
+
+ streamDefinition.setPayloadData(payloadData);
+ return streamDefinition;
+ } catch (Exception e) {
+ throw new RuntimeException("Could not create stream definition", e);
+ }
+ }
+
+ /**
+ * Publish health statistics to cep.
+ * @param clusterId
+ * @param clusterInstanceId
+ * @param networkPartitionId
+ * @param memberId
+ * @param partitionId
+ * @param health
+ * @param value
+ */
+ @Override
+ public void publish(String clusterId, String clusterInstanceId, String networkPartitionId, String memberId, String partitionId, String health, double value) {
+ if(log.isDebugEnabled()) {
+ log.debug(String.format("Publishing health statistics: [cluster] %s [network-partition] %s [partition] %s [member] %s [health] %s [value] %f",
+ clusterId, networkPartitionId, partitionId, memberId, health, value));
+ }
+ // Set payload values
+ List<Object> payload = new ArrayList<Object>();
+ payload.add(clusterId);
+ payload.add(clusterInstanceId);
+ payload.add(networkPartitionId);
+ payload.add(memberId);
+ payload.add(partitionId);
+ payload.add(health);
+ payload.add(value);
+
+ super.publish(payload.toArray());
+ }
+}
http://git-wip-us.apache.org/repos/asf/stratos/blob/dc863184/components/org.apache.stratos.common/src/main/java/org/apache/stratos/common/statistics/publisher/wso2/cep/WSO2CEPInFlightRequestPublisher.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.common/src/main/java/org/apache/stratos/common/statistics/publisher/wso2/cep/WSO2CEPInFlightRequestPublisher.java b/components/org.apache.stratos.common/src/main/java/org/apache/stratos/common/statistics/publisher/wso2/cep/WSO2CEPInFlightRequestPublisher.java
new file mode 100644
index 0000000..1a901df
--- /dev/null
+++ b/components/org.apache.stratos.common/src/main/java/org/apache/stratos/common/statistics/publisher/wso2/cep/WSO2CEPInFlightRequestPublisher.java
@@ -0,0 +1,86 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.stratos.common.statistics.publisher.wso2.cep;
+
+import org.apache.stratos.common.statistics.publisher.InFlightRequestPublisher;
+import org.wso2.carbon.databridge.commons.Attribute;
+import org.wso2.carbon.databridge.commons.AttributeType;
+import org.wso2.carbon.databridge.commons.StreamDefinition;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * WSO2 CEP in flight request count publisher.
+ * <p/>
+ * In-flight request count:
+ * Number of requests being served at a given moment could be identified as in-flight request count.
+ */
+public class WSO2CEPInFlightRequestPublisher extends WSO2CEPStatisticsPublisher implements InFlightRequestPublisher {
+
+ private static final String DATA_STREAM_NAME = "in_flight_requests";
+ private static final String VERSION = "1.0.0";
+
+ public WSO2CEPInFlightRequestPublisher() {
+ super(createStreamDefinition());
+ }
+
+ private static StreamDefinition createStreamDefinition() {
+ try {
+ // Create stream definition
+ StreamDefinition streamDefinition = new StreamDefinition(DATA_STREAM_NAME, VERSION);
+ streamDefinition.setNickName("lb stats");
+ streamDefinition.setDescription("lb stats");
+ List<Attribute> payloadData = new ArrayList<Attribute>();
+
+ // Set payload definition
+ payloadData.add(new Attribute("cluster_id", AttributeType.STRING));
+ payloadData.add(new Attribute("network_partition_id", AttributeType.STRING));
+ payloadData.add(new Attribute("active_instances_count", AttributeType.DOUBLE));
+ payloadData.add(new Attribute("in_flight_request_count", AttributeType.DOUBLE));
+ payloadData.add(new Attribute("served_request_count", AttributeType.DOUBLE));
+ streamDefinition.setPayloadData(payloadData);
+ return streamDefinition;
+ } catch (Exception e) {
+ throw new RuntimeException("Could not create stream definition", e);
+ }
+ }
+
+ /**
+ * Publish in-flight request count of a cluster.
+ *
+ * @param clusterId
+ * @param networkPartitionId
+ * @param inFlightRequestCount
+ * @param servedRequestCount
+ */
+ @Override
+ public void publish(String clusterId, String networkPartitionId, int activeInstancesCount, int inFlightRequestCount, int servedRequestCount) {
+ // Set payload values
+ List<Object> payload = new ArrayList<Object>();
+ payload.add(clusterId);
+ payload.add(networkPartitionId);
+ payload.add((double)activeInstancesCount);
+ payload.add((double)inFlightRequestCount);
+ payload.add((double)servedRequestCount);
+
+ super.publish(payload.toArray());
+ }
+}
http://git-wip-us.apache.org/repos/asf/stratos/blob/dc863184/components/org.apache.stratos.load.balancer.common/src/main/java/org/apache/stratos/load/balancer/common/statistics/notifier/LoadBalancerStatisticsNotifier.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.load.balancer.common/src/main/java/org/apache/stratos/load/balancer/common/statistics/notifier/LoadBalancerStatisticsNotifier.java b/components/org.apache.stratos.load.balancer.common/src/main/java/org/apache/stratos/load/balancer/common/statistics/notifier/LoadBalancerStatisticsNotifier.java
index 7d123e5..cf3b27c 100644
--- a/components/org.apache.stratos.load.balancer.common/src/main/java/org/apache/stratos/load/balancer/common/statistics/notifier/LoadBalancerStatisticsNotifier.java
+++ b/components/org.apache.stratos.load.balancer.common/src/main/java/org/apache/stratos/load/balancer/common/statistics/notifier/LoadBalancerStatisticsNotifier.java
@@ -22,8 +22,10 @@ package org.apache.stratos.load.balancer.common.statistics.notifier;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
+import org.apache.stratos.common.statistics.publisher.InFlightRequestPublisher;
+import org.apache.stratos.common.statistics.publisher.InFlightRequestPublisherFactory;
+import org.apache.stratos.common.statistics.publisher.StatisticsPublisherType;
import org.apache.stratos.load.balancer.common.statistics.LoadBalancerStatisticsReader;
-import org.apache.stratos.load.balancer.common.statistics.publisher.WSO2CEPInFlightRequestPublisher;
import org.apache.stratos.messaging.domain.topology.Cluster;
import org.apache.stratos.messaging.domain.topology.Service;
import org.apache.stratos.messaging.message.receiver.topology.TopologyManager;
@@ -35,14 +37,15 @@ public class LoadBalancerStatisticsNotifier implements Runnable {
private static final Log log = LogFactory.getLog(LoadBalancerStatisticsNotifier.class);
private final LoadBalancerStatisticsReader statsReader;
- private final WSO2CEPInFlightRequestPublisher inFlightRequestPublisher;
+ private final InFlightRequestPublisher inFlightRequestPublisher;
private long statsPublisherInterval = 15000;
private String networkPartitionId;
private boolean terminated;
public LoadBalancerStatisticsNotifier(LoadBalancerStatisticsReader statsReader) {
this.statsReader = statsReader;
- this.inFlightRequestPublisher = new WSO2CEPInFlightRequestPublisher();
+ this.inFlightRequestPublisher = InFlightRequestPublisherFactory.createInFlightRequestPublisher(
+ StatisticsPublisherType.WSO2CEP);
String interval = System.getProperty("stats.notifier.interval");
if (interval != null) {
http://git-wip-us.apache.org/repos/asf/stratos/blob/dc863184/components/org.apache.stratos.load.balancer.common/src/main/java/org/apache/stratos/load/balancer/common/statistics/publisher/WSO2CEPFaultyMemberPublisher.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.load.balancer.common/src/main/java/org/apache/stratos/load/balancer/common/statistics/publisher/WSO2CEPFaultyMemberPublisher.java b/components/org.apache.stratos.load.balancer.common/src/main/java/org/apache/stratos/load/balancer/common/statistics/publisher/WSO2CEPFaultyMemberPublisher.java
deleted file mode 100644
index 1a29813..0000000
--- a/components/org.apache.stratos.load.balancer.common/src/main/java/org/apache/stratos/load/balancer/common/statistics/publisher/WSO2CEPFaultyMemberPublisher.java
+++ /dev/null
@@ -1,75 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.stratos.load.balancer.common.statistics.publisher;
-
-import org.apache.stratos.common.statistics.publisher.wso2.cep.WSO2CEPStatisticsPublisher;
-import org.wso2.carbon.databridge.commons.Attribute;
-import org.wso2.carbon.databridge.commons.AttributeType;
-import org.wso2.carbon.databridge.commons.StreamDefinition;
-
-import java.util.ArrayList;
-import java.util.List;
-
-/**
- * WSO2 CEP faulty member publisher.
- * <p/>
- * Faulty members:
- * If a request was rejected by some of the members in a cluster while at least
- * one member could serve it, those members could be identified as faulty.
- */
-public class WSO2CEPFaultyMemberPublisher extends WSO2CEPStatisticsPublisher {
-
- private static final String DATA_STREAM_NAME = "stratos.lb.faulty.members";
- private static final String VERSION = "1.0.0";
-
- private static StreamDefinition createStreamDefinition() {
- try {
- StreamDefinition streamDefinition = new StreamDefinition(DATA_STREAM_NAME, VERSION);
- streamDefinition.setNickName("lb fault members");
- streamDefinition.setDescription("lb fault members");
- List<Attribute> payloadData = new ArrayList<Attribute>();
- // Payload definition
- payloadData.add(new Attribute("cluster_id", AttributeType.STRING));
- payloadData.add(new Attribute("member_id", AttributeType.STRING));
- streamDefinition.setPayloadData(payloadData);
- return streamDefinition;
- } catch (Exception e) {
- throw new RuntimeException("Could not create stream definition", e);
- }
- }
-
- public WSO2CEPFaultyMemberPublisher() {
- super(createStreamDefinition());
- }
-
- /**
- * Publish faulty members.
- *
- * @param clusterId
- * @param memberId
- */
- public void publish(String clusterId, String memberId) {
- List<Object> payload = new ArrayList<Object>();
- // Payload values
- payload.add(clusterId);
- payload.add(memberId);
- super.publish(payload.toArray());
- }
-}
http://git-wip-us.apache.org/repos/asf/stratos/blob/dc863184/components/org.apache.stratos.load.balancer.common/src/main/java/org/apache/stratos/load/balancer/common/statistics/publisher/WSO2CEPInFlightRequestPublisher.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.load.balancer.common/src/main/java/org/apache/stratos/load/balancer/common/statistics/publisher/WSO2CEPInFlightRequestPublisher.java b/components/org.apache.stratos.load.balancer.common/src/main/java/org/apache/stratos/load/balancer/common/statistics/publisher/WSO2CEPInFlightRequestPublisher.java
deleted file mode 100644
index 7a79015..0000000
--- a/components/org.apache.stratos.load.balancer.common/src/main/java/org/apache/stratos/load/balancer/common/statistics/publisher/WSO2CEPInFlightRequestPublisher.java
+++ /dev/null
@@ -1,82 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.stratos.load.balancer.common.statistics.publisher;
-
-import org.apache.stratos.common.statistics.publisher.wso2.cep.WSO2CEPStatisticsPublisher;
-import org.wso2.carbon.databridge.commons.Attribute;
-import org.wso2.carbon.databridge.commons.AttributeType;
-import org.wso2.carbon.databridge.commons.StreamDefinition;
-
-import java.util.ArrayList;
-import java.util.List;
-
-/**
- * WSO2 CEP in flight request count publisher.
- * <p/>
- * In-flight request count:
- * Number of requests being served at a given moment could be identified as in-flight request count.
- */
-public class WSO2CEPInFlightRequestPublisher extends WSO2CEPStatisticsPublisher {
-
- private static final String DATA_STREAM_NAME = "in_flight_requests";
- private static final String VERSION = "1.0.0";
-
- private static StreamDefinition createStreamDefinition() {
- try {
- StreamDefinition streamDefinition = new StreamDefinition(DATA_STREAM_NAME, VERSION);
- streamDefinition.setNickName("lb stats");
- streamDefinition.setDescription("lb stats");
- List<Attribute> payloadData = new ArrayList<Attribute>();
- // Payload definition
- payloadData.add(new Attribute("cluster_id", AttributeType.STRING));
- payloadData.add(new Attribute("network_partition_id", AttributeType.STRING));
- payloadData.add(new Attribute("active_instances_count", AttributeType.DOUBLE));
- payloadData.add(new Attribute("in_flight_request_count", AttributeType.DOUBLE));
- payloadData.add(new Attribute("served_request_count", AttributeType.DOUBLE));
- streamDefinition.setPayloadData(payloadData);
- return streamDefinition;
- } catch (Exception e) {
- throw new RuntimeException("Could not create stream definition", e);
- }
- }
-
- public WSO2CEPInFlightRequestPublisher() {
- super(createStreamDefinition());
- }
-
- /**
- * Publish in-flight request count of a cluster.
- *
- * @param clusterId
- * @param networkPartitionId
- * @param inFlightRequestCount
- * @param servedRequestCount
- */
- public void publish(String clusterId, String networkPartitionId,int activeInstancesCount, int inFlightRequestCount, int servedRequestCount) {
- List<Object> payload = new ArrayList<Object>();
- // Payload values
- payload.add(clusterId);
- payload.add(networkPartitionId);
- payload.add((double)activeInstancesCount);
- payload.add((double)inFlightRequestCount);
- payload.add((double)servedRequestCount);
- super.publish(payload.toArray());
- }
-}
http://git-wip-us.apache.org/repos/asf/stratos/blob/dc863184/components/org.apache.stratos.mock.iaas/src/main/java/org/apache/stratos/mock/iaas/statistics/publisher/MockHealthStatisticsNotifier.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.mock.iaas/src/main/java/org/apache/stratos/mock/iaas/statistics/publisher/MockHealthStatisticsNotifier.java b/components/org.apache.stratos.mock.iaas/src/main/java/org/apache/stratos/mock/iaas/statistics/publisher/MockHealthStatisticsNotifier.java
index fda5cc1..3060ae4 100644
--- a/components/org.apache.stratos.mock.iaas/src/main/java/org/apache/stratos/mock/iaas/statistics/publisher/MockHealthStatisticsNotifier.java
+++ b/components/org.apache.stratos.mock.iaas/src/main/java/org/apache/stratos/mock/iaas/statistics/publisher/MockHealthStatisticsNotifier.java
@@ -21,9 +21,12 @@ package org.apache.stratos.mock.iaas.statistics.publisher;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
-import org.apache.stratos.mock.iaas.services.impl.MockAutoscalingFactor;
+import org.apache.stratos.common.statistics.publisher.HealthStatisticsPublisher;
+import org.apache.stratos.common.statistics.publisher.HealthStatisticsPublisherFactory;
+import org.apache.stratos.common.statistics.publisher.StatisticsPublisherType;
import org.apache.stratos.mock.iaas.domain.MockInstanceContext;
import org.apache.stratos.mock.iaas.exceptions.NoStatisticsFoundException;
+import org.apache.stratos.mock.iaas.services.impl.MockAutoscalingFactor;
import org.apache.stratos.mock.iaas.statistics.MockHealthStatistics;
/**
@@ -36,11 +39,12 @@ public class MockHealthStatisticsNotifier implements Runnable {
public static final String LOAD_AVERAGE = "load_average";
private final MockInstanceContext mockMemberContext;
- private final MockHealthStatisticsPublisher statsPublisher;
+ private final HealthStatisticsPublisher statsPublisher;
public MockHealthStatisticsNotifier(MockInstanceContext mockMemberContext) {
this.mockMemberContext = mockMemberContext;
- this.statsPublisher = new MockHealthStatisticsPublisher();
+ this.statsPublisher = HealthStatisticsPublisherFactory.createHealthStatisticsPublisher(
+ StatisticsPublisherType.WSO2CEP);
this.statsPublisher.setEnabled(true);
}
http://git-wip-us.apache.org/repos/asf/stratos/blob/dc863184/components/org.apache.stratos.mock.iaas/src/main/java/org/apache/stratos/mock/iaas/statistics/publisher/MockHealthStatisticsPublisher.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.mock.iaas/src/main/java/org/apache/stratos/mock/iaas/statistics/publisher/MockHealthStatisticsPublisher.java b/components/org.apache.stratos.mock.iaas/src/main/java/org/apache/stratos/mock/iaas/statistics/publisher/MockHealthStatisticsPublisher.java
deleted file mode 100644
index caab566..0000000
--- a/components/org.apache.stratos.mock.iaas/src/main/java/org/apache/stratos/mock/iaas/statistics/publisher/MockHealthStatisticsPublisher.java
+++ /dev/null
@@ -1,93 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.stratos.mock.iaas.statistics.publisher;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.stratos.common.statistics.publisher.wso2.cep.WSO2CEPStatisticsPublisher;
-import org.wso2.carbon.databridge.commons.Attribute;
-import org.wso2.carbon.databridge.commons.AttributeType;
-import org.wso2.carbon.databridge.commons.StreamDefinition;
-
-import java.util.ArrayList;
-import java.util.List;
-
-/**
- * Health statistics publisher for publishing statistics to CEP.
- */
-public class MockHealthStatisticsPublisher extends WSO2CEPStatisticsPublisher {
- private static final Log log = LogFactory.getLog(MockHealthStatisticsPublisher.class);
-
- private static final String DATA_STREAM_NAME = "cartridge_agent_health_stats";
- private static final String VERSION = "1.0.0";
-
- private static StreamDefinition createStreamDefinition() {
- try {
- // Create stream definition
- StreamDefinition streamDefinition = new StreamDefinition(DATA_STREAM_NAME, VERSION);
- streamDefinition.setNickName("agent health stats");
- streamDefinition.setDescription("agent health stats");
-
- // Payload definition
- List<Attribute> payloadData = new ArrayList<Attribute>();
- payloadData.add(new Attribute("cluster_id", AttributeType.STRING));
- payloadData.add(new Attribute("cluster_instance_id", AttributeType.STRING));
- payloadData.add(new Attribute("network_partition_id", AttributeType.STRING));
- payloadData.add(new Attribute("member_id", AttributeType.STRING));
- payloadData.add(new Attribute("partition_id", AttributeType.STRING));
- payloadData.add(new Attribute("health_description", AttributeType.STRING));
- payloadData.add(new Attribute("value", AttributeType.DOUBLE));
- streamDefinition.setPayloadData(payloadData);
- return streamDefinition;
- } catch (Exception e) {
- throw new RuntimeException("Could not create stream definition", e);
- }
- }
-
- public MockHealthStatisticsPublisher() {
- super(createStreamDefinition());
- }
-
- /**
- * Publish health statistics to cep.
- * @param clusterId
- * @param networkPartitionId
- * @param memberId
- * @param partitionId
- * @param health
- * @param value
- */
- public void publish(String clusterId, String clusterInstanceId, String networkPartitionId, String memberId, String partitionId, String health, double value) {
- if(log.isDebugEnabled()) {
- log.debug(String.format("Publishing health statistics: [cluster] %s [network-partition] %s [partition] %s [member] %s [health] %s [value] %f",
- clusterId, networkPartitionId, partitionId, memberId, health, value));
- }
- // Set payload values
- List<Object> payload = new ArrayList<Object>();
- payload.add(clusterId);
- payload.add(clusterInstanceId);
- payload.add(networkPartitionId);
- payload.add(memberId);
- payload.add(partitionId);
- payload.add(health);
- payload.add(value);
- super.publish(payload.toArray());
- }
-}