Merge pull request #130 from jvlcek/add_kafka

Initial commit of kafka server configuration

bin/appliance_console

@@ -384,13 +384,15 @@ Static Network Configuration
           "Create Region in External Database" => "create_external",
           "Join Region in External Database"   => "join_external",
           "Reset Configured Database"          => "reset_region",
-          "No changes"                         => "no_changes"
+          "Make No Database Changes"           => "no_changes"
         }
         database_action = ask_with_menu("Database Operation", options)
 
         messaging_options = {
-          "Use an external messaging system" => "external_messaging",
-          "No changes"                       => "no_changes"
+          "Configure this appliance as a messaging server"           => "message_server",
+          "Connect to an appliance configured as a messaging server" => "message_client",
+          "Connect to an external messaging system"                  => "external_messaging",
+          "Make No messaging changes"                                => "no_changes"
         }
 
         messaging_action = ask_with_menu("Configure Messaging", messaging_options)
@@ -420,6 +422,33 @@ Static Network Configuration
         region = ManageIQ::ApplianceConsole::DatabaseConfiguration.region
 
         case messaging_action
+        when "message_server"
+          say("#{selection}\n\n")
+
+          message_server = MessageServerConfiguration.new
+          if message_server.ask_questions && message_server.activate
+            message_server.post_activation
+            say("\nMessage Server configured successfully.\n")
+            press_any_key
+          else
+            say("\nMessage Server configuration failed!\n")
+            press_any_key
+            raise MiqSignalError
+          end
+
+        when "message_client"
+          say("#{selection}\n\n")
+
+          message_client = MessageClientConfiguration.new
+          if message_client.ask_questions && message_client.activate
+            say("\nMessage Client configured successfully.\n")
+            press_any_key
+          else
+            say("\nMessage Client configuration failed!\n")
+            press_any_key
+            raise MiqSignalError
+          end
+
         when "external_messaging"
           ManageIQ::ApplianceConsole::MessagingConfiguration.new.run_interactive
         end

lib/manageiq-appliance_console.rb

@@ -43,6 +43,8 @@ def self.logger=(logger)
 require 'manageiq/appliance_console/key_configuration'
 require 'manageiq/appliance_console/logfile_configuration'
 require 'manageiq/appliance_console/logical_volume_management'
+require 'manageiq/appliance_console/message_configuration_client'
+require 'manageiq/appliance_console/message_configuration_server'
 require 'manageiq/appliance_console/messaging_configuration'
 require 'manageiq/appliance_console/oidc_authentication'
 require 'manageiq/appliance_console/principal'

lib/manageiq/appliance_console/message_configuration.rb

@@ -0,0 +1,152 @@
+require 'active_support/core_ext/module/delegation'
+require 'pathname'
+
+module ManageIQ
+  module ApplianceConsole
+    class MessageConfiguration
+      attr_reader :username, :password,
+                  :miq_config_dir_path, :config_dir_path, :sample_config_dir_path,
+                  :client_properties_path,
+                  :keystore_dir_path, :truststore_path, :keystore_path,
+                  :messaging_yaml_sample_path, :messaging_yaml_path
+
+      BASE_DIR                          = "/opt/kafka".freeze
+      LOGS_DIR                          = "#{BASE_DIR}/logs".freeze
+      CONFIG_DIR                        = "#{BASE_DIR}/config".freeze
+      SAMPLE_CONFIG_DIR                 = "#{BASE_DIR}/config-sample".freeze
+      MIQ_CONFIG_DIR                    = ManageIQ::ApplianceConsole::RAILS_ROOT.join("config").freeze
+
+      def initialize(options = {})
+        @username = options[:username] || "admin"
+        @password = options[:password]
+
+        @miq_config_dir_path               = Pathname.new(MIQ_CONFIG_DIR)
+        @config_dir_path                   = Pathname.new(CONFIG_DIR)
+        @sample_config_dir_path            = Pathname.new(SAMPLE_CONFIG_DIR)
+
+        @client_properties_path            = config_dir_path.join("client.properties")
+        @keystore_dir_path                 = config_dir_path.join("keystore")
+        @truststore_path                   = keystore_dir_path.join("truststore.jks")
+        @keystore_path                     = keystore_dir_path.join("keystore.jks")
+
+        @messaging_yaml_sample_path        = miq_config_dir_path.join("messaging.kafka.yml")
+        @messaging_yaml_path               = miq_config_dir_path.join("messaging.yml")
+      end
+
+      def already_configured?
+        installed_file_found = false
+        installed_files.each do |f|
+          if File.exist?(f)
+            installed_file_found = true
+            say("Installed file #{f} found.")
+          end
+        end
+        installed_file_found
+      end
+
+      def ask_questions
+        return false unless valid_environment?
+
+        ask_for_parameters
+        show_parameters
+        return false unless agree("\nProceed? (Y/N): ")
+
+        return false unless host_reachable?(server_hostname, "Message Server Hostname:")
+
+        true
+      end
+
+      def create_client_properties
+        say(__method__.to_s.tr("_", " ").titleize)
+
+        return if file_found?(client_properties_path)
+
+        content = <<~CLIENT_PROPERTIES
+          ssl.truststore.location=#{truststore_path}
+          ssl.truststore.password=#{password}
+
+          sasl.mechanism=PLAIN
+          security.protocol=SASL_SSL
+          sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \\
+            username=#{username} \\
+            password=#{password} ;
+        CLIENT_PROPERTIES
+
+        File.write(client_properties_path, content)
+      end
+
+      def configure_messaging_yaml
+        say(__method__.to_s.tr("_", " ").titleize)
+
+        return if file_found?(messaging_yaml_path)
+
+        messaging_yaml = YAML.load_file(messaging_yaml_sample_path)
+
+        messaging_yaml["production"]["hostname"] = server_hostname
+        messaging_yaml["production"]["port"] = 9093
+        messaging_yaml["production"]["username"] = username
+        messaging_yaml["production"]["password"] = ManageIQ::Password.try_encrypt(password)
+
+        File.write(messaging_yaml_path, messaging_yaml.to_yaml)
+      end
+
+      def remove_installed_files
+        say(__method__.to_s.tr("_", " ").titleize)
+
+        installed_files.each { |f| FileUtils.rm_rf(f) }
+      end
+
+      def valid_environment?
+        if already_configured?
+          return false unless agree("\nAlready configured on this Appliance, Un-Configure first? (Y/N): ")
+
+          deactivate
+          return false unless agree("\nProceed with Configuration? (Y/N): ")
+        end
+        true
+      end
+
+      def file_found?(path)
+        return false unless File.exist?(path)
+
+        say("\tWARNING: #{path} already exists. Taking no action.")
+        true
+      end
+
+      def files_found?(path_list)
+        return false unless path_list.all? { |path| File.exist?(path) }
+
+        path_list.each { |path| file_found?(path) }
+        true
+      end
+
+      def file_contains?(path, content)
+        return false unless File.exist?(path)
+
+        content.split("\n").each do |l|
+          l.gsub!("/", "\\/")
+          l.gsub!(/password=.*$/, "password=") # Remove the password as it can have special characters that grep can not match.
+          return false unless File.foreach(path).grep(/#{l}/).any?
+        end
+
+        say("Content already exists in #{path}. Taking no action.")
+        true
+      end
+
+      #
+      # Network validation
+      #
+      def host_reachable?(host, what)
+        require 'net/ping'
+        say("Checking connectivity to #{host} ... ")
+        unless Net::Ping::External.new(host).ping
+          say("Failed.\nCould not connect to #{host},")
+          say("the #{what} must be reachable by name.")
+          return false
+        end
+        say("Succeeded.")
+        true
+      end
+    end
+  end
+end

lib/manageiq/appliance_console/message_configuration_client.rb

@@ -0,0 +1,81 @@
+require "awesome_spawn"
+require "fileutils"
+require "linux_admin"
+require 'net/scp'
+require "manageiq/appliance_console/message_configuration"
+
+module ManageIQ
+  module ApplianceConsole
+    class MessageClientConfiguration < MessageConfiguration
+      attr_reader :server_hostname, :server_username, :server_password, :installed_files
+
+      def initialize(options = {})
+        super(options)
+
+        @server_hostname = options[:server_hostname]
+        @server_username = options[:server_usernamed] || "root"
+        @server_password = options[:server_password]
+
+        @installed_files = [client_properties_path, messaging_yaml_path, truststore_path]
+      end
+
+      def activate
+        begin
+          configure_messaging_yaml     # Set up the local message client in case EVM is actually running on this, Message Server
+          create_client_properties     # Create the client.properties configuration fle
+          fetch_truststore_from_server # Fetch the Java Keystore from the Kafka Server
+        rescue AwesomeSpawn::CommandResultError => e
+          say(e.result.output)
+          say(e.result.error)
+          say("")
+          say("Failed to Configure the Message Client- #{e}")
+          return false
+        rescue => e
+          say("Failed to Configure the Message Client- #{e}")
+          return false
+        end
+        true
+      end
+
+      def ask_for_parameters
+        say("\nMessage Client Parameters:\n\n")
+
+        @server_hostname = ask_for_string("Message Server Hostname")
+        @server_username = ask_for_string("Message Server Username", server_username)
+        @server_password = ask_for_password("Message Server Password")
+
+        @username  = ask_for_string("Message Key Username", username)
+        @password  = ask_for_password("Message Key Password")
+      end
+
+      def show_parameters
+        say("\nMessage Client Configuration:\n")
+        say("Message Client Details:\n")
+        say("  Message Server Hostname:   #{server_hostname}\n")
+        say("  Message Server Username:   #{server_username}\n")
+        say("  Message Key Username:      #{username}\n")
+      end
+
+      private
+
+      def fetch_truststore_from_server
+        say(__method__.to_s.tr("_", " ").titleize)
+
+        return if file_found?(truststore_path)
+
+        Net::SCP.start(server_hostname, server_username, :password => server_password) do |scp|
+          scp.download!(truststore_path, truststore_path)
+        end
+
+        File.exist?(truststore_path)
+      rescue => e
+        say("Failed to fetch server truststore: #{e.message}")
+        false
+      end
+
+      def deactivate
+        remove_installed_files
+      end
+    end
+  end
+end