in metrics/heron/tmaster/client.py [0:0]
def get_service_times(self, topology_id: str, cluster: str, environ: str,
start: dt.datetime, end: dt.datetime,
**kwargs: Union[str, int, float]) -> pd.DataFrame:
""" Gets the service times, as a timeseries, for every instance of the
of all the bolt components of the specified topology. The start and end
times define the window over which to gather the metrics. The window
duration should be less than 3 hours as this is the limit of what the
Topology master stores.
Arguments:
topology_id (str): The topology identification string.
start (datetime): utc datetime instance for the start of the
metrics gathering period.
end (datetime): utc datetime instance for the end of the
metrics gathering period.
**cluster (str): The cluster the topology is running in.
**environ (str): The environment the topology is running in (eg.
prod, devel, test, etc).
Returns:
pandas.DataFrame: A DataFrame containing the service time
measurements as a timeseries. Each row represents a measurement
(aggregated over one minute) with the following columns:
* timestamp:The UTC timestamp for the metric time period,
* component: The component this metric comes from,
* task: The instance ID number for the instance that the metric
comes from,
* container: The ID for the container this metric comes from,
* stream: The name of the incoming stream from which the tuples
that lead to this metric came from,
* latency_ms: The average execute latency measurement in
milliseconds for that metric time period.
"""
LOG.info("Getting service times for topology %s over a %d second "
"period from %s to %s", topology_id,
(end-start).total_seconds(), start.isoformat(),
end.isoformat())
logical_plan, start_time, end_time = self._query_setup(
topology_id, cluster, environ, start, end)
output: pd.DataFrame = None
bolts: Dict[str, Any] = logical_plan["bolts"]
bolt_component: str
for bolt_component in bolts:
try:
bolt_service_times: pd.DataFrame = \
self.get_component_service_times(topology_id,
cluster, environ,
bolt_component,
start_time, end_time,
logical_plan)
except HTTPError as http_error:
LOG.warning("Fetching execute latencies for component %s "
"failed with status code %s", bolt_component,
str(http_error.response.status_code))
else:
if output is None:
output = bolt_service_times
else:
output = output.append(bolt_service_times,
ignore_index=True)
return output