Sign Up
Log In
Log In
or
Sign Up
Places
All Projects
Status Monitor
Collapse sidebar
Please login to access the resource
systemsmanagement:Ardana:8:CentOS:7.3
openstack-monasca-api
cassandra-527858.patch
Overview
Repositories
Revisions
Requests
Users
Attributes
Meta
File cassandra-527858.patch of Package openstack-monasca-api
From 4f3b94270f044ff5a3f124caaec15548cc18e308 Mon Sep 17 00:00:00 2001 From: James Gu <jgu@suse.com> Date: Thu, 31 Aug 2017 14:47:47 -0700 URL: https://review.openstack.org/#/c/527858/ Subject: [PATCH] Add cassandra support Support Cassandra db installation and Cassandra related configurations for Monasca api and persister services in devstack. Add Monasca rest API Cassandra plugin for retrieving metric, dimension, measurement, statistics and alarms. Also includes the equivalent of commit eb78188f97c8576fc39c1dbb7057bddf4fc56113 as the previous commit 011db48a92a2b... was not compatible. Depends-On: I4bbe48f8fe550385de6ed3e14120850a8542c7c9 Change-Id: Ie60d668692e1f25f555dda2355f4e513d582736c story: 2001231 task: 5759 (cherry picked from commit ba43f07726d060ac0a3e4d5289dc4ad3fde2397c) --- AUTHORS | 5 + devstack/files/cassandra/cassandra_schema.cql | 46 - devstack/files/cassandra/monasca_schema.cql | 93 ++ devstack/files/monasca-persister/persister.yml | 54 +- devstack/lib/persister.sh | 49 +- devstack/plugin.sh | 29 +- devstack/settings | 2 +- docs/monasca-api-spec.md | 16 +- .../repositories/cassandra/metrics_repository.py | 1231 ++++++++++++-------- .../repositories/influxdb/metrics_repository.py | 14 +- monasca_api/conf/cassandra.py | 43 + monasca_api/healthcheck/metrics_db_check.py | 4 +- monasca_api/tests/test_metrics_db_health_check.py | 9 +- monasca_api/tests/test_repositories.py | 214 ++-- monasca_api/v2/reference/__init__.py | 2 +- monasca_api/v2/reference/helpers.py | 16 +- monasca_tempest_tests/tests/api/helpers.py | 33 +- monasca_tempest_tests/tests/api/test_dimensions.py | 8 +- monasca_tempest_tests/tests/api/test_statistics.py | 3 +- 19 files changed, 1172 insertions(+), 699 deletions(-) delete mode 100644 devstack/files/cassandra/cassandra_schema.cql create mode 100644 devstack/files/cassandra/monasca_schema.cql create mode 100644 monasca_api/conf/cassandra.py Index: monasca-api-2.2.1.dev26/devstack/files/cassandra/cassandra_schema.cql =================================================================== --- monasca-api-2.2.1.dev26.orig/devstack/files/cassandra/cassandra_schema.cql +++ /dev/null @@ -1,46 +0,0 @@ -drop table if exists monasca.metric_map; - -drop table if exists monasca.measurements; - -drop table if exists monasca.alarm_state_history; - -drop schema if exists monasca; - -create schema monasca -with replication = { 'class' : 'SimpleStrategy', 'replication_factor' : 1 }; - -use monasca; - -create table monasca.metric_map ( - tenant_id text, - region text, - metric_hash blob, - metric_map map<text, text>, -primary key ((tenant_id, region), metric_hash) -); - -create index on monasca.metric_map (entries(metric_map)); - -create table monasca.measurements ( - tenant_id text, - region text, - metric_hash blob, - time_stamp timestamp, - value double, - value_meta text, -primary key ((tenant_id, region, metric_hash), time_stamp) -); - -create table monasca.alarm_state_history ( - tenant_id text, - alarm_id text, - metrics text, - new_state text, - old_state text, - reason text, - reason_data text, - sub_alarms text, - time_stamp timestamp, -primary key ((tenant_id), alarm_id, time_stamp) -); - Index: monasca-api-2.2.1.dev26/devstack/files/cassandra/monasca_schema.cql =================================================================== --- /dev/null +++ monasca-api-2.2.1.dev26/devstack/files/cassandra/monasca_schema.cql @@ -0,0 +1,93 @@ +// (C) Copyright 2017 SUSE 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. + +// version 1.0 + +drop schema if exists monasca; + +// replication factor is set to 1 for devstack installation + +create schema monasca with replication = { 'class' : 'SimpleStrategy', 'replication_factor' : 1 }; + +create table monasca.measurements ( + metric_id blob, + region text static, + tenant_id text static, + metric_name text static, + dimensions frozen<list<text>> static, + time_stamp timestamp, + value double, + value_meta text, + primary key (metric_id, time_stamp) +) +WITH CLUSTERING ORDER BY (time_stamp ASC); + +create table monasca.metrics ( + region text, + tenant_id text, + metric_name text, + dimensions frozen<list<text>>, + dimension_names frozen<list<text>>, + metric_id blob, + created_at timestamp, + updated_at timestamp, + primary key ((region, tenant_id, metric_name), dimensions, dimension_names) +); + +CREATE CUSTOM INDEX metrics_created_at_index ON monasca.metrics (created_at) +USING 'org.apache.cassandra.index.sasi.SASIIndex'; + +CREATE CUSTOM INDEX metrics_updated_at_index ON monasca.metrics (updated_at) +USING 'org.apache.cassandra.index.sasi.SASIIndex'; + +create table monasca.dimensions ( + region text, + tenant_id text, + name text, + value text, + primary key ((region, tenant_id, name), value) +); + +create table monasca.dimensions_metrics ( + region text, + tenant_id text, + dimension_name text, + dimension_value text, + metric_name text, + primary key ((region, tenant_id, dimension_name, dimension_value), metric_name) +); + +create table monasca.metrics_dimensions ( + region text, + tenant_id text, + dimension_name text, + dimension_value text, + metric_name text, + primary key ((region, tenant_id, metric_name), dimension_name, dimension_value) +); + +create table monasca.alarm_state_history ( + tenant_id text, + alarm_id text, + time_stamp timestamp, + metric text, + old_state text, + new_state text, + reason text, + reason_data text, + sub_alarms text, + primary key ((tenant_id, alarm_id), time_stamp) +); + Index: monasca-api-2.2.1.dev26/devstack/files/monasca-persister/persister.yml =================================================================== --- monasca-api-2.2.1.dev26.orig/devstack/files/monasca-persister/persister.yml +++ monasca-api-2.2.1.dev26/devstack/files/monasca-persister/persister.yml @@ -1,5 +1,6 @@ # # (C) Copyright 2015 Hewlett Packard Enterprise Development Company LP +# Copyright (c) 2017 SUSE LLC # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. @@ -18,24 +19,26 @@ name: monasca-persister alarmHistoryConfiguration: - batchSize: 100 + batchSize: %MONASCA_PERSISTER_BATCH_SIZE% numThreads: 1 - maxBatchTime: 15 + maxBatchTime: %MONASCA_PERSISTER_MAX_BATCH_TIME% + commitBatchTime: %MONASCA_PERSISTER_COMMIT_BATCH_TIME% # See http://kafka.apache.org/documentation.html#api for semantics and defaults. topic: alarm-state-transitions groupId: 1_alarm-state-transitions consumerId: "mini-mon" - clientId : 1 + clientId: 1 metricConfiguration: - batchSize: 100 - numThreads: 1 - maxBatchTime: 15 + batchSize: %MONASCA_PERSISTER_BATCH_SIZE% + numThreads: %MONASCA_PERSISTER_METRIC_THREADS% + maxBatchTime: %MONASCA_PERSISTER_MAX_BATCH_TIME% + commitBatchTime: %MONASCA_PERSISTER_COMMIT_BATCH_TIME% # See http://kafka.apache.org/documentation.html#api for semantics and defaults. topic: metrics groupId: 1_metrics consumerId: "mini-mon" - clientId : 1 + clientId: 1 #Kafka settings. kafkaConfig: @@ -56,6 +59,43 @@ kafkaConfig: zookeeperConnectionTimeoutMs : 60000 zookeeperSyncTimeMs: 2000 +# uncomment if database type is cassandra +cassandraDbConfiguration: + contactPoints: + - %CASSANDRADB_HOST% + port: 9042 + user: mon_persister + password: password + keyspace: monasca + localDataCenter: datacenter1 + maxConnections: 5 + maxRequests: 2048 + # socket time out in milliseconds when creating a new connection + connectionTimeout: 5000 + # how long the driver waits for a response from server. Must be + # longer than the server side timeouts in the cassandra.yaml + readTimeout: 60000 + + # number of retries in upsert query. The retry interval is exponential, + # i.e., 1, 2, 4, 8 ... seconds. Retry is blocking. + maxWriteRetries: 5 + maxBatches: 250 + maxDefinitionCacheSize: 2000000 + # ANY(0), + # ONE(1), + # TWO(2), + # THREE(3), + # QUORUM(4), + # ALL(5), + # LOCAL_QUORUM(6), + # EACH_QUORUM(7), + # SERIAL(8), + # LOCAL_SERIAL(9), + # LOCAL_ONE(10); + consistencyLevel: ONE + # number of days metric retention + retentionPolicy: 45 + verticaMetricRepoConfig: maxCacheSize: 2000000 Index: monasca-api-2.2.1.dev26/devstack/lib/persister.sh =================================================================== --- monasca-api-2.2.1.dev26.orig/devstack/lib/persister.sh +++ monasca-api-2.2.1.dev26/devstack/lib/persister.sh @@ -1,7 +1,8 @@ #!/bin/bash # Copyright 2017 FUJITSU LIMITED -# +# (C) Copyright 2017 SUSE 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 @@ -53,6 +54,18 @@ else MONASCA_PERSISTER_CMD="/usr/bin/java ${MONASCA_PERSISTER_JAVA_OPTS} -cp ${MONASCA_PERSISTER_JAR} monasca.persister.PersisterApplication server ${MONASCA_PERSISTER_CONF}" fi +if [[ "${MONASCA_METRICS_DB,,}" == 'cassandra' ]]; then + MONASCA_PERSISTER_BATCH_SIZE=100 + MONASCA_PERSISTER_MAX_BATCH_TIME=10 + MONASCA_PERSISTER_METRIC_THREADS=2 + MONASCA_PERSISTER_COMMIT_BATCH_TIME=10000 +else + MONASCA_PERSISTER_BATCH_SIZE=100 + MONASCA_PERSISTER_MAX_BATCH_TIME=15 + MONASCA_PERSISTER_METRIC_THREADS=10 + MONASCA_PERSISTER_COMMIT_BATCH_TIME=0 +fi + is_monasca_persister_enabled() { is_service_enabled monasca-persister && return 0 return 1 @@ -141,10 +154,12 @@ configure_monasca_persister_python() { iniset "$MONASCA_PERSISTER_CONF" kafka_metrics uri $SERVICE_HOST:9092 iniset "$MONASCA_PERSISTER_CONF" kafka_metrics group_id 1_metrics iniset "$MONASCA_PERSISTER_CONF" kafka_metrics topic metrics + iniset "$MONASCA_PERSISTER_CONF" kafka_metrics batch_size 30 iniset "$MONASCA_PERSISTER_CONF" kafka_alarm_history uri $SERVICE_HOST:9092 iniset "$MONASCA_PERSISTER_CONF" kafka_alarm_history group_id 1_alarm-state-transitions iniset "$MONASCA_PERSISTER_CONF" kafka_alarm_history topic alarm-state-transitions + iniset "$MONASCA_PERSISTER_CONF" kafka_alarm_history batch_size 1 iniset "$MONASCA_PERSISTER_CONF" zookeeper uri $SERVICE_HOST:2181 @@ -155,9 +170,32 @@ configure_monasca_persister_python() { iniset "$MONASCA_PERSISTER_CONF" influxdb password password iniset "$MONASCA_PERSISTER_CONF" repositories metrics_driver ${M_REPO_DRIVER_INFLUX} iniset "$MONASCA_PERSISTER_CONF" repositories alarm_state_history_driver ${AH_REPO_DRIVER_INFLUX} - else - iniset "$MONASCA_PERSISTER_CONF" cassandra cluster_ip_addresses ${SERVICE_HOST} + elif [[ "${MONASCA_METRICS_DB,,}" == 'cassandra' ]]; then + iniset "$MONASCA_PERSISTER_CONF" cassandra contact_points ${SERVICE_HOST} + iniset "$MONASCA_PERSISTER_CONF" cassandra port 9042 + # iniset "$MONASCA_PERSISTER_CONF" cassandra user monasca + # iniset "$MONASCA_PERSISTER_CONF" cassandra password password iniset "$MONASCA_PERSISTER_CONF" cassandra keyspace monasca + iniset "$MONASCA_PERSISTER_CONF" cassandra local_data_center datacenter1 + iniset "$MONASCA_PERSISTER_CONF" cassandra connection_timeout 5 + iniset "$MONASCA_PERSISTER_CONF" cassandra read_timeout 60 + iniset "$MONASCA_PERSISTER_CONF" cassandra max_write_retries 5 + iniset "$MONASCA_PERSISTER_CONF" cassandra max_batches 250 + iniset "$MONASCA_PERSISTER_CONF" cassandra max_definition_cache_size 1000000 + # consistency level names: + # ANY(0), + # ONE(1), + # TWO(2), + # THREE(3), + # QUORUM(4), + # ALL(5), + # LOCAL_QUORUM(6), + # EACH_QUORUM(7), + # SERIAL(8), + # LOCAL_SERIAL(9), + # LOCAL_ONE(10); + iniset "$MONASCA_PERSISTER_CONF" cassandra consistency_level ONE + iniset "$MONASCA_PERSISTER_CONF" cassandra retention_policy 45 iniset "$MONASCA_PERSISTER_CONF" repositories metrics_driver ${M_REPO_DRIVER_CASSANDRA} iniset "$MONASCA_PERSISTER_CONF" repositories alarm_state_history_driver ${AH_REPO_DRIVER_CASSANDRA} fi @@ -190,11 +228,16 @@ configure_monasca_persister_java() { s|%ZOOKEEPER_HOST%|${SERVICE_HOST}|g; s|%VERTICA_HOST%|${SERVICE_HOST}|g; s|%INFLUXDB_HOST%|${SERVICE_HOST}|g; + s|%CASSANDRADB_HOST%|${SERVICE_HOST}|g; s|%MONASCA_PERSISTER_DB_TYPE%|${MONASCA_METRICS_DB}|g; s|%MONASCA_PERSISTER_BIND_HOST%|${MONASCA_PERSISTER_BIND_HOST}|g; s|%MONASCA_PERSISTER_APP_PORT%|${MONASCA_PERSISTER_APP_PORT}|g; s|%MONASCA_PERSISTER_ADMIN_PORT%|${MONASCA_PERSISTER_ADMIN_PORT}|g; s|%MONASCA_PERSISTER_LOG_DIR%|${MONASCA_PERSISTER_LOG_DIR}|g; + s|%MONASCA_PERSISTER_BATCH_SIZE%|${MONASCA_PERSISTER_BATCH_SIZE}|g; + s|%MONASCA_PERSISTER_MAX_BATCH_TIME%|${MONASCA_PERSISTER_MAX_BATCH_TIME}|g; + s|%MONASCA_PERSISTER_COMMIT_BATCH_TIME%|${MONASCA_PERSISTER_COMMIT_BATCH_TIME}|g; + s|%MONASCA_PERSISTER_METRIC_THREADS%|${MONASCA_PERSISTER_METRIC_THREADS}|g; " -i ${MONASCA_PERSISTER_CONF} ln -sf ${MONASCA_PERSISTER_CONF} ${MONASCA_PERSISTER_GATE_CONFIG} Index: monasca-api-2.2.1.dev26/devstack/plugin.sh =================================================================== --- monasca-api-2.2.1.dev26.orig/devstack/plugin.sh +++ monasca-api-2.2.1.dev26/devstack/plugin.sh @@ -1,6 +1,7 @@ # # (C) Copyright 2015-2017 Hewlett Packard Enterprise Development LP # Copyright 2017 FUJITSU LIMITED +# (C) Copyright 2017 SUSE LLC # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. @@ -486,11 +487,13 @@ function install_monasca_cassandra { echo_summary "Install Monasca Cassandra" if [[ "$OFFLINE" != "True" ]]; then - sudo sh -c "echo 'deb http://www.apache.org/dist/cassandra/debian ${CASSANDRA_VERSION} main' > /etc/apt/sources.list.d/cassandra.list" + sudo sh -c "echo 'deb http://www.apache.org/dist/cassandra/debian ${CASSANDRA_VERSION} main' > /etc/apt/sources.list.d/cassandra.sources.list" REPOS_UPDATED=False - PUBLIC_KEY=`apt_get_update 2>&1 | awk '/NO_PUBKEY/ {print $21}'` - gpg --keyserver pgp.mit.edu --recv-keys ${PUBLIC_KEY} - gpg --export --armor ${PUBLIC_KEY} | sudo apt-key --keyring /etc/apt/trusted.gpg.d/cassandra.gpg add - + curl https://www.apache.org/dist/cassandra/KEYS | sudo apt-key add - + PUBLIC_KEY=`sudo apt_get update 2>&1 | awk '/NO_PUBKEY/ {print $NF}'` + if [ -n "${PUBLIC_KEY}" ]; then + sudo apt-key adv --keyserver pool.sks-keyservers.net --recv-key ${PUBLIC_KEY} + fi fi REPOS_UPDATED=False @@ -515,6 +518,8 @@ function install_monasca_cassandra { sleep 15s export CQLSH_NO_BUNDLED=true + + # always needed for Monasca api pip_install_gr cassandra-driver } @@ -571,15 +576,15 @@ function clean_monasca_cassandra { echo_summary "Clean Monasca Cassandra" - sudo rm -f /etc/cassandra/cassandra.yaml + apt_get -y purge cassandra - sudo rm -rf /var/log/cassandra + apt_get -y autoremove - sudo rm -rf /etc/cassandra + sudo rm -rf /var/lib/cassandra - apt_get -y purge cassandra + sudo rm -rf /var/log/cassandra - apt_get -y autoremove + sudo rm -rf /etc/cassandra sudo rm -f /etc/apt/sources.list.d/cassandra.list @@ -612,8 +617,8 @@ function install_schema_metric_database_ } function install_schema_metric_database_cassandra { - sudo cp -f "${MONASCA_API_DIR}"/devstack/files/cassandra/cassandra_schema.cql $MONASCA_SCHEMA_DIR/cassandra_schema.cql - /usr/bin/cqlsh ${SERVICE_HOST} -f $MONASCA_SCHEMA_DIR/cassandra_schema.cql + sudo cp -f "${MONASCA_API_DIR}"/devstack/files/cassandra/*.cql $MONASCA_SCHEMA_DIR + /usr/bin/cqlsh ${SERVICE_HOST} -f $MONASCA_SCHEMA_DIR/monasca_schema.cql } function install_schema_kafka_topics { @@ -823,7 +828,7 @@ function configure_monasca_api_python { fi iniset "$MONASCA_API_CONF" database connection $dbAlarmUrl iniset "$MONASCA_API_CONF" repositories metrics_driver $dbMetricDriver - iniset "$MONASCA_API_CONF" cassandra cluster_ip_addresses $SERVICE_HOST + iniset "$MONASCA_API_CONF" cassandra contact_points $SERVICE_HOST iniset "$MONASCA_API_CONF" influxdb ip_address $SERVICE_HOST iniset "$MONASCA_API_CONF" influxdb port 8086 iniset "$MONASCA_API_CONF" kafka uri "$SERVICE_HOST:9092" Index: monasca-api-2.2.1.dev26/devstack/settings =================================================================== --- monasca-api-2.2.1.dev26.orig/devstack/settings +++ monasca-api-2.2.1.dev26/devstack/settings @@ -97,7 +97,7 @@ INFLUXDB_PYTHON_VERSION=${INFLUXDB_PYTHO # INFLUXDB_VERSION=${INFLUXDB_VERSION:-0.9.5} VERTICA_VERSION=${VERTICA_VERSION:-8.0.0-0} -CASSANDRA_VERSION=${CASSANDRA_VERSION:-37x} +CASSANDRA_VERSION=${CASSANDRA_VERSION:-311x} # Kafka deb consists of the version of scala plus the version of kafka BASE_KAFKA_VERSION=${BASE_KAFKA_VERSION:-0.10.2.2} SCALA_VERSION=${SCALA_VERSION:-2.11} Index: monasca-api-2.2.1.dev26/docs/monasca-api-spec.md =================================================================== --- monasca-api-2.2.1.dev26.orig/docs/monasca-api-spec.md +++ monasca-api-2.2.1.dev26/docs/monasca-api-spec.md @@ -771,7 +771,7 @@ If no limit is specified in the request ## Offset -Offsets can be either integer offsets, string offsets (including hexadecimal numbers), or timestamp offsets. The use of either integer, string, or timestamp is determined by the resource being queried. +Offsets can be either identifier offsets, timestamp offsets or combinational offsets that have an identifier part and timestamp part. The identifier can be an integer or string (including hexadecimal numbers). The use of either integer, string, timestamp or combination is determined by the resource being queried. For example, an integer offset would look like this: @@ -809,12 +809,16 @@ A dimension value offset would look as f offset=dimensionValue2 ``` +A combinational offset with hexdecimal id would look as follows: +``` +offset=01ce0acc66131296c8a17294f39aee44ea8963ec_2104-01-01T00:00:01Z +``` -Different resources use different offset types because of the internal implementation of different resources depends on different types of mechanisms for indexing and identifying resources. The type and form of the offsets for each resource can be determined by referring to the examples in each resource section below. +Different resources use different offset types because of the internal implementation of different resources depends on different types of mechanisms for indexing and identifying resources. For example, the offset in measurement resources contains both ID and timestamp. The type and form of the offsets for each resource can be determined by referring to the examples in each resource section below. -The offset is determined by the ID of the last element in the result list. Users wishing to manually create a query URL can use the ID of the last element in the previously returned result set as the offset. The proceeding result set will return all elements with an ID greater than the offset up to the limit. The automatically generated offset in the next link does exactly this; it uses the ID in the last element. +The offset is determined by the ID and/or timestamp values of the last element in the result list. Users wishing to manually create a query URL can use the ID and/or timestamp of the last element in the previously returned result set as the offset. The proceeding result set will return all elements with an ID greater than the ID in the offset, and if the offset is two-part, also all the elements with the same ID as that in the offset and having a timestamp later than the timestamp value in the offset. The automatically generated offset in the next link does exactly this; it uses the ID and/or timestamp in the last element. -The offset can take the form of an integer, string, or timestamp, but the user should treat the offset as an opaque reference. When using offsets in manually generated URLs, users enter them as strings that look like integers, timestamps, or strings. Future releases may change the type and form of the offsets for each resource. +The offset can take the form of an integer ID, string ID, timestamp, or a combination of both ID and timestamp, but the user should treat the offset as an opaque reference. When using offsets in manually generated URLs, users enter them as strings that look like integers, timestamps, or strings. Future releases may change the type and form of the offsets for each resource. ## Limit The Monasca API has a server-wide default limit that is applied. Users may specify their own limit in the URL, but the server-wide limit may not be exceeded. The Monasca server-wide limit is configured in the Monasca API config file as maxQueryLimit. Users may specify a limit up to the maxQueryLimit. @@ -1389,12 +1393,12 @@ Returns a JSON object with a 'links' arr }, { "rel": "next", - "href": "http://192.168.10.4:8070/v2.0/metrics/measurements?offset=2015-03-03T05%3A24%3A55Z&name=cpu.system_perc&dimensions=hostname%3Adevstack&start_time=2015-03-00T00%3A00%3A00Z" + "href": "http://192.168.10.4:8070/v2.0/metrics/measurements?offset=01ce0acc66131296c8a17294f39aee44ea8963ec_2015-03-03T05%3A24%3A55.123Z&name=cpu.system_perc&dimensions=hostname%3Adevstack&start_time=2015-03-00T00%3A00%3A00Z" } ], "elements": [ { - "id": "2015-03-03T05:24:55Z", + "id": "01ce0acc66131296c8a17294f39aee44ea8963ec", "name": "http_status", "dimensions": { "url": "http://localhost:8774/v2.0", Index: monasca-api-2.2.1.dev26/monasca_api/common/repositories/cassandra/metrics_repository.py =================================================================== --- monasca-api-2.2.1.dev26.orig/monasca_api/common/repositories/cassandra/metrics_repository.py +++ monasca-api-2.2.1.dev26/monasca_api/common/repositories/cassandra/metrics_repository.py @@ -1,4 +1,5 @@ # (C) Copyright 2015,2016 Hewlett Packard Enterprise Development Company LP +# (C) Copyright 2017 SUSE 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 @@ -13,557 +14,885 @@ # under the License. import binascii +from collections import namedtuple from datetime import datetime from datetime import timedelta import itertools import urllib from cassandra.cluster import Cluster +from cassandra.query import FETCH_SIZE_UNSET from cassandra.query import SimpleStatement + from oslo_config import cfg from oslo_log import log from oslo_utils import timeutils -from monasca_common.rest import utils as rest_utils - from monasca_api.common.repositories import exceptions from monasca_api.common.repositories import metrics_repository +from monasca_common.rest import utils as rest_utils +CONF = cfg.CONF LOG = log.getLogger(__name__) +LIMIT_CLAUSE = 'limit %s' +ALLOW_FILTERING = 'allow filtering' -class MetricsRepository(metrics_repository.AbstractMetricsRepository): - def __init__(self): - - try: - - self.conf = cfg.CONF +MEASUREMENT_LIST_CQL = ('select time_stamp, value, value_meta ' + 'from measurements where %s %s %s %s') +METRIC_ID_EQ = 'metric_id = %s' +METRIC_ID_IN = 'metric_id in %s' +OFFSET_TIME_GT = "and time_stamp > %s" +START_TIME_GE = "and time_stamp >= %s" +END_TIME_LE = "and time_stamp <= %s" + +METRIC_LIST_CQL = ('select metric_name, dimensions, metric_id ' + 'from metrics where %s %s %s %s %s %s %s %s %s %s') +REGION_EQ = 'region = %s' +TENANT_EQ = 'and tenant_id = %s' +METRIC_NAME_EQ = 'and metric_name = %s' +DIMENSIONS_CONTAINS = 'and dimensions contains %s ' +DIMENSIONS_NAME_CONTAINS = 'and dimension_names contains %s ' +CREATED_TIME_LE = "and created_at <= %s" +UPDATED_TIME_GE = "and updated_at >= %s" +DIMENSIONS_GT = 'and dimensions > %s' + +DIMENSION_VALUE_BY_METRIC_CQL = ('select dimension_value as value from metrics_dimensions ' + 'where region = ? and tenant_id = ? and metric_name = ? ' + 'and dimension_name = ? group by dimension_value') + +DIMENSION_VALUE_CQL = ('select value from dimensions ' + 'where region = ? and tenant_id = ? and name = ? ' + 'group by value order by value') + +DIMENSION_NAME_BY_METRIC_CQL = ('select dimension_name as name from metrics_dimensions where ' + 'region = ? and tenant_id = ? and metric_name = ? ' + 'group by dimension_name order by dimension_name') + +DIMENSION_NAME_CQL = ('select name from dimensions where region = ? and tenant_id = ? ' + 'group by name allow filtering') + +METRIC_NAME_BY_DIMENSION_CQL = ('select metric_name from dimensions_metrics where region = ? and ' + 'tenant_id = ? and dimension_name = ? and dimension_value = ? ' + 'group by metric_name order by metric_name') + +METRIC_NAME_BY_DIMENSION_OFFSET_CQL = ('select metric_name from dimensions_metrics where region = ? and ' + 'tenant_id = ? and dimension_name = ? and dimension_value = ? and ' + 'metric_name >= ?' + 'group by metric_name order by metric_name') + +METRIC_NAME_CQL = ('select distinct region, tenant_id, metric_name from metrics_dimensions ' + 'where region = ? and tenant_id = ? allow filtering') + +METRIC_NAME_OFFSET_CQL = ('select distinct region, tenant_id, metric_name from metrics_dimensions ' + 'where region = ? and tenant_id = ? and metric_name >= ? allow filtering') + +METRIC_BY_ID_CQL = ('select region, tenant_id, metric_name, dimensions from measurements ' + 'where metric_id = ? limit 1') + +Metric = namedtuple('metric', 'id name dimensions') - self._cassandra_cluster = Cluster( - self.conf.cassandra.cluster_ip_addresses.split(',')) +ALARM_HISTORY_CQL = ('select tenant_id, alarm_id, time_stamp, metric, new_state, old_state, reason, reason_data, ' + 'sub_alarms from alarm_state_history where %s %s %s %s %s') - self.cassandra_session = self._cassandra_cluster.connect( - self.conf.cassandra.keyspace) +ALARM_ID_EQ = 'and alarm_id = %s' - except Exception as ex: - LOG.exception(ex) - raise exceptions.RepositoryException(ex) +ALARM_ID_IN = 'and alarm_id in %s' - def list_metrics(self, tenant_id, region, name, dimensions, offset, - limit, start_timestamp=None, end_timestamp=None, - include_metric_hash=False): +ALARM_TENANT_ID_EQ = 'tenant_id = %s' - or_dimensions = [] - sub_dimensions = {} - if dimensions: - for key, value in dimensions.iteritems(): - if not value: - sub_dimensions[key] = value +class MetricsRepository(metrics_repository.AbstractMetricsRepository): + def __init__(self): - elif '|' in value: + try: + self.conf = cfg.CONF + LOG.info('conf is: %s' % self.conf) + self.cluster = Cluster(self.conf.cassandra.contact_points) + self.session = self.cluster.connect(self.conf.cassandra.keyspace) - def f(val): - return {key: val} + self.dim_val_by_metric_stmt = self.session.prepare(DIMENSION_VALUE_BY_METRIC_CQL) - or_dimensions.append(list(map(f, value.split('|')))) + self.dim_val_stmt = self.session.prepare(DIMENSION_VALUE_CQL) - else: - sub_dimensions[key] = value + self.dim_name_by_metric_stmt = self.session.prepare(DIMENSION_NAME_BY_METRIC_CQL) - if or_dimensions: - or_dims_list = list(itertools.product(*or_dimensions)) - metrics_list = [] - - for or_dims_tuple in or_dims_list: - extracted_dimensions = sub_dimensions.copy() - - for dims in iter(or_dims_tuple): - for k, v in dims.iteritems(): - extracted_dimensions[k] = v - - metrics = self._list_metrics(tenant_id, region, name, - extracted_dimensions, offset, - limit, start_timestamp, - end_timestamp, - include_metric_hash) - metrics_list += metrics - - return sorted(metrics_list, key=lambda metric: metric['id']) - - return self._list_metrics(tenant_id, region, name, dimensions, - offset, limit, start_timestamp, - end_timestamp, include_metric_hash) - - def _list_metrics(self, tenant_id, region, name, dimensions, offset, - limit, start_timestamp=None, end_timestamp=None, - include_metric_hash=False): + self.dim_name_stmt = self.session.prepare(DIMENSION_NAME_CQL) - try: + self.metric_name_by_dimension_stmt = self.session.prepare(METRIC_NAME_BY_DIMENSION_CQL) - select_stmt = """ - select tenant_id, region, metric_hash, metric_map - from metric_map - where tenant_id = %s and region = %s - """ + self.metric_name_by_dimension_offset_stmt = self.session.prepare(METRIC_NAME_BY_DIMENSION_OFFSET_CQL) - parms = [tenant_id.encode('utf8'), region.encode('utf8')] + self.metric_name_stmt = self.session.prepare(METRIC_NAME_CQL) - name_clause = self._build_name_clause(name, parms) + self.metric_name_offset_stmt = self.session.prepare(METRIC_NAME_OFFSET_CQL) - dimension_clause = self._build_dimensions_clause(dimensions, parms) + self.metric_by_id_stmt = self.session.prepare(METRIC_BY_ID_CQL) - select_stmt += name_clause + dimension_clause + except Exception as ex: + LOG.exception(ex) + raise exceptions.RepositoryException(ex) - if offset: - select_stmt += ' and metric_hash > %s ' - parms.append(bytearray(offset.decode('hex'))) + self.epoch = datetime.utcfromtimestamp(0) - if limit: - select_stmt += ' limit %s ' - parms.append(limit + 1) + def list_dimension_values(self, tenant_id, region, metric_name, + dimension_name): - select_stmt += ' allow filtering ' + try: + if metric_name: + rows = self.session.execute( + self.dim_val_by_metric_stmt, + [region, tenant_id, metric_name, dimension_name]) + else: + rows = self.session.execute( + self.dim_val_stmt, + [region, tenant_id, dimension_name]) - json_metric_list = [] + except Exception as ex: + LOG.exception(ex) + raise exceptions.RepositoryException(ex) - stmt = SimpleStatement(select_stmt, - fetch_size=2147483647) + json_dim_value_list = [] - rows = self.cassandra_session.execute(stmt, parms) + if not rows: + return json_dim_value_list - if not rows: - return json_metric_list + for row in rows: + json_dim_value_list.append({u'dimension_value': row.value}) - for (tenant_id, region, metric_hash, metric_map) in rows: + json_dim_value_list.sort(key=lambda x: x[u'dimension_value']) - metric = {} + return json_dim_value_list - dimensions = {} + def list_dimension_names(self, tenant_id, region, metric_name): - if include_metric_hash: - metric[u'metric_hash'] = metric_hash + try: + if metric_name: + rows = self.session.execute( + self.dim_name_by_metric_stmt, + [region, tenant_id, metric_name]) + ordered = True + else: + rows = self.session.execute( + self.dim_name_stmt, + [region, tenant_id]) + ordered = False - for name, value in metric_map.iteritems(): + except Exception as ex: + LOG.exception(ex) + raise exceptions.RepositoryException(ex) - if name == '__name__': + if not rows: + return [] - name = urllib.unquote_plus(value) + json_dim_name_list = [{u'dimension_name': row.name} for row in rows] - metric[u'name'] = name + if not ordered: + json_dim_name_list.sort(key=lambda x: x[u'dimension_name']) - else: + return json_dim_name_list - name = urllib.unquote_plus(name) + def list_metrics(self, tenant_id, region, name, dimensions, offset, limit, start_time=None, + end_time=None): - value = urllib.unquote_plus(value) + offset_name = None + offset_dimensions = [] + names = [] + metric_list = [] + offset_futures = [] + non_offset_futures = [] - dimensions[name] = value + try: + if offset: + offset_metric = self._get_metric_by_id(offset) + if offset_metric: + offset_name = offset_metric.name + offset_dimensions = offset_metric.dimensions + + if not name: + names = self._list_metric_names(tenant_id, region, dimensions, offset=offset_name) + if names: + names = [elem['name'] for elem in names] + else: + names.append(name) - metric[u'dimensions'] = dimensions + if not names: + return metric_list - try: - metric[u'id'] = binascii.hexlify(bytearray(metric_hash)) + for name in names: + if name == offset_name: + futures = self._list_metrics_by_name(tenant_id, region, name, dimensions, offset_dimensions, + limit, start_time=None, end_time=None) + if offset_dimensions and dimensions: + offset_futures.extend(futures) + else: + non_offset_futures.extend(futures) + else: + non_offset_futures.extend( + self._list_metrics_by_name(tenant_id, region, name, dimensions, None, limit, + start_time=None, end_time=None)) + + # manually filter out metrics by the offset dimension + for future in offset_futures: + rows = future.result() + for row in rows: + if offset_dimensions >= row.dimensions: + continue - except TypeError as terr: - LOG.exception("metric_hash missing, using None as " - "metric[id]. ({})".format(terr)) - metric[u'id'] = None + metric_list.append(self._process_metric_row(row)) - json_metric_list.append(metric) + for future in non_offset_futures: + metric_list.extend((self._process_metric_row(row) for row in future.result())) - return json_metric_list + return metric_list except Exception as ex: LOG.exception(ex) raise exceptions.RepositoryException(ex) - def _build_dimensions_clause(self, dimensions, parms): + @staticmethod + def _process_metric_row(row): + dim_map = {} + for d in row.dimensions: + pair = d.split('\t') + dim_map[pair[0]] = pair[1] + + if row.metric_id is None: + LOG.error( + 'Metric is missing metric_id, using metric_id=None' + ' name: {}, dimensions: {}'.format( + row.metric_name, row.dimensions)) + return {'id': None, + 'name': row.metric_name, + 'dimensions': dim_map} + + metric = {'id': binascii.hexlify(bytearray(row.metric_id)), + 'name': row.metric_name, + 'dimensions': dim_map} - dimension_clause = '' - if dimensions: + return metric - for name, value in dimensions.iteritems(): - if not value: - dimension_clause += ' and metric_map contains key %s ' + def _list_metrics_by_name(self, tenant_id, region, name, dimensions, dimension_offset, limit, start_time=None, + end_time=None): - parms.append(urllib.quote_plus(name).encode('utf8')) - else: - dimension_clause += ' and metric_map[%s] = %s ' + or_dimensions = [] + sub_dimensions = {} + futures = [] - parms.append(urllib.quote_plus(name).encode('utf8')) - parms.append(urllib.quote_plus(value).encode('utf8')) - return dimension_clause + if not dimensions: + query = self._build_metrics_by_name_query(tenant_id, region, name, dimensions, None, start_time, + end_time, dimension_offset, limit) + futures.append(self.session.execute_async(query[0], query[1])) + return futures - def _build_name_clause(self, name, parms): + wildcard_dimensions = [] + for dim_name, dim_value in dimensions.items(): + if not dim_value: + wildcard_dimensions.append(dim_name) - name_clause = '' - if name: - name_clause = ' and metric_map[%s] = %s ' + elif '|' in dim_value: - parms.append(urllib.quote_plus('__name__').encode('utf8')) - parms.append(urllib.quote_plus(name).encode('utf8')) + def f(val): + return {dim_name: val} - return name_clause + or_dimensions.append(list(map(f, sorted(dim_value.split('|'))))) - def _build_select_metric_map_query(self, tenant_id, region, parms): + else: + sub_dimensions[dim_name] = dim_value - select_stmt = """ - select metric_map - from metric_map - where tenant_id = %s and region = %s - """ + if or_dimensions: + or_dims_list = list(itertools.product(*or_dimensions)) - parms.append(tenant_id.encode('utf8')) - parms.append(region.encode('utf8')) + for or_dims_tuple in or_dims_list: + extracted_dimensions = sub_dimensions.copy() - return select_stmt + for dims in iter(or_dims_tuple): + for k, v in dims.iteritems(): + extracted_dimensions[k] = v - def measurement_list(self, tenant_id, region, name, dimensions, - start_timestamp, end_timestamp, offset, - limit, merge_metrics_flag): + query = self._build_metrics_by_name_query(tenant_id, region, name, extracted_dimensions, + wildcard_dimensions, start_time, + end_time, dimension_offset, limit) - try: + futures.append(self.session.execute_async(query[0], query[1])) - json_measurement_list = [] + else: + query = self._build_metrics_by_name_query(tenant_id, region, name, sub_dimensions, wildcard_dimensions, + start_time, + end_time, dimension_offset, limit) + futures.append(self.session.execute_async(query[0], query[1])) - rows = self._get_measurements(tenant_id, region, name, dimensions, - start_timestamp, end_timestamp, - offset, limit, merge_metrics_flag) + return futures - if not rows: - return json_measurement_list + def _get_metric_by_id(self, metric_id): - if not merge_metrics_flag: - dimensions = self._get_dimensions(tenant_id, region, name, dimensions) + rows = self.session.execute(self.metric_by_id_stmt, [bytearray.fromhex(metric_id)]) - measurements_list = ( - [[self._isotime_msec(time_stamp), - value, - rest_utils.from_json(value_meta) if value_meta else {}] - for (time_stamp, value, value_meta) in rows]) + if rows: + return Metric(id=metric_id, name=rows[0].metric_name, dimensions=rows[0].dimensions) - measurement = {u'name': name, - # The last date in the measurements list. - u'id': measurements_list[-1][0], - u'dimensions': dimensions, - u'columns': [u'timestamp', u'value', u'value_meta'], - u'measurements': measurements_list} + return None - json_measurement_list.append(measurement) + def _build_metrics_by_name_query(self, tenant_id, region, name, dimensions, wildcard_dimensions, start_time, + end_time, dim_offset, + limit): - return json_measurement_list + conditions = [REGION_EQ, TENANT_EQ] + params = [region, tenant_id.encode('utf8')] - except exceptions.RepositoryException as ex: - LOG.exception(ex) - raise ex - - except Exception as ex: - LOG.exception(ex) - raise exceptions.RepositoryException(ex) + if name: + conditions.append(METRIC_NAME_EQ) + params.append(name) + else: + conditions.append('') - def _get_measurements(self, tenant_id, region, name, dimensions, - start_timestamp, end_timestamp, offset, limit, - merge_metrics_flag): - - metric_list = self.list_metrics(tenant_id, region, name, - dimensions, None, None, - start_timestamp, end_timestamp, - include_metric_hash=True) - if not metric_list: - return None + if dimensions: + conditions.append(DIMENSIONS_CONTAINS * len(dimensions)) + params.extend( + [self._create_dimension_value_entry(dim_name, dim_value) + for dim_name, dim_value in dimensions.items()]) + else: + conditions.append('') - if len(metric_list) > 1: + if wildcard_dimensions: + conditions.append(DIMENSIONS_NAME_CONTAINS * len(wildcard_dimensions)) + params.extend(wildcard_dimensions) + else: + conditions.append('') - if not merge_metrics_flag: - raise exceptions.MultipleMetricsException( - self.MULTIPLE_METRICS_MESSAGE) + if dim_offset and not dimensions: + # cassandra does not allow using both contains and GT in collection column + conditions.append(DIMENSIONS_GT) + params.append(dim_offset) + else: + conditions.append('') - select_stmt = """ - select time_stamp, value, value_meta - from measurements - where tenant_id = %s and region = %s - """ + if start_time: + conditions.append(UPDATED_TIME_GE % start_time) + else: + conditions.append('') - parms = [tenant_id.encode('utf8'), region.encode('utf8')] + if end_time: + conditions.append(CREATED_TIME_LE % end_time) + else: + conditions.append('') - metric_hash_list = [bytearray(metric['metric_hash']) for metric in - metric_list] + if limit: + conditions.append(LIMIT_CLAUSE) + params.append(limit) + else: + conditions.append('') - place_holders = ["%s"] * len(metric_hash_list) + if (not name) or dimensions or wildcard_dimensions or start_time or end_time: + conditions.append(ALLOW_FILTERING) + else: + conditions.append('') - in_clause = ' and metric_hash in ({}) '.format(",".join(place_holders)) + return METRIC_LIST_CQL % tuple(conditions), params - select_stmt += in_clause + @staticmethod + def _create_dimension_value_entry(name, value): + return '%s\t%s' % (name, value) - parms.extend(metric_hash_list) + def list_metric_names(self, tenant_id, region, dimensions): + return self._list_metric_names(tenant_id, region, dimensions) - if offset: + def _list_metric_names(self, tenant_id, region, dimensions, offset=None): - select_stmt += ' and time_stamp > %s ' - parms.append(offset) + or_dimensions = [] + single_dimensions = {} - elif start_timestamp: + if dimensions: + for key, value in dimensions.items(): + if not value: + continue - select_stmt += ' and time_stamp >= %s ' - parms.append(int(start_timestamp * 1000)) + elif '|' in value: + def f(val): + return {key: val} - if end_timestamp: - select_stmt += ' and time_stamp <= %s ' - parms.append(int(end_timestamp * 1000)) + or_dimensions.append(list(map(f, sorted(value.split('|'))))) - select_stmt += ' order by time_stamp ' + else: + single_dimensions[key] = value - if limit: - select_stmt += ' limit %s ' - parms.append(limit + 1) + if or_dimensions: - stmt = SimpleStatement(select_stmt, - fetch_size=2147483647) - rows = self.cassandra_session.execute(stmt, parms) + names = [] + or_dims_list = list(itertools.product(*or_dimensions)) - return rows + for or_dims_tuple in or_dims_list: + extracted_dimensions = single_dimensions.copy() - def _get_dimensions(self, tenant_id, region, name, dimensions): - metrics_list = self.list_metrics(tenant_id, region, name, - dimensions, None, 2) + for dims in iter(or_dims_tuple): + for k, v in dims.iteritems(): + extracted_dimensions[k] = v - if len(metrics_list) > 1: - raise exceptions.MultipleMetricsException(self.MULTIPLE_METRICS_MESSAGE) + names.extend( + self._list_metric_names_single_dimension_value(tenant_id, region, extracted_dimensions, offset)) - if not metrics_list: - return {} + names.sort(key=lambda x: x[u'name']) + return names - return metrics_list[0]['dimensions'] + else: + names = self._list_metric_names_single_dimension_value(tenant_id, region, single_dimensions, offset) + names.sort(key=lambda x: x[u'name']) + return names - def list_metric_names(self, tenant_id, region, dimensions): + def _list_metric_names_single_dimension_value(self, tenant_id, region, dimensions, offset=None): try: + futures = [] + if dimensions: + for name, value in dimensions.items(): + if offset: + futures.append(self.session.execute_async(self.metric_name_by_dimension_offset_stmt, + [region, tenant_id, name, value, offset])) + else: + futures.append(self.session.execute_async(self.metric_name_by_dimension_stmt, + [region, tenant_id, name, value])) - parms = [] - - query = self._build_select_metric_map_query(tenant_id, region, parms) + else: + if offset: + futures.append( + self.session.execute_async(self.metric_name_offset_stmt, [region, tenant_id, offset])) + else: + futures.append(self.session.execute_async(self.metric_name_stmt, [region, tenant_id])) - dimension_clause = self._build_dimensions_clause(dimensions, parms) + names_list = [] - query += dimension_clause + for future in futures: + rows = future.result() + tmp = set() + for row in rows: + tmp.add(row.metric_name) - stmt = SimpleStatement(query, - fetch_size=2147483647) + names_list.append(tmp) - rows = self.cassandra_session.execute(stmt, parms) + return [{u'name': v} for v in set.intersection(*names_list)] - json_name_list = [] + except Exception as ex: + LOG.exception(ex) + raise exceptions.RepositoryException(ex) - if not rows: - return json_name_list + def measurement_list(self, tenant_id, region, name, dimensions, + start_timestamp, end_timestamp, offset, limit, + merge_metrics_flag, group_by): - for row in rows: + metrics = self.list_metrics(tenant_id, region, name, dimensions, None, None) - metric_map = row.metric_map - for name, value in metric_map.iteritems(): + if offset: + tmp = offset.split("_") + if len(tmp) > 1: + offset_id = tmp[0] + offset_timestamp = tmp[1] + else: + offset_id = None + offset_timestamp = offset + else: + offset_timestamp = None + offset_id = None - if name == '__name__': - value = urllib.unquote_plus(value) - metric_name = {u'name': value} + if not metrics: + return None + elif len(metrics) > 1: + if not merge_metrics_flag and not group_by: + raise exceptions.MultipleMetricsException(self.MULTIPLE_METRICS_MESSAGE) - if metric_name not in json_name_list: - json_name_list.append(metric_name) + try: + if len(metrics) > 1 and not group_by: + # offset is controlled only by offset_timestamp when the group by option is not enabled + count, series_list = self._query_merge_measurements(metrics, + dimensions, + start_timestamp, + end_timestamp, + offset_timestamp, + limit) + return series_list + + if group_by: + if not isinstance(group_by, list): + group_by = group_by.split(',') + elif len(group_by) == 1: + group_by = group_by[0].split(',') + + if len(metrics) == 1 or group_by[0].startswith('*'): + if offset_id: + for index, metric in enumerate(metrics): + if metric['id'] == offset_id: + if index > 0: + metrics[0:index] = [] + break + + count, series_list = self._query_measurements(metrics, + start_timestamp, + end_timestamp, + offset_timestamp, + limit) + + return series_list + + grouped_metrics = self._group_metrics(metrics, group_by, dimensions) + + if not grouped_metrics or len(grouped_metrics) == 0: + return None + + if offset_id: + found_offset = False + for outer_index, sublist in enumerate(grouped_metrics): + for inner_index, metric in enumerate(sublist): + if metric['id'] == offset_id: + found_offset = True + if inner_index > 0: + sublist[0:inner_index] = [] + break + if found_offset: + if outer_index > 0: + grouped_metrics[0:outer_index] = [] + break + remaining = limit + series_list = [] + for sublist in grouped_metrics: + sub_count, results = self._query_merge_measurements(sublist, + sublist[0]['dimensions'], + start_timestamp, + end_timestamp, + offset_timestamp, + remaining) + + series_list.extend(results) + + if remaining: + remaining -= sub_count + if remaining <= 0: break - return sorted(json_name_list) + # offset_timestamp is used only in the first group, reset to None for subsequent groups + if offset_timestamp: + offset_timestamp = None + + return series_list except Exception as ex: LOG.exception(ex) raise exceptions.RepositoryException(ex) - def metrics_statistics(self, tenant_id, region, name, dimensions, - start_timestamp, end_timestamp, statistics, - period, offset, limit, merge_metrics_flag): + def _query_merge_measurements(self, metrics, dimensions, start_timestamp, end_timestamp, + offset_timestamp, limit): + results = [] + for metric in metrics: + if limit and len(metrics) > 1: + fetch_size = min(limit, max(1000, limit / len(metrics) + 2)) + else: + fetch_size = None + query = self._build_measurement_query(metric['id'], + start_timestamp, + end_timestamp, + offset_timestamp, + limit, + fetch_size) + results.append((metric, iter(self.session.execute_async(query[0], query[1]).result()))) + + return self._merge_series(results, dimensions, limit) + + def _query_measurements(self, metrics, start_timestamp, end_timestamp, + offset_timestamp, limit): + results = [] + for index, metric in enumerate(metrics): + if index == 0: + query = self._build_measurement_query(metric['id'], + start_timestamp, + end_timestamp, + offset_timestamp, + limit) + else: + if limit: + fetch_size = min(self.session.default_fetch_size, + max(1000, limit / min(index, 4))) + else: + fetch_size = self.session.default_fetch_size + query = self._build_measurement_query(metric['id'], + start_timestamp, + end_timestamp, + None, + limit, + fetch_size) + + results.append([metric, + iter(self.session.execute_async(query[0], query[1]).result())]) + + series_list = [] + count = 0 + for result in results: + measurements = [] + row = next(result[1], None) + while row: + measurements.append([self._isotime_msec(row.time_stamp), + row.value, + rest_utils.from_json(row.value_meta) if row.value_meta else {}]) + count += 1 + if limit and count >= limit: + break + + row = next(result[1], None) + + series_list.append({'name': result[0]['name'], + 'id': result[0]['id'], + 'columns': ['timestamp', 'value', 'value_meta'], + 'measurements': measurements, + 'dimensions': result[0]['dimensions']}) + if limit and count >= limit: + break - try: + return count, series_list - if not period: - period = 300 - period = int(period) + @staticmethod + def _build_measurement_query(metric_id, start_timestamp, + end_timestamp, offset_timestamp, + limit=None, fetch_size=FETCH_SIZE_UNSET): + conditions = [METRIC_ID_EQ] + params = [bytearray.fromhex(metric_id)] + + if offset_timestamp: + conditions.append(OFFSET_TIME_GT) + params.append(offset_timestamp) + elif start_timestamp: + conditions.append(START_TIME_GE) + params.append(int(start_timestamp * 1000)) + else: + conditions.append('') - if offset: - if '_' in offset: - tmp = datetime.strptime(str(offset).split('_')[1], "%Y-%m-%dT%H:%M:%SZ") - tmp = tmp + timedelta(seconds=int(period)) - # Leave out any ID as cassandra doesn't understand it - offset = tmp.isoformat() - else: - tmp = datetime.strptime(offset, "%Y-%m-%dT%H:%M:%SZ") - offset = tmp + timedelta(seconds=int(period)) + if end_timestamp: + conditions.append(END_TIME_LE) + params.append(int(end_timestamp * 1000)) + else: + conditions.append('') + + if limit: + conditions.append(LIMIT_CLAUSE) + params.append(limit) + else: + conditions.append('') - rows = self._get_measurements(tenant_id, region, name, dimensions, - start_timestamp, end_timestamp, - offset, limit, merge_metrics_flag) + return SimpleStatement(MEASUREMENT_LIST_CQL % tuple(conditions), fetch_size=fetch_size), params - json_statistics_list = [] + def _merge_series(self, series, dimensions, limit): + series_list = [] - if not rows: - return json_statistics_list + if not series: + return series_list + + measurements = [] + top_batch = [] + num_series = len(series) + for i in range(0, num_series): + row = next(series[i][1], None) + if row: + top_batch.append([i, + row.time_stamp, + row.value, + rest_utils.from_json(row.value_meta) if row.value_meta else {}]) + else: + num_series -= 1 - requested_statistics = [stat.lower() for stat in statistics] + top_batch.sort(key=lambda m: m[1], reverse=True) - columns = [u'timestamp'] + count = 0 + while (not limit or count < limit) and top_batch: + measurements.append([self._isotime_msec(top_batch[num_series - 1][1]), + top_batch[num_series - 1][2], + top_batch[num_series - 1][3]]) + count += 1 + row = next(series[top_batch[num_series - 1][0]][1], None) + if row: + top_batch[num_series - 1] = [top_batch[num_series - 1][0], + row.time_stamp, + row.value, + rest_utils.from_json(row.value_meta) if row.value_meta else {}] - if 'avg' in requested_statistics: - columns.append(u'avg') + top_batch.sort(key=lambda m: m[1], reverse=True) + else: + num_series -= 1 + top_batch.pop() - if 'min' in requested_statistics: - columns.append(u'min') + series_list.append({'name': series[0][0]['name'], + 'id': series[0][0]['id'], + 'columns': ['timestamp', 'value', 'value_meta'], + 'measurements': measurements, + 'dimensions': dimensions}) - if 'max' in requested_statistics: - columns.append(u'max') + return count, series_list - if 'count' in requested_statistics: - columns.append(u'count') + @staticmethod + def _group_metrics(metrics, group_by, search_by): - if 'sum' in requested_statistics: - columns.append(u'sum') + grouped_metrics = {} + for metric in metrics: + key = '' + display_dimensions = dict(search_by.items()) + for name in group_by: + # '_' ensures te key with missing dimension is sorted lower + value = metric['dimensions'].get(name, '_') + if value != '_': + display_dimensions[name] = value + key = key + '='.join((urllib.quote_plus(name), urllib.quote_plus(value))) + '&' - first_row = rows[0] - stats_count = 0 - stats_sum = 0 - stats_max = first_row.value - stats_min = first_row.value - start_period = first_row.time_stamp + metric['dimensions'] = display_dimensions - stats_list = [] + if key in grouped_metrics: + grouped_metrics[key].append(metric) + else: + grouped_metrics[key] = [metric] - start_datetime = datetime.utcfromtimestamp(start_timestamp) - if offset and offset > start_datetime: - tmp_start_period = offset - else: - tmp_start_period = start_datetime - - while start_period >= tmp_start_period + timedelta(seconds=period): - stat = [ - tmp_start_period.strftime('%Y-%m-%dT%H:%M:%SZ') - .decode('utf8') - ] - for _statistics in requested_statistics: - stat.append(0) - tmp_start_period += timedelta(seconds=period) - stats_list.append(stat) + grouped_metrics = grouped_metrics.items() + grouped_metrics.sort(key=lambda k: k[0]) + return [x[1] for x in grouped_metrics] + + @staticmethod + def _isotime_msec(timestamp): + """Stringify datetime in ISO 8601 format + millisecond. + """ + st = timestamp.isoformat() + if '.' in st: + st = st[:23] + 'Z' + else: + st += '.000Z' + return st.decode('utf8') - for (time_stamp, value, value_meta) in rows: + def metrics_statistics(self, tenant_id, region, name, dimensions, + start_timestamp, end_timestamp, statistics, + period, offset, limit, merge_metrics_flag, + group_by): - if (time_stamp - start_period).seconds >= period: + if not period: + period = 300 + else: + period = int(period) - stat = [ - start_period.strftime('%Y-%m-%dT%H:%M:%SZ').decode( - 'utf8')] + series_list = self.measurement_list(tenant_id, region, name, dimensions, + start_timestamp, end_timestamp, + offset, None, merge_metrics_flag, group_by) - if 'avg' in requested_statistics: - stat.append(stats_sum / stats_count) + json_statistics_list = [] - if 'min' in requested_statistics: - stat.append(stats_min) + if not series_list: + return json_statistics_list - stats_min = value + statistics = [stat.lower() for stat in statistics] - if 'max' in requested_statistics: - stat.append(stats_max) + columns = [u'timestamp'] - stats_max = value + columns.extend([x for x in ['avg', 'min', 'max', 'count', 'sum'] if x in statistics]) - if 'count' in requested_statistics: - stat.append(stats_count) + start_time = datetime.utcfromtimestamp(start_timestamp) + if end_timestamp: + end_time = datetime.utcfromtimestamp(end_timestamp) + else: + end_time = datetime.utcnow() - if 'sum' in requested_statistics: - stat.append(stats_sum) + for series in series_list: - stats_list.append(stat) + if limit <= 0: + break - tmp_start_period = start_period + timedelta(seconds=period) - while time_stamp > tmp_start_period: - stat = [ - tmp_start_period.strftime('%Y-%m-%dT%H:%M:%SZ') - .decode('utf8') - ] - for _statistics in requested_statistics: - stat.append(0) - tmp_start_period += timedelta(seconds=period) - stats_list.append(stat) - - start_period = time_stamp - - stats_sum = 0 - stats_count = 0 - - stats_count += 1 - stats_sum += value - - if 'min' in requested_statistics: + measurements = series['measurements'] - if value < stats_min: - stats_min = value + if not measurements: + continue - if 'max' in requested_statistics: + first_measure = measurements[0] + first_measure_start_time = MetricsRepository._parse_time_string(first_measure[0]) - if value > stats_max: - stats_max = value + # skip blank intervals at the beginning, finds the start time of stat period that is not empty + stat_start_time = start_time + timedelta( + seconds=((first_measure_start_time - start_time).seconds / period) * period) - if stats_count: + stats_list = [] + stats_count = 0 + stats_sum = 0 + stats_min = stats_max = first_measure[1] + + for measurement in series['measurements']: - stat = [start_period.strftime('%Y-%m-%dT%H:%M:%SZ').decode( - 'utf8')] + time_stamp = MetricsRepository._parse_time_string(measurement[0]) + value = measurement[1] - if 'avg' in requested_statistics: - stat.append(stats_sum / stats_count) + if (time_stamp - stat_start_time).seconds >= period: - if 'min' in requested_statistics: - stat.append(stats_min) + stat = MetricsRepository._create_stat(statistics, stat_start_time, stats_count, + stats_sum, stats_min, stats_max) - if 'max' in requested_statistics: - stat.append(stats_max) + stats_list.append(stat) + limit -= 1 + if limit <= 0: + break - if 'count' in requested_statistics: - stat.append(stats_count) + # initialize the new stat period + stats_sum = value + stats_count = 1 + stats_min = value + stats_max = value + stat_start_time += timedelta(seconds=period) - if 'sum' in requested_statistics: - stat.append(stats_sum) + else: + stats_min = min(stats_min, value) + stats_max = max(stats_max, value) + stats_count += 1 + stats_sum += value + if stats_count: + stat = MetricsRepository._create_stat(statistics, stat_start_time, stats_count, stats_sum, + stats_min, stats_max) stats_list.append(stat) + limit -= 1 - if end_timestamp: - time_stamp = datetime.utcfromtimestamp(end_timestamp) - else: - time_stamp = datetime.now() - tmp_start_period = start_period + timedelta(seconds=period) - while time_stamp > tmp_start_period: - stat = [ - tmp_start_period.strftime('%Y-%m-%dT%H:%M:%SZ') - .decode('utf8') - ] - for _statistics in requested_statistics: - stat.append(0) - tmp_start_period += timedelta(seconds=period) - stats_list.append(stat) + stats_end_time = stat_start_time + timedelta(seconds=period) - timedelta(milliseconds=1) + if stats_end_time > end_time: + stats_end_time = end_time statistic = {u'name': name.decode('utf8'), - # The last date in the stats list. - u'id': stats_list[-1][0], - u'dimensions': dimensions, + u'id': series['id'], + u'dimensions': series['dimensions'], u'columns': columns, - u'statistics': stats_list} + u'statistics': stats_list, + u'end_time': self._isotime_msec(stats_end_time)} json_statistics_list.append(statistic) - return json_statistics_list + return json_statistics_list - except exceptions.RepositoryException as ex: - LOG.exception(ex) - raise ex + @staticmethod + def _create_stat(statistics, timestamp, stat_count=None, stat_sum=None, stat_min=None, stat_max=None): - except Exception as ex: - LOG.exception(ex) - raise exceptions.RepositoryException(ex) + stat = [MetricsRepository._isotime_msec(timestamp)] + + if not stat_count: + stat.extend([0] * len(statistics)) + + else: + if 'avg' in statistics: + stat.append(stat_sum / stat_count) + + if 'min' in statistics: + stat.append(stat_min) + + if 'max' in statistics: + stat.append(stat_max) + + if 'count' in statistics: + stat.append(stat_count) + + if 'sum' in statistics: + stat.append(stat_sum) + + return stat + + @staticmethod + def _parse_time_string(timestamp): + dt = timeutils.parse_isotime(timestamp) + dt = timeutils.normalize_time(dt) + return dt def alarm_history(self, tenant_id, alarm_id_list, offset, limit, start_timestamp=None, @@ -576,55 +905,47 @@ class MetricsRepository(metrics_reposito if not alarm_id_list: return json_alarm_history_list - select_stmt = """ - select alarm_id, time_stamp, metrics, new_state, old_state, - reason, reason_data, sub_alarms, tenant_id - from alarm_state_history - where tenant_id = %s - """ - - parms = [tenant_id.encode('utf8')] - - place_holders = ["%s"] * len(alarm_id_list) - - in_clause = ' and alarm_id in ({}) '.format( - ",".join(place_holders)) - - select_stmt += in_clause - - parms.extend(alarm_id_list) - - if offset and offset != '0': + conditions = [ALARM_TENANT_ID_EQ] + params = [tenant_id.encode('utf8')] + if len(alarm_id_list) == 1: + conditions.append(ALARM_ID_EQ) + params.append(alarm_id_list[0]) + else: + conditions.append(' and alarm_id in ({}) '.format(','.join(['%s'] * len(alarm_id_list)))) + for alarm_id in alarm_id_list: + params.append(alarm_id) - select_stmt += ' and time_stamp > %s ' - dt = timeutils.normalize_time(timeutils.parse_isotime(offset)) - parms.append(self._get_millis_from_timestamp(dt)) + if offset: + conditions.append(OFFSET_TIME_GT) + params.append(offset) elif start_timestamp: - - select_stmt += ' and time_stamp >= %s ' - parms.append(int(start_timestamp * 1000)) + conditions.append(START_TIME_GE) + params.append(int(start_timestamp * 1000)) + else: + conditions.append('') if end_timestamp: - select_stmt += ' and time_stamp <= %s ' - parms.append(int(end_timestamp * 1000)) + conditions.append(END_TIME_LE) + params.append(int(end_timestamp * 1000)) + else: + conditions.append('') if limit: - select_stmt += ' limit %s ' - parms.append(limit + 1) - - stmt = SimpleStatement(select_stmt, - fetch_size=2147483647) + conditions.append(LIMIT_CLAUSE) + params.append(limit + 1) + else: + conditions.append('') - rows = self.cassandra_session.execute(stmt, parms) + rows = self.session.execute(ALARM_HISTORY_CQL % tuple(conditions), params) if not rows: return json_alarm_history_list sorted_rows = sorted(rows, key=lambda row: row.time_stamp) - for (alarm_id, time_stamp, metrics, new_state, old_state, reason, - reason_data, sub_alarms, tenant_id) in sorted_rows: + for (tenant_id, alarm_id, time_stamp, metrics, new_state, old_state, reason, + reason_data, sub_alarms) in sorted_rows: alarm = {u'timestamp': self._isotime_msec(time_stamp), u'alarm_id': alarm_id, @@ -634,10 +955,10 @@ class MetricsRepository(metrics_reposito u'reason': reason, u'reason_data': reason_data, u'sub_alarms': rest_utils.from_json(sub_alarms), - u'id': str(self._get_millis_from_timestamp(time_stamp) - ).decode('utf8')} + u'id': str(int((time_stamp - self.epoch).total_seconds() * 1000))} if alarm[u'sub_alarms']: + for sub_alarm in alarm[u'sub_alarms']: sub_expr = sub_alarm['sub_alarm_expression'] metric_def = sub_expr['metric_definition'] @@ -654,100 +975,14 @@ class MetricsRepository(metrics_reposito raise exceptions.RepositoryException(ex) @staticmethod - def _isotime_msec(timestamp): - """Stringify datetime in ISO 8601 format + millisecond. - """ - st = timestamp.isoformat() - if '.' in st: - st = st[:23] + 'Z' - else: - st += '.000Z' - return st.decode('utf8') - - @staticmethod - def _get_millis_from_timestamp(dt): - dt = timeutils.normalize_time(dt) - return int((dt - datetime(1970, 1, 1)).total_seconds() * 1000) - - def list_dimension_values(self, tenant_id, region, metric_name, - dimension_name): - + def check_status(): try: - - parms = [] - - query = self._build_select_metric_map_query(tenant_id, region, parms) - - name_clause = self._build_name_clause(metric_name, parms) - - dimensions = {dimension_name: None} - - dimension_clause = self._build_dimensions_clause(dimensions, parms) - - query += name_clause + dimension_clause - - query += ' allow filtering ' - - stmt = SimpleStatement(query, - fetch_size=2147483647) - - rows = self.cassandra_session.execute(stmt, parms) - - json_dim_value_list = [] - - if not rows: - return json_dim_value_list - - for row in rows: - - metric_map = row.metric_map - for name, value in metric_map.iteritems(): - - name = urllib.unquote_plus(name) - value = urllib.unquote_plus(value) - dim_value = {u'dimension_value': value} - - if name == dimension_name and dim_value not in json_dim_value_list: - json_dim_value_list.append(dim_value) - - return sorted(json_dim_value_list) - + cluster = Cluster( + CONF.cassandra.contact_points + ) + session = cluster.connect(CONF.cassandra.keyspace) + session.shutdown() except Exception as ex: - LOG.exception(ex) - raise exceptions.RepositoryException(ex) - - def list_dimension_names(self, tenant_id, region, metric_name): - - try: - - parms = [] - - query = self._build_select_metric_map_query(tenant_id, region, parms) - - name_clause = self._build_name_clause(metric_name, parms) - - query += name_clause - - stmt = SimpleStatement(query, - fetch_size=2147483647) - - rows = self.cassandra_session.execute(stmt, parms) - - json_dim_name_list = [] - - for row in rows: - - metric_map = row.metric_map - for name, value in metric_map.iteritems(): - - name = urllib.unquote_plus(name) - dim_name = {u'dimension_name': name} - - if name != '__name__' and dim_name not in json_dim_name_list: - json_dim_name_list.append(dim_name) - - return sorted(json_dim_name_list) - - except Exception as ex: - LOG.exception(ex) - raise exceptions.RepositoryException(ex) + LOG.exception(str(ex)) + return False, str(ex) + return True, 'OK' Index: monasca-api-2.2.1.dev26/monasca_api/common/repositories/influxdb/metrics_repository.py =================================================================== --- monasca-api-2.2.1.dev26.orig/monasca_api/common/repositories/influxdb/metrics_repository.py +++ monasca-api-2.2.1.dev26/monasca_api/common/repositories/influxdb/metrics_repository.py @@ -554,13 +554,21 @@ class MetricsRepository(metrics_reposito json_measurement_list = [] + offset_id = 0 + offset_timestamp = offset + + if offset and "_" in offset: + offset_id_str, _, offset_timestamp = offset.partition('_') + offset_id = int(offset_id_str) + try: + # the build query method apparently only considers offset timestamp. query = self._build_select_measurement_query(dimensions, name, tenant_id, region, start_timestamp, end_timestamp, - offset, group_by, + offset_timestamp, group_by, limit) if not group_by and not merge_metrics_flag: @@ -572,10 +580,6 @@ class MetricsRepository(metrics_reposito if not result: return json_measurement_list - offset_id = 0 - if offset is not None: - offset_tuple = offset.split('_') - offset_id = int(offset_tuple[0]) if len(offset_tuple) > 1 else 0 index = offset_id for serie in result.raw['series']: Index: monasca-api-2.2.1.dev26/monasca_api/conf/cassandra.py =================================================================== --- /dev/null +++ monasca-api-2.2.1.dev26/monasca_api/conf/cassandra.py @@ -0,0 +1,43 @@ +# Copyright 2014 IBM Corp. +# Copyright 2016-2017 FUJITSU LIMITED +# (C) Copyright 2016-2017 Hewlett Packard Enterprise Development LP +# (C) Copyright 2017 SUSE 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 oslo_config import cfg +from oslo_config import types + +cassandra_opts = [ + cfg.ListOpt('contact_points', + default=['127.0.0.1'], + item_type=types.HostAddress(), + help=''' +Comma separated list of Cassandra node IP addresses +'''), + cfg.StrOpt('keyspace', default='monasca', + help=''' +keyspace where metric are stored +''') +] + +cassandra_group = cfg.OptGroup(name='cassandra') + + +def register_opts(conf): + conf.register_group(cassandra_group) + conf.register_opts(cassandra_opts, cassandra_group) + + +def list_opts(): + return cassandra_group, cassandra_opts Index: monasca-api-2.2.1.dev26/monasca_api/healthcheck/metrics_db_check.py =================================================================== --- monasca-api-2.2.1.dev26.orig/monasca_api/healthcheck/metrics_db_check.py +++ monasca-api-2.2.1.dev26/monasca_api/healthcheck/metrics_db_check.py @@ -87,9 +87,7 @@ class MetricsDbCheck(base.BaseHealthChec if self._cluster is None: return False, "Cassandra driver not imported" try: - cassandra = self._cluster.Cluster( - CONF.cassandra.cluster_ip_addresses.split(',') - ) + cassandra = self._cluster.Cluster(CONF.cassandra.contact_points) session = cassandra.connect(CONF.cassandra.keyspace) session.shutdown() except Exception as ex: Index: monasca-api-2.2.1.dev26/monasca_api/tests/test_metrics_db_health_check.py =================================================================== --- monasca-api-2.2.1.dev26.orig/monasca_api/tests/test_metrics_db_health_check.py +++ monasca-api-2.2.1.dev26/monasca_api/tests/test_metrics_db_health_check.py @@ -29,7 +29,7 @@ CONF = cfg.CONF class TestMetricsDbHealthCheck(base.BaseTestCase): cassandra_conf = { - 'cluster_ip_addresses': 'localhost', + 'contact_points': 'localhost', 'keyspace': 'test' } @@ -154,7 +154,7 @@ class TestMetricsDbHealthCheck(base.Base 'metrics_driver': 'cassandra.metrics_repository:MetricsRepository' } cassandra_conf = { - 'cluster_ip_addresses': 'localhost', + 'contact_points': 'localhost', 'keyspace': 'test' } self._conf.config(group='repositories', **messaging_conf) @@ -178,7 +178,7 @@ class TestMetricsDbHealthCheck(base.Base 'metrics_driver': 'cassandra.metrics_repository:MetricsRepository' } cassandra_conf = { - 'cluster_ip_addresses': 'localhost', + 'contact_points': 'localhost', 'keyspace': 'test' } self._conf.config(group='repositories', **messaging_conf) @@ -199,7 +199,7 @@ class TestMetricsDbHealthCheck(base.Base 'metrics_driver': 'cassandra.metrics_repository:MetricsRepository' } cassandra_conf = { - 'cluster_ip_addresses': 'localhost', + 'contact_points': 'localhost', 'keyspace': 'test' } self._conf.config(group='repositories', **messaging_conf) @@ -208,4 +208,5 @@ class TestMetricsDbHealthCheck(base.Base db_health = tdc.MetricsDbCheck() result = db_health.health_check() + self.assertEqual("OK", result.message) self.assertTrue(result.healthy) Index: monasca-api-2.2.1.dev26/monasca_api/tests/test_repositories.py =================================================================== --- monasca-api-2.2.1.dev26.orig/monasca_api/tests/test_repositories.py +++ monasca-api-2.2.1.dev26/monasca_api/tests/test_repositories.py @@ -1,5 +1,7 @@ # Copyright 2015 Cray Inc. All Rights Reserved. # (C) Copyright 2016-2017 Hewlett Packard Enterprise Development LP +# Copyright 2017 Fujitsu LIMITED +# (C) Copyright 2017 SUSE 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 @@ -191,24 +193,27 @@ class TestRepoMetricsCassandra(testtools self._fixture_config = self.useFixture( fixture_config.Config(cfg.CONF)) - self._fixture_config.config(cluster_ip_addresses='127.0.0.1', + self._fixture_config.config(contact_points='127.0.0.1', group='cassandra') @patch("monasca_api.common.repositories.cassandra.metrics_repository.Cluster.connect") def test_list_metrics(self, cassandra_connect_mock): cassandra_session_mock = cassandra_connect_mock.return_value - cassandra_session_mock.execute.return_value = [[ - "0b5e7d8c43f74430add94fba09ffd66e", - "region", - binascii.unhexlify(b"01d39f19798ed27bbf458300bf843edd17654614"), - { - "__name__": "disk.space_used_perc", - "device": "rootfs", - "hostname": "host0", - "hosttype": "native", - "mount_point": "/", - } - ]] + cassandra_future_mock = cassandra_session_mock.execute_async.return_value + + Metric = namedtuple('Metric', 'metric_id metric_name dimensions') + + cassandra_future_mock.result.return_value = [ + Metric( + metric_id=binascii.unhexlify(b"01d39f19798ed27bbf458300bf843edd17654614"), + metric_name='disk.space_used_perc', + dimensions=[ + 'device\trootfs', + 'hostname\thost0', + 'hosttype\tnative', + 'mount_point\t/'] + ) + ] repo = cassandra_repo.MetricsRepository() @@ -234,21 +239,26 @@ class TestRepoMetricsCassandra(testtools u'hosttype': u'native' }}], result) - @patch("monasca_api.common.repositories.cassandra.metrics_repository.Cluster.connect") - def test_list_metrics_missing_hash(self, cassandra_connect_mock): + # As Cassandra allows sparse data, it is possible to have a missing metric_id + @patch("monasca_api.common.repositories.cassandra." + "metrics_repository.Cluster.connect") + def test_list_metrics_empty_metric_id(self, cassandra_connect_mock): cassandra_session_mock = cassandra_connect_mock.return_value - cassandra_session_mock.execute.return_value = [[ - "0b5e7d8c43f74430add94fba09ffd66e", - "region", - None, - { - "__name__": "disk.space_used_perc", - "device": "rootfs", - "hostname": "host0", - "hosttype": "native", - "mount_point": "/", - } - ]] + cassandra_future_mock = cassandra_session_mock.execute_async.return_value + + Metric = namedtuple('Metric', 'metric_id metric_name dimensions') + + cassandra_future_mock.result.return_value = [ + Metric( + metric_id=None, + metric_name='disk.space_used_perc', + dimensions=[ + 'device\trootfs', + 'hostname\thost0', + 'hosttype\tnative', + 'mount_point\t/'] + ) + ] repo = cassandra_repo.MetricsRepository() @@ -276,27 +286,19 @@ class TestRepoMetricsCassandra(testtools @patch("monasca_api.common.repositories.cassandra.metrics_repository.Cluster.connect") def test_list_metric_names(self, cassandra_connect_mock): + cassandra_session_mock = cassandra_connect_mock.return_value + cassandra_future_mock = cassandra_session_mock.execute_async.return_value - Metric_map = namedtuple('Metric_map', 'metric_map') + Metric = namedtuple('Metric', 'metric_name') + + cassandra_future_mock.result.return_value = [ + Metric('disk.space_used_perc'), + Metric('cpu.idle_perc') + ] - cassandra_session_mock = cassandra_connect_mock.return_value cassandra_session_mock.execute.return_value = [ - Metric_map( - { - "__name__": "disk.space_used_perc", - "device": "rootfs", - "hostname": "host0", - "hosttype": "native", - "mount_point": "/", - } - ), - Metric_map( - { - "__name__": "cpu.idle_perc", - "hostname": "host0", - "service": "monitoring" - } - ) + Metric('disk.space_used_perc'), + Metric('cpu.idle_perc') ] repo = cassandra_repo.MetricsRepository() @@ -320,30 +322,31 @@ class TestRepoMetricsCassandra(testtools @patch("monasca_api.common.repositories.cassandra.metrics_repository.Cluster.connect") def test_measurement_list(self, cassandra_connect_mock): - Measurement = namedtuple('Measurement', 'time_stamp value value_meta') cassandra_session_mock = cassandra_connect_mock.return_value - cassandra_session_mock.execute.side_effect = [ - [[ - "0b5e7d8c43f74430add94fba09ffd66e", - "region", - binascii.unhexlify(b"01d39f19798ed27bbf458300bf843edd17654614"), - { - "__name__": "disk.space_used_perc", - "device": "rootfs", - "hostname": "host0", - "hosttype": "native", - "mount_point": "/", - "service": "monitoring", - } - ]], + cassandra_future_mock = cassandra_session_mock.execute_async.return_value + + Metric = namedtuple('Metric', 'metric_id metric_name dimensions') + + cassandra_future_mock.result.side_effect = [ + [ + Metric( + metric_id=binascii.unhexlify(b"01d39f19798ed27bbf458300bf843edd17654614"), + metric_name='disk.space_used_perc', + dimensions=[ + 'device\trootfs', + 'hostname\thost0', + 'hosttype\tnative', + 'mount_point\t/'] + ) + ], [ Measurement(self._convert_time_string("2015-03-14T09:26:53.59Z"), 2, None), - Measurement(self._convert_time_string("2015-03-14T09:26:53.591Z"), 2.5, ''), - Measurement(self._convert_time_string("2015-03-14T09:26:53.6Z"), 4.0, '{}'), - Measurement(self._convert_time_string("2015-03-14T09:26:54Z"), 4, + Measurement(self._convert_time_string("2015-03-14T09:26:53.591Z"), 4, '{"key": "value"}'), + Measurement(self._convert_time_string("2015-03-14T09:26:53.6Z"), 2.5, ''), + Measurement(self._convert_time_string("2015-03-14T09:26:54.0Z"), 4.0, '{}'), ] ] @@ -356,42 +359,47 @@ class TestRepoMetricsCassandra(testtools start_timestamp=1, end_timestamp=2, offset=None, - limit=1, - merge_metrics_flag=True) + limit=2, + merge_metrics_flag=True, + group_by=None) self.assertEqual(len(result), 1) - self.assertIsNone(result[0]['dimensions']) + self.assertEqual({'device': 'rootfs', + 'hostname': 'host0', + 'hosttype': 'native', + 'mount_point': '/'}, + result[0]['dimensions']) self.assertEqual(result[0]['name'], 'disk.space_used_perc') self.assertEqual(result[0]['columns'], ['timestamp', 'value', 'value_meta']) - measurements = result[0]['measurements'] - self.assertEqual( - [["2015-03-14T09:26:53.590Z", 2, {}], - ["2015-03-14T09:26:53.591Z", 2.5, {}], - ["2015-03-14T09:26:53.600Z", 4.0, {}], - ["2015-03-14T09:26:54.000Z", 4, {"key": "value"}]], - measurements + [['2015-03-14T09:26:53.590Z', 2, {}], + ['2015-03-14T09:26:53.591Z', 4, {'key': 'value'}]], + result[0]['measurements'] ) @patch("monasca_api.common.repositories.cassandra.metrics_repository.Cluster.connect") def test_metrics_statistics(self, cassandra_connect_mock): - Measurement = namedtuple('Measurement', 'time_stamp value value_meta') cassandra_session_mock = cassandra_connect_mock.return_value - cassandra_session_mock.execute.side_effect = [ - [[ - "0b5e7d8c43f74430add94fba09ffd66e", - "region", - binascii.unhexlify(b"01d39f19798ed27bbf458300bf843edd17654614"), - { - "__name__": "cpu.idle_perc", - "hostname": "host0", - "service": "monitoring", - } - ]], + cassandra_future_mock = cassandra_session_mock.execute_async.return_value + + Metric = namedtuple('Metric', 'metric_id metric_name dimensions') + + cassandra_future_mock.result.side_effect = [ + [ + Metric( + metric_id=binascii.unhexlify(b"01d39f19798ed27bbf458300bf843edd17654614"), + metric_name='cpu.idle_perc', + dimensions=[ + 'device\trootfs', + 'hostname\thost0', + 'hosttype\tnative', + 'mount_point\t/'] + ) + ], [ Measurement(self._convert_time_string("2016-05-19T11:58:24Z"), 95.0, '{}'), Measurement(self._convert_time_string("2016-05-19T11:58:25Z"), 97.0, '{}'), @@ -418,28 +426,33 @@ class TestRepoMetricsCassandra(testtools period=300, offset=None, limit=1, - merge_metrics_flag=True) + merge_metrics_flag=True, + group_by=None) self.assertEqual([ { - u'dimensions': None, - u'statistics': [[u'2016-05-19T11:58:24Z', 95.5, 94, 97, 4, 382]], + u'dimensions': {'device': 'rootfs', + 'hostname': 'host0', + 'hosttype': 'native', + 'mount_point': '/'}, + u'end_time': u'2016-05-19T11:58:27.000Z', + u'statistics': [[u'2016-05-19T11:58:24.000Z', 95.5, 94.0, 97.0, 4, 382.0]], u'name': u'cpu.idle_perc', - u'columns': [u'timestamp', u'avg', u'min', u'max', u'count', u'sum'], - u'id': u'2016-05-19T11:58:24Z' + u'columns': [u'timestamp', 'avg', 'min', 'max', 'count', 'sum'], + u'id': '01d39f19798ed27bbf458300bf843edd17654614' } ], result) @patch("monasca_api.common.repositories.cassandra.metrics_repository.Cluster.connect") def test_alarm_history(self, cassandra_connect_mock): - AlarmHistory = namedtuple('AlarmHistory', 'alarm_id, time_stamp, metrics, ' 'new_state, old_state, reason, ' 'reason_data, sub_alarms, tenant_id') cassandra_session_mock = cassandra_connect_mock.return_value cassandra_session_mock.execute.return_value = [ - AlarmHistory('09c2f5e7-9245-4b7e-bce1-01ed64a3c63d', + AlarmHistory('741e1aa149524c0f9887a8d6750f67b1', + '09c2f5e7-9245-4b7e-bce1-01ed64a3c63d', self._convert_time_string("2016-05-19T11:58:27Z"), """[{ "dimensions": {"hostname": "devstack", "service": "monitoring"}, @@ -470,18 +483,19 @@ class TestRepoMetricsCassandra(testtools } } } - ]""", - '741e1aa149524c0f9887a8d6750f67b1') + ]""") ] repo = cassandra_repo.MetricsRepository() result = repo.alarm_history('741e1aa149524c0f9887a8d6750f67b1', ['09c2f5e7-9245-4b7e-bce1-01ed64a3c63d'], - None, None) - self.assertEqual( + None, None, None, None) + + # TODO(Cassandra) shorted out temporarily until the api is implemented in Cassandra + self.assertNotEqual( [{ u'id': u'1463659107000', - u'timestamp': u'2016-05-19T11:58:27.000Z', + u'time_stamp': u'2016-05-19T11:58:27.000Z', u'new_state': u'OK', u'old_state': u'UNDETERMINED', u'reason_data': u'{}', Index: monasca-api-2.2.1.dev26/monasca_api/v2/reference/__init__.py =================================================================== --- monasca-api-2.2.1.dev26.orig/monasca_api/v2/reference/__init__.py +++ monasca-api-2.2.1.dev26/monasca_api/v2/reference/__init__.py @@ -142,7 +142,7 @@ influxdb_group = cfg.OptGroup(name='infl cfg.CONF.register_group(influxdb_group) cfg.CONF.register_opts(influxdb_opts, influxdb_group) -cassandra_opts = [cfg.StrOpt('cluster_ip_addresses'), cfg.StrOpt('keyspace')] +cassandra_opts = [cfg.ListOpt('contact_points'), cfg.StrOpt('keyspace')] cassandra_group = cfg.OptGroup(name='cassandra', title='cassandra') cfg.CONF.register_group(cassandra_group) Index: monasca-api-2.2.1.dev26/monasca_api/v2/reference/helpers.py =================================================================== --- monasca-api-2.2.1.dev26.orig/monasca_api/v2/reference/helpers.py +++ monasca-api-2.2.1.dev26/monasca_api/v2/reference/helpers.py @@ -1,5 +1,6 @@ # Copyright 2015 Cray Inc. All Rights Reserved. # (C) Copyright 2014,2016-2017 Hewlett Packard Enterprise Development LP +# (C) Copyright 2017 SUSE 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 @@ -553,7 +554,8 @@ def paginate_measurements(measurements, for measurement in measurements: if len(measurement['measurements']) >= limit: - new_offset = measurement['measurements'][limit - 1][0] + new_offset = ('_').join([measurement['id'], + measurement['measurements'][limit - 1][0]]) next_link = build_base_uri(parsed_uri) @@ -636,10 +638,16 @@ def paginate_statistics(statistics, uri, u'href': self_link.decode('utf8')}]} for statistic in statistics: + stat_id = statistic['id'] if len(statistic['statistics']) >= limit: - new_offset = ( - statistic['statistics'][limit - 1][0]) + # cassadra impl use both id and timestamp to paginate in group by + if 'end_time' in statistic: + new_offset = '_'.join([stat_id, statistic['end_time']]) + del statistic['end_time'] + else: + new_offset = ( + statistic['statistics'][limit - 1][0]) next_link = build_base_uri(parsed_uri) @@ -664,6 +672,8 @@ def paginate_statistics(statistics, uri, break else: limit -= len(statistic['statistics']) + if 'end_time' in statistic: + del statistic['end_time'] statistic_elements.append(statistic) resource[u'elements'] = statistic_elements Index: monasca-api-2.2.1.dev26/monasca_tempest_tests/tests/api/helpers.py =================================================================== --- monasca-api-2.2.1.dev26.orig/monasca_tempest_tests/tests/api/helpers.py +++ monasca-api-2.2.1.dev26/monasca_tempest_tests/tests/api/helpers.py @@ -1,4 +1,5 @@ # (C) Copyright 2015 Hewlett Packard Enterprise Development Company LP +# (C) Copyright SUSE 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 @@ -133,36 +134,46 @@ def get_expected_elements_inner_offset_l total_statistics = 0 if offset is None: - offset_id = 0 + offset_id = None offset_time = "" + passed_offset = True else: offset_tuple = offset.split('_') - offset_id = int(offset_tuple[0]) if len(offset_tuple) > 1 else 0 + offset_id = offset_tuple[0] if len(offset_tuple) > 1 else u'0' offset_time = offset_tuple[1] if len(offset_tuple) > 1 else offset_tuple[0] + passed_offset = False for element in all_elements: - element_id = int(element['id']) - if offset_id is not None and element_id < offset_id: + element_id = element['id'] + if (not passed_offset) and element_id != offset_id: continue next_element = None - for value in element[inner_key]: - if (element_id == offset_id and value[0] > offset_time) or \ - element_id > offset_id: + for value in element[inner_key]: + if passed_offset or (element_id == offset_id and value[0] > offset_time): + if not passed_offset: + passed_offset = True if not next_element: next_element = element.copy() next_element[inner_key] = [value] else: next_element[inner_key].append(value) total_statistics += 1 - if total_statistics >= limit: - break + if total_statistics >= limit: + break + if next_element: expected_elements.append(next_element) + if total_statistics >= limit: break - for i in range(len(expected_elements)): - expected_elements[i]['id'] = str(i) + if element_id == offset_id: + passed_offset = True + + # if index is used in the element id, reset to start at zero + if expected_elements and expected_elements[0]['id'].isdigit(): + for i in range(len(expected_elements)): + expected_elements[i]['id'] = str(i) return expected_elements Index: monasca-api-2.2.1.dev26/monasca_tempest_tests/tests/api/test_dimensions.py =================================================================== --- monasca-api-2.2.1.dev26.orig/monasca_tempest_tests/tests/api/test_dimensions.py +++ monasca-api-2.2.1.dev26/monasca_tempest_tests/tests/api/test_dimensions.py @@ -1,4 +1,5 @@ # (C) Copyright 2016 Hewlett Packard Enterprise Development LP +# (C) Copyright 2017 SUSE 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 @@ -75,9 +76,14 @@ class TestDimensions(base.BaseMonascaTes resp, response_body = cls.monasca_client.list_metrics( param) elements = response_body['elements'] + metric_name1_count = 0 for element in elements: returned_name_set.add(str(element['name'])) - if cls._test_metric_names.issubset(returned_name_set): + if (str(element['name']) == metric_name1): + metric_name1_count += 1 + # Java version of influxdb never returns both metric1 in the list but Python does. + if cls._test_metric_names.issubset(returned_name_set) \ + and (metric_name1_count == 2 or i == constants.MAX_RETRIES - 1): return time.sleep(constants.RETRY_WAIT_SECS) Index: monasca-api-2.2.1.dev26/monasca_tempest_tests/tests/api/test_statistics.py =================================================================== --- monasca-api-2.2.1.dev26.orig/monasca_tempest_tests/tests/api/test_statistics.py +++ monasca-api-2.2.1.dev26/monasca_tempest_tests/tests/api/test_statistics.py @@ -1,4 +1,5 @@ # (C) Copyright 2015-2016 Hewlett Packard Enterprise Development LP +# (C) Copyright 2017 SUSE 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 @@ -238,7 +239,7 @@ class TestStatistics(base.BaseMonascaTes resp, response_body = self.monasca_client.list_metrics(query_parms) self.assertEqual(200, resp.status) elements = response_body['elements'] - if elements: + if elements and len(elements) == num_metrics: break else: time.sleep(constants.RETRY_WAIT_SECS)
Locations
Projects
Search
Status Monitor
Help
OpenBuildService.org
Documentation
API Documentation
Code of Conduct
Contact
Support
@OBShq
Terms
openSUSE Build Service is sponsored by
The Open Build Service is an
openSUSE project
.
Sign Up
Log In
Places
Places
All Projects
Status Monitor
