WIP for Kafka Manager

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

@@ -94,6 +94,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"
@@ -134,6 +135,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
@@ -253,9 +286,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,13 @@
+#!/usr/bin/env bash
+
+if [ -d /usr/share/kafka-manager ]; then
+    echo "Kafka manager is installed"
+    exit
+fi
 
 # ---
 # Download
+cd /tmp
 git clone https://github.com/yahoo/kafka-manager
 
 
@@ -10,16 +17,9 @@ 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

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_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,62 @@
+require_relative 'kafka_manager_helpers'
+
+# Kafka Manager Configuration
+# https://github.com/yahoo/kafka-manager#configuration
+#
+namespace :kafka_manager do
+  namespace :service do
+    def kafka_manager_conf
+      @kafka_manager_conf ||= capture('ls /usr/share/kafka-manager/conf/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 configuration.basicAuthentication.enabled
+      enabled = "basicAuthentication.enabled=#{configuration.basicAuthentication.enabled}"
+      sudo("sed -i -e 's#basicAuthentication.enabled=.*##{enabled}#' #{kafka_manager_conf}")
+      # basicAuthentication.username="admin"
+      username = "basicAuthentication.username=\"#{configuration.basicAuthentication.username}\""
+      sudo("sed -i -e 's#basicAuthentication.username=.*##{username}#' #{kafka_manager_conf}")
+      # basicAuthentication.password="password"
+      password = "basicAuthentication.password=\"#{configuration.basicAuthentication.password}\""
+      sudo("sed -i -e 's#basicAuthentication.password=.*##{password}#' #{kafka_manager_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=[#{configuration.features.join(',')}]"
+      sudo("sed -i -e 's#application.features=.*##{features}#' #{kafka_manager_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
+      # 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_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,23 @@
+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.sbt)
+      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
+        KafkaManagerHelpers.manager.reboot_node(host_settings)
+      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
+      KafkaManagerHelpers.manager.describe_nodes
+    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,42 @@
+require_relative 'kafka_manager_helpers'
+
+namespace :kafka_manager do
+  namespace :service do
+    def kafka_manager_running?
+      pid = capture('ls /usr/share/kafka-manager/RUNNING_PID')
+      ! pid.nil?
+    end
+
+    desc 'Start Kafka Manager'
+    task :start do
+      on roles(:kafka_manager) do |host|
+        # TODO: Create 'kafka' user/group to run the service
+        if kafka_manager_running?
+          puts "#{host.hostname} is already running Kafka Manager"
+        else
+          sudo('kafka-manager')
+        end
+      end
+    end
+
+    desc 'Status of Kafka Manager'
+    task :status do
+      on roles(:kafka_manager) do |host|
+        if kafka_running?
+          puts "#{host.hostname} is running Kafka Manager"
+        else
+          puts "#{host.hostname} is not running Kafka Manager"
+        end
+      end
+    end
+
+    desc 'Stop Kafka Manager'
+    task :stop do
+      on roles(:kafka_manager) do
+        # Ignore the exit(1) status when it's not running already
+        sudo('${KAFKA_BIN}/kafka-server-stop.sh || true')
+      end
+    end
+  end
+end
+

lib/ubuntu/ubuntu_helper.rb

@@ -79,6 +79,10 @@ def kafka_bin(kafka_ver = '')
     "#{script_path}/kafka_bin.sh #{kafka_ver} > #{log_path}/kafka_bin.log"
   end
 
+  def kafka_manager
+    "#{script_path}/kafka_manager.sh #{kafka_ver} > #{log_path}/kafka_manager.log"
+  end
+
   def log_path_files
     "find #{log_path} -type f"
   end