From e1dfac9deddd8423675f2059012fa53befbcf8f0 Mon Sep 17 00:00:00 2001 From: Jarek Jarcec Cecho Date: Fri, 10 Nov 2017 13:41:47 -0800 Subject: [PATCH 1/3] Persist both zookeeper and kafka logs --- images/broker/start_kafka | 2 +- images/broker/start_zookeeper | 2 +- start.py | 4 ++-- 3 files changed, 4 insertions(+), 4 deletions(-) diff --git a/images/broker/start_kafka b/images/broker/start_kafka index 3a7c916..1430019 100644 --- a/images/broker/start_kafka +++ b/images/broker/start_kafka @@ -11,4 +11,4 @@ # See the License for the specific language governing permissions and # limitations under the License. -/kafka/bin/kafka-server-start.sh /kafka/config/server.properties & +/kafka/bin/kafka-server-start.sh /kafka/config/server.properties > /logs/kafka.log & diff --git a/images/broker/start_zookeeper b/images/broker/start_zookeeper index 39e67e6..e77756d 100644 --- a/images/broker/start_zookeeper +++ b/images/broker/start_zookeeper @@ -11,4 +11,4 @@ # See the License for the specific language governing permissions and # limitations under the License. -/kafka/bin/zookeeper-server-start.sh /kafka/config/zookeeper.properties & +/kafka/bin/zookeeper-server-start.sh /kafka/config/zookeeper.properties > /logs/zookeeper.log & diff --git a/start.py b/start.py index 4d4fd7a..68b34c5 100644 --- a/start.py +++ b/start.py @@ -45,5 +45,5 @@ def main(args): # start each node independently so they will all end up independent one-node clusters. for node in cluster: - node.execute('/kafka/bin/zookeeper-server-start.sh /kafka/config/zookeeper.properties &', detach=True) - node.execute('/kafka/bin/kafka-server-start.sh /kafka/config/server.properties &', detach=True) + node.execute('/start_zookeeper &', detach=True) + node.execute('/start_kafka &', detach=True) From ea63e0343e00d813c729fb7d25f497fc16a1e981 Mon Sep 17 00:00:00 2001 From: Jarek Jarcec Cecho Date: Fri, 10 Nov 2017 16:22:50 -0800 Subject: [PATCH 2/3] Add full cluster mode support This patch adds full cluster mode support, e.g. if more then one node is specified, then all brokers are part of one logical cluster rather then each node being it's own standalone cluster. This patch also adds ability for user to specify list of topics that should be auto created once the cluster is up and running. --- images/broker/Dockerfile | 6 ++- images/broker/create_topic | 14 ++++++ images/broker/start_kafka | 2 +- images/broker/start_zookeeper | 2 +- start.py | 83 +++++++++++++++++++++++++++++++++-- topology.yaml | 4 ++ 6 files changed, 103 insertions(+), 8 deletions(-) create mode 100755 images/broker/create_topic mode change 100644 => 100755 images/broker/start_kafka mode change 100644 => 100755 images/broker/start_zookeeper diff --git a/images/broker/Dockerfile b/images/broker/Dockerfile index 3c3962a..b2c8b41 100644 --- a/images/broker/Dockerfile +++ b/images/broker/Dockerfile @@ -29,5 +29,7 @@ RUN wget http://archive.apache.org/dist/kafka/${KAFKA_VERSION}/kafka_${SCALA_VER # And we provide custom start up scripts ADD start_kafka / ADD start_zookeeper / -RUN chmod +x start_kafka start_zookeeper \ - && mkdir /logs/ +ADD create_topic / + +# Create structures required by our automation +RUN mkdir /logs/ diff --git a/images/broker/create_topic b/images/broker/create_topic new file mode 100755 index 0000000..ae99f01 --- /dev/null +++ b/images/broker/create_topic @@ -0,0 +1,14 @@ +#!/bin/bash +# Licensed 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. + +/kafka/bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic $1 diff --git a/images/broker/start_kafka b/images/broker/start_kafka old mode 100644 new mode 100755 index 1430019..b69fa7f --- a/images/broker/start_kafka +++ b/images/broker/start_kafka @@ -11,4 +11,4 @@ # See the License for the specific language governing permissions and # limitations under the License. -/kafka/bin/kafka-server-start.sh /kafka/config/server.properties > /logs/kafka.log & +/kafka/bin/kafka-server-start.sh /kafka.properties > /logs/kafka.log & diff --git a/images/broker/start_zookeeper b/images/broker/start_zookeeper old mode 100644 new mode 100755 index e77756d..778ec1a --- a/images/broker/start_zookeeper +++ b/images/broker/start_zookeeper @@ -11,4 +11,4 @@ # See the License for the specific language governing permissions and # limitations under the License. -/kafka/bin/zookeeper-server-start.sh /kafka/config/zookeeper.properties > /logs/zookeeper.log & +/kafka/bin/zookeeper-server-start.sh /zookeeper.properties > /logs/zookeeper.log & diff --git a/start.py b/start.py index 68b34c5..e7180a2 100644 --- a/start.py +++ b/start.py @@ -11,6 +11,7 @@ # See the License for the specific language governing permissions and # limitations under the License. +import json import logging import tempfile import yaml @@ -20,11 +21,41 @@ from clusterdock.utils import wait_for_condition DEFAULT_NAMESPACE = 'clusterdock' +ZOOKEEPER_PORT = 2181 +BROKER_PORT = 9092 logger = logging.getLogger('clusterdock.{}'.format(__name__)) +def success(time): + logger.info('Conditions satisfied after %s seconds.', time) + + +def failure(timeout): + raise TimeoutError('Timed out after {} seconds waiting.'.format(timeout)) + + +# Validate that Zookeeper is up and running by connecting using shell +def validate_zookeeper(node, quiet): + return node.execute('/kafka/bin/zookeeper-shell.sh localhost:2181 ls /', quiet=quiet).exit_code == 0 + + +# Validate that Kafka is up by checking that all brokers are registered in zookeeper +def validate_kafka(node, broker_count, quiet): + command = node.execute('/kafka/bin/zookeeper-shell.sh localhost:2181 ls /brokers/ids | tail -n 1', quiet=quiet) + + if command.exit_code != 0: + return False + + nodes = command.output + if not nodes.startswith('['): + return False + + return len(json.loads(nodes)) == broker_count + def main(args): + quiet = not args.verbose + # Image name image = '{}/{}/topology_apache_kafka:kafka-{}-{}'.format(args.registry, args.namespace or DEFAULT_NAMESPACE, @@ -34,16 +65,60 @@ def main(args): # Nodes in the Kafka cluster nodes = [Node(hostname=hostname, group='brokers', - ports=[2181, 9092], + ports=[ZOOKEEPER_PORT, BROKER_PORT], image=image) for hostname in args.brokers] cluster = Cluster(*nodes) cluster.start(args.network, pull_images=args.always_pull) - # TODO: Add support for cluster mode (e.g. all nodes are part of the same cluster). Today we only - # start each node independently so they will all end up independent one-node clusters. + # Create distributed zookeeper configuration + zookeeper_config = ('tickTime=2000\n', + 'dataDir=/zookeeper\n', + 'clientPort=2181\n', + 'initLimit=5\n', + 'syncLimit=2\n') + for idx, node in enumerate(cluster): + zookeeper_config += 'server.{}={}:2888:3888\n'.format(idx, node.hostname) - for node in cluster: + # Start all zookeepers + for idx, node in enumerate(cluster): + logger.info('Starting Zookeeper on node {}'.format(node.hostname)) + node.execute('mkdir -p /zookeeper') + node.put_file('/zookeeper/myid', str(idx)) + node.put_file('/zookeeper.properties', zookeeper_config) node.execute('/start_zookeeper &', detach=True) + + # Validate that Zookeepr is alive from each node + for node in cluster: + logger.info('Validating Zookeeper on node %s', node.hostname) + wait_for_condition(condition=validate_zookeeper, + condition_args=[node, quiet], + time_between_checks=3, + timeout=60, + success=success, + failure=failure) + + # Start all brokers + for idx, node in enumerate(cluster): + logger.info('Starting Kafka on node {}'.format(node.hostname)) + + kafka_config = node.get_file('/kafka/config/server.properties') + kafka_config = kafka_config.replace('broker.id=0', 'broker.id={}'.format(idx)) + node.put_file('/kafka.properties', kafka_config) + node.execute('/start_kafka &', detach=True) + + # Verify that all Kafka brokers up + logger.info('Waiting on all brokers to register in zookeeper') + wait_for_condition(condition=validate_kafka, + condition_args=[nodes[0], len(nodes), quiet], + time_between_checks=3, + timeout=60, + success=success, + failure=failure) + + # Automatically create topics + for topic in args.topics.split(','): + logger.info('Creating topic %s', topic) + nodes[0].execute('/create_topic {}'.format(topic), quiet=quiet) diff --git a/topology.yaml b/topology.yaml index a2e597a..c50ae66 100644 --- a/topology.yaml +++ b/topology.yaml @@ -35,3 +35,7 @@ start args: default: 2.11 help: Scala version to use metavar: ver + --topics: + default: test + help: Comma-separated list of topics that should be auto-created + metavar: topic1,topic2 From ca341d1c8e35414354e7528809ad1bf610e9ebba Mon Sep 17 00:00:00 2001 From: Jarek Jarcec Cecho Date: Mon, 13 Nov 2017 09:49:29 -0800 Subject: [PATCH 3/3] Removing commas in zookeeper config generation --- start.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/start.py b/start.py index e7180a2..1a061a5 100644 --- a/start.py +++ b/start.py @@ -73,10 +73,10 @@ def main(args): cluster.start(args.network, pull_images=args.always_pull) # Create distributed zookeeper configuration - zookeeper_config = ('tickTime=2000\n', - 'dataDir=/zookeeper\n', - 'clientPort=2181\n', - 'initLimit=5\n', + zookeeper_config = ('tickTime=2000\n' + 'dataDir=/zookeeper\n' + 'clientPort=2181\n' + 'initLimit=5\n' 'syncLimit=2\n') for idx, node in enumerate(cluster): zookeeper_config += 'server.{}={}:2888:3888\n'.format(idx, node.hostname)