AMBARI-9464. Adding new host after version distribution does not contain 2 host versions (dlysnichenko)

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

@@ -24,10 +24,13 @@ import com.google.inject.Singleton;
 import com.google.inject.persist.Transactional;
 import org.apache.ambari.server.AmbariException;
 import org.apache.ambari.server.EagerSingleton;
+import org.apache.ambari.server.events.HostAddedEvent;
 import org.apache.ambari.server.events.ServiceComponentInstalledEvent;
 import org.apache.ambari.server.events.ServiceInstalledEvent;
 import org.apache.ambari.server.events.publishers.AmbariEventPublisher;
+import org.apache.ambari.server.orm.dao.HostDAO;
 import org.apache.ambari.server.orm.dao.HostVersionDAO;
+import org.apache.ambari.server.orm.entities.ClusterVersionEntity;
 import org.apache.ambari.server.orm.entities.HostVersionEntity;
 import org.apache.ambari.server.state.Cluster;
 import org.apache.ambari.server.state.Clusters;
@@ -36,6 +39,7 @@ import org.apache.ambari.server.state.StackId;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.Collection;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
@@ -59,6 +63,9 @@ public class HostVersionOutOfSyncListener {
   @Inject
   private Provider<HostVersionDAO> hostVersionDAO;
 
+  @Inject
+  private Provider<HostDAO> hostDAO;
+
   @Inject
   private Provider<Clusters> clusters;
 
@@ -123,4 +130,32 @@ public class HostVersionOutOfSyncListener {
       LOG.error("Can not update hosts about out of sync", e);
     }
   }
+
+  @Subscribe
+  @Transactional
+  public void onHostEvent(HostAddedEvent event) {
+    if (LOG.isDebugEnabled()) {
+      LOG.debug(event.toString());
+    }
+    try {
+      Cluster cluster = clusters.get().getClusterById(event.getClusterId());
+      Set<String> changedRepositoryVersions = new HashSet<String>();
+      Collection<ClusterVersionEntity> allClusterVersions = cluster.getAllClusterVersions();
+      for (ClusterVersionEntity clusterVersion : allClusterVersions) {
+        if (clusterVersion.getState() != RepositoryVersionState.CURRENT) { // Current version is taken care of automatically
+          String hostName = event.getHostName();
+          HostVersionEntity missingHostVersion = new HostVersionEntity(hostName,
+                  clusterVersion.getRepositoryVersion(), RepositoryVersionState.OUT_OF_SYNC);
+          missingHostVersion.setHostEntity(hostDAO.get().findByName(hostName));
+          hostVersionDAO.get().create(missingHostVersion);
+          changedRepositoryVersions.add(clusterVersion.getRepositoryVersion().getVersion());
+        }
+      }
+      for (String version : changedRepositoryVersions) {
+        cluster.recalculateClusterVersionState(version);
+      }
+    } catch (AmbariException e) {
+      LOG.error("Can not update hosts about out of sync", e);
+    }
+  }
 }

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

@@ -92,6 +92,10 @@ public class HostVersionEntity {
   public HostVersionEntity() {
   }
 
+  /**
+   * When using this constructor, you should also call setHostEntity(). Otherwise
+   * you will have persistence errors when persisting the instance.
+   */
   public HostVersionEntity(String hostName, RepositoryVersionEntity repositoryVersion, RepositoryVersionState state) {
     this.hostName = hostName;
     this.repositoryVersion = repositoryVersion;

ambari-server/src/test/java/org/apache/ambari/server/events/listeners/upgrade/HostVersionOutOfSyncListenerTest.java

@@ -0,0 +1,167 @@
+/**
+ * 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 static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+import static org.mockito.Mockito.mock;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import javax.persistence.RollbackException;
+
+import com.google.inject.AbstractModule;
+import com.google.inject.Singleton;
+import com.google.inject.persist.Transactional;
+import com.google.inject.util.Modules;
+
+import org.apache.ambari.server.AmbariException;
+import org.apache.ambari.server.api.services.AmbariMetaInfo;
+import org.apache.ambari.server.events.HostAddedEvent;
+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.HostVersionDAO;
+import org.apache.ambari.server.orm.entities.ClusterVersionEntity;
+import org.apache.ambari.server.orm.entities.HostVersionEntity;
+import org.apache.ambari.server.orm.entities.RepositoryVersionEntity;
+import org.apache.ambari.server.state.Cluster;
+import org.apache.ambari.server.state.RepositoryVersionState;
+import org.apache.ambari.server.state.Clusters;
+import org.apache.ambari.server.state.Host;
+import org.apache.ambari.server.state.HostState;
+import org.apache.ambari.server.state.StackId;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.inject.Guice;
+import com.google.inject.Injector;
+import com.google.inject.persist.PersistService;
+
+public class HostVersionOutOfSyncListenerTest {
+  private static final Logger LOG = LoggerFactory.getLogger(HostVersionOutOfSyncListenerTest.class);
+
+  private Clusters clusters;
+  private Cluster c1;
+  private Injector injector;
+  private OrmTestHelper helper;
+  private HostVersionDAO hostVersionDAO;
+  private AmbariMetaInfo metaInfo;
+
+
+  @Before
+  public void setup() throws Exception {
+    injector = Guice.createInjector(new InMemoryDefaultTestModule());
+    injector.getInstance(GuiceJpaInitializer.class);
+    clusters = injector.getInstance(Clusters.class);
+    metaInfo = injector.getInstance(AmbariMetaInfo.class);
+    helper = injector.getInstance(OrmTestHelper.class);
+    hostVersionDAO = injector.getInstance(HostVersionDAO.class);
+    metaInfo.init();
+    clusters.addCluster("c1");
+    c1 = clusters.getCluster("c1");
+    addHost("h1");
+
+    StackId stackId = new StackId("HDP-0.1");
+    c1.setDesiredStackVersion(stackId);
+    helper.getOrCreateRepositoryVersion(stackId.getStackName(), stackId.getStackVersion());
+    c1.createClusterVersion(stackId.getStackName(), stackId.getStackVersion(), "admin", RepositoryVersionState.UPGRADING);
+    c1.transitionClusterVersion(stackId.getStackName(), stackId.getStackVersion(), RepositoryVersionState.CURRENT);
+    clusters.mapHostToCluster("h1", "c1");
+  }
+
+  @After
+  public void teardown() {
+    injector.getInstance(PersistService.class).stop();
+  }
+
+  /**
+   * When a host is added to a cluster that has non-CURRENT cluster_version records,
+   * then Ambari needs to insert host_verion record for each one of
+   * those stack versions with a state of OUT_OF_SYNC
+   */
+  @Test
+  public void testOnHostEvent() throws AmbariException {
+    // Configuring single-node cluster with 2 repo versions
+    Host h1 = clusters.getHost("h1");
+    h1.setState(HostState.HEALTHY);
+
+    StackId stackId = new StackId("HDP-0.1");
+    RepositoryVersionEntity repositoryVersionEntity = helper.getOrCreateRepositoryVersion(stackId.getStackId(),
+            "1.0-1000");
+    c1.createClusterVersion(stackId.getStackId(), "1.0-1000", "admin", RepositoryVersionState.INSTALLING);
+    c1.setCurrentStackVersion(stackId);
+    c1.recalculateAllClusterVersionStates();
+    checkStackVersionState(stackId.getStackId(), "1.0-1000", RepositoryVersionState.INSTALLING);
+    checkStackVersionState(stackId.getStackId(), "1.0-2086", RepositoryVersionState.CURRENT);
+
+    HostVersionEntity hv1 = helper.createHostVersion("h1", repositoryVersionEntity, RepositoryVersionState.INSTALLED);
+    c1.recalculateAllClusterVersionStates();
+    checkStackVersionState(stackId.getStackId(), "1.0-1000", RepositoryVersionState.INSTALLED);
+    checkStackVersionState(stackId.getStackId(), "1.0-2086", RepositoryVersionState.CURRENT);
+
+    // Add new host and verify that it has all host versions present
+    addHost("h2");
+    clusters.mapHostToCluster("h2", "c1");
+
+    List<HostVersionEntity> h2Versions = hostVersionDAO.findByHost("h2");
+
+    for (HostVersionEntity hostVersionEntity : h2Versions) {
+      if (hostVersionEntity.getRepositoryVersion().toString().equals("1.0-2086")) {
+        assertEquals(hostVersionEntity.getState(), RepositoryVersionState.CURRENT);
+      } else {
+        assertEquals(hostVersionEntity.getState(), RepositoryVersionState.OUT_OF_SYNC);
+      }
+    }
+
+  }
+
+  private void addHost(String hostname) throws AmbariException{
+    clusters.addHost(hostname);
+
+    Host host1 = clusters.getHost(hostname);
+    host1.setIPv4("ipv4");
+    host1.setIPv6("ipv6");
+
+    Map<String, String> hostAttributes = new HashMap<String, String>();
+    hostAttributes.put("os_family", "redhat");
+    hostAttributes.put("os_release_version", "5.9");
+    host1.setHostAttributes(hostAttributes);
+
+    host1.persist();
+  }
+
+  private void checkStackVersionState(String stack, String version, RepositoryVersionState state) {
+    Collection<ClusterVersionEntity> allClusterVersions = c1.getAllClusterVersions();
+    for (ClusterVersionEntity entity : allClusterVersions) {
+      if (entity.getRepositoryVersion().getStack().equals(stack)
+              && entity.getRepositoryVersion().getVersion().equals(version)) {
+        assertEquals(state, entity.getState());
+      }
+    }
+  }
+}