v3(services): update managed service instance - async broker

update managed service instance when broker is async

Author: pivotal-marcela-campo

app/jobs/v3/update_service_instance_job.rb

@@ -1,15 +1,17 @@
-# require 'messages/service_instance_update_managed_message'
+require 'jobs/reoccurring_job'
 
 module VCAP::CloudController
   module V3
-    class UpdateServiceInstanceJob < VCAP::CloudController::Jobs::CCJob
+    class UpdateServiceInstanceJob < VCAP::CloudController::Jobs::ReoccurringJob
       attr_reader :warnings
 
       def initialize(service_instance_guid, message:, user_audit_info:)
         super()
         @service_instance_guid = service_instance_guid
         @message = message
         @user_audit_info = user_audit_info
+        @first_time = true
+        @broker_response = {}
         @warnings = []
       end
 
@@ -23,36 +25,25 @@ def perform
         aborted! if operation_in_progress != 'update'
 
         begin
-          compatibility_checks
-
-          client = VCAP::Services::ServiceClientProvider.provide({ instance: service_instance })
-
           maintenance_info = updated_maintenance_info(message, service_instance, service_plan)
+          client = VCAP::Services::ServiceClientProvider.provide({ instance: service_instance })
 
-          broker_response, err = client.update(
-            service_instance,
-            service_plan,
-            accepts_incomplete: false,
-            arbitrary_parameters: message.parameters || {},
-            previous_values: previous_values,
-            maintenance_info: maintenance_info,
-            name: message.requested?(:name) ? message.name : service_instance.name,
-          )
-          raise err if err
-
-          updates = message.updates.tap do |u|
-            u[:service_plan_guid] = service_plan.guid
-            u[:dashboard_url] = broker_response[:dashboard_url] if broker_response.key?(:dashboard_url)
-            u[:maintenance_info] = maintenance_info if maintenance_info_updated?(message, service_instance, service_plan)
+          if first_time
+            compute_maximum_duration
+            compatibility_checks
+            send_update_request(client, maintenance_info)
+            @first_time = false
           end
 
-          ServiceInstance.db.transaction do
-            service_instance.save_with_new_operation(updates, broker_response[:last_operation])
-            MetadataUpdate.update(service_instance, message)
-            record_event(service_instance, message.audit_hash)
+          if service_instance.operation_in_progress?
+            fetch_last_operation(client)
           end
 
-          logger.info("Service instance update complete #{service_instance_guid}")
+          if service_instance.last_operation.state == 'succeeded'
+            update_service_instance(broker_response, maintenance_info)
+            logger.info("Service instance update complete #{service_instance_guid}")
+            finish
+          end
         rescue => err
           logger.info("Service instance update failed: #{err.message}")
           service_instance.save_with_new_operation({}, {
@@ -62,6 +53,20 @@ def perform
           })
           raise err
         end
+
+        if service_instance.last_operation.state == 'failed'
+          logger.info("Service instance update failed: #{service_instance.last_operation.description}")
+          operation_failed!(service_instance.last_operation.description)
+        end
+      end
+
+      def handle_timeout
+        service_instance.save_and_update_operation(
+          last_operation: {
+            state: 'failed',
+            description: 'Service Broker failed to update within the required time.',
+          }
+        )
       end
 
       def job_name_in_configuration
@@ -86,7 +91,22 @@ def display_name
 
       private
 
-      attr_reader :service_instance_guid, :message, :user_audit_info
+      attr_reader :service_instance_guid, :message, :user_audit_info, :first_time, :broker_response
+
+      def send_update_request(client, maintenance_info)
+        @broker_response, err = client.update(
+          service_instance,
+          service_plan,
+          accepts_incomplete: true,
+          arbitrary_parameters: message.parameters || {},
+          previous_values: previous_values,
+          maintenance_info: maintenance_info,
+          name: message.requested?(:name) ? message.name : service_instance.name,
+        )
+        raise err if err
+
+        service_instance.save_with_new_operation({}, broker_response[:last_operation])
+      end
 
       def service_instance
         ManagedServiceInstance.first(guid: service_instance_guid)
@@ -145,6 +165,37 @@ def compatibility_checks
         end
       end
 
+      def compute_maximum_duration
+        max_poll_duration_on_plan = service_instance.service_plan.try(:maximum_polling_duration)
+        self.maximum_duration_seconds = max_poll_duration_on_plan if max_poll_duration_on_plan
+      end
+
+      def update_service_instance(broker_response, maintenance_info)
+        updates = message.updates.tap do |u|
+          u[:service_plan_guid] = service_plan.guid
+          u[:dashboard_url] = broker_response[:dashboard_url] if broker_response.key?(:dashboard_url)
+          u[:maintenance_info] = maintenance_info if maintenance_info_updated?(message, service_instance, service_plan)
+        end
+
+        ServiceInstance.db.transaction do
+          service_instance.update_service_instance(updates)
+          MetadataUpdate.update(service_instance, message)
+          record_event(service_instance, message.audit_hash)
+        end
+      end
+
+      def fetch_last_operation(client)
+        last_operation_result = client.fetch_service_instance_last_operation(service_instance)
+        self.polling_interval_seconds = last_operation_result[:retry_after] if last_operation_result[:retry_after]
+
+        service_instance.save_and_update_operation(
+          last_operation: last_operation_result[:last_operation].slice(:state, :description)
+        )
+      rescue HttpRequestError, HttpResponseError, Sequel::Error => e
+        logger = Steno.logger('cc-background')
+        logger.error("There was an error while fetching the service instance operation state: #{e}")
+      end
+
       def volume_services_disabled?
         !VCAP::CloudController::Config.config.get(:volume_services_enabled)
       end
@@ -164,6 +215,10 @@ def gone!
       def aborted!
         raise CloudController::Errors::ApiError.new_from_details('UnableToPerform', 'Update', 'delete in progress')
       end
+
+      def operation_failed!(msg)
+        raise CloudController::Errors::ApiError.new_from_details('UnableToPerform', 'Update', msg)
+      end
     end
   end
 end

docs/v3/source/includes/resources/service_instances/_update.md.erb

@@ -132,6 +132,7 @@ The response will be asynchronous if:
 
 * `parameters` are specified
 * `service_plan` is specified
+* `maintenance_info` is specified
 * `name` is specified and the service offering has the `allow_context_updates` feature enabled
 
 Otherwise the response will be synchronous.

spec/request/service_instances_spec.rb

@@ -1502,12 +1502,13 @@ def check_filtered_instances(*instances)
           )
         end
         let(:new_service_plan) { VCAP::CloudController::ServicePlan.make(service: service_offering) }
+        let(:original_maintenance_info) { { version: '1.1.0' } }
         let!(:service_instance) do
           si = VCAP::CloudController::ManagedServiceInstance.make(
             tags: %w(foo bar),
             space: space,
             service_plan: original_service_plan,
-            maintenance_info: { version: '1.1.0' }
+            maintenance_info: original_maintenance_info
           )
           si.annotation_ids = [
             VCAP::CloudController::ServiceInstanceAnnotationModel.make(key_prefix: 'pre.fix', key_name: 'to_delete', value: 'value').id,
@@ -1597,11 +1598,10 @@ def check_filtered_instances(*instances)
             api_call.call(space_dev_headers)
 
             instance = VCAP::CloudController::ServiceInstance.last
+
             stub_request(:patch, "#{instance.service_broker.broker_url}/v2/service_instances/#{instance.guid}").
+              with(query: { 'accepts_incomplete' => true }).
               to_return(status: broker_status_code, body: broker_response.to_json, headers: {})
-
-            # TODO: add this when doing async responses:
-            # with(query: { 'accepts_incomplete' => true }).
           end
 
           it 'sends a UPDATE request with the right arguments to the service broker' do
@@ -1610,8 +1610,7 @@ def check_filtered_instances(*instances)
             expect(
               a_request(:patch, "#{service_instance.service_broker.broker_url}/v2/service_instances/#{service_instance.guid}").
                 with(
-                  # TODO: add in when async part done:
-                  # query: { accepts_incomplete: true },
+                  query: { accepts_incomplete: true },
                   body: {
                     service_id: new_service_plan.service.unique_id,
                     plan_id: new_service_plan.unique_id,
@@ -1675,6 +1674,174 @@ def check_filtered_instances(*instances)
             end
           end
 
+          context 'when the update is asynchronous' do
+            let(:broker_status_code) { 202 }
+            let(:broker_response) { { operation: 'task12' } }
+            let(:last_operation_status_code) { 200 }
+            let(:last_operation_response) { { state: 'in progress' } }
+
+            before do
+              stub_request(:get, "#{service_instance.service_broker.broker_url}/v2/service_instances/#{service_instance.guid}/last_operation").
+                with(
+                  query: {
+                    operation: 'task12',
+                    service_id: service_instance.service_plan.service.unique_id,
+                    plan_id: service_instance.service_plan.unique_id,
+                  }).
+                to_return(status: last_operation_status_code, body: last_operation_response.to_json, headers: {})
+            end
+
+            it 'marks the job state as polling' do
+              execute_all_jobs(expected_successes: 1, expected_failures: 0)
+              expect(job.state).to eq(VCAP::CloudController::PollableJobModel::POLLING_STATE)
+            end
+
+            it 'calls last operation immediately' do
+              execute_all_jobs(expected_successes: 1, expected_failures: 0)
+              expect(
+                a_request(:get, "#{service_instance.service_broker.broker_url}/v2/service_instances/#{service_instance.guid}/last_operation").
+                  with(
+                    query: {
+                      operation: 'task12',
+                      service_id: service_instance.service_plan.service.unique_id,
+                      plan_id: service_instance.service_plan.unique_id,
+                    })
+              ).to have_been_made.once
+            end
+
+            it 'enqueues the next fetch last operation job' do
+              execute_all_jobs(expected_successes: 1, expected_failures: 0)
+              expect(Delayed::Job.count).to eq(1)
+            end
+
+            context 'when last operation eventually returns `update succeeded`' do
+              let(:last_operation_status_code) { 200 }
+              let(:last_operation_response) { { state: 'in progress' } }
+
+              before do
+                stub_request(:get, "#{service_instance.service_broker.broker_url}/v2/service_instances/#{service_instance.guid}/last_operation").
+                  with(
+                    query: {
+                      operation: 'task12',
+                      service_id: service_instance.service_plan.service.unique_id,
+                      plan_id: service_instance.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: 'succeeded' }.to_json, headers: {})
+
+                execute_all_jobs(expected_successes: 1, 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: 1, expected_failures: 0)
+                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::COMPLETE_STATE)
+              end
+
+              it 'sets the service instance last operation to create succeeded' do
+                expect(service_instance.last_operation.type).to eq('update')
+                expect(service_instance.last_operation.state).to eq('succeeded')
+              end
+            end
+
+            context 'when last operation eventually returns `update failed`' do
+              before do
+                stub_request(:get, "#{service_instance.service_broker.broker_url}/v2/service_instances/#{service_instance.guid}/last_operation").
+                  with(
+                    query: {
+                      operation: 'task12',
+                      service_id: service_instance.service_plan.service.unique_id,
+                      plan_id: service_instance.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: 1, 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, jobs_to_execute: 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 update failed' do
+                expect(service_instance.last_operation.type).to eq('update')
+                expect(service_instance.last_operation.state).to eq('failed')
+              end
+            end
+
+            context 'when last operation eventually returns error 400' do
+              before do
+                stub_request(:get, "#{service_instance.service_broker.broker_url}/v2/service_instances/#{service_instance.guid}/last_operation").
+                  with(
+                    query: {
+                      operation: 'task12',
+                      service_id: service_instance.service_plan.service.unique_id,
+                      plan_id: service_instance.service_plan.unique_id,
+                    }).
+                  to_return(status: last_operation_status_code, body: last_operation_response.to_json, headers: {}).times(1).then.
+                  to_return(status: 400, body: {}.to_json, headers: {})
+
+                execute_all_jobs(expected_successes: 1, 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, jobs_to_execute: 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 update failed' do
+                expect(service_instance.last_operation.type).to eq('update')
+                expect(service_instance.last_operation.state).to eq('failed')
+              end
+
+              it 'does not update the instance' do
+                #  TODO maybe look in the client to add this test and make sure what it returns? so we can test at a unit level in the job as well
+                service_instance.reload
+                expect(service_instance.reload.tags).to eq(%w(foo bar))
+                expect(service_instance.service_plan).to eq(original_service_plan)
+                expect_metadata(
+                  service_instance,
+                  annotations: [
+                    { prefix: 'pre.fix', key: 'to_delete', value: 'value' },
+                    { prefix: 'pre.fix', key: 'fox', value: 'bushy' },
+                  ],
+                  labels: [
+                    { prefix: 'pre.fix', key: 'to_delete', value: 'value' },
+                    { prefix: 'pre.fix', key: 'tail', value: 'fluffy' }
+                  ]
+                )
+              end
+
+              context 'when changing maintenance_info' do
+                let(:request_body) do
+                  {
+                    maintenance_info: { version: '1.1.1' },
+                  }
+                end
+
+                it 'does not update the instance' do
+                  service_instance.reload
+                  expect(service_instance.maintenance_info.symbolize_keys).to eq(original_maintenance_info)
+                end
+              end
+            end
+          end
+
           context 'changing maintenance_info alongside other parameters' do
             let(:new_maintenance_info) { { version: '1.1.1' } }
             let(:request_body) do