Cleanup job for operations stuck in create/initial state (#3742)

* check in short frequency for operations in state 'create'/'initial' where 'updated_at' is older than current time plus service broker connection timeout plus 10% safety margin
* set all matches to 'failed' so that users are able to delete the associated entity manually (service instance / service binding / key binding / route service binding)
* call orphan mitigation for each affected service

Author: jochenehret

README.md

@@ -114,13 +114,15 @@ When the Docker containers have been set up as described above, you can start th
 ```
 Then start a local worker:
 ```
-export CLOUD_CONTROLLER_NG_CONFIG=./tmp/cloud_controller.yml
-bundle exec rake jobs:local
+CLOUD_CONTROLLER_NG_CONFIG=./tmp/cloud_controller.yml bundle exec rake jobs:local
 ```
-And finally start a delayed_job worker:
+Start a delayed_job worker:
 ```
-export CLOUD_CONTROLLER_NG_CONFIG=./tmp/cloud_controller.yml
-bundle exec rake jobs:generic
+CLOUD_CONTROLLER_NG_CONFIG=./tmp/cloud_controller.yml bundle exec rake jobs:generic
+```
+And finally start the scheduler:
+```
+CLOUD_CONTROLLER_NG_CONFIG=./tmp/cloud_controller.yml bundle exec rake clock:start
 ```
 
 Known limitations:

app/jobs/runtime/service_operations_initial_cleanup.rb

@@ -0,0 +1,65 @@
+module VCAP::CloudController
+  module Jobs
+    module Runtime
+      class ServiceOperationsInitialCleanup < VCAP::CloudController::Jobs::CCJob
+        def perform
+          cleanup_operations(ServiceInstanceOperation, ServiceInstance, :service_instance_id, :cleanup_failed_provision)
+          cleanup_operations(ServiceBindingOperation, ServiceBinding, :service_binding_id, :cleanup_failed_bind)
+          cleanup_operations(ServiceKeyOperation, ServiceKey, :service_key_id, :cleanup_failed_key)
+          cleanup_operations(RouteBindingOperation, RouteBinding, :route_binding_id, nil)
+        end
+
+        def cleanup_operations(operation_model, instance_model, foreign_key, orphan_mitigator_method)
+          operations_create_initial = operation_model.
+                                      where(type: 'create', state: 'initial').
+                                      where(updated_at_past_broker_timeout).
+                                      select(:id, foreign_key)
+
+          return if operations_create_initial.empty?
+
+          operations_create_initial.each do |result|
+            instance = instance_model.first(id: result[foreign_key])
+            logger.info("#{instance_model.to_s.split('::').last} #{instance[:guid]} is stuck in state 'create'/'initial'. " \
+                        "Setting state to 'failed' and triggering orphan mitigation.")
+            operation_model.first(id: result[:id]).update(state: 'failed',
+                                                          description: 'Operation was stuck in "initial" state. Set to "failed" by cleanup job.')
+
+            orphan_mitigator.send(orphan_mitigator_method, instance) unless orphan_mitigator_method.nil?
+          end
+        end
+
+        def updated_at_past_broker_timeout
+          Sequel.lit(
+            "updated_at < ? - INTERVAL '?' SECOND",
+            Sequel::CURRENT_TIMESTAMP,
+            broker_client_timeout_plus_margin
+          )
+        end
+
+        def broker_client_timeout_plus_margin
+          (config.get(:broker_client_timeout_seconds).to_i * 1.1).round
+        end
+
+        def job_name_in_configuration
+          :service_operations_initial_cleanup
+        end
+
+        def max_attempts
+          1
+        end
+
+        def config
+          @config ||= Config.config
+        end
+
+        def orphan_mitigator
+          @orphan_mitigator ||= VCAP::Services::ServiceBrokers::V2::OrphanMitigator.new
+        end
+
+        def logger
+          @logger ||= Steno.logger('cc.background.service-operations-initial-cleanup')
+        end
+      end
+    end
+  end
+end

config/cloud_controller.yml

@@ -47,6 +47,9 @@ failed_jobs:
   cutoff_age_in_days: 31
   frequency_in_seconds: 144000 #4h
 
+service_operations_initial_cleanup:
+  frequency_in_seconds: 300
+
 completed_tasks:
   cutoff_age_in_days: 31
 

lib/cloud_controller/clock/scheduler.rb

@@ -22,7 +22,8 @@ class Scheduler
     FREQUENTS = [
       { name: 'pending_droplets', class: Jobs::Runtime::PendingDropletCleanup },
       { name: 'pending_builds', class: Jobs::Runtime::PendingBuildCleanup },
-      { name: 'failed_jobs', class: Jobs::Runtime::FailedJobsCleanup }
+      { name: 'failed_jobs', class: Jobs::Runtime::FailedJobsCleanup },
+      { name: 'service_operations_initial_cleanup', class: Jobs::Runtime::ServiceOperationsInitialCleanup }
     ].freeze
 
     def initialize(config)

lib/cloud_controller/config_schemas/base/clock_schema.rb

@@ -26,6 +26,9 @@ class ClockSchema < VCAP::Config
               optional(:max_number_of_failed_delayed_jobs) => Integer,
               frequency_in_seconds: Integer
             },
+            service_operations_initial_cleanup: {
+              frequency_in_seconds: Integer
+            },
             completed_tasks: {
               cutoff_age_in_days: Integer
             },

lib/cloud_controller/jobs.rb

@@ -26,6 +26,7 @@
 require 'jobs/runtime/expired_orphaned_blob_cleanup'
 require 'jobs/runtime/expired_resource_cleanup'
 require 'jobs/runtime/failed_jobs_cleanup'
+require 'jobs/runtime/service_operations_initial_cleanup'
 require 'jobs/runtime/legacy_jobs'
 require 'jobs/runtime/model_deletion'
 require 'jobs/runtime/orphaned_blobs_cleanup'

lib/tasks/jobs.rake

@@ -34,6 +34,7 @@ namespace :jobs do
       'app_usage_events',
       'audit_events',
       'failed_jobs',
+      'service_operations_initial_cleanup',
       'service_usage_events',
       'completed_tasks',
       'expired_blob_cleanup',

spec/unit/jobs/runtime/service_operations_initial_cleanup_spec.rb

@@ -0,0 +1,228 @@
+require 'spec_helper'
+
+module VCAP::CloudController
+  module Jobs::Runtime
+    RSpec.describe ServiceOperationsInitialCleanup, job_context: :worker do
+      subject(:job) { ServiceOperationsInitialCleanup.new }
+
+      let(:fake_orphan_mitigator) { double(VCAP::Services::ServiceBrokers::V2::OrphanMitigator) }
+      let(:fake_logger) { instance_double(Steno::Logger, info: nil) }
+
+      it { is_expected.to be_a_valid_job }
+
+      it 'knows its job name' do
+        expect(job.job_name_in_configuration).to equal(:service_operations_initial_cleanup)
+      end
+
+      it 'adds 10% to the broker_client_timeout' do
+        expect(job.broker_client_timeout_plus_margin).to equal(66)
+      end
+
+      def expect_no_orphan_mitigator_calls(orphan_mitigator)
+        %i[cleanup_failed_provision cleanup_failed_bind cleanup_failed_key].each do |call|
+          expect(orphan_mitigator).not_to have_received(call)
+        end
+      end
+
+      describe 'perform' do
+        let!(:service_instance) { ManagedServiceInstance.make }
+        let!(:service_binding) { ServiceBinding.make }
+        let!(:service_key) { ServiceKey.make }
+        let!(:route_binding) { RouteBinding.make }
+
+        before do
+          allow(Steno).to receive(:logger).and_return(fake_logger)
+
+          allow(VCAP::Services::ServiceBrokers::V2::OrphanMitigator).to receive(:new).and_return(fake_orphan_mitigator)
+          allow(fake_orphan_mitigator).to receive(:cleanup_failed_provision).with(service_instance)
+          allow(fake_orphan_mitigator).to receive(:cleanup_failed_bind).with(service_binding)
+          allow(fake_orphan_mitigator).to receive(:cleanup_failed_key).with(service_key)
+        end
+
+        context 'when there are no service instance operations in state create/initial' do
+          let!(:service_instance_operation) { ServiceInstanceOperation.make(service_instance_id: service_instance.id, type: 'create', state: 'succeeded') }
+
+          it 'does nothing' do
+            job.perform
+
+            expect(service_instance_operation.reload.type).to eq('create')
+            expect(service_instance_operation.reload.state).to eq('succeeded')
+            expect_no_orphan_mitigator_calls(fake_orphan_mitigator)
+          end
+        end
+
+        context 'when there is one service instance operation in state create/initial' do
+          let!(:service_instance_operation) { ServiceInstanceOperation.make(service_instance_id: service_instance.id, type: 'create', state: 'initial') }
+
+          context 'and the service broker connection timeout has not yet passed' do
+            it 'does nothing' do
+              job.perform
+
+              expect(service_instance_operation.reload.type).to eq('create')
+              expect(service_instance_operation.reload.state).to eq('initial')
+              expect_no_orphan_mitigator_calls(fake_orphan_mitigator)
+            end
+          end
+
+          context 'and the service broker connection timeout has passed' do
+            let(:expired_time) { Time.now.utc - 2.seconds }
+
+            before do
+              TestConfig.override(broker_client_timeout_seconds: 1)
+              service_instance_operation.this.update(updated_at: expired_time)
+            end
+
+            it 'sets the operation state to create/failed and triggers the orphan mitigation' do
+              job.perform
+
+              expect(service_instance_operation.reload.type).to eq('create')
+              expect(service_instance_operation.reload.state).to eq('failed')
+              expect(service_instance_operation.reload.description).to eq('Operation was stuck in "initial" state. Set to "failed" by cleanup job.')
+              expect(fake_logger).to have_received(:info).with("ServiceInstance #{service_instance[:guid]} is stuck in state 'create'/'initial'. " \
+                                                               "Setting state to 'failed' and triggering orphan mitigation.")
+              expect(fake_orphan_mitigator).to have_received(:cleanup_failed_provision).with(service_instance)
+            end
+          end
+        end
+
+        context 'when there are no service instance binding operations in state create/initial' do
+          let!(:service_binding_operation) { ServiceBindingOperation.make(service_binding_id: service_binding.id, type: 'create', state: 'succeeded') }
+
+          it 'does nothing' do
+            job.perform
+
+            expect(service_binding_operation.reload.type).to eq('create')
+            expect(service_binding_operation.reload.state).to eq('succeeded')
+            expect_no_orphan_mitigator_calls(fake_orphan_mitigator)
+          end
+        end
+
+        context 'when there is one service instance binding operation in state create/initial' do
+          let!(:service_binding_operation) { ServiceBindingOperation.make(service_binding_id: service_binding.id, type: 'create', state: 'initial') }
+
+          context 'and the service broker connection timeout has not yet passed' do
+            it 'does nothing' do
+              job.perform
+
+              expect(service_binding_operation.reload.type).to eq('create')
+              expect(service_binding_operation.reload.state).to eq('initial')
+              expect_no_orphan_mitigator_calls(fake_orphan_mitigator)
+            end
+          end
+
+          context 'and the service broker connection timeout has passed' do
+            let(:expired_time) { Time.now.utc - 2.seconds }
+
+            before do
+              TestConfig.override(broker_client_timeout_seconds: 1)
+              service_binding_operation.this.update(updated_at: expired_time)
+            end
+
+            it 'sets the operation state to create/failed and triggers the orphan mitigation' do
+              job.perform
+
+              expect(service_binding_operation.reload.type).to eq('create')
+              expect(service_binding_operation.reload.state).to eq('failed')
+              expect(service_binding_operation.reload.description).to eq('Operation was stuck in "initial" state. Set to "failed" by cleanup job.')
+              expect(fake_logger).to have_received(:info).with("ServiceBinding #{service_binding[:guid]} is stuck in state 'create'/'initial'. " \
+                                                               "Setting state to 'failed' and triggering orphan mitigation.")
+              expect(fake_orphan_mitigator).to have_received(:cleanup_failed_bind).with(service_binding)
+            end
+          end
+        end
+
+        context 'when there are no service key operations in state create/initial' do
+          let!(:service_key_operation) { ServiceKeyOperation.make(service_key_id: service_key.id, type: 'create', state: 'succeeded') }
+
+          it 'does nothing' do
+            job.perform
+
+            expect(service_key_operation.reload.type).to eq('create')
+            expect(service_key_operation.reload.state).to eq('succeeded')
+            expect_no_orphan_mitigator_calls(fake_orphan_mitigator)
+          end
+        end
+
+        context 'when there is one service key operation in state create/initial' do
+          let!(:service_key_operation) { ServiceKeyOperation.make(service_key_id: service_key.id, type: 'create', state: 'initial') }
+
+          context 'and the service broker connection timeout has not yet passed' do
+            it 'does nothing' do
+              job.perform
+
+              expect(service_key_operation.reload.type).to eq('create')
+              expect(service_key_operation.reload.state).to eq('initial')
+              expect_no_orphan_mitigator_calls(fake_orphan_mitigator)
+            end
+          end
+
+          context 'and the service broker connection timeout has passed' do
+            let(:expired_time) { Time.now.utc - 2.seconds }
+
+            before do
+              TestConfig.override(broker_client_timeout_seconds: 1)
+              service_key_operation.this.update(updated_at: expired_time)
+            end
+
+            it 'sets the operation state to create/failed and triggers the orphan mitigation' do
+              job.perform
+
+              expect(service_key_operation.reload.type).to eq('create')
+              expect(service_key_operation.reload.state).to eq('failed')
+              expect(service_key_operation.reload.description).to eq('Operation was stuck in "initial" state. Set to "failed" by cleanup job.')
+              expect(fake_logger).to have_received(:info).with("ServiceKey #{service_key[:guid]} is stuck in state 'create'/'initial'. " \
+                                                               "Setting state to 'failed' and triggering orphan mitigation.")
+              expect(fake_orphan_mitigator).to have_received(:cleanup_failed_key).with(service_key)
+            end
+          end
+        end
+
+        context 'when there are no route binding operations in state create/initial' do
+          let!(:route_binding_operation) { RouteBindingOperation.make(route_binding_id: route_binding.id, type: 'create', state: 'succeeded') }
+
+          it 'does nothing' do
+            job.perform
+
+            expect(route_binding_operation.reload.type).to eq('create')
+            expect(route_binding_operation.reload.state).to eq('succeeded')
+            expect_no_orphan_mitigator_calls(fake_orphan_mitigator)
+          end
+        end
+
+        context 'when there is one route binding operation in state create/initial' do
+          let!(:route_binding_operation) { RouteBindingOperation.make(route_binding_id: route_binding.id, type: 'create', state: 'initial') }
+
+          context 'and the service broker connection timeout has not yet passed' do
+            it 'does nothing' do
+              job.perform
+
+              expect(route_binding_operation.reload.type).to eq('create')
+              expect(route_binding_operation.reload.state).to eq('initial')
+              expect_no_orphan_mitigator_calls(fake_orphan_mitigator)
+            end
+          end
+
+          context 'and the service broker connection timeout has passed' do
+            let(:expired_time) { Time.now.utc - 2.seconds }
+
+            before do
+              TestConfig.override(broker_client_timeout_seconds: 1)
+              route_binding_operation.this.update(updated_at: expired_time)
+            end
+
+            it 'sets the operation state to create/failed and triggers the orphan mitigation' do
+              job.perform
+
+              expect(route_binding_operation.reload.type).to eq('create')
+              expect(route_binding_operation.reload.state).to eq('failed')
+              expect(route_binding_operation.reload.description).to eq('Operation was stuck in "initial" state. Set to "failed" by cleanup job.')
+              expect(fake_logger).to have_received(:info).with("RouteBinding #{route_binding[:guid]} is stuck in state 'create'/'initial'. " \
+                                                               "Setting state to 'failed' and triggering orphan mitigation.")
+              expect_no_orphan_mitigator_calls(fake_orphan_mitigator)
+            end
+          end
+        end
+      end
+    end
+  end
+end

spec/unit/lib/cloud_controller/clock/scheduler_spec.rb

@@ -19,6 +19,7 @@ module VCAP::CloudController
           app_usage_events: { cutoff_age_in_days: 1 },
           audit_events: { cutoff_age_in_days: 3 },
           failed_jobs: { frequency_in_seconds: 400, cutoff_age_in_days: 4, max_number_of_failed_delayed_jobs: 10 },
+          service_operations_initial_cleanup: { frequency_in_seconds: 600 },
           service_usage_events: { cutoff_age_in_days: 5 },
           completed_tasks: { cutoff_age_in_days: 6 },
           pending_droplets: { frequency_in_seconds: 300, expiration_in_seconds: 600 },
@@ -153,6 +154,12 @@ module VCAP::CloudController
           expect(block.call).to be_instance_of(Jobs::Runtime::FailedJobsCleanup)
         end
 
+        expect(clock).to receive(:schedule_frequent_worker_job) do |args, &block|
+          expect(args).to eql(name: 'service_operations_initial_cleanup', interval: 600)
+          expect(Jobs::Runtime::ServiceOperationsInitialCleanup).to receive(:new).and_call_original
+          expect(block.call).to be_instance_of(Jobs::Runtime::ServiceOperationsInitialCleanup)
+        end
+
         schedule.start
       end