ecs.rb in hako-2.3.0

- old
+ new

@@ -11,10 +11,11 @@
 require 'hako/schedulers/ecs_autoscaling'
 require 'hako/schedulers/ecs_definition_comparator'
 require 'hako/schedulers/ecs_elb'
 require 'hako/schedulers/ecs_elb_v2'
 require 'hako/schedulers/ecs_service_comparator'
+require 'hako/schedulers/ecs_volume_comparator'
 
 module Hako
   module Schedulers
     class Ecs < Scheduler
       class NoTasksStarted < Error
@@ -43,11 +44,11 @@
           @ecs_elb_v2_options['target_type'] = 'ip'
         end
         @dynamic_port_mapping = options.fetch('dynamic_port_mapping', @ecs_elb_options.nil?)
         @health_check_grace_period_seconds = options.fetch('health_check_grace_period_seconds', nil)
         if options.key?('autoscaling')
-          @autoscaling = EcsAutoscaling.new(options.fetch('autoscaling'), @region, dry_run: @dry_run)
+          @autoscaling = EcsAutoscaling.new(options.fetch('autoscaling'), @region, ecs_elb_client, dry_run: @dry_run)
         end
         @autoscaling_group_for_oneshot = options.fetch('autoscaling_group_for_oneshot', nil)
         @autoscaling_topic_for_oneshot = options.fetch('autoscaling_topic_for_oneshot', nil)
         if @autoscaling_topic_for_oneshot && !@autoscaling_group_for_oneshot
           validation_error!('autoscaling_group_for_oneshot must be set when autoscaling_topic_for_oneshot is set')
@@ -92,10 +93,13 @@
         front_port = determine_front_port
         @scripts.each { |script| script.deploy_started(containers, front_port) }
         definitions = create_definitions(containers)
 
         if @dry_run
+          volumes_definition.each do |d|
+            print_volume_definition_in_cli_format(d)
+          end
           definitions.each do |d|
             print_definition_in_cli_format(d)
           end
           if @autoscaling
             @autoscaling.apply(Aws::ECS::Types::Service.new(cluster_arn: @cluster, service_name: @app_id))
@@ -174,10 +178,13 @@
         definitions.each do |definition|
           definition.delete(:essential)
         end
 
         if @dry_run
+          volumes_definition.each do |d|
+            print_volume_definition_in_cli_format(d)
+          end
           definitions.each do |d|
             if d[:name] == 'app'
               d[:command] = commands
             end
             print_definition_in_cli_format(d, additional_env: env)
@@ -403,21 +410,28 @@
         end
         unless actual_definition
           # Initial deployment
           return true
         end
+        actual_volume_definitions = {}
+        actual_definition.volumes.each do |v|
+          actual_volume_definitions[v.name] = v
+        end
         container_definitions = {}
         actual_definition.container_definitions.each do |c|
           container_definitions[c.name] = c
         end
 
         if actual_definition.task_role_arn != @task_role_arn
           return true
         end
-        if different_volumes?(actual_definition.volumes)
+        if volumes_definition.any? { |definition| different_volume?(definition, actual_volume_definitions.delete(definition[:name])) }
           return true
         end
+        unless actual_volume_definitions.empty?
+          return true
+        end
         if desired_definitions.any? { |definition| different_definition?(definition, container_definitions.delete(definition[:name])) }
           return true
         end
         unless container_definitions.empty?
           return true
@@ -439,27 +453,15 @@
         end
 
         false
       end
 
-      # @param [Hash<String, Hash<String, String>>] actual_volumes
+      # @param [Hash] expected_volume
+      # @param [Aws::ECS::Types::Volume] actual_volume
       # @return [Boolean]
-      def different_volumes?(actual_volumes)
-        if @volumes.size != actual_volumes.size
-          return true
-        end
-        actual_volumes.each do |actual_volume|
-          expected_volume = @volumes[actual_volume.name]
-          if expected_volume.nil?
-            return true
-          end
-          if expected_volume['source_path'] != actual_volume.host.source_path
-            return true
-          end
-        end
-
-        false
+      def different_volume?(expected_volume, actual_volume)
+        EcsVolumeComparator.new(expected_volume).different?(actual_volume)
       end
 
       # @param [Hash] expected_container
       # @param [Aws::ECS::Types::ContainerDefinition] actual_container
       # @return [Boolean]
@@ -532,17 +534,32 @@
           end
         end
         raise Error.new('Unable to register task definition for oneshot due to too many client errors')
       end
 
-      # @return [Hash]
+      # @return [Array<Hash>]
       def volumes_definition
-        @volumes.map do |name, volume|
-          {
-            name: name,
-            host: { source_path: volume['source_path'] },
-          }
+        @volumes_definition ||= @volumes.map do |name, volume|
+          definition = { name: name }
+          if volume.key?('docker_volume_configuration')
+            configuration = volume['docker_volume_configuration']
+            definition[:docker_volume_configuration] = {
+              autoprovision: configuration['autoprovision'],
+              driver: configuration['driver'],
+              # ECS API doesn't allow 'driver_opts' to be an empty hash.
+              driver_opts: configuration['driver_opts'],
+              # ECS API doesn't allow 'labels' to be an empty hash.
+              labels: configuration['labels'],
+              scope: configuration['scope'],
+            }
+          else
+            # When neither 'host' nor 'docker_volume_configuration' is
+            # specified, ECS API treats it as if 'host' is specified without
+            # 'source_path'.
+            definition[:host] = { source_path: volume['source_path'] }
+          end
+          definition
         end
       end
 
       def describe_task_definition(family)
         ecs_client.describe_task_definition(task_definition: family).task_definition
@@ -572,10 +589,11 @@
           privileged: container.privileged,
           linux_parameters: container.linux_parameters,
           volumes_from: container.volumes_from,
           user: container.user,
           log_configuration: container.log_configuration,
+          health_check: container.health_check,
           ulimits: container.ulimits,
           extra_hosts: container.extra_hosts,
         }
       end
 
@@ -1042,10 +1060,35 @@
           required_cpu <= cpu && required_memory <= memory
         end
       end
 
       # @param [Hash] definition
+      # @return [nil]
+      def print_volume_definition_in_cli_format(definition)
+        return if definition.dig(:docker_volume_configuration, :autoprovision)
+        # From version 1.20.0 of ECS agent, a local volume is provisioned when
+        # 'host' is specified without 'source_path'.
+        return if definition.dig(:host, :source_path)
+
+        cmd = %w[docker volume create]
+        if (configuration = definition[:docker_volume_configuration])
+          if configuration[:driver]
+            cmd << '--driver' << configuration[:driver]
+          end
+          (configuration[:driver_opts] || {}).each do |k, v|
+            cmd << '--opt' << "#{k}=#{v}"
+          end
+          (configuration[:labels] || {}).each do |k, v|
+            cmd << '--label' << "#{k}=#{v}"
+          end
+        end
+        cmd << definition[:name]
+        puts cmd.join(' ')
+        nil
+      end
+
+      # @param [Hash] definition
       # @param [Hash<String, String>] additional_env
       # @return [nil]
       def print_definition_in_cli_format(definition, additional_env: {})
         cmd = %w[docker run]
         cmd << '--name' << definition.fetch(:name)
@@ -1067,17 +1110,18 @@
             cmd << '--label' << "#{key}=#{val}"
           end
         end
         definition.fetch(:mount_points).each do |mount_point|
           source_volume = mount_point.fetch(:source_volume)
-          v = @volumes[source_volume]
-          if v
-            cmd << '--volume' << "#{v.fetch('source_path')}:#{mount_point.fetch(:container_path)}#{mount_point[:read_only] ? ':ro' : ''}"
-          else
-            raise "Could not find volume #{source_volume}"
-          end
+          v = volumes_definition.find { |d| d[:name] == source_volume }
+          raise "Could not find volume #{source_volume}" unless v
+          source = v.dig(:host, :source_path) || source_volume
+          cmd << '--volume' << "#{source}:#{mount_point.fetch(:container_path)}#{mount_point[:read_only] ? ':ro' : ''}"
         end
+        definition.fetch(:volumes_from).each do |volumes_from|
+          cmd << '--volumes-from' << "#{volumes_from.fetch(:source_container)}#{volumes_from[:read_only] ? ':ro' : ''}"
+        end
         if definition[:privileged]
           cmd << '--privileged'
         end
         if definition[:linux_parameters]
           if definition[:linux_parameters][:capabilities]
@@ -1101,11 +1145,20 @@
               end
               cmd << "--device=#{opts}"
             end
           end
 
-          if definition[:init_process_enabled]
+          if definition[:linux_parameters][:init_process_enabled]
             cmd << '--init'
+          end
+
+          if definition[:linux_parameters][:shared_memory_size]
+            cmd << '--shm-size' << "#{definition[:linux_parameters][:shared_memory_size]}m"
+          end
+
+          definition[:linux_parameters].fetch(:tmpfs, []).each do |tmpfs|
+            options = ["size=#{tmpfs[:size]}m"].concat(tmpfs[:mount_options])
+            cmd << '--tmpfs' << "#{tmpfs[:container_path]}:#{options.join(',')}"
           end
         end
         definition.fetch(:volumes_from).each do |volumes_from|
           p volumes_from
         end