cfcdbfdfd1
- Create a new gnocchi dir for the collectd-gnocchi plugin. - Get gnocchi endpoint from keystone - Add new _base_url format - Create metrics if they don't already exist - Add instructions to doc/source/devstackGSG.rst Change-Id: Id7ce8130cb22f33147b7f031cd65564375db10d6
243 lines
8.3 KiB
Python
243 lines
8.3 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
# 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.
|
|
"""Gnocchi collectd plugin implementation"""
|
|
|
|
from __future__ import division
|
|
from __future__ import unicode_literals
|
|
|
|
import collectd_ceilometer
|
|
from collectd_ceilometer.common.keystone_light import ClientV3
|
|
from collectd_ceilometer.common.keystone_light import KeystoneException
|
|
from collectd_ceilometer.common.settings import Config
|
|
|
|
import json
|
|
import logging
|
|
import requests
|
|
from requests.exceptions import RequestException
|
|
import six
|
|
import threading
|
|
|
|
LOGGER = logging.getLogger(__name__)
|
|
ROOT_LOGGER = logging.getLogger(collectd_ceilometer.__name__)
|
|
|
|
|
|
# HTTP status codes
|
|
HTTP_CREATED = 201
|
|
HTTP_UNAUTHORIZED = 401
|
|
HTTP_NOT_FOUND = 404
|
|
|
|
|
|
class Sender(object):
|
|
"""Sends the JSON serialized data to Gnocchi"""
|
|
|
|
def __init__(self):
|
|
"""Create the Sender instance
|
|
|
|
The cofinguration must be initialized before the object is created.
|
|
"""
|
|
self._url_base = None
|
|
self._keystone = None
|
|
self._auth_token = None
|
|
self._auth_lock = threading.Lock()
|
|
self._failed_auth = False
|
|
self._meter_ids = {}
|
|
|
|
def _authenticate(self):
|
|
"""Authenticate and renew the authentication token"""
|
|
|
|
# if auth_token is available, just return it
|
|
if self._auth_token is not None:
|
|
return self._auth_token
|
|
|
|
# aquire the authentication lock
|
|
with self._auth_lock:
|
|
# re-check the auth_token as another thread could set it
|
|
if self._auth_token is not None:
|
|
return self._auth_token
|
|
|
|
LOGGER.debug('Authenticating request')
|
|
# pylint: disable=broad-except
|
|
try:
|
|
# create a keystone client if it doesn't exist
|
|
if self._keystone is None:
|
|
cfg = Config.instance()
|
|
self._keystone = ClientV3(
|
|
auth_url=cfg.OS_AUTH_URL,
|
|
username=cfg.OS_USERNAME,
|
|
password=cfg.OS_PASSWORD,
|
|
tenant_name=cfg.OS_TENANT_NAME
|
|
)
|
|
# store the authentication token
|
|
self._auth_token = self._keystone.auth_token
|
|
|
|
# get the uri of service endpoint
|
|
endpoint = self._get_endpoint("gnocchi")
|
|
|
|
self._url_base = "{}/v1/metric/%s/measures".format(endpoint)
|
|
|
|
LOGGER.info('Authenticating request - success')
|
|
self._failed_auth = False
|
|
|
|
except KeystoneException as exc:
|
|
log_level = logging.DEBUG
|
|
|
|
if not self._failed_auth:
|
|
log_level = logging.ERROR
|
|
LOGGER.error(
|
|
'Suspending error logs until successful auth'
|
|
)
|
|
|
|
LOGGER.log(log_level, 'Authentication error: %s',
|
|
six.text_type(exc),
|
|
exc_info=0)
|
|
|
|
if exc.response:
|
|
LOGGER.debug('Response: %s', exc.response)
|
|
|
|
self._auth_token = None
|
|
self._failed_auth = True
|
|
|
|
return self._auth_token
|
|
|
|
def send(self, metername, payload, unit):
|
|
"""Send the payload to Gnocchi"""
|
|
|
|
# get the auth_token
|
|
auth_token = self._authenticate()
|
|
LOGGER.info('Auth_token: %s',
|
|
auth_token,
|
|
)
|
|
# if auth_token is not set, there is nothing to do
|
|
if auth_token is None:
|
|
LOGGER.debug('Unable to send data. Not authenticated')
|
|
return
|
|
|
|
if self._url_base is None:
|
|
LOGGER.debug(
|
|
'Unable to send data. Missing endpoint from ident server')
|
|
return
|
|
|
|
# create request URL
|
|
metric_id = self._get_metric_id(metername, unit)
|
|
url = self._url_base % (metric_id)
|
|
|
|
# send the POST request
|
|
result = self._perform_request(url, payload, auth_token)
|
|
|
|
if result is None:
|
|
return
|
|
|
|
LOGGER.info('Result: %s %s',
|
|
six.text_type(result.status_code),
|
|
result.text)
|
|
|
|
# if the request failed due to an auth error
|
|
if result.status_code == HTTP_UNAUTHORIZED:
|
|
# reset the auth token in order to force the subsequent
|
|
# _authenticate() call to renew it
|
|
# Here, it can happen that the token is reset right after
|
|
# another thread has finished the authentication and thus
|
|
# the authentication may be performed twice
|
|
self._auth_token = None
|
|
|
|
# renew the authentication token
|
|
auth_token = self._authenticate()
|
|
|
|
if auth_token is not None:
|
|
# and try to repost
|
|
result = self._perform_request(url, payload, auth_token)
|
|
|
|
if result.status_code == HTTP_NOT_FOUND:
|
|
LOGGER.debug("Received 404 error when submitting %s sample, \
|
|
creating a new metric",
|
|
metername)
|
|
|
|
# create metric (endpoint, metername)
|
|
metric_id = self._get_metric_id(metername, unit)
|
|
|
|
LOGGER.info('metername: %s, meter_id: %s', metername, metric_id)
|
|
# Set a new url for the request
|
|
url = self._url_base % (metric_id)
|
|
# TODO(emma-l-foley): Add error checking
|
|
# Submit the sample
|
|
result = self._perform_request(url, payload, auth_token)
|
|
|
|
if result.status_code == HTTP_CREATED:
|
|
LOGGER.debug('Result: %s', HTTP_CREATED)
|
|
else:
|
|
LOGGER.info('Result: %s %s',
|
|
result.status_code,
|
|
result.text)
|
|
|
|
def _get_endpoint(self, service):
|
|
# get the uri of service endpoint
|
|
endpoint = self._keystone.get_service_endpoint(
|
|
service,
|
|
Config.instance().CEILOMETER_URL_TYPE)
|
|
return endpoint
|
|
|
|
def _get_metric_id(self, metername, unit):
|
|
|
|
try:
|
|
return self._meter_ids[metername]
|
|
except KeyError as ke:
|
|
LOGGER.warn(ke)
|
|
LOGGER.warn('No known ID for %s', metername)
|
|
|
|
endpoint = self._get_endpoint("gnocchi")
|
|
self._meter_ids[metername] = \
|
|
self._create_metric(metername, endpoint, unit)
|
|
|
|
return self._meter_ids[metername]
|
|
|
|
def _create_metric(self, metername, endpoint, unit):
|
|
url = "{}/v1/metric/".format(endpoint)
|
|
payload = json.dumps({"archive_policy_name": "high",
|
|
"name": metername,
|
|
"unit": unit,
|
|
})
|
|
result = self._perform_request(url, payload, self._auth_token)
|
|
metric_id = json.loads(result.text)['id']
|
|
LOGGER.debug("metric_id=%s", metric_id)
|
|
return metric_id
|
|
|
|
@classmethod
|
|
def _perform_request(cls, url, payload, auth_token):
|
|
"""Perform the POST request"""
|
|
|
|
LOGGER.debug('Performing request to %s', url)
|
|
|
|
# request headers
|
|
headers = {'X-Auth-Token': auth_token,
|
|
'Content-type': 'application/json'}
|
|
# perform request and return its result
|
|
response = None
|
|
try:
|
|
LOGGER.debug(
|
|
"Performing request to: %s with data=%s and headers=%s",
|
|
url, payload, headers)
|
|
|
|
response = requests.post(
|
|
url, data=payload, headers=headers,
|
|
timeout=(Config.instance().CEILOMETER_TIMEOUT / 1000.))
|
|
LOGGER.info('Response: %s: %s',
|
|
response.status_code, response.text
|
|
)
|
|
except RequestException as exc:
|
|
LOGGER.error('gnocchi request error: %s', six.text_type(exc))
|
|
finally:
|
|
LOGGER.debug('Returning response from _perform_request(): %s',
|
|
response.status_code)
|
|
return response
|