Class: Floe::ContainerRunner::Docker

Inherits:
Runner
  • Object
show all
Includes:
DockerMixin
Defined in:
lib/floe/container_runner/docker.rb

Direct Known Subclasses

Podman

Constant Summary collapse

DOCKER_COMMAND = 
"docker"

Constants included from DockerMixin

Floe::ContainerRunner::DockerMixin::MAX_CONTAINER_NAME_SIZE

Constants inherited from Runner

Runner::OUTPUT_MARKER

Instance Method Summary collapse

Methods included from DockerMixin

#container_name, #image_name

Methods inherited from Runner

for_resource, register_scheme

Constructor Details

#initialize(options = {}) ⇒ Docker

Returns a new instance of Docker.



10
11
12
13
14
15
16
17
18
19
 
# File 'lib/floe/container_runner/docker.rb', line 10

def initialize(options = {})
  require "awesome_spawn"
  require "io/wait"
  require "tempfile"

  super

  @network     = options.fetch("network", "bridge")
  @pull_policy = options["pull-policy"]
end

Instance Method Details

#cleanup(runner_context) ⇒ Object 



43
44
45
46
47
48
 
# File 'lib/floe/container_runner/docker.rb', line 43

def cleanup(runner_context)
  container_id, secrets_file = runner_context.values_at("container_ref", "secrets_ref")

  delete_container(container_id) if container_id
  delete_secret(secrets_file)    if secrets_file
end

#output(runner_context) ⇒ Object 



122
123
124
125
126
127
 
# File 'lib/floe/container_runner/docker.rb', line 122

def output(runner_context)
  return runner_context.slice("Error", "Cause") if runner_context.key?("Error")

  output = docker!("logs", runner_context["container_ref"], :combined_output => true).output
  runner_context["output"] = output
end

#run_async!(resource, env, secrets, context, volumes: [], entrypoint: nil, command: nil) ⇒ Object

Raises:

  • (ArgumentError) 


21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
 
# File 'lib/floe/container_runner/docker.rb', line 21

def run_async!(resource, env, secrets, context, volumes: [], entrypoint: nil, command: nil)
  raise ArgumentError, "Invalid resource" unless resource&.start_with?("docker://")
  raise ArgumentError, "entrypoint must be a String" if entrypoint && !entrypoint.kind_of?(String)
  raise ArgumentError, "command must be an Array"    if command && !command.kind_of?(Array)

  image          = resource.sub("docker://", "")
  execution_id   = context.execution["Id"]
  runner_context = {}

  if secrets && !secrets.empty?
    runner_context["secrets_ref"] = create_secret(secrets)
  end

  begin
    runner_context["container_ref"] = run_container(image, env, execution_id, runner_context["secrets_ref"], context.logger, volumes, entrypoint, command)
    runner_context
  rescue AwesomeSpawn::CommandResultError => err
    cleanup(runner_context)
    {"Error" => "States.TaskFailed", "Cause" => err.to_s}
  end
end

#running?(runner_context) ⇒ Boolean

Returns:

  • (Boolean) 


110
111
112
113
114
 
# File 'lib/floe/container_runner/docker.rb', line 110

def running?(runner_context)
  return false if runner_context.key?("Error")

  !!runner_context.dig("container_state", "Running")
end

#status!(runner_context) ⇒ Object 



104
105
106
107
108
 
# File 'lib/floe/container_runner/docker.rb', line 104

def status!(runner_context)
  return if runner_context.key?("Error")

  runner_context["container_state"] = inspect_container(runner_context["container_ref"])&.dig("State")
end

#success?(runner_context) ⇒ Boolean

Returns:

  • (Boolean) 


116
117
118
119
120
 
# File 'lib/floe/container_runner/docker.rb', line 116

def success?(runner_context)
  return false if runner_context.key?("Error")

  runner_context.dig("container_state", "ExitCode") == 0
end

#wait(timeout: nil, events: %i[create update delete],, &block) ⇒ Object 



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
 
# File 'lib/floe/container_runner/docker.rb', line 50

def wait(timeout: nil, events: %i[create update delete], &block)
  until_timestamp = (Time.now.utc + timeout).to_i if timeout

  r, w = IO.pipe

  pid = Kernel.spawn({}, self.class::DOCKER_COMMAND, *wait_params(until_timestamp), :err => :out, :out => w)
  Process.detach(pid)

  w.close

  loop do
    readable_timeout = until_timestamp - Time.now.utc if until_timestamp

    # Wait for our end of the pipe to be readable and if it didn't timeout
    # get the events from stdout
    next if r.wait_readable(readable_timeout).nil?

    # Get all events while the pipe is readable
    notices = []
    while r.wait_readable(0)
      notice = r.gets

      # If the process has exited `r.gets` returns `nil` and the pipe is
      # always `ready?`
      break if notice.nil?

      event, runner_context = parse_notice(notice)
      next if event.nil? || !events.include?(event)

      notices << [event, runner_context]
    end

    # If we're given a block yield the events otherwise return them
    if block
      notices.each(&block)
    else
      # Terminate the `docker events` process before returning the events
      sigterm(pid)

      return notices
    end

    # Check that the `docker events` process is still alive
    Process.kill(0, pid)
  rescue Errno::ESRCH
    # Break out of the loop if the `docker events` process has exited
    pid = nil
    break
  end
ensure
  r.close
  sigterm(pid) if pid
end