AMBARI-14565 - Renaming cluster in Ambari breaks alerts (jonathanhurley)

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

@@ -55,6 +55,9 @@ class AlertSchedulerHandler():
     self.host_scripts_dir = host_scripts_dir
 
     self._cluster_configuration = cluster_configuration
+
+    # a mapping between a cluster name and a unique hash for all definitions
+    self._cluster_hashes = {}
     
     if not os.path.exists(cachedir):
       try:
@@ -111,8 +114,17 @@ class AlertSchedulerHandler():
     with open(os.path.join(self.cachedir, self.FILENAME), 'w') as f:
       json.dump(alert_definitions, f, indent=2)
 
-    # reschedule only the jobs that have changed
-    self.reschedule()
+    # determine how to reschedule the jobs
+    reschedule_all = False
+    if "clusterName" in command_copy and command_copy["clusterName"] not in self._cluster_hashes:
+      reschedule_all = True
+
+    if reschedule_all is True:
+      # reschedule all jobs, creating new instances
+      self.reschedule_all()
+    else:
+      # reschedule only the jobs that have changed
+      self.reschedule()
 
 
   def __make_function(self, alert_def):
@@ -170,7 +182,7 @@ class AlertSchedulerHandler():
           break
 
       # jobs without valid UUIDs should be unscheduled
-      if uuid_valid == False:
+      if uuid_valid is False:
         jobs_removed += 1
         logger.info("[AlertScheduler] Unscheduling {0}".format(scheduled_job.name))
         self._collector.remove_by_uuid(scheduled_job.name)
@@ -186,7 +198,7 @@ class AlertSchedulerHandler():
           break
 
       # if no jobs are found with the definitions UUID, schedule it
-      if definition_scheduled == False:
+      if definition_scheduled is False:
         jobs_scheduled += 1
         self.schedule_definition(definition)
 
@@ -199,6 +211,8 @@ class AlertSchedulerHandler():
     Removes jobs that are scheduled where their UUID no longer is valid.
     Schedules jobs where the definition UUID is not currently scheduled.
     """
+    logger.info("[AlertScheduler] Rescheduling all jobs...")
+
     jobs_scheduled = 0
     jobs_removed = 0
 
@@ -207,18 +221,18 @@ class AlertSchedulerHandler():
 
     # unschedule all scheduled jobs
     for scheduled_job in scheduled_jobs:
-        jobs_removed += 1
-        logger.info("[AlertScheduler] Unscheduling {0}".format(scheduled_job.name))
-        self._collector.remove_by_uuid(scheduled_job.name)
-        self.__scheduler.unschedule_job(scheduled_job)
+      jobs_removed += 1
+      logger.info("[AlertScheduler] Unscheduling {0}".format(scheduled_job.name))
+      self._collector.remove_by_uuid(scheduled_job.name)
+      self.__scheduler.unschedule_job(scheduled_job)
 
     # for every definition, schedule a job
     for definition in definitions:
-        jobs_scheduled += 1
-        self.schedule_definition(definition)
+      jobs_scheduled += 1
+      self.schedule_definition(definition)
 
-    logger.info("[AlertScheduler] Reschedule Summary: {0} rescheduled, {1} unscheduled".format(
-      str(jobs_scheduled), str(jobs_removed)))
+    logger.info("[AlertScheduler] Reschedule Summary: {0} unscheduled, {0} rescheduled".format(
+      str(jobs_removed), str(jobs_scheduled)))
 
 
   def collector(self):
@@ -229,12 +243,11 @@ class AlertSchedulerHandler():
   def __load_definitions(self):
     """
     Loads all alert definitions from a file. All clusters are stored in
-    a single file.
+    a single file. This wil also populate the cluster-to-hash dictionary.
     :return:
     """
     definitions = []
 
-    all_commands = None
     alerts_definitions_path = os.path.join(self.cachedir, self.FILENAME)
     try:
       with open(alerts_definitions_path) as fp:
@@ -246,6 +259,12 @@ class AlertSchedulerHandler():
     for command_json in all_commands:
       clusterName = '' if not 'clusterName' in command_json else command_json['clusterName']
       hostName = '' if not 'hostName' in command_json else command_json['hostName']
+      clusterHash = None if not 'hash' in command_json else command_json['hash']
+
+      # cache the cluster and cluster hash after loading the JSON
+      if clusterName != '' and clusterHash is not None:
+        logger.info('[AlertScheduler] Caching cluster {0} with alert hash {1}'.format(clusterName, clusterHash))
+        self._cluster_hashes[clusterName] = clusterHash
 
       for definition in command_json['alertDefinitions']:
         alert = self.__json_to_callable(clusterName, hostName, definition)

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

@@ -76,7 +76,7 @@ class ClusterConfiguration():
     Updates the in-memory and disk-based cluster configurations based on
     the heartbeat. This will only update configurations on the following
     types of commands in the heartbeat: execution, and alert definition.
-    :param new_configurations:
+    :param heartbeat: the heartbeat response
     :return:
     """
     heartbeat_keys = heartbeat.keys()
@@ -88,7 +88,7 @@ class ClusterConfiguration():
 
     # if this heartbeat doesn't contain a command with configurations, then
     # don't process it
-    if not heartbeat_contains_configurations:
+    if heartbeat_contains_configurations is False:
       return
 
     if self.EXECUTION_COMMANDS in heartbeat_keys:
@@ -99,8 +99,6 @@ class ClusterConfiguration():
           configurations = command['configurations']
           self._update_configurations(cluster_name, configurations)
 
-      return
-
     if self.ALERT_DEFINITION_COMMANDS in heartbeat_keys:
       alert_definition_commands = heartbeat[self.ALERT_DEFINITION_COMMANDS]
       for command in alert_definition_commands:
@@ -109,8 +107,6 @@ class ClusterConfiguration():
           configurations = command['configurations']
           self._update_configurations(cluster_name, configurations)
 
-      return
-
 
   def _update_configurations(self, cluster_name, configuration):
     """

ambari-server/src/main/java/org/apache/ambari/server/events/AggregateAlertRecalculateEvent.java

@@ -39,7 +39,7 @@ public class AggregateAlertRecalculateEvent extends AlertEvent {
    */
   @Override
   public String toString() {
-    StringBuilder buffer = new StringBuilder("AggregateAlertRecalculateEvent{ ");
+    StringBuilder buffer = new StringBuilder("AggregateAlertRecalculateEvent{");
     buffer.append("cluserId=").append(m_clusterId);
     buffer.append("}");
     return buffer.toString();

ambari-server/src/main/java/org/apache/ambari/server/events/AlertReceivedEvent.java

@@ -17,10 +17,10 @@
  */
 package org.apache.ambari.server.events;
 
-import org.apache.ambari.server.state.Alert;
-
 import java.util.List;
 
+import org.apache.ambari.server.state.Alert;
+
 /**
  * The {@link AlertReceivedEvent} is fired when an {@link Alert} is received or
  * generated.
@@ -46,7 +46,7 @@ public final class AlertReceivedEvent extends AlertEvent {
    */
   @Override
   public String toString() {
-    StringBuilder buffer = new StringBuilder("AlertReceivedEvent{ ");
+    StringBuilder buffer = new StringBuilder("AlertReceivedEvent{");
     buffer.append("cluserId=").append(m_clusterId);
     buffer.append(", alerts=").append(getAlerts());
     buffer.append("}");

ambari-server/src/main/java/org/apache/ambari/server/events/AlertStateChangeEvent.java

@@ -90,7 +90,7 @@ public class AlertStateChangeEvent extends AlertEvent {
    */
   @Override
   public String toString() {
-    StringBuilder buffer = new StringBuilder("AlertStateChangeEvent{ ");
+    StringBuilder buffer = new StringBuilder("AlertStateChangeEvent{");
     buffer.append("cluserId=").append(m_clusterId);
     buffer.append(", fromState=").append(m_fromState);
     buffer.append(", alert=").append(m_alert);

ambari-server/src/main/java/org/apache/ambari/server/events/AmbariEvent.java

@@ -100,7 +100,12 @@ public abstract class AmbariEvent {
     /**
      * Received a final command report for some action
      */
-    ACTION_EXECUTION_FINISHED
+    ACTION_EXECUTION_FINISHED,
+
+    /**
+     * The cluster was renamed.
+     */
+    CLUSTER_RENAME;
   }
 
   /**
@@ -126,4 +131,15 @@ public abstract class AmbariEvent {
   public AmbariEventType getType() {
     return m_eventType;
   }
+
+  /**
+   * {@inheritDoc}
+   */
+  @Override
+  public String toString() {
+    StringBuilder buffer = new StringBuilder(getClass().getSimpleName());
+    buffer.append("{eventType=").append(m_eventType);
+    buffer.append("}");
+    return buffer.toString();
+  }
 }

ambari-server/src/main/java/org/apache/ambari/server/events/HostComponentVersionEvent.java

@@ -54,11 +54,11 @@ public class HostComponentVersionEvent extends ClusterEvent {
    */
   @Override
   public String toString() {
-    StringBuilder buffer = new StringBuilder("HostComponentVersionEvent{ ");
+    StringBuilder buffer = new StringBuilder("HostComponentVersionEvent{");
     buffer.append("cluserId=").append(m_clusterId);
-    buffer.append(", serviceName=").append(this.sch.getServiceName());
-    buffer.append(", componentName=").append(this.sch.getServiceComponentName());
-    buffer.append(", hostName=").append(this.sch.getHostName());
+    buffer.append(", serviceName=").append(sch.getServiceName());
+    buffer.append(", componentName=").append(sch.getServiceComponentName());
+    buffer.append(", hostName=").append(sch.getHostName());
     buffer.append("}");
     return buffer.toString();
   }

ambari-server/src/main/java/org/apache/ambari/server/events/HostRemovedEvent.java

@@ -61,7 +61,7 @@ public class HostRemovedEvent extends HostEvent {
    */
   @Override
   public String toString() {
-    StringBuilder buffer = new StringBuilder("HostRemovedEvent{ ");
+    StringBuilder buffer = new StringBuilder("HostRemovedEvent{");
     buffer.append("hostName=").append(m_hostName);
     buffer.append("}");
     return buffer.toString();

ambari-server/src/main/java/org/apache/ambari/server/events/InitialAlertEvent.java

@@ -80,7 +80,7 @@ public class InitialAlertEvent extends AlertEvent {
    */
   @Override
   public String toString() {
-    StringBuilder buffer = new StringBuilder("InitialAlertEvent{ ");
+    StringBuilder buffer = new StringBuilder("InitialAlertEvent{");
     buffer.append("cluserId=").append(m_clusterId);
     buffer.append(", alert=").append(m_alert);
 

ambari-server/src/main/java/org/apache/ambari/server/events/MaintenanceModeEvent.java

@@ -165,10 +165,10 @@ public class MaintenanceModeEvent extends AmbariEvent {
       object = m_serviceComponentHost;
     }
 
-    StringBuilder buffer = new StringBuilder("MaintenanceModeEvent{ ");
+    StringBuilder buffer = new StringBuilder("MaintenanceModeEvent{");
     buffer.append("state=").append(m_state);
     buffer.append(", object=").append(object);
-    buffer.append(" }");
+    buffer.append("}");
 
     return buffer.toString();
   }

ambari-server/src/main/java/org/apache/ambari/server/events/ServiceComponentInstalledEvent.java

@@ -59,7 +59,7 @@ public class ServiceComponentInstalledEvent extends ServiceEvent {
    */
   @Override
   public String toString() {
-    StringBuilder buffer = new StringBuilder("ServiceComponentInstalledEvent{ ");
+    StringBuilder buffer = new StringBuilder("ServiceComponentInstalledEvent{");
     buffer.append("cluserId=").append(m_clusterId);
     buffer.append(", stackName=").append(m_stackName);
     buffer.append(", stackVersion=").append(m_stackVersion);

ambari-server/src/main/java/org/apache/ambari/server/events/ServiceComponentUninstalledEvent.java

@@ -65,8 +65,7 @@ public class ServiceComponentUninstalledEvent extends ServiceEvent {
    */
   @Override
   public String toString() {
-    StringBuilder buffer = new StringBuilder(
-        "ServiceComponentUninstalledEvent{ ");
+    StringBuilder buffer = new StringBuilder("ServiceComponentUninstalledEvent{");
     buffer.append("cluserId=").append(m_clusterId);
     buffer.append(", stackName=").append(m_stackName);
     buffer.append(", stackVersion=").append(m_stackVersion);

ambari-server/src/main/java/org/apache/ambari/server/events/ServiceInstalledEvent.java

@@ -41,7 +41,7 @@ public class ServiceInstalledEvent extends ServiceEvent {
    */
   @Override
   public String toString() {
-    StringBuilder buffer = new StringBuilder("ServiceInstalledEvent{ ");
+    StringBuilder buffer = new StringBuilder("ServiceInstalledEvent{");
     buffer.append("cluserId=").append(m_clusterId);
     buffer.append(", stackName=").append(m_stackName);
     buffer.append(", stackVersion=").append(m_stackVersion);

ambari-server/src/main/java/org/apache/ambari/server/events/ServiceRemovedEvent.java

@@ -41,7 +41,7 @@ public class ServiceRemovedEvent extends ServiceEvent {
    */
   @Override
   public String toString() {
-    StringBuilder buffer = new StringBuilder("ServiceRemovedEvent{ ");
+    StringBuilder buffer = new StringBuilder("ServiceRemovedEvent{");
     buffer.append("cluserId=").append(m_clusterId);
     buffer.append(", stackName=").append(m_stackName);
     buffer.append(", stackVersion=").append(m_stackVersion);

ambari-server/src/main/java/org/apache/ambari/server/events/listeners/alerts/AlertHashInvalidationListener.java

@@ -24,6 +24,8 @@ import org.apache.ambari.server.EagerSingleton;
 import org.apache.ambari.server.agent.AlertDefinitionCommand;
 import org.apache.ambari.server.agent.HeartBeatResponse;
 import org.apache.ambari.server.events.AlertHashInvalidationEvent;
+import org.apache.ambari.server.events.AmbariEvent.AmbariEventType;
+import org.apache.ambari.server.events.ClusterEvent;
 import org.apache.ambari.server.events.ServiceComponentUninstalledEvent;
 import org.apache.ambari.server.events.publishers.AmbariEventPublisher;
 import org.apache.ambari.server.state.alert.AlertDefinitionHash;
@@ -38,9 +40,15 @@ import com.google.inject.Singleton;
 
 /**
  * The {@link AlertHashInvalidationListener} is used to respond to
- * {@link AlertHashInvalidationEvent} instances and ensure that the
- * {@link AlertDefinitionCommand}s are enqueued for the
+ * {@link AlertHashInvalidationEvent} and {@link ClusterEvent} instances and
+ * ensure that the {@link AlertDefinitionCommand}s are enqueued for the
  * {@link HeartBeatResponse}.
+ * <p/>
+ * <ul>
+ * <li>{@link ClusterEvent} - invalidates all alerts across the cluster</li>
+ * <li>{@link AlertHashInvalidationEvent} - invalidates a specific alert across
+ * affected hosts</li>
+ * </ul>
  */
 @Singleton
 @EagerSingleton
@@ -119,4 +127,32 @@ public class AlertHashInvalidationListener {
     m_alertDefinitionHash.get().enqueueAgentCommands(clusterId,
         Collections.singletonList(hostName));
   }
+
+  /**
+   * Handles {@link ClusterEvent} by performing the following tasks:
+   * <ul>
+   * <li>Invalidates all alerts across all hosts. This is because agents use the
+   * cluster name as an identifier and if the cluster is renamed, then they need
+   * to be rescheduled.</li>
+   * </ul>
+   *
+   * @param event
+   *          the event being handled.
+   */
+  @Subscribe
+  @AllowConcurrentEvents
+  public void onAmbariEvent(ClusterEvent event) {
+    LOG.debug("Received event {}", event);
+
+    if( event.getType() != AmbariEventType.CLUSTER_RENAME ) {
+      return;
+    }
+
+
+    AlertDefinitionHash hash = m_alertDefinitionHash.get();
+    hash.invalidateAll();
+
+    long clusterId = event.getClusterId();
+    hash.enqueueAgentCommands(clusterId);
+  }
 }

ambari-server/src/main/java/org/apache/ambari/server/state/alert/AlertDefinitionHash.java

@@ -433,6 +433,39 @@ public class AlertDefinitionHash {
     return affectedHosts;
   }
 
+  /**
+   * Enqueue {@link AlertDefinitionCommand}s for every host in the cluster so
+   * that they will receive a payload of alert definitions that they should be
+   * running.
+   * <p/>
+   * This method is typically called after {@link #invalidateAll()} has caused a
+   * cache invalidation of all alert definitions.
+   *
+   * @param clusterId
+   *          the ID of the cluster.
+   * @param hosts
+   *          the hosts to push {@link AlertDefinitionCommand}s for.
+   */
+  public void enqueueAgentCommands(long clusterId) {
+    String clusterName = null;
+    Collection<String> hostNames;
+
+    try {
+      Cluster cluster = m_clusters.get().getClusterById(clusterId);
+      clusterName = cluster.getClusterName();
+      Collection<Host> hosts = cluster.getHosts();
+
+      hostNames = new ArrayList<>(hosts.size());
+      for (Host host : hosts) {
+        hostNames.add(host.getHostName());
+      }
+
+      enqueueAgentCommands(clusterName, hostNames);
+    } catch (AmbariException ae) {
+      LOG.error("Unable to lookup cluster for alert definition commands", ae);
+    }
+  }
+
   /**
    * Enqueue {@link AlertDefinitionCommand}s for every host specified so that
    * they will receive a payload of alert definitions that they should be
@@ -442,8 +475,8 @@ public class AlertDefinitionHash {
    * {@link #invalidateHosts(AlertDefinitionEntity)} has caused a cache
    * invalidation of the alert definition hash.
    *
-   * @param clusterName
-   *          the name of the cluster (not {@code null}).
+   * @param clusterId
+   *          the ID of the cluster.
    * @param hosts
    *          the hosts to push {@link AlertDefinitionCommand}s for.
    */

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

@@ -55,6 +55,9 @@ import org.apache.ambari.server.controller.ConfigurationResponse;
 import org.apache.ambari.server.controller.MaintenanceStateHelper;
 import org.apache.ambari.server.controller.RootServiceResponseFactory.Services;
 import org.apache.ambari.server.controller.ServiceConfigVersionResponse;
+import org.apache.ambari.server.events.AmbariEvent.AmbariEventType;
+import org.apache.ambari.server.events.ClusterEvent;
+import org.apache.ambari.server.events.publishers.AmbariEventPublisher;
 import org.apache.ambari.server.orm.RequiresSession;
 import org.apache.ambari.server.orm.cache.HostConfigMapping;
 import org.apache.ambari.server.orm.dao.AlertDefinitionDAO;
@@ -273,6 +276,12 @@ public class ClusterImpl implements Cluster {
 
   private volatile Multimap<String, String> serviceConfigTypes;
 
+  /**
+   * Used to publish events relating to cluster CRUD operations.
+   */
+  @Inject
+  private AmbariEventPublisher eventPublisher;
+
 
   @Inject
   public ClusterImpl(@Assisted ClusterEntity clusterEntity,
@@ -631,9 +640,10 @@ public class ClusterImpl implements Cluster {
 
   @Override
   public void setClusterName(String clusterName) {
+    String oldName = clusterEntity.getClusterName();
     clusterGlobalLock.writeLock().lock();
+
     try {
-      String oldName = clusterEntity.getClusterName();
       clusterEntity.setClusterName(clusterName);
 
       // RollbackException possibility if UNIQUE constraint violated
@@ -642,6 +652,12 @@ public class ClusterImpl implements Cluster {
     } finally {
       clusterGlobalLock.writeLock().unlock();
     }
+
+    // if the name changed, fire an event
+    if (!StringUtils.equals(oldName, clusterName)) {
+      ClusterEvent clusterNameChangedEvent = new ClusterEvent(AmbariEventType.CLUSTER_RENAME, clusterId);
+      eventPublisher.publish(clusterNameChangedEvent);
+    }
   }
 
   @Override

ambari-server/src/test/java/org/apache/ambari/server/events/EventsTest.java

@@ -20,9 +20,9 @@ package org.apache.ambari.server.events;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.UUID;
 
-import junit.framework.Assert;
-
+import org.apache.ambari.server.events.AmbariEvent.AmbariEventType;
 import org.apache.ambari.server.orm.GuiceJpaInitializer;
 import org.apache.ambari.server.orm.InMemoryDefaultTestModule;
 import org.apache.ambari.server.orm.OrmTestHelper;
@@ -52,6 +52,8 @@ import com.google.inject.Guice;
 import com.google.inject.Injector;
 import com.google.inject.persist.PersistService;
 
+import junit.framework.Assert;
+
 /**
  * Tests that {@link EventsTest} instances are fired correctly and
  * that alert data is bootstrapped into the database.
@@ -242,6 +244,25 @@ public class EventsTest {
     Assert.assertEquals(2, m_listener.getAmbariEventReceivedCount(eventClass));
   }
 
+  /**
+   * Tests that {@link ServiceComponentUninstalledEvent}s are fired correctly.
+   *
+   * @throws Exception
+   */
+  @Test
+  public void testClusterRenameEvent() throws Exception {
+    Class<? extends AmbariEvent> eventClass = ClusterEvent.class;
+    installHdfsService();
+
+    Assert.assertFalse(m_listener.isAmbariEventReceived(eventClass));
+    m_cluster.setClusterName(UUID.randomUUID().toString());
+
+    Assert.assertTrue(m_listener.isAmbariEventReceived(eventClass));
+    List<AmbariEvent> ambariEvents = m_listener.getAmbariEventInstances(eventClass);
+    Assert.assertEquals(1, ambariEvents.size());
+    Assert.assertEquals(AmbariEventType.CLUSTER_RENAME, ambariEvents.get(0).getType());
+  }
+
   /**
    * Calls {@link Service#persist()} to mock a service install along with
    * creating a single {@link Host} and {@link ServiceComponentHost}.