AMBARI-11438 - RU: Storm actions require Zookeeper host (jonathanhurley)

ambari-common/src/main/python/ambari_commons/yaml_utils.py

@@ -0,0 +1,68 @@
+#!/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.
+
+"""
+import re
+
+def escape_yaml_property(value):
+  unquouted = False
+  unquouted_values = ["null", "Null", "NULL", "true", "True", "TRUE", "false",
+    "False", "FALSE", "YES", "Yes", "yes", "NO", "No", "no", "ON", "On", "on",
+    "OFF", "Off", "off"]
+
+  if value in unquouted_values:
+    unquouted = True
+
+  # if is list [a,b,c] or dictionary {a: v, b: v2, c: v3}
+  if re.match('^\w*\[.+\]\w*$', value) or re.match('^\w*\{.+\}\w*$', value):
+    unquouted = True
+
+  try:
+    int(value)
+    unquouted = True
+  except ValueError:
+    pass
+
+  try:
+    float(value)
+    unquouted = True
+  except ValueError:
+    pass
+
+  if not unquouted:
+    value = value.replace("'", "''")
+    value = "'" + value + "'"
+
+  return value
+
+def get_values_from_yaml_array(yaml_array):
+  """
+  Converts a YAML array into a normal array of values. For example, this
+  will turn ['c6401','c6402']
+  into an array with two strings, "c6401" and "c6402".
+  :param yaml_array: the YAML array to convert
+  :return:  a python array or None if the value could not be converted correctly.
+  """
+  if yaml_array is None:
+    return None
+
+  matches = re.findall(r'[\'|\"](.+?)[\'|\"]', yaml_array)
+  if matches is None or len(matches) == 0:
+    return None
+
+  return matches

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

@@ -72,12 +72,12 @@ public class ClusterConfigEntity {
   private String tag;
 
   @Basic(fetch = FetchType.LAZY)
-  @Column(name = "config_data", nullable = false, insertable = true, updatable = false)
+  @Column(name = "config_data", nullable = false, insertable = true)
   @Lob
   private String configJson;
 
   @Basic(fetch = FetchType.LAZY)
-  @Column(name = "config_attributes", nullable = true, insertable = true, updatable = false)
+  @Column(name = "config_attributes", nullable = true, insertable = true)
   @Lob
   private String configAttributesJson;
 

ambari-server/src/main/java/org/apache/ambari/server/state/ConfigImpl.java

@@ -28,6 +28,8 @@ import org.apache.ambari.server.orm.dao.ClusterDAO;
 import org.apache.ambari.server.orm.dao.ServiceConfigDAO;
 import org.apache.ambari.server.orm.entities.ClusterConfigEntity;
 import org.apache.ambari.server.orm.entities.ClusterEntity;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import com.google.gson.Gson;
 import com.google.inject.Inject;
@@ -37,6 +39,11 @@ import com.google.inject.assistedinject.AssistedInject;
 import com.google.inject.persist.Transactional;
 
 public class ConfigImpl implements Config {
+  /**
+   * Logger.
+   */
+  private final static Logger LOG = LoggerFactory.getLogger(ConfigImpl.class);
+
   public static final String GENERATED_TAG_PREFIX = "generatedTag_";
 
   private Cluster cluster;
@@ -47,12 +54,16 @@ public class ConfigImpl implements Config {
   private Map<String, String> properties;
   private Map<String, Map<String, String>> propertiesAttributes;
   private ClusterConfigEntity entity;
+
   @Inject
   private ClusterDAO clusterDAO;
+
   @Inject
   private Gson gson;
+
   @Inject
   private ServiceConfigDAO serviceConfigDAO;
+
   @AssistedInject
   public ConfigImpl(@Assisted Cluster cluster, @Assisted String type, @Assisted Map<String, String> properties,
       @Assisted Map<String, Map<String, String>> propertiesAttributes, Injector injector) {
@@ -66,7 +77,6 @@ public class ConfigImpl implements Config {
     stackId = cluster.getDesiredStackVersion();
 
     injector.injectMembers(this);
-
   }
 
 
@@ -204,16 +214,20 @@ public class ConfigImpl implements Config {
       entity.setTimestamp(new Date().getTime());
       entity.setStack(clusterEntity.getDesiredStack());
       entity.setData(gson.toJson(getProperties()));
+
       if (null != getPropertiesAttributes()) {
         entity.setAttributes(gson.toJson(getPropertiesAttributes()));
       }
 
       clusterDAO.createConfig(entity);
-
       clusterEntity.getClusterConfigEntities().add(entity);
+      clusterDAO.merge(clusterEntity);
+      cluster.refresh();
     } else {
       // only supporting changes to the properties
       ClusterConfigEntity entity = null;
+
+      // find the existing configuration to update
       for (ClusterConfigEntity cfe : clusterEntity.getClusterConfigEntities()) {
         if (getTag().equals(cfe.getTag()) &&
             getType().equals(cfe.getType()) &&
@@ -221,15 +235,17 @@ public class ConfigImpl implements Config {
           entity = cfe;
           break;
         }
-
       }
 
+      // if the configuration was found, then update it
       if (null != entity) {
+        LOG.debug(
+            "Updating {} version {} with new configurations; a new version will not be created",
+            getType(), getVersion());
+
         entity.setData(gson.toJson(getProperties()));
+        clusterDAO.merge(clusterEntity);
       }
     }
-
-    clusterDAO.merge(clusterEntity);
-    cluster.refresh();
   }
 }

ambari-server/src/main/resources/common-services/STORM/0.9.1.2.1/package/scripts/params_linux.py

@@ -83,7 +83,8 @@ java64_home = config['hostLevelParams']['java_home']
 jps_binary = format("{java64_home}/bin/jps")
 nimbus_port = config['configurations']['storm-site']['nimbus.thrift.port']
 storm_zookeeper_root_dir = default('/configurations/storm-site/storm.zookeeper.root', None)
-storm_zookeeper_servers = default('/configurations/storm-site/storm.zookeeper.servers', None)
+storm_zookeeper_servers = config['configurations']['storm-site']['storm.zookeeper.servers']
+storm_zookeeper_port = config['configurations']['storm-site']['storm.zookeeper.port']
 
 # nimbus.seeds is supported in HDP 2.3.0.0 and higher
 nimbus_seeds_supported = default('/configurations/storm-env/nimbus_seeds_supported', False)

ambari-server/src/main/resources/common-services/STORM/0.9.1.2.1/package/scripts/storm.py

@@ -27,7 +27,7 @@ from resource_management.libraries.functions.format import format
 from resource_management.libraries.script.script import Script
 from resource_management.core.source import Template
 from resource_management.libraries.functions import compare_versions
-from yaml_utils import yaml_config_template, yaml_config
+from storm_yaml_utils import yaml_config_template, yaml_config
 from ambari_commons.os_family_impl import OsFamilyFuncImpl, OsFamilyImpl
 from ambari_commons import OSConst
 

ambari-server/src/main/resources/common-services/STORM/0.9.1.2.1/package/scripts/storm_upgrade.py

@@ -19,6 +19,7 @@ limitations under the License.
 import json
 import os
 
+from ambari_commons import yaml_utils
 from resource_management.core.logger import Logger
 from resource_management.core.exceptions import Fail
 from resource_management.core.resources.system import Directory
@@ -48,24 +49,49 @@ class StormUpgrade(Script):
     """
     import params
 
+
     Logger.info('Clearing Storm data from ZooKeeper')
 
     storm_zookeeper_root_dir = params.storm_zookeeper_root_dir
     if storm_zookeeper_root_dir is None:
       raise Fail("The storm ZooKeeper directory specified by storm-site/storm.zookeeper.root must be specified")
 
-    # create the ZooKeeper delete command
-    if params.version is not None:
-      command = "/usr/hdp/{0}/zookeeper/bin/zkCli.sh rmr /storm".format(params.version)
-    else:
-      command = "/usr/hdp/current/zookeeper-client/bin/zkCli.sh rmr /storm"
+    # the zookeeper client must be given a zookeeper host to contact
+    storm_zookeeper_server_list = yaml_utils.get_values_from_yaml_array(params.storm_zookeeper_servers)
+    if storm_zookeeper_server_list is None:
+      Logger.info("Unable to extract ZooKeeper hosts from '{0}', assuming localhost").format(params.storm_zookeeper_servers)
+      storm_zookeeper_server_list = ["localhost"]
 
+    # kinit if needed
     if params.security_enabled:
       kinit_command=format("{kinit_path_local} -kt {smoke_user_keytab} {smokeuser_principal}; ")
       Execute(kinit_command,user=params.smokeuser)
 
-    # clean out ZK
-    Execute(command, user=params.storm_user, tries=1)
+    # for every zk server, try to remove /storm
+    zookeeper_data_cleared = False
+    for storm_zookeeper_server in storm_zookeeper_server_list:
+      # determine where the zkCli.sh shell script is
+      zk_command_location = "/usr/hdp/current/zookeeper-client/bin/zkCli.sh"
+      if params.version is not None:
+        zk_command_location = "/usr/hdp/{0}/zookeeper/bin/zkCli.sh".format(params.version)
+
+      # create the ZooKeeper delete command
+      command = "{0} -server {1}:{2} rmr /storm".format(
+        zk_command_location, storm_zookeeper_server, params.storm_zookeeper_port)
+
+      # clean out ZK
+      try:
+        Execute(command, user=params.storm_user, logoutput=True, tries=1)
+        zookeeper_data_cleared = True
+        break
+      except:
+        # the command failed, try a different ZK server
+        pass
+
+    # fail if the ZK data could not be cleared
+    if not zookeeper_data_cleared:
+      raise Fail("Unable to clear ZooKeeper Storm data on any of the following ZooKeeper hosts: {0}".format(
+        storm_zookeeper_server_list))
 
 
   def delete_storm_local_data(self, env):

ambari-server/src/main/resources/common-services/STORM/0.9.1.2.1/package/scripts/yaml_utils.py → ambari-server/src/main/resources/common-services/STORM/0.9.1.2.1/package/scripts/storm_yaml_utils.py

@@ -17,41 +17,13 @@ See the License for the specific language governing permissions and
 limitations under the License.
 
 """
-
-import re
 import os
 import resource_management
+
+from ambari_commons.yaml_utils import escape_yaml_property
 from resource_management.core.source import InlineTemplate
 from resource_management.core.resources.system import File
 
-def escape_yaml_property(value):
-  unquouted = False
-  unquouted_values = ["null","Null","NULL","true","True","TRUE","false","False","FALSE","YES","Yes","yes","NO","No","no","ON","On","on","OFF","Off","off"]
-  if value in unquouted_values:
-    unquouted = True
-
-  # if is list [a,b,c] or dictionary {a: v, b: v2, c: v3}
-  if re.match('^\w*\[.+\]\w*$', value) or re.match('^\w*\{.+\}\w*$', value):
-    unquouted = True
-
-  try:
-    int(value)
-    unquouted = True
-  except ValueError:
-    pass
-  
-  try:
-    float(value)
-    unquouted = True
-  except ValueError:
-    pass
-  
-  if not unquouted:
-    value = value.replace("'","''")
-    value = "'"+value+"'"
-    
-  return value
-
 def replace_jaas_placeholder(name, security_enabled, conf_dir):
   if name.find('_JAAS_PLACEHOLDER') > -1:
     if security_enabled:
@@ -78,4 +50,4 @@ def yaml_config(filename, configurations = None, conf_dir = None, owner = None,
         content = config_content,
         owner = owner,
         mode = "f"
-  )
+  )

ambari-server/src/test/java/org/apache/ambari/server/state/cluster/ClusterTest.java

@@ -42,8 +42,6 @@ import java.util.Set;
 import javax.persistence.EntityManager;
 import javax.persistence.RollbackException;
 
-import com.google.gson.Gson;
-import com.google.inject.persist.UnitOfWork;
 import junit.framework.Assert;
 
 import org.apache.ambari.server.AmbariException;
@@ -65,6 +63,7 @@ import org.apache.ambari.server.orm.dao.HostVersionDAO;
 import org.apache.ambari.server.orm.dao.RepositoryVersionDAO;
 import org.apache.ambari.server.orm.dao.ResourceTypeDAO;
 import org.apache.ambari.server.orm.dao.StackDAO;
+import org.apache.ambari.server.orm.entities.ClusterConfigEntity;
 import org.apache.ambari.server.orm.entities.ClusterEntity;
 import org.apache.ambari.server.orm.entities.ClusterServiceEntity;
 import org.apache.ambari.server.orm.entities.ClusterStateEntity;
@@ -108,12 +107,14 @@ import org.junit.Before;
 import org.junit.Test;
 import org.mockito.ArgumentCaptor;
 
+import com.google.gson.Gson;
 import com.google.inject.AbstractModule;
 import com.google.inject.Guice;
 import com.google.inject.Injector;
 import com.google.inject.Singleton;
 import com.google.inject.persist.PersistService;
 import com.google.inject.persist.Transactional;
+import com.google.inject.persist.UnitOfWork;
 import com.google.inject.util.Modules;
 
 public class ClusterTest {
@@ -1905,4 +1906,60 @@ public class ClusterTest {
     verify(hostVersionDAOMock).merge(hostVersionCaptor.capture());
     assertEquals(hostVersionCaptor.getValue().getState(), RepositoryVersionState.CURRENT);
   }
+
+  /**
+   * Tests that an existing configuration can be successfully updated without
+   * creating a new version.
+   *
+   * @throws Exception
+   */
+  @Test
+  public void testClusterConfigMergingWithoutNewVersion() throws Exception {
+    createDefaultCluster();
+
+    Cluster cluster = clusters.getCluster("c1");
+    ClusterEntity clusterEntity = clusterDAO.findByName("c1");
+    assertEquals(0, clusterEntity.getClusterConfigEntities().size());
+
+    final Config originalConfig = configFactory.createNew(cluster, "foo-site",
+        new HashMap<String, String>() {
+          {
+            put("one", "two");
+          }
+        }, new HashMap<String, Map<String, String>>());
+
+    originalConfig.setTag("version3");
+    originalConfig.persist();
+    cluster.addConfig(originalConfig);
+
+    ConfigGroup configGroup = configGroupFactory.createNew(cluster, "g1", "t1", "",
+        new HashMap<String, Config>() {
+          {
+            put("foo-site", originalConfig);
+          }
+        }, Collections.<Long, Host> emptyMap());
+
+    configGroup.persist();
+    cluster.addConfigGroup(configGroup);
+
+    clusterEntity = clusterDAO.findByName("c1");
+    assertEquals(1, clusterEntity.getClusterConfigEntities().size());
+
+    Map<String, Config> configsByType = cluster.getConfigsByType("foo-site");
+    Config config = configsByType.entrySet().iterator().next().getValue();
+
+    Map<String, String> properties = config.getProperties();
+    properties.put("three", "four");
+    config.setProperties(properties);
+
+    config.persist(false);
+
+    clusterEntity = clusterDAO.findByName("c1");
+    assertEquals(1, clusterEntity.getClusterConfigEntities().size());
+    ClusterConfigEntity clusterConfigEntity = clusterEntity.getClusterConfigEntities().iterator().next();
+    assertTrue(clusterConfigEntity.getData().contains("one"));
+    assertTrue(clusterConfigEntity.getData().contains("two"));
+    assertTrue(clusterConfigEntity.getData().contains("three"));
+    assertTrue(clusterConfigEntity.getData().contains("four"));
+  }
 }

ambari-server/src/test/python/TestYAMLUtils.py

@@ -0,0 +1,44 @@
+#!/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 unittest import TestCase
+
+from ambari_commons import yaml_utils
+
+class TestYAMLUtils(TestCase):
+  def setUp(self):
+    pass
+
+  def test_convert_yaml_array(self):
+    expected_values = []
+
+    expected_values.append("c6401.ambari.apache.org")
+    values = yaml_utils.get_values_from_yaml_array("['c6401.ambari.apache.org']")
+    self.assertEquals(expected_values, values)
+
+    expected_values.append("c6402.ambari.apache.org")
+    values = yaml_utils.get_values_from_yaml_array("['c6401.ambari.apache.org', 'c6402.ambari.apache.org']")
+    self.assertEquals(expected_values, values)
+
+    values = yaml_utils.get_values_from_yaml_array('["c6401.ambari.apache.org", "c6402.ambari.apache.org"]')
+    self.assertEquals(expected_values, values)
+
+    values = yaml_utils.get_values_from_yaml_array('[\'c6401.ambari.apache.org\', "c6402.ambari.apache.org"]')
+    self.assertEquals(expected_values, values)

ambari-server/src/test/python/stacks/2.1/STORM/test_storm_base.py

@@ -113,10 +113,10 @@ class TestStormBase(RMFTestCase):
     return storm_yarn_content
 
   def call_storm_template_and_assert(self, confDir="/etc/storm/conf"):
-    import yaml_utils
+    import storm_yaml_utils
 
     with RMFTestCase.env as env:
-      storm_yarn_temlate = yaml_utils.yaml_config_template(self.getConfig()['configurations']['storm-site'])
+      storm_yarn_temlate = storm_yaml_utils.yaml_config_template(self.getConfig()['configurations']['storm-site'])
 
       self.assertResourceCalled('File', confDir + '/storm.yaml',
         owner = 'storm',

ambari-server/src/test/python/stacks/2.3/STORM/test_storm_base.py

@@ -110,10 +110,10 @@ class TestStormBase(RMFTestCase):
     return storm_yarn_content
 
   def call_storm_template_and_assert(self, confDir="/etc/storm/conf"):
-    import yaml_utils
+    import storm_yaml_utils
 
     with RMFTestCase.env as env:
-      storm_yarn_temlate = yaml_utils.yaml_config_template(self.getConfig()['configurations']['storm-site'])
+      storm_yarn_temlate = storm_yaml_utils.yaml_config_template(self.getConfig()['configurations']['storm-site'])
 
       self.assertResourceCalled('File', confDir + '/storm.yaml',
         owner = 'storm',