You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ambari.apache.org by nc...@apache.org on 2015/11/18 17:51:21 UTC
[25/50] [abbrv] ambari git commit: AMBARI-13918. Stack Upgrade: Kafka
to call migration script when upgrading across HDP 2.3.4.0 (alejandro)
AMBARI-13918. Stack Upgrade: Kafka to call migration script when upgrading across HDP 2.3.4.0 (alejandro)
Project: http://git-wip-us.apache.org/repos/asf/ambari/repo
Commit: http://git-wip-us.apache.org/repos/asf/ambari/commit/6212d9d6
Tree: http://git-wip-us.apache.org/repos/asf/ambari/tree/6212d9d6
Diff: http://git-wip-us.apache.org/repos/asf/ambari/diff/6212d9d6
Branch: refs/heads/branch-dev-patch-upgrade
Commit: 6212d9d657fbd832e7cde4f3a1734397c33064f3
Parents: 09c9134
Author: Alejandro Fernandez <af...@hortonworks.com>
Authored: Fri Nov 13 14:28:08 2015 -0800
Committer: Alejandro Fernandez <af...@hortonworks.com>
Committed: Tue Nov 17 13:02:17 2015 -0800
----------------------------------------------------------------------
.../0.8.1.2.2/package/scripts/kafka_broker.py | 35 ++++++++++--
.../KAFKA/0.8.1.2.2/package/scripts/params.py | 25 +++++++-
.../KAFKA/0.8.1.2.2/package/scripts/upgrade.py | 60 +++++++++++++++++---
.../stacks/HDP/2.2/upgrades/upgrade-2.3.xml | 9 +++
.../stacks/HDP/2.3/upgrades/upgrade-2.3.xml | 9 +++
5 files changed, 124 insertions(+), 14 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ambari/blob/6212d9d6/ambari-server/src/main/resources/common-services/KAFKA/0.8.1.2.2/package/scripts/kafka_broker.py
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/resources/common-services/KAFKA/0.8.1.2.2/package/scripts/kafka_broker.py b/ambari-server/src/main/resources/common-services/KAFKA/0.8.1.2.2/package/scripts/kafka_broker.py
index 8efd564..3caeb6d 100644
--- a/ambari-server/src/main/resources/common-services/KAFKA/0.8.1.2.2/package/scripts/kafka_broker.py
+++ b/ambari-server/src/main/resources/common-services/KAFKA/0.8.1.2.2/package/scripts/kafka_broker.py
@@ -16,11 +16,17 @@ See the License for the specific language governing permissions and
limitations under the License.
"""
+from resource_management import Script
+from resource_management.core.logger import Logger
+from resource_management.core.resources.system import Execute, File, Directory
+from resource_management.libraries.functions import conf_select
+from resource_management.libraries.functions import hdp_select
+from resource_management.libraries.functions import Direction
+from resource_management.libraries.functions.version import compare_versions, format_hdp_stack_version
+from resource_management.libraries.functions.format import format
+from resource_management.libraries.functions.check_process_status import check_process_status
-from resource_management import *
-import sys
import upgrade
-
from kafka import kafka
from setup_ranger_kafka import setup_ranger_kafka
@@ -40,7 +46,26 @@ class KafkaBroker(Script):
def pre_upgrade_restart(self, env, upgrade_type=None):
import params
env.set_params(params)
- upgrade.prestart(env, "kafka-broker")
+
+ if params.version and compare_versions(format_hdp_stack_version(params.version), '2.2.0.0') >= 0:
+ conf_select.select(params.stack_name, "kafka", params.version)
+ hdp_select.select("kafka-broker", params.version)
+
+ # This is extremely important since it should only be called if crossing the HDP 2.3.4.0 boundary.
+ if params.current_version and params.version and params.upgrade_direction:
+ src_version = dst_version = None
+ if params.upgrade_direction == Direction.UPGRADE:
+ src_version = format_hdp_stack_version(params.current_version)
+ dst_version = format_hdp_stack_version(params.version)
+ else:
+ # These represent the original values during the UPGRADE direction
+ src_version = format_hdp_stack_version(params.version)
+ dst_version = format_hdp_stack_version(params.downgrade_from_version)
+
+ if compare_versions(src_version, '2.3.4.0') < 0 and compare_versions(dst_version, '2.3.4.0') >= 0:
+ # Calling the acl migration script requires the configs to be present.
+ self.configure(env)
+ upgrade.run_migration(env, upgrade_type)
def start(self, env, upgrade_type=None):
import params
@@ -62,7 +87,7 @@ class KafkaBroker(Script):
Execute(daemon_cmd,
user=params.kafka_user,
)
- File (params.kafka_pid_file,
+ File(params.kafka_pid_file,
action = "delete"
)
http://git-wip-us.apache.org/repos/asf/ambari/blob/6212d9d6/ambari-server/src/main/resources/common-services/KAFKA/0.8.1.2.2/package/scripts/params.py
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/resources/common-services/KAFKA/0.8.1.2.2/package/scripts/params.py b/ambari-server/src/main/resources/common-services/KAFKA/0.8.1.2.2/package/scripts/params.py
index dc3a61b..04490fc 100644
--- a/ambari-server/src/main/resources/common-services/KAFKA/0.8.1.2.2/package/scripts/params.py
+++ b/ambari-server/src/main/resources/common-services/KAFKA/0.8.1.2.2/package/scripts/params.py
@@ -26,6 +26,7 @@ from resource_management.libraries.functions.get_hdp_version import get_hdp_vers
from resource_management.libraries.functions.is_empty import is_empty
import status_params
+from resource_management.core.logger import Logger
# server configurations
@@ -34,11 +35,23 @@ tmp_dir = Script.get_tmp_dir()
stack_name = default("/hostLevelParams/stack_name", None)
retryAble = default("/commandParams/command_retry_enabled", False)
+# Version being upgraded/downgraded to
version = default("/commandParams/version", None)
+
+# Version that is CURRENT.
+current_version = default("/hostLevelParams/current_version", None)
+
host_sys_prepped = default("/hostLevelParams/host_sys_prepped", False)
stack_version_unformatted = str(config['hostLevelParams']['stack_version'])
hdp_stack_version = format_hdp_stack_version(stack_version_unformatted)
+upgrade_direction = default("/commandParams/upgrade_direction", None)
+
+# When downgrading the 'version' and 'current_version' are both pointing to the downgrade-target version
+# downgrade_from_version provides the source-version the downgrade is happening from
+downgrade_from_version = default("/commandParams/downgrade_from_version", None)
+
+hostname = config['hostname']
# default kafka parameters
kafka_home = '/usr/lib/kafka/'
@@ -46,6 +59,17 @@ kafka_bin = kafka_home+'/bin/kafka'
conf_dir = "/etc/kafka/conf"
limits_conf_dir = "/etc/security/limits.d"
+# Find the port number of the Zookeeper Server running on this host.
+zookeeper_port = None
+zookeeper_connect = default("/configurations/kafka-broker/zookeeper.connect", None)
+zookeeper_list = [e.strip() for e in zookeeper_connect.split(",")] if zookeeper_connect else []
+if len(zookeeper_list) > 0:
+ for e in zookeeper_list:
+ if e.find(":") > 0 and len(e.split(":")) == 2 and e.split(":")[0].strip() == hostname:
+ zookeeper_port = e.split(":")[1].strip()
+ break
+
+
kafka_user_nofile_limit = config['configurations']['kafka-env']['kafka_user_nofile_limit']
kafka_user_nproc_limit = config['configurations']['kafka-env']['kafka_user_nproc_limit']
@@ -63,7 +87,6 @@ kafka_pid_file = kafka_pid_dir+"/kafka.pid"
# This is hardcoded on the kafka bash process lifecycle on which we have no control over
kafka_managed_pid_dir = "/var/run/kafka"
kafka_managed_log_dir = "/var/log/kafka"
-hostname = config['hostname']
user_group = config['configurations']['cluster-env']['user_group']
java64_home = config['hostLevelParams']['java_home']
kafka_env_sh_template = config['configurations']['kafka-env']['content']
http://git-wip-us.apache.org/repos/asf/ambari/blob/6212d9d6/ambari-server/src/main/resources/common-services/KAFKA/0.8.1.2.2/package/scripts/upgrade.py
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/resources/common-services/KAFKA/0.8.1.2.2/package/scripts/upgrade.py b/ambari-server/src/main/resources/common-services/KAFKA/0.8.1.2.2/package/scripts/upgrade.py
index 22036f7..0457525 100644
--- a/ambari-server/src/main/resources/common-services/KAFKA/0.8.1.2.2/package/scripts/upgrade.py
+++ b/ambari-server/src/main/resources/common-services/KAFKA/0.8.1.2.2/package/scripts/upgrade.py
@@ -18,15 +18,59 @@ See the License for the specific language governing permissions and
limitations under the License.
"""
-from resource_management import *
+import os
+
+
from resource_management.core.resources.system import Execute
-from resource_management.libraries.functions import conf_select
-from resource_management.libraries.functions import hdp_select
-from resource_management.libraries.functions.version import compare_versions, format_hdp_stack_version
+from resource_management.libraries.functions import format
+from resource_management.libraries.functions import Direction
+from resource_management.core.exceptions import Fail
+from resource_management.core.logger import Logger
-def prestart(env, hdp_component):
+def run_migration(env, upgrade_type):
+ """
+ If the acl migration script is present, then run it for either upgrade or downgrade.
+ That script was introduced in HDP 2.3.4.0 and requires stopping all Kafka brokers first.
+ Requires configs to be present.
+ :param env: Environment.
+ :param upgrade_type: "rolling" or "nonrolling
+ """
import params
- if params.version and compare_versions(format_hdp_stack_version(params.version), '2.2.0.0') >= 0:
- conf_select.select(params.stack_name, "kafka", params.version)
- hdp_select.select(hdp_component, params.version)
+ if upgrade_type is None:
+ raise Fail('Parameter "upgrade_type" is missing.')
+
+ if params.upgrade_direction is None:
+ raise Fail('Parameter "upgrade_direction" is missing.')
+
+ if params.upgrade_direction == Direction.DOWNGRADE and params.downgrade_from_version is None:
+ raise Fail('Parameter "downgrade_from_version" is missing.')
+
+ Logger.info("Upgrade type: {0}, direction: {1}".format(str(upgrade_type), params.upgrade_direction))
+
+ # If the schema upgrade script exists in the version upgrading to, then attempt to upgrade/downgrade it while still using the present bits.
+ kafka_acls_script = None
+ command_suffix = ""
+ if params.upgrade_direction == Direction.UPGRADE:
+ kafka_acls_script = format("/usr/hdp/{version}/kafka/bin/kafka-acls.sh")
+ command_suffix = "--upgradeAcls"
+ elif params.upgrade_direction == Direction.DOWNGRADE:
+ kafka_acls_script = format("/usr/hdp/{downgrade_from_version}/kafka/bin/kafka-acls.sh")
+ command_suffix = "--downgradeAcls"
+
+ if kafka_acls_script is not None:
+ if os.path.exists(kafka_acls_script):
+ Logger.info("Found Kafka acls script: {0}".format(kafka_acls_script))
+ if params.zookeeper_port is None:
+ raise Fail("Could not retrieve Zookeeper Port number for localhost from property kafka-broker/zookeeper.connect")
+
+ zookeeper_localhost_port = "localhost:" + params.zookeeper_port
+
+ acls_command = "{0} --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect={1} {2}".\
+ format(kafka_acls_script, zookeeper_localhost_port, command_suffix)
+
+ Execute(acls_command,
+ user=params.kafka_user,
+ logoutput=True)
+ else:
+ Logger.info("Did not find Kafka acls script: {0}".format(kafka_acls_script))
http://git-wip-us.apache.org/repos/asf/ambari/blob/6212d9d6/ambari-server/src/main/resources/stacks/HDP/2.2/upgrades/upgrade-2.3.xml
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/resources/stacks/HDP/2.2/upgrades/upgrade-2.3.xml b/ambari-server/src/main/resources/stacks/HDP/2.2/upgrades/upgrade-2.3.xml
index fe87ed2..6a85479 100644
--- a/ambari-server/src/main/resources/stacks/HDP/2.2/upgrades/upgrade-2.3.xml
+++ b/ambari-server/src/main/resources/stacks/HDP/2.2/upgrades/upgrade-2.3.xml
@@ -784,6 +784,15 @@
<service name="KAFKA">
<component name="KAFKA_BROKER">
+ <!-- In HDP 2.3.4.0, Kafka had changes that required stopping all Kafka Brokers, running a migration script,
+ and then starting one at a time. This is needed during both upgrade and downgrade. -->
+ <pre-upgrade>
+ <task xsi:type="execute" hosts="all" summary="Shut down all Kafka Brokers">
+ <script>scripts/kafka_broker.py</script>
+ <function>stop</function>
+ </task>
+ </pre-upgrade>
+
<upgrade>
<task xsi:type="restart-task"/>
</upgrade>
http://git-wip-us.apache.org/repos/asf/ambari/blob/6212d9d6/ambari-server/src/main/resources/stacks/HDP/2.3/upgrades/upgrade-2.3.xml
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/resources/stacks/HDP/2.3/upgrades/upgrade-2.3.xml b/ambari-server/src/main/resources/stacks/HDP/2.3/upgrades/upgrade-2.3.xml
index 90f50a4..1d3ee79 100644
--- a/ambari-server/src/main/resources/stacks/HDP/2.3/upgrades/upgrade-2.3.xml
+++ b/ambari-server/src/main/resources/stacks/HDP/2.3/upgrades/upgrade-2.3.xml
@@ -713,6 +713,15 @@
<service name="KAFKA">
<component name="KAFKA_BROKER">
+ <!-- In HDP 2.3.4.0, Kafka had changes that required stopping all Kafka Brokers, running a migration script,
+ and then starting one at a time. This is needed during both upgrade and downgrade. -->
+ <pre-upgrade>
+ <task xsi:type="execute" hosts="all" summary="Shut down all Kafka Brokers">
+ <script>scripts/kafka_broker.py</script>
+ <function>stop</function>
+ </task>
+ </pre-upgrade>
+
<upgrade>
<task xsi:type="restart-task"/>
</upgrade>