Class: Wrapbox::Runner::Ecs

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

Defined Under Namespace

Classes: Cli, ContainerAbnormalEnd, ExecutionFailure, ExecutionTimeout, LackResource, LaunchFailure, Parameter

Constant Summary collapse

EXECUTION_RETRY_INTERVAL = 
3
WAIT_DELAY = 
5
TERM_TIMEOUT = 
120
HOST_TERMINATED_REASON_REGEXP = 
/Host EC2.*terminated/

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(options) ⇒ Ecs

Returns a new instance of Ecs.



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
 
# File 'lib/wrapbox/runner/ecs.rb', line 47

def initialize(options)
  @name = options[:name]
  @revision = options[:revision]
  @cluster = options[:cluster]
  @region = options[:region]
  @volumes = options[:volumes]
  @placement_constraints = options[:placement_constraints]
  @placement_strategy = options[:placement_strategy]
  @launch_type = options[:launch_type]
  @requires_compatibilities = options[:requires_compatibilities]
  @network_mode = options[:network_mode]
  @network_configuration = options[:network_configuration]
  @cpu = options[:cpu]
  @memory = options[:memory]

  @container_definitions = options[:container_definition] ? [options[:container_definition]] : options[:container_definitions] || []
  @container_definitions.concat(options[:additional_container_definitions]) if options[:additional_container_definitions] # deprecated

  if !@container_definitions.empty? && options[:task_definition]
    raise "Please set only one of `container_definition` and `task_definition`"
  end

  if options[:additional_container_definitions] && !options[:additional_container_definitions].empty?
    warn "`additional_container_definitions` is deprecated parameter, Use `container_definitions` instead of it"
  end

  @task_definition_info = options[:task_definition]

  if !@container_definitions.empty?
    @task_definition_name = "wrapbox_#{@name}"
    @main_container_name = @container_definitions[0][:name] || @task_definition_name
  elsif @task_definition_info
    @task_definition_name = @task_definition_info[:task_definition_name]
    @main_container_name = @task_definition_info[:main_container_name]
    unless @main_container_name
      raise "Please set `task_definition[:main_container_name]`"
    end
  end

  @container_definitions.each do |d|
    d[:docker_labels]&.stringify_keys!
    d.dig(:log_configuration, :options)&.stringify_keys!
  end

  @task_role_arn = options[:task_role_arn]
  $stdout.sync = true
  @logger = Logger.new($stdout)
  if options[:log_fetcher]
    type = options[:log_fetcher].delete(:type)
    @log_fetcher = LogFetcher.new(type, options[:log_fetcher])
  end
end

Instance Attribute Details

#cluster ⇒ Object (readonly)

Returns the value of attribute cluster.



28
29
30
 
# File 'lib/wrapbox/runner/ecs.rb', line 28

def cluster
  @cluster
end

#container_definitions ⇒ Object (readonly)

Returns the value of attribute container_definitions.



28
29
30
 
# File 'lib/wrapbox/runner/ecs.rb', line 28

def container_definitions
  @container_definitions
end

#cpu ⇒ Object (readonly)

Returns the value of attribute cpu.



28
29
30
 
# File 'lib/wrapbox/runner/ecs.rb', line 28

def cpu
  @cpu
end

#launch_type ⇒ Object (readonly)

Returns the value of attribute launch_type.



28
29
30
 
# File 'lib/wrapbox/runner/ecs.rb', line 28

def launch_type
  @launch_type
end

#main_container_name ⇒ Object (readonly)

Returns the value of attribute main_container_name.



28
29
30
 
# File 'lib/wrapbox/runner/ecs.rb', line 28

def main_container_name
  @main_container_name
end

#memory ⇒ Object (readonly)

Returns the value of attribute memory.



28
29
30
 
# File 'lib/wrapbox/runner/ecs.rb', line 28

def memory
  @memory
end

#name ⇒ Object (readonly)

Returns the value of attribute name.



28
29
30
 
# File 'lib/wrapbox/runner/ecs.rb', line 28

def name
  @name
end

#network_configuration ⇒ Object (readonly)

Returns the value of attribute network_configuration.



28
29
30
 
# File 'lib/wrapbox/runner/ecs.rb', line 28

def network_configuration
  @network_configuration
end

#network_mode ⇒ Object (readonly)

Returns the value of attribute network_mode.



28
29
30
 
# File 'lib/wrapbox/runner/ecs.rb', line 28

def network_mode
  @network_mode
end

#placement_constraints ⇒ Object (readonly)

Returns the value of attribute placement_constraints.



28
29
30
 
# File 'lib/wrapbox/runner/ecs.rb', line 28

def placement_constraints
  @placement_constraints
end

#placement_strategy ⇒ Object (readonly)

Returns the value of attribute placement_strategy.



28
29
30
 
# File 'lib/wrapbox/runner/ecs.rb', line 28

def placement_strategy
  @placement_strategy
end

#region ⇒ Object (readonly)

Returns the value of attribute region.



28
29
30
 
# File 'lib/wrapbox/runner/ecs.rb', line 28

def region
  @region
end

#requires_compatibilities ⇒ Object (readonly)

Returns the value of attribute requires_compatibilities.



28
29
30
 
# File 'lib/wrapbox/runner/ecs.rb', line 28

def requires_compatibilities
  @requires_compatibilities
end

#revision ⇒ Object (readonly)

Returns the value of attribute revision.



28
29
30
 
# File 'lib/wrapbox/runner/ecs.rb', line 28

def revision
  @revision
end

#task_definition_name ⇒ Object (readonly)

Returns the value of attribute task_definition_name.



28
29
30
 
# File 'lib/wrapbox/runner/ecs.rb', line 28

def task_definition_name
  @task_definition_name
end

#task_role_arn ⇒ Object (readonly)

Returns the value of attribute task_role_arn.



28
29
30
 
# File 'lib/wrapbox/runner/ecs.rb', line 28

def task_role_arn
  @task_role_arn
end

#volumes ⇒ Object (readonly)

Returns the value of attribute volumes.



28
29
30
 
# File 'lib/wrapbox/runner/ecs.rb', line 28

def volumes
  @volumes
end

Instance Method Details

#run(class_name, method_name, args, container_definition_overrides: {}, **parameters) ⇒ Object 



121
122
123
124
125
126
127
128
129
130
 
# File 'lib/wrapbox/runner/ecs.rb', line 121

def run(class_name, method_name, args, container_definition_overrides: {}, **parameters)
  task_definition = prepare_task_definition(container_definition_overrides)
  parameter = Parameter.new(**parameters)

  run_task(
    task_definition.task_definition_arn, class_name, method_name, args,
    ["bundle", "exec", "rake", "wrapbox:run"],
    parameter
  )
end

#run_cmd(cmds, container_definition_overrides: {}, ignore_signal: false, **parameters) ⇒ Object 



132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
 
# File 'lib/wrapbox/runner/ecs.rb', line 132

def run_cmd(cmds, container_definition_overrides: {}, ignore_signal: false, **parameters)
  ths = []

  task_definition = prepare_task_definition(container_definition_overrides)

  cmds << nil if cmds.empty?
  cmds.each_with_index do |cmd, idx|
    ths << Thread.new(cmd, idx) do |c, i|
      Thread.current[:cmd_index] = i
      envs = (parameters[:environments] || []) + [{name: "WRAPBOX_CMD_INDEX", value: i.to_s}]
      run_task(
        task_definition.task_definition_arn, nil, nil, nil,
        c&.shellsplit,
        Parameter.new(**parameters.merge(environments: envs))
      )
    end
  end
  ths.each(&:join)

  true
rescue SignalException => e
  sig = "SIG#{Signal.signame(e.signo)}"
  if ignore_signal
    @logger.info("Receive #{sig} signal. But ECS Tasks continue running")
  else
    @logger.info("Receive #{sig} signal. Stop All tasks")
    ths.each do |th|
      th.report_on_exception = false
      th.raise(e)
    end
    wait_until = Time.now + TERM_TIMEOUT + 15 # thread_timeout_buffer
    ths.each do |th|
      wait = wait_until - Time.now
      th.join(wait) if wait.positive?
    end
  end
  nil
end