Set last-applied-label during create or replace operation atomically (#…

…972)

* Fix apply ordering of created and replaced resources

lib/krane/resource_deployer.rb

@@ -22,15 +22,15 @@ def initialize(task_config:, prune_allowlist:, global_timeout:, current_sha: nil
 
     def deploy!(resources, verify_result, prune)
       if verify_result
-        deploy_all_resources(resources, prune: prune, verify: true)
+        deploy_all_resources(resources, prune: prune, verify: true, annotate_individuals: true)
         failed_resources = resources.reject(&:deploy_succeeded?)
         success = failed_resources.empty?
         if !success && failed_resources.all?(&:deploy_timed_out?)
           raise DeploymentTimeoutError
         end
         raise FatalDeploymentError unless success
       else
-        deploy_all_resources(resources, prune: prune, verify: false)
+        deploy_all_resources(resources, prune: prune, verify: false, annotate_individuals: true)
         logger.summary.add_action("deployed #{resources.length} #{'resource'.pluralize(resources.length)}")
         warning = <<~MSG
           Deploy result verification is disabled for this deploy.
@@ -73,12 +73,13 @@ def predeploy_priority_resources(resource_list, predeploy_sequence)
 
     private
 
-    def deploy_all_resources(resources, prune: false, verify:, record_summary: true)
-      deploy_resources(resources, prune: prune, verify: verify, record_summary: record_summary)
+    def deploy_all_resources(resources, prune: false, verify:, record_summary: true, annotate_individuals: false)
+      deploy_resources(resources, prune: prune, verify: verify, record_summary: record_summary,
+      annotate_individuals: annotate_individuals)
     end
     measure_method(:deploy_all_resources, 'normal_resources.duration')
 
-    def deploy_resources(resources, prune: false, verify:, record_summary: true)
+    def deploy_resources(resources, prune: false, verify:, record_summary: true, annotate_individuals: false)
       return if resources.empty?
       deploy_started_at = Time.now.utc
 
@@ -96,7 +97,6 @@ def deploy_resources(resources, prune: false, verify:, record_summary: true)
       applyables, individuals = resources.partition { |r| r.deploy_method == :apply }
       # Prunable resources should also applied so that they can  be pruned
       pruneable_types = @prune_allowlist.map { |t| t.split("/").last }
-      applyables += individuals.select { |r| pruneable_types.include?(r.type) && !r.deploy_method_override }
 
       individuals.each do |individual_resource|
         individual_resource.deploy_started_at = Time.now.utc
@@ -122,6 +122,11 @@ def deploy_resources(resources, prune: false, verify:, record_summary: true)
 
       apply_all(applyables, prune)
 
+      if annotate_individuals
+        to_annotate = individuals.select { |r| pruneable_types.include?(r.type) && !r.deploy_method_override }
+        update_last_applied_annotations(to_annotate)
+      end
+
       if verify
         watcher = Krane::ResourceWatcher.new(resources: resources, deploy_started_at: deploy_started_at,
           timeout: @global_timeout, task_config: @task_config, sha: @current_sha)
@@ -252,6 +257,19 @@ def create_resource(resource)
       [err, status]
     end
 
+    def update_last_applied_annotations(resources)
+      resources.each do |resource|
+        err, status = set_last_applied_annotation(resource)
+        raise FatalDeploymentError, "Failed to set last applied annotation: #{err}" unless status.success?
+      end
+    end
+
+    def set_last_applied_annotation(resource)
+      _, err, status = kubectl.run("apply", "set-last-applied", "--create-annotation", "-f", resource.file_path, log_failure: false,
+        output_is_sensitive: resource.sensitive_template_content?, use_namespace: !resource.global?)
+      [err, status]
+    end
+
     # Inspect the file referenced in the kubectl stderr
     # to make it easier for developer to understand what's going on
     def find_bad_files_from_kubectl_output(line)