[EAGLE-890] Add Job Scheduling Framework and HadoopHealthCheckApp

eagle-core/eagle-alert-parent/eagle-alert/alert-common/src/main/java/org/apache/eagle/alert/engine/coordinator/AlertSeverity.java

@@ -16,6 +16,9 @@
  */
 package org.apache.eagle.alert.engine.coordinator;
 
+/**
+ * Severity Levels for alerts, events, logs, etc.
+ */
 public enum AlertSeverity {
     UNKNOWN, OK, WARNING, CRITICAL, FATAL
 }

eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/ExecutableApplication.java

@@ -23,7 +23,7 @@
 public abstract class ExecutableApplication<E extends Environment, P> implements Application<E, P>, ApplicationTool {
     @Override
     public void run(Config config) {
-        ExecutionRuntimeManager.getInstance().getRuntime(getEnvironmentType(), config).start(this, config);
+        ExecutionRuntimeManager.getInstance().getRuntimeSingleton(getEnvironmentType(), config).start(this, config);
     }
 
     @Override

eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/SchedulingApplication.java

@@ -0,0 +1,55 @@
+/*
+ * 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;
+
+import com.typesafe.config.Config;
+import org.apache.eagle.app.environment.ExecutionRuntimeManager;
+import org.apache.eagle.app.environment.impl.ScheduledEnvironment;
+import org.apache.eagle.app.environment.impl.ScheduledExecutionRuntime;
+import org.apache.eagle.app.environment.impl.SchedulingPlan;
+
+public abstract class SchedulingApplication extends ExecutableApplication<ScheduledEnvironment, SchedulingPlan> {
+    @Override
+    public Class<? extends ScheduledEnvironment> getEnvironmentType() {
+        return ScheduledEnvironment.class;
+    }
+
+    @Override
+    public void run(Config config) {
+        ScheduledExecutionRuntime runtime = (ScheduledExecutionRuntime) ExecutionRuntimeManager.getInstance().getRuntimeSingleton(getEnvironmentType(), config);
+        try {
+            Runtime.getRuntime().addShutdownHook(new Thread() {
+                @Override
+                public void run() {
+                    try {
+                        runtime.stop();
+                    } catch (Exception e) {
+                        throw new IllegalStateException(e);
+                    }
+                }
+            });
+            runtime.start();
+            runtime.start(this, config);
+        } catch (Exception e) {
+            throw new IllegalStateException(e);
+        }
+    }
+
+    public static XmlSchedulingApplication buildFromXml(String schedulingXmlFile) {
+        return new XmlSchedulingApplication(schedulingXmlFile);
+    }
+}

eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/XmlSchedulingApplication.java

@@ -0,0 +1,141 @@
+/*
+ * 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;
+
+import com.google.common.base.Preconditions;
+import com.typesafe.config.Config;
+import org.apache.eagle.app.environment.impl.AbstractSchedulingPlan;
+import org.apache.eagle.app.environment.impl.ScheduledEnvironment;
+import org.apache.eagle.app.environment.impl.SchedulingPlan;
+import org.apache.eagle.app.job.listener.ApplicationMonitorJobListener;
+import org.apache.eagle.app.job.plugin.XMLStreamSchedulingDataProcessor;
+import org.quartz.SchedulerException;
+import org.quartz.impl.matchers.KeyMatcher;
+import org.quartz.spi.MutableTrigger;
+import org.quartz.xml.ValidationException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.xml.sax.SAXException;
+
+import javax.xml.parsers.ParserConfigurationException;
+import javax.xml.xpath.XPathException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.text.ParseException;
+import java.util.stream.Collectors;
+
+public class XmlSchedulingApplication extends SchedulingApplication {
+    private static final Logger LOGGER = LoggerFactory.getLogger(XmlSchedulingPlan.class);
+
+    private final String schedulingXmlFile;
+
+    public XmlSchedulingApplication(String schedulingXmlFile) {
+        this.schedulingXmlFile = schedulingXmlFile;
+    }
+
+    @Override
+    public SchedulingPlan execute(Config config, ScheduledEnvironment environment) {
+        return new XmlSchedulingPlan(schedulingXmlFile, config, environment);
+    }
+
+    private class XmlSchedulingPlan extends AbstractSchedulingPlan {
+        private final String schedulingXmlFile;
+        private final ApplicationMonitorJobListener appJobListener;
+
+        public XmlSchedulingPlan(String schedulingXmlFile, Config config, ScheduledEnvironment environment) {
+            super(config, environment);
+            this.schedulingXmlFile = schedulingXmlFile;
+            this.appJobListener = new ApplicationMonitorJobListener(config, getAppId() + "_LISTENER");
+        }
+
+        private InputStream getSchedulingXmlAsStream() {
+            String metaInfoSchedulingXmlFile = "/META-INF/providers/jobs/" + this.schedulingXmlFile;
+
+            InputStream inputStream = XmlSchedulingPlan.class.getResourceAsStream(metaInfoSchedulingXmlFile);
+            if (inputStream == null) {
+                inputStream = XmlSchedulingPlan.class.getResourceAsStream(this.schedulingXmlFile);
+            }
+            Preconditions.checkNotNull(inputStream, "Unable to load resource " + metaInfoSchedulingXmlFile);
+            return inputStream;
+        }
+
+        @Override
+        public void schedule() throws SchedulerException {
+            try {
+                XMLStreamSchedulingDataProcessor processor = new XMLStreamSchedulingDataProcessor();
+                processor.processStreamWithConfigAndScheduleJobs(
+                    getSchedulingXmlAsStream(),
+                    this.getConfig(),
+                    getAppId(),
+                    getScheduler()
+                );
+                LOGGER.info("Adding job listener: {} for application: {}", this.appJobListener, this.getAppId());
+                this.getScheduler().getListenerManager().addJobListener(this.appJobListener,
+                        processor.getParsedJobs().stream().map(jobDetail -> KeyMatcher.keyEquals(jobDetail.getKey())).collect(Collectors.toList()));
+            } catch (ParserConfigurationException | IOException | XPathException | ParseException | SAXException | ValidationException | ClassNotFoundException e) {
+                LOGGER.error("Failed to schedule {}: {}", this.getAppId(), e.getMessage(), e);
+                throw new SchedulerException(e.getMessage(), e);
+            }
+        }
+
+        @Override
+        public boolean unschedule() throws SchedulerException {
+            try {
+                XMLStreamSchedulingDataProcessor processor = new XMLStreamSchedulingDataProcessor();
+                processor.processStreamWithConfigAndUnscheduleJobs(
+                    getSchedulingXmlAsStream(),
+                    getConfig(),
+                    getAppId(),
+                    getScheduler()
+                );
+
+                LOGGER.info("Adding job listener: {} for application: {}", this.appJobListener, this.getAppId());
+                this.getScheduler().getListenerManager().removeJobListener(this.appJobListener.getName());
+                return true;
+            } catch (ParserConfigurationException | IOException | XPathException | ParseException | SAXException | ValidationException | ClassNotFoundException e) {
+                LOGGER.error("Failed to unschedule {}: {}", this.getAppId(), e.getMessage(), e);
+                throw new SchedulerException(e.getMessage(), e);
+            }
+        }
+
+        @Override
+        public boolean scheduling() throws SchedulerException {
+            try {
+                XMLStreamSchedulingDataProcessor processor = new XMLStreamSchedulingDataProcessor();
+                processor.processStreamWithConfig(
+                    getSchedulingXmlAsStream(),
+                    getConfig(),
+                    getAppId()
+                );
+                boolean scheduling = false;
+                for (MutableTrigger trigger : processor.getParsedTriggers()) {
+                    if (this.getScheduler().checkExists(trigger.getKey())) {
+                        LOGGER.debug("Job {} is scheduling with trigger {}", trigger.getJobKey(), trigger.getKey());
+                        scheduling = true;
+                    } else {
+                        LOGGER.debug("Trigger {} not exists", trigger.getJobKey(), trigger.getKey());
+                    }
+                }
+                return scheduling;
+            } catch (ParserConfigurationException | IOException | XPathException | ParseException | SAXException | ValidationException | ClassNotFoundException e) {
+                LOGGER.error("Failed to check {} scheduling status: {}", this.getAppId(), e.getMessage(), e);
+                throw new SchedulerException(e.getMessage(), e);
+            }
+        }
+    }
+}

eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/config/ApplicationConfig.java

@@ -0,0 +1,67 @@
+/*
+ * 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.config;
+
+import org.apache.eagle.metadata.model.ApplicationEntity;
+
+import java.io.Serializable;
+
+public class ApplicationConfig implements Serializable {
+    @Config("mode")
+    private ApplicationEntity.Mode mode;
+
+    @Config("siteId")
+    private String siteId;
+
+    @Config("appId")
+    private String appId;
+
+    @Config("jarPath")
+    private String jarPath;
+
+    public ApplicationEntity.Mode getMode() {
+        return mode;
+    }
+
+    public void setMode(ApplicationEntity.Mode mode) {
+        this.mode = mode;
+    }
+
+    public String getAppId() {
+        return appId;
+    }
+
+    public void setAppId(String appId) {
+        this.appId = appId;
+    }
+
+    public String getJarPath() {
+        return jarPath;
+    }
+
+    public void setJarPath(String jarPath) {
+        this.jarPath = jarPath;
+    }
+
+    public String getSiteId() {
+        return siteId;
+    }
+
+    public void setSiteId(String siteId) {
+        this.siteId = siteId;
+    }
+}

eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/config/ApplicationConfigMeta.java

@@ -0,0 +1,71 @@
+/*
+ * 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.config;
+
+import java.lang.reflect.Field;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+
+class ApplicationConfigMeta<T extends Object> {
+    private final Class<? extends T> configurationClass;
+    private final Map<Field, Config> fieldProperties;
+    private final Map<String, Field> fieldNameFields;
+
+
+    ApplicationConfigMeta(Class<? extends T> configurationClass) {
+        this.configurationClass = configurationClass;
+        fieldNameFields = new HashMap<>();
+        fieldProperties = new HashMap<>();
+        load();
+    }
+
+    private void load() {
+        Field[] fields = this.configurationClass.getDeclaredFields();
+        for (Field field : fields) {
+            fieldNameFields.put(field.getName(), field);
+            if (field.isAnnotationPresent(Config.class)) {
+                Config config = field.getAnnotation(Config.class);
+                fieldProperties.put(field, config);
+            }
+        }
+    }
+
+    public boolean hasAnyDeclaredProperties() {
+        return fieldProperties.size() > 0;
+    }
+
+    public Class<? extends T> getConfigClass() {
+        return configurationClass;
+    }
+
+    public Set<String> getFieldNames() {
+        return fieldNameFields.keySet();
+    }
+
+    public Field getField(String field) {
+        return fieldNameFields.get(field);
+    }
+
+    public Map<Field, Config> getFieldProperties() {
+        return fieldProperties;
+    }
+
+    public T newInstance() throws IllegalAccessException, InstantiationException {
+        return configurationClass.newInstance();
+    }
+}

eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/config/Config.java

@@ -0,0 +1,32 @@
+/*
+ * 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.config;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+@Target({ElementType.FIELD})
+@Retention(RetentionPolicy.RUNTIME)
+public @interface Config {
+    String value() default "";
+    String name() default "";
+    boolean required() default false;
+    String displayName() default "";
+    String description() default "";
+}