AMBARI-8675. Rolling Upgrade - Upgrade Pack to complete HBase (ncole)

ambari-server/src/main/java/org/apache/ambari/server/actionmanager/ActionDBAccessor.java

@@ -149,9 +149,14 @@ public interface ActionDBAccessor {
   public Collection<HostRoleCommand> getTasks(Collection<Long> taskIds);
 
   /**
-   * Get a List of host role commands where the role and status are as specified
+   * Get a List of host role commands where the host, role and status are as specified
    */
   public List<HostRoleCommand> getTasksByHostRoleAndStatus(String hostname, String role, HostRoleStatus status);
+  
+  /**
+   * Get a List of host role commands where the role and status are as specified
+   */
+  public List<HostRoleCommand> getTasksByRoleAndStatus(String role, HostRoleStatus status);
 
   /**
    * Get all stages that contain tasks with specified host role statuses

ambari-server/src/main/java/org/apache/ambari/server/actionmanager/ActionDBAccessorImpl.java

@@ -554,6 +554,11 @@ public class ActionDBAccessorImpl implements ActionDBAccessor {
     return getTasks(hostRoleCommandDAO.findTaskIdsByHostRoleAndStatus(hostname, role, status));
   }
 
+  @Override
+  public List<HostRoleCommand> getTasksByRoleAndStatus(String role, HostRoleStatus status) {
+    return getTasks(hostRoleCommandDAO.findTaskIdsByRoleAndStatus(role, status));
+  }
+  
   @Override
   public List<Stage> getStagesByHostRoleStatus(Set<HostRoleStatus> statuses) {
     List<Stage> stages = new ArrayList<Stage>();

ambari-server/src/main/java/org/apache/ambari/server/api/services/UpgradeItemService.java

@@ -72,7 +72,7 @@ public class UpgradeItemService extends BaseService {
       @Context HttpHeaders headers,
       @Context UriInfo ui,
       @PathParam("upgradeItemId") Long id) {
-    return new TaskService(m_clusterName, m_upgradeId, m_upgradeId.toString());
+    return new TaskService(m_clusterName, m_upgradeId, id.toString());
   }
 
   /**
@@ -88,6 +88,7 @@ public class UpgradeItemService extends BaseService {
 
     if (null != upgradeItemId) {
       mapIds.put(Resource.Type.UpgradeItem, upgradeItemId.toString());
+      mapIds.put(Resource.Type.Stage, upgradeItemId.toString());
     }
 
     return createResource(Resource.Type.UpgradeItem, mapIds);

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

@@ -410,7 +410,10 @@ public class AmbariCustomCommandExecutionHelper {
     final Cluster cluster = clusters.getCluster(clusterName);
     final String componentName = actionMetadata.getClient(resourceFilter.getServiceName());
     final String serviceName = resourceFilter.getServiceName();
-    String smokeTestRole = actionExecutionContext.getActionName();
+    String smokeTestRole = actionMetadata.getServiceCheckAction(serviceName);
+    if (null == smokeTestRole) {
+      smokeTestRole = actionExecutionContext.getActionName();
+    }
     long nowTimestamp = System.currentTimeMillis();
     Map<String, String> actionParameters = actionExecutionContext.getParameters();
     final Set<String> candidateHosts;

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

@@ -310,43 +310,39 @@ public class UpgradeResourceProvider extends AbstractControllerResourceProvider
   }
 
   /**
-   * Inject variables into the list of {@link org.apache.ambari.server.orm.entities.UpgradeItemEntity} items, whose
+   * Inject variables into the {@link org.apache.ambari.server.orm.entities.UpgradeItemEntity}, whose
    * tasks may use strings like {{configType/propertyName}} that need to be retrieved from the properties.
    * @param configHelper Configuration Helper
    * @param cluster Cluster
-   * @param items Collection of items whose tasks will be injected.
-   * @return Return the collection of items with the injected properties.
+   * @param upgradeItem the item whose tasks will be injected.
    */
-  private List<UpgradeItemEntity> injectVariables(ConfigHelper configHelper, Cluster cluster, List<UpgradeItemEntity> items) {
+  private void injectVariables(ConfigHelper configHelper, Cluster cluster, UpgradeItemEntity upgradeItem) {
     final String regexp = "(\\{\\{.*?\\}\\})";
 
-    for (UpgradeItemEntity upgradeItem : items) {
-      String task = upgradeItem.getTasks();
-      if (task != null && !task.isEmpty()) {
-        Matcher m = Pattern.compile(regexp).matcher(task);
-        while(m.find()) {
-          String origVar = m.group(1);
-          String formattedVar = origVar.substring(2, origVar.length() - 2).trim();
-
-          int posConfigFile = formattedVar.indexOf("/");
-          if (posConfigFile > 0) {
-            String configType = formattedVar.substring(0, posConfigFile);
-            String propertyName = formattedVar.substring(posConfigFile + 1, formattedVar.length());
-            try {
-              // TODO, some properties use 0.0.0.0 to indicate the current host.
-              // Right now, ru_execute_tasks.py is responsible for replacing 0.0.0.0 with the hostname.
-
-              String configValue = configHelper.getPropertyValueFromStackDefinitions(cluster, configType, propertyName);
-              task = task.replace(origVar, configValue);
-            } catch (Exception err) {
-              LOG.error(String.format("Exception trying to retrieve property %s/%s. Error: %s", configType, propertyName, err.getMessage()));
-            }
+    String task = upgradeItem.getTasks();
+    if (task != null && !task.isEmpty()) {
+      Matcher m = Pattern.compile(regexp).matcher(task);
+      while(m.find()) {
+        String origVar = m.group(1);
+        String formattedVar = origVar.substring(2, origVar.length() - 2).trim();
+
+        int posConfigFile = formattedVar.indexOf("/");
+        if (posConfigFile > 0) {
+          String configType = formattedVar.substring(0, posConfigFile);
+          String propertyName = formattedVar.substring(posConfigFile + 1, formattedVar.length());
+          try {
+            // TODO, some properties use 0.0.0.0 to indicate the current host.
+            // Right now, ru_execute_tasks.py is responsible for replacing 0.0.0.0 with the hostname.
+
+            String configValue = configHelper.getPropertyValueFromStackDefinitions(cluster, configType, propertyName);
+            task = task.replace(origVar, configValue);
+          } catch (Exception err) {
+            LOG.error(String.format("Exception trying to retrieve property %s/%s. Error: %s", configType, propertyName, err.getMessage()));
           }
         }
-        upgradeItem.setTasks(task);
       }
+      upgradeItem.setTasks(task);
     }
-    return items;
   }
 
   private UpgradeEntity createUpgrade(UpgradePack pack, Map<String, Object> requestMap)
@@ -381,13 +377,13 @@ public class UpgradeResourceProvider extends AbstractControllerResourceProvider
         itemEntity.setTasks(wrapper.getTasksJson());
         itemEntity.setHosts(wrapper.getHostsJson());
         itemEntities.add(itemEntity);
+        
+        injectVariables(configHelper, cluster, itemEntity);
 
         // upgrade items match a stage
         createStage(cluster, req, version, itemEntity, wrapper);
       }
 
-      itemEntities = injectVariables(configHelper, cluster, itemEntities);
-
       groupEntity.setItems(itemEntities);
 
       groupEntities.add(groupEntity);
@@ -622,7 +618,7 @@ public class UpgradeResourceProvider extends AbstractControllerResourceProvider
         RoleCommand.EXECUTE,
         cluster.getClusterName(), host,
         new ServiceComponentHostServerActionEvent(StageUtils.getHostName(), System.currentTimeMillis()),
-        Collections.<String, String>emptyMap(), 1200);
+        Collections.<String, String>emptyMap(), 15);
 
     request.addStages(Collections.singletonList(stage));
 

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

@@ -137,6 +137,16 @@ public class HostRoleCommandDAO {
     return daoUtils.selectList(query, hostname, role, status);
   }
 
+  public List<Long> findTaskIdsByRoleAndStatus(String role, HostRoleStatus status) {
+    TypedQuery<Long> query = entityManagerProvider.get().createQuery(
+        "SELECT DISTINCT task.taskId FROM HostRoleCommandEntity task " +
+            "WHERE task.role=?1 AND task.status=?2 " +
+            "ORDER BY task.taskId", Long.class);
+
+    return daoUtils.selectList(query, role, status);
+  }
+  
+  
   @RequiresSession
   public List<HostRoleCommandEntity> findSortedCommandsByStageAndHost(StageEntity stageEntity, HostEntity hostEntity) {
     TypedQuery<HostRoleCommandEntity> query = entityManagerProvider.get().createQuery("SELECT hostRoleCommand " +

ambari-server/src/main/java/org/apache/ambari/server/serveraction/ServerActionExecutor.java

@@ -319,6 +319,15 @@ public class ServerActionExecutor {
   public void doWork() throws InterruptedException {
     List<HostRoleCommand> tasks = db.getTasksByHostRoleAndStatus(serverHostName,
         Role.AMBARI_SERVER_ACTION.toString(), HostRoleStatus.QUEUED);
+    
+    if (null == tasks || tasks.isEmpty()) {
+      // !!! if the server is not a part of the cluster,
+      // !!! just look for anything designated AMBARI_SERVER_ACTION.
+      // !!! do we even need to worry about servername in the first place?  We're
+      // !!! _on_ the ambari server!
+      tasks = db.getTasksByRoleAndStatus(Role.AMBARI_SERVER_ACTION.name(),
+          HostRoleStatus.QUEUED);
+    }
 
     if ((tasks != null) && !tasks.isEmpty()) {
       for (HostRoleCommand task : tasks) {

ambari-server/src/main/java/org/apache/ambari/server/state/stack/upgrade/ColocatedGrouping.java

@@ -130,15 +130,17 @@ public class ColocatedGrouping extends Grouping {
       results.addAll(fromProxies(initialBatch));
 
       // !!! TODO when manual tasks are ready
-//      StageWrapper wrapper = new StageWrapper(
-//      ManualTask task = new ManualTask();
-//      task.message = batch.message;
-//      wrapper.tasks.add(new TaskWrapper(null, null, null, task));
-//      results.add(wrapper);
+      ManualTask task = new ManualTask();
+      task.message = batch.message;
+      
+      StageWrapper wrapper = new StageWrapper(
+          StageWrapper.Type.MANUAL,
+          "Validate partial upgrade",
+          new TaskWrapper(null, null, Collections.<String>emptySet(), task));
+      results.add(wrapper);
 
       results.addAll(fromProxies(finalBatches));
 
-
       return results;
     }
 

ambari-server/src/main/resources/key_properties.json

@@ -40,6 +40,7 @@
   "Task": {
     "Cluster": "Tasks/cluster_name",
     "Request": "Tasks/request_id",
+    "Stage": "Tasks/stage_id",
     "Task": "Tasks/id"
   },
   "User": {

ambari-server/src/main/resources/stacks/HDP/2.0.6/services/HBASE/package/scripts/hbase_master.py

@@ -24,6 +24,7 @@ from resource_management import *
 from hbase import hbase
 from hbase_service import hbase_service
 from hbase_decommission import hbase_decommission
+import upgrade
 
          
 class HbaseMaster(Script):
@@ -36,6 +37,11 @@ class HbaseMaster(Script):
 
     hbase(name='master')
     
+  def pre_rolling_restart(self, env):
+    import params
+    env.set_params(params)
+    upgrade.prestart(env, "hbase-master")
+
   def start(self, env, rolling_restart=False):
     import params
     env.set_params(params)

ambari-server/src/main/resources/stacks/HDP/2.0.6/services/HBASE/package/scripts/hbase_regionserver.py

@@ -23,6 +23,7 @@ from resource_management import *
 
 from hbase import hbase
 from hbase_service import hbase_service
+import upgrade
 
          
 class HbaseRegionServer(Script):
@@ -35,6 +36,16 @@ class HbaseRegionServer(Script):
 
     hbase(name='regionserver')
       
+  def pre_rolling_restart(self, env):
+    import params
+    env.set_params(params)
+    upgrade.prestart(env, "hbase-regionserver")
+
+  def post_rolling_restart(self, env):
+    import params
+    env.set_params(params)
+    upgrade.post_regionserver(env)
+
   def start(self, env, rolling_restart=False):
     import params
     env.set_params(params)
@@ -43,6 +54,7 @@ class HbaseRegionServer(Script):
     hbase_service( 'regionserver',
       action = 'start'
     )
+
     
   def stop(self, env, rolling_restart=False):
     import params

ambari-server/src/main/resources/stacks/HDP/2.0.6/services/HBASE/package/scripts/params.py

@@ -27,6 +27,8 @@ import status_params
 config = Script.get_config()
 exec_tmp_dir = Script.get_tmp_dir()
 
+version = default("/commandParams/version", None)
+
 hdp_stack_version = str(config['hostLevelParams']['stack_version'])
 hdp_stack_version = format_hdp_stack_version(hdp_stack_version)
 

ambari-server/src/main/resources/stacks/HDP/2.0.6/services/HBASE/package/scripts/upgrade.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.
+
+"""
+from resource_management import *
+from resource_management.core.resources.system import Execute
+from resource_management.core.shell import call
+from resource_management.libraries.functions.version import compare_versions, format_hdp_stack_version
+from resource_management.libraries.functions.decorator import retry
+
+def prestart(env, hdp_component):
+  import params
+
+  if params.version and compare_versions(format_hdp_stack_version(params.version), '2.2.0.0') >= 0:
+    Execute("hdp-select set {0} {1}".format(hdp_component, params.version))
+
+def post_regionserver(env):
+  import params
+  env.set_params(params)
+
+  check_cmd = "echo 'status \"simple\"' | {0} shell".format(params.hbase_cmd)
+
+  exec_cmd = "{0} {1}".format(params.kinit_cmd, check_cmd)
+  call_and_match(exec_cmd, params.hbase_user, params.hostname.lower() + ":")
+
+
+@retry(times=15, sleep_time=2, err_class=Fail)
+def call_and_match(cmd, user, regex):
+
+  code, out = call(cmd, user=user)
+
+  if not (out and re.search(regex, out)):
+    raise Fail("Could not verify RS available")

ambari-server/src/main/resources/stacks/HDP/2.2/upgrades/upgrade-2.2.xml

@@ -55,6 +55,9 @@
         <component>RESOURCEMANAGER</component>
         <component>NODEMANAGER</component>
       </service>
+      <service name="HBASE">
+        <component>HBASE_MASTER</component>
+      </service>
     </group>
 
     <group name="CORE_SLAVES" title="Core Slaves" xsi:type="colocated">
@@ -63,7 +66,7 @@
       </service>
 
       <service name="HBASE">
-        <component>REGIONSERVER</component>
+        <component>HBASE_REGIONSERVER</component>
       </service>
 
       <service name="YARN">
@@ -76,6 +79,7 @@
       </batch>
     </group>
     
+    <!--
     <group xsi:type="cluster" name="POST_CLUSTER" title="Finalize Upgrade">
       <execute-stage title="Confirm Finalize">
         <task xsi:type="manual">
@@ -88,7 +92,7 @@
         </task>
       </execute-stage>
     </group>
-    
+    -->    
   </order>
   
   
@@ -280,5 +284,23 @@
       </component>
 
     </service>
+    <service name="HBASE">
+      <component name="HBASE_MASTER">
+        <pre-upgrade>
+          <task xsi:type="execute">
+            <command>su - {{hbase-env/hbase_user}} -c 'echo "snapshot_all" | hbase shell'</command>
+          </task>
+        </pre-upgrade>
+        <upgrade>
+          <task xsi:type="restart" />
+        </upgrade>
+      </component>
+
+      <component name="HBASE_REGIONSERVER">
+        <upgrade>
+          <task xsi:type="restart" />
+        </upgrade>
+      </component>
+    </service>
   </processing>
 </upgrade>

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

@@ -184,11 +184,11 @@ public class UpgradeResourceProviderTest {
 
     List<Stage> stages = am.getRequestStatus(requests.get(0).longValue());
 
-    assertEquals(7, stages.size());
+    assertEquals(8, stages.size());
 
     List<HostRoleCommand> tasks = am.getRequestTasks(requests.get(0).longValue());
     // same number of tasks as stages here
-    assertEquals(7, tasks.size());
+    assertEquals(8, tasks.size());
 
     return status;
   }

ambari-server/src/test/java/org/apache/ambari/server/state/UpgradeHelperTest.java

@@ -87,7 +87,7 @@ public class UpgradeHelperTest {
 
     assertEquals(6, groups.get(1).items.size());
     assertEquals(2, groups.get(2).items.size());
-    assertEquals(6, groups.get(3).items.size());
+    assertEquals(7, groups.get(3).items.size());
   }
 
   public Cluster makeCluster() throws AmbariException {