Unmanaged pods should fail fast when evicted/preempted/deleted

lib/kubernetes-deploy/deploy_task.rb

@@ -93,8 +93,6 @@ def server_version
       kubectl.server_version
     end
 
-    NOT_FOUND_ERROR = 'NotFound'
-
     def initialize(namespace:, context:, current_sha:, template_dir:, logger:, kubectl_instance: nil, bindings: {},
       max_watch_seconds: nil)
       @namespace = namespace
@@ -399,7 +397,7 @@ def deploy_resources(resources, prune: false, verify:, record_summary: true)
 
     def apply_all(resources, prune)
       return unless resources.present?
-      command = ["apply"]
+      command = %w(apply)
 
       Dir.mktmpdir do |tmp_dir|
         resources.each do |r|
@@ -500,7 +498,7 @@ def confirm_namespace_exists
       st, err = nil
       with_retries(2) do
         _, err, st = kubectl.run("get", "namespace", @namespace, use_namespace: false, log_failure: true)
-        st.success? || err.include?(NOT_FOUND_ERROR)
+        st.success? || err.include?(KubernetesDeploy::Kubectl::NOT_FOUND_ERROR_TEXT)
       end
       raise FatalDeploymentError, "Failed to find namespace. #{err}" unless st.success?
       @logger.info("Namespace #{@namespace} found")

lib/kubernetes-deploy/kubectl.rb

@@ -3,6 +3,9 @@
 module KubernetesDeploy
   class Kubectl
     DEFAULT_TIMEOUT = 30
+    NOT_FOUND_ERROR_TEXT = 'NotFound'
+
+    class ResourceNotFoundError < StandardError; end
 
     def initialize(namespace:, context:, logger:, log_failure_by_default:, default_timeout: DEFAULT_TIMEOUT,
       output_is_sensitive: false)
@@ -17,7 +20,7 @@ def initialize(namespace:, context:, logger:, log_failure_by_default:, default_t
       raise ArgumentError, "context is required" if context.blank?
     end
 
-    def run(*args, log_failure: nil, use_context: true, use_namespace: true)
+    def run(*args, log_failure: nil, use_context: true, use_namespace: true, raise_on_404: false)
       log_failure = @log_failure_by_default if log_failure.nil?
 
       args = args.unshift("kubectl")
@@ -29,10 +32,17 @@ def run(*args, log_failure: nil, use_context: true, use_namespace: true)
       out, err, st = Open3.capture3(*args)
       @logger.debug(out.shellescape) unless output_is_sensitive?
 
-      if !st.success? && log_failure
-        @logger.warn("The following command failed: #{Shellwords.join(args)}")
-        @logger.warn(err) unless output_is_sensitive?
+      unless st.success?
+        if log_failure
+          @logger.warn("The following command failed: #{Shellwords.join(args)}")
+          @logger.warn(err) unless output_is_sensitive?
+        end
+
+        if raise_on_404 && err.match(NOT_FOUND_ERROR_TEXT)
+          raise ResourceNotFoundError, err
+        end
       end
+
       [out.chomp, err.chomp, st]
     end
 

lib/kubernetes-deploy/kubernetes_resource.rb

@@ -85,6 +85,7 @@ def initialize(namespace:, context:, definition:, logger:, statsd_tags: [])
       @logger = logger
       @definition = definition
       @statsd_report_done = false
+      @disappeared = false
       @validation_errors = []
       @instance_data = {}
     end
@@ -121,12 +122,23 @@ def file_path
     end
 
     def sync(mediator)
-      @instance_data = mediator.get_instance(kubectl_resource_type, name)
+      @instance_data = mediator.get_instance(kubectl_resource_type, name, raise_on_404: true)
+    rescue KubernetesDeploy::Kubectl::ResourceNotFoundError
+      @disappeared = true if deploy_started?
+      @instance_data = {}
     end
 
     def after_sync
     end
 
+    def deleted?
+      @instance_data.dig('metadata', 'deletionTimestamp').present?
+    end
+
+    def disappeared?
+      @disappeared
+    end
+
     def deploy_failed?
       false
     end

lib/kubernetes-deploy/kubernetes_resource/pod.rb

@@ -69,9 +69,18 @@ def timeout_message
       header + probe_failure_msgs.join("\n") + "\n"
     end
 
+    def permanent_failed_phase?
+      return false unless phase == FAILED_PHASE_NAME
+      unmanaged? || !TRANSIENT_FAILURE_REASONS.include?(reason)
+    end
+
     def failure_message
-      if phase == FAILED_PHASE_NAME && !TRANSIENT_FAILURE_REASONS.include?(reason)
-        phase_problem = "Pod status: #{status}. "
+      phase_problem = if permanent_failed_phase?
+        "Pod status: #{status}. "
+      elsif unmanaged? && deleted?
+        "Pod status: Terminating. "
+      elsif unmanaged? && disappeared?
+        "Pod status: Disappeared. "
       end
 
       doomed_containers = @containers.select(&:doomed?)

lib/kubernetes-deploy/sync_mediator.rb

@@ -10,12 +10,16 @@ def initialize(namespace:, context:, logger:)
       clear_cache
     end
 
-    def get_instance(kind, resource_name)
-      if @cache.key?(kind)
-        @cache.dig(kind, resource_name) || {}
-      else
-        request_instance(kind, resource_name)
+    def get_instance(kind, resource_name, raise_on_404: false)
+      unless @cache.key?(kind)
+        return request_instance(kind, resource_name, raise_on_404: raise_on_404)
       end
+
+      cached_instance = @cache[kind].fetch(resource_name, {})
+      if cached_instance.blank? && raise_on_404
+        raise KubernetesDeploy::Kubectl::ResourceNotFoundError, "Resource does not exist (used cache for kind #{kind})"
+      end
+      cached_instance
     end
 
     def get_all(kind, selector = nil)
@@ -55,8 +59,8 @@ def clear_cache
       @cache = {}
     end
 
-    def request_instance(kind, iname)
-      raw_json, _, st = kubectl.run("get", kind, iname, "-a", "--output=json")
+    def request_instance(kind, iname, raise_on_404:)
+      raw_json, _err, st = kubectl.run("get", kind, iname, "-a", "--output=json", raise_on_404: raise_on_404)
       st.success? ? JSON.parse(raw_json) : {}
     end
 

test/integration/runner_task_test.rb

@@ -18,7 +18,7 @@ def test_run_without_verify_result_succeeds_as_soon_as_pod_is_successfully_creat
       "Result: SUCCESS",
       "Result verification is disabled for this task",
       "The following status was observed immediately after pod creation:",
-      %r{Pod/task-runner-\w+\s+Pending},
+      %r{Pod/task-runner-\w+\s+(Pending|Running)},
     ], in_order: true)
 
     pods = kubeclient.get_pods(namespace: @namespace)
@@ -37,7 +37,7 @@ def test_run_global_timeout_with_max_watch_seconds
       "Result: TIMED OUT",
       "Timed out waiting for 1 resource to run",
       %r{Pod/task-runner-\w+: GLOBAL WATCH TIMEOUT \(5 seconds\)},
-      "Final status: Running"
+      /Final status\: (Pending|Running)/
     ], in_order: true)
   end
 
@@ -88,6 +88,40 @@ def test_run_with_verify_result_success
     assert_equal task_runner.pod_name, pods.first.metadata.name, "Pod name should be available after run"
   end
 
+  def test_run_with_verify_result_fails_quickly_if_the_pod_is_deleted_out_of_band
+    deploy_task_template
+
+    task_runner = build_task_runner
+    deleter_thread = Thread.new do
+      loop do
+        if task_runner.pod_name.present?
+          begin
+            kubeclient.delete_pod(task_runner.pod_name, @namespace)
+            break
+          rescue Kubeclient::ResourceNotFoundError
+            sleep 0.1
+            retry
+          end
+        end
+        sleep 0.1
+      end
+    end
+
+    result = task_runner.run(run_params(log_lines: 20, log_interval: 1))
+    assert_task_run_failure(result)
+
+    assert_logs_match_all([
+      "Pod creation succeeded",
+      "Result: FAILURE",
+      /Pod status\: (Terminating|Disappeared)/,
+    ])
+  ensure
+    if deleter_thread
+      deleter_thread.join
+      deleter_thread.kill
+    end
+  end
+
   def test_run_with_verify_result_neither_misses_nor_duplicates_logs_across_pollings
     deploy_task_template
     task_runner = build_task_runner

test/test_helper.rb

@@ -194,13 +194,12 @@ def fixture_path(set_name)
       source_dir
     end
 
-    def stub_kubectl_response(*args, resp:, err: "", success: true, json: true, times: 1)
+    def stub_kubectl_response(*args, resp:, err: "", raise_on_404: nil, success: true, json: true, times: 1)
       resp = resp.to_json if json
       response = [resp, err, stub(success?: success)]
-      KubernetesDeploy::Kubectl.any_instance.expects(:run)
-        .with(*args)
-        .returns(response)
-        .times(times)
+      expectation = KubernetesDeploy::Kubectl.any_instance.expects(:run)
+      expectation = raise_on_404.nil? ? expectation.with(*args) : expectation.with(*args, raise_on_404: raise_on_404)
+      expectation.returns(response).times(times)
     end
 
     def build_runless_kubectl

test/unit/kubernetes-deploy/kubectl_test.rb

@@ -137,6 +137,22 @@ def test_version_info_raises_if_command_fails
     end
   end
 
+  def test_run_with_raise_err_on_404_raises_the_correct_thing
+    err = 'Error from server (NotFound): pods "foobar" not found'
+    stub_open3(%w(kubectl get pod foobar --namespace=testn --context=testc --request-timeout=30),
+      resp: "", err: err, success: false)
+    assert_raises_message(KubernetesDeploy::Kubectl::ResourceNotFoundError, err) do
+      build_kubectl.run("get", "pod", "foobar", raise_on_404: true)
+    end
+  end
+
+  def test_run_with_raise_err_on_404_does_not_raise_on_other_errors
+    err = 'Error from server (TooManyRequests): Please try again later'
+    stub_open3(%w(kubectl get pod foobar --namespace=testn --context=testc --request-timeout=30),
+      resp: "", err: err, success: false)
+    build_kubectl.run("get", "pod", "foobar", raise_on_404: true)
+  end
+
   private
 
   def stub_version_request(client:, server:)

test/unit/kubernetes-deploy/kubernetes_resource/daemon_set_test.rb

@@ -51,9 +51,8 @@ def build_ds_template(status: {})
   def build_synced_ds(template:)
     ds = KubernetesDeploy::DaemonSet.new(namespace: "test", context: "nope", logger: logger, definition: template)
     sync_mediator = build_sync_mediator
-    sync_mediator.kubectl.expects(:run).with("get", "DaemonSet", "ds-app", "-a", "--output=json").returns(
-      [template.to_json, "", SystemExit.new(0)]
-    )
+    sync_mediator.kubectl.expects(:run).with("get", "DaemonSet", "ds-app", "-a", "--output=json", raise_on_404: true)
+      .returns([template.to_json, "", SystemExit.new(0)])
 
     sync_mediator.kubectl.expects(:run).with("get", "Pod", "-a", "--output=json", anything).returns(
       ['{ "items": [] }', "", SystemExit.new(0)]

test/unit/kubernetes-deploy/kubernetes_resource/deployment_test.rb

@@ -367,9 +367,9 @@ def build_rs_template(status: { 'replicas' => 3 })
   def build_synced_deployment(template:, replica_sets:, server_version: Gem::Version.new("1.8"))
     deploy = KubernetesDeploy::Deployment.new(namespace: "test", context: "nope", logger: logger, definition: template)
     sync_mediator = build_sync_mediator
-    sync_mediator.kubectl.expects(:run).with("get", "Deployment", "web", "-a", "--output=json").returns(
-      [template.to_json, "", SystemExit.new(0)]
-    )
+    sync_mediator.kubectl.expects(:run)
+      .with("get", "Deployment", "web", "-a", "--output=json", raise_on_404: true)
+      .returns([template.to_json, "", SystemExit.new(0)])
     sync_mediator.kubectl.expects(:server_version).returns(server_version)
 
     if replica_sets.present?

test/unit/kubernetes-deploy/kubernetes_resource/pod_disruption_budget_test.rb

@@ -24,9 +24,9 @@ def build_synced_pdb(template:)
     pdb = KubernetesDeploy::PodDisruptionBudget.new(namespace: "test", context: "nope",
       logger: logger, definition: template)
     sync_mediator = KubernetesDeploy::SyncMediator.new(namespace: 'test', context: 'minikube', logger: logger)
-    sync_mediator.kubectl.expects(:run).with("get", "PodDisruptionBudget", "test", "-a", "--output=json").returns(
-      [template.to_json, "", SystemExit.new(0)]
-    )
+    sync_mediator.kubectl.expects(:run)
+      .with("get", "PodDisruptionBudget", "test", "-a", "--output=json", raise_on_404: true)
+      .returns([template.to_json, "", SystemExit.new(0)])
     pdb.sync(sync_mediator)
     pdb
   end