aodh/ceilometer/storage/impl_sqlalchemy.py

# -*- encoding: utf-8 -*-
#
# Author: John Tran <jhtran@att.com>
#         Julien Danjou <julien@danjou.info>
#
# 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.

"""SQLAlchemy storage backend."""

from __future__ import absolute_import

import copy
import os
from sqlalchemy import func

from ceilometer.openstack.common import log
from ceilometer.openstack.common import timeutils
from ceilometer.storage import base
from ceilometer.storage import models as api_models
from ceilometer.storage.sqlalchemy import migration
from ceilometer.storage.sqlalchemy.models import Meter, Project, Resource
from ceilometer.storage.sqlalchemy.models import Source, User, Base
import ceilometer.storage.sqlalchemy.session as sqlalchemy_session

LOG = log.getLogger(__name__)


class SQLAlchemyStorage(base.StorageEngine):
    """Put the data into a SQLAlchemy database.

    Tables::

        - user
          - { id: user uuid }
        - source
          - { id: source id }
        - project
          - { id: project uuid }
        - meter
          - the raw incoming data
          - { id: meter id
              counter_name: counter name
              user_id: user uuid            (->user.id)
              project_id: project uuid      (->project.id)
              resource_id: resource uuid    (->resource.id)
              resource_metadata: metadata dictionaries
              counter_type: counter type
              counter_unit: counter unit
              counter_volume: counter volume
              timestamp: datetime
              message_signature: message signature
              message_id: message uuid
              }
        - resource
          - the metadata for resources
          - { id: resource uuid
              resource_metadata: metadata dictionaries
              project_id: project uuid      (->project.id)
              user_id: user uuid            (->user.id)
              }
        - sourceassoc
          - the relationships
          - { meter_id: meter id            (->meter.id)
              project_id: project uuid      (->project.id)
              resource_id: resource uuid    (->resource.id)
              user_id: user uuid            (->user.id)
              source_id: source id          (->source.id)
              }
    """

    OPTIONS = []

    def register_opts(self, conf):
        """Register any configuration options used by this engine."""
        conf.register_opts(self.OPTIONS)

    @staticmethod
    def get_connection(conf):
        """Return a Connection instance based on the configuration settings.
        """
        return Connection(conf)


def make_query_from_filter(query, event_filter, require_meter=True):
    """Return a query dictionary based on the settings in the filter.

    :param filter: EventFilter instance
    :param require_meter: If true and the filter does not have a meter,
                          raise an error.
    """

    if event_filter.meter:
        query = query.filter(Meter.counter_name == event_filter.meter)
    elif require_meter:
        raise RuntimeError('Missing required meter specifier')
    if event_filter.source:
        query = query.filter(Meter.sources.any(id=event_filter.source))
    if event_filter.start:
        ts_start = event_filter.start
        query = query.filter(Meter.timestamp >= ts_start)
    if event_filter.end:
        ts_end = event_filter.end
        query = query.filter(Meter.timestamp < ts_end)
    if event_filter.user:
        query = query.filter_by(user_id=event_filter.user)
    if event_filter.project:
        query = query.filter_by(project_id=event_filter.project)
    if event_filter.resource:
        query = query.filter_by(resource_id=event_filter.resource)

    if event_filter.metaquery:
        raise NotImplementedError('metaquery not implemented')

    return query


class Connection(base.Connection):
    """SqlAlchemy connection."""

    def __init__(self, conf):
        url = conf.database_connection
        if url == 'sqlite://':
            url = os.environ.get('CEILOMETER_TEST_SQL_URL', url)
        LOG.info('connecting to %s', url)
        self.session = sqlalchemy_session.get_session(url, conf)

    def upgrade(self, version=None):
        migration.db_sync(self.session.get_bind(), version=version)

    def clear(self):
        engine = self.session.get_bind()
        for table in reversed(Base.metadata.sorted_tables):
            engine.execute(table.delete())

    def record_metering_data(self, data):
        """Write the data to the backend storage system.

        :param data: a dictionary such as returned by
                     ceilometer.meter.meter_message_from_counter
        """
        if data['source']:
            source = self.session.query(Source).get(data['source'])
            if not source:
                source = Source(id=data['source'])
                self.session.add(source)
        else:
            source = None

        # create/update user && project, add/update their sources list
        if data['user_id']:
            user = self.session.merge(User(id=str(data['user_id'])))
            if not filter(lambda x: x.id == source.id, user.sources):
                user.sources.append(source)
        else:
            user = None

        if data['project_id']:
            project = self.session.merge(Project(id=str(data['project_id'])))
            if not filter(lambda x: x.id == source.id, project.sources):
                project.sources.append(source)
        else:
            project = None

        # Record the updated resource metadata
        rmetadata = data['resource_metadata']

        resource = self.session.merge(Resource(id=str(data['resource_id'])))
        if not filter(lambda x: x.id == source.id, resource.sources):
            resource.sources.append(source)
        resource.project = project
        resource.user = user
        # Current metadata being used and when it was last updated.
        resource.resource_metadata = rmetadata
        # autoflush didn't catch this one, requires manual flush
        self.session.flush()

        # Record the raw data for the event.
        meter = Meter(counter_type=data['counter_type'],
                      counter_unit=data['counter_unit'],
                      counter_name=data['counter_name'], resource=resource)
        self.session.add(meter)
        if not filter(lambda x: x.id == source.id, meter.sources):
            meter.sources.append(source)
        meter.project = project
        meter.user = user
        meter.timestamp = data['timestamp']
        meter.resource_metadata = rmetadata
        meter.counter_volume = data['counter_volume']
        meter.message_signature = data['message_signature']
        meter.message_id = data['message_id']

        return

    def get_users(self, source=None):
        """Return an iterable of user id strings.

        :param source: Optional source filter.
        """
        query = self.session.query(User.id)
        if source is not None:
            query = query.filter(User.sources.any(id=source))
        return (x[0] for x in query.all())

    def get_projects(self, source=None):
        """Return an iterable of project id strings.

        :param source: Optional source filter.
        """
        query = self.session.query(Project.id)
        if source:
            query = query.filter(Project.sources.any(id=source))
        return (x[0] for x in query.all())

    def get_resources(self, user=None, project=None, source=None,
                      start_timestamp=None, end_timestamp=None,
                      metaquery={}, resource=None):
        """Return an iterable of api_models.Resource instances

        :param user: Optional ID for user that owns the resource.
        :param project: Optional ID for project that owns the resource.
        :param source: Optional source filter.
        :param start_timestamp: Optional modified timestamp start range.
        :param end_timestamp: Optional modified timestamp end range.
        :param metaquery: Optional dict with metadata to match on.
        :param resource: Optional resource filter.
        """
        query = self.session.query(Meter,).group_by(Meter.resource_id)
        if user is not None:
            query = query.filter(Meter.user_id == user)
        if source is not None:
            query = query.filter(Meter.sources.any(id=source))
        if start_timestamp:
            query = query.filter(Meter.timestamp >= start_timestamp)
        if end_timestamp:
            query = query.filter(Meter.timestamp < end_timestamp)
        if project is not None:
            query = query.filter(Meter.project_id == project)
        if resource is not None:
            query = query.filter(Meter.resource_id == resource)
        if metaquery:
            raise NotImplementedError('metaquery not implemented')

        for meter in query.all():
            yield api_models.Resource(
                resource_id=meter.resource_id,
                project_id=meter.project_id,
                user_id=meter.user_id,
                metadata=meter.resource_metadata,
                meter=[
                    api_models.ResourceMeter(
                        counter_name=m.counter_name,
                        counter_type=m.counter_type,
                        counter_unit=m.counter_unit,
                    )
                    for m in meter.resource.meters
                ],
            )

    def get_meters(self, user=None, project=None, resource=None, source=None,
                   metaquery={}):
        """Return an iterable of api_models.Meter instances

        :param user: Optional ID for user that owns the resource.
        :param project: Optional ID for project that owns the resource.
        :param resource: Optional ID of the resource.
        :param source: Optional source filter.
        :param metaquery: Optional dict with metadata to match on.
        """
        query = self.session.query(Resource)
        if user is not None:
            query = query.filter(Resource.user_id == user)
        if source is not None:
            query = query.filter(Resource.sources.any(id=source))
        if resource:
            query = query.filter(Resource.id == resource)
        if project is not None:
            query = query.filter(Resource.project_id == project)
        query = query.options(
            sqlalchemy_session.sqlalchemy.orm.joinedload('meters'))
        if metaquery:
            raise NotImplementedError('metaquery not implemented')

        for resource in query.all():
            meter_names = set()
            for meter in resource.meters:
                if meter.counter_name in meter_names:
                    continue
                meter_names.add(meter.counter_name)
                yield api_models.Meter(
                    name=meter.counter_name,
                    type=meter.counter_type,
                    unit=meter.counter_unit,
                    resource_id=resource.id,
                    project_id=resource.project_id,
                    user_id=resource.user_id,
                )

    def get_samples(self, event_filter):
        """Return an iterable of api_models.Samples
        """
        query = self.session.query(Meter)
        query = make_query_from_filter(query, event_filter,
                                       require_meter=False)
        samples = query.all()

        for s in samples:
            # Remove the id generated by the database when
            # the event was inserted. It is an implementation
            # detail that should not leak outside of the driver.
            yield api_models.Sample(
                # Replace 'sources' with 'source' to meet the caller's
                # expectation, Meter.sources contains one and only one
                # source in the current implementation.
                source=s.sources[0].id,
                counter_name=s.counter_name,
                counter_type=s.counter_type,
                counter_unit=s.counter_unit,
                counter_volume=s.counter_volume,
                user_id=s.user_id,
                project_id=s.project_id,
                resource_id=s.resource_id,
                timestamp=s.timestamp,
                resource_metadata=s.resource_metadata,
                message_id=s.message_id,
                message_signature=s.message_signature,
            )

    def _make_volume_query(self, event_filter, counter_volume_func):
        """Returns complex Meter counter_volume query for max and sum."""
        subq = self.session.query(Meter.id)
        subq = make_query_from_filter(subq, event_filter, require_meter=False)
        subq = subq.subquery()
        mainq = self.session.query(Resource.id, counter_volume_func)
        mainq = mainq.join(Meter).group_by(Resource.id)
        return mainq.filter(Meter.id.in_(subq))

    def get_event_interval(self, event_filter):
        """Return the min and max timestamps from samples,
        using the event_filter to limit the samples seen.

        ( datetime.datetime(), datetime.datetime() )
        """
        query = self.session.query(func.min(Meter.timestamp),
                                   func.max(Meter.timestamp))
        query = make_query_from_filter(query, event_filter)
        results = query.all()
        a_min, a_max = results[0]
        return (a_min, a_max)

    def _make_stats_query(self, event_filter):
        query = self.session.query(
            func.min(Meter.timestamp).label('tsmin'),
            func.max(Meter.timestamp).label('tsmax'),
            func.avg(Meter.counter_volume).label('avg'),
            func.sum(Meter.counter_volume).label('sum'),
            func.min(Meter.counter_volume).label('min'),
            func.max(Meter.counter_volume).label('max'),
            func.count(Meter.counter_volume).label('count'))

        return make_query_from_filter(query, event_filter)

    @staticmethod
    def _stats_result_to_model(result, period, period_start, period_end):
        duration = (timeutils.delta_seconds(result.tsmin, result.tsmax)
                    if result.tsmin is not None and result.tsmax is not None
                    else None)
        return api_models.Statistics(
            count=int(result.count),
            min=result.min,
            max=result.max,
            avg=result.avg,
            sum=result.sum,
            duration_start=result.tsmin,
            duration_end=result.tsmax,
            duration=duration,
            period=period,
            period_start=period_start,
            period_end=period_end,
        )

    def get_meter_statistics(self, event_filter, period=None):
        """Return an iterable of api_models.Statistics instances containing
        meter statistics described by the query parameters.

        The filter must have a meter value set.

        """
        if not period or not event_filter.start or not event_filter.end:
            res = self._make_stats_query(event_filter).all()[0]

        if not period:
            yield self._stats_result_to_model(res, 0, res.tsmin, res.tsmax)
            return

        query = self._make_stats_query(event_filter)
        # HACK(jd) This is an awful method to compute stats by period, but
        # since we're trying to be SQL agnostic we have to write portable
        # code, so here it is, admire! We're going to do one request to get
        # stats by period. We would like to use GROUP BY, but there's no
        # portable way to manipulate timestamp in SQL, so we can't.
        for period_start, period_end in base.iter_period(
                event_filter.start or res.tsmin,
                event_filter.end or res.tsmax,
                period):
            q = query.filter(Meter.timestamp >= period_start)
            q = q.filter(Meter.timestamp < period_end)
            r = q.all()[0]
            # Don't return results that didn't have any event
            if r.count:
                yield self._stats_result_to_model(
                    result=r,
                    period=int(timeutils.delta_seconds(period_start,
                                                       period_end)),
                    period_start=period_start,
                    period_end=period_end,
                )