KAFKA-2641: Upgrade path for ZK authentication #313
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,17 @@ | ||
| #!/bin/bash | ||
| # 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. | ||
|
|
||
| exec $(dirname $0)/kafka-run-class.sh kafka.admin.ZkSecurityMigrator $@ |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,240 @@ | ||
| /** | ||
| * 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 kafka.admin | ||
|
|
||
|
|
||
| import java.util.concurrent.LinkedBlockingQueue | ||
| import java.util.concurrent.ThreadPoolExecutor | ||
| import java.util.concurrent.TimeUnit | ||
| import joptsimple.OptionParser | ||
| import org.I0Itec.zkclient.exception.ZkException | ||
| import kafka.utils.{Logging, ZkUtils, CommandLineUtils} | ||
| import org.apache.log4j.Level | ||
| import org.apache.kafka.common.security.JaasUtils | ||
| import org.apache.zookeeper.AsyncCallback.{ChildrenCallback, StatCallback} | ||
| import org.apache.zookeeper.data.Stat | ||
| import org.apache.zookeeper.KeeperException | ||
| import org.apache.zookeeper.KeeperException.Code | ||
| import scala.annotation.tailrec | ||
| import scala.collection.JavaConverters._ | ||
| import scala.collection._ | ||
| import scala.collection.mutable.Queue | ||
| import scala.concurrent._ | ||
| import scala.concurrent.duration._ | ||
|
|
||
| /** | ||
| * This tool is to be used when making access to ZooKeeper authenticated or | ||
| * the other way around, when removing authenticated access. The exact steps | ||
| * to migrate a Kafka cluster from unsecure to secure with respect to ZooKeeper | ||
| * access are the following: | ||
| * | ||
| * 1- Perform a rolling upgrade of Kafka servers, setting zookeeper.set.acl to false | ||
| * and passing a valid JAAS login file via the system property | ||
| * java.security.auth.login.config | ||
| * 2- Perform a second rolling upgrade keeping the system property for the login file | ||
| * and now setting zookeeper.set.acl to true | ||
| * 3- Finally run this tool. There is a script under ./bin. Run | ||
| * ./bin/zookeeper-security-migration --help | ||
| * to see the configuration parameters. An example of running it is the following: | ||
| * ./bin/zookeeper-security-migration --zookeeper.acl=secure --zookeeper.connection=localhost:2181 | ||
| * | ||
| * To convert a cluster from secure to unsecure, we need to perform the following | ||
| * steps: | ||
| * 1- Perform a rolling upgrade setting zookeeper.set.acl to false for each server | ||
| * 2- Run this migration tool, setting zookeeper.acl to unsecure | ||
| * 3- Perform another rolling upgrade to remove the system property setting the | ||
| * login file (java.security.auth.login.config). | ||
| */ | ||
|
|
||
| object ZkSecurityMigrator extends Logging { | ||
| val usageMessage = ("ZooKeeper Migration Tool Help. This tool updates the ACLs of " | ||
| + "znodes as part of the process of setting up ZooKeeper " | ||
| + "authentication.") | ||
|
|
||
| def run(args: Array[String]) { | ||
| var jaasFile = System.getProperty(JaasUtils.JAVA_LOGIN_CONFIG_PARAM) | ||
| val parser = new OptionParser() | ||
|
|
||
| val goOpt = parser.accepts("zookeeper.acl", "Indicates whether to make the Kafka znodes in ZooKeeper secure or unsecure." | ||
| + " The options are 'secure' and 'unsecure'").withRequiredArg().ofType(classOf[String]) | ||
| val jaasFileOpt = parser.accepts("jaas.file", "JAAS Config file.").withOptionalArg().ofType(classOf[String]) | ||
| val zkUrlOpt = parser.accepts("zookeeper.connect", "Sets the ZooKeeper connect string (ensemble). This parameter " + | ||
| "takes a comma-separated list of host:port pairs.").withRequiredArg().defaultsTo("localhost:2181"). | ||
| ofType(classOf[String]) | ||
| val zkSessionTimeoutOpt = parser.accepts("zookeeper.session.timeout", "Sets the ZooKeeper session timeout."). | ||
| withRequiredArg().ofType(classOf[java.lang.Integer]).defaultsTo(30000) | ||
| val zkConnectionTimeoutOpt = parser.accepts("zookeeper.connection.timeout", "Sets the ZooKeeper connection timeout."). | ||
| withRequiredArg().ofType(classOf[java.lang.Integer]).defaultsTo(30000) | ||
| val helpOpt = parser.accepts("help", "Print usage information.") | ||
|
Contributor
There was a problem hiding this comment. To support going from secured ZK to unsecured ZK, we will need an option to enable/disable the CREATOR_ALL acl. |
||
|
|
||
| val options = parser.parse(args : _*) | ||
| if (options.has(helpOpt)) | ||
| CommandLineUtils.printUsageAndDie(parser, usageMessage) | ||
|
|
||
| if ((jaasFile == null) && !options.has(jaasFileOpt)) { | ||
| val errorMsg = ("No JAAS configuration file has been specified. Please make sure that you have set either " + | ||
| "the system property %s or the option %s".format(JaasUtils.JAVA_LOGIN_CONFIG_PARAM, "--jaas.file")) | ||
| error(errorMsg) | ||
|
Contributor
There was a problem hiding this comment. If we print to stdout, there is probably no need to log the same message to log4j. Otherwise, if log4j is enabled, we will get duplicated output.
Contributor
Author
There was a problem hiding this comment. We are also running this code on tests, so I'd rather keep the log4j output. I had an if statement to avoid redundant messages, but I removed after a comment I got here. If you want, I can bring it back so that if print to stderr in the case log4j is disabled..
Contributor
There was a problem hiding this comment. If this is just for testing, perhaps we can log in debug mode?
Member
There was a problem hiding this comment. @fpj, the code you had was: This is the same as
Contributor
Author
There was a problem hiding this comment. This isn't only for testing, the user needs to be informed of the configuration error. What I'm saying is that all error messages should be printed via log4j because we are exercising this code in junit tests. You suggested that users might not have log4j on so we need to have messages printed to stdout as well. |
||
| System.err.println("ERROR: %s".format(errorMsg)) | ||
| throw new IllegalArgumentException("Incorrect configuration") | ||
| } | ||
|
|
||
| if (jaasFile == null) { | ||
| jaasFile = options.valueOf(jaasFileOpt) | ||
| System.setProperty(JaasUtils.JAVA_LOGIN_CONFIG_PARAM, jaasFile) | ||
| } | ||
|
|
||
| if (!JaasUtils.isZkSecurityEnabled(jaasFile)) { | ||
| val errorMsg = "Security isn't enabled, most likely the file isn't set properly: %s".format(jaasFile) | ||
| error(errorMsg) | ||
| System.err.println("ERROR: %s".format(errorMsg)) | ||
| throw new IllegalArgumentException("Incorrect configuration") | ||
| } | ||
|
|
||
| val goSecure: Boolean = options.valueOf(goOpt) match { | ||
|
Contributor
There was a problem hiding this comment. Could we rename goSecure to sth like zkAclMode? |
||
| case "secure" => | ||
| info("Making it secure") | ||
|
Contributor
There was a problem hiding this comment. Do we want the user to see this? If so, we should print it instead of logging it.
Contributor
Author
There was a problem hiding this comment. Same argument as above, we are also exercising this code in tests, so I'd rather keep the log4j output. As this isn't strictly necessary, I don't think we need to have it on stdout. |
||
| true | ||
| case "unsecure" => | ||
| info("Making it unsecure") | ||
| false | ||
| case _ => | ||
| CommandLineUtils.printUsageAndDie(parser, usageMessage) | ||
| } | ||
| val zkUrl = options.valueOf(zkUrlOpt) | ||
| val zkSessionTimeout = options.valueOf(zkSessionTimeoutOpt).intValue | ||
| val zkConnectionTimeout = options.valueOf(zkConnectionTimeoutOpt).intValue | ||
| val zkUtils = ZkUtils(zkUrl, zkSessionTimeout, zkConnectionTimeout, goSecure) | ||
| val migrator = new ZkSecurityMigrator(zkUtils) | ||
| migrator.run() | ||
| } | ||
|
|
||
| def main(args: Array[String]) { | ||
| try { | ||
| run(args) | ||
| } catch { | ||
| case e: Exception => | ||
| e.printStackTrace() | ||
| } | ||
| } | ||
| } | ||
|
|
||
| class ZkSecurityMigrator(zkUtils: ZkUtils) extends Logging { | ||
| private val workQueue = new LinkedBlockingQueue[Runnable] | ||
| private val futures = new Queue[Future[String]] | ||
|
|
||
| private def setAclsRecursively(path: String) = { | ||
| info("Setting ACL for path %s".format(path)) | ||
| val setPromise = Promise[String] | ||
| val childrenPromise = Promise[String] | ||
| futures.synchronized { | ||
| futures += setPromise.future | ||
| futures += childrenPromise.future | ||
| } | ||
| zkUtils.zkConnection.getZookeeper.setACL(path, ZkUtils.DefaultAcls(zkUtils.isSecure), -1, SetACLCallback, setPromise) | ||
| zkUtils.zkConnection.getZookeeper.getChildren(path, false, GetChildrenCallback, childrenPromise) | ||
| } | ||
|
|
||
| private object GetChildrenCallback extends ChildrenCallback { | ||
| def processResult(rc: Int, | ||
| path: String, | ||
| ctx: Object, | ||
| children: java.util.List[String]) { | ||
| val zkHandle = zkUtils.zkConnection.getZookeeper | ||
| val promise = ctx.asInstanceOf[Promise[String]] | ||
| Code.get(rc) match { | ||
| case Code.OK => | ||
| // Set ACL for each child | ||
| for (child <- children.asScala) | ||
| setAclsRecursively(s"$path/$child") | ||
| promise success "done" | ||
| case Code.CONNECTIONLOSS => | ||
| zkHandle.getChildren(path, false, GetChildrenCallback, ctx) | ||
| case Code.NONODE => | ||
| warn("Node is gone, it could be have been legitimately deleted: %s".format(path)) | ||
| promise success "done" | ||
| case Code.SESSIONEXPIRED => | ||
| // Starting a new session isn't really a problem, but it'd complicate | ||
| // the logic of the tool, so we quit and let the user re-run it. | ||
| error("ZooKeeper session expired while changing ACLs") | ||
|
Contributor
There was a problem hiding this comment. It seems that we need to print this instead of logging it. |
||
| promise failure ZkException.create(KeeperException.create(Code.get(rc))) | ||
| case _ => | ||
| error("Unexpected return code: %d".format(rc)) | ||
|
Contributor
There was a problem hiding this comment. It seems that we need to print this instead of logging it. |
||
| promise failure ZkException.create(KeeperException.create(Code.get(rc))) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| private object SetACLCallback extends StatCallback { | ||
| def processResult(rc: Int, | ||
| path: String, | ||
| ctx: Object, | ||
| stat: Stat) { | ||
| val zkHandle = zkUtils.zkConnection.getZookeeper | ||
| val promise = ctx.asInstanceOf[Promise[String]] | ||
|
|
||
| Code.get(rc) match { | ||
| case Code.OK => | ||
| info("Successfully set ACLs for %s".format(path)) | ||
|
Contributor
There was a problem hiding this comment. It seems that we need to print this instead of logging it. |
||
| promise success "done" | ||
| case Code.CONNECTIONLOSS => | ||
| zkHandle.setACL(path, ZkUtils.DefaultAcls(zkUtils.isSecure), -1, SetACLCallback, ctx) | ||
| case Code.NONODE => | ||
| warn("Node is gone, it could be have been legitimately deleted: %s".format(path)) | ||
| promise success "done" | ||
| case Code.SESSIONEXPIRED => | ||
| // Starting a new session isn't really a problem, but it'd complicate | ||
| // the logic of the tool, so we quit and let the user re-run it. | ||
| error("ZooKeeper session expired while changing ACLs") | ||
|
Contributor
There was a problem hiding this comment. It seems that we need to print this instead of logging it. |
||
| promise failure ZkException.create(KeeperException.create(Code.get(rc))) | ||
| case _ => | ||
| error("Unexpected return code: %d".format(rc)) | ||
|
Contributor
There was a problem hiding this comment. It seems that we need to print this instead of logging it. |
||
| promise failure ZkException.create(KeeperException.create(Code.get(rc))) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| private def run(): Unit = { | ||
| try { | ||
| for (path <- zkUtils.securePersistentZkPaths) { | ||
| info("Securing " + path) | ||
|
Contributor
There was a problem hiding this comment. We could be either securing or unsecuring the path. Also, is this sth to show the user or for debugging?
Contributor
Author
There was a problem hiding this comment. I've made it a debug call and removed the securing comment to say that we are setting the ACL for this path. |
||
| zkUtils.makeSurePersistentPathExists(path) | ||
| setAclsRecursively(path) | ||
| } | ||
|
|
||
| @tailrec | ||
| def recurse(): Unit = { | ||
| val future = futures.synchronized { | ||
| info("Size of future list is %d".format(futures.size)) | ||
|
Contributor
There was a problem hiding this comment. Perhaps this should be debug level.
Contributor
Author
There was a problem hiding this comment. I've removed this altogether. It was there for debugging indeed. |
||
| futures.headOption | ||
| } | ||
| future match { | ||
| case Some(a) => | ||
| Await.result(a, 6000 millis) | ||
| futures.synchronized { futures.dequeue } | ||
| recurse | ||
| case None => | ||
| } | ||
| } | ||
| recurse() | ||
|
|
||
| } finally { | ||
| zkUtils.close | ||
| } | ||
| } | ||
|
|
||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
As we were discussing with @junrao, I think we need to either find a way not to need this or to at least do it in a safe way within our app (ie reload it atomically).
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
yeah, I'm not comfortable with this at all, it doesn't sound robust. at the same time, I want to separate bugs in my code from bugs elsewhere, so I'm leaving this here for now, I need to understand what's causing this.