[#125] - Enhancing clustering functionality

README.md

@@ -67,10 +67,44 @@ Enables any users listed in the `node['rabbitmq']['enabled_users']` and disables
 ### virtualhost_management
 Enables any vhosts listed in the `node['rabbitmq']['virtualhosts']` and disables any listed in `node['rabbitmq']['disabled_virtualhosts']` attributes.
 
+### cluster
+Configure the cluster between the nodes in the `node['rabbitmq']['clustering']['cluster_nodes']` attribute. It also, supports the auto or manual clustering.
+* Auto clustering : Use auto-configuration of RabbitMQ, http://www.rabbitmq.com/clustering.html#auto-config
+* Manual clustering : Configure the cluster by executing `rabbitmqctl join_cluster` command.
+
+#### Attributes that related to clustering
+* `node['rabbitmq']['cluster']` : Default decision flag of clustering
+* `node['rabbitmq']['erlang_cookie']` : Same erlang cookie is required for the cluster
+* `node['rabbitmq']['clustering']['use\_auto_clustering']` : Default is false. (manual clustering is default)
+* `node['rabbitmq']['clustering']['cluster_name']` : Name of cluster. default value is nil. In case of nil or '' is set for `cluster_name`, first node name in `node['rabbitmq']['clustering']['cluster_nodes']` attribute will be set for manual clustering. for the auto clustering, one of the node name will be set. 
+* `node['rabbitmq']['clustering']['cluster_nodes']` : List of cluster nodes. it required node name and cluster node type. please refer to example in below.
+
+Attributes example
+```ruby
+node['rabbitmq']['cluster'] = true
+node['rabbitmq']['erlang_cookie'] = 'AnyAlphaNumericStringWillDo'
+node['rabbitmq']['cluster_partition_handling'] = 'ignore'
+node['rabbitmq']['clustering']['use_auto_clustering'] = false
+node['rabbitmq']['clustering']['cluster_name'] = 'seoul_tokyo_newyork'
+node['rabbitmq']['clustering']['cluster_nodes'] = [
+    {
+        :name => 'rabbit@rabbit1',
+        :type => 'disc'
+    },
+    {
+        :name => 'rabbit@rabbit2',
+        :type => 'ram'
+    },
+    {
+        :name => 'rabbit@rabbit3',
+        :type => 'disc'
+    }
+]
+```
 
 ## Resources/Providers
 
-There are 4 LWRPs for interacting with RabbitMQ.
+There are 5 LWRPs for interacting with RabbitMQ.
 
 ### plugin
 Enables or disables a rabbitmq plugin. Plugins are not supported for releases prior to 2.7.0.
@@ -167,6 +201,53 @@ rabbitmq_vhost "/nova" do
 end
 ```
 
+### cluster
+Join cluster, set cluster name and change cluster node type.
+
+- `:join` join in cluster as a manual clustering. node will join in first node of json string data.
+
+ - cluster nodes data json format : Data should have all the cluster nodes information.
+
+ ```json
+ [
+     {
+         "name" : "rabbit@rabbit1",
+         "type" : "disc"
+     },
+     {
+         "name" : "rabbit@rabbit2",
+         "type" : "ram"
+     },
+     {
+         "name" "rabbit@rabbit3",
+         "type" : "disc"
+     }
+]
+ ```
+
+- `:set_cluster_name` set the cluster name.
+- `:change_cluster_node_type` change cluster type of node. `disc` or `ram` should be set.
+
+#### Examples
+```ruby
+rabbitmq_cluster '[{"name":"rabbit@rabbit1","type":"disc"},{"name":"rabbit@rabbit2","type":"ram"},{"name":"rabbit@rabbit3","type":"disc"}]' do
+  action :join
+end
+```
+
+```ruby
+rabbitmq_cluster '[{"name":"rabbit@rabbit1","type":"disc"},{"name":"rabbit@rabbit2","type":"ram"},{"name":"rabbit@rabbit3","type":"disc"}]' do
+  cluster_name 'seoul_tokyo_newyork'
+  action :set_cluster_name
+end
+```
+
+```ruby
+rabbitmq_cluster '[{"name":"rabbit@rabbit1","type":"disc"},{"name":"rabbit@rabbit2","type":"ram"},{"name":"rabbit@rabbit3","type":"disc"}]' do
+  action :change_cluster_node_type
+end
+```
+
 
 ## Limitations
 

attributes/default.rb

@@ -46,6 +46,18 @@
 default['rabbitmq']['erlang_cookie'] = 'AnyAlphaNumericStringWillDo'
 default['rabbitmq']['cluster_partition_handling'] = 'ignore'
 
+default['rabbitmq']['clustering']['use_auto_clustering'] = false
+default['rabbitmq']['clustering']['cluster_name'] = nil
+default['rabbitmq']['clustering']['cluster_nodes'] = []
+
+# Manual clustering
+# - Node type : master | slave
+default['rabbitmq']['clustering']['node_type'] = 'master'
+# - Master node name : ex) rabbit@rabbit1
+default['rabbitmq']['clustering']['master_node_name'] = 'rabbit@rabbit1'
+# - Cluster node type : disc | ram
+default['rabbitmq']['clustering']['cluster_node_type'] = 'disc'
+
 # resource usage
 default['rabbitmq']['disk_free_limit_relative'] = nil
 default['rabbitmq']['vm_memory_high_watermark'] = nil

libraries/matchers.rb

@@ -54,4 +54,16 @@ def enable_rabbitmq_plugin(resource_name)
   def disable_rabbitmq_plugin(resource_name)
     ChefSpec::Matchers::ResourceMatcher.new(:rabbitmq_plugin, :disable, resource_name)
   end
+
+  def join_cluster(resource_name)
+    ChefSpec::Matchers::ResourceMatcher.new(:rabbitmq_cluster, :join, resource_name)
+  end
+
+  def set_cluster_name(resource_name) # rubocop:disable AccessorMethodName
+    ChefSpec::Matchers::ResourceMatcher.new(:rabbitmq_cluster, :change_cluster_node_type, resource_name)
+  end
+
+  def change_cluster_node_type(resource_name)
+    ChefSpec::Matchers::ResourceMatcher.new(:rabbitmq_cluster, :change_cluster_node_type, resource_name)
+  end
 end

providers/cluster.rb

@@ -0,0 +1,258 @@
+#
+# Cookbook Name:: rabbitmq
+# Provider:: cluster
+#
+# Author: Sunggun Yu <sunggun.dev@gmail.com>
+# Copyright (C) 2015 Sunggun Yu
+#
+# 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.
+#
+
+include Chef::Mixin::ShellOut
+
+use_inline_resources
+
+# Get ShellOut
+def get_shellout(cmd)
+  sh_cmd = Mixlib::ShellOut.new(cmd)
+  sh_cmd.environment['HOME'] = ENV.fetch('HOME', '/root')
+  sh_cmd
+end
+
+# Execute rabbitmqctl command with args
+def run_rabbitmqctl(*args)
+  cmd = "rabbitmqctl #{args.join(' ')}"
+  Chef::Log.debug("[rabbitmq_cluster] Executing #{cmd}")
+  cmd = get_shellout(cmd)
+  cmd.run_command
+  begin
+    cmd.error!
+    Chef::Log.debug("[rabbitmq_cluster] #{cmd.stdout}")
+  rescue
+    Chef::Application.fatal!("[rabbitmq_cluster] #{cmd.stderr}")
+  end
+end
+
+# Get cluster status result
+def cluster_status
+  # execute > rabbitmqctl cluster_status | sed "1d" | tr "\n" " " | tr -d " "
+  # rabbitmqctl cluster_status returns "Cluster status of node rabbit@rabbit1 ..." at the first line.
+  # To parse the result string, it is removed by sed "1d"
+  cmd = 'rabbitmqctl cluster_status | sed "1d" | tr "\n" " " | tr -d " "'
+  Chef::Log.debug("[rabbitmq_cluster] Executing #{cmd}")
+  cmd = get_shellout(cmd)
+  cmd.run_command
+  cmd.error!
+  result = cmd.stdout.chomp
+  Chef::Log.debug("[rabbitmq_cluster] rabbitmqctl cluster_status : #{result}")
+  result
+end
+
+# Match regex pattern from result of rabbitmqctl cluster_status
+def match_pattern_cluster_status(cluster_status, pattern)
+  if cluster_status.nil? || cluster_status.to_s.empty?
+    Chef::Application.fatal!('[rabbitmq_cluster] cluster_status should not be empty')
+  end
+  match = cluster_status.match(pattern)
+  match[2]
+end
+
+# Get currently joined cluster name from result string of "rabbitmqctl cluster_status"
+def current_cluster_name(cluster_status)
+  pattern = '({cluster_name,<<")(.*?)(">>})'
+  result = match_pattern_cluster_status(cluster_status, pattern)
+  Chef::Log.debug("[rabbitmq_cluster] current_cluster_name : #{result}")
+  result
+end
+
+# Get running nodes
+def running_nodes(cluster_status)
+  pattern = '({running_nodes,\[)(.*?)(\]})'
+  result = match_pattern_cluster_status(cluster_status, pattern)
+  Chef::Log.debug("[rabbitmq_cluster] running_nodes : #{result}")
+  result.split(',')
+end
+
+# Get disc nodes
+def disc_nodes(cluster_status)
+  pattern = '({disc,\[)(.*?)(\]})'
+  result = match_pattern_cluster_status(cluster_status, pattern)
+  Chef::Log.debug("[rabbitmq_cluster] disc_nodes : #{result}")
+  result.split(',')
+end
+
+# Get ram nodes
+def ram_nodes(cluster_status)
+  pattern = '({ram,\[)(.*?)(\]})'
+  result = match_pattern_cluster_status(cluster_status, pattern)
+  Chef::Log.debug("[rabbitmq_cluster] ram_nodes : #{result}")
+  result.split(',')
+end
+
+# Get node name
+def node_name
+  # execute > rabbitmqctl eval 'node().'
+  cmd = 'rabbitmqctl eval "node()."'
+  Chef::Log.debug("[rabbitmq_cluster] Executing #{cmd}")
+  cmd = get_shellout(cmd)
+  cmd.run_command
+  cmd.error!
+  result = cmd.stdout.chomp
+  Chef::Log.debug("[rabbitmq_cluster] node name : #{result}")
+  result
+end
+
+# Get cluster_node_type of current node
+def current_cluster_node_type(node_name, cluster_status)
+  var_cluster_node_type = ''
+  if disc_nodes(cluster_status).include?(node_name)
+    var_cluster_node_type = 'disc'
+  elsif ram_nodes(cluster_status).include?(node_name)
+    var_cluster_node_type = 'ram'
+  end
+  var_cluster_node_type
+end
+
+# Parse hash string of cluster_nodes to JSON object
+def parse_cluster_nodes_string(cluster_nodes)
+  JSON.parse(cluster_nodes.gsub('=>', ':'))
+end
+
+# Checking node is joined in cluster
+def joined_cluster?(node_name, cluster_status)
+  running_nodes(cluster_status).include?(node_name)
+end
+
+# Join cluster.
+def join_cluster(cluster_name)
+  cmd = "rabbitmqctl join_cluster --ram #{cluster_name}"
+  Chef::Log.debug("[rabbitmq_cluster] Executing #{cmd}")
+  cmd = get_shellout(cmd)
+  cmd.run_command
+  begin
+    cmd.error!
+    Chef::Log.info("[rabbitmq_cluster] #{cmd.stdout}")
+  rescue
+    err = cmd.stderr
+    Chef::Log.warn("[rabbitmq_cluster] #{err}")
+    if err.include?('{ok,already_member}')
+      Chef::Log.info('[rabbitmq_cluster] Node is already member of cluster, error will be ignored.')
+    elsif err.include?('cannot_cluster_node_with_itself')
+      Chef::Log.info('[rabbitmq_cluster] Cannot cluster node itself, error will be ignored.')
+    else
+      Chef::Application.fatal!("[rabbitmq_cluster] #{err}")
+    end
+  end
+end
+
+# Change cluster node type
+def change_cluster_node_type(cluster_node_type)
+  cmd = "rabbitmqctl change_cluster_node_type #{cluster_node_type}"
+  Chef::Log.debug("[rabbitmq_cluster] Executing #{cmd}")
+  cmd = get_shellout(cmd)
+  cmd.run_command
+  begin
+    cmd.error!
+    Chef::Log.debug("[rabbitmq_cluster] #{cmd.stdout}")
+  rescue
+    err = cmd.stderr
+    Chef::Log.warn("[rabbitmq_cluster] #{err}")
+    if err.include?('{not_clustered,"Non-clustered nodes can only be disc nodes."}')
+      Chef::Log.info('[rabbitmq_cluster] Node is not clustered yet, error will be ignored.')
+    else
+      Chef::Application.fatal!("[rabbitmq_cluster] #{err}")
+    end
+  end
+end
+
+########################################################################################################################
+# Actions
+#  :join
+#  :change_cluster_node_type
+########################################################################################################################
+
+# Action for joining cluster
+action :join do
+  Chef::Log.info('[rabbitmq_cluster] Action join ... ')
+
+  Chef::Application.fatal!('rabbitmq_cluster with action :join requires a non-nil/empty cluster_nodes.') if new_resource.cluster_nodes.nil? || new_resource.cluster_nodes.empty?
+
+  var_cluster_status = cluster_status
+  var_node_name = node_name
+  var_node_name_to_join = parse_cluster_nodes_string(new_resource.cluster_nodes).first['name']
+
+  if var_node_name == var_node_name_to_join
+    Chef::Log.warn('[rabbitmq_cluster] Trying to join cluster node itself. Joining cluster will be skipped.')
+  elsif joined_cluster?(var_node_name_to_join, var_cluster_status)
+    Chef::Log.warn("[rabbitmq_cluster] Node is already member of #{current_cluster_name(var_cluster_status)}. Joining cluster will be skipped.")
+  else
+    run_rabbitmqctl('stop_app')
+    join_cluster(var_node_name_to_join)
+    run_rabbitmqctl('start_app')
+    Chef::Log.info("[rabbitmq_cluster] Node #{var_node_name} joined in #{var_node_name_to_join}")
+    Chef::Log.info(cluster_status)
+  end
+end
+
+# Action for set cluster name
+action :set_cluster_name do
+  Chef::Application.fatal!('rabbitmq_cluster with action :join requires a non-nil/empty cluster_nodes.') if new_resource.cluster_nodes.nil? || new_resource.cluster_nodes.empty?
+  var_cluster_status = cluster_status
+  var_cluster_name = new_resource.cluster_name
+  unless current_cluster_name(var_cluster_status) == var_cluster_name
+    unless var_cluster_name.empty?
+      run_rabbitmqctl("set_cluster_name #{var_cluster_name}")
+      Chef::Log.info("[rabbitmq_cluster] Cluster name has been set : #{current_cluster_name(cluster_status)}")
+    end
+  end
+end
+
+# Action for changing cluster node type
+action :change_cluster_node_type do
+  Chef::Log.info('[rabbitmq_cluster] Action change_cluster_node_type ... ')
+
+  Chef::Application.fatal!('rabbitmq_cluster with action :join requires a non-nil/empty cluster_nodes.') if new_resource.cluster_nodes.nil? || new_resource.cluster_nodes.empty?
+
+  var_cluster_status = cluster_status
+  var_node_name = node_name
+  var_current_cluster_node_type = current_cluster_node_type(var_node_name, var_cluster_status)
+  var_cluster_node_type = parse_cluster_nodes_string(new_resource.cluster_nodes).select { |node| node['name'] == var_node_name }.first['type'] # ~FC039
+
+  if var_current_cluster_node_type == var_cluster_node_type
+    Chef::Log.warn('[rabbitmq_cluster] Skip changing cluster node type : trying to change to same cluster node type')
+    node_type_changeable = false
+  else
+    if var_cluster_node_type == 'ram'
+      if var_current_cluster_node_type == 'disc' && disc_nodes(var_cluster_status).length < 2
+        Chef::Log.warn('[rabbitmq_cluster] At least one disc node is required for rabbitmq cluster. Changing cluster node type will be ignored.')
+        node_type_changeable = false
+      else
+        node_type_changeable = true
+      end
+    elsif var_cluster_node_type == 'disc'
+      node_type_changeable = true
+    else
+      Chef::Log.warn("[rabbitmq_cluster] Unexpected cluster_note_type #{var_cluster_node_type}. Changing cluster node type will be ignored.")
+      node_type_changeable = false
+    end
+  end
+
+  # Change cluster node type
+  if node_type_changeable
+    run_rabbitmqctl('stop_app')
+    change_cluster_node_type(var_cluster_node_type)
+    run_rabbitmqctl('start_app')
+    Chef::Log.info("[rabbitmq_cluster] The cluster node type of #{var_node_name} has been changed into #{var_cluster_node_type}")
+    Chef::Log.info(cluster_status)
+  end
+end