Class: Amazon

Inherits:
Infrastructure show all
Defined in:
lib/cluster/infrastructures/amazon.rb

Class Method Summary collapse

Instance Method Summary collapse

Methods inherited from Infrastructure

cluster_name, connect, #credentials, #credentials?, current, dns, in_cluster?, #initialize, #machines, #names, #services, sizes, #to_credentials

Methods included from Cluster::Logging

#logger

Constructor Details

This class inherits a constructor from Infrastructure

Class Method Details

.from_sdb_results(res) ⇒ Object 



588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
 
# File 'lib/cluster/infrastructures/amazon.rb', line 588

def from_sdb_results(res)
  res[:items].inject([]) {|m, obj|
    aws_id = obj.keys.first
    attrs = obj[aws_id]
    args = attrs.keys.inject({:aws_id => aws_id}) {|n, attr|
      val = attrs[attr]
      if val.empty?
        n.merge attr => nil
      elsif val.length == 1
        n.merge attr => val.first
      else
        n.merge attr => val
      end
    }
    
    m << args
  }
end

.to_sdb_attributes(args) ⇒ Object 



579
580
581
582
583
584
585
586
 
# File 'lib/cluster/infrastructures/amazon.rb', line 579

def to_sdb_attributes(args)
  attrs = {}
  # FIXME this needs a way to deal with arrays that are empty
  args.each do |k, v|
    v and attrs.merge!(k => Array(v))
  end
  attrs
end

Instance Method Details

#alter_instances!(*iss) ⇒ Object 



230
231
232
233
234
235
236
237
238
239
240
241
242
 
# File 'lib/cluster/infrastructures/amazon.rb', line 230

def alter_instances!(*iss)
  list = iss.empty? ? instances : iss

  for ins in list.flatten
    yield ins if block_given?
    attrs = ins.attributes
    remove = attrs.keys.select {|k| attrs[k].empty? and attrs.delete(k) }
    unless remove.empty?
      sdb.delete_attributes domain, ins.aws_id, remove
    end
    sdb.put_attributes domain, ins.aws_id, ins.attributes, :replace
  end
end

#authorize(ips) ⇒ Object 



562
563
564
565
566
 
# File 'lib/cluster/infrastructures/amazon.rb', line 562

def authorize(ips)
  ips.map {|ip|
    ecc.authorize_security_group_IP_ingress('access', 22, 22, 'tcp', ip) and ip
  }
end

#balance ⇒ Object 



370
371
372
373
374
 
# File 'lib/cluster/infrastructures/amazon.rb', line 370

def balance
  eid = current_instance.ec2_id

  elb.register_instances_with_load_balancer @options.elb_name, eid
end

#bucket ⇒ Object 



510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
 
# File 'lib/cluster/infrastructures/amazon.rb', line 510

def bucket
  unless @options.cluster_bucket
    puts "#{Cluster::NAME} has not been configured with a bucket for client materials."
    exit 2
  end

  @bucket ||= sss.bucket(@options.cluster_bucket, true)

  unless @bucket
    puts "#{Cluster::NAME} bucket named #{@options.cluster_bucket} cannot be created or accessed."
    exit 2
  else
    @bucket
  end
end

#clear_instances ⇒ Object 



107
108
109
 
# File 'lib/cluster/infrastructures/amazon.rb', line 107

def clear_instances
  @@instances = nil
end

#configure ⇒ Object 



5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
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
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
 
# File 'lib/cluster/infrastructures/amazon.rb', line 5

def configure
  super
  @@in_cluster = in_cluster?
  @@instances = nil

  if credentials? and credentials.include? 'amazon'
    creds = credentials['amazon']
    @options.key = creds['key']
    @options.secret = creds['secret']
    @options.owner = creds['owner']
    @options.cluster_bucket = creds['cluster_bucket']
    @options.bucket_key = creds['bucket_key']
    @options.cluster_domain = creds['cluster_domain']
    @options.zones = creds['zones']
    @options.volumes = creds['volumes']
    @options.elb_name = creds['web']['elb_name']
  end

  @options.key ||= ENV['AWS_ACCESS_KEY'] 
  @options.secret ||= ENV['AWS_SECRET_KEY'] 
  @options.owner ||= ENV['AMAZON_OWNER_ID'] 
  @options.cluster_bucket ||= ENV['CLUSTER_BUCKET']
  @options.bucket_key ||= 'cluster_credentials.yml'
  @options.zones ||= if ENV['AMAZON_ZONE'].empty?
                       nil
                     else
                       Array(ENV['AMAZON_ZONE'])
                     end

  @options.cluster_domain ||= ENV['CLUSTER_DOMAIN'] || self.class.cluster_name
  @options.elb_name ||= ENV['INGAMER_LOAD_BALANCER']
  @options.volumes ||= {}

  @options.role = (ENV['CLUSTER_ROLE'] or ENV['RAILS_ENV'] or 'production')

  @options.cluster_image_key = 'cluster_images.yml'
  @options.spot_instances = false
  @options.price = false

  OptionParser.new {|o|
    o.banner = "Amazon Infrastructure Options"

    o.on('-k', '--key VAL', "Amazon Access Key ID") do |v|
      @options.key = v
    end

    o.on('-s', '--secret VAL', 'Amazon Access Secret') do |v|
      @options.secret = v
    end

    o.on('-o', '--owenr VAL', 'Amazon User Code') do |v|
      @options.owner = v
    end

    o.on('-d', '--domain VAL', 'Amazon Domain to use') do |v|
      @options.cluster_domain = v
    end

    o.on('-b', '--bucket VAL', 'Cluster Bucket') do |v|
      @options.cluster_bucket = v
    end

    o.on('-f', '--bucket-credentials-file VAL', 'Cluster credentials file location on the bucket.') do |v|
      @options.bucket_key = v
    end

    o.on('--source-bucket VAL', 'Bucket that has the host data.') do |v|
    end

    o.on('-r', '--role VAL', 'Role in which to operate.') do |v|
      @options.role = v
    end

    o.on('-z', '--zone VAL', "Availability Zone") do |v|
      @options.zones = Array(v)
    end

    o.on('--spot', "Use Spot Instances") do |v|
      @options.spot_instances = true
    end

    o.on('--price=VAL', 'Maximum price for the spot instance') do |v|
      @options.price = v
    end

  }.parse(@arguments)

  unless @options.key and @options.secret
    $stderr.puts "Amazon Infrastructure cannot communicate without secret and key."
    exit 2
  end

  unless @options.cluster_domain
    $stderr.puts "Amazon Infrastructure needs to know what domain to connect to."
    exit 2
  end
end

#connect_to_active_sdb(o = nil) ⇒ Object 



382
383
384
385
386
 
# File 'lib/cluster/infrastructures/amazon.rb', line 382

def connect_to_active_sdb(o = nil)
  params = connection_params o
  params.delete :region
  RightAws::ActiveSdb.establish_connection key, secret, params
end

#connection_params(o = nil) ⇒ Object 



439
440
441
442
443
444
445
446
447
 
# File 'lib/cluster/infrastructures/amazon.rb', line 439

def connection_params(o = nil)
  params = o ? o.clone : {}
  params = {:multi_thread => true} if params.empty?

  params[:logger] ||= logger
  params[:region] ||= @options.zones.first

  params
end

#cost(sizes) ⇒ Object 



111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
 
# File 'lib/cluster/infrastructures/amazon.rb', line 111

def cost(sizes)
  unless @options.spot_instances
    puts "Cost only works for spot instances currently (ie. supply infrastructure argument of --spot"
    exit 3
  end
  args = {:start_time => (Time.now - (7 * 24 * 60 * 60)),
        :end_time => Time.now,
        :product_description => "Linux/UNIX"}
  sizes.map {|s|
    t = [size_to_instance_type(s)]
    prices = ecc.describe_spot_price_history(args.merge(:instance_types => t))
    p = prices.reduce(0) {|c,p| c + p[:spot_price] } / prices.length
    [s, "%0.3f" % p]
  }
end

#create_data_store(name) ⇒ Object 



131
132
133
 
# File 'lib/cluster/infrastructures/amazon.rb', line 131

def create_data_store(name)
  sdb.create_domain(name)
end

#create_file_store(name) ⇒ Object 



127
128
129
 
# File 'lib/cluster/infrastructures/amazon.rb', line 127

def create_file_store(name)
  sss.bucket(name, true)
end

#credentials_url(seconds = 1200) ⇒ Object 



467
468
469
470
 
# File 'lib/cluster/infrastructures/amazon.rb', line 467

def credentials_url(seconds = 1200)
  keygen = RightAws::S3Generator::Key.new(bucket, @options.bucket_key)
  keygen.get(seconds)
end

#current_instance ⇒ Object 



316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
 
# File 'lib/cluster/infrastructures/amazon.rb', line 316

def current_instance
  unless in_cluster?
    puts "Are we in the cluster?"
    exit 3
  end

  require 'open-uri'
  ec2_id = open('http://169.254.169.254/latest/meta-data/instance-id') {|f|  f.read }
  ins = self.instances.detect {|i| i.ec2_id.eql? ec2_id }

  unless ins
    puts "#{Cluster::NAME} cannot determine the current instance." 
    exit 2
  end

  if @options.current_instance_id
    aws = self.instances.detect {|i| i.aws_id.eql? @options.current_instance_id }
    if aws and aws.aws_id != ins.aws_id
      ins.services = (ins.services + aws.services).uniq
      ins.disabled_services = (ins.disabled_services + aws.disabled_services).uniq
      ins.spot_price = aws.spot_price unless ins.spot_price
      ins.friendly_name = aws.friendly_name unless ins.friendly_name
      sdb.delete_attributes domain, aws.aws_id
      sdb.put_attributes domain, ins.aws_id, ins.attributes, :replace
    end
  end

  ins
end

#domain ⇒ Object 



366
367
368
 
# File 'lib/cluster/infrastructures/amazon.rb', line 366

def domain
  @options.cluster_domain
end

#domains(reload = false) ⇒ Object 



412
413
414
415
416
417
418
 
# File 'lib/cluster/infrastructures/amazon.rb', line 412

def domains(reload = false)
  @sdb_domains = if !@sdb_domains or reload
                   sdb.list_domains[:domains]
                 else
                   @sdb_domains
                 end
end

#ecc(o = nil) ⇒ Object 



406
407
408
409
410
 
# File 'lib/cluster/infrastructures/amazon.rb', line 406

def ecc(o = nil)
  params = connection_params o
  @ec2 ||= {}
  @ec2[params[:region]] ||= RightAws::Ec2.new key, secret, params
end

#elb(o = nil) ⇒ Object 



400
401
402
403
404
 
# File 'lib/cluster/infrastructures/amazon.rb', line 400

def elb(o = nil)
  params = connection_params o
  @elb ||= {}
  @elb[params[:region]] ||= RightAws::ElbInterface.new(key, secret, params)
end

#fetch_monitor ⇒ Object 



309
310
311
312
313
314
 
# File 'lib/cluster/infrastructures/amazon.rb', line 309

def fetch_monitor
  res = sdb.select "select * from #{domain} where entry = 'monitor'"
  return nil if res[:items].empty?
  monitor = self.class.from_sdb_results(res).first
  bucket.get(monitor['key'])
end

#get_image(bits) ⇒ Object 



531
532
533
534
535
536
537
538
539
 
# File 'lib/cluster/infrastructures/amazon.rb', line 531

def get_image(bits)
  @image_file ||= get_image_file
  case bits
  when 32
    @image_file['thirtytwo']
  when 64
    @image_file['sixtyfour']
  end
end

#get_image_file ⇒ Object 



526
527
528
529
 
# File 'lib/cluster/infrastructures/amazon.rb', line 526

def get_image_file
  require 'open-uri'
  open(Cluster::IMAGES) {|f| YAML::load(f) }
end

#in_cluster? ⇒ Boolean

Returns:

  • (Boolean) 


244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
 
# File 'lib/cluster/infrastructures/amazon.rb', line 244

def in_cluster?
  return @cluster_check if @cluster_check

  check = false
  begin
    Timeout::timeout(1) do
      begin
        s = TCPSocket.new('169.254.169.254', 80)
        s.close
        check = true
      rescue Errno::ECONNREFUSED, Errno::EHOSTUNREACH
        # NOP
      end
    end
  rescue Timeout::Error
  end

  @@in_cluster = check
end

#instance_list ⇒ Object

def add_instance(*services)

  update_machines
  ins = current_instance
  ins.services = services
  sdb.put_attributes domain, ins.aws_id, ins.attributes, :replace
end
alias :add_machine :add_instance



163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
 
# File 'lib/cluster/infrastructures/amazon.rb', line 163

def instance_list
  terminated = []
  iss = @options.zones.map {|z|
    e = ecc region: z
    e.describe_instances.map {|ins|
      if %w(terminated shutting-down).include? ins[:aws_state]
        terminated << ins[:aws_instance_id]
        nil
      else
        AmazonInstance.new(ins)
      end
    }.compact
  }.flatten

  [iss, terminated]
end

#instances ⇒ Object 



103
104
105
 
# File 'lib/cluster/infrastructures/amazon.rb', line 103

def instances
  @@instances ||= load_instances
end

#key ⇒ Object

The methods below here are typically only used internally, but could also be called by anything that would like to access the Amazon tools directly.



354
355
356
 
# File 'lib/cluster/infrastructures/amazon.rb', line 354

def key
  @options.key
end

#load_instances ⇒ Object 



135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
 
# File 'lib/cluster/infrastructures/amazon.rb', line 135

def load_instances
  begin
    res = sdb.select "select * from #{domain} where entry='machine'"
  rescue RightAws::AwsError
    unless domains.include? domain
      sdb.create_domain domain
      retry
    end
  end
  sdbs = self.class.from_sdb_results res

  sdbs.map {|sd|
    ins = AmazonInstance.new
    ins.set_sdb_attributes sd
    ins
  }
end

#new_instance(size, services) ⇒ Object 



272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
 
# File 'lib/cluster/infrastructures/amazon.rb', line 272

def new_instance(size, services)
  args = { :services => services, :size => size }
  if @options.spot_instances
    unless @options.price
      puts "Amazon Spot instances need a '--price=??' argument"
      exit 3
    end
    args.merge! :spot_price => @options.price
  end
=begin
     type = size_to_type size
  groups = services_to_groups services
  image = type_to_image type
  puts "type #{type} Groups #{groups.inspect} K #{key} I #{image} UD #{user}"
=end

  ins = AmazonInstance.create args
  puts ins.inspect
  ins
end

#options ⇒ Object 



541
542
543
 
# File 'lib/cluster/infrastructures/amazon.rb', line 541

def options
  @options
end

#owner ⇒ Object 



362
363
364
 
# File 'lib/cluster/infrastructures/amazon.rb', line 362

def owner
  @options.owner
end

#period(args) ⇒ Object 



472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
 
# File 'lib/cluster/infrastructures/amazon.rb', line 472

def period(args)
  shots = {}
  ecc.describe_snapshots.each do |shot|
    next unless shot[:aws_owner].to_s == @options.owner.to_s
    next unless shot[:aws_status] == 'completed'
    shot[:started_at] = Time.parse shot[:aws_started_at]
    aid = shot[:aws_volume_id]
    if shots.include? aid
      shots[aid].push shot
    else
      shots.merge! aid => [shot]
    end
  end

  stores = @options.volumes['stores'] || []
  defaults = if @options.volumes.include?('defaults')
               @options.volumes['defaults']
             else
               {}
             end

  shots.keys.each do |k|
    opts = if v = stores.detect {|v| v['aws_id'] == k} 
             defaults.merge v
           else
             defaults
           end
    snap_count = opts['snaps'] || 2
    snap_count = 2 unless snap_count > 1

    ordered = shots[k].sort {|a,b| a[:started_at] <=> b[:started_at] }.reverse
    ordered.slice(snap_count, ordered.length).map {|shot|
      ecc.delete_snapshot shot[:aws_id]
      shot[:aws_id]
    }
  end
end

#query(qry) ⇒ Object 



568
569
570
571
572
573
574
575
 
# File 'lib/cluster/infrastructures/amazon.rb', line 568

def query(qry)
  res = sdb.select qry
  if res
    self.class.from_sdb_results res
  else
    nil
  end
end

#release_class ⇒ Object 



346
347
348
 
# File 'lib/cluster/infrastructures/amazon.rb', line 346

def release_class
  AmazonRelease
end

#retrieve(key) ⇒ Object 



453
454
455
 
# File 'lib/cluster/infrastructures/amazon.rb', line 453

def retrieve(key)
  bucket.get key
end

#revoke(ips) ⇒ Object 



556
557
558
559
560
 
# File 'lib/cluster/infrastructures/amazon.rb', line 556

def revoke(ips)
  ips.map {|ip|
    ecc.revoke_security_group_IP_ingress('access', 22, 22, 'tcp', ip) and ip
  }
end

#save_credentials(creds, key = nil) ⇒ Object 



457
458
459
460
 
# File 'lib/cluster/infrastructures/amazon.rb', line 457

def save_credentials(creds, key = nil)
  key ||= @options.bucket_key
  bucket.put(key, creds)
end

#save_images(input) ⇒ Object 



462
463
464
465
 
# File 'lib/cluster/infrastructures/amazon.rb', line 462

def save_images(input)
  key ||= @options.cluster_image_key
  bucket.put key, input, {}, 'public-read'
end

#save_monitor(io, key) ⇒ Object 



292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
 
# File 'lib/cluster/infrastructures/amazon.rb', line 292

def save_monitor(io, key)
  if bucket.put key, io
    old = sdb.select "select * from #{domain} where entry = 'monitor'"
    monitor = unless old[:items].empty?
                self.class.from_sdb_results(old).first
              else
                { 'aws_id' => UUIDTools::UUID.timestamp_create.to_s,
                  'entry' => 'monitor' }
              end
    monitor.merge! 'updated_at' => Time.now.xmlschema(3),
      'key' => key,
      'bucket' => @cluster_bucket
    sdb.put_attributes domain, monitor['aws_id'], self.class.to_sdb_attributes(monitor), :replace
    monitor
  end
end

#sdb(o = nil) ⇒ Object 



376
377
378
379
380
 
# File 'lib/cluster/infrastructures/amazon.rb', line 376

def sdb(o = nil)
  params = connection_params o
  params.delete :region
  @sdb ||= RightAws::SdbInterface.new(key, secret, params)
end

#secret ⇒ Object 



358
359
360
 
# File 'lib/cluster/infrastructures/amazon.rb', line 358

def secret
  @options.secret
end

#security(groups) ⇒ Object 



545
546
547
548
549
550
551
552
553
554
 
# File 'lib/cluster/infrastructures/amazon.rb', line 545

def security(groups)
  ecc.describe_security_groups(groups).inject({}) {|m,g|
    p = g[:aws_perms].map {|p|
      next unless p[:cidr_ips]
      [p[:cidr_ips], p[:from_port]..p[:to_port]]
    }.compact

    m.merge g[:aws_group_name] => p
  }
end

#size_to_instance_type(size) ⇒ Object 



264
265
266
267
268
269
270
 
# File 'lib/cluster/infrastructures/amazon.rb', line 264

def size_to_instance_type(size)
  unless self.class.sizes.include? size.downcase
    puts "#{Cluster::NAME} does not have a machine size of #{size}\n\tAvailable Sizes: (#{self.class.sizes.join(', ')})"
    exit 2
  end
  AmazonInstance.size_to_type(size)
end

#sqs(o = nil) ⇒ Object 



394
395
396
397
398
 
# File 'lib/cluster/infrastructures/amazon.rb', line 394

def sqs(o = nil)
  params = connection_params o
  @sqs ||= {}
  @sqs[params[:region]] ||= RightAws::ElbInterface.new(key, secret, params)
end

#sss(o = nil) ⇒ Object 



388
389
390
391
392
 
# File 'lib/cluster/infrastructures/amazon.rb', line 388

def sss(o = nil)
  params = connection_params o
  params.delete :region
  @sqs ||= RightAws::ElbInterface.new(key, secret, params)
end

#store(key, io) ⇒ Object 



449
450
451
 
# File 'lib/cluster/infrastructures/amazon.rb', line 449

def store(key, io)
  bucket.put(key, io)
end

#update_instances ⇒ Object Also known as: update_machines 



180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
 
# File 'lib/cluster/infrastructures/amazon.rb', line 180

def update_instances
  iss, terminated = instance_list

  begin
    res = sdb.select "select * from #{domain} where entry='machine'"
  rescue RightAws::AwsError
    unless domains.include? domain
      sdb.create_domain domain 
      retry
    end
  end
  sdbs = self.class.from_sdb_results res

  sdbs.each do |sd|
    aid = sd[:aws_id]
    iid = sd['ec2_id']
    if !iid
      started = sd['start_time_sorted'] && Time.parse(sd['start_time_sorted'])
      diff = started && (Time.now - started)
      if diff and diff < (12 * 3600)
        ins = AmazonInstance.new
        ins.set_sdb_attributes sd
        iss.push ins
      else
        $stderr.puts "Cannot find machine #{aid} -- old entry being removed."
        sdb.delete_attributes domain, aid
      end
    else
      if ins = iss.detect {|i| i.id.eql? iid }
        ins.set_sdb_attributes sd
      elsif terminated.include? iid
        $stderr.puts "Removing terminated entry #{iid}"
        sdb.delete_attributes domain, aid
      else
        $stderr.puts "Orphaned cluster record of #{aid}.  (Just started?) [#{sd.inspect}]"
      end
    end
  end

  iss.each do |ins|
    if ins.no_sdb?
      puts "Cannot find cluster registration for #{ins.ec2_id} -- creating."
    end
    sdb.put_attributes domain, ins.aws_id, ins.attributes, :replace
  end

  iss
end