eagle-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From h..@apache.org
Subject [3/6] incubator-eagle git commit: [EAGLE-787] add healthy check for hadoop-queue/topology-health/spark-history apps
Date Tue, 22 Nov 2016 07:19:00 GMT
[EAGLE-787] add healthy check for hadoop-queue/topology-health/spark-history apps

Author: wujinhu <wujinhu920@126.com>

Closes #671 from wujinhu/EAGLE-787.


Project: http://git-wip-us.apache.org/repos/asf/incubator-eagle/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-eagle/commit/7471eced
Tree: http://git-wip-us.apache.org/repos/asf/incubator-eagle/tree/7471eced
Diff: http://git-wip-us.apache.org/repos/asf/incubator-eagle/diff/7471eced

Branch: refs/heads/master
Commit: 7471eced641b132df7e2950a93a00914be75cc97
Parents: 620959b
Author: wujinhu <wujinhu920@126.com>
Authored: Tue Nov 22 09:53:04 2016 +0800
Committer: Hao Chen <hao@apache.org>
Committed: Tue Nov 22 15:15:08 2016 +0800

----------------------------------------------------------------------
 .../impl/ApplicationHealthCheckBase.java        |  44 ++++++++
 .../impl/ApplicationHealthCheckServiceImpl.java |  10 +-
 .../eagle/app/spi/ApplicationProvider.java      |   1 +
 .../src/main/resources/HealthCheckTemplate.vm   |   2 +-
 .../queue/HadoopQueueRunningAppProvider.java    |   7 ++
 ...adoopQueueRunningApplicationHealthCheck.java | 100 +++++++++++++++++++
 .../MRHistoryJobApplicationHealthCheck.java     |  41 +++++---
 .../history/SparkHistoryJobAppProvider.java     |   7 ++
 .../SparkHistoryJobApplicationHealthCheck.java  |  99 ++++++++++++++++++
 .../topology/TopologyCheckAppProvider.java      |   7 ++
 .../TopologyCheckApplicationHealthCheck.java    |  98 ++++++++++++++++++
 .../eagle/topology/TopologyConstants.java       |   1 +
 12 files changed, 399 insertions(+), 18 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/7471eced/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/service/impl/ApplicationHealthCheckBase.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/service/impl/ApplicationHealthCheckBase.java
b/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/service/impl/ApplicationHealthCheckBase.java
new file mode 100644
index 0000000..104e76d
--- /dev/null
+++ b/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/service/impl/ApplicationHealthCheckBase.java
@@ -0,0 +1,44 @@
+/*
+ * 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
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
+ * 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.eagle.app.service.impl;
+
+import com.codahale.metrics.health.HealthCheck;
+import com.google.inject.Inject;
+import com.typesafe.config.Config;
+import org.apache.eagle.metadata.model.ApplicationEntity;
+import org.apache.eagle.metadata.service.ApplicationEntityService;
+
+public abstract class ApplicationHealthCheckBase extends HealthCheck {
+    private static final String APP_ID_PATH = "appId";
+    protected static final long DEFAULT_MAX_DELAY_TIME = 2 * 60 * 60 * 1000L;
+    protected static final String MAX_DELAY_TIME_KEY = "application.maxDelayTime";
+
+    protected Config config;
+
+    @Inject
+    private ApplicationEntityService applicationEntityService;
+
+    protected ApplicationHealthCheckBase(Config config) {
+        this.config = config;
+    }
+
+    protected ApplicationEntity.Status getApplicationStatus() {
+        ApplicationEntity applicationEntity = applicationEntityService.getByUUIDOrAppId(null,
config.getString(APP_ID_PATH));
+        return applicationEntity.getStatus();
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/7471eced/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/service/impl/ApplicationHealthCheckServiceImpl.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/service/impl/ApplicationHealthCheckServiceImpl.java
b/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/service/impl/ApplicationHealthCheckServiceImpl.java
index 9d98b2f..4aea963 100644
--- a/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/service/impl/ApplicationHealthCheckServiceImpl.java
+++ b/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/service/impl/ApplicationHealthCheckServiceImpl.java
@@ -19,6 +19,7 @@ package org.apache.eagle.app.service.impl;
 
 import com.codahale.metrics.health.HealthCheck;
 import com.google.inject.Inject;
+import com.google.inject.Injector;
 import com.typesafe.config.Config;
 import com.typesafe.config.ConfigFactory;
 import io.dropwizard.setup.Environment;
@@ -56,6 +57,9 @@ public class ApplicationHealthCheckServiceImpl extends ApplicationHealthCheckSer
     private static final String HEALTH_PUBLISHER_IMPL_PATH = "application.healthCheck.publisher.publisherImpl";
 
     @Inject
+    private Injector currentInjector;
+
+    @Inject
     public ApplicationHealthCheckServiceImpl(ApplicationProviderService applicationProviderService,
                                              ApplicationEntityService applicationEntityService,
                                              Config config) {
@@ -105,11 +109,12 @@ public class ApplicationHealthCheckServiceImpl extends ApplicationHealthCheckSer
         }
         ApplicationProvider<?> appProvider = applicationProviderService.getApplicationProviderByType(appEntity.getDescriptor().getType());
         HealthCheck applicationHealthCheck = appProvider.getAppHealthCheck(
-                ConfigFactory.parseMap(appEntity.getConfiguration())
+                        ConfigFactory.parseMap(appEntity.getContext())
                         .withFallback(config)
-                        .withFallback(ConfigFactory.parseMap(appEntity.getContext()))
+                        .withFallback(ConfigFactory.parseMap(appEntity.getConfiguration()))
         );
         this.environment.healthChecks().register(appEntity.getAppId(), applicationHealthCheck);
+        currentInjector.injectMembers(applicationHealthCheck);
         synchronized (lock) {
             if (!appHealthChecks.containsKey(appEntity.getAppId())) {
                 appHealthChecks.put(appEntity.getAppId(), applicationHealthCheck);
@@ -137,7 +142,6 @@ public class ApplicationHealthCheckServiceImpl extends ApplicationHealthCheckSer
         registerAll();
         synchronized (lock) {
             for (String appId : appHealthChecks.keySet()) {
-                LOG.info("check application {}", appId);
                 HealthCheck.Result result = appHealthChecks.get(appId).execute();
                 if (result.isHealthy()) {
                     LOG.info("application {} is healthy", appId);

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/7471eced/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/spi/ApplicationProvider.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/spi/ApplicationProvider.java
b/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/spi/ApplicationProvider.java
index 143e026..d9c1eff 100644
--- a/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/spi/ApplicationProvider.java
+++ b/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/spi/ApplicationProvider.java
@@ -23,6 +23,7 @@ import org.apache.eagle.app.Application;
 import org.apache.eagle.app.service.ApplicationListener;
 import org.apache.eagle.common.module.ModuleRegistry;
 import org.apache.eagle.metadata.model.ApplicationDesc;
+import org.apache.eagle.metadata.service.ApplicationEntityService;
 
 import java.lang.reflect.ParameterizedType;
 import java.util.Optional;

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/7471eced/eagle-core/eagle-app/eagle-app-base/src/main/resources/HealthCheckTemplate.vm
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-app/eagle-app-base/src/main/resources/HealthCheckTemplate.vm
b/eagle-core/eagle-app/eagle-app-base/src/main/resources/HealthCheckTemplate.vm
index b368458..51c1186 100644
--- a/eagle-core/eagle-app/eagle-app-base/src/main/resources/HealthCheckTemplate.vm
+++ b/eagle-core/eagle-app/eagle-app-base/src/main/resources/HealthCheckTemplate.vm
@@ -23,7 +23,7 @@
 <body>
     #set ( $elem = $unHealthyContext )
 
-<p><b>Message: </b>$elem["appId"] is delayed</p>
+<p><b>Message: </b>$elem["appId"] has delayed</p>
 <p><b>Detail: </b>$elem["unHealthyMessage"]</p>
 
 </body>

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/7471eced/eagle-jpm/eagle-hadoop-queue/src/main/java/org/apache/eagle/hadoop/queue/HadoopQueueRunningAppProvider.java
----------------------------------------------------------------------
diff --git a/eagle-jpm/eagle-hadoop-queue/src/main/java/org/apache/eagle/hadoop/queue/HadoopQueueRunningAppProvider.java
b/eagle-jpm/eagle-hadoop-queue/src/main/java/org/apache/eagle/hadoop/queue/HadoopQueueRunningAppProvider.java
index 916dd5b..5d4078c 100644
--- a/eagle-jpm/eagle-hadoop-queue/src/main/java/org/apache/eagle/hadoop/queue/HadoopQueueRunningAppProvider.java
+++ b/eagle-jpm/eagle-hadoop-queue/src/main/java/org/apache/eagle/hadoop/queue/HadoopQueueRunningAppProvider.java
@@ -16,10 +16,17 @@
  */
 package org.apache.eagle.hadoop.queue;
 
+import com.codahale.metrics.health.HealthCheck;
+import com.typesafe.config.Config;
 import org.apache.eagle.app.spi.AbstractApplicationProvider;
 
 public class HadoopQueueRunningAppProvider extends AbstractApplicationProvider<HadoopQueueRunningApp>
{
     public HadoopQueueRunningApp getApplication() {
         return new HadoopQueueRunningApp();
     }
+
+    @Override
+    public HealthCheck getAppHealthCheck(Config config) {
+        return new HadoopQueueRunningApplicationHealthCheck(config);
+    }
 }

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/7471eced/eagle-jpm/eagle-hadoop-queue/src/main/java/org/apache/eagle/hadoop/queue/HadoopQueueRunningApplicationHealthCheck.java
----------------------------------------------------------------------
diff --git a/eagle-jpm/eagle-hadoop-queue/src/main/java/org/apache/eagle/hadoop/queue/HadoopQueueRunningApplicationHealthCheck.java
b/eagle-jpm/eagle-hadoop-queue/src/main/java/org/apache/eagle/hadoop/queue/HadoopQueueRunningApplicationHealthCheck.java
new file mode 100644
index 0000000..c73021b
--- /dev/null
+++ b/eagle-jpm/eagle-hadoop-queue/src/main/java/org/apache/eagle/hadoop/queue/HadoopQueueRunningApplicationHealthCheck.java
@@ -0,0 +1,100 @@
+/*
+ * 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
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
+ * 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.eagle.hadoop.queue;
+
+import com.typesafe.config.Config;
+import org.apache.commons.lang.exception.ExceptionUtils;
+import org.apache.eagle.app.service.impl.ApplicationHealthCheckBase;
+import org.apache.eagle.hadoop.queue.common.HadoopClusterConstants;
+import org.apache.eagle.jpm.util.Constants;
+import org.apache.eagle.log.entity.GenericServiceAPIResponseEntity;
+import org.apache.eagle.metadata.model.ApplicationEntity;
+import org.apache.eagle.service.client.IEagleServiceClient;
+import org.apache.eagle.service.client.impl.EagleServiceClientImpl;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+import java.util.Map;
+
+public class HadoopQueueRunningApplicationHealthCheck extends ApplicationHealthCheckBase
{
+    private static final Logger LOG = LoggerFactory.getLogger(HadoopQueueRunningApplicationHealthCheck.class);
+
+    private HadoopQueueRunningAppConfig hadoopQueueRunningAppConfig;
+
+    public HadoopQueueRunningApplicationHealthCheck(Config config) {
+        super(config);
+        this.hadoopQueueRunningAppConfig = new HadoopQueueRunningAppConfig(config);
+    }
+
+    @Override
+    public Result check() {
+        HadoopQueueRunningAppConfig.EagleProps eagleServiceConfig = this.hadoopQueueRunningAppConfig.eagleProps;
+        IEagleServiceClient client = new EagleServiceClientImpl(
+                eagleServiceConfig.eagleService.host,
+                eagleServiceConfig.eagleService.port,
+                eagleServiceConfig.eagleService.username,
+                eagleServiceConfig.eagleService.password);
+
+        client.getJerseyClient().setReadTimeout(60000);
+
+        try {
+            ApplicationEntity.Status status = getApplicationStatus();
+            if (!status.toString().equals(ApplicationEntity.Status.RUNNING.toString())) {
+                String message = String.format("Application is not running, status is %s",
status.toString());
+                return Result.unhealthy(message);
+            }
+
+            String query = String.format("%s[@site=\"%s\"]<@site>{max(timestamp)}",
+                    Constants.GENERIC_METRIC_SERVICE,
+                    hadoopQueueRunningAppConfig.eagleProps.site);
+
+            GenericServiceAPIResponseEntity response = client
+                    .search(query)
+                    .metricName(HadoopClusterConstants.MetricName.HADOOP_CLUSTER_ALLOCATED_MEMORY)
+                    .startTime(System.currentTimeMillis() - 2 * 60 * 60000L)
+                    .endTime(System.currentTimeMillis())
+                    .pageSize(Integer.MAX_VALUE)
+                    .send();
+            List<Map<List<String>, List<Double>>> results = response.getObj();
+            long currentProcessTimeStamp = results.get(0).get("value").get(0).longValue();
+            long currentTimeStamp = System.currentTimeMillis();
+            long maxDelayTime = DEFAULT_MAX_DELAY_TIME;
+            if (hadoopQueueRunningAppConfig.getConfig().hasPath(MAX_DELAY_TIME_KEY)) {
+                maxDelayTime = hadoopQueueRunningAppConfig.getConfig().getLong(MAX_DELAY_TIME_KEY);
+            }
+
+            if (currentTimeStamp - currentProcessTimeStamp > maxDelayTime) {
+                String message = String.format("Current process time is %sms, delay %s hours",
+                        currentProcessTimeStamp, (currentTimeStamp - currentProcessTimeStamp)
* 1.0 / 60000L / 60);
+                return Result.unhealthy(message);
+            } else {
+                return Result.healthy();
+            }
+        } catch (Exception e) {
+            return Result.unhealthy(ExceptionUtils.getStackTrace(e.getCause()));
+        } finally {
+            client.getJerseyClient().destroy();
+            try {
+                client.close();
+            } catch (Exception e) {
+                LOG.warn("{}", e);
+            }
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/7471eced/eagle-jpm/eagle-jpm-mr-history/src/main/java/org/apache/eagle/jpm/mr/history/MRHistoryJobApplicationHealthCheck.java
----------------------------------------------------------------------
diff --git a/eagle-jpm/eagle-jpm-mr-history/src/main/java/org/apache/eagle/jpm/mr/history/MRHistoryJobApplicationHealthCheck.java
b/eagle-jpm/eagle-jpm-mr-history/src/main/java/org/apache/eagle/jpm/mr/history/MRHistoryJobApplicationHealthCheck.java
index ab078f5..91e16df 100644
--- a/eagle-jpm/eagle-jpm-mr-history/src/main/java/org/apache/eagle/jpm/mr/history/MRHistoryJobApplicationHealthCheck.java
+++ b/eagle-jpm/eagle-jpm-mr-history/src/main/java/org/apache/eagle/jpm/mr/history/MRHistoryJobApplicationHealthCheck.java
@@ -17,11 +17,12 @@
 
 package org.apache.eagle.jpm.mr.history;
 
-import com.codahale.metrics.health.HealthCheck;
 import com.typesafe.config.Config;
 import org.apache.commons.lang.exception.ExceptionUtils;
+import org.apache.eagle.app.service.impl.ApplicationHealthCheckBase;
 import org.apache.eagle.jpm.util.Constants;
 import org.apache.eagle.log.entity.GenericServiceAPIResponseEntity;
+import org.apache.eagle.metadata.model.ApplicationEntity;
 import org.apache.eagle.service.client.IEagleServiceClient;
 import org.apache.eagle.service.client.impl.EagleServiceClientImpl;
 import org.slf4j.Logger;
@@ -30,28 +31,33 @@ import org.slf4j.LoggerFactory;
 import java.util.List;
 import java.util.Map;
 
-public class MRHistoryJobApplicationHealthCheck extends HealthCheck {
+public class MRHistoryJobApplicationHealthCheck extends ApplicationHealthCheckBase {
     private static final Logger LOG = LoggerFactory.getLogger(MRHistoryJobApplicationHealthCheck.class);
 
     private MRHistoryJobConfig mrHistoryJobConfig;
-    private static final long DEFAULT_MAX_DELAY_TIME = 2 * 60 * 60 * 1000L;
-    private static final String MAX_DELAY_TIME_KEY = "application.maxDelayTime";
 
     public MRHistoryJobApplicationHealthCheck(Config config) {
+        super(config);
         mrHistoryJobConfig = MRHistoryJobConfig.newInstance(config);
     }
 
     @Override
     public Result check() {
-        try {
-            MRHistoryJobConfig.EagleServiceConfig eagleServiceConfig = mrHistoryJobConfig.getEagleServiceConfig();
-            IEagleServiceClient client = new EagleServiceClientImpl(
-                    eagleServiceConfig.eagleServiceHost,
-                    eagleServiceConfig.eagleServicePort,
-                    eagleServiceConfig.username,
-                    eagleServiceConfig.password);
+        MRHistoryJobConfig.EagleServiceConfig eagleServiceConfig = mrHistoryJobConfig.getEagleServiceConfig();
+        IEagleServiceClient client = new EagleServiceClientImpl(
+                eagleServiceConfig.eagleServiceHost,
+                eagleServiceConfig.eagleServicePort,
+                eagleServiceConfig.username,
+                eagleServiceConfig.password);
+
+        client.getJerseyClient().setReadTimeout(eagleServiceConfig.readTimeoutSeconds * 1000);
 
-            client.getJerseyClient().setReadTimeout(eagleServiceConfig.readTimeoutSeconds
* 1000);
+        try {
+            ApplicationEntity.Status status = getApplicationStatus();
+            if (!status.toString().equals(ApplicationEntity.Status.RUNNING.toString())) {
+                String message = String.format("Application is not running, status is %s",
status.toString());
+                return Result.unhealthy(message);
+            }
 
             String query = String.format("%s[@site=\"%s\"]<@site>{max(currentTimeStamp)}",
                     Constants.JPA_JOB_PROCESS_TIME_STAMP_NAME,
@@ -73,14 +79,21 @@ public class MRHistoryJobApplicationHealthCheck extends HealthCheck {
             }
 
             if (currentTimeStamp - currentProcessTimeStamp > maxDelayTime) {
-                String message = String.format("current process time %sms, delay %sms",
-                        currentProcessTimeStamp, currentTimeStamp - currentProcessTimeStamp);
+                String message = String.format("Current process time is %sms, delay %s hours",
+                        currentProcessTimeStamp, (currentTimeStamp - currentProcessTimeStamp)
* 1.0 / 60000L / 60);
                 return Result.unhealthy(message);
             } else {
                 return Result.healthy();
             }
         } catch (Exception e) {
             return Result.unhealthy(ExceptionUtils.getStackTrace(e.getCause()));
+        } finally {
+            client.getJerseyClient().destroy();
+            try {
+                client.close();
+            } catch (Exception e) {
+                LOG.warn("{}", e);
+            }
         }
     }
 }

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/7471eced/eagle-jpm/eagle-jpm-spark-history/src/main/java/org/apache/eagle/jpm/spark/history/SparkHistoryJobAppProvider.java
----------------------------------------------------------------------
diff --git a/eagle-jpm/eagle-jpm-spark-history/src/main/java/org/apache/eagle/jpm/spark/history/SparkHistoryJobAppProvider.java
b/eagle-jpm/eagle-jpm-spark-history/src/main/java/org/apache/eagle/jpm/spark/history/SparkHistoryJobAppProvider.java
index 343d9c2..2b962c9 100644
--- a/eagle-jpm/eagle-jpm-spark-history/src/main/java/org/apache/eagle/jpm/spark/history/SparkHistoryJobAppProvider.java
+++ b/eagle-jpm/eagle-jpm-spark-history/src/main/java/org/apache/eagle/jpm/spark/history/SparkHistoryJobAppProvider.java
@@ -17,6 +17,8 @@
 
 package org.apache.eagle.jpm.spark.history;
 
+import com.codahale.metrics.health.HealthCheck;
+import com.typesafe.config.Config;
 import org.apache.eagle.app.spi.AbstractApplicationProvider;
 
 public class SparkHistoryJobAppProvider extends AbstractApplicationProvider<SparkHistoryJobApp>
{
@@ -24,4 +26,9 @@ public class SparkHistoryJobAppProvider extends AbstractApplicationProvider<Spar
     public SparkHistoryJobApp getApplication() {
         return new SparkHistoryJobApp();
     }
+
+    @Override
+    public HealthCheck getAppHealthCheck(Config config) {
+        return new SparkHistoryJobApplicationHealthCheck(config);
+    }
 }

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/7471eced/eagle-jpm/eagle-jpm-spark-history/src/main/java/org/apache/eagle/jpm/spark/history/SparkHistoryJobApplicationHealthCheck.java
----------------------------------------------------------------------
diff --git a/eagle-jpm/eagle-jpm-spark-history/src/main/java/org/apache/eagle/jpm/spark/history/SparkHistoryJobApplicationHealthCheck.java
b/eagle-jpm/eagle-jpm-spark-history/src/main/java/org/apache/eagle/jpm/spark/history/SparkHistoryJobApplicationHealthCheck.java
new file mode 100644
index 0000000..bbca566
--- /dev/null
+++ b/eagle-jpm/eagle-jpm-spark-history/src/main/java/org/apache/eagle/jpm/spark/history/SparkHistoryJobApplicationHealthCheck.java
@@ -0,0 +1,99 @@
+/*
+ * 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
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
+ * 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.eagle.jpm.spark.history;
+
+import com.typesafe.config.Config;
+import org.apache.commons.lang.exception.ExceptionUtils;
+import org.apache.eagle.app.service.impl.ApplicationHealthCheckBase;
+import org.apache.eagle.jpm.util.Constants;
+import org.apache.eagle.log.entity.GenericServiceAPIResponseEntity;
+import org.apache.eagle.metadata.model.ApplicationEntity;
+import org.apache.eagle.service.client.IEagleServiceClient;
+import org.apache.eagle.service.client.impl.EagleServiceClientImpl;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+import java.util.Map;
+
+public class SparkHistoryJobApplicationHealthCheck extends ApplicationHealthCheckBase {
+    private static final Logger LOG = LoggerFactory.getLogger(SparkHistoryJobApplicationHealthCheck.class);
+
+    private SparkHistoryJobAppConfig sparkHistoryJobAppConfig;
+
+    public SparkHistoryJobApplicationHealthCheck(Config config) {
+        super(config);
+        this.sparkHistoryJobAppConfig = SparkHistoryJobAppConfig.newInstance(config);
+    }
+
+    @Override
+    public Result check() {
+        SparkHistoryJobAppConfig.EagleInfo eagleServiceConfig = sparkHistoryJobAppConfig.eagleInfo;
+        IEagleServiceClient client = new EagleServiceClientImpl(
+                eagleServiceConfig.host,
+                eagleServiceConfig.port,
+                eagleServiceConfig.username,
+                eagleServiceConfig.password);
+
+        client.getJerseyClient().setReadTimeout(eagleServiceConfig.timeout * 1000);
+
+        try {
+            ApplicationEntity.Status status = getApplicationStatus();
+            if (!status.toString().equals(ApplicationEntity.Status.RUNNING.toString())) {
+                String message = String.format("Application is not running, status is %s",
status.toString());
+                return Result.unhealthy(message);
+            }
+
+            String query = String.format("%s[@site=\"%s\"]<@site>{max(endTime)}",
+                    Constants.SPARK_APP_SERVICE_ENDPOINT_NAME,
+                    sparkHistoryJobAppConfig.stormConfig.siteId);
+
+            GenericServiceAPIResponseEntity response = client
+                    .search(query)
+                    .startTime(0L)
+                    .endTime(System.currentTimeMillis())
+                    .pageSize(10)
+                    .send();
+
+            List<Map<List<String>, List<Double>>> results = response.getObj();
+            long currentProcessTimeStamp = results.get(0).get("value").get(0).longValue();
+            long currentTimeStamp = System.currentTimeMillis();
+            long maxDelayTime = DEFAULT_MAX_DELAY_TIME;
+            if (sparkHistoryJobAppConfig.getConfig().hasPath(MAX_DELAY_TIME_KEY)) {
+                maxDelayTime = sparkHistoryJobAppConfig.getConfig().getLong(MAX_DELAY_TIME_KEY);
+            }
+
+            if (currentTimeStamp - currentProcessTimeStamp > maxDelayTime * 3) {
+                String message = String.format("Current process time is %sms, delay %s hours",
+                        currentProcessTimeStamp, (currentTimeStamp - currentProcessTimeStamp)
* 1.0 / 60000L / 60);
+                return Result.unhealthy(message);
+            } else {
+                return Result.healthy();
+            }
+        } catch (Exception e) {
+            return Result.unhealthy(ExceptionUtils.getStackTrace(e.getCause()));
+        } finally {
+            client.getJerseyClient().destroy();
+            try {
+                client.close();
+            } catch (Exception e) {
+                LOG.warn("{}", e);
+            }
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/7471eced/eagle-topology-check/eagle-topology-app/src/main/java/org/apache/eagle/topology/TopologyCheckAppProvider.java
----------------------------------------------------------------------
diff --git a/eagle-topology-check/eagle-topology-app/src/main/java/org/apache/eagle/topology/TopologyCheckAppProvider.java
b/eagle-topology-check/eagle-topology-app/src/main/java/org/apache/eagle/topology/TopologyCheckAppProvider.java
index 5969a7a..5766454 100644
--- a/eagle-topology-check/eagle-topology-app/src/main/java/org/apache/eagle/topology/TopologyCheckAppProvider.java
+++ b/eagle-topology-check/eagle-topology-app/src/main/java/org/apache/eagle/topology/TopologyCheckAppProvider.java
@@ -18,6 +18,8 @@
 
 package org.apache.eagle.topology;
 
+import com.codahale.metrics.health.HealthCheck;
+import com.typesafe.config.Config;
 import org.apache.eagle.app.spi.AbstractApplicationProvider;
 
 public class TopologyCheckAppProvider extends AbstractApplicationProvider<TopologyCheckApp>
{
@@ -25,4 +27,9 @@ public class TopologyCheckAppProvider extends AbstractApplicationProvider<Topolo
     public TopologyCheckApp getApplication() {
         return new TopologyCheckApp();
     }
+
+    @Override
+    public HealthCheck getAppHealthCheck(Config config) {
+        return new TopologyCheckApplicationHealthCheck(config);
+    }
 }

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/7471eced/eagle-topology-check/eagle-topology-app/src/main/java/org/apache/eagle/topology/TopologyCheckApplicationHealthCheck.java
----------------------------------------------------------------------
diff --git a/eagle-topology-check/eagle-topology-app/src/main/java/org/apache/eagle/topology/TopologyCheckApplicationHealthCheck.java
b/eagle-topology-check/eagle-topology-app/src/main/java/org/apache/eagle/topology/TopologyCheckApplicationHealthCheck.java
new file mode 100644
index 0000000..25730c8
--- /dev/null
+++ b/eagle-topology-check/eagle-topology-app/src/main/java/org/apache/eagle/topology/TopologyCheckApplicationHealthCheck.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
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
+ * 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.eagle.topology;
+
+import com.typesafe.config.Config;
+import org.apache.commons.lang.exception.ExceptionUtils;
+import org.apache.eagle.app.service.impl.ApplicationHealthCheckBase;
+import org.apache.eagle.log.entity.GenericServiceAPIResponseEntity;
+import org.apache.eagle.metadata.model.ApplicationEntity;
+import org.apache.eagle.service.client.IEagleServiceClient;
+import org.apache.eagle.service.client.impl.EagleServiceClientImpl;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+import java.util.Map;
+
+public class TopologyCheckApplicationHealthCheck extends ApplicationHealthCheckBase {
+    private static final Logger LOG = LoggerFactory.getLogger(TopologyCheckApplicationHealthCheck.class);
+
+    private TopologyCheckAppConfig topologyCheckAppConfig;
+
+    public TopologyCheckApplicationHealthCheck(Config config) {
+        super(config);
+        topologyCheckAppConfig = TopologyCheckAppConfig.newInstance(config);
+    }
+
+    @Override
+    public Result check() {
+        //FIXME, this application owner please add eagle server config to Class TopologyCheckAppConfig
+        IEagleServiceClient client = new EagleServiceClientImpl(
+                topologyCheckAppConfig.getConfig().getString("service.host"),
+                topologyCheckAppConfig.getConfig().getInt("service.port"),
+                topologyCheckAppConfig.getConfig().getString("service.username"),
+                topologyCheckAppConfig.getConfig().getString("service.password"));
+
+        client.getJerseyClient().setReadTimeout(topologyCheckAppConfig.getConfig().getInt("service.readTimeOutSeconds")
* 1000);
+
+        try {
+            ApplicationEntity.Status status = getApplicationStatus();
+            if (!status.toString().equals(ApplicationEntity.Status.RUNNING.toString())) {
+                String message = String.format("Application is not running, status is %s",
status.toString());
+                return Result.unhealthy(message);
+            }
+
+            String query = String.format("%s[@site=\"%s\"]<@site>{max(timestamp)}",
+                    TopologyConstants.GENERIC_METRIC_SERVICE,
+                    topologyCheckAppConfig.dataExtractorConfig.site);
+
+            GenericServiceAPIResponseEntity response = client
+                    .search(query)
+                    .metricName(String.format(TopologyConstants.METRIC_LIVE_RATIO_NAME_FORMAT,
TopologyConstants.REGIONSERVER_ROLE))
+                    .startTime(System.currentTimeMillis() - 2 * 60 * 60000L)
+                    .endTime(System.currentTimeMillis())
+                    .pageSize(Integer.MAX_VALUE)
+                    .send();
+            List<Map<List<String>, List<Double>>> results = response.getObj();
+            long currentProcessTimeStamp = results.get(0).get("value").get(0).longValue();
+            long currentTimeStamp = System.currentTimeMillis();
+            long maxDelayTime = DEFAULT_MAX_DELAY_TIME;
+            if (topologyCheckAppConfig.getConfig().hasPath(MAX_DELAY_TIME_KEY)) {
+                maxDelayTime = topologyCheckAppConfig.getConfig().getLong(MAX_DELAY_TIME_KEY);
+            }
+
+            if (currentTimeStamp - currentProcessTimeStamp > maxDelayTime) {
+                String message = String.format("Current process time is %sms, delay %s hours",
+                        currentProcessTimeStamp, (currentTimeStamp - currentProcessTimeStamp)
* 1.0 / 60000L / 60);
+                return Result.unhealthy(message);
+            } else {
+                return Result.healthy();
+            }
+        } catch (Exception e) {
+            return Result.unhealthy(ExceptionUtils.getStackTrace(e.getCause()));
+        } finally {
+            client.getJerseyClient().destroy();
+            try {
+                client.close();
+            } catch (Exception e) {
+                LOG.warn("{}", e);
+            }
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/7471eced/eagle-topology-check/eagle-topology-entity/src/main/java/org/apache/eagle/topology/TopologyConstants.java
----------------------------------------------------------------------
diff --git a/eagle-topology-check/eagle-topology-entity/src/main/java/org/apache/eagle/topology/TopologyConstants.java
b/eagle-topology-check/eagle-topology-entity/src/main/java/org/apache/eagle/topology/TopologyConstants.java
index 74f446b..3e535a8 100644
--- a/eagle-topology-check/eagle-topology-entity/src/main/java/org/apache/eagle/topology/TopologyConstants.java
+++ b/eagle-topology-check/eagle-topology-entity/src/main/java/org/apache/eagle/topology/TopologyConstants.java
@@ -26,6 +26,7 @@ public class TopologyConstants {
     public static final String HBASE_INSTANCE_SERVICE_NAME = "HbaseServiceInstance";
     public static final String MR_INSTANCE_SERVICE_NAME = "MRServiceInstance";
     public static final String JN_INSTANCE_SERVICE_NAME = "JNServiceInstance";
+    public static final String GENERIC_METRIC_SERVICE = "GenericMetricService";
 
     public static final int DEFAULT_READ_TIMEOUT = 30 * 60 * 1000; // in milliseconds
     public static final Pattern HTTP_HOST_MATCH_PATTERN = Pattern.compile("^https?://(.+?):-?(\\d+)/?");


Mime
View raw message