in metrics/heron/influxdb/client.py [0:0]
def get_metric_measurement_names(
self, database: str, metric_name: str, metric_regex: str,
force: bool=False) -> List[str]:
""" Gets a list of measurement names from the supplied database that
are related to the supplied metric using the supplied regex string.
Metric name search as cached and so repeated calls will not query the
database unless the force flag is used.
Arguments:
database (str): The name of the influx database to be searched.
metric_name (str): The name of the metric whose measurement names
are required (this is used for cache keys and
logging).
metric_regex (str): The regex to be used to search for
measurement names.
force (bool): Flag indicating if the cache should be bypassed.
Defaults to false.
Returns:
List[str]: A list of measurement name strings from the specified
database.
Raises:
RuntimeError: If the specified database has no measurements
matching the supplied metric regex.
"""
# Check to see if we have already queried Influx for the metric
# measurement names, if not query them and cache the results.
if database not in self.metric_name_cache[metric_name] or force:
LOG.info("Finding measurement names for metric: %s from "
"database: %s", metric_name, database)
# Find all the measurements for each bolt component
measurement_query: str = (f"SHOW MEASUREMENTS ON \"{database}\" "
f"WITH MEASUREMENT =~ {metric_regex}")
measurement_names: List[str] = \
[point["name"] for point in
self.client.query(measurement_query).get_points()]
if not measurement_names:
msg: str = (f"No measurements found in database: {database} "
f"for metric: {metric_name}")
LOG.error(msg)
raise RuntimeError(msg)
else:
LOG.info("Found %d measurement names for metric: %s",
len(measurement_names), metric_name)
self.metric_name_cache[metric_name][database] = \
measurement_names
else:
LOG.info("Using cached measurement names for metric: %s from "
"database: %s", metric_name, database)
return self.metric_name_cache[metric_name][database]