AMBARI-7118. Alerts: populate incoming alert data to history and current (ncole)

ambari-agent/src/main/python/ambari_agent/AlertSchedulerHandler.py

@@ -57,8 +57,6 @@ class AlertSchedulerHandler():
       except:
         logger.critical("Could not create the cache directory {0}".format(cachedir))
         pass
-    
-
 
     self.__scheduler = Scheduler(AlertSchedulerHandler.APS_CONFIG)
     self.__in_minutes = in_minutes

ambari-agent/src/main/python/ambari_agent/alerts/base_alert.py

@@ -20,6 +20,7 @@ limitations under the License.
 
 import logging
 import re
+import time
 import traceback
 
 logger = logging.getLogger()
@@ -78,6 +79,7 @@ class BaseAlert(object):
     data['cluster'] = self.cluster
     data['service'] = self._find_value('serviceName')
     data['component'] = self._find_value('componentName')
+    data['timestamp'] = long(time.time() * 1000)
     
     self.collector.put(self.cluster, data)
   

ambari-server/src/main/java/org/apache/ambari/server/agent/HeartBeatHandler.java

@@ -60,6 +60,7 @@ import org.apache.ambari.server.state.StackInfo;
 import org.apache.ambari.server.state.State;
 import org.apache.ambari.server.state.alert.AlertDefinition;
 import org.apache.ambari.server.state.alert.AlertDefinitionHash;
+import org.apache.ambari.server.state.cluster.AlertDataManager;
 import org.apache.ambari.server.state.fsm.InvalidStateTransitionException;
 import org.apache.ambari.server.state.host.HostHealthyHeartbeatEvent;
 import org.apache.ambari.server.state.host.HostRegistrationRequestEvent;
@@ -114,6 +115,9 @@ public class HeartBeatHandler {
 
   @Inject
   private AlertDefinitionHash alertDefinitionHash;
+  
+  @Inject
+  private AlertDataManager alertManager;
 
   private Map<String, Long> hostResponseIds = new ConcurrentHashMap<String, Long>();
 
@@ -232,6 +236,15 @@ public class HeartBeatHandler {
     if (null == hostname || null == heartbeat) {
       return;
     }
+    
+    if (null != heartbeat.getAlerts()) {
+      for (Alert alert : heartbeat.getAlerts()) {
+        if (null == alert.getHost())
+          alert.setHost(hostname);
+        Cluster cluster = clusterFsm.getCluster(alert.getCluster());
+        alertManager.add(cluster.getClusterId(), alert);
+      }
+    }
 
     for (Cluster cluster : clusterFsm.getClustersForHost(hostname)) {
       cluster.addAlerts(heartbeat.getNodeStatus().getAlerts());

ambari-server/src/main/java/org/apache/ambari/server/orm/dao/AlertsDAO.java

@@ -239,6 +239,20 @@ public class AlertsDAO {
 
     return daoUtils.selectList(query);
   }
+  
+  @RequiresSession
+  public AlertCurrentEntity findCurrentByHostAndName(long clusterId, String hostName,
+      String alertName) {
+    
+    TypedQuery<AlertCurrentEntity> query = entityManagerProvider.get().createNamedQuery(
+        "AlertCurrentEntity.findByHostAndName", AlertCurrentEntity.class);
+
+    query.setParameter("clusterId", Long.valueOf(clusterId));
+    query.setParameter("hostName", hostName);
+    query.setParameter("definitionName", alertName);
+
+    return daoUtils.selectOne(query);
+  }
 
   /**
    * Removes alert history and current alerts for the specified alert defintiion

ambari-server/src/main/java/org/apache/ambari/server/orm/entities/AlertCurrentEntity.java

@@ -17,6 +17,7 @@
  */
 package org.apache.ambari.server.orm.entities;
 
+import javax.persistence.CascadeType;
 import javax.persistence.Column;
 import javax.persistence.Entity;
 import javax.persistence.EnumType;
@@ -48,6 +49,7 @@ import org.apache.ambari.server.state.MaintenanceState;
     @NamedQuery(name = "AlertCurrentEntity.findAll", query = "SELECT alert FROM AlertCurrentEntity alert"),
     @NamedQuery(name = "AlertCurrentEntity.findByService", query = "SELECT alert FROM AlertCurrentEntity alert JOIN alert.alertHistory history WHERE history.clusterId = :clusterId AND history.serviceName = :serviceName"),
     @NamedQuery(name = "AlertCurrentEntity.findByHost", query = "SELECT alert FROM AlertCurrentEntity alert JOIN alert.alertHistory history WHERE history.clusterId = :clusterId AND history.hostName = :hostName"),
+    @NamedQuery(name = "AlertCurrentEntity.findByHostAndName", query = "SELECT alert FROM AlertCurrentEntity alert JOIN alert.alertHistory history WHERE history.clusterId = :clusterId AND history.alertDefinition.definitionName = :definitionName AND history.hostName = :hostName"),
     @NamedQuery(name = "AlertCurrentEntity.removeByHistoryId", query = "DELETE FROM AlertCurrentEntity alert WHERE alert.alertHistory.alertId = :historyId"),
     @NamedQuery(name = "AlertCurrentEntity.removeByDefinitionId", query = "DELETE FROM AlertCurrentEntity alert WHERE alert.alertDefinition.definitionId = :definitionId") })
 public class AlertCurrentEntity {
@@ -70,7 +72,7 @@ public class AlertCurrentEntity {
   /**
    * Unidirectional one-to-one association to {@link AlertHistoryEntity}
    */
-  @OneToOne
+  @OneToOne(cascade = { CascadeType.PERSIST })
   @JoinColumn(name = "history_id", unique = true, nullable = false)
   private AlertHistoryEntity alertHistory;
 

ambari-server/src/main/java/org/apache/ambari/server/state/Alert.java

@@ -22,6 +22,7 @@ import org.codehaus.jackson.annotate.JsonProperty;
  * An alert represents a problem or notice for a cluster.
  */
 public class Alert {
+  private String cluster = null;
   private String name = null;
   private String instance = null;
   private String service = null;
@@ -30,6 +31,7 @@ public class Alert {
   private AlertState state = AlertState.UNKNOWN;
   private String label = null;
   private String text = null;
+  private long timestamp = 0L;
   
  
   /**
@@ -162,7 +164,23 @@ public class Alert {
   public void setState(AlertState state) {
     this.state = state;
   }
-
+  
+  @JsonProperty("timestamp")
+  public void setTimestamp(long ts) {
+    timestamp = ts;
+  }
+  
+  @JsonProperty("timestamp")
+  public long getTimestamp() {
+    return timestamp;
+  }
+  
+  /**
+   * @return
+   */
+  public String getCluster() {
+    return cluster;
+  }
   
   @Override
   public int hashCode() {
@@ -224,6 +242,7 @@ public class Alert {
     sb.append('}');
     return sb.toString();
   }
+
   
 
 }

ambari-server/src/main/java/org/apache/ambari/server/state/cluster/AlertDataManager.java

@@ -0,0 +1,110 @@
+/**
+ * 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.cluster;
+
+import org.apache.ambari.server.orm.dao.AlertDefinitionDAO;
+import org.apache.ambari.server.orm.dao.AlertsDAO;
+import org.apache.ambari.server.orm.entities.AlertCurrentEntity;
+import org.apache.ambari.server.orm.entities.AlertDefinitionEntity;
+import org.apache.ambari.server.orm.entities.AlertHistoryEntity;
+import org.apache.ambari.server.state.Alert;
+
+import com.google.inject.Inject;
+import com.google.inject.Singleton;
+
+/**
+ * The AlertManager is reponsible for tracking all alerts generated for a cluster. 
+ */
+@Singleton
+public class AlertDataManager {
+
+  @Inject
+  private AlertsDAO m_alertsDao;
+  @Inject
+  private AlertDefinitionDAO m_definitionDao;
+
+  
+  AlertDataManager() {
+  }
+  
+  /**
+   * Adds an alert.  Checks for a new state before creating a new history record.
+   * 
+   * @param clusterId the id for the cluster
+   * @param alert the alert to add
+   */
+  public void add(long clusterId, Alert alert) {
+    
+    AlertCurrentEntity current = m_alertsDao.findCurrentByHostAndName(clusterId,
+        alert.getHost(), alert.getName());
+    
+    if (null == current) {
+      AlertDefinitionEntity definition = m_definitionDao.findByName(clusterId,
+          alert.getName());
+      
+      AlertHistoryEntity history = createHistory(clusterId, definition, alert);
+      
+      current = new AlertCurrentEntity();
+      current.setAlertHistory(history);
+      current.setLatestTimestamp(Long.valueOf(alert.getTimestamp()));
+      current.setOriginalTimestamp(Long.valueOf(alert.getTimestamp()));
+      
+      m_alertsDao.create(current);
+      
+    } else if (alert.getState() == current.getAlertHistory().getAlertState()) {
+      current.setLatestTimestamp(Long.valueOf(alert.getTimestamp()));
+      
+      m_alertsDao.merge(current);
+    } else {
+      // insert history, update current
+      AlertHistoryEntity history = createHistory(clusterId,
+          current.getAlertHistory().getAlertDefinition(), alert);
+      
+      current.setAlertHistory(history);
+      current.setLatestTimestamp(Long.valueOf(alert.getTimestamp()));
+      current.setOriginalTimestamp(Long.valueOf(alert.getTimestamp()));
+      
+      m_alertsDao.merge(current);
+    }
+    
+  }
+  /**
+   * Convenience to create a new alert.
+   * @param clusterId the cluster id
+   * @param definition the definition
+   * @param alert the alert data
+   * @return the new history record
+   */
+  private AlertHistoryEntity createHistory(long clusterId, AlertDefinitionEntity definition, Alert alert) {
+    AlertHistoryEntity history = new AlertHistoryEntity();
+    history.setAlertDefinition(definition);
+    history.setAlertInstance(alert.getInstance());
+    history.setAlertLabel(alert.getLabel());
+    history.setAlertState(alert.getState());
+    history.setAlertText(alert.getText());
+    history.setAlertTimestamp(Long.valueOf(alert.getTimestamp()));
+    history.setClusterId(Long.valueOf(clusterId));
+    history.setComponentName(alert.getComponent());
+    history.setHostName(alert.getHost());
+    history.setServiceName(alert.getService());
+    
+    return history;
+  }
+  
+
+}

ambari-server/src/test/java/org/apache/ambari/server/orm/dao/AlertDefinitionDAOTest.java

@@ -287,7 +287,7 @@ public class AlertDefinitionDAOTest {
     history.setAlertState(AlertState.OK);
     history.setAlertText("Alert Text");
     history.setAlertTimestamp(calendar.getTimeInMillis());
-    alertsDao.create(history);
+//    alertsDao.create(history);
 
     AlertCurrentEntity current = new AlertCurrentEntity();
     current.setAlertHistory(history);

ambari-server/src/test/java/org/apache/ambari/server/orm/dao/AlertsDAOTest.java

@@ -21,6 +21,7 @@ package org.apache.ambari.server.orm.dao;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
 
 import java.util.ArrayList;
 import java.util.Calendar;
@@ -104,6 +105,7 @@ public class AlertsDAOTest {
         history.setAlertLabel(definition.getDefinitionName() + " " + i);
         history.setAlertText(definition.getDefinitionName() + " " + i);
         history.setAlertTimestamp(calendar.getTimeInMillis());
+        history.setHostName("h1");
 
         history.setAlertState(AlertState.OK);
         if (i == 0 || i == 5) {
@@ -261,4 +263,16 @@ public class AlertsDAOTest {
     assertNotNull(history);
     assertEquals(0, history.size());
   }
+  
+  @Test
+  public void testFindCurrentByHostAndName() throws Exception {
+    AlertCurrentEntity entity = dao.findCurrentByHostAndName(clusterId.longValue(), "h2", "Alert Definition 1");
+    assertNull(entity);
+    
+    entity = dao.findCurrentByHostAndName(clusterId.longValue(), "h1", "Alert Definition 1");
+    
+    assertNotNull(entity);
+    assertNotNull(entity.getAlertHistory());
+    assertNotNull(entity.getAlertHistory().getAlertDefinition());
+  }
 }

ambari-server/src/test/java/org/apache/ambari/server/state/cluster/AlertDataManagerTest.java

@@ -0,0 +1,181 @@
+/**
+ * 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.cluster;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+
+import java.util.List;
+import java.util.UUID;
+
+import org.apache.ambari.server.orm.GuiceJpaInitializer;
+import org.apache.ambari.server.orm.InMemoryDefaultTestModule;
+import org.apache.ambari.server.orm.OrmTestHelper;
+import org.apache.ambari.server.orm.dao.AlertDefinitionDAO;
+import org.apache.ambari.server.orm.dao.AlertsDAO;
+import org.apache.ambari.server.orm.entities.AlertCurrentEntity;
+import org.apache.ambari.server.orm.entities.AlertDefinitionEntity;
+import org.apache.ambari.server.orm.entities.AlertHistoryEntity;
+import org.apache.ambari.server.state.Alert;
+import org.apache.ambari.server.state.AlertState;
+import org.apache.ambari.server.state.alert.Scope;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import com.google.inject.Guice;
+import com.google.inject.Injector;
+import com.google.inject.persist.PersistService;
+
+/**
+ * Tests the alert manager.
+ */
+public class AlertDataManagerTest {
+  
+  private static final String ALERT_DEFINITION = "Alert Definition 1";
+  private static final String SERVICE = "service1";
+  private static final String COMPONENT = "component1";
+  private static final String HOST1 = "h1";
+  private static final String HOST2 = "h2";
+  private static final String ALERT_LABEL = "My Label";
+  
+  private Long clusterId;
+  private Injector injector;
+  private OrmTestHelper helper;
+  private AlertsDAO dao;
+  
+  @Before
+  public void setup() throws Exception {
+    injector = Guice.createInjector(new InMemoryDefaultTestModule());
+    injector.getInstance(GuiceJpaInitializer.class);
+    helper = injector.getInstance(OrmTestHelper.class);
+    clusterId = helper.createCluster();
+    dao = injector.getInstance(AlertsDAO.class);
+    AlertDefinitionDAO definitionDao = injector.getInstance(AlertDefinitionDAO.class);
+
+    // 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(clusterId);
+      definition.setHash(UUID.randomUUID().toString());
+      definition.setScheduleInterval(Integer.valueOf(60));
+      definition.setScope(Scope.SERVICE);
+      definition.setSource("Source " + i);
+      definition.setSourceType("SCRIPT");
+      definitionDao.create(definition);
+    }
+  
+  }
+  
+  @After
+  public void teardown() {
+    injector.getInstance(PersistService.class).stop();
+    injector = null;
+  }
+  
+  @Test
+  public void testAlertRecords() {
+    
+    Alert alert1 = new Alert(ALERT_DEFINITION, null, SERVICE, COMPONENT, HOST1, AlertState.OK);
+    alert1.setLabel(ALERT_LABEL);
+    alert1.setText("Component component1 is OK");
+    alert1.setTimestamp(1L);
+
+    Alert alert2 = new Alert(ALERT_DEFINITION, null, SERVICE, COMPONENT, HOST2, AlertState.CRITICAL);
+    alert2.setLabel(ALERT_LABEL);
+    alert2.setText("Component component2 is not OK");
+    
+    AlertDataManager am = injector.getInstance(AlertDataManager.class);
+    
+    am.add(clusterId.longValue(), alert1);
+    am.add(clusterId.longValue(), alert2);
+    
+    List<AlertCurrentEntity> allCurrent = dao.findCurrentByService(clusterId.longValue(), SERVICE);
+    assertEquals(2, allCurrent.size());
+    
+    List<AlertHistoryEntity> allHistory = dao.findAll(clusterId.longValue());
+    assertEquals(2, allHistory.size());
+    
+    AlertCurrentEntity current = dao.findCurrentByHostAndName(clusterId.longValue(), HOST1, ALERT_DEFINITION);
+    assertNotNull(current);
+    assertEquals(HOST1, current.getAlertHistory().getHostName());
+    assertEquals(ALERT_DEFINITION, current.getAlertHistory().getAlertDefinition().getDefinitionName());
+    assertEquals(ALERT_LABEL, current.getAlertHistory().getAlertLabel());
+    assertEquals("Component component1 is OK", current.getAlertHistory().getAlertText());
+    assertEquals(current.getAlertHistory().getAlertState(), AlertState.OK);
+    assertEquals(1L, current.getOriginalTimestamp().longValue());
+    assertEquals(1L, current.getLatestTimestamp().longValue());
+    
+    Long currentId = current.getAlertId();
+    Long historyId = current.getAlertHistory().getAlertId();
+    
+    // no new history since the state is the same
+    Alert alert3 = new Alert(ALERT_DEFINITION, null, SERVICE, COMPONENT, HOST1, AlertState.OK);
+    alert3.setLabel(ALERT_LABEL);
+    alert3.setText("Component component1 is OK");
+    alert3.setTimestamp(2L);
+    am.add(clusterId.longValue(), alert3);
+    
+    current = dao.findCurrentByHostAndName(clusterId.longValue(), HOST1, ALERT_DEFINITION);
+    assertNotNull(current);
+    assertEquals(currentId, current.getAlertId());
+    assertEquals(historyId, current.getAlertHistory().getAlertId());
+    assertEquals(HOST1, current.getAlertHistory().getHostName());
+    assertEquals(ALERT_DEFINITION, current.getAlertHistory().getAlertDefinition().getDefinitionName());
+    assertEquals(ALERT_LABEL, current.getAlertHistory().getAlertLabel());
+    assertEquals("Component component1 is OK", current.getAlertHistory().getAlertText());
+    assertEquals(current.getAlertHistory().getAlertState(), AlertState.OK);
+    assertEquals(1L, current.getOriginalTimestamp().longValue());
+    assertEquals(2L, current.getLatestTimestamp().longValue());
+   
+    allCurrent = dao.findCurrentByService(clusterId.longValue(), SERVICE);
+    assertEquals(2, allCurrent.size());
+    
+    allHistory = dao.findAll(clusterId.longValue());
+    assertEquals(2, allHistory.size());
+    
+    // change to warning
+    Alert alert4 = new Alert(ALERT_DEFINITION, null, SERVICE, COMPONENT, HOST1, AlertState.WARNING);
+    alert4.setLabel(ALERT_LABEL);
+    alert4.setText("Component component1 is about to go down");
+    alert4.setTimestamp(3L);
+    am.add(clusterId.longValue(), alert4);
+
+    current = dao.findCurrentByHostAndName(clusterId.longValue(), HOST1, ALERT_DEFINITION);
+    assertNotNull(current);
+    assertEquals(current.getAlertId(), currentId);
+    assertFalse(historyId.equals(current.getAlertHistory().getAlertId()));    
+    assertEquals(HOST1, current.getAlertHistory().getHostName());
+    assertEquals(ALERT_DEFINITION, current.getAlertHistory().getAlertDefinition().getDefinitionName());
+    assertEquals(ALERT_LABEL, current.getAlertHistory().getAlertLabel());
+    assertEquals("Component component1 is about to go down", current.getAlertHistory().getAlertText());
+    assertEquals(current.getAlertHistory().getAlertState(), AlertState.WARNING);
+    assertEquals(3L, current.getOriginalTimestamp().longValue());
+    assertEquals(3L, current.getLatestTimestamp().longValue());
+
+    allCurrent = dao.findCurrentByService(clusterId.longValue(), SERVICE);
+    assertEquals(2, allCurrent.size());
+    
+    allHistory = dao.findAll(clusterId.longValue());
+    assertEquals(3, allHistory.size());
+  }
+}