Class: Asger::Runner

Inherits:
Object
  • Object
show all
Defined in:
lib/asger/runner.rb

Instance Method Summary collapse

Constructor Details

#initialize(logger:, aws_logger:, credentials:, region:, queue_url:, parameters:, task_files:, no_delete_messages:) ⇒ Runner

Returns a new instance of Runner.

Parameters:

  • logger (Logger) —

    the logger for Asger to use

  • sqs_client (Aws::SQS::Client) —

    the SQS client to use for polling

  • ec2_client (Aws::EC2::Client) —

    the EC2 client to use to get instance information

  • asg_client (Aws::AutoScaling::Client) —

    the ASG client to use to get ASG information

  • queue_url (String) —

    the queue URL to poll

  • parameters (Hash) —

    a hash of parameters to pass to Tasks

  • task_files (Array<String>) —

    list of file paths to load as Tasks

  • no_delete_messages (TrueClass, FalseClass) —

    if true, don't call sqs:DeleteMessage



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
 
# File 'lib/asger/runner.rb', line 18

def initialize(logger:, aws_logger:, credentials:,
               region:, queue_url:,
               parameters:, task_files:, no_delete_messages:)
  @logger = logger
  @region = region
  @parameters = parameters.merge(
    region: region, credentials: credentials
  ).deep_symbolize_keys

  @sqs_client = Aws::SQS::Client.new(logger: aws_logger,
    region: region, credentials: credentials)
  @ec2_client = Aws::EC2::Client.new(logger: aws_logger,
    region: region, credentials: credentials)
  @asg_client = Aws::AutoScaling::Client.new(logger: aws_logger,
    region: region, credentials: credentials)
  @ec2_resource_client = Aws::EC2::Resource.new(client: @ec2_client)
  @asg_resource_client = Aws::AutoScaling::Resource.new(client: @asg_client)
  @queue_url = queue_url
  @tasks = task_files.map { |tf| Task.from_file(@logger, tf) }
  @no_delete_messages = no_delete_messages

  @logger.info "#{@tasks.length} task(s) set up."
  @logger.warn('no_delete_messages is set; will not clear SQS messages!') \
    if @no_delete_messages

  @tasks.each { |t| t.invoke_init(@parameters) }
end

Instance Method Details

#delete_message(msg) ⇒ Object (private) 



107
108
109
110
 
# File 'lib/asger/runner.rb', line 107

def delete_message(msg)
  @logger.debug "Deleting message '#{msg[:receipt_handle]}'"
  @sqs_client.delete_message(queue_url: @queue_url, receipt_handle: msg[:receipt_handle])
end

#poll ⇒ Object 



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
102
103
 
# File 'lib/asger/runner.rb', line 47

def poll()
  poller = Aws::SQS::QueuePoller.new(@queue_url, client: @sqs_client,
    max_number_of_messages: 10, skip_delete: true)

  poller.poll do |msgs|
    [ msgs ].flatten.each do |msg|
      notification = JSON.parse(JSON.parse(msg.body)["Message"])
      if notification["Event"] != nil
        asg = @asg_resource_client.group(notification['AutoScalingGroupName'])
        instance_id = notification["EC2InstanceId"]

        @logger.warn("ASG '#{asg}' has fired event, but does not exist - already cleaned up?") \
          unless asg.exists?

        case notification["Event"].gsub("autoscaling:", "")
        when "EC2_INSTANCE_LAUNCH"
          @logger.info "Instance launched in '#{asg.name}': #{instance_id}"

          instance = @ec2_resource_client.instance(instance_id)
          @tasks.each do |task|
            task.invoke_up(instance, asg, @parameters)
          end

          delete_message(msg) unless @no_delete_messages
        when "EC2_INSTANCE_LAUNCH_ERROR"
          @logger.warn "Instance failed to launch in '#{asg.name}'."

          @tasks.each do |task|
            task.invoke_up_failed(asg, @parameters)
          end

          delete_message(msg) unless @no_delete_messages
        when "EC2_INSTANCE_TERMINATE"
          @logger.info "Instance terminated in '#{asg.name}': #{instance_id}"

          @tasks.reverse_each do |task|
            task.invoke_down(instance_id, asg, @parameters)
          end

          delete_message(msg) unless @no_delete_messages
        when "EC2_INSTANCE_TERMINATE_ERROR"
          @logger.warn "Instance failed to terminate in '#{asg.name}': #{instance_id}"

          @tasks.reverse_each do |task|
            task.invoke_down_failed(instance_id, asg, @parameters)
          end
          delete_message(msg) unless @no_delete_messages
        when "TEST_NOTIFICATION"
          @logger.info "Found test notification in queue."
          delete_message(msg) unless @no_delete_messages
        else
          @logger.debug "Unrecognized notification '#{notification["Event"]}', ignoring."
        end
      end
    end
  end
end