Class: MCollective::Application::Federation

Inherits:
Application
  • Object
show all
Defined in:
lib/mcollective/application/federation.rb

Instance Method Summary collapse

Instance Method Details

#choria ⇒ Util::Choria

Creates and cache a Choria helper class

Returns:



195
196
197
 
# File 'lib/mcollective/application/federation.rb', line 195

def choria
  @_choria ||= Util::Choria.new
end

#destination_hop(route) ⇒ Object 



66
67
68
 
# File 'lib/mcollective/application/federation.rb', line 66

def destination_hop(route)
  route[route_to(route, destination).size]
end

#display_federated_route(route) ⇒ Object

TODO:

this is aweful brute force rubbish while I dont know what I want, make some algo



129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
 
# File 'lib/mcollective/application/federation.rb', line 129

def display_federated_route(route)
  c_out, fb1, node, fb2, c_in = route

  puts "  %s" % Util.colorize(:bold, "Request:")
  puts "    %s" % Util.colorize(:cyan, c_out[0])
  puts draw_cluster(c_out, fb1, :out, 7)
  puts "            └─ %s" % Util.colorize(:green, fb1[1])
  puts draw_cluster(fb1, node, :out, 15)
  puts "                    └── %s" % Util.colorize(:yellow, node[1])
  puts
  puts "  %s" % Util.colorize(:bold, "Reply:")
  puts "                    ┌── %s" % Util.colorize(:yellow, node[1])
  puts draw_cluster(node, fb2, :int, 15)
  puts "            ┌─ %s" % Util.colorize(:green, fb2[1])
  puts draw_cluster(fb2, c_in, :in, 7)
  puts "    %s" % Util.colorize(:cyan, c_in[1])
  puts
  puts "[%s] Client [%s] Federation Broker [%s] Server [%s] Middleware" % [
    Util.colorize(:cyan, "█"), Util.colorize(:green, "█"), Util.colorize(:yellow, "█"), "█"
  ]
  puts
end

#display_unfederated_route(route) ⇒ Object

TODO:

this is aweful brute force rubbish while I dont know what I want, make some algo



95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
 
# File 'lib/mcollective/application/federation.rb', line 95

def display_unfederated_route(route)
  c_out, node, c_in = route

  if c_out[1] == node[0] && node[2] == c_out[1] && node[0] == c_out[1]
    puts "  Shared Middleware"
    puts
    puts "    %s" % Util.colorize(:cyan, c_out[0])
    puts draw_cluster(c_out, node, :out, 7)
    puts "           └── %s" % Util.colorize(:yellow, node[1])

  elsif c_out[1] == c_in[0] && node[0] == node[2]
    puts "  NATS Cluster with Symetrical Path"
    puts
    puts "    %s" % Util.colorize(:cyan, c_out[0])
    puts draw_cluster(c_out, node, :out, 7)
    puts "           └── %s" % Util.colorize(:yellow, node[1])

  else
    puts "  NATS Cluster with Asymmetrical Path"
    puts
    puts "  %s" % Util.colorize(:cyan, c_out[0])
    puts draw_cluster(c_out, node, :out, 5)
    puts "         ├── %s" % Util.colorize(:yellow, node[1])
    puts draw_cluster(node, c_in, :in, 5)
    puts "  %s" % Util.colorize(:cyan, c_in[1])
  end
  puts
  puts "[%s] Client [%s] Server [%s] Middleware" % [
    Util.colorize(:cyan, "█"), Util.colorize(:yellow, "█"), "█"
  ]
  puts
end

#draw_cluster(left, right, direction, indent) ⇒ Object 



76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
 
# File 'lib/mcollective/application/federation.rb', line 76

def draw_cluster(left, right, direction, indent)
  if left.last == right.first && direction == :out
    "%s└── %s" % [" " * indent, left.last]
  elsif left.last == right.first
    "%s┌── %s" % [" " * indent, left.last]
  elsif direction == :out
    [
      "%s└─┐ %s" % [" " * indent, left.last],
      "%s  └ %s" % [" " * indent, right.first]
    ].join("\n")
  else
    [
      "%s  ┌ %s" % [" " * indent, left.last],
      "%s┌─┘ %s" % [" " * indent, right.first]
    ].join("\n")
  end
end

#extract_federation_brokers(route, destination) ⇒ Object 



70
71
72
73
74
 
# File 'lib/mcollective/application/federation.rb', line 70

def extract_federation_brokers(route, destination)
  abort("Unexpected route size %d found" % route.size) unless route.size == 5

  [route_to(route, destination).last[1], route_back(route, destination).first[1]]
end

#main ⇒ Object 



223
224
225
226
227
 
# File 'lib/mcollective/application/federation.rb', line 223

def main
  send("%s_command" % configuration[:command])
rescue Interrupt
  exit
end

#post_option_parser(configuration) ⇒ Object 



199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
 
# File 'lib/mcollective/application/federation.rb', line 199

def post_option_parser(configuration)
  if ARGV.length >= 1
    configuration[:command] = ARGV.shift
  else
    abort("Please specify a command, valid commands are: %s" % valid_commands.join(", "))
  end

  if configuration[:command] == "trace"
    if ARGV.length >= 1
      configuration[:host] = ARGV.shift
    else
      abort("Please specify a host to trace, example: mco federation trace node1.prod.example.net")
    end
  end
end

#route_back(route, destination) ⇒ Object 



62
63
64
 
# File 'lib/mcollective/application/federation.rb', line 62

def route_back(route, destination)
  route.reverse.take_while {|hop| hop.size == 2 || hop[1] != destination}.reverse
end

#route_to(route, destination) ⇒ Object 



58
59
60
 
# File 'lib/mcollective/application/federation.rb', line 58

def route_to(route, destination)
  route.take_while {|hop| hop[1] != destination}
end

#trace_command ⇒ Object 



152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
 
# File 'lib/mcollective/application/federation.rb', line 152

def trace_command
  result = trace_node(configuration[:host])

  abort("No routes were reported, are your nodes running a supported version?") if result[:route].empty?

  puts "Received response from %s in %.2fms for message %s" % [configuration[:host], result[:response_time] * 1000, result[:request]]
  puts

  route = result[:route]
  puts "Reported Route:"
  puts
  case route.size
  when 5
    display_federated_route(route)
  when 3
    display_unfederated_route(route)
  end

  puts

  puts "Federation Brokers Instances:"
  puts
  if choria.federated?
    extract_federation_brokers(result[:route], configuration[:host]).sort.uniq.each do |broker|
      puts "  %s" % broker
    end
  else
    puts "  Unfederated"
  end
  puts

  puts "Known Federation Broker Clusters:"
  puts
  if choria.federated?
    puts "  %s" % choria.federation_collectives.join(", ")
  else
    puts "  Unfederated"
  end
end

#trace_node(node) ⇒ Object

Publish a specially crafted 'ping' with seen-by headers embedded, this signals to the entire collective to record their route



26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
 
# File 'lib/mcollective/application/federation.rb', line 26

def trace_node(node)
  options[:filter] = Util.empty_filter

  request = Message.new("ping", nil,
                        :agent => "discovery",
                        :filter => options[:filter],
                        :collective => options[:collective],
                        :type => :request,
                        :headers => {"seen-by" => []},
                        :options => options)

  request.discovered_hosts = [node]

  client = Client.new(options)
  found_time = 0
  route = []

  stats = client.req(request) do |reply, message|
    abort("Tracing requests requires MCollective 2.10.3 or newer") unless message

    found_time = Time.now.to_f

    abort("Received a response from %s while expecting a response from %s" % [reply[:senderid], node]) unless reply[:senderid] == node

    route = message.headers["seen-by"]
  end

  raise("Did not receive any responses to trace request") if stats[:responses] == 0

  {:route => route, :stats => stats, :response_time => (found_time - stats[:starttime]), :request => request.requestid}
end

#valid_commands ⇒ Array<String>

List of valid commands this application respond to

Returns:



232
233
234
 
# File 'lib/mcollective/application/federation.rb', line 232

def valid_commands
  methods.grep(/_command$/).map {|c| c.to_s.gsub("_command", "")}
end

#validate_configuration(configuration) ⇒ Object 



215
216
217
218
219
220
221
 
# File 'lib/mcollective/application/federation.rb', line 215

def validate_configuration(configuration)
  Util.loadclass("MCollective::Util::Choria")

  abort("Unknown command %s, valid commands are: %s" % [configuration[:command], valid_commands.join(", ")]) unless valid_commands.include?(configuration[:command])

  abort("A certificate is needed from the Puppet CA for `%s`, please use the `choria enroll` command" % choria.certname) unless choria.has_client_public_cert?
end