[WIP] Kafka manager

README.md

@@ -6,6 +6,8 @@ Utils to provision services on AWS
 
 Incomplete - work in progress.
 
+See also `lib/{service}/README.md` documentation.
+
 # Goals
  - use ruby AWS API client to provision AWS resources
  - use capistrano for service installation/configuration

config/deploy/test.rb

@@ -42,6 +42,8 @@
 server 'test_kafka2', user: 'ubuntu', roles: %w[ubuntu kafka]
 server 'test_kafka3', user: 'ubuntu', roles: %w[ubuntu kafka]
 
+server 'test_kafka_manager', user: 'ubuntu', roles: %w[ubuntu kafka_manager]
+
 
 # role-based syntax
 # ==================

config/settings/test.yml

@@ -108,6 +108,7 @@ aws:
   #   - advertised.listeners
   test_kafka_configuration:
     resource: configuration
+    tag_service: kafka
     kafka_home: "/opt/kafka" # symlinks to /opt/kafka_{SCALA_VERSION}-{KAFKA_VERSION}
     kafka_version: "0.11.0.0"
     scala_version: "2.11"
@@ -148,6 +149,38 @@ aws:
     tag_name: test_kafka3
     availability_zone: us-west-2c  # need one node in each zone
 
+  # ---
+  # Kafka Manager
+
+  test_kafka_manager_configuration:
+    resource: configuration
+    tag_service: kafka_manager
+    basicAuthentication:
+      enabled: false
+      username: "admin"
+      password: "password"
+      realm: "Kafka-Manager"
+      excluded:
+        - "/api/health"
+    # KMClusterManagerFeature - allows adding, updating, deleting cluster from Kafka Manager
+    # KMTopicManagerFeature - allows adding, updating, deleting topic from a Kafka cluster
+    # KMPreferredReplicaElectionFeature - allows running of preferred replica election for a Kafka cluster
+    # KMReassignPartitionsFeature - allows generating partition assignments and reassigning partitions
+    features:
+      - KMClusterManagerFeature
+      - KMTopicManagerFeature
+      - KMPreferredReplicaElectionFeature
+      - KMReassignPartitionsFeature
+
+  test_kafka_manager:
+    <<: *TestDefault
+    tag_service: kafka_manager
+    tag_group: test_kafka
+    tag_name: test_kafka_manager
+    security_groups:
+      - test_ssh_security_group
+      - test_kafka_security_group
+
   # ---
   # Mesos Nodes
   # - note definition and use of defaults: MesosMasterDefaults, MesosAgentDefaults
@@ -267,9 +300,16 @@ aws:
       group_id:
       ip_permissions:
         - ip_protocol: tcp
-          from_port: 9092
+          from_port: 9092  # kafka clients
           to_port: 9092
           ip_ranges:
             - cidr_ip: '0.0.0.0/0'
           ipv_6_ranges:
             - cidr_ipv_6: '::/0'
+        - ip_protocol: tcp
+          from_port: 9000  # kafka manager
+          to_port: 9000
+          ip_ranges:
+            - cidr_ip: '0.0.0.0/0'
+          ipv_6_ranges:
+            - cidr_ipv_6: '::/0'

lib/bash/debian/kafka_manager.sh

@@ -1,6 +1,46 @@
+#!/usr/bin/env bash
+
+if which kafka-manager; then
+    echo "Kafka manager is installed"
+    exit
+fi
+
+curl -s https://packagecloud.io/install/repositories/spuder/kafka-manager/script.deb.sh | sudo bash
+
+sudo apt-get install -y -q kafka-manager
+
+
+#---
+# Enable systemd
+# https://github.com/yahoo/kafka-manager/issues/373
+
+cat > /tmp/kafka-manager.service <<EOF
+[Unit]
+Description=Kafka Manager
+After=network.target
+
+[Install]
+WantedBy=multi-user.target
+
+[Service]
+ExecStart=/usr/bin/kafka-manager -Dpidfile.path=/var/run/kafka-manager.pid -Dconfig.file=/etc/kafka-manager/application.conf -Dlogger.file=/etc/kafka-manager/logger.xml
+Type=simple
+Restart=always
+EOF
+
+sudo mv /tmp/kafka-manager.service /lib/systemd/system/
+sudo chown root:root /lib/systemd/system/kafka-manager.service
+sudo chmod 644       /lib/systemd/system/kafka-manager.service
+sudo systemctl enable kafka-manager.service
+
+# TODO: Create 'kafka' user/group to run the service
+
+exit
+
 
 # ---
 # Download
+cd /tmp
 git clone https://github.com/yahoo/kafka-manager
 
 
@@ -10,16 +50,14 @@ cd kafka-manager
 sbt debian:packageBin
 
 
-exit
-
-
 # ---
 # Install
-cd ../
-DIST=$(basename ${SRC}/core/build/distributions/kafka_*-${VER}.tgz .tgz)
-tar zxf ${SRC}/core/build/distributions/${DIST}.tgz -C /usr/local/
-rm -f /usr/local/kafka
-ln -s /usr/local/$DIST /usr/local/kafka
 
+sudo dpkg -i -R target/
 
+# dpkg -L kafka-manager
+
+# ---
+# Cleanup
 
+sbt clean clean-files

lib/boot.rb

@@ -24,5 +24,6 @@
 require_relative 'redhat/redhat_helper'
 
 require_relative 'kafka/kafka_helpers'
+require_relative 'kafka_manager/kafka_manager_helpers'
 require_relative 'zookeeper/zookeeper_helpers'
 

lib/kafka/kafka_nodes.rake

@@ -16,7 +16,7 @@ namespace :kafka do
 
     desc 'Find and describe all nodes'
     task :find do
-      KafkaHelpers.manager.describe_nodes
+      puts KafkaHelpers.manager.describe_nodes.join("\n")
     end
 
     desc 'Reboot Kafka systems - WARNING, can reset IPs'

lib/kafka_manager/README.md

@@ -0,0 +1,19 @@
+# Kafka Manager
+
+- https://github.com/yahoo/kafka-manager
+
+## Configuration
+
+- https://github.com/yahoo/kafka-manager#configuration
+
+See `lib/kafka/kafka_manager_configure.rake`
+
+Look for "kafka_manager" in:
+ - `config/settings/{stage}.yml`
+ - `config/deploy/{stage}.rb`
+
+## Capistrano tasks
+
+```bash
+$ bundle exec cap -T | grep kafka_manager
+```

lib/kafka_manager/kafka_manager_configure.rake

@@ -0,0 +1,88 @@
+require_relative 'kafka_manager_helpers'
+
+# Kafka Manager Configuration
+# https://github.com/yahoo/kafka-manager#configuration
+#
+namespace :kafka_manager do
+  namespace :service do
+
+
+    # /etc/default/kafka-manager defines:
+    #
+    #JAVA_OPTS="-Dpidfile.path=/var/run/kafka-manager.pid -Dconfig.file=/etc/kafka-manager/application.conf -Dlogger.file=/etc/kafka-manager/logger.xml $JAVA_OPTS"
+    #PIDFILE="/var/run/kafka-manager.pid"
+
+    ##########################################################################
+    # BEGIN from https://github.com/yahoo/kafka-manager/issues/373
+
+    # In /opt/kafka-manager-1.3.3.6/conf/application.ini
+    #
+    # -Dapplication.home=/opt/kafka-manager-1.3.3.6
+    # -Dpidfile.path=/opt/kafka-manager-1.3.3.6/kafka-manager.pid
+    # -Dhttp.port=8080
+
+    # END from https://github.com/yahoo/kafka-manager/issues/373
+    ##########################################################################
+
+
+
+    def kafka_manager_application_conf
+      # @kafka_manager_application_conf ||= capture('ls /usr/share/kafka-manager/conf/application.conf')
+      @kafka_manager_application_conf ||= capture('ls /etc/kafka-manager/application.conf')
+    end
+
+    # basicAuthentication
+    # basicAuthentication.enabled=false
+    # basicAuthentication.username="admin"
+    # basicAuthentication.password="password"
+    # basicAuthentication.realm="Kafka-Manager"
+    # basicAuthentication.excluded=["/api/health"] # ping the health of your instance without authentification
+    def kafka_manager_authentication
+      return unless KafkaManagerHelpers.configuration.basicAuthentication.enabled
+      auth = KafkaManagerHelpers.configuration.basicAuthentication
+      enabled = 'basicAuthentication.enabled=true'
+      sudo("sed -i -e 's#basicAuthentication.enabled=.*##{enabled}#' #{kafka_manager_application_conf}")
+      # basicAuthentication.username="admin"
+      username = "basicAuthentication.username=\"#{auth.username}\""
+      sudo("sed -i -e 's#basicAuthentication.username=.*##{username}#' #{kafka_manager_application_conf}")
+      # basicAuthentication.password="password"
+      password = "basicAuthentication.password=\"#{auth.password}\""
+      sudo("sed -i -e 's#basicAuthentication.password=.*##{password}#' #{kafka_manager_application_conf}")
+    end
+
+    # application.features=["KMClusterManagerFeature","KMTopicManagerFeature",
+    #                       "KMPreferredReplicaElectionFeature","KMReassignPartitionsFeature"]
+    #
+    # KMClusterManagerFeature - allows adding, updating, deleting cluster from Kafka Manager
+    # KMTopicManagerFeature - allows adding, updating, deleting topic from a Kafka cluster
+    # KMPreferredReplicaElectionFeature - allows running of preferred replica election for a Kafka cluster
+    # KMReassignPartitionsFeature - allows generating partition assignments and reassigning partitions
+    def kafka_manager_features
+      features = "application.features=[#{KafkaManagerHelpers.configuration.features.join(',')}]"
+      sudo("sed -i -e 's#application.features=.*##{features}#' #{kafka_manager_application_conf}")
+    end
+
+    # Set kafka-manager.zkhosts (note the /kafka chroot path)
+    # - for multiple ZooKeeper instances, the kafka-manager.zkhosts should be a
+    #   comma-separated string listing the IP addresses and port numbers
+    #   of all the ZooKeeper instances.
+    def kafka_manager_zookeeper_connect
+      # kafka-manager.zkhosts="my-zookeeper-connection-string"
+      # kafka-manager.base-zk-path="/a-chroot"  - don't see this in the installed file
+      #
+      # Note the use of a '#' in sed delimiter, because connections may contain `/` chars
+      zk = ZookeeperHelpers.connections(false).join(',')
+      zoo_connect = "kafka-manager.zkhosts=#{zk}/kafka"
+      sudo("sed -i -e 's#kafka-manager.zkhosts=.*##{zoo_connect}#' #{kafka_manager_application_conf}")
+    end
+
+    desc 'Configure Kafka Manager'
+    task :configure do
+      on roles(:kafka_manager), in: :parallel do |host|
+        kafka_manager_features
+        kafka_manager_zookeeper_connect
+      end
+    end
+  end
+end
+

lib/kafka_manager/kafka_manager_helpers.rb

@@ -0,0 +1,24 @@
+
+# Utilities for working with Kafka
+module KafkaManagerHelpers
+
+  module_function
+
+  SERVICE = 'kafka_manager'.freeze
+
+  # KAFKA_HOME_DEFAULT = '/opt/kafka'.freeze
+
+  def settings
+    @settings ||= ServiceSettings.new SERVICE
+  end
+
+  def manager
+    @manager ||= ServiceManager.new SERVICE
+  end
+
+  def configuration
+    settings.configuration
+  end
+
+end
+

lib/kafka_manager/kafka_manager_install.rake

@@ -0,0 +1,21 @@
+require_relative 'kafka_manager_helpers'
+
+# Kafka Manager Installation
+# https://github.com/yahoo/kafka-manager
+#
+namespace :kafka_manager do
+  namespace :service do
+    def install_kafka_manager
+      install_java8
+      sudo(ubuntu_helper.kafka_manager)
+    end
+
+    desc 'Install Kafka Manager service'
+    task :install do
+      on roles(:kafka_manager), in: :parallel do |host|
+        install_kafka_manager
+      end
+    end
+  end
+end
+

lib/kafka_manager/kafka_manager_nodes.rake

@@ -0,0 +1,43 @@
+require_relative 'kafka_manager_helpers'
+
+namespace :kafka_manager do
+  namespace :nodes do
+    desc 'List settings in this project'
+    task :check_settings do
+      KafkaManagerHelpers.settings.nodes.each do |params|
+        puts JSON.pretty_generate(JSON.parse(params.to_json))
+      end
+    end
+
+    desc 'Create nodes'
+    task :create do
+      KafkaManagerHelpers.manager.create_nodes
+    end
+
+    desc 'Find and describe all nodes'
+    task :find do
+      puts KafkaManagerHelpers.manager.describe_nodes.join("\n")
+    end
+
+    desc 'Reboot Kafka systems - WARNING, can reset IPs'
+    task :reboot do
+      KafkaManagerHelpers.manager.reboot_nodes
+    end
+
+    desc 'Terminate nodes'
+    task :terminate do
+      KafkaManagerHelpers.manager.terminate_nodes
+    end
+
+    desc 'Compose public entries for ~/.ssh/config for nodes'
+    task :ssh_config_public do
+      puts KafkaManagerHelpers.manager.ssh_config
+    end
+
+    desc 'Compose entries for /etc/hosts using public IPs'
+    task :etc_hosts_public do
+      puts KafkaManagerHelpers.manager.etc_hosts.join("\n")
+    end
+  end
+end
+

lib/kafka_manager/kafka_manager_service.rake

@@ -0,0 +1,27 @@
+require_relative 'kafka_manager_helpers'
+
+namespace :kafka_manager do
+  namespace :service do
+    desc 'Start Kafka Manager'
+    task :start do
+      on roles(:kafka_manager) do
+        sudo('service kafka-manager start')
+      end
+    end
+
+    desc 'Status of Kafka Manager'
+    task :status do
+      on roles(:kafka_manager) do
+        execute('service kafka-manager status || true')
+      end
+    end
+
+    desc 'Stop Kafka Manager'
+    task :stop do
+      on roles(:kafka_manager) do
+        sudo('service kafka-manager stop')
+      end
+    end
+  end
+end
+