# Copyright 2023 Google LLC # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. from .base import Processor, NotConfiguredException import google.auth from googleapiclient import discovery import time class ClouddeployProcessor(Processor): """ Operate Cloud Deploy applications, releases and other such things. Args: project (str, optional): Google Cloud project ID. name (str): Name of the delivery pipeline. region (str): Google Cloud Deploy reiog.n mode (str): One of: releases.get, releases.create, releases.rollouts.create, releases.rollouts.get, releases.rollouts.approve, releases.rollouts.reject """ def get_default_config_key(): return 'clouddeploy' def wait_for_operation_done(self, deploy_service, operation_name): end_time = start_time = time.monotonic() while True and (end_time - start_time) < 60: op_request = deploy_service.projects().locations().operations().get( name=operation_name).execute() if 'done' in op_request and op_request['done']: if 'error' in op_request: self.logger.error( 'Error while waiting for long running operation %s to complete.' % (operation_name), extra={'error': op_request['error']}) return op_request['error'] response = op_request['response'] del response['@type'] return response time.sleep(5) end_time = time.monotonic() def process(self, output_var='clouddeploy'): if 'region' not in self.config: raise NotConfiguredException('No Cloud Deploy region specified.') if 'name' not in self.config: raise NotConfiguredException( 'No Cloud Deploy delivery pipeline name specified.') if 'mode' not in self.config: raise NotConfiguredException('No Cloud Deploy operation specified.') credentials, credentials_project_id = google.auth.default() project = self._jinja_expand_string( self.config['project'], 'project') if 'project' in self.config else credentials_project_id if not project: project = credentials.quota_project_id region_name = self._jinja_expand_string(self.config['region'], 'region') pipeline_name = self._jinja_expand_string(self.config['name'], 'name') pipeline_full_name = "projects/%s/locations/%s/deliveryPipelines/%s" % ( project, region_name, pipeline_name) deploy_service = discovery.build( 'clouddeploy', 'v1', http=self._get_branded_http(credentials)) deploy_base_request = deploy_service.projects().locations( ).deliveryPipelines() if self.config['mode'] == 'pipelines.get': deploy_response = deploy_base_request.get( name=pipeline_full_name).execute() return { output_var: deploy_response, } if self.config['mode'] == 'releases.get': deploy_response = deploy_base_request.releases().get( name=pipeline_full_name).execute() return { output_var: deploy_response, } if self.config['mode'] == 'releases.create': if 'releaseId' not in self.config: raise NotConfiguredException( 'No releaseId parameter configured for releases.create.') release_id = self._jinja_expand_string(self.config['releaseId'], 'release_id') if 'release' not in self.config: raise NotConfiguredException( 'No release parameter configured for releases.create.') body = self._jinja_expand_dict_all(self.config['release'], 'release') deploy_response = deploy_base_request.releases().create( body=body, parent=pipeline_full_name, releaseId=release_id).execute() deploy_response = self.wait_for_operation_done( deploy_service, deploy_response['name']) return { output_var: deploy_response, } if self.config['mode'] == 'releases.rollouts.create': if 'releaseId' not in self.config: raise NotConfiguredException( 'No releaseId parameter configured for releases.rollouts.create.' ) if 'rolloutId' not in self.config: raise NotConfiguredException( 'No rolloutId parameter configured for releases.rollouts.create.' ) release_id = self._jinja_expand_string(self.config['releaseId'], 'release_id') rollout_id = self._jinja_expand_string(self.config['rolloutId'], 'rollout_id') if 'rollout' not in self.config: raise NotConfiguredException( 'No rollout parameter configured for releases.rollouts.create.' ) body = self._jinja_expand_dict_all(self.config['rollout'], 'rollout') if '/releases/' not in release_id: release_full_name = '%s/releases/%s' % (pipeline_full_name, release_id) else: release_full_name = release_id deploy_response = deploy_base_request.releases().rollouts().create( body=body, parent=release_full_name, rolloutId=rollout_id).execute() deploy_response = self.wait_for_operation_done( deploy_service, deploy_response['name']) return { output_var: deploy_response, } if self.config['mode'] == 'releases.rollouts.approve' or self.config[ 'mode'] == 'releases.rollouts.reject': if 'releaseId' not in self.config: raise NotConfiguredException( 'No releaseId parameter configured for releases.rollouts.approve/reject.' ) if 'rolloutId' not in self.config: raise NotConfiguredException( 'No rolloutId parameter configured for releases.rollouts.approve/reject.' ) release_id = self._jinja_expand_string(self.config['releaseId'], 'release_id') rollout_id = self._jinja_expand_string(self.config['rolloutId'], 'rollout_id') if self.config['mode'] == 'releases.rollouts.approve': body = {'approved': True} else: body = {'approved': False} if '/releases/' not in release_id: release_full_name = '%s/releases/%s' % (pipeline_full_name, release_id) else: release_full_name = release_id if '/rollouts/' in rollout_id: rollout_full_name = rollout_id else: rollout_full_name = '%s/rollouts/%s' % (release_full_name, rollout_id) deploy_response = deploy_base_request.releases().rollouts().approve( body=body, name=rollout_full_name).execute() return { output_var: deploy_response, } return { output_var: None, }