v3: Create service instance handles last operation errors

[#171930482](https://www.pivotaltracker.com/story/show/171930482)

Co-authored-by: George Blue <gblue@pivotal.io>
Co-authored-by: Derik Evangelista <devangelista@pivotal.io>

Author: Derik Evangelista

app/jobs/v3/services/fetch_last_operation_job.rb

@@ -44,6 +44,10 @@ def perform
         last_operation_result = client.fetch_service_instance_last_operation(service_instance)
         update_with_attributes(last_operation_result[:last_operation], service_instance, intended_operation)
 
+        if last_operation_result[:last_operation][:state] == 'failed'
+          operation_failed!(last_operation_result[:last_operation][:description])
+        end
+
         @retry_after = last_operation_result[:retry_after]
       rescue HttpRequestError, HttpResponseError, Sequel::Error => e
         logger = Steno.logger('cc-background')
@@ -66,6 +70,10 @@ def aborted!
         raise CloudController::Errors::ApiError.new_from_details('UnableToPerform', 'Create', 'delete in progress')
       end
 
+      def operation_failed!(msg)
+        raise CloudController::Errors::ApiError.new_from_details('ServiceInstanceProvisionFailed', msg)
+      end
+
       def pollable_job
         PollableJobModel.where(guid: @pollable_job_guid)
       end
@@ -100,6 +108,7 @@ def end_timestamp_reached
             description: 'Service Broker failed to provision within the required time.',
           }
         )
+        raise CloudController::Errors::ApiError.new_from_details('JobTimeout')
       end
 
       def apply_proposed_changes(service_instance)

lib/services/service_brokers/v2/client.rb

@@ -256,12 +256,10 @@ def fetch_service_instance_last_operation(instance)
       parsed_response     = @response_parser.parse_fetch_state(path, response)
       last_operation_hash = parsed_response.delete('last_operation') || {}
 
-      state = extract_state(instance, last_operation_hash)
-
       result = {
         last_operation:
           {
-            state: state
+            state: extract_state(instance, last_operation_hash)
           }
       }
 

lib/services/service_brokers/v2/response_parser.rb

@@ -182,19 +182,25 @@ def parse_fetch_state(path, response)
           validator =
             case unvalidated_response.code
             when 200
-              JsonObjectValidator.new(@logger,
-                StateValidator.new(['succeeded', 'failed', 'in progress'],
-                  SuccessValidator.new))
+              JsonObjectValidator.new(
+                @logger,
+                StateValidator.new(
+                  ['succeeded', 'failed', 'in progress'],
+                  SuccessValidator.new
+                )
+              )
             when 201, 202
               JsonObjectValidator.new(@logger,
-                FailingValidator.new(Errors::ServiceBrokerBadResponse))
+                FailingValidator.new(Errors::ServiceBrokerBadResponse)
+              )
+            when 400
+              BadRequestValidator.new
             when 410
               SuccessValidator.new { |res| {} }
             else
-              FailingValidator.new(Errors::ServiceBrokerBadResponse)
+              CommonErrorValidator.new(FailingValidator.new(Errors::ServiceBrokerBadResponse))
             end
 
-          validator = CommonErrorValidator.new(validator)
           validator.validate(unvalidated_response.to_hash)
         end
 
@@ -698,6 +704,24 @@ def remove_trailing_validation_schema_id(err_msg)
             err_msg.sub(/ in schema.*$/, '')
           end
         end
+
+        class BadRequestValidator
+          def validate(method:, uri:, code:, response:)
+            description = 'Bad request'
+            begin
+              parsed_response = MultiJson.load(response.body)
+              description = parsed_response['description'] if parsed_response.is_a?(Hash) && parsed_response.key?('description')
+            rescue MultiJson::ParseError
+            end
+
+            {
+              'last_operation' => {
+                'state' => 'failed',
+                'description' => description
+              }
+            }
+          end
+        end
       end
     end
   end

spec/request/service_instances_spec.rb

@@ -1119,6 +1119,37 @@ def check_filtered_instances(*instances)
               expect(instance.last_operation.state).to eq('succeeded')
             end
           end
+
+          context 'when last operation eventually returns `create failed`' do
+            before do
+              stub_request(:get, "#{instance.service_broker.broker_url}/v2/service_instances/#{instance.guid}/last_operation").
+                with(
+                  query: {
+                    operation: 'task12',
+                    service_id: service_plan.service.unique_id,
+                    plan_id: service_plan.unique_id,
+                  }).
+                to_return(status: last_operation_status_code, body: last_operation_response.to_json, headers: {}).times(1).then.
+                to_return(status: 200, body: { state: 'failed' }.to_json, headers: {})
+
+              execute_all_jobs(expected_successes: 2, expected_failures: 0)
+              expect(job.state).to eq(VCAP::CloudController::PollableJobModel::POLLING_STATE)
+
+              Timecop.freeze(Time.now + 1.hour) do
+                execute_all_jobs(expected_successes: 0, expected_failures: 1)
+              end
+            end
+
+            it 'completes the job' do
+              updated_job = VCAP::CloudController::PollableJobModel.find(guid: job.guid)
+              expect(updated_job.state).to eq(VCAP::CloudController::PollableJobModel::FAILED_STATE)
+            end
+
+            it 'sets the service instance last operation to create failed' do
+              expect(instance.last_operation.type).to eq('create')
+              expect(instance.last_operation.state).to eq('failed')
+            end
+          end
         end
       end
     end

spec/unit/jobs/v3/services/fetch_last_operation_job_spec.rb

@@ -206,7 +206,7 @@ def run_job(job, successes: 1, failures: 0)
           end
         end
 
-        context 'when the broker responds with state is `in progress`' do
+        context 'when the broker responds with state `in progress`' do
           let(:state) { 'in progress' }
 
           it 'fetches and updates the service instance state' do
@@ -248,6 +248,92 @@ def run_job(job, successes: 1, failures: 0)
           end
         end
 
+        context 'when the broker responds with state `failed`' do
+          let(:state) { 'failed' }
+          let(:description) { 'soz' }
+
+          it 'fetches and updates the service instance state' do
+            run_job(job, successes: 0, failures: 1)
+
+            db_service_instance = ManagedServiceInstance.first(guid: service_instance.guid)
+            expect(db_service_instance.last_operation.state).to eq('failed')
+          end
+
+          it 'fails with an appropriate message' do
+            run_job(job, successes: 0, failures: 1)
+
+            pollable_job.reload
+
+            expect(pollable_job.state).to eq(PollableJobModel::FAILED_STATE)
+            expect(pollable_job.cf_api_error).
+              to include('The service broker reported an error during provisioning: soz')
+          end
+
+          it 'should not enqueue another fetch job' do
+            run_job(job, successes: 0, failures: 1)
+            expect(Delayed::Job.all).to have(1).jobs
+            expect(Delayed::Job.where(failed_at: nil).all).to have(0).jobs
+          end
+
+          it 'does not apply the instance attributes that were proposed in the operation' do
+            run_job(job, successes: 0, failures: 1)
+
+            db_service_instance = ManagedServiceInstance.first(guid: service_instance.guid)
+            expect(db_service_instance.service_plan).to_not eq(proposed_service_plan)
+            expect(db_service_instance.name).to eq(service_instance.name)
+          end
+
+          it 'should not create an audit event' do
+            run_job(job, successes: 0, failures: 1)
+
+            expect(Event.find(type: 'audit.service_instance.create')).to be_nil
+          end
+        end
+
+        context 'when the job has fetched for more than the max poll duration' do
+          let(:state) { 'in progress' }
+
+          before do
+            run_job(job)
+            Timecop.travel(Time.now + max_duration.minutes + 1.minute) do
+              execute_all_jobs(expected_successes: 0, expected_failures: 1)
+            end
+          end
+
+          it 'should not enqueue another fetch job' do
+            Timecop.freeze(Time.now + max_duration.minutes + 1.minute) do
+              execute_all_jobs(expected_successes: 0, expected_failures: 0)
+            end
+          end
+
+          it 'should mark the service instance operation as failed' do
+            service_instance.reload
+
+            expect(service_instance.last_operation.state).to eq('failed')
+            expect(service_instance.last_operation.description).to eq('Service Broker failed to provision within the required time.')
+          end
+
+          it 'fails the job with an appropriate message' do
+            pollable_job.reload
+
+            expect(pollable_job.state).to eq(PollableJobModel::FAILED_STATE)
+            expect(pollable_job.cf_api_error).
+              to include('The job execution has timed out.')
+          end
+        end
+
+        context 'when enqueuing the job would exceed the max poll duration by the time it runs' do
+          let(:state) { 'in progress' }
+
+          it 'should not enqueue another fetch job' do
+            Timecop.freeze(job.end_timestamp - (job.poll_interval * 0.5))
+            run_job(job, failures: 1, successes: 0)
+
+            Timecop.freeze(Time.now + job.poll_interval * 2)
+            execute_all_jobs(expected_successes: 0, expected_failures: 0)
+          end
+        end
+
         context 'failures' do
           context 'when fetching the last operation from the broker fails' do
             context 'due to an HttpRequestError' do
@@ -407,43 +493,6 @@ def run_job(job, successes: 1, failures: 0)
             end
           end
 
-          context 'when the last operation type is `create`' do
-            before do
-              service_instance.save_with_new_operation({}, { type: 'create' })
-            end
-
-            context 'when during create, a delete operation was started' do
-              before do
-                allow(client).to receive(:fetch_service_instance_last_operation) do
-                  service_instance.save_with_new_operation(
-                    {},
-                    {
-                      type: 'delete',
-                      state: 'in progress'
-                    }
-                  )
-
-                  last_operation_response
-                end
-
-                run_job(job)
-              end
-
-              let(:db_service_instance) { ManagedServiceInstance.first(guid: service_instance.guid) }
-
-              it 'does not finish creating' do
-                expect(db_service_instance.last_operation.type).to eq('delete')
-                expect(db_service_instance.last_operation.state).to eq('in progress')
-                expect(Event.find(type: 'audit.service_instance.create')).not_to be
-              end
-
-              it 'does not delete' do
-                expect(db_service_instance).not_to be_nil
-                expect(Event.find(type: 'audit.service_instance.delete')).not_to be
-              end
-            end
-          end
-
           context 'when the last operation type is `update`' do
             before do
               service_instance.last_operation.type = 'update'
@@ -485,101 +534,6 @@ def run_job(job, successes: 1, failures: 0)
           end
         end
 
-        context 'when the state is `failed`' do
-          pending('do not run')
-          let(:state) { 'failed' }
-
-          it 'does not apply the instance attributes that were proposed in the operation' do
-            run_job(job)
-
-            db_service_instance = ManagedServiceInstance.first(guid: service_instance.guid)
-            expect(db_service_instance.service_plan).to_not eq(proposed_service_plan)
-            expect(db_service_instance.name).to eq(service_instance.name)
-          end
-
-          it 'fetches and updates the service instance state' do
-            run_job(job)
-
-            db_service_instance = ManagedServiceInstance.first(guid: service_instance.guid)
-            expect(db_service_instance.last_operation.state).to eq('failed')
-          end
-
-          it 'should not enqueue another fetch job' do
-            run_job(job)
-
-            expect(Delayed::Job.count).to eq 0
-          end
-
-          it 'should not create an audit event' do
-            run_job(job)
-
-            expect(Event.find(type: 'audit.service_instance.create')).to be_nil
-          end
-        end
-
-        context 'when the job has fetched for more than the max poll duration' do
-          let(:state) { 'in progress' }
-
-          before do
-            run_job(job)
-            Timecop.travel(Time.now + max_duration.minutes + 1.minute) do
-              execute_all_jobs(expected_successes: 1, expected_failures: 0)
-            end
-          end
-
-          it 'should not enqueue another fetch job' do
-            Timecop.freeze(Time.now + max_duration.minutes + 1.minute) do
-              execute_all_jobs(expected_successes: 0, expected_failures: 0)
-            end
-          end
-
-          it 'should mark the service instance operation as failed' do
-            service_instance.reload
-
-            expect(service_instance.last_operation.state).to eq('failed')
-            expect(service_instance.last_operation.description).to eq('Service Broker failed to provision within the required time.')
-          end
-        end
-
-        context 'when enqueuing the job would exceed the max poll duration by the time it runs' do
-          let(:state) { 'in progress' }
-
-          it 'should not enqueue another fetch job' do
-            Timecop.freeze(job.end_timestamp - (job.poll_interval * 0.5))
-            run_job(job)
-
-            Timecop.freeze(Time.now + job.poll_interval * 2)
-            execute_all_jobs(expected_successes: 0, expected_failures: 0)
-          end
-        end
-
-        context 'when the job was migrated before the addition of end_timestamp' do
-          let(:state) { 'in progress' }
-
-          it 'should compute the end_timestamp based on the current time' do
-            Timecop.freeze(Time.now)
-
-            run_job(job)
-
-            # should run enqueued job
-            Timecop.travel(Time.now + max_duration.minutes - 1.minute) do
-              execute_all_jobs(expected_successes: 1, expected_failures: 0)
-            end
-
-            # should not run enqueued job
-            Timecop.travel(Time.now + max_duration.minutes) do
-              execute_all_jobs(expected_successes: 0, expected_failures: 0)
-            end
-          end
-
-          it 'should enqueue another fetch job' do
-            run_job(job)
-
-            expect(Delayed::Job.count).to eq 1
-            expect(Delayed::Job.first).to be_a_fully_wrapped_job_of(FetchLastOperationJob)
-          end
-        end
-
         context 'when the poll_interval is changed after the job was created' do
           let(:default_polling_interval) { VCAP::CloudController::Config.config.get(:broker_client_default_async_poll_interval_seconds) }
           let(:new_polling_interval) { default_polling_interval * 2 }