ambari-server/src/main/java/org/apache/ambari/server/events/listeners/upgrade/HostVersionOutOfSyncListener.java

/*
 * 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.events.listeners.upgrade;

import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.locks.Lock;

import org.apache.ambari.annotations.Experimental;
import org.apache.ambari.annotations.ExperimentalFeature;
import org.apache.ambari.server.AmbariException;
import org.apache.ambari.server.EagerSingleton;
import org.apache.ambari.server.api.services.AmbariMetaInfo;
import org.apache.ambari.server.events.HostsAddedEvent;
import org.apache.ambari.server.events.HostsRemovedEvent;
import org.apache.ambari.server.events.ServiceComponentInstalledEvent;
import org.apache.ambari.server.events.ServiceComponentUninstalledEvent;
import org.apache.ambari.server.events.ServiceInstalledEvent;
import org.apache.ambari.server.events.publishers.AmbariEventPublisher;
import org.apache.ambari.server.logging.LockFactory;
import org.apache.ambari.server.orm.dao.HostDAO;
import org.apache.ambari.server.orm.dao.HostVersionDAO;
import org.apache.ambari.server.orm.dao.RepositoryVersionDAO;
import org.apache.ambari.server.orm.entities.HostComponentDesiredStateEntity;
import org.apache.ambari.server.orm.entities.HostEntity;
import org.apache.ambari.server.orm.entities.HostVersionEntity;
import org.apache.ambari.server.orm.entities.RepositoryVersionEntity;
import org.apache.ambari.server.orm.entities.StackEntity;
import org.apache.ambari.server.state.Cluster;
import org.apache.ambari.server.state.Clusters;
import org.apache.ambari.server.state.ComponentInfo;
import org.apache.ambari.server.state.RepositoryVersionState;
import org.apache.ambari.server.state.Service;
import org.apache.ambari.server.state.ServiceComponent;
import org.apache.ambari.server.state.StackId;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.eventbus.Subscribe;
import com.google.inject.Inject;
import com.google.inject.Provider;
import com.google.inject.Singleton;
import com.google.inject.persist.Transactional;

/**
 * The {@link org.apache.ambari.server.events.listeners.upgrade.HostVersionOutOfSyncListener} class
 * handles {@link org.apache.ambari.server.events.ServiceInstalledEvent} and
 * {@link org.apache.ambari.server.events.ServiceComponentInstalledEvent}
 * to update {@link org.apache.ambari.server.state.RepositoryVersionState}
 */
@Singleton
@EagerSingleton
public class HostVersionOutOfSyncListener {
  /**
   * Logger.
   */
  private static final Logger LOG = LoggerFactory.getLogger(HostVersionOutOfSyncListener.class);

  @Inject
  private Provider<HostVersionDAO> hostVersionDAO;

  @Inject
  private Provider<HostDAO> hostDAO;

  @Inject
  private Provider<Clusters> clusters;

  @Inject
  private Provider<AmbariMetaInfo> ami;

  @Inject
  private Provider<RepositoryVersionDAO> repositoryVersionDAO;

  /**
   * The publisher may be an asynchronous, multi-threaded one, so to avoid the (rare, but possible) case
   * of both an Install and Uninstall event occurring at the same time, we use a Lock.
   */
  private final Lock m_lock;

  @Inject
  public HostVersionOutOfSyncListener(AmbariEventPublisher ambariEventPublisher, LockFactory lockFactory) {
    ambariEventPublisher.register(this);

    m_lock = lockFactory.newLock("hostVersionOutOfSyncListenerLock");
  }

  @Subscribe
  @Transactional
  public void onServiceComponentEvent(ServiceComponentInstalledEvent event) {
    if (LOG.isDebugEnabled()) {
      LOG.debug(event.toString());
    }

    m_lock.lock();

    try {
      Cluster cluster = clusters.get().getClusterById(event.getClusterId());
      List<HostVersionEntity> hostVersionEntities =
          hostVersionDAO.get().findByClusterAndHost(cluster.getClusterName(), event.getHostName());

      Service service = cluster.getService(event.getServiceName());
      ServiceComponent serviceComponent = service.getServiceComponent(event.getComponentName());
      RepositoryVersionEntity componentRepo = serviceComponent.getDesiredRepositoryVersion();

      for (HostVersionEntity hostVersionEntity : hostVersionEntities) {
        StackEntity hostStackEntity = hostVersionEntity.getRepositoryVersion().getStack();
        StackId hostStackId = new StackId(hostStackEntity);

        // If added components do not advertise version, it makes no sense to mark version OUT_OF_SYNC
        // We perform check per-stack version, because component may be not versionAdvertised in current
        // stack, but become versionAdvertised in some future (installed, but not yet upgraded to) stack
        String serviceName = event.getServiceName();
        String componentName = event.getComponentName();

        // Skip lookup if stack does not contain the component
        if (!ami.get().isValidServiceComponent(hostStackId.getStackName(),
            hostStackId.getStackVersion(), serviceName, componentName)) {
          LOG.debug("Component not found is host stack, stack={}, version={}, service={}, component={}",
              hostStackId.getStackName(), hostStackId.getStackVersion(), serviceName, componentName);
          continue;
        }
        ComponentInfo component = ami.get().getComponent(hostStackId.getStackName(),
                hostStackId.getStackVersion(), serviceName, componentName);

        if (!component.isVersionAdvertised()) {
          RepositoryVersionState state = checkAllHostComponents(hostStackId, hostVersionEntity.getHostEntity());
          if (null != state) {
            hostVersionEntity.setState(state);
            hostVersionDAO.get().merge(hostVersionEntity);
          }
          continue;
        }

        // !!! we shouldn't be changing other versions to OUT_OF_SYNC if the event
        // component repository doesn't match
        if (!hostVersionEntity.getRepositoryVersion().equals(componentRepo)) {
          continue;
        }

        switch (hostVersionEntity.getState()) {
          case INSTALLED:
          case NOT_REQUIRED:
            hostVersionEntity.setState(RepositoryVersionState.OUT_OF_SYNC);
            hostVersionDAO.get().merge(hostVersionEntity);
            break;
          default:
            break;
        }
      }
    } catch (AmbariException e) {
      LOG.error("Can not update hosts about out of sync", e);
    } finally {
      m_lock.unlock();
    }
  }

  @Subscribe
  @Transactional
  public void onServiceComponentHostEvent(ServiceComponentUninstalledEvent event) {

    m_lock.lock();

    try {
      Cluster cluster = clusters.get().getClusterById(event.getClusterId());
      List<HostVersionEntity> hostVersionEntities =
          hostVersionDAO.get().findByClusterAndHost(cluster.getClusterName(), event.getHostName());

      for (HostVersionEntity hostVersionEntity : hostVersionEntities) {
        HostEntity hostEntity = hostVersionEntity.getHostEntity();
        RepositoryVersionEntity repoVersionEntity = hostVersionEntity.getRepositoryVersion();
        StackId stackId = repoVersionEntity.getStackId();

        if (null == stackId) {
          LOG.info("Stack id could not be loaded for host version {}, repo {}", hostVersionEntity.getHostName(),
              repoVersionEntity.getVersion());
          continue;
        }

        RepositoryVersionState repoState = checkAllHostComponents(stackId, hostEntity);
        if (null != repoState) {
          hostVersionEntity.setState(repoState);
          hostVersionDAO.get().merge(hostVersionEntity);
        }
      }

    } catch (AmbariException e) {
      LOG.error("Cannot update states after a component was uninstalled: {}", event, e);
    } finally {
      m_lock.unlock();
    }
  }

  /**
   * Checks if all the components advertise version.  If additional states need to be
   * computed, add on to the logic of this method; make sure the usages are checked for
   * correctness.
   *
   * @param stackId the stack id
   * @param host    the host entity to find components
   * @return {@code null} if there should be no state change.  non-{@code null} to change it
   */
  private RepositoryVersionState checkAllHostComponents(StackId stackId,
      HostEntity host) throws AmbariException {

    Collection<HostComponentDesiredStateEntity> hostComponents = host.getHostComponentDesiredStateEntities();

    for (HostComponentDesiredStateEntity hostComponent : hostComponents) {
      // Skip lookup if stack does not contain the component
      if (!ami.get().isValidServiceComponent(stackId.getStackName(),
          stackId.getStackVersion(), hostComponent.getServiceName(), hostComponent.getComponentName())) {
        LOG.debug("Component not found is host stack, stack={}, version={}, service={}, component={}",
            stackId.getStackName(), stackId.getStackVersion(),
            hostComponent.getServiceName(), hostComponent.getComponentName());
        continue;
      }
      ComponentInfo ci = ami.get().getComponent(stackId.getStackName(), stackId.getStackVersion(),
          hostComponent.getServiceName(), hostComponent.getComponentName());

      if (ci.isVersionAdvertised()) {
        return null;
      }
    }

    return RepositoryVersionState.NOT_REQUIRED;
  }

  @Subscribe
  @Transactional
  public void onServiceEvent(ServiceInstalledEvent event) {
    if (LOG.isDebugEnabled()) {
      LOG.debug(event.toString());
    }

    try {
      Cluster cluster = clusters.get().getClusterById(event.getClusterId());

      Map<String, ServiceComponent> serviceComponents = cluster.getService(event.getServiceName()).getServiceComponents();
      // Determine hosts that become OUT_OF_SYNC when adding components for new service
      Map<String, List<ServiceComponent>> affectedHosts =
        new HashMap<>();
      for (ServiceComponent component : serviceComponents.values()) {
        for (String hostname : component.getServiceComponentHosts().keySet()) {
          if (! affectedHosts.containsKey(hostname)) {
            affectedHosts.put(hostname, new ArrayList<>());
          }
          affectedHosts.get(hostname).add(component);
        }
      }
      for (String hostName : affectedHosts.keySet()) {
        List<HostVersionEntity> hostVersionEntities =
            hostVersionDAO.get().findByClusterAndHost(cluster.getClusterName(), hostName);
        for (HostVersionEntity hostVersionEntity : hostVersionEntities) {
          RepositoryVersionEntity repositoryVersion = hostVersionEntity.getRepositoryVersion();
          // If added components do not advertise version, it makes no sense to mark version OUT_OF_SYNC
          // We perform check per-stack version, because component may be not versionAdvertised in current
          // stack, but become versionAdvertised in some future (installed, but not yet upgraded to) stack
          boolean hasChangedComponentsWithVersions = false;
          String serviceName = event.getServiceName();
          for (ServiceComponent comp : affectedHosts.get(hostName)) {
            String componentName = comp.getName();

            // Skip lookup if stack does not contain the component
            if (!ami.get().isValidServiceComponent(repositoryVersion.getStackName(),
                repositoryVersion.getStackVersion(), serviceName, componentName)) {
              LOG.debug("Component not found is host stack, stack={}, version={}, service={}, component={}",
                  repositoryVersion.getStackName(), repositoryVersion.getStackVersion(), serviceName, componentName);
              continue;
            }
            ComponentInfo component = ami.get().getComponent(repositoryVersion.getStackName(),
                    repositoryVersion.getStackVersion(), serviceName, componentName);
            if (component.isVersionAdvertised()) {
              hasChangedComponentsWithVersions = true;
            }
          }
          if (! hasChangedComponentsWithVersions) {
            continue;
          }

          if (hostVersionEntity.getState().equals(RepositoryVersionState.INSTALLED)) {
            hostVersionEntity.setState(RepositoryVersionState.OUT_OF_SYNC);
            hostVersionDAO.get().merge(hostVersionEntity);
          }
        }
      }

    } catch (AmbariException e) {
      LOG.error("Can not update hosts about out of sync", e);
    }
  }

  /**
   * When hosts are added, add a host_version record for every repo_version in the database.
   *
   * @param event the add event
   */
  @Subscribe
  @Transactional
  public void onHostEvent(HostsAddedEvent event) {
    if (LOG.isDebugEnabled()) {
      LOG.debug(event.toString());
    }

    // create host version entries for every repository
    @Experimental(feature=ExperimentalFeature.PATCH_UPGRADES, comment="Eventually take into account deleted repositories")
    List<RepositoryVersionEntity> repos = repositoryVersionDAO.get().findAll();

    for (String hostName : event.getHostNames()) {
      HostEntity hostEntity = hostDAO.get().findByName(hostName);

      for (RepositoryVersionEntity repositoryVersion : repos) {

        // we don't have the knowledge yet to know if we need the record
        HostVersionEntity missingHostVersion = new HostVersionEntity(hostEntity,
            repositoryVersion, RepositoryVersionState.NOT_REQUIRED);

        LOG.info("Creating host version for {}, state={}, repo={} (repo_id={})",
          missingHostVersion.getHostName(), missingHostVersion.getState(),
          missingHostVersion.getRepositoryVersion().getVersion(), missingHostVersion.getRepositoryVersion().getId());

        hostVersionDAO.get().create(missingHostVersion);
        hostDAO.get().merge(hostEntity);

        hostEntity.getHostVersionEntities().add(missingHostVersion);
        hostEntity = hostDAO.get().merge(hostEntity);
      }
    }
  }

  /**
   * Host repo_version entities are removed via cascade.
   *
   * @param event
   *          the removal event.
   */
  @Subscribe
  @Transactional
  public void onHostEvent(HostsRemovedEvent event) {
    if (LOG.isDebugEnabled()) {
      LOG.debug(event.toString());
    }
  }

}