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 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 six.moves.urllib import parse as urlparse

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_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.context.region_name or
                                cfg.CONF.region_name_for_services)}

    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 = urlparse.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 = urlparse.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 = urlparse.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 = urlparse.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={})