AMBARI-8400 - Alerts: Template Engine for Dispatching (jonathanhurley)

ambari-server/src/main/java/org/apache/ambari/server/AmbariService.java

@@ -0,0 +1,43 @@
+/**
+ * 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.ambari.server;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+import com.google.common.util.concurrent.AbstractScheduledService;
+import com.google.common.util.concurrent.ServiceManager;
+import com.google.inject.ScopeAnnotation;
+import com.google.inject.Singleton;
+
+/**
+ * The {@link AmbariService} annotation is used to register a class that
+ * implements Guava's {@link AbstractScheduledService} with the
+ * {@link ServiceManager}.
+ * <p/>
+ * Classes with this annotation are bound as singletons and automatically
+ * injected with their members. There is not need to use {@link Singleton} or
+ * {@link EagerSingleton}.
+ */
+@Retention(RetentionPolicy.RUNTIME)
+@Target({ ElementType.TYPE })
+@ScopeAnnotation
+public @interface AmbariService {
+}

ambari-server/src/main/java/org/apache/ambari/server/api/query/render/AlertSummaryGroupedRenderer.java

@@ -120,7 +120,7 @@ public class AlertSummaryGroupedRenderer extends AlertSummaryRenderer {
     for (TreeNode<Resource> node : resultTree.getChildren()) {
       Resource resource = node.getObject();
 
-      Long definitionId = (Long) resource.getPropertyValue(AlertResourceProvider.ALERT_ID);
+      Long definitionId = (Long) resource.getPropertyValue(AlertResourceProvider.ALERT_DEFINITION_ID);
       String definitionName = (String) resource.getPropertyValue(AlertResourceProvider.ALERT_DEFINITION_NAME);
       AlertState state = (AlertState) resource.getPropertyValue(AlertResourceProvider.ALERT_STATE);
       Long originalTimestampObject = (Long) resource.getPropertyValue(AlertResourceProvider.ALERT_ORIGINAL_TIMESTAMP);

ambari-server/src/main/java/org/apache/ambari/server/configuration/Configuration.java

@@ -331,8 +331,14 @@ public class Configuration {
 
   private static final String SERVER_HTTP_SESSION_INACTIVE_TIMEOUT = "server.http.session.inactive_timeout";
 
+  /**
+   * The full path to the XML file that describes the different alert templates.
+   */
+  private static final String ALERT_TEMPLATE_FILE = "alerts.template.file";
+
   private static final Logger LOG = LoggerFactory.getLogger(
       Configuration.class);
+
   private Properties properties;
   private Map<String, String> configsMap;
   private CredentialProvider credentialProvider = null;
@@ -810,8 +816,9 @@ public class Configuration {
     String passwdProp = properties.getProperty(SCOM_JDBC_SINK_USER_PASSWD_KEY);
     if (passwdProp != null) {
       String dbpasswd = readPasswordFromStore(passwdProp);
-      if (dbpasswd != null)
+      if (dbpasswd != null) {
         return dbpasswd;
+      }
     }
     return readPasswordFromFile(passwdProp, SCOM_JDBC_SINK_USER_PASSWD_DEFAULT);
   }
@@ -1209,4 +1216,14 @@ public class Configuration {
         SERVER_HTTP_SESSION_INACTIVE_TIMEOUT,
         "1800"));
   }
+
+  /**
+   * Gets the location of the XML alert template file which contains the
+   * velocity templates for outbound notifications.
+   *
+   * @return the location of the template file, or {@code null} if not defined.
+   */
+  public String getAlertTemplateFile() {
+    return properties.getProperty(ALERT_TEMPLATE_FILE);
+  }
 }

ambari-server/src/main/java/org/apache/ambari/server/controller/ControllerModule.java

@@ -34,6 +34,7 @@ import static org.eclipse.persistence.config.PersistenceUnitProperties.THROW_EXC
 
 import java.lang.annotation.Annotation;
 import java.security.SecureRandom;
+import java.text.MessageFormat;
 import java.util.Arrays;
 import java.util.HashSet;
 import java.util.List;
@@ -42,6 +43,7 @@ import java.util.Map.Entry;
 import java.util.Properties;
 import java.util.Set;
 
+import org.apache.ambari.server.AmbariService;
 import org.apache.ambari.server.EagerSingleton;
 import org.apache.ambari.server.StaticallyInject;
 import org.apache.ambari.server.actionmanager.ActionDBAccessor;
@@ -91,7 +93,6 @@ import org.apache.ambari.server.state.host.HostImpl;
 import org.apache.ambari.server.state.scheduler.RequestExecution;
 import org.apache.ambari.server.state.scheduler.RequestExecutionFactory;
 import org.apache.ambari.server.state.scheduler.RequestExecutionImpl;
-import org.apache.ambari.server.state.services.AlertNoticeDispatchService;
 import org.apache.ambari.server.state.stack.OsFamily;
 import org.apache.ambari.server.state.svccomphost.ServiceComponentHostImpl;
 import org.apache.ambari.server.view.ViewInstanceHandlerList;
@@ -109,6 +110,7 @@ import org.springframework.security.crypto.password.StandardPasswordEncoder;
 import org.springframework.util.ClassUtils;
 import org.springframework.web.filter.DelegatingFilterProxy;
 
+import com.google.common.util.concurrent.AbstractScheduledService;
 import com.google.common.util.concurrent.ServiceManager;
 import com.google.gson.Gson;
 import com.google.gson.GsonBuilder;
@@ -242,7 +244,6 @@ public class ControllerModule extends AbstractModule {
 
     requestStaticInjection(ExecutionCommandWrapper.class);
 
-    bindServices();
     bindByAnnotation();
   }
 
@@ -324,46 +325,26 @@ public class ControllerModule extends AbstractModule {
   }
 
   /**
-   * Bind all {@link com.google.common.util.concurrent.Service} singleton
-   * instances and then register them with a singleton {@link ServiceManager}.
-   */
-  private void bindServices() {
-    Set<com.google.common.util.concurrent.Service> services =
-        new HashSet<com.google.common.util.concurrent.Service>();
-
-    AlertNoticeDispatchService alertNoticeDispatchService =
-        new AlertNoticeDispatchService();
-
-    bind(AlertNoticeDispatchService.class).toInstance(
-        alertNoticeDispatchService);
-
-    services.add(alertNoticeDispatchService);
-    ServiceManager manager = new ServiceManager(services);
-    bind(ServiceManager.class).toInstance(manager);
-  }
-
-  /**
-   * Initializes specially-marked interfaces that require injection.  All eager singletons that should be instantiated as soon as
-   * possible and not wait for injection.
+   * Initializes specially-marked interfaces that require injection.
    * <p/>
    * An example of where this is needed is with a singleton that is headless; in
    * other words, it doesn't have any injections but still needs to be part of
    * the Guice framework.
    * <p/>
-   * <p/>
    * A second example of where this is needed is when classes require static
    * members that are available via injection.
    * <p/>
    * This currently scans {@code org.apache.ambari.server} for any
-   * {@link EagerSingleton} or {@link StaticallyInject} instances.
+   * {@link EagerSingleton} or {@link StaticallyInject} or {@link AmbariService}
+   * instances.
    */
+  @SuppressWarnings("unchecked")
   private void bindByAnnotation() {
     ClassPathScanningCandidateComponentProvider scanner =
         new ClassPathScanningCandidateComponentProvider(false);
 
-    @SuppressWarnings("unchecked")
-    List<Class<? extends Annotation>> classes = Arrays.asList(EagerSingleton.class,
-        StaticallyInject.class);
+    List<Class<? extends Annotation>> classes = Arrays.asList(
+        EagerSingleton.class, StaticallyInject.class, AmbariService.class);
 
     // match only singletons that are eager listeners
     for (Class<? extends Annotation> cls : classes) {
@@ -377,6 +358,9 @@ public class ControllerModule extends AbstractModule {
       return;
     }
 
+    Set<com.google.common.util.concurrent.Service> services =
+        new HashSet<com.google.common.util.concurrent.Service>();
+
     for (BeanDefinition beanDefinition : beanDefinitions) {
       String className = beanDefinition.getBeanClassName();
       Class<?> clazz = ClassUtils.resolveClassName(className,
@@ -391,7 +375,34 @@ public class ControllerModule extends AbstractModule {
         requestStaticInjection(clazz);
         LOG.debug("Statically injecting {} ", clazz);
       }
+
+      // Ambari services are registered with Guava
+      if (null != clazz.getAnnotation(AmbariService.class)) {
+        // safety check to ensure it's actually a Guava service
+        if (!AbstractScheduledService.class.isAssignableFrom(clazz)) {
+          String message = MessageFormat.format(
+              "Unable to register service {0} because it is not an AbstractScheduledService",
+              clazz);
+
+          LOG.warn(message);
+          throw new RuntimeException(message);
+        }
+
+        // instantiate the service, register as singleton via toInstance()
+        AbstractScheduledService service = null;
+        try {
+          service = (AbstractScheduledService) clazz.newInstance();
+          bind((Class<AbstractScheduledService>) clazz).toInstance(service);
+          services.add(service);
+          LOG.debug("Registering service {} ", clazz);
+        } catch (Exception exception) {
+          LOG.error("Unable to register {} as a service", clazz, exception);
+          throw new RuntimeException(exception);
+        }
+      }
     }
-  }
 
-}
+    ServiceManager manager = new ServiceManager(services);
+    bind(ServiceManager.class).toInstance(manager);
+  }
+}

ambari-server/src/main/java/org/apache/ambari/server/notifications/DispatchFactory.java

@@ -31,7 +31,7 @@ import com.google.inject.Singleton;
  * {@link NotificationDispatcher} based on a supplied type.
  */
 @Singleton
-public final class DispatchFactory {
+public class DispatchFactory {
 
   /**
    * Mapping of dispatch type to dispatcher singleton.

ambari-server/src/main/java/org/apache/ambari/server/notifications/dispatchers/EmailDispatcher.java

@@ -22,7 +22,6 @@ import java.util.Properties;
 import java.util.Timer;
 
 import javax.mail.Authenticator;
-import javax.mail.Message;
 import javax.mail.Message.RecipientType;
 import javax.mail.MessagingException;
 import javax.mail.PasswordAuthentication;
@@ -113,7 +112,8 @@ public class EmailDispatcher implements NotificationDispatcher {
     session = Session.getInstance(properties, authenticator);
 
     try {
-      Message message = new MimeMessage(session);
+      // !!! at some point in the future we can worry about multipart
+      MimeMessage message = new MimeMessage(session);
 
       for (Recipient recipient : notification.Recipients) {
         InternetAddress address = new InternetAddress(recipient.Identifier);
@@ -121,7 +121,7 @@ public class EmailDispatcher implements NotificationDispatcher {
       }
 
       message.setSubject(notification.Subject);
-      message.setText(notification.Body);
+      message.setText(notification.Body, "UTF-8", "html");
 
       Transport.send(message);
 

ambari-server/src/main/java/org/apache/ambari/server/state/services/AlertNoticeDispatchService.java

@@ -17,20 +17,36 @@
  */
 package org.apache.ambari.server.state.services;
 
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.StringWriter;
+import java.io.Writer;
 import java.lang.reflect.Type;
-import java.text.MessageFormat;
 import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Set;
+import java.util.concurrent.Executor;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
+import javax.xml.bind.JAXBContext;
+import javax.xml.bind.Unmarshaller;
+import javax.xml.bind.annotation.XmlAttribute;
+import javax.xml.bind.annotation.XmlElement;
+import javax.xml.bind.annotation.XmlRootElement;
+
+import org.apache.ambari.server.AmbariService;
+import org.apache.ambari.server.api.services.AmbariMetaInfo;
+import org.apache.ambari.server.configuration.Configuration;
 import org.apache.ambari.server.events.AlertEvent;
 import org.apache.ambari.server.notifications.DispatchCallback;
 import org.apache.ambari.server.notifications.DispatchCredentials;
@@ -45,6 +61,9 @@ import org.apache.ambari.server.orm.entities.AlertNoticeEntity;
 import org.apache.ambari.server.orm.entities.AlertTargetEntity;
 import org.apache.ambari.server.state.AlertState;
 import org.apache.ambari.server.state.NotificationState;
+import org.apache.commons.io.IOUtils;
+import org.apache.velocity.VelocityContext;
+import org.apache.velocity.app.Velocity;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -59,7 +78,7 @@ import com.google.gson.JsonObject;
 import com.google.gson.JsonParseException;
 import com.google.gson.reflect.TypeToken;
 import com.google.inject.Inject;
-import com.google.inject.Singleton;
+import com.google.inject.Provider;
 
 /**
  * The {@link AlertNoticeDispatchService} is used to scan the database for
@@ -69,15 +88,31 @@ import com.google.inject.Singleton;
  * The dispatch system will then make a callback to
  * {@link AlertNoticeDispatchCallback} so that the {@link NotificationState} can
  * be updated to its final value.
+ * <p/>
+ * This class uses the templates that are defined via
+ * {@link Configuration#getAlertTemplateFile()} or the fallback internal
+ * template {@code alert-templates.xml}. These files are parsed during
+ * {@link #startUp()}. If there is a problem parsing them, the service will
+ * still startup normally, producing an error in logs. It will fall back to
+ * simple string concatenation for {@link Notification} content in this case.
  */
-@Singleton
+@AmbariService
 public class AlertNoticeDispatchService extends AbstractScheduledService {
-
   /**
    * Logger.
    */
   private static final Logger LOG = LoggerFactory.getLogger(AlertNoticeDispatchService.class);
 
+  /**
+   * The log tag to pass to Apache Velocity during rendering.
+   */
+  private static final String VELOCITY_LOG_TAG = "ambari-alerts";
+
+  /**
+   * The internal Ambari templates that ship.
+   */
+  private static final String AMBARI_ALERT_TEMPLATES = "alert-templates.xml";
+
   /**
    * The property containing the dispatch authentication username.
    */
@@ -105,16 +140,34 @@ public class AlertNoticeDispatchService extends AbstractScheduledService {
   private AlertDispatchDAO m_dao;
 
   /**
-   * The factory used to get an {@link NotificationDispatcher} instance to submit to the
-   * {@link #m_executor}.
+   * The factory used to get an {@link NotificationDispatcher} instance to
+   * submit to the {@link #m_executor}.
    */
   @Inject
   private DispatchFactory m_dispatchFactory;
 
+  /**
+   * The alert templates to use when rendering content for a
+   * {@link Notification}.
+   */
+  private AlertTemplates m_alertTemplates;
+
+  /**
+   * The configuration instance to get Ambari properties.
+   */
+  @Inject
+  private Configuration m_configuration;
+
+  /**
+   * Ambari meta information used fro alert {@link Notification}s.
+   */
+  @Inject
+  private Provider<AmbariMetaInfo> m_metaInfo;
+
   /**
    * The executor responsible for dispatching.
    */
-  private final ThreadPoolExecutor m_executor;
+  private Executor m_executor;
 
   /**
    * Constructor.
@@ -132,6 +185,67 @@ public class AlertNoticeDispatchService extends AbstractScheduledService {
     m_gson = gsonBuilder.create();
   }
 
+  /**
+   * {@inheritDoc}
+   * <p/>
+   * Parse the XML template for {@link Notification} content. If there is a
+   * problem parsing the content, the service will still startup normally but
+   * the {@link Notification} content will fallback to plaintext.
+   */
+  @Override
+  protected void startUp() throws Exception {
+    super.startUp();
+
+    InputStream inputStream = null;
+    String alertTemplatesFile = null;
+
+    try {
+      alertTemplatesFile = m_configuration.getAlertTemplateFile();
+      if (null != alertTemplatesFile) {
+        File file = new File(alertTemplatesFile);
+        inputStream = new FileInputStream(file);
+      }
+    } catch (Exception exception) {
+      LOG.warn("Unable to load alert template file {}", alertTemplatesFile,
+          exception);
+    }
+
+    try {
+      JAXBContext context = JAXBContext.newInstance(AlertTemplates.class);
+      Unmarshaller unmarshaller = context.createUnmarshaller();
+
+      // if the file provided via the configuration is not available, use
+      // the internal one
+      if (null == inputStream) {
+        inputStream = ClassLoader.getSystemResourceAsStream(AMBARI_ALERT_TEMPLATES);
+      }
+
+      m_alertTemplates = (AlertTemplates) unmarshaller.unmarshal(inputStream);
+    } catch (Exception exception) {
+      LOG.error(
+          "Unable to load alert template file {}, outbound notifications will not be formatted",
+          AMBARI_ALERT_TEMPLATES,
+          exception);
+    } finally {
+      if (null != inputStream) {
+        IOUtils.closeQuietly(inputStream);
+      }
+    }
+  }
+
+
+  /**
+   * Sets the {@link Executor} to use when dispatching {@link Notification}s.
+   * This should only be used by unit tests to provide a mock executor.
+   *
+   * @param executor
+   *          the executor to use (not {@code null).
+
+   */
+  protected void setExecutor(Executor executor) {
+    m_executor = executor;
+  }
+
   /**
    * {@inheritDoc}
    */
@@ -145,10 +259,10 @@ public class AlertNoticeDispatchService extends AbstractScheduledService {
     LOG.info("There are {} pending alert notices about to be dispatched..."
         + pending.size());
 
-    // combine all histories by target
-    Map<AlertTargetEntity, List<AlertNoticeEntity>> aggregateMap = new HashMap<AlertTargetEntity, List<AlertNoticeEntity>>(
-        pending.size());
+    Map<AlertTargetEntity, List<AlertNoticeEntity>> aggregateMap =
+        new HashMap<AlertTargetEntity, List<AlertNoticeEntity>>(pending.size());
 
+    // combine all histories by target
     for (AlertNoticeEntity notice : pending) {
       AlertTargetEntity target = notice.getAlertTarget();
 
@@ -161,6 +275,7 @@ public class AlertNoticeDispatchService extends AbstractScheduledService {
       notices.add(notice);
     }
 
+    // now that all of the notices are grouped by target, dispatch them
     Set<AlertTargetEntity> targets = aggregateMap.keySet();
     for (AlertTargetEntity target : targets) {
       List<AlertNoticeEntity> notices = aggregateMap.get(target);
@@ -168,7 +283,9 @@ public class AlertNoticeDispatchService extends AbstractScheduledService {
         continue;
       }
 
+      String targetType = target.getNotificationType();
       String propertiesJson = target.getProperties();
+
       AlertTargetProperties targetProperties = m_gson.fromJson(propertiesJson,
           AlertTargetProperties.class);
 
@@ -178,47 +295,37 @@ public class AlertNoticeDispatchService extends AbstractScheduledService {
       notification.Callback = new AlertNoticeDispatchCallback();
       notification.CallbackIds = new ArrayList<String>(notices.size());
 
-      // !!! FIXME: temporary until velocity templates are implemented
-      String subject = "OK ({0}), Warning ({1}), Critical ({2})";
-      StringBuilder buffer = new StringBuilder(512);
-
-      int okCount = 0;
-      int warningCount = 0;
-      int criticalCount = 0;
+      List<AlertHistoryEntity> histories = new ArrayList<AlertHistoryEntity>(
+          notices.size());
 
+      // add callback IDs so that the notices can be marked as DELIVERED or
+      // FAILED, and create a list of just the alert histories
       for (AlertNoticeEntity notice : notices) {
         AlertHistoryEntity history = notice.getAlertHistory();
+        histories.add(history);
+
         notification.CallbackIds.add(notice.getUuid());
+      }
 
-        AlertState alertState = history.getAlertState();
-        switch (alertState) {
-          case CRITICAL:
-            criticalCount++;
-            break;
-          case OK:
-            okCount++;
-            break;
-          case UNKNOWN:
-            // !!! hmmmmmm
-            break;
-          case WARNING:
-            warningCount++;
-            break;
-          default:
-            break;
+      // populate the subject and body fields; if there is a problem
+      // generating the content, then mark the notices as FAILED
+      try {
+        renderNotificationContent(notification, histories, target);
+      } catch (Exception exception) {
+        LOG.error("Unable to create notification for alerts", exception);
+
+        // there was a problem generating content for the target; mark all
+        // notices as FAILED and skip this target
+        List<String> failedNoticeIds = new ArrayList<String>(notices.size());
+        for (AlertNoticeEntity notice : notices) {
+          failedNoticeIds.add(notice.getUuid());
         }
 
-        buffer.append(history.getAlertLabel());
-        buffer.append(": ");
-        buffer.append(history.getAlertText());
-        buffer.append("\n");
+        // mark these as failed
+        notification.Callback.onFailure(failedNoticeIds);
+        continue;
       }
 
-      notification.Subject = MessageFormat.format(subject, okCount,
-          warningCount, criticalCount);
-
-      notification.Body = buffer.toString();
-
       // set dispatch credentials
       if (properties.containsKey(AMBARI_DISPATCH_CREDENTIAL_USERNAME)
           && properties.containsKey(AMBARI_DISPATCH_CREDENTIAL_PASSWORD)) {
@@ -228,6 +335,7 @@ public class AlertNoticeDispatchService extends AbstractScheduledService {
         notification.Credentials = credentials;
       }
 
+      // create recipients
       if (null != targetProperties.Recipients) {
         List<Recipient> recipients = new ArrayList<Recipient>(
             targetProperties.Recipients.size());
@@ -244,7 +352,8 @@ public class AlertNoticeDispatchService extends AbstractScheduledService {
       // set all other dispatch properties
       notification.DispatchProperties = properties;
 
-      NotificationDispatcher dispatcher = m_dispatchFactory.getDispatcher(target.getNotificationType());
+      // dispatch
+      NotificationDispatcher dispatcher = m_dispatchFactory.getDispatcher(targetType);
       DispatchRunnable runnable = new DispatchRunnable(dispatcher, notification);
 
       m_executor.execute(runnable);
@@ -254,12 +363,74 @@ public class AlertNoticeDispatchService extends AbstractScheduledService {
   /**
    * {@inheritDoc}
    * <p/>
-   * Returns a schedule that starts after 1 minute and runs every minute after
-   * {@link #runOneIteration()} completes.
+   * Returns a schedule that starts after 2 minute and runs every 2 minutes
+   * after {@link #runOneIteration()} completes.
    */
   @Override
   protected Scheduler scheduler() {
-    return Scheduler.newFixedDelaySchedule(1, 1, TimeUnit.MINUTES);
+    return Scheduler.newFixedDelaySchedule(2, 2, TimeUnit.MINUTES);
+  }
+
+  /**
+   * Generates the content for the {@link Notification} from the
+   * {@link #m_alertTemplates}. If there is a problem with the templates, this
+   * will fallback to non-formatted content.
+   *
+   * @param notification
+   *          the notification (not {@code null}).
+   * @param histories
+   *          the alerts to generate the content fron (not {@code null}.
+   * @param target
+   *          the target of the {@link Notification}.
+   */
+  private void renderNotificationContent(Notification notification,
+      List<AlertHistoryEntity> histories, AlertTargetEntity target)
+      throws IOException {
+    String targetType = target.getNotificationType();
+
+    // build the velocity objects for template rendering
+    AmbariInfo ambari = new AmbariInfo(m_metaInfo.get());
+    AlertInfo summary = new AlertInfo(histories);
+    DispatchInfo dispatch = new DispatchInfo(target);
+
+    // get the template for this target type
+    final Writer subjectWriter = new StringWriter();
+    final Writer bodyWriter = new StringWriter();
+    final AlertTemplate template = m_alertTemplates.getTemplate(targetType);
+    if (null != template) {
+      // create the velocity context for template rendering
+      VelocityContext velocityContext = new VelocityContext();
+      velocityContext.put("ambari", ambari);
+      velocityContext.put("summary", summary);
+      velocityContext.put("dispatch", dispatch);
+
+      // render the template and assign the content to the notification
+      String subjectTemplate = template.getSubject();
+      String bodyTemplate = template.getBody();
+
+      // render the subject
+      Velocity.evaluate(velocityContext, subjectWriter, VELOCITY_LOG_TAG,
+          subjectTemplate);
+
+      // render the body
+      Velocity.evaluate(velocityContext, bodyWriter, VELOCITY_LOG_TAG,
+          bodyTemplate);
+    } else {
+      // a null template is possible from parsing incorrectly or not
+      // having the correct type defined for the target
+      for (AlertHistoryEntity alert : histories) {
+        subjectWriter.write("Apache Ambari Alert Summary");
+        bodyWriter.write(alert.getAlertState().name());
+        bodyWriter.write(" ");
+        bodyWriter.write(alert.getAlertDefinition().getLabel());
+        bodyWriter.write(" ");
+        bodyWriter.write(alert.getAlertText());
+        bodyWriter.write("\n");
+      }
+    }
+
+    notification.Subject = subjectWriter.toString();
+    notification.Body = bodyWriter.toString();
   }
 
   /**
@@ -395,4 +566,398 @@ public class AlertNoticeDispatchService extends AbstractScheduledService {
       }
     }
   }
-}
+
+  /**
+   * The {@link AlertInfo} class encapsulates all of the alert information for
+   * the {@link Notification}. This includes customized structures to better
+   * organize information about each of the services, hosts, and alert states.
+   */
+  public final static class AlertInfo {
+    private int m_okCount = 0;
+    private int m_warningCount = 0;
+    private int m_criticalCount = 0;
+    private int m_unknownCount = 0;
+
+    /**
+     * The hosts that have at least 1 alert reported.
+     */
+    private final Set<String> m_hosts = new HashSet<String>();
+
+    /**
+     * The services that have at least 1 alert reported.
+     */
+    private final Set<String> m_services = new HashSet<String>();
+
+    /**
+     * All of the alerts for the {@link Notification}.
+     */
+    private final List<AlertHistoryEntity> m_alerts;
+
+    /**
+     * A mapping of service to alerts where the alerts are also grouped by state
+     * for that service.
+     */
+    private final Map<String, Map<AlertState, List<AlertHistoryEntity>>> m_alertsByServiceAndState =
+        new HashMap<String, Map<AlertState, List<AlertHistoryEntity>>>();
+
+    /**
+     * A mapping of all services by state.
+     */
+    private final Map<String, Set<String>> m_servicesByState = new HashMap<String, Set<String>>();
+
+    /**
+     * A mapping of all alerts by the service that owns them.
+     */
+    private final Map<String, List<AlertHistoryEntity>> m_alertsByService = new HashMap<String, List<AlertHistoryEntity>>();
+
+    /**
+     * Constructor.
+     *
+     * @param histories
+     */
+    protected AlertInfo(List<AlertHistoryEntity> histories) {
+      m_alerts = histories;
+
+      // group all alerts by their service and severity
+      for (AlertHistoryEntity history : m_alerts) {
+        AlertState alertState = history.getAlertState();
+        String serviceName = history.getServiceName();
+        String hostName = history.getHostName();
+
+        if( null != hostName ){
+          m_hosts.add(hostName);
+        }
+
+        if (null != serviceName) {
+          m_services.add(serviceName);
+        }
+
+        // group alerts by service name & state
+        Map<AlertState, List<AlertHistoryEntity>> service = m_alertsByServiceAndState.get(serviceName);
+        if (null == service) {
+          service = new HashMap<AlertState, List<AlertHistoryEntity>>();
+          m_alertsByServiceAndState.put(serviceName, service);
+        }
+
+        List<AlertHistoryEntity> alertList = service.get(alertState);
+        if (null == alertList) {
+          alertList = new ArrayList<AlertHistoryEntity>();
+          service.put(alertState, alertList);
+        }
+
+        alertList.add(history);
+
+        // group services by alert states
+        Set<String> services = m_servicesByState.get(alertState.name());
+        if (null == services) {
+          services = new HashSet<String>();
+          m_servicesByState.put(alertState.name(), services);
+        }
+
+        services.add(serviceName);
+
+        // group alerts by service
+        List<AlertHistoryEntity> alertsByService = m_alertsByService.get(serviceName);
+        if (null == alertsByService) {
+          alertsByService = new ArrayList<AlertHistoryEntity>();
+          m_alertsByService.put(serviceName, alertsByService);
+        }
+
+        alertsByService.add(history);
+
+        // keep track of totals
+        switch (alertState) {
+          case CRITICAL:
+            m_criticalCount++;
+            break;
+          case OK:
+            m_okCount++;
+            break;
+          case UNKNOWN:
+            m_unknownCount++;
+            break;
+          case WARNING:
+            m_warningCount++;
+            break;
+          default:
+            m_unknownCount++;
+            break;
+        }
+      }
+    }
+
+    /**
+     * Gets the total number of OK alerts in the {@link Notification}.
+     *
+     * @return the OK count.
+     */
+    public int getOkCount() {
+      return m_okCount;
+    }
+
+    /**
+     * Gets the total number of WARNING alerts in the {@link Notification}.
+     *
+     * @return the WARNING count.
+     */
+    public int getWarningCount() {
+      return m_warningCount;
+    }
+
+    /**
+     * Gets the total number of CRITICAL alerts in the {@link Notification}.
+     *
+     * @return the CRITICAL count.
+     */
+    public int getCriticalCount() {
+      return m_criticalCount;
+    }
+
+    /**
+     * Gets the total number of UNKNOWN alerts in the {@link Notification}.
+     *
+     * @return the UNKNOWN count.
+     */
+    public int getUnknownCount() {
+      return m_unknownCount;
+    }
+
+    /**
+     * Gets the total count of all alerts in the {@link Notification}
+     *
+     * @return the total count of all alerts.
+     */
+    public int getTotalCount() {
+      return m_okCount + m_warningCount + m_criticalCount + m_unknownCount;
+    }
+
+    /**
+     * Gets all of the services that have alerts being reporting in this
+     * notification dispatch.
+     *
+     * @return the list of services.
+     */
+    public Set<String> getServices() {
+      return m_services;
+    }
+
+    /**
+     * Gets all of the alerts in the {@link Notification}.
+     *
+     * @return all of the alerts.
+     */
+    public List<AlertHistoryEntity> getAlerts() {
+      return m_alerts;
+    }
+
+    /**
+     * Gets all of the alerts in the {@link Notification} by service name.
+     *
+     * @param serviceName
+     *          the service name.
+     * @return the alerts for that service, or {@code null} none.
+     */
+    public List<AlertHistoryEntity> getAlerts(String serviceName) {
+      return m_alertsByService.get(serviceName);
+    }
+
+    /**
+     * Gets all of the alerts for a given service and alert state level.
+     *
+     * @param serviceName
+     *          the name of the service.
+     * @param alertState
+     *          the alert state level.
+     * @return the list of alerts or {@code null} for none.
+     */
+    public List<AlertHistoryEntity> getAlerts(String serviceName,
+        String alertState) {
+
+      Map<AlertState, List<AlertHistoryEntity>> serviceAlerts = m_alertsByServiceAndState.get(serviceName);
+      if (null == serviceAlerts) {
+        return null;
+      }
+
+      AlertState state = AlertState.valueOf(alertState);
+      return serviceAlerts.get(state);
+    }
+
+    /**
+     * Gets a list of services that have an alert being reporting for the given
+     * state.
+     *
+     * @param alertState
+     *          the state to get the services for.
+     * @return the services or {@code null} if none.
+     */
+    public Set<String> getServicesByAlertState(String alertState) {
+      return m_servicesByState.get(alertState);
+    }
+  }
+
+  /**
+   * The {@link AmbariInfo} class is used to provide the template engine with
+   * information about the Ambari installation.
+   */
+  public final static class AmbariInfo {
+    private String m_hostName = null;
+    private String m_url = null;
+    private String m_version = null;
+
+    /**
+     * Constructor.
+     *
+     * @param metaInfo
+     */
+    protected AmbariInfo(AmbariMetaInfo metaInfo) {
+      m_version = metaInfo.getServerVersion();
+    }
+
+    /**
+     * @return the hostName
+     */
+    public String getHostName() {
+      return m_hostName;
+    }
+
+    /**
+     * @return the url
+     */
+    public String getUrl() {
+      return m_url;
+    }
+
+    /**
+     * Gets the Ambari server version.
+     *
+     * @return the version
+     */
+    public String getServerVersion() {
+      return m_version;
+    }
+  }
+
+  /**
+   * The {@link DispatchInfo} class is used to provide the template engine with
+   * information about the intended target of the notification.
+   */
+  public static final class DispatchInfo {
+    private String m_targetName;
+    private String m_targetDescription;
+
+    /**
+     * Constructor.
+     *
+     * @param target
+     *          the {@link AlertTargetEntity} receiving the notification.
+     */
+    protected DispatchInfo(AlertTargetEntity target) {
+      m_targetName = target.getTargetName();
+      m_targetDescription = target.getDescription();
+    }
+
+    /**
+     * Gets the name of the notification target.
+     *
+     * @return the name of the target.
+     */
+    public String getTargetName() {
+      return m_targetName;
+    }
+
+    /**
+     * Gets the description of the notification target.
+     *
+     * @return the target description.
+     */
+    public String getTargetDescription() {
+      return m_targetDescription;
+    }
+  }
+
+  /**
+   * The {@link AlertTemplates} class represnts the {@link AlertTemplates} that
+   * have been loaded, either by the {@link Configuration} or by the backup
+   * {@code alert-templates.xml} file.
+   */
+  @XmlRootElement(name = "alert-templates")
+  private final static class AlertTemplates {
+    /**
+     * The alert templates defined.
+     */
+    @XmlElement(name = "alert-template", required = true)
+    private List<AlertTemplate> m_templates;
+
+    /**
+     * Gets the alert template given the specified template type.
+     *
+     * @param type
+     *          the template type.
+     * @return the template, or {@code null} if none.
+     * @see AlertTargetEntity#getNotificationType()
+     */
+    public AlertTemplate getTemplate(String type) {
+      for (AlertTemplate template : m_templates) {
+        if (type.equals(template.getType())) {
+          return template;
+        }
+      }
+
+      return null;
+    }
+  }
+
+  /**
+   * The {@link AlertTemplate} class represents a template for a specified alert
+   * target type that can be used when creating the content for dispatching
+   * {@link Notification}s.
+   */
+  private final static class AlertTemplate {
+    /**
+     * The type that this template is for.
+     *
+     * @see AlertTargetEntity#getNotificationType()
+     */
+    @XmlAttribute(name = "type", required = true)
+    private String m_type;
+
+    /**
+     * The subject template for the {@link Notification}.
+     */
+    @XmlElement(name = "subject", required = true)
+    private String m_subject;
+
+    /**
+     * The body template for the {@link Notification}.
+     */
+    @XmlElement(name = "body", required = true)
+    private String m_body;
+
+    /**
+     * Gets the template type.
+     *
+     * @return the template type.
+     */
+    public String getType() {
+      return m_type;
+    }
+
+    /**
+     * Gets the subject template.
+     *
+     * @return the subject template.
+     */
+    public String getSubject() {
+      return m_subject;
+    }
+
+    /**
+     * Gets the body template.
+     *
+     * @return the body template.
+     */
+    public String getBody() {
+      return m_body;
+    }
+  }
+}

ambari-server/src/main/resources/alert-templates.xml

@@ -0,0 +1,144 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+   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.
+-->
+<alert-templates>
+  <alert-template type="EMAIL">
+    <subject>
+      <![CDATA[Alert Summary: OK[$summary.getOkCount()], Warning[$summary.getWarningCount()], Critical[$summary.getCriticalCount()], Unknown[$summary.getUnknownCount()]]]>
+    </subject>
+    <body>
+      <![CDATA[
+#set( $alertStates = ["OK", "WARNING", "CRITICAL", "UNKNOWN"] )
+#set( $services = $summary.getServices() )
+<html>
+  <style type="text/css">
+    .ok{
+      color: green;
+    }
+    .warning{
+      color: ffff00;
+    }
+    .critical{
+      color: red;
+    }
+    .unknown{
+      color: brown;
+    }
+    .section{
+      background-color: #eee;
+      padding: 20px;
+      margin-top: 10px;
+      margin-left: 20px;
+      margin-right: 20px;
+      text-align: left;
+      border-radius: 10px;
+      font-family: Arial, Helvetica, sans-serif;
+    }
+    .footer{
+      margin-top: 10px;
+      margin-left: 20px;
+      margin-right: 20px;
+      font-family: Arial, Helvetica, sans-serif;
+      font-size: 12px;
+    }
+    .alert{    
+      margin: 5px;
+      font-weight: normal;
+      font-size: 14px;   
+    }
+    .alertText{
+      font-size: 12px;
+    }
+    h1{
+      margin: 0 0 0.5em 0;
+      color: #343434;
+      font-weight: normal;
+      font-size: 20px;
+    }
+    table {
+      width: 100%;
+    }
+    tr{
+      margin-botton: 5px;
+    }    
+    td{
+      text-align: left;
+      vertical-align: top;
+    }
+  </style>
+  <div class="section">
+    <h1>Services Reporting Alerts</h1>  
+      #foreach( $alertState in $alertStates )
+        #if( $summary.getServicesByAlertState($alertState)  )
+          <div class="alert">
+            <span class="$alertState">$alertState</span> $summary.getServicesByAlertState($alertState)
+          </div>
+        #end
+      #end
+  </div>
+
+  #foreach( $service in $services )
+    <div class="section">
+      <h1>$service</h1>
+      <table>
+        #foreach( $alertState in $alertStates )
+            #foreach( $alert in $summary.getAlerts($service,$alertState) )
+              <tr class="alert">
+                <td width="100px">
+                  <div class="$alertState">
+                    $alertState
+                  </div>
+                </td>
+                <td>
+                  $alert.getAlertDefinition().getLabel()
+                  <br/>
+                  <span class="alertText">
+                    $alert.getAlertText()
+                  <span>
+                </td>
+              </tr>
+            #end
+        #end
+      </table>
+    </div>
+  #end
+  <div class="footer">
+    This notification was sent to $dispatch.getTargetName()
+    <br/>
+    Apache Ambari $ambari.getServerVersion()
+  </div>
+</html>
+      ]]>
+    </body>
+  </alert-template>
+  <alert-template type="SNMP">
+    <subject>
+      <![CDATA[Alert Summary: OK[$summary.getOkCount()], Warning[$summary.getWarningCount()], Critical[$summary.getCriticalCount()], Unknown[$summary.getUnknownCount()]]]>
+    </subject>
+    <body>
+      <![CDATA[
+#set( $alertStates = ["OK", "WARNING", "CRITICAL", "UNKNOWN"] )
+#set( $services = $summary.getServices() )
+#foreach( $service in $services )
+#foreach( $alert in $summary.getAlerts($service) )
+[$service] [$alert.getAlertState()] [$alert.getAlertDefinition().getLabel()] [$alert.getAlertText()]
+#end
+#end
+      ]]>
+    </body>
+  </alert-template>  
+</alert-templates>

ambari-server/src/main/resources/stacks/BIGTOP/0.8/services/HDFS/alerts.json

@@ -372,7 +372,7 @@
     "DATANODE": [
       {
         "name": "datanode_process",
-        "label": "DateNode Process",
+        "label": "DataNode Process",
         "interval": 1,
         "scope": "HOST",
         "enabled": true,

ambari-server/src/main/resources/stacks/HDP/1.3.2/services/HDFS/alerts.json

@@ -349,7 +349,7 @@
     "DATANODE": [
       {
         "name": "datanode_process",
-        "label": "DateNode Process",
+        "label": "DataNode Process",
         "interval": 1,
         "scope": "HOST",
         "enabled": true,

ambari-server/src/main/resources/stacks/HDP/2.0.6/services/HDFS/alerts.json

@@ -386,8 +386,8 @@
     "DATANODE": [
       {
         "name": "datanode_process",
-        "label": "DateNode Process",
-        "description": "Checks that the DateNode process responds to a TCP port request.",
+        "label": "DataNode Process",
+        "description": "Checks that the DataNode process responds to a TCP port request.",
         "interval": 1,
         "scope": "HOST",
         "enabled": true,

ambari-server/src/test/java/org/apache/ambari/server/state/services/AlertNoticeDispatchServiceTest.java

@@ -0,0 +1,339 @@
+/**
+ * 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.ambari.server.state.services;
+
+import static org.easymock.EasyMock.createNiceMock;
+import static org.easymock.EasyMock.createStrictMock;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+import java.util.ArrayList;
+import java.util.Calendar;
+import java.util.EnumSet;
+import java.util.List;
+import java.util.TimeZone;
+import java.util.UUID;
+import java.util.concurrent.Executor;
+
+import org.apache.ambari.server.api.services.AmbariMetaInfo;
+import org.apache.ambari.server.notifications.DispatchFactory;
+import org.apache.ambari.server.notifications.Notification;
+import org.apache.ambari.server.notifications.NotificationDispatcher;
+import org.apache.ambari.server.orm.InMemoryDefaultTestModule;
+import org.apache.ambari.server.orm.dao.AlertDispatchDAO;
+import org.apache.ambari.server.orm.entities.AlertDefinitionEntity;
+import org.apache.ambari.server.orm.entities.AlertHistoryEntity;
+import org.apache.ambari.server.orm.entities.AlertNoticeEntity;
+import org.apache.ambari.server.orm.entities.AlertTargetEntity;
+import org.apache.ambari.server.state.AlertState;
+import org.apache.ambari.server.state.NotificationState;
+import org.apache.ambari.server.state.alert.Scope;
+import org.apache.ambari.server.state.alert.SourceType;
+import org.easymock.EasyMock;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import com.google.inject.Binder;
+import com.google.inject.Guice;
+import com.google.inject.Injector;
+import com.google.inject.Module;
+import com.google.inject.util.Modules;
+
+/**
+ * Tests the {@link AlertNoticeDispatchService}.
+ */
+public class AlertNoticeDispatchServiceTest extends AlertNoticeDispatchService {
+
+  final static String ALERT_NOTICE_UUID = UUID.randomUUID().toString();
+  final static String ALERT_UNIQUE_TEXT = "0eeda438-2b13-4869-a416-137e35ff76e9";
+  final static String HOSTNAME = "c6401.ambari.apache.org";
+  final static Calendar calendar = Calendar.getInstance(TimeZone.getTimeZone("UTC"));
+
+  private AmbariMetaInfo m_metaInfo = null;
+  private DispatchFactory m_dispatchFactory = null;
+  private AlertDispatchDAO m_dao = null;
+  private Injector m_injector;
+
+  List<AlertDefinitionEntity> m_definitions = new ArrayList<AlertDefinitionEntity>();
+  List<AlertHistoryEntity> m_histories = new ArrayList<AlertHistoryEntity>();
+
+  @Before
+  public void before() {
+    m_dao = createStrictMock(AlertDispatchDAO.class);
+    m_dispatchFactory = createStrictMock(DispatchFactory.class);
+    m_metaInfo = createNiceMock(AmbariMetaInfo.class);
+
+    // create an injector which will inject the mocks
+    m_injector = Guice.createInjector(Modules.override(
+        new InMemoryDefaultTestModule()).with(new MockModule()));
+
+    Assert.assertNotNull(m_injector);
+
+    // create 5 definitions
+    for (int i = 0; i < 5; i++) {
+      AlertDefinitionEntity definition = new AlertDefinitionEntity();
+      definition.setDefinitionName("Alert Definition " + i);
+      definition.setServiceName("Service " + i);
+      definition.setComponentName(null);
+      definition.setClusterId(1L);
+      definition.setHash(UUID.randomUUID().toString());
+      definition.setScheduleInterval(Integer.valueOf(60));
+      definition.setScope(Scope.SERVICE);
+      definition.setSource("{\"type\" : \"SCRIPT\"}");
+      definition.setSourceType(SourceType.SCRIPT);
+
+      m_definitions.add(definition);
+    }
+
+
+    // create 10 historical alerts for each definition, 8 OK and 2 CRIT
+    calendar.clear();
+    calendar.set(2014, Calendar.JANUARY, 1);
+
+    for (AlertDefinitionEntity definition : m_definitions) {
+      for (int i = 0; i < 10; i++) {
+        AlertHistoryEntity history = new AlertHistoryEntity();
+        history.setServiceName(definition.getServiceName());
+        history.setClusterId(1L);
+        history.setAlertDefinition(definition);
+        history.setAlertLabel(definition.getDefinitionName() + " " + i);
+        history.setAlertText(definition.getDefinitionName() + " " + i);
+        history.setAlertTimestamp(calendar.getTimeInMillis());
+        history.setHostName(HOSTNAME);
+
+        history.setAlertState(AlertState.OK);
+        if (i == 0 || i == 5) {
+          history.setAlertState(AlertState.CRITICAL);
+        }
+
+        // increase the days for each
+        calendar.add(Calendar.DATE, 1);
+        m_histories.add(history);
+      }
+    }
+  }
+
+  /**
+   * Tests the parsing of the {@link AlertHistoryEntity} list into
+   * {@link AlertInfo}.
+   *
+   * @throws Exception
+   */
+  @Test
+  public void testAlertInfo() throws Exception {
+    AlertInfo alertInfo = new AlertInfo(m_histories);
+    assertEquals(50, alertInfo.getAlerts().size());
+    assertEquals(10, alertInfo.getAlerts("Service 1").size());
+    assertEquals(10, alertInfo.getAlerts("Service 2").size());
+
+    assertEquals(8, alertInfo.getAlerts("Service 1", "OK").size());
+    assertEquals(2, alertInfo.getAlerts("Service 1", "CRITICAL").size());
+    assertNull(alertInfo.getAlerts("Service 1", "WARNING"));
+    assertNull(alertInfo.getAlerts("Service 1", "UNKNOWN"));
+
+    assertEquals(5, alertInfo.getServices().size());
+  }
+
+  /**
+   * Tests that the dispatcher is not called when there are no notices.
+   *
+   * @throws Exception
+   */
+  @Test
+  public void testNoDispatch() throws Exception {
+    EasyMock.expect(m_dao.findPendingNotices()).andReturn(
+        new ArrayList<AlertNoticeEntity>()).once();
+
+    // m_dispatchFactory should not be called at all
+    EasyMock.replay(m_dao, m_dispatchFactory);
+
+    // "startup" the service so that its initialization is done
+    AlertNoticeDispatchService service = m_injector.getInstance(AlertNoticeDispatchService.class);
+    service.startUp();
+
+    // service trigger
+    service.runOneIteration();
+
+    EasyMock.verify(m_dao, m_dispatchFactory);
+  }
+
+  /**
+   * Tests that the dispatcher is not called when there are no notices.
+   *
+   * @throws Exception
+   */
+  @Test
+  public void testDispatch() throws Exception {
+    MockDispatcher dispatcher = new MockDispatcher();
+
+    EasyMock.expect(m_dao.findPendingNotices()).andReturn(getMockNotices()).once();
+
+    EasyMock.expect(m_dispatchFactory.getDispatcher("EMAIL")).andReturn(
+        dispatcher).once();
+
+    EasyMock.replay(m_dao, m_dispatchFactory);
+
+    // "startup" the service so that its initialization is done
+    AlertNoticeDispatchService service = m_injector.getInstance(AlertNoticeDispatchService.class);
+    service.startUp();
+
+    // service trigger with mock executor that blocks
+    service.setExecutor(new MockExecutor());
+    service.runOneIteration();
+
+    EasyMock.verify(m_dao, m_dispatchFactory);
+
+    Notification notification = dispatcher.getNotification();
+    assertNotNull(notification);
+
+    assertTrue(notification.Subject.contains("OK[1]"));
+    assertTrue(notification.Subject.contains("Critical[0]"));
+    assertTrue(notification.Body.contains(ALERT_UNIQUE_TEXT));
+  }
+
+  /**
+   * Tests that a failed dispatch invokes the callback to mark the UUIDs of the
+   * notices as FAILED.
+   *
+   * @throws Exception
+   */
+  @Test
+  public void testFailedDispatch() throws Exception {
+    MockDispatcher dispatcher = new MockDispatcher();
+    List<AlertNoticeEntity> notices = getMockNotices();
+    AlertNoticeEntity notice = notices.get(0);
+
+    // these expectations happen b/c we need to mark the notice as FAILED
+    EasyMock.expect(m_dao.findPendingNotices()).andReturn(notices).once();
+    EasyMock.expect(m_dao.findNoticeByUuid(ALERT_NOTICE_UUID)).andReturn(notice).once();
+    EasyMock.expect(m_dao.merge(getMockNotices().get(0))).andReturn(notice).once();
+
+    EasyMock.replay(m_dao, m_dispatchFactory);
+
+    // do NOT startup the service which will force a template NPE
+    AlertNoticeDispatchService service = m_injector.getInstance(AlertNoticeDispatchService.class);
+
+    // service trigger with mock executor that blocks
+    service.setExecutor(new MockExecutor());
+    service.runOneIteration();
+
+    EasyMock.verify(m_dao, m_dispatchFactory);
+
+    Notification notification = dispatcher.getNotification();
+    assertNull(notification);
+  }
+
+  /**
+   * Gets PENDING notices.
+   *
+   * @return
+   */
+  private List<AlertNoticeEntity> getMockNotices(){
+    AlertHistoryEntity history = new AlertHistoryEntity();
+    history.setServiceName("HDFS");
+    history.setClusterId(1L);
+    history.setAlertDefinition(null);
+    history.setAlertLabel("Label");
+    history.setAlertState(AlertState.OK);
+    history.setAlertText(ALERT_UNIQUE_TEXT);
+    history.setAlertTimestamp(System.currentTimeMillis());
+
+    AlertTargetEntity target = new AlertTargetEntity();
+    target.setAlertStates(EnumSet.allOf(AlertState.class));
+    target.setTargetName("Alert Target");
+    target.setDescription("Mock Target");
+    target.setNotificationType("EMAIL");
+
+    String properties = "{ \"foo\" : \"bar\" }";
+    target.setProperties(properties);
+
+    AlertNoticeEntity notice = new AlertNoticeEntity();
+    notice.setUuid(ALERT_NOTICE_UUID);
+    notice.setAlertTarget(target);
+    notice.setAlertHistory(history);
+    notice.setNotifyState(NotificationState.PENDING);
+
+    ArrayList<AlertNoticeEntity> notices = new ArrayList<AlertNoticeEntity>();
+    notices.add(notice);
+
+    return notices;
+  }
+
+  /**
+   * A mock dispatcher that captures the {@link Notification}.
+   */
+  private static final class MockDispatcher implements NotificationDispatcher {
+
+    private Notification m_notificaiton;
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public String getType() {
+      return null;
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public void dispatch(Notification notification) {
+      m_notificaiton = notification;
+    }
+
+    public Notification getNotification() {
+      return m_notificaiton;
+    }
+  }
+
+  /**
+   * An {@link Executor} that calls {@link Runnable#run()} directly in the
+   * current thread.
+   */
+  private static final class MockExecutor implements Executor {
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public void execute(Runnable runnable) {
+      runnable.run();
+    }
+  }
+
+  /**
+   *
+   */
+  private class MockModule implements Module {
+    /**
+     *
+     */
+    @Override
+    public void configure(Binder binder) {
+      binder.bind(AlertDispatchDAO.class).toInstance(m_dao);
+      binder.bind(DispatchFactory.class).toInstance(m_dispatchFactory);
+      binder.bind(AmbariMetaInfo.class).toInstance(m_metaInfo);
+
+      EasyMock.expect(m_metaInfo.getServerVersion()).andReturn("2.0.0").anyTimes();
+    }
+  }
+}