AMBARI-8392. Upgrade Execute: create request/stage/tasks out of upgrade items (ncole)

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

@@ -95,6 +95,10 @@ public class ActionExecutionContext {
     return timeout;
   }
 
+  public void setTimeout(Short timeout) {
+    this.timeout = timeout;
+  }
+
   public List<RequestResourceFilter> getResourceFilters() {
     return resourceFilters;
   }

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

@@ -48,6 +48,7 @@ import org.apache.ambari.server.state.ServiceInfo;
 import org.apache.ambari.server.state.StackId;
 import org.apache.ambari.server.state.svccomphost.ServiceComponentHostOpInProgressEvent;
 import org.apache.ambari.server.utils.StageUtils;
+import org.apache.commons.lang.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -335,7 +336,7 @@ public class AmbariActionExecutionHelper {
       Map<String, Map<String, String>> configurations = new TreeMap<String, Map<String, String>>();
       Map<String, Map<String, Map<String, String>>> configurationAttributes = new TreeMap<String, Map<String, Map<String, String>>>();
       Map<String, Map<String, String>> configTags = null;
-      if (!serviceName.isEmpty() && null != cluster) {
+      if (!StringUtils.isEmpty(serviceName) && null != cluster) {
         configTags = managementController.findConfigurationTagsWithOverrides(cluster, hostName);
       }
 

ambari-server/src/main/java/org/apache/ambari/server/controller/internal/RequestStageContainer.java

@@ -60,6 +60,8 @@ public class RequestStageContainer {
    */
   private ActionManager actionManager;
 
+  private String requestContext = null;
+
   /**
    * Logger
    */
@@ -119,6 +121,15 @@ public class RequestStageContainer {
     return stages.isEmpty() ? -1 : stages.get(stages.size() - 1).getStageId();
   }
 
+  /**
+   * Sets the context for the request (optional operation)
+   *
+   * @param context the new context
+   */
+  public void setRequestContext(String context) {
+    requestContext = context;
+  }
+
   /**
    * Determine the projected state for a host component from the existing stages.
    *
@@ -173,6 +184,11 @@ public class RequestStageContainer {
   public void persist() throws AmbariException {
     if (!stages.isEmpty()) {
       Request request = requestFactory.createNewFromStages(stages);
+
+      if (null != requestContext) {
+        request.setRequestContext(requestContext);
+      }
+
       if (request != null && request.getStages()!= null && !request.getStages().isEmpty()) {
         if (LOG.isDebugEnabled()) {
           LOG.debug(String.format("Triggering Action Manager, request=%s", request));

ambari-server/src/main/java/org/apache/ambari/server/controller/internal/UpgradeResourceProvider.java

@@ -17,6 +17,8 @@
  */
 package org.apache.ambari.server.controller.internal;
 
+import static org.apache.ambari.server.agent.ExecutionCommand.KeyNames.JDK_LOCATION;
+
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
@@ -30,9 +32,13 @@ import java.util.Set;
 import org.apache.ambari.server.AmbariException;
 import org.apache.ambari.server.StaticallyInject;
 import org.apache.ambari.server.actionmanager.ActionManager;
+import org.apache.ambari.server.actionmanager.HostRoleCommand;
 import org.apache.ambari.server.actionmanager.RequestFactory;
+import org.apache.ambari.server.actionmanager.Stage;
 import org.apache.ambari.server.actionmanager.StageFactory;
 import org.apache.ambari.server.api.services.AmbariMetaInfo;
+import org.apache.ambari.server.controller.ActionExecutionContext;
+import org.apache.ambari.server.controller.AmbariActionExecutionHelper;
 import org.apache.ambari.server.controller.AmbariManagementController;
 import org.apache.ambari.server.controller.spi.NoSuchParentResourceException;
 import org.apache.ambari.server.controller.spi.NoSuchResourceException;
@@ -54,9 +60,8 @@ import org.apache.ambari.server.state.ServiceComponent;
 import org.apache.ambari.server.state.StackId;
 import org.apache.ambari.server.state.stack.UpgradePack;
 import org.apache.ambari.server.state.stack.UpgradePack.ProcessingComponent;
-import org.apache.ambari.server.state.stack.upgrade.CountBatch;
-import org.apache.ambari.server.state.stack.upgrade.PercentBatch;
 import org.apache.ambari.server.state.stack.upgrade.Task;
+import org.apache.ambari.server.utils.StageUtils;
 
 import com.google.gson.Gson;
 import com.google.inject.Inject;
@@ -85,9 +90,11 @@ public class UpgradeResourceProvider extends AbstractControllerResourceProvider
   @Inject
   private static RepositoryVersionDAO m_repoVersionDAO = null;
   @Inject
-  private RequestFactory requestFactory;
+  private static Provider<RequestFactory> requestFactory;
+  @Inject
+  private static Provider<StageFactory> stageFactory;
   @Inject
-  private StageFactory stageFactory;
+  private static Provider<AmbariActionExecutionHelper> actionExecutionHelper;
 
 
   static {
@@ -305,7 +312,6 @@ public class UpgradeResourceProvider extends AbstractControllerResourceProvider
       }
     }
 
-
     Gson gson = new Gson();
 
     UpgradeEntity entity = new UpgradeEntity();
@@ -333,11 +339,24 @@ public class UpgradeResourceProvider extends AbstractControllerResourceProvider
 
     m_upgradeDAO.create(entity);
 
-//    RequestStageContainer req = createRequest();
-//
-//    req.getRequestStatusResponse();
+    RequestStageContainer req = createRequest((String) requestMap.get(UPGRADE_VERSION));
+
+    for (StageHolder holder : preUpgrades) {
+      createUpgradeTaskStage(cluster, req, holder);
+    }
+
+    for (StageHolder holder : restart) {
+      createRestartStage(cluster, req, holder);
+    }
+
+    for (StageHolder holder : postUpgrades) {
+      createUpgradeTaskStage(cluster, req, holder);
+    }
+
+    req.getRequestStatusResponse();
+
+    req.persist();
 
-//  TODO req.persist();
     return entity;
   }
 
@@ -449,32 +468,81 @@ public class UpgradeResourceProvider extends AbstractControllerResourceProvider
     private TaskHolder taskHolder;
     private UpgradeItemEntity upgradeItemEntity;
     private Set<String> hosts;
-
-    @Override
-    public String toString() {
-      // TODO Auto-generated method stub
-      return upgradeItemEntity.toString();
-    }
   }
 
 
-  private RequestStageContainer createRequest() {
-
+  private RequestStageContainer createRequest(String version) {
     ActionManager actionManager = getManagementController().getActionManager();
 
     RequestStageContainer requestStages = new RequestStageContainer(
-        actionManager.getNextRequestId(), null, requestFactory, actionManager);
+        actionManager.getNextRequestId(), null, requestFactory.get(), actionManager);
+    requestStages.setRequestContext(String.format("Upgrading to %s", version));
 
-    /*
-    List<Stage> stages = doStageCreation(requestStages, cluster, changedServices, changedComponents,
-        changedHosts, requestParameters, requestProperties,
-        runSmokeTest, reconfigureClients);
-    LOG.debug("Created {} stages", ((stages != null) ? stages.size() : 0));
-
-    requestStages.addStages(stages);
-    updateServiceStates(changedServices, changedComponents, changedHosts, ignoredHosts);
-    */
     return requestStages;
   }
 
+  /**
+   * Creates a stage and appends it to the request.
+   * @param cluster the cluster
+   * @param request the request container
+   * @param holder the holder
+   * @throws AmbariException
+   */
+  private void createUpgradeTaskStage(Cluster cluster, RequestStageContainer request,
+      StageHolder holder) throws AmbariException {
+
+    Map<String, String> hostLevelParams = new HashMap<String, String>();
+    hostLevelParams.put(JDK_LOCATION, getManagementController().getJdkResourceUrl());
+
+    Stage stage = stageFactory.get().createNew(request.getId().longValue(),
+        "/tmp/ambari",
+        cluster.getClusterName(),
+        cluster.getClusterId(),
+        holder.upgradeItemEntity.getText(),
+        "{}", "{}",
+        StageUtils.getGson().toJson(hostLevelParams));
+
+    long stageId = request.getLastStageId() + 1;
+    if (0L == stageId) {
+      stageId = 1L;
+    }
+
+    stage.setStageId(stageId);
+
+    // add each host to this stage
+    RequestResourceFilter filter = new RequestResourceFilter("", "",
+        new ArrayList<String>(holder.hosts));
+
+    // !!! TODO when the custom action is underway, change this
+    Map<String, String> params = new HashMap<String, String>();
+    params.put("tasks", "TheTaskInfo");
+
+    ActionExecutionContext actionContext = new ActionExecutionContext(
+        cluster.getClusterName(), "ru_execute_tasks",
+        Collections.singletonList(filter),
+        params);
+    actionContext.setTimeout(Short.valueOf((short)60));
+
+    // !!! TODO validate the action is valid
+
+    actionExecutionHelper.get().addExecutionCommandsToStage(actionContext, stage);
+
+    // need to set meaningful text on the command
+    for (Map<String, HostRoleCommand> map : stage.getHostRoleCommands().values()) {
+      for (HostRoleCommand hrc : map.values()) {
+        hrc.setCommandDetail(holder.upgradeItemEntity.getText());
+      }
+    }
+
+    request.addStages(Collections.singletonList(stage));
+  }
+
+  private void createRestartStage(Cluster cluster, RequestStageContainer request,
+      StageHolder holder) throws AmbariException {
+
+    // !!! TODO make a restart command
+    createUpgradeTaskStage(cluster, request, holder);
+
+  }
+
 }

ambari-server/src/main/resources/custom_action_definitions/system_action_definitions.xml

@@ -49,4 +49,13 @@
     <description>Distribute repositories and install packages</description>
     <targetType>ALL</targetType>
   </actionDefinition>
+  <actionDefinition>
+    <actionName>ru_execute_tasks</actionName>
+    <actionType>SYSTEM</actionType>
+    <inputs>tasks</inputs>
+    <targetService/>
+    <targetComponent/>
+    <description>Perform upgrade action</description>
+    <targetType>ANY</targetType>
+  </actionDefinition>
 </actionDefinitions>

ambari-server/src/main/resources/custom_actions/scripts/ru_execute_tasks.py

@@ -0,0 +1,49 @@
+#!/usr/bin/env python
+"""
+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.
+
+Ambari Agent
+
+"""
+
+import json
+import sys
+import traceback
+from resource_management import *
+from resource_management.libraries.functions.list_ambari_managed_repos import *
+from ambari_commons.os_check import OSCheck, OSConst
+
+
+class ExecuteUpgradeTasks(Script):
+  """
+  This script is a part of Rolling Upgrade workflow and is described at
+  appropriate design doc.
+
+  It executes tasks used for rolling upgrades.
+  """
+
+  def actionexecute(self, env):
+
+    # Parse parameters
+    config = Script.get_config()
+    #tasks = json.loads(config['roleParams']['tasks'])
+
+    print str(config)
+
+
+if __name__ == "__main__":
+  ExecuteUpgradeTasks().execute()

ambari-server/src/test/java/org/apache/ambari/server/controller/internal/UpgradeResourceProviderTest.java

@@ -17,22 +17,18 @@
  */
 package org.apache.ambari.server.controller.internal;
 
-import static org.easymock.EasyMock.anyObject;
-import static org.easymock.EasyMock.capture;
-import static org.easymock.EasyMock.createMock;
-import static org.easymock.EasyMock.createStrictMock;
-import static org.easymock.EasyMock.expect;
-import static org.easymock.EasyMock.expectLastCall;
-import static org.easymock.EasyMock.replay;
-import static org.easymock.EasyMock.verify;
 import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertTrue;
 
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 
+import org.apache.ambari.server.actionmanager.ActionManager;
+import org.apache.ambari.server.actionmanager.HostRoleCommand;
+import org.apache.ambari.server.actionmanager.RequestStatus;
+import org.apache.ambari.server.actionmanager.Stage;
 import org.apache.ambari.server.controller.AmbariManagementController;
 import org.apache.ambari.server.controller.spi.Request;
 import org.apache.ambari.server.controller.utilities.PropertyHelper;
@@ -44,21 +40,17 @@ import org.apache.ambari.server.orm.entities.RepositoryVersionEntity;
 import org.apache.ambari.server.orm.entities.UpgradeEntity;
 import org.apache.ambari.server.state.Cluster;
 import org.apache.ambari.server.state.Clusters;
+import org.apache.ambari.server.state.Host;
 import org.apache.ambari.server.state.Service;
 import org.apache.ambari.server.state.ServiceComponent;
-import org.apache.ambari.server.state.ServiceComponentHost;
 import org.apache.ambari.server.state.StackId;
-import org.easymock.Capture;
 import org.junit.After;
 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.persist.PersistService;
-import com.google.inject.util.Modules;
 
 /**
  * UpgradeResourceDefinition tests.
@@ -67,30 +59,57 @@ public class UpgradeResourceProviderTest {
 
   private UpgradeDAO upgradeDao = null;
   private RepositoryVersionDAO repoVersionDao = null;
-  private Injector m_injector;
+  private Injector injector;
+  private Clusters clusters;
 
   @Before
-  public void before() {
-    upgradeDao = createStrictMock(UpgradeDAO.class);
-    repoVersionDao = createStrictMock(RepositoryVersionDAO.class);
-
-    m_injector = Guice.createInjector(Modules.override(
-        new InMemoryDefaultTestModule()).with(new Module() {
-
-          @Override
-          public void configure(Binder binder) {
-            // TODO Auto-generated method stub
-            binder.bind(UpgradeDAO.class).toInstance(upgradeDao);
-            binder.bind(RepositoryVersionDAO.class).toInstance(repoVersionDao);
-          }
-        }));
-    m_injector.getInstance(GuiceJpaInitializer.class);
+  public void before() throws Exception {
+
+    injector = Guice.createInjector(new InMemoryDefaultTestModule());
+    injector.getInstance(GuiceJpaInitializer.class);
+
+    upgradeDao = injector.getInstance(UpgradeDAO.class);
+    repoVersionDao = injector.getInstance(RepositoryVersionDAO.class);
+
+    RepositoryVersionEntity repoVersionEntity = new RepositoryVersionEntity();
+    repoVersionEntity.setDisplayName("My New Version");
+    repoVersionEntity.setRepositories("");
+    repoVersionEntity.setStack("HDP-2.1.1");
+    repoVersionEntity.setUpgradePackage("upgrade_test");
+    repoVersionEntity.setVersion("2.2.2.2");
+
+    repoVersionDao.create(repoVersionEntity);
+
+    clusters = injector.getInstance(Clusters.class);
+    clusters.addCluster("c1");
+    Cluster cluster = clusters.getCluster("c1");
+    cluster.setDesiredStackVersion(new StackId("HDP-2.1.1"));
+
+    clusters.addHost("h1");
+    Host host = clusters.getHost("h1");
+    Map<String, String> hostAttributes = new HashMap<String, String>();
+    hostAttributes.put("os_family", "redhat");
+    hostAttributes.put("os_release_version", "6.3");
+    host.setHostAttributes(hostAttributes);
+    host.persist();
+
+    clusters.mapHostToCluster("h1", "c1");
+
+    // add a single ZK server
+    Service service = cluster.addService("ZOOKEEPER");
+    service.setDesiredStackVersion(cluster.getDesiredStackVersion());
+    service.persist();
+
+    ServiceComponent component = service.addServiceComponent("ZOOKEEPER_SERVER");
+    component.addServiceComponentHost("h1");
+
+
   }
 
   @After
   public void after() {
-    m_injector.getInstance(PersistService.class).stop();
-    m_injector = null;
+    injector.getInstance(PersistService.class).stop();
+    injector = null;
   }
 
   /**
@@ -98,23 +117,13 @@ public class UpgradeResourceProviderTest {
    */
   @Test
   public void testCreateResources() throws Exception {
-    AmbariManagementController amc = createMock(AmbariManagementController.class);
-    Clusters clusters = createMock(Clusters.class);
-    Cluster cluster = createCluster();
-
-    expect(amc.getClusters()).andReturn(clusters).atLeastOnce();
-    expect(clusters.getCluster((String) anyObject())).andReturn(cluster).atLeastOnce();
 
-    RepositoryVersionEntity repoVersionEntity = new RepositoryVersionEntity();
-    repoVersionEntity.setUpgradePackage("upgrade_test");
+    Cluster cluster = clusters.getCluster("c1");
 
-    expect(repoVersionDao.findByStackAndVersion((String)anyObject(),(String)anyObject())).andReturn(repoVersionEntity).atLeastOnce();
+    List<UpgradeEntity> upgrades = upgradeDao.findUpgrades(cluster.getClusterId());
+    assertEquals(0, upgrades.size());
 
-    Capture<UpgradeEntity> entityCapture = new Capture<UpgradeEntity>();
-    upgradeDao.create(capture(entityCapture));
-    expectLastCall();
-
-    replay(amc, clusters, cluster, upgradeDao, repoVersionDao);
+    AmbariManagementController amc = injector.getInstance(AmbariManagementController.class);
 
     UpgradeResourceProvider provider = createProvider(amc);
 
@@ -125,17 +134,27 @@ public class UpgradeResourceProviderTest {
     Request request = PropertyHelper.getCreateRequest(Collections.singleton(requestProps), null);
     provider.createResources(request);
 
-    assertTrue(entityCapture.hasCaptured());
-    UpgradeEntity entity = entityCapture.getValue();
-    assertNotNull(entity);
-    assertEquals(Long.valueOf(1), entity.getClusterId());
+
+    upgrades = upgradeDao.findUpgrades(cluster.getClusterId());
+    assertEquals(1, upgrades.size());
+
+    UpgradeEntity entity = upgrades.get(0);
+    assertEquals(cluster.getClusterId(), entity.getClusterId().longValue());
     assertEquals(3, entity.getUpgradeItems().size());
 
     assertTrue(entity.getUpgradeItems().get(0).getText().contains("Preparing"));
     assertTrue(entity.getUpgradeItems().get(1).getText().contains("Restarting"));
     assertTrue(entity.getUpgradeItems().get(2).getText().contains("Finalizing"));
 
-    verify(amc, clusters, cluster, upgradeDao);
+    ActionManager am = injector.getInstance(ActionManager.class);
+    List<Long> requests = am.getRequestsByStatus(RequestStatus.IN_PROGRESS, 100, true);
+
+    assertEquals(1, requests.size());
+    List<Stage> stages = am.getRequestStatus(requests.get(0).longValue());
+    assertEquals(3, stages.size());
+
+    List<HostRoleCommand> tasks = am.getRequestTasks(requests.get(0).longValue());
+    assertEquals(3, tasks.size());
   }
 
 
@@ -147,36 +166,5 @@ public class UpgradeResourceProviderTest {
     return new UpgradeResourceProvider(amc);
   }
 
-  private Cluster createCluster() {
-    Cluster cluster = createMock(Cluster.class);
-
-    expect(cluster.getClusterId()).andReturn(Long.valueOf(1)).anyTimes();
-    expect(cluster.getDesiredStackVersion()).andReturn(new StackId("HDP-2.1.1")).atLeastOnce();
-
-    final ServiceComponentHost zk_server_host_comp = createStrictMock(ServiceComponentHost.class);
-    Map<String, ServiceComponentHost> zk_host_comp_map = new HashMap<String, ServiceComponentHost>() {{
-      put("h1", zk_server_host_comp);
-    }};
-
-    final ServiceComponent zk_server_comp = createStrictMock(ServiceComponent.class);
-    expect(zk_server_comp.getServiceComponentHosts()).andReturn(zk_host_comp_map).atLeastOnce();
-
-    Map<String, ServiceComponent> zk_comp_map = new HashMap<String, ServiceComponent>() {{
-      put("ZOOKEEPER_SERVER", zk_server_comp);
-    }};
-
-    final Service zk = createStrictMock(Service.class);
-    expect(zk.getServiceComponents()).andReturn(zk_comp_map).atLeastOnce();
-
-    Map<String, Service> servicesMap = new HashMap<String, Service>() {{
-      put("ZOOKEEPER", zk);
-    }};
-    expect(cluster.getServices()).andReturn(servicesMap).anyTimes();
-
-    replay(zk_server_host_comp, zk_server_comp, zk);
-
-
-    return cluster;
-  }
 
 }