"""
Sprockets InfluxDB
==================
`sprockets_influxdb` includes both a buffering InfluxDB client and a Tornado
RequestHandler mixin.
Measurements will be sent in batches to InfluxDB when there are
``INFLUXDB_TRIGGER_SIZE`` measurements in the buffer or after
``INFLUXDB_INTERVAL`` milliseconds have passed since the last measurement was
added, which ever occurs first.
The timeout timer for submitting a buffer of < ``INFLUXDB_TRIGGER_SIZE``
measurements is only started when there isn't an active timer, there is not a
batch currently being written, and a measurement is added to the buffer.
"""
import contextlib
import logging
import os
import random
import select
import socket
import ssl
import time
import uuid
try:
from tornado import concurrent, httpclient, ioloop
except ImportError: # pragma: no cover
logging.critical('Could not import Tornado')
concurrent, httpclient, ioloop = None, None, None
try:
from tornado import routing
except ImportError: # Not needed for Tornado<4.5
pass
version_info = (2, 2, 0)
__version__ = '.'.join(str(v) for v in version_info)
__all__ = ['__version__', 'version_info', 'add_measurement', 'flush',
'install', 'shutdown', 'Measurement']
LOGGER = logging.getLogger(__name__)
REQUEST_DATABASE = 'sprockets_influxdb.database'
USER_AGENT = 'sprockets-influxdb/v{}'.format(__version__)
try:
TimeoutError
except NameError: # Python 2.7 compatibility
class TimeoutError(Exception):
pass
_base_tags = {}
_base_url = 'http://localhost:8086/write'
_batch_future = None
_buffer_size = 0
_credentials = None, None
_dirty = False
_enabled = True
_http_client = None
_installed = False
_last_warning = None
_measurements = {}
_max_batch_size = 10000
_max_buffer_size = 25000
_max_clients = 10
_sample_probability = 1.0
_stopping = False
_timeout_interval = 60000
_timeout = None
_trigger_size = 5000
_warn_threshold = 15000
_writing = False
[docs]class InfluxDBMixin(object):
"""Mixin that automatically submits per-request measurements to InfluxDB
with the request duration.
The measurements will automatically add the following tags:
- Request :data:`handler`
- Request :data:`endpoint` (if enabled via a named URL)
- Request :data:`method`
- Request :data:`correlation_id` (if set)
- Response :data:`status_code`
To add additional tags and fields, use the
:meth:`~sprockets_influxdb.Measurement.set_field`,
:meth:`~sprockets_influxdb.Measurement.set_tag`,
:meth:`~sprockets_influxdb.Measurement.set_tags`, and
:meth:`~sprockets_influxdb.Measurement.timer` methods of the
``influxdb`` attribute of the :class:`~tornado.web.RequestHandler`.
"""
def __init__(self, application, request, **kwargs):
self.influxdb = Measurement(
application.settings[REQUEST_DATABASE],
application.settings.get('service', 'request'))
super(InfluxDBMixin, self).__init__(application, request, **kwargs)
if _enabled:
handler = '{}.{}'.format(self.__module__, self.__class__.__name__)
self.influxdb.set_tags({'handler': handler,
'method': request.method})
pattern = None
if hasattr(application, 'handlers'):
pattern = self._get_path_pattern_tornado4()
else:
pattern = self._get_path_pattern_tornado45()
if pattern:
endpoint = pattern.rstrip('$')
else:
LOGGER.warning('Unable to determine routing pattern')
endpoint = request.path
self.influxdb.set_tags({'endpoint': endpoint})
def _get_path_pattern_tornado4(self):
"""Return the path pattern used when routing a request. (Tornado<4.5)
:rtype: str
"""
for host, handlers in self.application.handlers:
if host.match(self.request.host):
for handler in handlers:
if handler.regex.match(self.request.path):
return handler.regex.pattern
def _get_path_pattern_tornado45(self, router=None):
"""Return the path pattern used when routing a request. (Tornado>=4.5)
:param tornado.routing.Router router: (Optional) The router to scan.
Defaults to the application's router.
:rtype: str
"""
if router is None:
router = self.application.default_router
for rule in router.rules:
if rule.matcher.match(self.request) is not None:
if isinstance(rule.matcher, routing.PathMatches):
return rule.matcher.regex.pattern
elif isinstance(rule.target, routing.Router):
return self._get_path_pattern_tornado45(rule.target)
def on_finish(self):
if _enabled:
self.influxdb.set_field(
'content_length', int(self._headers.get('Content-Length', 0)))
self.influxdb.set_field('duration', self.request.request_time())
self.influxdb.set_tag('status_code', self._status_code)
self.influxdb.set_tag('remote_ip', self.request.remote_ip)
add_measurement(self.influxdb)
[docs]def add_measurement(measurement):
"""Add measurement data to the submission buffer for eventual writing to
InfluxDB.
Example:
.. code:: python
import sprockets_influxdb as influxdb
measurement = influxdb.Measurement('example', 'measurement-name')
measurement.set_tag('foo', 'bar')
measurement.set_field('baz', 1.05)
influxdb.add_measurement(measurement)
:param :class:`~sprockets_influxdb.Measurement` measurement: The
measurement to add to the buffer for submission to InfluxDB.
"""
global _buffer_size
if not _enabled:
LOGGER.debug('Discarding measurement for %s while not enabled',
measurement.database)
return
if _stopping:
LOGGER.warning('Discarding measurement for %s while stopping',
measurement.database)
return
if _buffer_size > _max_buffer_size:
LOGGER.warning('Discarding measurement due to buffer size limit')
return
if not measurement.fields:
raise ValueError('Measurement does not contain a field')
if measurement.database not in _measurements:
_measurements[measurement.database] = []
value = measurement.marshall()
_measurements[measurement.database].append(value)
# Ensure that len(measurements) < _trigger_size are written
if not _timeout:
if (_batch_future and _batch_future.done()) or not _batch_future:
_start_timeout()
# Check to see if the batch should be triggered
_buffer_size = _pending_measurements()
if _buffer_size >= _trigger_size:
_trigger_batch_write()
[docs]def flush():
"""Flush all pending measurements to InfluxDB. This will ensure that all
measurements that are in the buffer for any database are written. If the
requests fail, it will continue to try and submit the metrics until they
are successfully written.
:rtype: :class:`~tornado.concurrent.Future`
"""
flush_future = concurrent.Future()
if _batch_future and not _batch_future.done():
LOGGER.debug('Flush waiting on incomplete _batch_future')
_flush_wait(flush_future, _batch_future)
else:
LOGGER.info('Flushing buffer with %i measurements to InfluxDB',
_pending_measurements())
_flush_wait(flush_future, _write_measurements())
return flush_future
[docs]def install(url=None, auth_username=None, auth_password=None,
submission_interval=None, max_batch_size=None, max_clients=10,
base_tags=None, max_buffer_size=None, trigger_size=None,
sample_probability=1.0):
"""Call this to install/setup the InfluxDB client collector. All arguments
are optional.
:param str url: The InfluxDB API URL. If URL is not specified, the
``INFLUXDB_SCHEME``, ``INFLUXDB_HOST`` and ``INFLUXDB_PORT``
environment variables will be used to construct the base URL. Default:
``http://localhost:8086/write``
:param str auth_username: A username to use for InfluxDB authentication. If
not specified, the ``INFLUXDB_USER`` environment variable will
be used. Default: ``None``
:param str auth_password: A password to use for InfluxDB authentication. If
not specified, the ``INFLUXDB_PASSWORD`` environment variable will
be used. Default: ``None``
:param int submission_interval: The maximum number of milliseconds to wait
after the last batch submission before submitting a batch that is
smaller than ``trigger_size``. Default: ``60000``
:param int max_batch_size: The number of measurements to be submitted in a
single HTTP request. Default: ``10000``
:param int max_clients: The number of simultaneous batch submissions that
may be made at any given time. Default: ``10``
:param dict base_tags: Default tags that are to be submitted with each
measurement. Default: ``None``
:param int max_buffer_size: The maximum number of pending measurements
in the buffer before new measurements are discarded.
Default: ``25000``
:param int trigger_size: The minimum number of measurements that
are in the buffer before a batch can be submitted. Default: ``5000``
:param float sample_probability: Value between 0 and 1.0 specifying the
probability that a batch will be submitted (0.25 == 25%)
:returns: :data:`True` if the client was installed by this call
and :data:`False` otherwise.
If ``INFLUXDB_PASSWORD`` is specified as an environment variable, it will
be masked in the Python process.
"""
global _base_tags, _base_url, _credentials, _enabled, _installed, \
_max_batch_size, _max_buffer_size, _max_clients, \
_sample_probability, _timeout, _timeout_interval, _trigger_size
_enabled = os.environ.get('INFLUXDB_ENABLED', 'true') == 'true'
if not _enabled:
LOGGER.warning('Disabling InfluxDB support')
return
if _installed:
LOGGER.warning('InfluxDB client already installed')
return False
_base_url = url or '{}://{}:{}/write'.format(
os.environ.get('INFLUXDB_SCHEME', 'http'),
os.environ.get('INFLUXDB_HOST', 'localhost'),
os.environ.get('INFLUXDB_PORT', 8086))
_credentials = (auth_username or os.environ.get('INFLUXDB_USER', None),
auth_password or os.environ.get('INFLUXDB_PASSWORD', None))
# Don't leave the environment variable out there with the password
if os.environ.get('INFLUXDB_PASSWORD'):
os.environ['INFLUXDB_PASSWORD'] = \
'X' * len(os.environ['INFLUXDB_PASSWORD'])
# Submission related values
_timeout_interval = submission_interval or \
int(os.environ.get('INFLUXDB_INTERVAL', _timeout_interval))
_max_batch_size = max_batch_size or \
int(os.environ.get('INFLUXDB_MAX_BATCH_SIZE', _max_batch_size))
_max_clients = max_clients
_max_buffer_size = max_buffer_size or \
int(os.environ.get('INFLUXDB_MAX_BUFFER_SIZE', _max_buffer_size))
_sample_probability = sample_probability or \
float(os.environ.get('INFLUXDB_SAMPLE_PROBABILITY',
_sample_probability))
_trigger_size = trigger_size or \
int(os.environ.get('INFLUXDB_TRIGGER_SIZE', _trigger_size))
# Set the base tags
if os.environ.get('INFLUXDB_TAG_HOSTNAME', 'true') == 'true':
_base_tags.setdefault('hostname', socket.gethostname())
if os.environ.get('ENVIRONMENT'):
_base_tags.setdefault('environment', os.environ['ENVIRONMENT'])
_base_tags.update(base_tags or {})
# Seed the random number generator for batch sampling
random.seed()
# Don't let this run multiple times
_installed = True
LOGGER.info('sprockets_influxdb v%s installed; %i measurements or %.2f '
'seconds will trigger batch submission', __version__,
_trigger_size, _timeout_interval / 1000.0)
return True
[docs]def set_auth_credentials(username, password):
"""Override the default authentication credentials obtained from the
environment variable configuration.
:param str username: The username to use
:param str password: The password to use
"""
global _credentials, _dirty
LOGGER.debug('Setting authentication credentials')
_credentials = username, password
_dirty = True
[docs]def set_base_url(url):
"""Override the default base URL value created from the environment
variable configuration.
:param str url: The base URL to use when submitting measurements
"""
global _base_url, _dirty
LOGGER.debug('Setting base URL to %s', url)
_base_url = url
_dirty = True
[docs]def set_max_batch_size(limit):
"""Set a limit to the number of measurements that are submitted in
a single batch that is submitted per databases.
:param int limit: The maximum number of measurements per batch
"""
global _max_batch_size
LOGGER.debug('Setting maximum batch size to %i', limit)
_max_batch_size = limit
[docs]def set_max_buffer_size(limit):
"""Set the maximum number of pending measurements allowed in the buffer
before new measurements are discarded.
:param int limit: The maximum number of measurements per batch
"""
global _max_buffer_size
LOGGER.debug('Setting maximum buffer size to %i', limit)
_max_buffer_size = limit
def set_max_clients(limit):
"""Set the maximum number of simultaneous batch submission that can execute
in parallel.
:param int limit: The maximum number of simultaneous batch submissions
"""
global _dirty, _max_clients
LOGGER.debug('Setting maximum client limit to %i', limit)
_dirty = True
_max_clients = limit
def set_sample_probability(probability):
"""Set the probability that a batch will be submitted to the InfluxDB
server. This should be a value that is greater than or equal to ``0`` and
less than or equal to ``1.0``. A value of ``0.25`` would represent a
probability of 25% that a batch would be written to InfluxDB.
:param float probability: The value between 0 and 1.0 that represents the
probability that a batch will be submitted to the InfluxDB server.
"""
global _sample_probability
if not 0.0 <= probability <= 1.0:
raise ValueError('Invalid probability value')
LOGGER.debug('Setting sample probability to %.2f', probability)
_sample_probability = float(probability)
[docs]def set_timeout(milliseconds):
"""Override the maximum duration to wait for submitting measurements to
InfluxDB.
:param int milliseconds: Maximum wait in milliseconds
"""
global _timeout, _timeout_interval
LOGGER.debug('Setting batch wait timeout to %i ms', milliseconds)
_timeout_interval = milliseconds
_maybe_stop_timeout()
_timeout = ioloop.IOLoop.current().add_timeout(milliseconds, _on_timeout)
[docs]def set_trigger_size(limit):
"""Set the number of pending measurements that trigger the writing of data
to InfluxDB
:param int limit: The minimum number of measurements to trigger a batch
"""
global _trigger_size
LOGGER.debug('Setting trigger buffer size to %i', limit)
_trigger_size = limit
[docs]def shutdown():
"""Invoke on shutdown of your application to stop the periodic
callbacks and flush any remaining metrics.
Returns a future that is complete when all pending metrics have been
submitted.
:rtype: :class:`~tornado.concurrent.Future`
"""
global _stopping
if _stopping:
LOGGER.warning('Already shutting down')
return
_stopping = True
_maybe_stop_timeout()
return flush()
def _create_http_client():
"""Create the HTTP client with authentication credentials if required."""
global _http_client
defaults = {'user_agent': USER_AGENT}
auth_username, auth_password = _credentials
if auth_username and auth_password:
defaults['auth_username'] = auth_username
defaults['auth_password'] = auth_password
_http_client = httpclient.AsyncHTTPClient(
force_instance=True, defaults=defaults,
max_clients=_max_clients)
def _flush_wait(flush_future, write_future):
"""Pause briefly allowing any pending metric writes to complete before
shutting down.
:param tornado.concurrent.Future flush_future: The future to resolve
when the shutdown is complete.
:param tornado.concurrent.Future write_future: The future that is for the
current batch write operation.
"""
if write_future.done():
if not _pending_measurements():
flush_future.set_result(True)
return
else:
write_future = _write_measurements()
ioloop.IOLoop.current().add_timeout(
ioloop.IOLoop.current().time() + 0.25,
_flush_wait, flush_future, write_future)
def _futures_wait(wait_future, futures):
"""Waits for all futures to be completed. If the futures are not done,
wait 100ms and then invoke itself via the ioloop and check again. If
they are done, set a result on `wait_future` indicating the list of
futures are done.
:param wait_future: The future to complete when all `futures` are done
:type wait_future: tornado.concurrent.Future
:param list futures: The list of futures to watch for completion
"""
global _buffer_size, _writing
remaining = []
for (future, batch, database, measurements) in futures:
# If the future hasn't completed, add it to the remaining stack
if not future.done():
remaining.append((future, batch, database, measurements))
continue
# Get the result of the HTTP request, processing any errors
error = future.exception()
if isinstance(error, httpclient.HTTPError):
if error.code == 400:
_write_error_batch(batch, database, measurements)
elif error.code >= 500:
_on_5xx_error(batch, error, database, measurements)
else:
LOGGER.error('Error submitting %s batch %s to InfluxDB (%s): '
'%s', database, batch, error.code,
error.response.body)
elif isinstance(error, (TimeoutError, OSError, socket.error,
select.error, ssl.socket_error)):
_on_5xx_error(batch, error, database, measurements)
# If there are futures that remain, try again in 100ms.
if remaining:
return ioloop.IOLoop.current().add_timeout(
ioloop.IOLoop.current().time() + 0.1,
_futures_wait, wait_future, remaining)
else: # Start the next timeout or trigger the next batch
_buffer_size = _pending_measurements()
LOGGER.debug('Batch submitted, %i measurements remain', _buffer_size)
if _buffer_size >= _trigger_size:
ioloop.IOLoop.current().add_callback(_trigger_batch_write)
elif _buffer_size:
_start_timeout()
_writing = False
wait_future.set_result(True)
def _maybe_stop_timeout():
"""If there is a pending timeout, remove it from the IOLoop and set the
``_timeout`` global to None.
"""
global _timeout
if _timeout is not None:
LOGGER.debug('Removing the pending timeout (%r)', _timeout)
ioloop.IOLoop.current().remove_timeout(_timeout)
_timeout = None
def _maybe_warn_about_buffer_size():
"""Check the buffer size and issue a warning if it's too large and
a warning has not been issued for more than 60 seconds.
"""
global _last_warning
if not _last_warning:
_last_warning = time.time()
if _buffer_size > _warn_threshold and (time.time() - _last_warning) > 120:
LOGGER.warning('InfluxDB measurement buffer has %i entries',
_buffer_size)
def _on_5xx_error(batch, error, database, measurements):
"""Handle a batch submission error, logging the problem and adding the
measurements back to the stack.
:param str batch: The batch ID
:param mixed error: The error that was returned
:param str database: The database the submission failed for
:param list measurements: The measurements to add back to the stack
"""
LOGGER.info('Appending %s measurements to stack due to batch %s %r',
database, batch, error)
_measurements[database] = _measurements[database] + measurements
def _on_timeout():
"""Invoked periodically to ensure that metrics that have been collected
are submitted to InfluxDB.
:rtype: tornado.concurrent.Future or None
"""
global _buffer_size
LOGGER.debug('No metrics submitted in the last %.2f seconds',
_timeout_interval / 1000.0)
_buffer_size = _pending_measurements()
if _buffer_size:
return _trigger_batch_write()
_start_timeout()
def _pending_measurements():
"""Return the number of measurements that have not been submitted to
InfluxDB.
:rtype: int
"""
return sum([len(_measurements[dbname]) for dbname in _measurements])
def _sample_batch():
"""Determine if a batch should be processed and if not, pop off all of
the pending metrics for that batch.
:rtype: bool
"""
if _sample_probability == 1.0 or random.random() < _sample_probability:
return True
# Pop off all the metrics for the batch
for database in _measurements:
_measurements[database] = _measurements[database][_max_batch_size:]
return False
def _start_timeout():
"""Stop a running timeout if it's there, then create a new one."""
global _timeout
LOGGER.debug('Adding a new timeout in %i ms', _timeout_interval)
_maybe_stop_timeout()
_timeout = ioloop.IOLoop.current().add_timeout(
ioloop.IOLoop.current().time() + _timeout_interval / 1000.0,
_on_timeout)
def _trigger_batch_write():
"""Stop a timeout if it's running, and then write the measurements."""
global _batch_future
LOGGER.debug('Batch write triggered (%r/%r)',
_buffer_size, _trigger_size)
_maybe_stop_timeout()
_maybe_warn_about_buffer_size()
_batch_future = _write_measurements()
return _batch_future
def _write_measurements():
"""Write out all of the metrics in each of the databases,
returning a future that will indicate all metrics have been written
when that future is done.
:rtype: tornado.concurrent.Future
"""
global _timeout, _writing
future = concurrent.Future()
if _writing:
LOGGER.warning('Currently writing measurements, skipping write')
future.set_result(False)
elif not _pending_measurements():
future.set_result(True)
elif not _sample_batch():
LOGGER.debug('Skipping batch submission due to sampling')
future.set_result(True)
# Exit early if there's an error condition
if future.done():
return future
if not _http_client or _dirty:
_create_http_client()
# Keep track of the futures for each batch submission
futures = []
# Submit a batch for each database
for database in _measurements:
url = '{}?db={}&precision=ms'.format(_base_url, database)
# Get the measurements to submit
measurements = _measurements[database][:_max_batch_size]
# Pop them off the stack of pending measurements
_measurements[database] = _measurements[database][_max_batch_size:]
# Create the request future
LOGGER.debug('Submitting %r measurements to %r',
len(measurements), url)
request = _http_client.fetch(
url, method='POST', body='\n'.join(measurements).encode('utf-8'))
# Keep track of each request in our future stack
futures.append((request, str(uuid.uuid4()), database, measurements))
# Start the wait cycle for all the requests to complete
_writing = True
_futures_wait(future, futures)
return future
def _write_error_batch(batch, database, measurements):
"""Invoked when a batch submission fails, this method will submit one
measurement to InfluxDB. It then adds a timeout to the IOLoop which will
invoke :meth:`_write_error_batch_wait` which will evaluate the result and
then determine what to do next.
:param str batch: The batch ID for correlation purposes
:param str database: The database name for the measurements
:param list measurements: The measurements that failed to write as a batch
"""
if not measurements:
LOGGER.info('All %s measurements from batch %s processed',
database, batch)
return
LOGGER.debug('Processing batch %s for %s by measurement, %i left',
batch, database, len(measurements))
url = '{}?db={}&precision=ms'.format(_base_url, database)
measurement = measurements.pop(0)
# Create the request future
future = _http_client.fetch(
url, method='POST', body=measurement.encode('utf-8'))
# Check in 25ms to see if it's done
ioloop.IOLoop.current().add_timeout(
ioloop.IOLoop.current().time() + 0.025,
_write_error_batch_wait, future, batch, database, measurement,
measurements)
def _write_error_batch_wait(future, batch, database, measurement,
measurements):
"""Invoked by the IOLoop, this method checks if the HTTP request future
created by :meth:`_write_error_batch` is done. If it's done it will
evaluate the result, logging any error and moving on to the next
measurement. If there are no measurements left in the `measurements`
argument, it will consider the batch complete.
:param tornado.concurrent.Future future: The AsyncHTTPClient request future
:param str batch: The batch ID
:param str database: The database name for the measurements
:param str measurement: The measurement the future is for
:param list measurements: The measurements that failed to write as a batch
"""
if not future.done():
ioloop.IOLoop.current().add_timeout(
ioloop.IOLoop.current().time() + 0.025,
_write_error_batch_wait, future, batch, database, measurement,
measurements)
return
error = future.exception()
if isinstance(error, httpclient.HTTPError):
if error.code == 400:
LOGGER.error('Error writing %s measurement from batch %s to '
'InfluxDB (%s): %s', database, batch, error.code,
error.response.body)
LOGGER.info('Bad %s measurement from batch %s: %s',
database, batch, measurement)
else:
LOGGER.error('Error submitting individual metric for %s from '
'batch %s to InfluxDB (%s): %s',
database, batch, error.code)
measurements = measurements + [measurement]
elif isinstance(error, (TimeoutError, OSError, socket.error,
select.error, ssl.socket_error)):
LOGGER.error('Error submitting individual metric for %s from batch '
'%s to InfluxDB (%s)', database, batch, error)
_write_error_batch(batch, database, measurements + [measurement])
measurements = measurements + [measurement]
if not measurements:
LOGGER.info('All %s measurements from batch %s processed',
database, batch)
return
# Continue writing measurements
_write_error_batch(batch, database, measurements)
[docs]class Measurement(object):
"""The :class:`Measurement` class represents what will become a single row
in an InfluxDB database. Measurements are added to InfluxDB via the
:meth:`~sprockets_influxdb.add_measurement` method.
Example:
.. code:: python
import sprockets_influxdb as influxdb
measurement = Measurement('database-name', 'measurement-name')
measurement.set_tag('foo', 'bar')
measurement.set_field('baz', 1.05)
influxdb.add_measurement(measurement)
:param str database: The database name to use when submitting
:param str name: The measurement name
"""
def __init__(self, database, name):
self.database = database
self.name = name
self.fields = {}
self.tags = dict(_base_tags)
self.timestamp = time.time()
[docs] @contextlib.contextmanager
def duration(self, name):
"""Record the time it takes to run an arbitrary code block.
:param str name: The field name to record the timing in
This method returns a context manager that records the amount
of time spent inside of the context, adding the timing to the
measurement.
"""
start = time.time()
try:
yield
finally:
self.set_field(name, max(time.time(), start) - start)
[docs] def marshall(self):
"""Return the measurement in the line protocol format.
:rtype: str
"""
return '{},{} {} {}'.format(
self._escape(self.name),
','.join(['{}={}'.format(self._escape(k), self._escape(v))
for k, v in self.tags.items()]),
self._marshall_fields(),
int(self.timestamp * 1000))
[docs] def set_field(self, name, value):
"""Set the value of a field in the measurement.
:param str name: The name of the field to set the value for
:param int|float|bool|str value: The value of the field
:raises: ValueError
"""
if not any([isinstance(value, t) for t in {int, float, bool, str}]):
LOGGER.debug('Invalid field value: %r', value)
raise ValueError('Value must be a str, bool, integer, or float')
self.fields[name] = value
[docs] def set_tag(self, name, value):
"""Set a tag on the measurement.
:param str name: name of the tag to set
:param str value: value to assign
This will overwrite the current value assigned to a tag
if one exists.
"""
self.tags[name] = value
[docs] def set_timestamp(self, value):
"""Override the timestamp of a measurement.
:param float value: The timestamp to assign to the measurement
"""
self.timestamp = value
@staticmethod
def _escape(value):
"""Escape a string (key or value) for InfluxDB's line protocol.
:param str|int|float|bool value: The value to be escaped
:rtype: str
"""
value = str(value)
for char, escaped in {' ': '\ ', ',': '\,', '"': '\"'}.items():
value = value.replace(char, escaped)
return value
def _marshall_fields(self):
"""Convert the field dict into the string segment of field key/value
pairs.
:rtype: str
"""
values = {}
for key, value in self.fields.items():
if (isinstance(value, int) or
(isinstance(value, str) and value.isdigit() and
'.' not in value)):
values[key] = '{}i'.format(value)
elif isinstance(value, bool):
values[key] = self._escape(value)
elif isinstance(value, float):
values[key] = '{}'.format(value)
elif isinstance(value, str):
values[key] = '"{}"'.format(self._escape(value))
return ','.join(['{}={}'.format(self._escape(k), v)
for k, v in values.items()])