apache_beam.testing.load_tests.load_test_metrics_utils module

Utility functions used for integrating Metrics API into load tests pipelines.

Metrics are send to BigQuery in following format: test_id | submit_timestamp | metric_type | value

The ‘test_id’ is common for all metrics for one run. Currently it is possible to have following metrics types: * runtime * total_bytes_count

apache_beam.testing.load_tests.load_test_metrics_utils.parse_step(step_name)

Replaces white spaces and removes ‘Step:’ label

Parameters:step_name (str) – step name passed in metric ParDo Returns:lower case step name without namespace and step label

apache_beam.testing.load_tests.load_test_metrics_utils.split_metrics_by_namespace_and_name(metrics, namespace, name)

Splits metrics list namespace and name.

Parameters:

• metrics – list of metrics from pipeline result
• namespace (str) – filter metrics by namespace
• name (str) – filter metrics by name

Returns:

two lists - one of metrics which are matching filters and second of not matching

apache_beam.testing.load_tests.load_test_metrics_utils.get_generic_distributions(generic_dists, metric_id)

Creates flatten list of distributions per its value type. A generic distribution is the one which is not processed but saved in the most raw version.

Parameters:

• generic_dists – list of distributions to be saved
• metric_id (uuid) – id of the current test run

Returns:

list of dictionaries made from DistributionMetric

apache_beam.testing.load_tests.load_test_metrics_utils.get_all_distributions_by_type(dist, metric_id)

Creates new list of objects with type of each distribution metric value.

Parameters:

• dist (object) – DistributionMetric object to be parsed
• metric_id (uuid) – id of the current test run

Returns:

list of DistributionMetric objects

apache_beam.testing.load_tests.load_test_metrics_utils.get_distribution_dict(metric_type, submit_timestamp, dist, metric_id)

Function creates DistributionMetric

Parameters:

• metric_type (str) – type of value from distribution metric which will be saved (ex. max, min, mean, sum)
• submit_timestamp – timestamp when metric is saved
• dist (object) –
• metric_id (uuid) – id of the current test run

Returns:

dictionary prepared for saving according to schema

class apache_beam.testing.load_tests.load_test_metrics_utils.MetricsReader(project_name=None, bq_table=None, bq_dataset=None, publish_to_bq=False, influxdb_options=None, namespace=None, filters=None)

Bases: object

A MetricsReader retrieves metrics from pipeline result, prepares it for publishers and setup publishers.

Initializes MetricsReader .

Parameters:

• project_name (str) – project with BigQuery where metrics will be saved
• bq_table (str) – BigQuery table where metrics will be saved
• bq_dataset (str) – BigQuery dataset where metrics will be saved
• namespace (str) – Namespace of the metrics
• filters – MetricFilter to query only filtered metrics

get_counter_metric(result: apache_beam.runners.runner.PipelineResult, name: str) → int

Return the current value for a long counter, or -1 if can’t be retrieved. Note this uses only attempted metrics because some runners don’t support committed metrics.

publish_metrics(result: apache_beam.runners.runner.PipelineResult, extra_metrics: Optional[dict] = None)

Publish metrics from pipeline result to registered publishers.

publish_values(labeled_values)

The method to publish simple labeled values.

Parameters:labeled_values (List[Tuple(str, int)]) – list of (label, value)

class apache_beam.testing.load_tests.load_test_metrics_utils.Metric(submit_timestamp, metric_id, value, metric=None, label=None)

Bases: object

Metric base class in ready-to-save format.

Initializes Metric

Parameters:

• metric (object) – object of metric result
• submit_timestamp (float) – date-time of saving metric to database
• metric_id (uuid) – unique id to identify test run
• value – value of metric
• label – custom metric name to be saved in database

as_dict()

class apache_beam.testing.load_tests.load_test_metrics_utils.CounterMetric(counter_metric, submit_timestamp, metric_id)

Bases: apache_beam.testing.load_tests.load_test_metrics_utils.Metric

The Counter Metric in ready-to-publish format.

Parameters:

• counter_metric (object) – counter metric object from MetricResult
• submit_timestamp (float) – date-time of saving metric to database
• metric_id (uuid) – unique id to identify test run

class apache_beam.testing.load_tests.load_test_metrics_utils.DistributionMetric(dist_metric, submit_timestamp, metric_id, metric_type)

Bases: apache_beam.testing.load_tests.load_test_metrics_utils.Metric

The Distribution Metric in ready-to-publish format.

Parameters:

• dist_metric (object) – distribution metric object from MetricResult
• submit_timestamp (float) – date-time of saving metric to database
• metric_id (uuid) – unique id to identify test run

class apache_beam.testing.load_tests.load_test_metrics_utils.RuntimeMetric(runtime_list, metric_id)

Bases: apache_beam.testing.load_tests.load_test_metrics_utils.Metric

The Distribution Metric in ready-to-publish format.

Parameters:

• runtime_list – list of distributions metrics from MetricResult with runtime name
• metric_id (uuid) – unique id to identify test run

class apache_beam.testing.load_tests.load_test_metrics_utils.MetricsPublisher

Bases: object

Base class for metrics publishers.

publish(results)

class apache_beam.testing.load_tests.load_test_metrics_utils.ConsoleMetricsPublisher

Bases: apache_beam.testing.load_tests.load_test_metrics_utils.MetricsPublisher

A ConsoleMetricsPublisher publishes collected metrics to console output.

publish(results)

class apache_beam.testing.load_tests.load_test_metrics_utils.BigQueryMetricsPublisher(project_name, table, dataset, bq_schema=None)

Bases: apache_beam.testing.load_tests.load_test_metrics_utils.MetricsPublisher

A BigQueryMetricsPublisher publishes collected metrics to BigQuery output.

publish(results)

class apache_beam.testing.load_tests.load_test_metrics_utils.BigQueryClient(project_name, table, dataset, bq_schema=None)

Bases: object

A BigQueryClient publishes collected metrics to BigQuery output.

save(results)

class apache_beam.testing.load_tests.load_test_metrics_utils.InfluxDBMetricsPublisherOptions(measurement, db_name, hostname, user=None, password=None)

Bases: object

validate()

http_auth_enabled()

class apache_beam.testing.load_tests.load_test_metrics_utils.InfluxDBMetricsPublisher(options)

Bases: apache_beam.testing.load_tests.load_test_metrics_utils.MetricsPublisher

Publishes collected metrics to InfluxDB database.

publish(results)

class apache_beam.testing.load_tests.load_test_metrics_utils.MeasureTime(namespace)

Bases: apache_beam.transforms.core.DoFn

A distribution metric prepared to be added to pipeline as ParDo to measure runtime.

Initializes MeasureTime.

namespace(str): namespace of metric

start_bundle()

finish_bundle()

process(element)

class apache_beam.testing.load_tests.load_test_metrics_utils.MeasureBytes(namespace, extractor=None)

Bases: apache_beam.transforms.core.DoFn

Metric to measure how many bytes was observed in pipeline.

Initializes MeasureBytes.

Parameters:

• namespace (str) – metric namespace
• extractor – function to extract elements to be count

LABEL = 'total_bytes'

process(element, *args)

class apache_beam.testing.load_tests.load_test_metrics_utils.CountMessages(namespace)

Bases: apache_beam.transforms.core.DoFn

LABEL = 'total_messages'

process(element)

class apache_beam.testing.load_tests.load_test_metrics_utils.MeasureLatency(namespace)

Bases: apache_beam.transforms.core.DoFn

A distribution metric which captures the latency based on the timestamps of the processed elements.

Initializes MeasureLatency.

namespace(str): namespace of metric

LABEL = 'latency'

process(element, timestamp=TimestampParam)

class apache_beam.testing.load_tests.load_test_metrics_utils.AssignTimestamps

Bases: apache_beam.transforms.core.DoFn

DoFn to assigned timestamps to elements.

process(element)

class apache_beam.testing.load_tests.load_test_metrics_utils.BigQueryMetricsFetcher

Bases: object

fetch(query) → pandas.core.frame.DataFrame