You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ambari.apache.org by sw...@apache.org on 2014/09/22 20:02:25 UTC
[15/22] AMBARI-5707. Metrics system prototype implementation. (swagle)
http://git-wip-us.apache.org/repos/asf/ambari/blob/865d187e/ambari-metrics/ambari-metrics-hadoop-timelineservice/src/test/java/org/apache/hadoop/yarn/server/applicationhistoryservice/webapp/TestAHSWebServices.java
----------------------------------------------------------------------
diff --git a/ambari-metrics/ambari-metrics-hadoop-timelineservice/src/test/java/org/apache/hadoop/yarn/server/applicationhistoryservice/webapp/TestAHSWebServices.java b/ambari-metrics/ambari-metrics-hadoop-timelineservice/src/test/java/org/apache/hadoop/yarn/server/applicationhistoryservice/webapp/TestAHSWebServices.java
new file mode 100644
index 0000000..7ca5a03
--- /dev/null
+++ b/ambari-metrics/ambari-metrics-hadoop-timelineservice/src/test/java/org/apache/hadoop/yarn/server/applicationhistoryservice/webapp/TestAHSWebServices.java
@@ -0,0 +1,303 @@
+/**
+ * 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.hadoop.yarn.server.applicationhistoryservice.webapp;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+import javax.ws.rs.core.MediaType;
+
+import junit.framework.Assert;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
+import org.apache.hadoop.yarn.api.records.ApplicationId;
+import org.apache.hadoop.yarn.api.records.ContainerId;
+import org.apache.hadoop.yarn.api.records.ContainerState;
+import org.apache.hadoop.yarn.api.records.FinalApplicationStatus;
+import org.apache.hadoop.yarn.api.records.NodeId;
+import org.apache.hadoop.yarn.api.records.Priority;
+import org.apache.hadoop.yarn.api.records.YarnApplicationAttemptState;
+import org.apache.hadoop.yarn.api.records.YarnApplicationState;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.hadoop.yarn.server.api.ApplicationContext;
+import org.apache.hadoop.yarn.server.applicationhistoryservice.ApplicationHistoryManager;
+import org.apache.hadoop.yarn.server.applicationhistoryservice.ApplicationHistoryStore;
+import org.apache.hadoop.yarn.server.applicationhistoryservice.MemoryApplicationHistoryStore;
+import org.apache.hadoop.yarn.webapp.GenericExceptionHandler;
+import org.apache.hadoop.yarn.webapp.WebServicesTestUtils;
+import org.apache.hadoop.yarn.webapp.util.WebAppUtils;
+import org.codehaus.jettison.json.JSONArray;
+import org.codehaus.jettison.json.JSONException;
+import org.codehaus.jettison.json.JSONObject;
+import org.junit.Before;
+import org.junit.Test;
+
+import com.google.inject.Guice;
+import com.google.inject.Injector;
+import com.google.inject.servlet.GuiceServletContextListener;
+import com.google.inject.servlet.ServletModule;
+import com.sun.jersey.api.client.ClientResponse;
+import com.sun.jersey.api.client.ClientResponse.Status;
+import com.sun.jersey.api.client.UniformInterfaceException;
+import com.sun.jersey.api.client.WebResource;
+import com.sun.jersey.guice.spi.container.servlet.GuiceContainer;
+import com.sun.jersey.test.framework.JerseyTest;
+import com.sun.jersey.test.framework.WebAppDescriptor;
+
+public class TestAHSWebServices extends JerseyTest {
+
+ private static ApplicationHistoryManager ahManager;
+
+ private Injector injector = Guice.createInjector(new ServletModule() {
+
+ @Override
+ protected void configureServlets() {
+ bind(JAXBContextResolver.class);
+ bind(AHSWebServices.class);
+ bind(GenericExceptionHandler.class);
+ try {
+ ahManager = mockApplicationHistoryManager();
+ } catch (Exception e) {
+ Assert.fail();
+ }
+ bind(ApplicationContext.class).toInstance(ahManager);
+ serve("/*").with(GuiceContainer.class);
+ }
+ });
+
+ public class GuiceServletConfig extends GuiceServletContextListener {
+
+ @Override
+ protected Injector getInjector() {
+ return injector;
+ }
+ }
+
+ private ApplicationHistoryManager mockApplicationHistoryManager()
+ throws Exception {
+ ApplicationHistoryStore store = new MemoryApplicationHistoryStore();
+ TestAHSWebApp testAHSWebApp = new TestAHSWebApp();
+ testAHSWebApp.setApplicationHistoryStore(store);
+ ApplicationHistoryManager ahManager =
+ testAHSWebApp.mockApplicationHistoryManager(5, 5, 5);
+ return ahManager;
+ }
+
+ public TestAHSWebServices() {
+ super(new WebAppDescriptor.Builder(
+ "org.apache.hadoop.yarn.server.applicationhistoryservice.webapp")
+ .contextListenerClass(GuiceServletConfig.class)
+ .filterClass(com.google.inject.servlet.GuiceFilter.class)
+ .contextPath("jersey-guice-filter").servletPath("/").build());
+ }
+
+ @Before
+ @Override
+ public void setUp() throws Exception {
+ super.setUp();
+ }
+
+ @Test
+ public void testInvalidUri() throws JSONException, Exception {
+ WebResource r = resource();
+ String responseStr = "";
+ try {
+ responseStr =
+ r.path("ws").path("v1").path("applicationhistory").path("bogus")
+ .accept(MediaType.APPLICATION_JSON).get(String.class);
+ fail("should have thrown exception on invalid uri");
+ } catch (UniformInterfaceException ue) {
+ ClientResponse response = ue.getResponse();
+ assertEquals(Status.NOT_FOUND, response.getClientResponseStatus());
+
+ WebServicesTestUtils.checkStringMatch(
+ "error string exists and shouldn't", "", responseStr);
+ }
+ }
+
+ @Test
+ public void testInvalidUri2() throws JSONException, Exception {
+ WebResource r = resource();
+ String responseStr = "";
+ try {
+ responseStr = r.accept(MediaType.APPLICATION_JSON).get(String.class);
+ fail("should have thrown exception on invalid uri");
+ } catch (UniformInterfaceException ue) {
+ ClientResponse response = ue.getResponse();
+ assertEquals(Status.NOT_FOUND, response.getClientResponseStatus());
+ WebServicesTestUtils.checkStringMatch(
+ "error string exists and shouldn't", "", responseStr);
+ }
+ }
+
+ @Test
+ public void testInvalidAccept() throws JSONException, Exception {
+ WebResource r = resource();
+ String responseStr = "";
+ try {
+ responseStr =
+ r.path("ws").path("v1").path("applicationhistory")
+ .accept(MediaType.TEXT_PLAIN).get(String.class);
+ fail("should have thrown exception on invalid uri");
+ } catch (UniformInterfaceException ue) {
+ ClientResponse response = ue.getResponse();
+ assertEquals(Status.INTERNAL_SERVER_ERROR,
+ response.getClientResponseStatus());
+ WebServicesTestUtils.checkStringMatch(
+ "error string exists and shouldn't", "", responseStr);
+ }
+ }
+
+ @Test
+ public void testAppsQuery() throws Exception {
+ WebResource r = resource();
+ ClientResponse response =
+ r.path("ws").path("v1").path("applicationhistory").path("apps")
+ .queryParam("state", YarnApplicationState.FINISHED.toString())
+ .accept(MediaType.APPLICATION_JSON).get(ClientResponse.class);
+ assertEquals(MediaType.APPLICATION_JSON_TYPE, response.getType());
+ JSONObject json = response.getEntity(JSONObject.class);
+ assertEquals("incorrect number of elements", 1, json.length());
+ JSONObject apps = json.getJSONObject("apps");
+ assertEquals("incorrect number of elements", 1, apps.length());
+ JSONArray array = apps.getJSONArray("app");
+ assertEquals("incorrect number of elements", 5, array.length());
+ }
+
+ @Test
+ public void testSingleApp() throws Exception {
+ ApplicationId appId = ApplicationId.newInstance(0, 1);
+ WebResource r = resource();
+ ClientResponse response =
+ r.path("ws").path("v1").path("applicationhistory").path("apps")
+ .path(appId.toString()).accept(MediaType.APPLICATION_JSON)
+ .get(ClientResponse.class);
+ assertEquals(MediaType.APPLICATION_JSON_TYPE, response.getType());
+ JSONObject json = response.getEntity(JSONObject.class);
+ assertEquals("incorrect number of elements", 1, json.length());
+ JSONObject app = json.getJSONObject("app");
+ assertEquals(appId.toString(), app.getString("appId"));
+ assertEquals(appId.toString(), app.get("name"));
+ assertEquals(appId.toString(), app.get("diagnosticsInfo"));
+ assertEquals("test queue", app.get("queue"));
+ assertEquals("test user", app.get("user"));
+ assertEquals("test type", app.get("type"));
+ assertEquals(FinalApplicationStatus.UNDEFINED.toString(),
+ app.get("finalAppStatus"));
+ assertEquals(YarnApplicationState.FINISHED.toString(), app.get("appState"));
+ }
+
+ @Test
+ public void testMultipleAttempts() throws Exception {
+ ApplicationId appId = ApplicationId.newInstance(0, 1);
+ WebResource r = resource();
+ ClientResponse response =
+ r.path("ws").path("v1").path("applicationhistory").path("apps")
+ .path(appId.toString()).path("appattempts")
+ .accept(MediaType.APPLICATION_JSON).get(ClientResponse.class);
+ assertEquals(MediaType.APPLICATION_JSON_TYPE, response.getType());
+ JSONObject json = response.getEntity(JSONObject.class);
+ assertEquals("incorrect number of elements", 1, json.length());
+ JSONObject appAttempts = json.getJSONObject("appAttempts");
+ assertEquals("incorrect number of elements", 1, appAttempts.length());
+ JSONArray array = appAttempts.getJSONArray("appAttempt");
+ assertEquals("incorrect number of elements", 5, array.length());
+ }
+
+ @Test
+ public void testSingleAttempt() throws Exception {
+ ApplicationId appId = ApplicationId.newInstance(0, 1);
+ ApplicationAttemptId appAttemptId =
+ ApplicationAttemptId.newInstance(appId, 1);
+ WebResource r = resource();
+ ClientResponse response =
+ r.path("ws").path("v1").path("applicationhistory").path("apps")
+ .path(appId.toString()).path("appattempts")
+ .path(appAttemptId.toString()).accept(MediaType.APPLICATION_JSON)
+ .get(ClientResponse.class);
+ assertEquals(MediaType.APPLICATION_JSON_TYPE, response.getType());
+ JSONObject json = response.getEntity(JSONObject.class);
+ assertEquals("incorrect number of elements", 1, json.length());
+ JSONObject appAttempt = json.getJSONObject("appAttempt");
+ assertEquals(appAttemptId.toString(), appAttempt.getString("appAttemptId"));
+ assertEquals(appAttemptId.toString(), appAttempt.getString("host"));
+ assertEquals(appAttemptId.toString(),
+ appAttempt.getString("diagnosticsInfo"));
+ assertEquals("test tracking url", appAttempt.getString("trackingUrl"));
+ assertEquals(YarnApplicationAttemptState.FINISHED.toString(),
+ appAttempt.get("appAttemptState"));
+ }
+
+ @Test
+ public void testMultipleContainers() throws Exception {
+ ApplicationId appId = ApplicationId.newInstance(0, 1);
+ ApplicationAttemptId appAttemptId =
+ ApplicationAttemptId.newInstance(appId, 1);
+ WebResource r = resource();
+ ClientResponse response =
+ r.path("ws").path("v1").path("applicationhistory").path("apps")
+ .path(appId.toString()).path("appattempts")
+ .path(appAttemptId.toString()).path("containers")
+ .accept(MediaType.APPLICATION_JSON).get(ClientResponse.class);
+ assertEquals(MediaType.APPLICATION_JSON_TYPE, response.getType());
+ JSONObject json = response.getEntity(JSONObject.class);
+ assertEquals("incorrect number of elements", 1, json.length());
+ JSONObject containers = json.getJSONObject("containers");
+ assertEquals("incorrect number of elements", 1, containers.length());
+ JSONArray array = containers.getJSONArray("container");
+ assertEquals("incorrect number of elements", 5, array.length());
+ }
+
+ @Test
+ public void testSingleContainer() throws Exception {
+ ApplicationId appId = ApplicationId.newInstance(0, 1);
+ ApplicationAttemptId appAttemptId =
+ ApplicationAttemptId.newInstance(appId, 1);
+ ContainerId containerId = ContainerId.newInstance(appAttemptId, 1);
+ WebResource r = resource();
+ ClientResponse response =
+ r.path("ws").path("v1").path("applicationhistory").path("apps")
+ .path(appId.toString()).path("appattempts")
+ .path(appAttemptId.toString()).path("containers")
+ .path(containerId.toString()).accept(MediaType.APPLICATION_JSON)
+ .get(ClientResponse.class);
+ assertEquals(MediaType.APPLICATION_JSON_TYPE, response.getType());
+ JSONObject json = response.getEntity(JSONObject.class);
+ assertEquals("incorrect number of elements", 1, json.length());
+ JSONObject container = json.getJSONObject("container");
+ assertEquals(containerId.toString(), container.getString("containerId"));
+ assertEquals(containerId.toString(), container.getString("diagnosticsInfo"));
+ assertEquals("0", container.getString("allocatedMB"));
+ assertEquals("0", container.getString("allocatedVCores"));
+ assertEquals(NodeId.newInstance("localhost", 0).toString(),
+ container.getString("assignedNodeId"));
+ assertEquals(Priority.newInstance(containerId.getId()).toString(),
+ container.getString("priority"));
+ Configuration conf = new YarnConfiguration();
+ assertEquals(WebAppUtils.getHttpSchemePrefix(conf) +
+ WebAppUtils.getAHSWebAppURLWithoutScheme(conf) +
+ "/applicationhistory/logs/localhost:0/container_0_0001_01_000001/" +
+ "container_0_0001_01_000001/test user",
+ container.getString("logUrl"));
+ assertEquals(ContainerState.COMPLETE.toString(),
+ container.getString("containerState"));
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/ambari/blob/865d187e/ambari-metrics/ambari-metrics-hadoop-timelineservice/src/test/java/org/apache/hadoop/yarn/server/applicationhistoryservice/webapp/TestTimelineWebServices.java
----------------------------------------------------------------------
diff --git a/ambari-metrics/ambari-metrics-hadoop-timelineservice/src/test/java/org/apache/hadoop/yarn/server/applicationhistoryservice/webapp/TestTimelineWebServices.java b/ambari-metrics/ambari-metrics-hadoop-timelineservice/src/test/java/org/apache/hadoop/yarn/server/applicationhistoryservice/webapp/TestTimelineWebServices.java
new file mode 100644
index 0000000..2b93190
--- /dev/null
+++ b/ambari-metrics/ambari-metrics-hadoop-timelineservice/src/test/java/org/apache/hadoop/yarn/server/applicationhistoryservice/webapp/TestTimelineWebServices.java
@@ -0,0 +1,391 @@
+/**
+ * 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.hadoop.yarn.server.applicationhistoryservice.webapp;
+
+import static org.junit.Assert.assertEquals;
+
+import javax.ws.rs.core.MediaType;
+
+import junit.framework.Assert;
+
+import org.apache.hadoop.metrics2.sink.timeline.TimelineMetrics;
+import org.apache.hadoop.yarn.api.records.timeline.TimelineEntities;
+import org.apache.hadoop.yarn.api.records.timeline.TimelineEntity;
+import org.apache.hadoop.yarn.api.records.timeline.TimelineEvent;
+import org.apache.hadoop.yarn.api.records.timeline.TimelineEvents;
+import org.apache.hadoop.yarn.api.records.timeline.TimelinePutResponse;
+import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TestTimelineMetricStore;
+import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricStore;
+import org.apache.hadoop.yarn.server.applicationhistoryservice.timeline.TimelineStore;
+import org.apache.hadoop.yarn.server.applicationhistoryservice.timeline.TestMemoryTimelineStore;
+import org.apache.hadoop.yarn.webapp.GenericExceptionHandler;
+import org.apache.hadoop.yarn.webapp.YarnJacksonJaxbJsonProvider;
+import org.junit.Test;
+
+import com.google.inject.Guice;
+import com.google.inject.Injector;
+import com.google.inject.servlet.GuiceServletContextListener;
+import com.google.inject.servlet.ServletModule;
+import com.sun.jersey.api.client.ClientResponse;
+import com.sun.jersey.api.client.WebResource;
+import com.sun.jersey.api.client.config.DefaultClientConfig;
+import com.sun.jersey.guice.spi.container.servlet.GuiceContainer;
+import com.sun.jersey.test.framework.JerseyTest;
+import com.sun.jersey.test.framework.WebAppDescriptor;
+
+
+public class TestTimelineWebServices extends JerseyTest {
+
+ private static TimelineStore store;
+ private static TimelineMetricStore metricStore;
+ private long beforeTime;
+
+ private Injector injector = Guice.createInjector(new ServletModule() {
+
+ @Override
+ protected void configureServlets() {
+ bind(YarnJacksonJaxbJsonProvider.class);
+ bind(TimelineWebServices.class);
+ bind(GenericExceptionHandler.class);
+ try{
+ store = mockTimelineStore();
+ metricStore = new TestTimelineMetricStore();
+ } catch (Exception e) {
+ Assert.fail();
+ }
+ bind(TimelineStore.class).toInstance(store);
+ bind(TimelineMetricStore.class).toInstance(metricStore);
+ serve("/*").with(GuiceContainer.class);
+ }
+
+ });
+
+ public class GuiceServletConfig extends GuiceServletContextListener {
+
+ @Override
+ protected Injector getInjector() {
+ return injector;
+ }
+ }
+
+ private TimelineStore mockTimelineStore()
+ throws Exception {
+ beforeTime = System.currentTimeMillis() - 1;
+ TestMemoryTimelineStore store = new TestMemoryTimelineStore();
+ store.setup();
+ return store.getTimelineStore();
+ }
+
+ public TestTimelineWebServices() {
+ super(new WebAppDescriptor.Builder(
+ "org.apache.hadoop.yarn.server.applicationhistoryservice.webapp")
+ .contextListenerClass(GuiceServletConfig.class)
+ .filterClass(com.google.inject.servlet.GuiceFilter.class)
+ .contextPath("jersey-guice-filter")
+ .servletPath("/")
+ .clientConfig(new DefaultClientConfig(YarnJacksonJaxbJsonProvider.class))
+ .build());
+ }
+
+ @Test
+ public void testAbout() throws Exception {
+ WebResource r = resource();
+ ClientResponse response = r.path("ws").path("v1").path("timeline")
+ .accept(MediaType.APPLICATION_JSON)
+ .get(ClientResponse.class);
+ assertEquals(MediaType.APPLICATION_JSON_TYPE, response.getType());
+ TimelineWebServices.AboutInfo about =
+ response.getEntity(TimelineWebServices.AboutInfo.class);
+ Assert.assertNotNull(about);
+ Assert.assertEquals("Timeline API", about.getAbout());
+ }
+
+ private static void verifyEntities(TimelineEntities entities) {
+ Assert.assertNotNull(entities);
+ Assert.assertEquals(2, entities.getEntities().size());
+ TimelineEntity entity1 = entities.getEntities().get(0);
+ Assert.assertNotNull(entity1);
+ Assert.assertEquals("id_1", entity1.getEntityId());
+ Assert.assertEquals("type_1", entity1.getEntityType());
+ Assert.assertEquals(123l, entity1.getStartTime().longValue());
+ Assert.assertEquals(2, entity1.getEvents().size());
+ Assert.assertEquals(4, entity1.getPrimaryFilters().size());
+ Assert.assertEquals(4, entity1.getOtherInfo().size());
+ TimelineEntity entity2 = entities.getEntities().get(1);
+ Assert.assertNotNull(entity2);
+ Assert.assertEquals("id_2", entity2.getEntityId());
+ Assert.assertEquals("type_1", entity2.getEntityType());
+ Assert.assertEquals(123l, entity2.getStartTime().longValue());
+ Assert.assertEquals(2, entity2.getEvents().size());
+ Assert.assertEquals(4, entity2.getPrimaryFilters().size());
+ Assert.assertEquals(4, entity2.getOtherInfo().size());
+ }
+
+ private static void verifyMetrics(TimelineMetrics metrics) {
+ Assert.assertNotNull(metrics);
+ Assert.assertEquals("cpu_user", metrics.getMetrics().get(0).getMetricName());
+ Assert.assertEquals(3, metrics.getMetrics().get(0).getMetricValues().size());
+ Assert.assertEquals("mem_free", metrics.getMetrics().get(1).getMetricName());
+ Assert.assertEquals(3, metrics.getMetrics().get(1).getMetricValues().size());
+ }
+
+ @Test
+ public void testGetEntities() throws Exception {
+ WebResource r = resource();
+ ClientResponse response = r.path("ws").path("v1").path("timeline")
+ .path("type_1")
+ .accept(MediaType.APPLICATION_JSON)
+ .get(ClientResponse.class);
+ assertEquals(MediaType.APPLICATION_JSON_TYPE, response.getType());
+ verifyEntities(response.getEntity(TimelineEntities.class));
+ }
+
+ @Test
+ public void testFromId() throws Exception {
+ WebResource r = resource();
+ ClientResponse response = r.path("ws").path("v1").path("timeline")
+ .path("type_1").queryParam("fromId", "id_2")
+ .accept(MediaType.APPLICATION_JSON)
+ .get(ClientResponse.class);
+ assertEquals(MediaType.APPLICATION_JSON_TYPE, response.getType());
+ assertEquals(1, response.getEntity(TimelineEntities.class).getEntities()
+ .size());
+
+ response = r.path("ws").path("v1").path("timeline")
+ .path("type_1").queryParam("fromId", "id_1")
+ .accept(MediaType.APPLICATION_JSON)
+ .get(ClientResponse.class);
+ assertEquals(MediaType.APPLICATION_JSON_TYPE, response.getType());
+ assertEquals(2, response.getEntity(TimelineEntities.class).getEntities()
+ .size());
+ }
+
+ @Test
+ public void testFromTs() throws Exception {
+ WebResource r = resource();
+ ClientResponse response = r.path("ws").path("v1").path("timeline")
+ .path("type_1").queryParam("fromTs", Long.toString(beforeTime))
+ .accept(MediaType.APPLICATION_JSON)
+ .get(ClientResponse.class);
+ assertEquals(MediaType.APPLICATION_JSON_TYPE, response.getType());
+ assertEquals(0, response.getEntity(TimelineEntities.class).getEntities()
+ .size());
+
+ response = r.path("ws").path("v1").path("timeline")
+ .path("type_1").queryParam("fromTs", Long.toString(
+ System.currentTimeMillis()))
+ .accept(MediaType.APPLICATION_JSON)
+ .get(ClientResponse.class);
+ assertEquals(MediaType.APPLICATION_JSON_TYPE, response.getType());
+ assertEquals(2, response.getEntity(TimelineEntities.class).getEntities()
+ .size());
+ }
+
+ @Test
+ public void testPrimaryFilterString() {
+ WebResource r = resource();
+ ClientResponse response = r.path("ws").path("v1").path("timeline")
+ .path("type_1").queryParam("primaryFilter", "user:username")
+ .accept(MediaType.APPLICATION_JSON)
+ .get(ClientResponse.class);
+ assertEquals(MediaType.APPLICATION_JSON_TYPE, response.getType());
+ verifyEntities(response.getEntity(TimelineEntities.class));
+ }
+
+ @Test
+ public void testPrimaryFilterInteger() {
+ WebResource r = resource();
+ ClientResponse response = r.path("ws").path("v1").path("timeline")
+ .path("type_1").queryParam("primaryFilter",
+ "appname:" + Integer.toString(Integer.MAX_VALUE))
+ .accept(MediaType.APPLICATION_JSON)
+ .get(ClientResponse.class);
+ assertEquals(MediaType.APPLICATION_JSON_TYPE, response.getType());
+ verifyEntities(response.getEntity(TimelineEntities.class));
+ }
+
+ @Test
+ public void testPrimaryFilterLong() {
+ WebResource r = resource();
+ ClientResponse response = r.path("ws").path("v1").path("timeline")
+ .path("type_1").queryParam("primaryFilter",
+ "long:" + Long.toString((long)Integer.MAX_VALUE + 1l))
+ .accept(MediaType.APPLICATION_JSON)
+ .get(ClientResponse.class);
+ assertEquals(MediaType.APPLICATION_JSON_TYPE, response.getType());
+ verifyEntities(response.getEntity(TimelineEntities.class));
+ }
+
+ @Test
+ public void testPrimaryFilterNumericString() {
+ // without quotes, 123abc is interpreted as the number 123,
+ // which finds no entities
+ WebResource r = resource();
+ ClientResponse response = r.path("ws").path("v1").path("timeline")
+ .path("type_1").queryParam("primaryFilter", "other:123abc")
+ .accept(MediaType.APPLICATION_JSON)
+ .get(ClientResponse.class);
+ assertEquals(MediaType.APPLICATION_JSON_TYPE, response.getType());
+ assertEquals(0, response.getEntity(TimelineEntities.class).getEntities()
+ .size());
+ }
+
+ @Test
+ public void testPrimaryFilterNumericStringWithQuotes() {
+ WebResource r = resource();
+ ClientResponse response = r.path("ws").path("v1").path("timeline")
+ .path("type_1").queryParam("primaryFilter", "other:\"123abc\"")
+ .accept(MediaType.APPLICATION_JSON)
+ .get(ClientResponse.class);
+ assertEquals(MediaType.APPLICATION_JSON_TYPE, response.getType());
+ verifyEntities(response.getEntity(TimelineEntities.class));
+ }
+
+ @Test
+ public void testSecondaryFilters() {
+ WebResource r = resource();
+ ClientResponse response = r.path("ws").path("v1").path("timeline")
+ .path("type_1")
+ .queryParam("secondaryFilter",
+ "user:username,appname:" + Integer.toString(Integer.MAX_VALUE))
+ .accept(MediaType.APPLICATION_JSON)
+ .get(ClientResponse.class);
+ assertEquals(MediaType.APPLICATION_JSON_TYPE, response.getType());
+ verifyEntities(response.getEntity(TimelineEntities.class));
+ }
+
+ @Test
+ public void testGetEntity() throws Exception {
+ WebResource r = resource();
+ ClientResponse response = r.path("ws").path("v1").path("timeline")
+ .path("type_1").path("id_1")
+ .accept(MediaType.APPLICATION_JSON)
+ .get(ClientResponse.class);
+ assertEquals(MediaType.APPLICATION_JSON_TYPE, response.getType());
+ TimelineEntity entity = response.getEntity(TimelineEntity.class);
+ Assert.assertNotNull(entity);
+ Assert.assertEquals("id_1", entity.getEntityId());
+ Assert.assertEquals("type_1", entity.getEntityType());
+ Assert.assertEquals(123l, entity.getStartTime().longValue());
+ Assert.assertEquals(2, entity.getEvents().size());
+ Assert.assertEquals(4, entity.getPrimaryFilters().size());
+ Assert.assertEquals(4, entity.getOtherInfo().size());
+ }
+
+ @Test
+ public void testGetEntityFields1() throws Exception {
+ WebResource r = resource();
+ ClientResponse response = r.path("ws").path("v1").path("timeline")
+ .path("type_1").path("id_1").queryParam("fields", "events,otherinfo")
+ .accept(MediaType.APPLICATION_JSON)
+ .get(ClientResponse.class);
+ assertEquals(MediaType.APPLICATION_JSON_TYPE, response.getType());
+ TimelineEntity entity = response.getEntity(TimelineEntity.class);
+ Assert.assertNotNull(entity);
+ Assert.assertEquals("id_1", entity.getEntityId());
+ Assert.assertEquals("type_1", entity.getEntityType());
+ Assert.assertEquals(123l, entity.getStartTime().longValue());
+ Assert.assertEquals(2, entity.getEvents().size());
+ Assert.assertEquals(0, entity.getPrimaryFilters().size());
+ Assert.assertEquals(4, entity.getOtherInfo().size());
+ }
+
+ @Test
+ public void testGetEntityFields2() throws Exception {
+ WebResource r = resource();
+ ClientResponse response = r.path("ws").path("v1").path("timeline")
+ .path("type_1").path("id_1").queryParam("fields", "lasteventonly," +
+ "primaryfilters,relatedentities")
+ .accept(MediaType.APPLICATION_JSON)
+ .get(ClientResponse.class);
+ assertEquals(MediaType.APPLICATION_JSON_TYPE, response.getType());
+ TimelineEntity entity = response.getEntity(TimelineEntity.class);
+ Assert.assertNotNull(entity);
+ Assert.assertEquals("id_1", entity.getEntityId());
+ Assert.assertEquals("type_1", entity.getEntityType());
+ Assert.assertEquals(123l, entity.getStartTime().longValue());
+ Assert.assertEquals(1, entity.getEvents().size());
+ Assert.assertEquals(4, entity.getPrimaryFilters().size());
+ Assert.assertEquals(0, entity.getOtherInfo().size());
+ }
+
+ @Test
+ public void testGetEvents() throws Exception {
+ WebResource r = resource();
+ ClientResponse response = r.path("ws").path("v1").path("timeline")
+ .path("type_1").path("events")
+ .queryParam("entityId", "id_1")
+ .accept(MediaType.APPLICATION_JSON)
+ .get(ClientResponse.class);
+ assertEquals(MediaType.APPLICATION_JSON_TYPE, response.getType());
+ TimelineEvents events = response.getEntity(TimelineEvents.class);
+ Assert.assertNotNull(events);
+ Assert.assertEquals(1, events.getAllEvents().size());
+ TimelineEvents.EventsOfOneEntity partEvents = events.getAllEvents().get(0);
+ Assert.assertEquals(2, partEvents.getEvents().size());
+ TimelineEvent event1 = partEvents.getEvents().get(0);
+ Assert.assertEquals(456l, event1.getTimestamp());
+ Assert.assertEquals("end_event", event1.getEventType());
+ Assert.assertEquals(1, event1.getEventInfo().size());
+ TimelineEvent event2 = partEvents.getEvents().get(1);
+ Assert.assertEquals(123l, event2.getTimestamp());
+ Assert.assertEquals("start_event", event2.getEventType());
+ Assert.assertEquals(0, event2.getEventInfo().size());
+ }
+
+ @Test
+ public void testPostEntities() throws Exception {
+ TimelineEntities entities = new TimelineEntities();
+ TimelineEntity entity = new TimelineEntity();
+ entity.setEntityId("test id");
+ entity.setEntityType("test type");
+ entity.setStartTime(System.currentTimeMillis());
+ entities.addEntity(entity);
+ WebResource r = resource();
+ ClientResponse response = r.path("ws").path("v1").path("timeline")
+ .accept(MediaType.APPLICATION_JSON)
+ .type(MediaType.APPLICATION_JSON)
+ .post(ClientResponse.class, entities);
+ assertEquals(MediaType.APPLICATION_JSON_TYPE, response.getType());
+ TimelinePutResponse putResposne = response.getEntity(TimelinePutResponse.class);
+ Assert.assertNotNull(putResposne);
+ Assert.assertEquals(0, putResposne.getErrors().size());
+ // verify the entity exists in the store
+ response = r.path("ws").path("v1").path("timeline")
+ .path("test type").path("test id")
+ .accept(MediaType.APPLICATION_JSON)
+ .get(ClientResponse.class);
+ assertEquals(MediaType.APPLICATION_JSON_TYPE, response.getType());
+ entity = response.getEntity(TimelineEntity.class);
+ Assert.assertNotNull(entity);
+ Assert.assertEquals("test id", entity.getEntityId());
+ Assert.assertEquals("test type", entity.getEntityType());
+ }
+
+ @Test
+ public void testGetMetrics() throws Exception {
+ WebResource r = resource();
+ ClientResponse response = r.path("ws").path("v1").path("timeline")
+ .path("metrics").queryParam("metricNames", "cpu_user")
+ .accept(MediaType.APPLICATION_JSON)
+ .get(ClientResponse.class);
+ assertEquals(MediaType.APPLICATION_JSON_TYPE, response.getType());
+ verifyMetrics(response.getEntity(TimelineMetrics.class));
+ }
+}
http://git-wip-us.apache.org/repos/asf/ambari/blob/865d187e/ambari-metrics/ambari-metrics-host-monitoring/pom.xml
----------------------------------------------------------------------
diff --git a/ambari-metrics/ambari-metrics-host-monitoring/pom.xml b/ambari-metrics/ambari-metrics-host-monitoring/pom.xml
new file mode 100644
index 0000000..20c6642
--- /dev/null
+++ b/ambari-metrics/ambari-metrics-host-monitoring/pom.xml
@@ -0,0 +1,152 @@
+<!--
+ 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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
+ http://maven.apache.org/xsd/maven-4.0.0.xsd">
+
+ <parent>
+ <artifactId>ambari-metrics</artifactId>
+ <groupId>org.apache.ambari</groupId>
+ <version>1.3.0-SNAPSHOT</version>
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+ <packaging>pom</packaging>
+ <version>1.3.0-SNAPSHOT</version>
+ <artifactId>ambari-metrics-host-monitoring</artifactId>
+ <properties>
+ <resmonitor.install.dir>
+ /usr/lib/python2.6/site-packages/resource_monitoring
+ </resmonitor.install.dir>
+ </properties>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>build-helper-maven-plugin</artifactId>
+ <version>1.8</version>
+ <executions>
+ <execution>
+ <id>parse-version</id>
+ <phase>validate</phase>
+ <goals>
+ <goal>parse-version</goal>
+ </goals>
+ </execution>
+ <execution>
+ <id>regex-property</id>
+ <goals>
+ <goal>regex-property</goal>
+ </goals>
+ <configuration>
+ <name>ambariVersion</name>
+ <value>${project.version}</value>
+ <regex>^([0-9]+)\.([0-9]+)\.([0-9]+)(\.|-).*</regex>
+ <replacement>$1.$2.$3</replacement>
+ <failIfNoMatch>false</failIfNoMatch>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <artifactId>maven-compiler-plugin</artifactId>
+ <version>3.0</version>
+ </plugin>
+ <plugin>
+ <artifactId>maven-assembly-plugin</artifactId>
+ <configuration>
+ <tarLongFileMode>gnu</tarLongFileMode>
+ <descriptors>
+ <descriptor>src/packages/tarball/all.xml</descriptor>
+ </descriptors>
+ </configuration>
+ <executions>
+ <execution>
+ <id>build-tarball</id>
+ <phase>prepare-package</phase>
+ <goals>
+ <goal>single</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>exec-maven-plugin</artifactId>
+ <version>1.2</version>
+ <executions>
+ <execution>
+ <configuration>
+ <executable>python</executable>
+ <workingDirectory>
+ ${basedir}/src/main/python/resource_monitoring/psutil
+ </workingDirectory>
+ <arguments>
+ <argument>build.py</argument>
+ </arguments>
+ </configuration>
+ <id>generate-psutil-binaries</id>
+ <phase>package</phase>
+ <goals>
+ <goal>exec</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>rpm-maven-plugin</artifactId>
+ <version>2.0.1</version>
+ <executions>
+ <execution>
+ <!-- unbinds rpm creation from maven lifecycle -->
+ <phase>none</phase>
+ <goals>
+ <goal>rpm</goal>
+ </goals>
+ </execution>
+ </executions>
+
+ <configuration>
+ <group>Development</group>
+ <needarch>x86_64</needarch>
+ <autoRequires>false</autoRequires>
+ <mappings>
+ <mapping>
+ <directory>${resmonitor.install.dir}</directory>
+ <username>root</username>
+ <groupname>root</groupname>
+ <sources>
+ <source>
+ <location>
+ ${project.build.directory}/${project.artifactId}-${project.version}/resource_monitoring/external/build/*
+ </location>
+ <location>
+ ${project.build.directory}/${project.artifactId}-${project.version}/resource_monitoring/core
+ </location>
+ <location>
+ ${project.build.directory}/${project.artifactId}-${project.version}/resource_monitoring/__init__.py
+ </location>
+ </source>
+ </sources>
+ </mapping>
+ </mappings>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
+</project>
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ambari/blob/865d187e/ambari-metrics/ambari-metrics-host-monitoring/src/main/python/__init__.py
----------------------------------------------------------------------
diff --git a/ambari-metrics/ambari-metrics-host-monitoring/src/main/python/__init__.py b/ambari-metrics/ambari-metrics-host-monitoring/src/main/python/__init__.py
new file mode 100644
index 0000000..38daf22
--- /dev/null
+++ b/ambari-metrics/ambari-metrics-host-monitoring/src/main/python/__init__.py
@@ -0,0 +1,21 @@
+#!/usr/bin/env python
+
+'''
+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.
+'''
+
+from resource_monitoring.core import *
http://git-wip-us.apache.org/repos/asf/ambari/blob/865d187e/ambari-metrics/ambari-metrics-host-monitoring/src/main/python/core/__init__.py
----------------------------------------------------------------------
diff --git a/ambari-metrics/ambari-metrics-host-monitoring/src/main/python/core/__init__.py b/ambari-metrics/ambari-metrics-host-monitoring/src/main/python/core/__init__.py
new file mode 100644
index 0000000..1629c9f
--- /dev/null
+++ b/ambari-metrics/ambari-metrics-host-monitoring/src/main/python/core/__init__.py
@@ -0,0 +1,34 @@
+#!/usr/bin/env python
+"""
+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.
+
+"""
+
+from resource_monitoring.core import *
+import os, sys
+path = os.path.abspath(__file__)
+path = os.path.join(os.path.dirname(os.path.dirname(path)), "psutil/build/")
+
+for dir in os.walk(path).next()[1]:
+ if 'lib' in dir:
+ sys.path.append(os.path.join(path, dir))
+
+try:
+ import psutil
+except ImportError:
+ print 'psutil binaries need to be built by running, psutil/build.py ' \
+ 'manually or by running a, mvn clean package, command.'
http://git-wip-us.apache.org/repos/asf/ambari/blob/865d187e/ambari-metrics/ambari-metrics-host-monitoring/src/main/python/core/application_metric_map.py
----------------------------------------------------------------------
diff --git a/ambari-metrics/ambari-metrics-host-monitoring/src/main/python/core/application_metric_map.py b/ambari-metrics/ambari-metrics-host-monitoring/src/main/python/core/application_metric_map.py
new file mode 100644
index 0000000..8c32661
--- /dev/null
+++ b/ambari-metrics/ambari-metrics-host-monitoring/src/main/python/core/application_metric_map.py
@@ -0,0 +1,127 @@
+#!/usr/bin/env python
+
+'''
+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.
+'''
+
+import logging
+import json
+from threading import RLock
+
+logger = logging.getLogger()
+
+class ApplicationMetricMap:
+ """
+ A data structure to buffer metrics in memory.
+ The in-memory dict stores metrics as shown below:
+ { application_id : { metric_id : { timestamp : metric_value } } }
+ application_id => uniquely identify the metrics for an application / host.
+ metric_id => identify the metric
+ timestamp => collection time
+ metric_value => numeric value
+ """
+
+ app_metric_map = {}
+
+ def __init__(self, hostname, ip_address):
+ self.hostname = hostname
+ self.ip_address = ip_address
+ self.lock = RLock()
+ pass
+
+ def acquire_lock(self):
+ self.lock.acquire()
+
+ def release_lock(self):
+ self.lock.release()
+
+ def put_metric(self, application_id, metric_id, timestamp, value):
+ metric_map = self.app_metric_map.get(application_id)
+ if not metric_map:
+ metric_map = { metric_id : { timestamp : value } }
+ self.app_metric_map[ application_id ] = metric_map
+ else:
+ metric_id_map = metric_map.get(metric_id)
+ if not metric_id_map:
+ metric_id_map = { timestamp : value }
+ metric_map[ metric_id ] = metric_id_map
+ else:
+ metric_map[ metric_id ].update( { timestamp : value } )
+ pass
+ pass
+ pass
+
+ def delete_application_metrics(self, app_id):
+ del self.app_metric_map[ app_id ]
+ pass
+
+ def flatten(self, application_id = None):
+ """
+ Return flatten dict to caller in json format.
+ Json format:
+ {"metrics":[{"hostname":"a","metricname":"b","appid":"c",
+ "instanceid":"d","starttime":"e","metrics":{"t":"v"}}]}
+ """
+
+ timeline_metrics = { "metrics" : [] }
+ local_metric_map = {}
+
+ if application_id:
+ if self.app_metric_map.has_key(application_id):
+ local_metric_map = { application_id : self.app_metric_map[application_id] }
+ else:
+ logger.info("application_id: {0}, not present in the map.".format(application_id))
+ else:
+ local_metric_map = self.app_metric_map.copy()
+ pass
+
+ for appId, metrics in local_metric_map.iteritems():
+ for metricId, metricData in dict(metrics).iteritems():
+ # Create a timeline metric object
+ timeline_metric = {
+ "hostname" : self.hostname,
+ "metricname" : metricId,
+ "appid" : self.get_app_id(appId),
+ "instanceid" : self.get_instance_id(appId),
+ "starttime" : self.get_start_time(appId, metricId),
+ "metrics" : metricData
+ }
+ timeline_metrics[ "metrics" ].append( timeline_metric )
+ pass
+ pass
+
+ return json.dumps(timeline_metrics)
+ pass
+
+ def get_start_time(self, app_id, metric_id):
+ if self.app_metric_map.has_key(app_id):
+ if self.app_metric_map.get(app_id).has_key(metric_id):
+ return self.app_metric_map.get(app_id).get(metric_id).iteritems().next()[0]
+ pass
+
+ def format_app_id(self, app_id, instance_id = None):
+ return app_id + "_" + instance_id if instance_id else app_id
+ pass
+
+ def get_app_id(self, app_id):
+ return app_id.split("_")[0]
+ pass
+
+ def get_instance_id(self, app_id):
+ parts = app_id.split("_")
+ return parts[1] if len(parts) > 1 else ''
+ pass
http://git-wip-us.apache.org/repos/asf/ambari/blob/865d187e/ambari-metrics/ambari-metrics-host-monitoring/src/main/python/core/config_reader.py
----------------------------------------------------------------------
diff --git a/ambari-metrics/ambari-metrics-host-monitoring/src/main/python/core/config_reader.py b/ambari-metrics/ambari-metrics-host-monitoring/src/main/python/core/config_reader.py
new file mode 100644
index 0000000..7726925
--- /dev/null
+++ b/ambari-metrics/ambari-metrics-host-monitoring/src/main/python/core/config_reader.py
@@ -0,0 +1,127 @@
+#!/usr/bin/env python
+
+'''
+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.
+'''
+
+import ConfigParser
+import StringIO
+import json
+import os
+
+config = ConfigParser.RawConfigParser()
+CONFIG_FILE_PATH = "/etc/metric-monitor/conf/metric_monitor.ini"
+METRIC_FILE_PATH = "/etc/metric-monitor/conf/metric_groups.conf"
+
+config_content = """
+[default]
+debug_level = INFO
+metrics_server = host:port
+enable_time_threshold = false
+enable_value_threshold = false
+
+[emitter]
+send_interval = 60
+
+[collector]
+collector_sleep_interval = 5
+max_queue_size = 5000
+"""
+
+metric_group_info = """
+{
+ "host_metric_groups": {
+ "cpu_info": {
+ "collect_every": "15",
+ "metrics": [
+ {
+ "name": "cpu_user",
+ "value_threshold": "1.0"
+ }
+ ]
+ },
+ "disk_info": {
+ "collect_every": "30",
+ "metrics": [
+ {
+ "name": "disk_free",
+ "value_threshold": "5.0"
+ }
+ ]
+ },
+ "network_info": {
+ "collect_every": "20",
+ "metrics": [
+ {
+ "name": "bytes_out",
+ "value_threshold": "128"
+ }
+ ]
+ }
+ },
+ "process_metric_groups": {
+ "": {
+ "collect_every": "15",
+ "metrics": []
+ }
+ }
+}
+"""
+
+class Configuration:
+
+ def __init__(self):
+ global config_content
+ self.config = ConfigParser.RawConfigParser()
+ if os.path.exists(CONFIG_FILE_PATH):
+ self.config.read(CONFIG_FILE_PATH)
+ else:
+ self.config.readfp(StringIO.StringIO(config_content))
+ pass
+ if os.path.exists(METRIC_FILE_PATH):
+ self.metric_groups = json.load(open(METRIC_FILE_PATH))
+ else:
+ print 'No metric configs found at {0}'.format(METRIC_FILE_PATH)
+ pass
+
+ def getConfig(self):
+ return self.config
+
+ def getMetricGroupConfig(self):
+ return self.metric_groups
+
+ def get(self, section, key, default=None):
+ try:
+ value = self.config.get(section, key)
+ except:
+ return default
+ return value
+
+ def get_send_interval(self):
+ return int(self.get("emitter", "send_interval", 60))
+
+ def get_collector_sleep_interval(self):
+ return int(self.get("collector", "collector_sleep_interval", 5))
+
+ def get_server_address(self):
+ return self.get("default", "metrics_server")
+
+ def get_log_level(self):
+ return self.get("default", "debug_level", "INFO")
+
+ def get_max_queue_size(self):
+ return int(self.get("collector", "max_queue_size", 5000))
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ambari/blob/865d187e/ambari-metrics/ambari-metrics-host-monitoring/src/main/python/core/controller.py
----------------------------------------------------------------------
diff --git a/ambari-metrics/ambari-metrics-host-monitoring/src/main/python/core/controller.py b/ambari-metrics/ambari-metrics-host-monitoring/src/main/python/core/controller.py
new file mode 100644
index 0000000..3d1f487
--- /dev/null
+++ b/ambari-metrics/ambari-metrics-host-monitoring/src/main/python/core/controller.py
@@ -0,0 +1,126 @@
+#!/usr/bin/env python
+
+'''
+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.
+'''
+
+import logging
+import signal
+import threading
+import time
+import sys
+from Queue import Queue
+from threading import Timer
+from application_metric_map import ApplicationMetricMap
+from config_reader import Configuration
+from event_definition import HostMetricCollectEvent, ProcessMetricCollectEvent
+from metric_collector import MetricsCollector
+from emitter import Emitter
+from host_info import HostInfo
+
+logger = logging.getLogger()
+
+class Controller(threading.Thread):
+
+ def __init__(self, config):
+ # Process initialization code
+ threading.Thread.__init__(self)
+ logger.debug('Initializing Controller thread.')
+ self.lock = threading.Lock()
+ self.config = config
+ self.metrics_config = config.getMetricGroupConfig()
+ self.events_cache = []
+ hostinfo = HostInfo()
+ self.application_metric_map = ApplicationMetricMap(hostinfo.get_hostname(),
+ hostinfo.get_ip_address())
+ self.event_queue = Queue(config.get_max_queue_size())
+ self.metric_collector = MetricsCollector(self.event_queue, self.application_metric_map)
+ self.server_url = config.get_server_address()
+ self.sleep_interval = config.get_collector_sleep_interval()
+ self.initialize_events_cache()
+ self.emitter = Emitter(self.config, self.application_metric_map)
+
+ def run(self):
+ logger.info('Running Controller thread: %s' % threading.currentThread().getName())
+ # Wake every 5 seconds to push events to the queue
+ while True:
+ if (self.event_queue.full()):
+ logger.warn('Event Queue full!! Suspending further collections.')
+ else:
+ self.enqueque_events()
+ pass
+ time.sleep(self.sleep_interval)
+ pass
+
+ # TODO: Optimize to not use Timer class and use the Queue instead
+ def enqueque_events(self):
+ # Queue events for up to a minute
+ for event in self.events_cache:
+ t = Timer(event.get_collect_interval(), self.metric_collector.process_event(event))
+ t.start()
+ pass
+
+ def initialize_events_cache(self):
+ self.events_cache = []
+ try:
+ host_metrics_groups = self.metrics_config['host_metric_groups']
+ process_metrics_groups = self.metrics_config['process_metric_groups']
+ except KeyError, ke:
+ logger.warn('Error loading metric groups.')
+ raise ke
+ pass
+
+ if host_metrics_groups:
+ for name, properties in host_metrics_groups.iteritems():
+ event = HostMetricCollectEvent(properties, name)
+ logger.info('Adding event to cache, {0} : {1}'.format(name, properties))
+ self.events_cache.append(event)
+ pass
+ pass
+
+ if process_metrics_groups:
+ for name, properties in process_metrics_groups.iteritems():
+ event = ProcessMetricCollectEvent(properties, name)
+ logger.info('Adding event to cache, {0} : {1}'.format(name, properties))
+ #self.events_cache.append(event)
+ pass
+ pass
+
+ pass
+
+ def start_emitter(self):
+ self.emitter.start()
+
+def main(argv=None):
+ # Allow Ctrl-C
+ signal.signal(signal.SIGINT, signal.SIG_DFL)
+
+ config = Configuration()
+ collector = Controller(config)
+
+ logger.setLevel(config.get_log_level())
+ #formatter = logging.Formatter("%(asctime)s %(filename)s:%(lineno)d - %(message)s")
+ stream_handler = logging.StreamHandler(sys.stdout)
+ stream_handler.setFormatter(formatter)
+ logger.addHandler(stream_handler)
+ logger.info('Starting Server RPC Thread: {0}'.format(sys.argv))
+
+ collector.start()
+ collector.start_emitter()
+
+if __name__ == '__main__':
+ main()
http://git-wip-us.apache.org/repos/asf/ambari/blob/865d187e/ambari-metrics/ambari-metrics-host-monitoring/src/main/python/core/emitter.py
----------------------------------------------------------------------
diff --git a/ambari-metrics/ambari-metrics-host-monitoring/src/main/python/core/emitter.py b/ambari-metrics/ambari-metrics-host-monitoring/src/main/python/core/emitter.py
new file mode 100644
index 0000000..c907445
--- /dev/null
+++ b/ambari-metrics/ambari-metrics-host-monitoring/src/main/python/core/emitter.py
@@ -0,0 +1,90 @@
+#!/usr/bin/env python
+
+'''
+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.
+'''
+
+import logging
+import threading
+import time
+import urllib2
+
+logger = logging.getLogger()
+COLLECTOR_URL = "http://{0}/ws/v1/timeline/metrics"
+RETRY_SLEEP_INTERVAL = 5
+MAX_RETRY_COUNT = 3
+
+class Emitter(threading.Thread):
+ """
+ Wake up every send interval seconds and empty the application metric map.
+ """
+
+ def __init__(self, config, application_metric_map):
+ threading.Thread.__init__(self)
+ logger.debug('Initializing Emitter thread.')
+ self.lock = threading.Lock()
+ self.collector_address = config.get_server_address()
+ self.send_interval = config.get_send_interval()
+ self.application_metric_map = application_metric_map
+
+ def run(self):
+ logger.info('Running Emitter thread: %s' % threading.currentThread().getName())
+ while True:
+ try:
+ retry_count = 0
+ while retry_count < MAX_RETRY_COUNT:
+ self.application_metric_map.acquire_lock()
+ json_data = self.application_metric_map.flatten()
+ if json_data is None:
+ logger.info("Nothing to emit, resume waiting.")
+ break
+ pass
+ response = self.push_metrics(json_data)
+
+ if response and response.status == '201':
+ retry_count = MAX_RETRY_COUNT
+ self.application_metric_map.clear()
+ self.application_metric_map.release_lock()
+ else:
+ logger.warn("Error sending metrics to server. Retrying after {0} "
+ "...".format(RETRY_SLEEP_INTERVAL))
+ self.application_metric_map.release_lock()
+ retry_count += 1
+ time.sleep(RETRY_SLEEP_INTERVAL)
+ pass
+ pass
+
+ time.sleep(self.send_interval)
+ except Exception, e:
+ logger.warn('Unable to emit events. %s' % str(e))
+ time.sleep(RETRY_SLEEP_INTERVAL)
+ logger.info('Retrying emit after %s seconds.' % RETRY_SLEEP_INTERVAL)
+ pass
+
+ def push_metrics(self, data):
+ headers = {"Content-Type" : "application/json", "Accept" : "*/*"}
+ server = COLLECTOR_URL.format(self.collector_address.strip())
+ logger.info("server: %s" % server)
+ req = urllib2.Request(server, data, headers)
+ response = urllib2.urlopen(req, timeout=int(self.send_interval - 10))
+ if response:
+ logger.debug("POST response from server: status = {0}, code = {1}".format(
+ response.status, response.reason))
+ logger.debug(response.read())
+ pass
+ return response
+
http://git-wip-us.apache.org/repos/asf/ambari/blob/865d187e/ambari-metrics/ambari-metrics-host-monitoring/src/main/python/core/event_definition.py
----------------------------------------------------------------------
diff --git a/ambari-metrics/ambari-metrics-host-monitoring/src/main/python/core/event_definition.py b/ambari-metrics/ambari-metrics-host-monitoring/src/main/python/core/event_definition.py
new file mode 100644
index 0000000..f356b03
--- /dev/null
+++ b/ambari-metrics/ambari-metrics-host-monitoring/src/main/python/core/event_definition.py
@@ -0,0 +1,85 @@
+#!/usr/bin/env python
+
+'''
+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.
+'''
+
+import logging
+
+DEFAULT_COLLECT_INTERVAL = 10
+
+logger = logging.getLogger()
+
+class Event:
+ def __init__(self):
+ self._classname = self.__class__.__name__
+
+ def get_classname(self):
+ return self._classname
+
+ def get_collect_interval(self):
+ return DEFAULT_COLLECT_INTERVAL
+
+
+class EmmitEvent(Event):
+
+ def __init__(self, application_metric_map, config):
+ Event.__init__(self)
+ self.collector_address = config.get_server_address()
+ self.application_metric_map = application_metric_map
+
+ def get_emmit_payload(self):
+ return self.application_metric_map.flatten()
+
+
+class HostMetricCollectEvent(Event):
+
+ def __init__(self, group_config, group_name):
+ Event.__init__(self)
+ self.group_config = group_config
+ self.group_name = group_name
+ try:
+ self.group_interval = group_config['collect_every']
+ self.metrics = group_config['metrics']
+ except KeyError, ex:
+ logger.warn('Unable to create event from metric group. {0}'.format(
+ group_config))
+ raise ex
+
+ def get_metric_value_thresholds(self):
+ metric_value_thresholds = {}
+
+ for metric in self.metrics:
+ try:
+ metric_value_thresholds[metric['name']] = metric['value_threshold']
+ except:
+ logger.warn('Error parsing metric configuration. {0}'.format(metric))
+ pass
+
+ return metric_value_thresholds
+
+ def get_group_name(self):
+ return self.group_name
+
+ def get_collect_interval(self):
+ return int(self.group_interval if self.group_interval else DEFAULT_COLLECT_INTERVAL)
+
+class ProcessMetricCollectEvent:
+
+ def __init__(self, group_config, group_name):
+ # Initialize the Process metric event
+ pass
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ambari/blob/865d187e/ambari-metrics/ambari-metrics-host-monitoring/src/main/python/core/host_info.py
----------------------------------------------------------------------
diff --git a/ambari-metrics/ambari-metrics-host-monitoring/src/main/python/core/host_info.py b/ambari-metrics/ambari-metrics-host-monitoring/src/main/python/core/host_info.py
new file mode 100644
index 0000000..4b39119
--- /dev/null
+++ b/ambari-metrics/ambari-metrics-host-monitoring/src/main/python/core/host_info.py
@@ -0,0 +1,187 @@
+#!/usr/bin/env python
+
+'''
+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.
+'''
+
+import logging
+import psutil
+import os
+from collections import namedtuple
+import platform
+import socket
+
+logger = logging.getLogger()
+
+def bytes2human(n):
+ bytes = float(n)
+ gigabytes = bytes / 1073741824
+ return '%.2f' % gigabytes
+pass
+
+
+class HostInfo():
+
+
+ def get_cpu_times(self):
+ """
+ Return cpu stats at current time
+ """
+ cpu_times = psutil.cpu_times()
+ load_avg = os.getloadavg()
+
+ return {
+ 'cpu_user' : cpu_times.user if hasattr(cpu_times, 'user') else '',
+ 'cpu_system' : cpu_times.system if hasattr(cpu_times, 'system') else '',
+ 'cpu_idle' : cpu_times.idle if hasattr(cpu_times, 'idle') else '',
+ 'cpu_nice' : cpu_times.nice if hasattr(cpu_times, 'nice') else '',
+ 'cpu_wio' : cpu_times.iowait if hasattr(cpu_times, 'iowait') else '',
+ 'cpu_intr' : cpu_times.irq if hasattr(cpu_times, 'irq') else '',
+ 'cpu_sintr' : cpu_times.softirq if hasattr(cpu_times, 'softirq') else '',
+ 'load_one' : load_avg[0] if hasattr(cpu_times, 'load_avg') else '',
+ 'load_five' : load_avg[1] if hasattr(cpu_times, 'load_avg') else '',
+ 'load_fifteen' : load_avg[2] if hasattr(cpu_times, 'load_avg') else ''
+ }
+ pass
+
+ def get_mem_info(self):
+ """
+ Return memory statistics at current time
+ """
+
+ mem_stats = psutil.virtual_memory()
+ swap_stats = psutil.swap_memory()
+ disk_usage = self.get_combined_disk_usage()
+
+ return {
+ 'mem_free' : mem_stats.free if hasattr(mem_stats, 'free') else '',
+ 'mem_shared' : mem_stats.shared if hasattr(mem_stats, 'shared') else '',
+ 'mem_buffered' : mem_stats.buffers if hasattr(mem_stats, 'buffers') else '',
+ 'mem_cached' : mem_stats.cached if hasattr(mem_stats, 'cached') else '',
+ 'swap_free' : swap_stats.free if hasattr(mem_stats, 'free') else '',
+ 'disk_free' : disk_usage.get("disk_free"),
+ 'part_max_used' : disk_usage.get("max_part_used")[0],
+ 'disk_total' : disk_usage.get("disk_total")
+ }
+ pass
+
+ def get_network_info(self):
+ """
+ Return network counters
+ """
+ net_stats = psutil.net_io_counters()
+
+ return {
+ 'bytes_out' : net_stats.bytes_sent,
+ 'bytes_in' : net_stats.bytes_recv,
+ 'pkts_out' : net_stats.packets_sent,
+ 'pkts_in' : net_stats.packets_recv
+ }
+ pass
+
+ # Faster version
+ def get_combined_disk_usage(self):
+ disk_usage = namedtuple('disk_usage', [ 'total', 'used', 'free',
+ 'percent', 'part_max_used' ])
+ combined_disk_total = 0
+ combined_disk_used = 0
+ combined_disk_free = 0
+ combined_disk_percent = 0
+ max_percent_usage = ('', 0)
+
+ for part in psutil.disk_partitions(all=False):
+ if os.name == 'nt':
+ if 'cdrom' in part.opts or part.fstype == '':
+ # skip cd-rom drives with no disk in it; they may raise
+ # ENOENT, pop-up a Windows GUI error for a non-ready
+ # partition or just hang.
+ continue
+ pass
+ pass
+ usage = psutil.disk_usage(part.mountpoint)
+
+ combined_disk_total += usage.total if hasattr(usage, 'total') else 0
+ combined_disk_used += usage.used if hasattr(usage, 'used') else 0
+ combined_disk_free += usage.free if hasattr(usage, 'free') else 0
+ combined_disk_percent += usage.percent if hasattr(usage, 'percent') else 0
+
+ if hasattr(usage, 'percent') and max_percent_usage[1] < int(usage.percent):
+ max_percent_usage = (part.mountpoint, usage.percent)
+ pass
+ pass
+
+ return { "disk_total" : bytes2human(combined_disk_total),
+ "disk_used" : bytes2human(combined_disk_used),
+ "disk_free" : bytes2human(combined_disk_free),
+ "disk_percent" : bytes2human(combined_disk_percent),
+ "max_part_used" : max_percent_usage }
+ pass
+
+ def get_host_static_info(self):
+
+ boot_time = psutil.boot_time()
+ cpu_count_logical = psutil.cpu_count()
+ swap_stats = psutil.swap_memory()
+ mem_info = psutil.virtual_memory()
+
+ return {
+ 'cpu_num' : cpu_count_logical,
+ 'cpu_speed' : '',
+ 'swap_total' : swap_stats.total,
+ 'boottime' : boot_time,
+ 'machine_type' : platform.processor(),
+ 'os_name' : platform.system(),
+ 'os_release' : platform.release(),
+ 'location' : '',
+ 'mem_total' : mem_info.total
+ }
+
+
+
+ def get_disk_usage(self):
+ disk_usage = {}
+
+ for part in psutil.disk_partitions(all=False):
+ if os.name == 'nt':
+ if 'cdrom' in part.opts or part.fstype == '':
+ # skip cd-rom drives with no disk in it; they may raise
+ # ENOENT, pop-up a Windows GUI error for a non-ready
+ # partition or just hang.
+ continue
+ pass
+ pass
+ usage = psutil.disk_usage(part.mountpoint)
+ disk_usage.update(
+ { part.device :
+ {
+ "total" : bytes2human(usage.total),
+ "user" : bytes2human(usage.used),
+ "free" : bytes2human(usage.free),
+ "percent" : int(usage.percent),
+ "fstype" : part.fstype,
+ "mount" : part.mountpoint
+ }
+ }
+ )
+ pass
+ pass
+
+ def get_hostname(self):
+ return socket.getfqdn()
+
+ def get_ip_address(self):
+ return socket.gethostbyname(socket.getfqdn())
http://git-wip-us.apache.org/repos/asf/ambari/blob/865d187e/ambari-metrics/ambari-metrics-host-monitoring/src/main/python/core/metric_collector.py
----------------------------------------------------------------------
diff --git a/ambari-metrics/ambari-metrics-host-monitoring/src/main/python/core/metric_collector.py b/ambari-metrics/ambari-metrics-host-monitoring/src/main/python/core/metric_collector.py
new file mode 100644
index 0000000..d4b96dc
--- /dev/null
+++ b/ambari-metrics/ambari-metrics-host-monitoring/src/main/python/core/metric_collector.py
@@ -0,0 +1,91 @@
+#!/usr/bin/env python
+
+'''
+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.
+'''
+
+import logging
+from time import time
+from host_info import HostInfo
+from event_definition import HostMetricCollectEvent, ProcessMetricCollectEvent
+
+logger = logging.getLogger()
+
+DEFAULT_HOST_APP_ID = '_HOST'
+
+class MetricsCollector():
+ """
+ The main Reader thread that dequeues events from the event queue and
+ submits a metric record to the emit buffer. Implementation of dequeue is
+ not required if Timer class is used for metric groups.
+ """
+
+ def __init__(self, emit_queue, application_metric_map):
+ self.emit_queue = emit_queue
+ self.application_metric_map = application_metric_map
+ self.host_info = HostInfo()
+ pass
+
+ def process_event(self, event):
+ if event.get_classname() == HostMetricCollectEvent.__name__:
+ self.process_host_collection_event(event)
+ elif event.get_classname() == ProcessMetricCollectEvent.__name__:
+ self.process_process_collection_event(event)
+ else:
+ logger.warn('Unknown event in queue')
+ pass
+
+ def process_host_collection_event(self, event):
+ startTime = int(round(time() * 1000))
+ metrics = None
+
+ if 'cpu' in event.get_group_name():
+ metrics = self.host_info.get_cpu_times()
+
+ elif 'disk' in event.get_group_name():
+ metrics = self.host_info.get_combined_disk_usage()
+
+ elif 'network' in event.get_group_name():
+ metrics = self.host_info.get_network_info()
+
+ elif 'mem' in event.get_group_name():
+ metrics = self.host_info.get_mem_info()
+
+ elif 'all' in event.get_group_name():
+ metrics = {}
+ metrics.update(self.host_info.get_cpu_times())
+ metrics.update(self.host_info.get_combined_disk_usage())
+ metrics.update(self.host_info.get_network_info())
+ metrics.update(self.host_info.get_mem_info())
+
+ else:
+ logger.warn('Unknown metric group.')
+ pass
+
+ if metrics:
+ self.application_metric_map.acquire_lock()
+ for metric_name, value in metrics.iteritems():
+ self.application_metric_map.put_metric(DEFAULT_HOST_APP_ID, metric_name, startTime, value)
+ pass
+ self.application_metric_map.release_lock()
+ pass
+
+ def process_process_collection_event(self, event):
+ """
+ Collect Process level metrics and update the application metric map
+ """
+ pass
http://git-wip-us.apache.org/repos/asf/ambari/blob/865d187e/ambari-metrics/ambari-metrics-host-monitoring/src/main/python/main.py
----------------------------------------------------------------------
diff --git a/ambari-metrics/ambari-metrics-host-monitoring/src/main/python/main.py b/ambari-metrics/ambari-metrics-host-monitoring/src/main/python/main.py
new file mode 100644
index 0000000..0a54d6b
--- /dev/null
+++ b/ambari-metrics/ambari-metrics-host-monitoring/src/main/python/main.py
@@ -0,0 +1,49 @@
+#!/usr/bin/env python
+
+'''
+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.
+'''
+
+import resource_monitoring.core
+from core.controller import Controller, Configuration
+import logging
+import signal
+import sys
+
+logger = logging.getLogger()
+
+def main(argv=None):
+ # Allow Ctrl-C
+ signal.signal(signal.SIGINT, signal.SIG_DFL)
+
+ config = Configuration()
+ collector = Controller(config)
+
+ logger.setLevel(config.get_log_level())
+ formatter = logging.Formatter("%(asctime)s %(filename)s:%(lineno)d - %(message)s")
+ stream_handler = logging.StreamHandler()
+ stream_handler.setFormatter(formatter)
+ logger.addHandler(stream_handler)
+ logger.info('Starting Server RPC Thread: %s' % ' '.join(sys.argv))
+
+ collector.start()
+ collector.start_emitter()
+
+
+if __name__ == '__main__':
+ main()
+
http://git-wip-us.apache.org/repos/asf/ambari/blob/865d187e/ambari-metrics/ambari-metrics-host-monitoring/src/main/python/psutil/LICENSE
----------------------------------------------------------------------
diff --git a/ambari-metrics/ambari-metrics-host-monitoring/src/main/python/psutil/LICENSE b/ambari-metrics/ambari-metrics-host-monitoring/src/main/python/psutil/LICENSE
new file mode 100644
index 0000000..e91b135
--- /dev/null
+++ b/ambari-metrics/ambari-metrics-host-monitoring/src/main/python/psutil/LICENSE
@@ -0,0 +1,27 @@
+psutil is distributed under BSD license reproduced below.
+
+Copyright (c) 2009, Jay Loden, Dave Daeschler, Giampaolo Rodola'
+All rights reserved.
+
+Redistribution and use in source and binary forms, with or without modification,
+are permitted provided that the following conditions are met:
+
+ * Redistributions of source code must retain the above copyright notice, this
+ list of conditions and the following disclaimer.
+ * Redistributions in binary form must reproduce the above copyright notice,
+ this list of conditions and the following disclaimer in the documentation
+ and/or other materials provided with the distribution.
+ * Neither the name of the psutil authors nor the names of its contributors
+ may be used to endorse or promote products derived from this software without
+ specific prior written permission.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
+ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR
+ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
+(INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
+LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON
+ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
http://git-wip-us.apache.org/repos/asf/ambari/blob/865d187e/ambari-metrics/ambari-metrics-host-monitoring/src/main/python/psutil/MANIFEST.in
----------------------------------------------------------------------
diff --git a/ambari-metrics/ambari-metrics-host-monitoring/src/main/python/psutil/MANIFEST.in b/ambari-metrics/ambari-metrics-host-monitoring/src/main/python/psutil/MANIFEST.in
new file mode 100644
index 0000000..d930257
--- /dev/null
+++ b/ambari-metrics/ambari-metrics-host-monitoring/src/main/python/psutil/MANIFEST.in
@@ -0,0 +1,14 @@
+include CREDITS
+include HISTORY
+include LICENSE
+include make.bat
+include Makefile
+include MANIFEST.in
+include README
+include setup.py
+include TODO
+recursive-include docs *
+recursive-exclude docs/_build *
+recursive-include examples *.py
+recursive-include psutil *.py *.c *.h
+recursive-include test *.py README
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ambari/blob/865d187e/ambari-metrics/ambari-metrics-host-monitoring/src/main/python/psutil/Makefile
----------------------------------------------------------------------
diff --git a/ambari-metrics/ambari-metrics-host-monitoring/src/main/python/psutil/Makefile b/ambari-metrics/ambari-metrics-host-monitoring/src/main/python/psutil/Makefile
new file mode 100644
index 0000000..b812527
--- /dev/null
+++ b/ambari-metrics/ambari-metrics-host-monitoring/src/main/python/psutil/Makefile
@@ -0,0 +1,77 @@
+# Shortcuts for various tasks (UNIX only).
+# To use a specific Python version run:
+# $ make install PYTHON=python3.3
+
+# You can set these variables from the command line.
+PYTHON = python
+TSCRIPT = test/test_psutil.py
+
+all: test
+
+clean:
+ rm -f `find . -type f -name \*.py[co]`
+ rm -f `find . -type f -name \*.so`
+ rm -f `find . -type f -name .\*~`
+ rm -f `find . -type f -name \*.orig`
+ rm -f `find . -type f -name \*.bak`
+ rm -f `find . -type f -name \*.rej`
+ rm -rf `find . -type d -name __pycache__`
+ rm -rf *.egg-info
+ rm -rf *\$testfile*
+ rm -rf build
+ rm -rf dist
+ rm -rf docs/_build
+
+build: clean
+ $(PYTHON) setup.py build
+
+install: build
+ if test $(PYTHON) = python2.4; then \
+ $(PYTHON) setup.py install; \
+ elif test $(PYTHON) = python2.5; then \
+ $(PYTHON) setup.py install; \
+ else \
+ $(PYTHON) setup.py install --user; \
+ fi
+
+uninstall:
+ if test $(PYTHON) = python2.4; then \
+ pip-2.4 uninstall -y -v psutil; \
+ else \
+ cd ..; $(PYTHON) -m pip uninstall -y -v psutil; \
+ fi
+
+test: install
+ $(PYTHON) $(TSCRIPT)
+
+test-process: install
+ $(PYTHON) -m unittest -v test.test_psutil.TestProcess
+
+test-system: install
+ $(PYTHON) -m unittest -v test.test_psutil.TestSystemAPIs
+
+# Run a specific test by name; e.g. "make test-by-name disk_" will run
+# all test methods containing "disk_" in their name.
+# Requires "pip install nose".
+test-by-name:
+ @$(PYTHON) -m nose test/test_psutil.py --nocapture -v -m $(filter-out $@,$(MAKECMDGOALS))
+
+memtest: install
+ $(PYTHON) test/test_memory_leaks.py
+
+pep8:
+ @hg locate '*py' | xargs pep8
+
+pyflakes:
+ @export PYFLAKES_NODOCTEST=1 && \
+ hg locate '*py' | xargs pyflakes
+
+# Upload source tarball on https://pypi.python.org/pypi/psutil.
+upload-src: clean
+ $(PYTHON) setup.py sdist upload
+
+# Build and upload doc on https://pythonhosted.org/psutil/.
+# Requires "pip install sphinx-pypi-upload".
+upload-doc:
+ cd docs; make html
+ $(PYTHON) setup.py upload_sphinx --upload-dir=docs/_build/html