Source code for heat.engine.resources.signal_responder

#
#    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 urllib import parse

from keystoneclient.contrib.ec2 import utils as ec2_utils
from oslo_config import cfg
from oslo_log import log as logging
from oslo_serialization import jsonutils
from oslo_utils import timeutils

from heat.common import exception
from heat.common.i18n import _
from heat.common import password_gen
from heat.engine.clients.os import swift
from heat.engine.resources import stack_user

LOG = logging.getLogger(__name__)

SIGNAL_TYPES = (
    WAITCONDITION, SIGNAL
) = (
    '/waitcondition', '/signal'
)
SIGNAL_VERB = {WAITCONDITION: 'PUT',
               SIGNAL: 'POST'}


[docs]class SignalResponder(stack_user.StackUser): PROPERTIES = ( SIGNAL_TRANSPORT, ) = ( 'signal_transport', ) ATTRIBUTES = ( SIGNAL_ATTR, ) = ( 'signal', ) # Anything which subclasses this may trigger authenticated # API operations as a consequence of handling a signal requires_deferred_auth = True
[docs] def handle_delete(self): self._delete_signals() return super(SignalResponder, self).handle_delete()
def _delete_signals(self): self._delete_ec2_signed_url() self._delete_heat_signal_url() self._delete_swift_signal_url() self._delete_zaqar_signal_queue() @property def password(self): return self.data().get('password') @password.setter def password(self, password): if password is None: self.data_delete('password') else: self.data_set('password', password, True) def _signal_transport_cfn(self): return self.properties[ self.SIGNAL_TRANSPORT] == self.CFN_SIGNAL def _signal_transport_heat(self): return self.properties[ self.SIGNAL_TRANSPORT] == self.HEAT_SIGNAL def _signal_transport_none(self): return self.properties[ self.SIGNAL_TRANSPORT] == self.NO_SIGNAL def _signal_transport_temp_url(self): return self.properties[ self.SIGNAL_TRANSPORT] == self.TEMP_URL_SIGNAL def _signal_transport_zaqar(self): return self.properties.get( self.SIGNAL_TRANSPORT) == self.ZAQAR_SIGNAL def _get_region_name(self): return self.client_plugin('heat')._get_region_name() def _get_heat_signal_credentials(self): """Return OpenStack credentials that can be used to send a signal. These credentials are for the user associated with this resource in the heat stack user domain. """ if self._get_user_id() is None: if self.password is None: self.password = password_gen.generate_openstack_password() self._create_user() return {'auth_url': self.keystone().server_keystone_endpoint_url( fallback_endpoint=self.keystone().v3_endpoint), 'username': self.physical_resource_name(), 'user_id': self._get_user_id(), 'password': self.password, 'project_id': self.stack.stack_user_project_id, 'domain_id': self.keystone().stack_domain_id, 'region_name': self._get_region_name()} def _get_ec2_signed_url(self, signal_type=SIGNAL, never_expire=False): """Create properly formatted and pre-signed URL. This uses the created user for the credentials. See boto/auth.py::QuerySignatureV2AuthHandler :param signal_type: either WAITCONDITION or SIGNAL. """ access_key = self.data().get('access_key') secret_key = self.data().get('secret_key') if not access_key or not secret_key: if self.id is None: # it is too early return if self._get_user_id() is None: self._create_user() self._create_keypair() access_key = self.data().get('access_key') secret_key = self.data().get('secret_key') if not access_key or not secret_key: LOG.warning('Cannot generate signed url, ' 'unable to create keypair') return config_url = cfg.CONF.heat_waitcondition_server_url if config_url: signal_url = config_url.replace('/waitcondition', signal_type) else: heat_client_plugin = self.stack.clients.client_plugin('heat') endpoint = heat_client_plugin.get_heat_cfn_url() signal_url = ''.join([endpoint, signal_type]) host_url = parse.urlparse(signal_url) path = self.identifier().arn_url_path() # Note the WSGI spec apparently means that the webob request we end up # processing in the CFN API (ec2token.py) has an unquoted path, so we # need to calculate the signature with the path component unquoted, but # ensure the actual URL contains the quoted version... unquoted_path = parse.unquote(host_url.path + path) params = {'SignatureMethod': 'HmacSHA256', 'SignatureVersion': '2', 'AWSAccessKeyId': access_key} if not never_expire: params['Timestamp'] = timeutils.utcnow().strftime( "%Y-%m-%dT%H:%M:%SZ") request = {'host': host_url.netloc.lower(), 'verb': SIGNAL_VERB[signal_type], 'path': unquoted_path, 'params': params} # Sign the request signer = ec2_utils.Ec2Signer(secret_key) request['params']['Signature'] = signer.generate(request) qs = parse.urlencode(request['params']) url = "%s%s?%s" % (signal_url.lower(), path, qs) return url def _delete_ec2_signed_url(self): self.data_delete('ec2_signed_url') self._delete_keypair() def _get_heat_signal_url(self, project_id=None): """Return a heat-api signal URL for this resource. This URL is not pre-signed, valid user credentials are required. If a project_id is provided, it is used in place of the original project_id. This is useful to generate a signal URL that uses the heat stack user project instead of the user's. """ stored = self.data().get('heat_signal_url') if stored is not None: return stored if self.id is None: # it is too early return url = self.client_plugin('heat').get_heat_url() path = self.identifier().url_path() if project_id is not None: path = project_id + path[path.find('/'):] url = parse.urljoin(url, '%s/signal' % path) self.data_set('heat_signal_url', url) return url def _delete_heat_signal_url(self): self.data_delete('heat_signal_url') def _get_swift_signal_url(self, multiple_signals=False): """Create properly formatted and pre-signed Swift signal URL. This uses a Swift pre-signed temp_url. If multiple_signals is requested, the Swift object referenced by the returned URL will have versioning enabled. """ put_url = self.data().get('swift_signal_url') if put_url: return put_url if self.id is None: # it is too early return container = self.stack.id object_name = self.physical_resource_name() self.client('swift').put_container(container) if multiple_signals: put_url = self.client_plugin('swift').get_signal_url(container, object_name) else: put_url = self.client_plugin('swift').get_temp_url(container, object_name) self.client('swift').put_object(container, object_name, '') self.data_set('swift_signal_url', put_url) self.data_set('swift_signal_object_name', object_name) return put_url def _delete_swift_signal_url(self): object_name = self.data().get('swift_signal_object_name') if not object_name: return with self.client_plugin('swift').ignore_not_found: container_name = self.stack.id swift = self.client('swift') # delete all versions of the object, in case there are some # signals that are waiting to be handled container = swift.get_container(container_name) filtered = [obj for obj in container[1] if object_name in obj['name']] for obj in filtered: # we delete the main object every time, swift takes # care of restoring the previous version after each delete swift.delete_object(container_name, object_name) headers = swift.head_container(container_name) if int(headers['x-container-object-count']) == 0: swift.delete_container(container_name) self.data_delete('swift_signal_object_name') self.data_delete('swift_signal_url') def _get_zaqar_signal_queue_id(self): """Return a zaqar queue_id for signaling this resource. This uses the created user for the credentials. """ queue_id = self.data().get('zaqar_signal_queue_id') if queue_id: return queue_id if self.id is None: # it is too early return if self._get_user_id() is None: if self.password is None: self.password = password_gen.generate_openstack_password() self._create_user() queue_id = self.physical_resource_name() zaqar_plugin = self.client_plugin('zaqar') zaqar = zaqar_plugin.create_for_tenant( self.stack.stack_user_project_id, self._user_token()) queue = zaqar.queue(queue_id) signed_url_data = queue.signed_url( ['messages'], methods=['GET', 'DELETE']) self.data_set('zaqar_queue_signed_url_data', jsonutils.dumps(signed_url_data)) self.data_set('zaqar_signal_queue_id', queue_id) return queue_id def _delete_zaqar_signal_queue(self): queue_id = self.data().get('zaqar_signal_queue_id') if not queue_id: return zaqar_plugin = self.client_plugin('zaqar') zaqar = zaqar_plugin.create_for_tenant( self.stack.stack_user_project_id, self._user_token()) with zaqar_plugin.ignore_not_found: zaqar.queue(queue_id).delete() self.data_delete('zaqar_signal_queue_id') def _get_signal(self, signal_type=SIGNAL, multiple_signals=False): """Return a dictionary with signal details. Subclasses can invoke this method to retrieve information of the resource signal for the specified transport. """ signal = None if self._signal_transport_cfn(): signal = {'alarm_url': self._get_ec2_signed_url( signal_type=signal_type)} elif self._signal_transport_heat(): signal = self._get_heat_signal_credentials() signal['alarm_url'] = self._get_heat_signal_url( project_id=self.stack.stack_user_project_id) elif self._signal_transport_temp_url(): signal = {'alarm_url': self._get_swift_signal_url( multiple_signals=multiple_signals)} elif self._signal_transport_zaqar(): signal = self._get_heat_signal_credentials() signal['queue_id'] = self._get_zaqar_signal_queue_id() elif self._signal_transport_none(): signal = {} return signal def _service_swift_signal(self): swift_client = self.client('swift') try: container = swift_client.get_container(self.stack.id) except Exception as exc: self.client_plugin('swift').ignore_not_found(exc) LOG.debug("Swift container %s was not found", self.stack.id) return index = container[1] if not index: # Swift objects were deleted by user LOG.debug("Swift objects in container %s were not found", self.stack.id) return # Remove objects that are for other resources, given that # multiple swift signals in the same stack share a container object_name = self.physical_resource_name() filtered = [obj for obj in index if object_name in obj['name']] # Fetch objects from Swift and filter results signal_names = [] for obj in filtered: try: signal = swift_client.get_object(self.stack.id, obj['name']) except Exception as exc: self.client_plugin('swift').ignore_not_found(exc) continue body = signal[1] if body == swift.IN_PROGRESS: # Ignore the initial object continue signal_names.append(obj['name']) if body == "": self.signal(details={}) continue try: self.signal(details=jsonutils.loads(body)) except ValueError: raise exception.Error(_("Failed to parse JSON data: %s") % body) # remove the signals that were consumed for signal_name in signal_names: if signal_name != object_name: swift_client.delete_object(self.stack.id, signal_name) if object_name in signal_names: swift_client.delete_object(self.stack.id, object_name) def _service_zaqar_signal(self): zaqar_plugin = self.client_plugin('zaqar') zaqar = zaqar_plugin.create_for_tenant( self.stack.stack_user_project_id, self._user_token()) try: queue = zaqar.queue(self._get_zaqar_signal_queue_id()) except Exception as ex: self.client_plugin('zaqar').ignore_not_found(ex) return messages = list(queue.pop()) for message in messages: self.signal(details=message.body) def _service_signal(self): """Service the signal, when necessary. This method must be called repeatedly by subclasses to update the state of the signals that require polling, which are the ones based on Swift temp URLs and Zaqar queues. The "NO_SIGNAL" case is also handled here by triggering the signal once per call. """ if self._signal_transport_temp_url(): self._service_swift_signal() elif self._signal_transport_zaqar(): self._service_zaqar_signal() elif self._signal_transport_none(): self.signal(details={})