Remove dataflow (in packages and recipe) Bug: 1040991 Change-Id: I246b1b9d7a15edf25dc145a68421eb7ced57575f Reviewed-on: https://chromium-review.googlesource.com/c/infra/infra/+/2128354 Commit-Queue: Quinten Yearsley <qyearsley@chromium.org> Reviewed-by: Andrii Shyshkalov <tandrii@google.com> Cr-Original-Commit-Position: refs/heads/master@{#30378} Cr-Mirrored-From: https://chromium.googlesource.com/infra/infra Cr-Mirrored-Commit: dc7de4675108065b876d36eadc72721016879676
diff --git a/.coveragerc b/.coveragerc deleted file mode 100644 index 92cab5e..0000000 --- a/.coveragerc +++ /dev/null
@@ -1,5 +0,0 @@ -[run] -include = ./packages/dataflow/* - -[expect_tests] -expected_coverage_min = 70
diff --git a/README.md b/README.md deleted file mode 100644 index 5179635..0000000 --- a/README.md +++ /dev/null
@@ -1,135 +0,0 @@ -# packages/dataflow - -The purpose of this package is to simplify the development of Dataflow -workflows. - -See the modules in [common](./common/README.md) for some generally useful -abstractions. - -You'll notice that workflows are included in this package. Workflows are located -here to simplify job execution, as all -non-[standard](https://beam.apache.org/documentation/) and -non-[beam](https://beam.apache.org/documentation/) modules must be packaged -together for job execution. - -See -[Scheduling a Dataflow workflow](https://chromium.googlesource.com/infra/infra/+/master/doc/users/event_pipeline.md#scheduling-a-dataflow-workflow) -for more information on automating a workflow you'd like to run regularly. - -It's possible that you may only care about running your pipeline locally. In -that case, you can simply import the common modules. - -Note that Dataflow supports continuous pipelines. Chrome Operations hasn't -experimented with these yet, but they are worth exploring! - -[TOC] - -# References - -[Beam Docs](https://beam.apache.org/documentation/) - -# Unit Testing - -From the root of the infra repository, run the command `./test.py test -packages/dataflow`. - -# Workflow Testing - -There are a couple requirements to testing your Dataflow workflow. - -First, you must activate the infra Python environment. Assuming you have that -set up already, run `source ENV/bin/activate` from the root of your infra -checkout. If you need to set up or update your environment, see -[bootstrap/README](../../bootstrap/README.md) for more info. - -Next, you must have Google Storage buckets to pass with the `--staging_location` -and `--temp_location` options. The name is not important, but for example you -could use `gs://my-dataflow-job/staging`. -[Create these](https://cloud.google.com/storage/docs/creating-buckets) if you -don't have them already. - -Next, you must have permission within the project to schedule a Dataflow job, -and be authenticated to do so. - -To check your authentication status, ensure that you have the Cloud SDK -[installed](https://cloud.google.com/sdk/docs/quickstarts), then run `gcloud -info`. - -If you don't see the correct project ID, reach out to an -[editor](https://pantheon.corp.google.com/iam-admin/iam) of that project to -request access. - -Finally, run the command below to test your workflow as a remote job. Note: Job -names should match the regular expression `[a-z]\([-a-z0-9]{0,38}[a-z0-9])`. - -``` -python <path-to-dataflow-job> --job_name <pick-a-job-name> \ ---project <project> --runner DataflowRunner \ ---setup_file <infra-checkout-path>/packages/dataflow/setup.py \ ---staging_location <staging bucket> \ ---temp_location <temp bucket> --save_main_session -``` - -Navigate to the [Dataflow console](https://console.cloud.google.com/project) in -your browser and you should see your job running. Wait until it succeeds. - -Running the test will leave behind a directory, -`packages/dataflow/dataflow.egg-info`, that you must manually clean up. - -To run the workflow locally, first set credentials using `export -GOOGLE_APPLICATION_CREDENTIALS=<path_to_credentials>` - -Then `python cq_attempts.py --output <dummy_path> --project -<name_of_test_project>` - -# Updating the package - -Changes to this directory are automatically mirrored in a synthesized -[repo](https://chromium.googlesource.com/infra/infra/packages/dataflow/). To -deploy changes to this repository: - - * Land the changes. - * Submit a separate CL that updates the version in `setup.py`. - * Build and upload a new wheel. - * Submit a single CL that updates the remote execution recipe and deps.pyl. - -Jobs scheduled with the -[`remote_execute_dataflow_workflow`](../../recipes/recipes/remote_execute_dataflow_workflow.py) -recipe use the version of the job at HEAD but the version of the package pinned -in [bootstrap/deps.pyl](../../bootstrap/deps.pyl). So, if you make a breaking -change to the package, submit the update first (which will automatically be -picked up by the -[package mirror](https://chromium.googlesource.com/infra/infra/packages/dataflow/)), -then submit the change to the job along with the ref update in `deps.pyl` together -in one commit. Be sure to follow the instructions in -[bootstrap/README.md](../../bootstrap/README.md) to build and upload the new -wheel before submitting the change to `deps.pyl`. - -# Limits - -Please see the [Dataflow docs](https://cloud.google.com/dataflow/quotas) for the -most up to date information on quotas and limits. - -At the time of writing, there are limits on Dataflow requests per minute, number -of GCE instances (`--numWorkers`), number of concurrent jobs, monitoring requests, -job creation request size, and number of side input shards. - -Some of these limits are per user, others are per project, others are per -organization. In general, users and project owners are responsible for ensuring -they do not hit user and project limits. Currently these limits are, in general, -not restrictive for us, and we are not concerned about hitting them. - -## Interaction with other cloud services - -If you run a job on GCE, GCE limits apply. If you use BigQuery as a source or -sink, BigQuery limits apply. Project owners should monitor usage and be mindful -of these limits. - -### BigQuery - -The relevant limits are query and insert limits. -[Query](https://cloud.google.com/bigquery/quotas#queries) and -[insert](https://cloud.google.com/bigquery/quotas#streaminginserts) limits can -be found in the Dataflow documentation. As query and insert rates are highly -project-specific, project owners should be responsible for monitoring this -limit.
diff --git a/__init__.py b/__init__.py deleted file mode 100644 index e69de29..0000000 --- a/__init__.py +++ /dev/null
diff --git a/common/README.md b/common/README.md deleted file mode 100644 index 56b1c6b..0000000 --- a/common/README.md +++ /dev/null
@@ -1,30 +0,0 @@ -The `common` module provides reusable classes for writing Dataflow workflows. - -## chops\_beam - -For easily constructing readable pipelines with standard defaults. - -``` -q = ('SELECT blah FROM `example_project.example_dataset.example_table`') -p = chops_beam.EventsPipeline() -_ = (p - | chops_beam.BQRead(q) - | ... # do some transforms - | chops_beam.BQWrite('example_project', 'destination_table')) -p.run() -``` - -## objects - -Convenient classes for BigQuery tables. Can be used for reading from or writing -to BigQuery. - -``` -for row in input_rows: - event = objects.CQEvent.from_bigquery_row(row) -``` - -## combine\_fns - -Generally useful [Combine -Functions](https://beam.apache.org/documentation/programming-guide/#transforms-combine).
diff --git a/common/__init__.py b/common/__init__.py deleted file mode 100644 index e69de29..0000000 --- a/common/__init__.py +++ /dev/null
diff --git a/common/chops_beam.py b/common/chops_beam.py deleted file mode 100644 index 4a9036c..0000000 --- a/common/chops_beam.py +++ /dev/null
@@ -1,44 +0,0 @@ -# Copyright 2017 The Chromium Authors. All rights reserved. -# Use of this source code is governed by a BSD-style license that can be -# found in the LICENSE file. - -import apache_beam as beam - -from apache_beam.pipeline import PipelineOptions - - -class EventsPipeline(beam.Pipeline): - """Pipeline that reads options from the command line.""" - def __init__(self): - super(EventsPipeline, self).__init__(options=PipelineOptions()) - - -class BQRead(beam.io.iobase.Read): - """Read transform created from a BigQuerySource with convenient defaults.""" - - def __init__(self, query, validate=True, coder=None, use_standard_sql=True, - flatten_results=False): - """ - Args: - query: The query to be run. Should specify table in - `project.dataset.table` form for standard SQL and - [project:dataset.table] form is use_standard_sql is False. - See beam.io.BigQuerySource for explanation of remaining arguments. - """ - source = beam.io.BigQuerySource(query=query, validate=validate, coder=coder, - flatten_results=flatten_results, - use_standard_sql=use_standard_sql) - super(BQRead, self).__init__(source) - - -class BQWrite(beam.io.Write): - """Write transform created from a BigQuerySink with convenient defaults. - - beam.io.BigQuerySink will automatically add unique insert IDs to rows, - which BigQuery uses to prevent duplicate inserts. - """ - def __init__(self, project, table, dataset='aggregated', - write_disposition=beam.io.BigQueryDisposition.WRITE_TRUNCATE): - sink = beam.io.BigQuerySink(table, dataset, project, - write_disposition=write_disposition) - super(BQWrite, self).__init__(sink)
diff --git a/common/combine_fns.py b/common/combine_fns.py deleted file mode 100644 index 2ecd6fc..0000000 --- a/common/combine_fns.py +++ /dev/null
@@ -1,51 +0,0 @@ -# Copyright 2017 The Chromium Authors. All rights reserved. -# Use of this source code is governed by a BSD-style license that can be -# found in the LICENSE file - -import apache_beam as beam - - -class ConvertToCSV(beam.CombineFn): - """Convert elements to CSV format to be written out. - - Transform for writing elements out in a CSV format. Can process elements - of type dictonary or list. This transform only supports consistent elements, - meaning dictionaries must all have the same keys and lists must have the - same length and order. If provided a header, a list must already be in that - order and a dictionary must have at least those fields. - """ - def __init__(self, header=None): - self.header = header - - def create_accumulator(self): - return [] - - def iterable(self, obj): - """Returns an iterable for a dictionary or list. - - Sorting dictionary keys assures that the CSV is in the same order - for all dictionaries. If given a header, use the header fields as - the iterable for the dictionary to preserve that order. - """ - if isinstance(obj, dict): - if self.header: - return self.header - keys = obj.keys() - return sorted(keys) - else: - return range(len(obj)) - - def add_input(self, accumulator, element): - element_string = [] - for i in self.iterable(element): - element_string.append(str(element[i])) - accumulator.append(','.join(element_string) + '\n') - return accumulator - - def merge_accumulators(self, accumulators): - return sum(accumulators, []) - - def extract_output(self, accumulator): - if self.header: - accumulator.insert(0, ','.join(self.header) + '\n') - return ''.join(accumulator)
diff --git a/common/objects.py b/common/objects.py deleted file mode 100644 index 8045790..0000000 --- a/common/objects.py +++ /dev/null
@@ -1,172 +0,0 @@ -# Copyright 2017 The Chromium Authors. All rights reserved. -# Use of this source code is governed by a BSD-style license that can be -# found in the LICENSE file. - - -class BigQueryObject(object): - """A BigQueryObject holds data that will be read from/written to BigQuery.""" - - def __eq__(self, other): - return self.__dict__ == other.__dict__ - - @staticmethod - def get_bigquery_attributes(): - """Returns a list of attributes that exist in the BigQuery schema. - - These attributes should be strings. - """ - raise NotImplementedError() - - def as_bigquery_row(self): - """Returns data in a suitable format for writing to BigQuery. - - The default behavior constructs a dictionary from the attributes listed - by get_bigquery_attributes and their values. This behavior can be - overridden. - """ - return {attr: self.__dict__.get(attr) - for attr in self.get_bigquery_attributes()} - - @classmethod - def from_bigquery_row(cls, row): - """Creates an instance of cls from a BigQuery row. - - Args: - row: dictionary in the form {field: value} where field is in - get_bigquery_attributes(). - """ - obj = cls() - for field, value in row.items(): - obj.__dict__[field] = value - return obj - - -class CQAttempt(BigQueryObject): - """A CQAttempt represents a single CQ attempt. - - It is created by aggregating all CQEvents for a given attempt. - """ - def __init__(self): - # Consistent between events for a given attempt - self.attempt_start_msec = None - self.cq_name = None - self.issue = None - self.patchset = None - self.dry_run = False - - # Patch event timestamps - self.first_start_msec = None - self.last_start_msec = None - self.first_stop_msec = None - self.last_stop_msec = None - self.patch_committed_msec = None - self.patch_started_to_commit_msec = None - self.patch_failed_msec = None - self.vcs_commit_latency_sec = None - self.click_to_failure_sec = None - self.click_to_patch_committed_sec = None - self.click_to_result_sec = None - - # Patch event bools - self.committed = False - self.was_throttled = False - self.waited_for_tree = False - self.failed = False - - # Verifier event timestamps - self.first_verifier_trigger_msec = None - self.patch_verifier_pass_msec = None - self.cq_launch_latency_sec = None - self.verifier_pass_latency_sec = None - self.tree_check_and_throttle_latency_sec = None - - # Verifier event bools - self.no_tryjobs_launched = False - self.custom_trybots = False - - self.failure_reason = None - self.max_failure_msec = None - self.fail_type = None - - self.infra_failures = 0 - self.compile_failures = 0 - self.test_failures = 0 - self.invalid_test_results_failures = 0 - self.patch_failures = 0 - self.total_failures = 0 - - self.contributing_bbucket_ids = None - self.earliest_equivalent_patchset = None - self.attempt_key = None - - @staticmethod - def get_bigquery_attributes(): - return [ - 'attempt_start_msec', - 'first_start_msec', - 'last_start_msec', - 'cq_name', - 'first_stop_msec', - 'last_stop_msec', - 'committed', - 'was_throttled', - 'waited_for_tree', - 'issue', - 'patchset', - 'dry_run', - 'cq_launch_latency_sec', - 'verifier_pass_latency_sec', - 'tree_check_and_throttle_latency_sec', - 'no_tryjobs_launched', - 'custom_trybots', - 'failed', - 'infra_failures', - 'compile_failures', - 'test_failures', - 'invalid_test_results_failures', - 'patch_failures', - 'total_failures', - 'fail_type', - 'contributing_bbucket_ids', - 'vcs_commit_latency_sec', - 'click_to_patch_committed_sec', - 'click_to_failure_sec', - 'click_to_result_sec', - 'earliest_equivalent_patchset', - 'attempt_key', - ] - - -class CQEvent(BigQueryObject): - """A CQEvent represents event data reported to BigQuery from CQ. - - CQEvents are aggregated to make CQAttempts. - """ - def __init__(self): - self.timestamp_millis = None - self.action = None - self.attempt_start_usec = None - self.cq_name = None - self.issue = None - self.patchset = None - self.failure_reason = None - self.dry_run = False - self.contributing_buildbucket_ids = None - self.earliest_equivalent_patchset = None - self.attempt_key = None - - @staticmethod - def get_bigquery_attributes(): - return [ - 'timestamp_millis', - 'action', - 'attempt_start_usec', - 'cq_name', - 'issue', - 'patchset', - 'dry_run', - 'failure_reason', - 'contributing_buildbucket_ids', - 'earliest_equivalent_patchset', - 'attempt_key', - ]
diff --git a/common/test/__init__.py b/common/test/__init__.py deleted file mode 100644 index e69de29..0000000 --- a/common/test/__init__.py +++ /dev/null
diff --git a/common/test/combine_fns_test.py b/common/test/combine_fns_test.py deleted file mode 100644 index a5833dc..0000000 --- a/common/test/combine_fns_test.py +++ /dev/null
@@ -1,44 +0,0 @@ -# Copyright 2017 The Chromium Authors. All rights reserved. -# Use of this source code is governed by a BSD-style license that can be -# found in the LICENSE file. - -import unittest - -import apache_beam as beam - -from apache_beam.testing import test_pipeline -from apache_beam.testing import util -from dataflow.common import combine_fns - - -class TestCombineFns(unittest.TestCase): - def test_convert_to_CSV_with_dicts(self): - elements = {'a': 1, 'b': 2, 'c': 3} - pipeline = test_pipeline.TestPipeline() - result = (pipeline - | beam.Create([elements, elements]) - | beam.CombineGlobally(combine_fns.ConvertToCSV()) - ) - util.assert_that(result, util.equal_to(['1,2,3\n1,2,3\n'])) - pipeline.run() - - def test_convert_to_CSV_with_lists(self): - elements = [1, 2, 3] - pipeline = test_pipeline.TestPipeline() - result = (pipeline - | beam.Create([elements, elements]) - | beam.CombineGlobally(combine_fns.ConvertToCSV()) - ) - util.assert_that(result, util.equal_to(['1,2,3\n1,2,3\n'])) - pipeline.run() - - def test_convert_to_CSV_with_header(self): - elements = {'a': 1, 'b': 2, 'c': 3} - header = ['a', 'b', 'c'] - pipeline = test_pipeline.TestPipeline() - result = (pipeline - | beam.Create([elements, elements]) - | beam.CombineGlobally(combine_fns.ConvertToCSV(header)) - ) - util.assert_that(result, util.equal_to(['a,b,c\n1,2,3\n1,2,3\n'])) - pipeline.run()
diff --git a/cq_attempts.py b/cq_attempts.py deleted file mode 100644 index 05775e6..0000000 --- a/cq_attempts.py +++ /dev/null
@@ -1,248 +0,0 @@ -# Copyright 2017 The Chromium Authors. All rights reserved. -# Use of this source code is governed by a BSD-style license that can be -# found in the LICENSE file. - -import logging -import time - -import apache_beam as beam - -from dataflow.common import chops_beam -from dataflow.common import objects - - -class CombineEventsToAttempt(beam.CombineFn): - ACTION_PATCH_START = 'PATCH_START' - ACTION_PATCH_COMMITTED = 'PATCH_COMMITTED' - ACTION_PATCH_COMMITTING = 'PATCH_COMMITTING' - ACTION_PATCH_STOP = 'PATCH_STOP' - ACTION_PATCH_THROTTLED = 'PATCH_THROTTLED' - ACTION_PATCH_TREE_CLOSED = 'PATCH_TREE_CLOSED' - ACTION_VERIFIER_TRIGGER = 'VERIFIER_TRIGGER' - ACTION_VERIFIER_PASS = 'VERIFIER_PASS' - ACTION_VERIFIER_NOTRY = 'VERIFIER_NOTRY' - ACTION_VERIFIER_CUSTOM_TRYBOTS = 'VERIFIER_CUSTOM_TRYBOTS' - ACTION_PATCH_FAILED = 'PATCH_FAILED' - # Try job fail types - FAIL_TYPE_PATCH = 'FAIL_TYPE_PATCH' - FAIL_TYPE_INFRA = 'FAIL_TYPE_INFRA' - FAIL_TYPE_COMPILE = 'FAIL_TYPE_COMPILE' - FAIL_TYPE_TEST = 'FAIL_TYPE_TEST' - FAIL_TYPE_INVALID = 'FAIL_TYPE_INVALID' - - def __init__(self): - super(CombineEventsToAttempt, self).__init__() - self.action_affects_fields = { - self.ACTION_PATCH_START: set(['first_start_msec', 'last_start_msec']), - self.ACTION_PATCH_STOP: set(['first_stop_msec', 'last_stop_msec']), - self.ACTION_PATCH_COMMITTED: set(['patch_committed_msec', 'committed']), - self.ACTION_PATCH_COMMITTING: set(['patch_started_to_commit_msec']), - self.ACTION_PATCH_THROTTLED: set(['was_throttled']), - self.ACTION_PATCH_TREE_CLOSED: set(['waited_for_tree']), - self.ACTION_VERIFIER_TRIGGER: set(['first_verifier_trigger_msec']), - self.ACTION_VERIFIER_PASS: set(['patch_verifier_pass_msec']), - self.ACTION_VERIFIER_NOTRY: set(['no_tryjobs_launched']), - self.ACTION_VERIFIER_CUSTOM_TRYBOTS: set(['custom_trybots']), - self.ACTION_PATCH_FAILED: set(['patch_failed_msec', 'failed', - 'failure_reason']), - } - self.min_timestamp_fields = set([ - 'first_start_msec', - 'first_stop_msec', - 'patch_committed_msec', - 'patch_started_to_commit_msec', - 'first_verifier_trigger_msec', - 'patch_verifier_pass_msec', - 'patch_failed_msec' - ]) - self.max_timestamp_fields = set([ - 'last_start_msec', - 'last_stop_msec', - ]) - self.logical_or_fields = set([ - 'committed', - 'was_throttled', - 'waited_for_tree', - 'failed', - 'custom_trybots', - ]) - # Fields that are copied from event to attempt. Values for these fields are - # the same for all events for a given attempt. - self.consistent_fields = set([ - 'cq_name', - 'issue', - 'patchset', - 'dry_run', - ]) - - @staticmethod - def choose_min(old, new): - if new is not None and (old is None or new < old): - return new - return old - - @staticmethod - def compute_difference(minuend, subtrahend): - if minuend is None or subtrahend is None: - return None - return minuend - subtrahend - - @staticmethod - def ms_to_sec(ms): - return ms / 1000.0 if ms is not None else None - - def create_accumulator(self): - return [] - - def add_input(self, accumulator, input_rows): - for row in input_rows: - event = objects.CQEvent.from_bigquery_row(row) - if event.attempt_start_usec is None: - logging.warn('recieved row with null attempt_start_usec: %s', row) - continue - - if event.timestamp_millis is None: - logging.warn('recieved raw with null timestamp: %s', row) - continue - - accumulator.append(event) - return accumulator - - def merge_accumulators(self, accumulators): - merged = self.create_accumulator() - for a in list(accumulators): - merged += a - return merged - - def extract_output(self, accumulator): - attempt = objects.CQAttempt() - - for event in accumulator: - attempt_start_msec = float(event.attempt_start_usec) / 1000 - if (attempt.attempt_start_msec and - attempt.attempt_start_msec != attempt_start_msec): - logging.error(('tried to combine events with different ' - 'attempt_start_msec')) - return - - attempt.attempt_start_msec = attempt_start_msec - - # Here we search for the last-reported value of some field. - if (event.failure_reason and (attempt.max_failure_msec is None - or event.timestamp_millis > attempt.max_failure_msec)): - attempt.failure_reason = event.failure_reason - attempt.max_failure_msec = event.timestamp_millis - attempt.fail_type = attempt.failure_reason['fail_type'] - if event.contributing_buildbucket_ids: - attempt_bb_ids = set(attempt.contributing_bbucket_ids or []) - attempt_bb_ids.update(event.contributing_buildbucket_ids) - attempt.contributing_bbucket_ids = sorted(attempt_bb_ids) - if event.earliest_equivalent_patchset: - attempt.earliest_equivalent_patchset = ( - event.earliest_equivalent_patchset) - if event.attempt_key: - attempt.attempt_key = event.attempt_key - - for field in self.consistent_fields: - attempt_value = attempt.__dict__.get(field) - event_value = event.__dict__.get(field) - if attempt_value and attempt_value != event_value: - logging.error('tried to combine events with inconsistent %s', field) - return - attempt.__dict__[field] = event_value - - affected_fields = self.action_affects_fields.get(event.action, []) - for field in affected_fields: - if field in self.min_timestamp_fields: - attempt.__dict__[field] = self.choose_min(attempt.__dict__.get(field), - event.timestamp_millis) - if field in self.max_timestamp_fields: - attempt.__dict__[field] = max(attempt.__dict__.get(field), - event.timestamp_millis) - if field in self.logical_or_fields: - attempt.__dict__[field] = True - - attempt.cq_launch_latency_sec = self.ms_to_sec( - self.compute_difference(attempt.first_verifier_trigger_msec, - attempt.attempt_start_msec)) - - attempt.verifier_pass_latency_sec = self.ms_to_sec( - self.compute_difference(attempt.patch_verifier_pass_msec, - attempt.attempt_start_msec)) - - attempt.tree_check_and_throttle_latency_sec = self.ms_to_sec( - self.compute_difference(attempt.patch_started_to_commit_msec, - attempt.patch_verifier_pass_msec)) - - attempt.vcs_commit_latency_sec = self.ms_to_sec( - self.compute_difference(attempt.patch_committed_msec, - attempt.patch_started_to_commit_msec)) - - attempt.click_to_failure_sec = self.ms_to_sec( - self.compute_difference(attempt.patch_failed_msec, - attempt.attempt_start_msec)) - - attempt.click_to_patch_committed_sec = self.ms_to_sec( - self.compute_difference(attempt.patch_committed_msec, - attempt.attempt_start_msec)) - - attempt.click_to_result_sec = self.ms_to_sec( - self.compute_difference(attempt.last_stop_msec, - attempt.attempt_start_msec)) - - # TODO: Deprecate. Now that we have contributing buildbucket ids, we can - # join with completed_builds to get this information. - if attempt.failure_reason: - for job in attempt.failure_reason.get('failed_try_jobs', []): - fail_type = job['fail_type'] - attempt.total_failures += 1 - if fail_type == self.FAIL_TYPE_INFRA: - attempt.infra_failures += 1 - if fail_type == self.FAIL_TYPE_COMPILE: - attempt.compile_failures += 1 - if fail_type == self.FAIL_TYPE_TEST: - attempt.test_failures += 1 - if fail_type == self.FAIL_TYPE_INVALID: - attempt.invalid_test_results_failures += 1 - if fail_type == self.FAIL_TYPE_PATCH: - attempt.patch_failures += 1 - return attempt.as_bigquery_row() - - -class ComputeAttempts(beam.PTransform): - @staticmethod - def key(event): - parts = [event.get('attempt_start_usec'), event.get('cq_name'), - event.get('issue'), event.get('patchset')] - return ':'.join([str(part) or '' for part in parts]) - - @staticmethod - def filter_incomplete_attempts(attempt): - if (attempt is not None and attempt.get('first_start_msec') - and attempt.get('last_stop_msec')): - yield attempt - - def expand(self, pcoll): - return (pcoll - | beam.Map(lambda e: (self.key(e), e)) - | beam.GroupByKey() - | beam.CombinePerKey(CombineEventsToAttempt()) - | beam.Map(lambda (k, v): v) - | beam.FlatMap(self.filter_incomplete_attempts)) - - -def main(): - q = ('SELECT timestamp_millis, action, attempt_start_usec, cq_name, issue,' - ' patchset, dry_run, failure_reason, contributing_buildbucket_ids, ' - ' earliest_equivalent_patchset, attempt_key ' - 'FROM `chrome-infra-events.raw_events.cq`') - p = chops_beam.EventsPipeline() - _ = (p - | chops_beam.BQRead(q) - | ComputeAttempts() - | chops_beam.BQWrite('chrome-infra-events', 'cq_attempts')) - p.run() - - -if __name__ == '__main__': - main()
diff --git a/new_cq_attempts.py b/new_cq_attempts.py deleted file mode 100644 index fd32303..0000000 --- a/new_cq_attempts.py +++ /dev/null
@@ -1,126 +0,0 @@ -# Copyright 2019 The Chromium Authors. All rights reserved. -# Use of this source code is governed by a BSD-style license that can be -# found in the LICENSE file. - -import apache_beam as beam - -from dataflow import cq_attempts as sanitize_cq_attempts -from dataflow.common import chops_beam - - -class ExtractBuildBucketIdFn(beam.DoFn): - def process(self, cq_attempt_with_key): - # For a CQ attempt, we create one row for each contributing BuildBucket id. - key = cq_attempt_with_key[0] - cq_attempt_dict = cq_attempt_with_key[1] - - bb_ids = cq_attempt_dict.get('contributing_bbucket_ids') - if bb_ids: - for bb_id in bb_ids: - yield str(bb_id), key - -class FilterJoinedBuildBucketCQAttempt(beam.DoFn): - def process(self, joined_result): - # The key is BuildBucket ID. We expect there to be exactly 1 cq_attempt, and - # up to 1 BuildBucket entry. - cq_attempt_key = joined_result[1]['cq_attempt_key'] - bb_entry = joined_result[1]['bb_entries'] - if len(bb_entry) != 1 or len(cq_attempt_key) != 1: - return - yield cq_attempt_key[0], bb_entry[0] - -def update_with_presubmit_failure(input_tuple): - value = input_tuple[1] - cq_attempts = value['cq_attempts'] - assert len(cq_attempts) == 1, "There must be 1 cq_attempt." - cq_attempt = cq_attempts[0] - - if cq_attempt['fail_type'] == 'FAILED_JOBS': - buildbucket_results = value['bb_entries'] - presubmit_failures = 0 - other_failures = 0 - for bb_result in buildbucket_results: - if (bb_result['status'] == 'FAILURE' and - bb_result['builder'] == 'chromium_presubmit'): - presubmit_failures += 1 - elif bb_result['status'] != 'SUCCESS': - other_failures += 1 - if presubmit_failures >= 1 and other_failures == 0: - cq_attempt['fail_type'] = 'FAILED_PRESUBMIT_BOT' - - # Dictionaries are supposed to be returned in a single element list. - return [cq_attempt] - -def process_input(cq_events_pcol, bb_entries_pcol): - """Sets up the pipeline stages to return aggregated cq attempts pcol. - - This function performs two tasks: - 1) Computes CQ attempts from raw CQ events. This includes data sanitization. - 2) If a CQ attempt fails only because of 'chromium_presubmit' builder, sets - the failure status to 'FAILED_PRESUBMIT_BOT'. - """ - # Pcol of cq_attempt_as_dict - sanitized_cq_attempts = ( - cq_events_pcol | sanitize_cq_attempts.ComputeAttempts()) - - # Create Pcol of tuples: (cq_attempt_key, cq_attempt_as_dict) - def extract_key(cq_attempt_dict): - key_parts = [ - cq_attempt_dict.get('attempt_start_msec'), - cq_attempt_dict.get('cq_name'), - cq_attempt_dict.get('issue'), - cq_attempt_dict.get('patchset') - ] - key = ':'.join([str(part) or '' for part in key_parts]) - return key, cq_attempt_dict - cq_attempts_with_key = sanitized_cq_attempts | beam.Map(extract_key) - - # Create Pcol of tuples: (build_bucket_id, cq_attempt_key) - cq_attempt_key_keyed_by_bb_id = cq_attempts_with_key | beam.ParDo( - ExtractBuildBucketIdFn()) - - # Create Pcol of tuples: (build_bucket_id, build_bucket_entry) - bb_entry_keyed_by_bb_id = bb_entries_pcol | beam.Map( - lambda e: (str(e.get('id')), e)) - - # Create Pcol of tuples: (cq_attempt_key, BuildBucket entry) - bb_entries_keyed_by_cq_attempt_key = ({ - 'bb_entries' : bb_entry_keyed_by_bb_id, - 'cq_attempt_key': cq_attempt_key_keyed_by_bb_id - } | 'Join BuildBucket with cq attempts' >> beam.CoGroupByKey() - | beam.ParDo(FilterJoinedBuildBucketCQAttempt()) - ) - - # Uses BuildBucket entries associated with a CQ attempt to potentially change - # the failure reason to FAILED_PRESUBMIT_BOT. Creates a Pcol of - # cq_attempt_as_dict. - results = ({ - 'cq_attempts' : cq_attempts_with_key, - 'bb_entries' : bb_entries_keyed_by_cq_attempt_key - } | beam.CoGroupByKey() - | beam.FlatMap(update_with_presubmit_failure) - ) - return results - -def main(): - p = chops_beam.EventsPipeline() - q = ('SELECT timestamp_millis, action, attempt_start_usec, cq_name, issue,' - ' patchset, dry_run, failure_reason, contributing_buildbucket_ids, ' - ' earliest_equivalent_patchset ' - 'FROM `chrome-infra-events.raw_events.cq`') - cq_events_pcol = p | 'read raw CQ events' >> chops_beam.BQRead(q) - - q = ('SELECT id, builder.builder, status from ' - '`cr-buildbucket.chromium.builds`') - bb_entries_pcol = p | 'read BuildBucket' >> chops_beam.BQRead(q) - - results = process_input(cq_events_pcol, bb_entries_pcol) - - # pylint: disable=expression-not-assigned - results | chops_beam.BQWrite('chrome-infra-events', 'cq_attempts') - - p.run() - - -if __name__ == '__main__': - main()
diff --git a/setup.py b/setup.py deleted file mode 100644 index fc596b9..0000000 --- a/setup.py +++ /dev/null
@@ -1,18 +0,0 @@ -# Copyright 2017 The Chromium Authors. All rights reserved. -# Use of this source code is governed by a BSD-style license that can be -# found in the LICENSE file. - -from setuptools import setup - -setup( - name='dataflow', - version='0.0.6', - description='Chrome Infra Dataflow Workflows', - long_description=('This package includes Chrome Infra workflows as well as ' - 'common modules.'), - classifiers=[ - 'Programming Language :: Python :: 2.7', - ], - package_dir={'dataflow': ''}, - packages=['dataflow', 'dataflow.common'], -)
diff --git a/test/__init__.py b/test/__init__.py deleted file mode 100644 index e69de29..0000000 --- a/test/__init__.py +++ /dev/null
diff --git a/test/cq_attempts_test.py b/test/cq_attempts_test.py deleted file mode 100644 index c93bb9b..0000000 --- a/test/cq_attempts_test.py +++ /dev/null
@@ -1,301 +0,0 @@ -# Copyright 2017 The Chromium Authors. All rights reserved. -# Use of this source code is governed by a BSD-style license that can be -# found in the LICENSE file. - -import unittest - -import apache_beam as beam -from dataflow import cq_attempts as job - -from apache_beam.testing import test_pipeline -from apache_beam.testing import util -from dataflow.common import objects - - -class TestCQAttemptAccumulator(unittest.TestCase): - def setUp(self): - self.attempt_start_usec = 1493833887566000 - self.attempt_start_msec = self.attempt_start_usec / 1000 - self.timestamp_msec = 1493833887688 - self.earlier_timestamp = self.timestamp_msec - 1 - self.later_timestamp = self.timestamp_msec + 1 - self.combFn = job.CombineEventsToAttempt() - - @staticmethod - def construct_attempt_values(attempt_start_usec, actions, - failure_reasons=None): - cq_name = 'test_cq' - issue = '1' - patchset = '1' - - event_basic = { - 'attempt_start_usec': attempt_start_usec, - 'cq_name': cq_name, - 'issue': issue, - 'patchset': patchset, - } - - events = [] - for i, action in enumerate(actions): - event = event_basic.copy() - event.update({ - 'action': action[0], - 'timestamp_millis': action[1], - 'failure_reason': failure_reasons[i] if failure_reasons else None, - }) - events.append(event) - - attempt = objects.CQAttempt() - attempt.cq_name = cq_name - attempt.attempt_start_msec = float(attempt_start_usec) / 1000 - attempt.issue = issue - attempt.patchset = patchset - - return (events, attempt) - - - def failed_attempt_values(self, attempt_start_usec): - actions = [ - (self.combFn.ACTION_PATCH_START, 1000), - (self.combFn.ACTION_VERIFIER_CUSTOM_TRYBOTS, 2000), - (self.combFn.ACTION_PATCH_FAILED, 5000), - (self.combFn.ACTION_PATCH_FAILED, 6000), - (self.combFn.ACTION_PATCH_STOP, 9000), - ] - - failure_reasons = [ - None, - None, - { - 'fail_type': 'FAIL_TYPE_1', - 'failed_try_jobs': [ - {'fail_type': self.combFn.FAIL_TYPE_TEST}, - {'fail_type': self.combFn.FAIL_TYPE_TEST}, - {'fail_type': self.combFn.FAIL_TYPE_TEST}, - {'fail_type': self.combFn.FAIL_TYPE_PATCH}, - ] - }, - { - 'fail_type': 'FAIL_TYPE_2', - 'failed_try_jobs': [ - {'fail_type': self.combFn.FAIL_TYPE_COMPILE}, - {'fail_type': self.combFn.FAIL_TYPE_INVALID}, - ] - }, - None, - ] - - events, attempt = self.construct_attempt_values(attempt_start_usec, actions, - failure_reasons) - - attempt.first_start_msec = 1000 - attempt.last_start_msec = 1000 - attempt.last_stop_msec = 9000 - attempt.first_stop_msec = 9000 - attempt.fail_type = 'FAIL_TYPE_2' - attempt.failed = True - attempt.patch_failed_msec = 5000 - attempt.invalid_test_results_failures = 1 - attempt.compile_failures = 1 - attempt.total_failures = 2 - attempt.custom_trybots = True - attempt.click_to_failure_sec = 4.0 - attempt.click_to_result_sec = 8.0 - - return (events, attempt.as_bigquery_row()) - - - def complete_attempt_values(self, attempt_start_usec): - actions = [ - (self.combFn.ACTION_PATCH_START, 1000), - (self.combFn.ACTION_VERIFIER_TRIGGER, 3000), - (self.combFn.ACTION_VERIFIER_PASS, 4000), - (self.combFn.ACTION_PATCH_COMMITTING, 5000), - (self.combFn.ACTION_PATCH_COMMITTED, 6000), - (self.combFn.ACTION_PATCH_STOP, 9000), - ] - - events, attempt = self.construct_attempt_values(attempt_start_usec, actions) - - events[-2]['contributing_buildbucket_ids'] = [1] - events[-1]['contributing_buildbucket_ids'] = [2, 3] - - attempt.first_start_msec = 1000 - attempt.last_start_msec = 1000 - attempt.last_stop_msec = 9000 - attempt.first_stop_msec = 9000 - attempt.cq_launch_latency_sec = 3.0 - attempt.verifier_pass_latency_sec = 4.0 - attempt.tree_check_and_throttle_latency_sec = 1.0 - attempt.vcs_commit_latency_sec = 1.0 - attempt.click_to_patch_committed_sec = 6.0 - attempt.click_to_result_sec = 9.0 - attempt.committed = True - attempt.contributing_bbucket_ids = [1, 2, 3] - - return (events, attempt.as_bigquery_row()) - - def test_compute_attempts(self): - complete_attempt_events, complete_attempt = self.complete_attempt_values( - attempt_start_usec=0) - failed_attempt_events, failed_attempt = self.failed_attempt_values( - attempt_start_usec=1000000) # 1 second - - incomplete_attempt_events = [ - { - 'timestamp_millis': 2, - 'action': self.combFn.ACTION_PATCH_START, - 'attempt_start_usec': 1, - 'cq_name': 'test_cq', - 'issue': '2', - 'patchset': '1', - }, - ] - - events = (complete_attempt_events + failed_attempt_events + - incomplete_attempt_events) - expected_attempts = [complete_attempt, failed_attempt] - - p = test_pipeline.TestPipeline() - pcoll = (p - | beam.Create(events) - | job.ComputeAttempts()) - util.assert_that(pcoll, util.equal_to(expected_attempts)) - p.run() - - def basic_event(self, action=None, timestamp_millis=None, - attempt_start_usec=None, cq_name=None): - event = objects.CQEvent() - event.attempt_start_usec = (attempt_start_usec if attempt_start_usec else - self.attempt_start_usec) - event.timestamp_millis = (timestamp_millis if timestamp_millis else - self.timestamp_msec) - event.action = action if action else self.combFn.ACTION_PATCH_START - event.cq_name = cq_name if cq_name else 'test_cq' - event.issue = '123' - event.patchset = '456' - event.dry_run = True - return event - - def test_null_attempt_start_not_included(self): - accumulator = self.combFn.add_input(self.combFn.create_accumulator(), - [{'attempt_start_usec': None}]) - self.assertEqual(accumulator, []) - - def test_null_timestamp_not_included(self): - accumulator = self.combFn.add_input(self.combFn.create_accumulator(), - [{'timestamp_millis': None}]) - self.assertEqual(accumulator, []) - - def test_add_input(self): - row = { - 'attempt_start_usec': self.attempt_start_usec, - 'timestamp_millis': self.timestamp_msec, - 'action': self.combFn.ACTION_PATCH_START, - } - event = objects.CQEvent.from_bigquery_row(row) - accumulator = self.combFn.add_input(self.combFn.create_accumulator(), [row]) - self.assertEqual(accumulator, [event]) - - def test_extract_min_timestamp_one_timestamp(self): - accumulator = [self.basic_event()] - attempt = self.combFn.extract_output(accumulator) - self.assertEqual(attempt['first_start_msec'], self.timestamp_msec) - - def test_extract_min_timestamp(self): - accumulator = [ - self.basic_event(timestamp_millis=self.timestamp_msec), - self.basic_event(timestamp_millis=self.earlier_timestamp), - self.basic_event(timestamp_millis=self.later_timestamp), - ] - attempt = self.combFn.extract_output(accumulator) - self.assertEqual(attempt['first_start_msec'], self.earlier_timestamp) - - def test_extract_max_timestamp(self): - accumulator = [ - self.basic_event(timestamp_millis=self.timestamp_msec), - self.basic_event(timestamp_millis=self.later_timestamp), - self.basic_event(timestamp_millis=self.earlier_timestamp), - ] - attempt = self.combFn.extract_output(accumulator) - self.assertEqual(attempt['last_start_msec'], self.later_timestamp) - - def test_extract_attempt_start(self): - accumulator = [self.basic_event()] - attempt = self.combFn.extract_output(accumulator) - self.assertEqual(attempt['attempt_start_msec'], self.attempt_start_msec) - - def test_extract_different_attempt_start(self): - accumulator = [ - self.basic_event(attempt_start_usec=self.attempt_start_usec), - self.basic_event(attempt_start_usec=self.attempt_start_usec+1000) - ] - self.assertIsNone(self.combFn.extract_output(accumulator)) - - def test_earliest_equivalent_patchset(self): - event = self.basic_event() - event.earliest_equivalent_patchset = 455 - attempt = self.combFn.extract_output([event]) - self.assertEqual(attempt['earliest_equivalent_patchset'], 455) - - def test_attempt_key(self): - event = self.basic_event() - event.attempt_key = 'deadbeef512' - attempt = self.combFn.extract_output([event]) - self.assertEqual(attempt['attempt_key'], 'deadbeef512') - - def test_extract_consistent_field(self): - event = self.basic_event() - attempt = self.combFn.extract_output([event]) - for field in self.combFn.consistent_fields: - self.assertEqual(attempt[field], event.__dict__[field]) - - def test_extract_different_consistent_field(self): - accumulator = [ - self.basic_event(), - self.basic_event(cq_name='different_cq_name') - ] - self.assertIsNone(self.combFn.extract_output(accumulator)) - - def test_extract_logical_or(self): - accumulator = [self.basic_event(action=self.combFn.ACTION_PATCH_COMMITTED)] - attempt = self.combFn.extract_output(accumulator) - self.assertTrue(attempt['committed']) - - def test_filter_incomplete_attempts(self): - test_cases = [ - { - 'attempt': { - 'first_start_msec': self.timestamp_msec, - 'last_stop_msec': self.timestamp_msec, - }, - 'filtered_expected': False - }, - { - 'attempt': { - 'first_start_msec': None, - 'last_stop_msec': self.timestamp_msec, - }, - 'filtered_expected': True - }, - { - 'attempt': { - 'first_start_msec': self.timestamp_msec, - 'last_stop_msec': None, - }, - 'filtered_expected': True - } - ] - for test_case in test_cases: - attempt = test_case['attempt'] - filter_attempts = job.ComputeAttempts.filter_incomplete_attempts - if test_case['filtered_expected']: - with self.assertRaises(StopIteration): - filtered_attempt = filter_attempts(attempt).next() - else: - filtered_attempt = filter_attempts(attempt).next() - self.assertEqual(filtered_attempt, attempt) - - -if __name__ == '__main__': - unittest.main()
diff --git a/test/new_cq_attempts_test.py b/test/new_cq_attempts_test.py deleted file mode 100644 index b585328..0000000 --- a/test/new_cq_attempts_test.py +++ /dev/null
@@ -1,151 +0,0 @@ -# Copyright 2019 The Chromium Authors. All rights reserved. -# Use of this source code is governed by a BSD-style license that can be -# found in the LICENSE file. - - -import unittest - -import apache_beam as beam -from apache_beam.testing import test_pipeline -from apache_beam.testing import util - -from dataflow import cq_attempts as sanitize_cq_attempts -from dataflow import new_cq_attempts - - -def _beam_equals(expected, actual): - if expected != actual: - raise util.BeamAssertException( - 'Expected {} to equal {}'.format(expected, actual)) - - -class IntegrationTest(unittest.TestCase): - def setUp(self): - self.pipeline = test_pipeline.TestPipeline() - - def construct_cq_events_pcol(self, failure_reason=None): - attempt_start_usec = 1000000 - cq_name = 'test_cq' - issue = '1' - patchset = '1' - combine_class = sanitize_cq_attempts.CombineEventsToAttempt - actions = [ - combine_class.ACTION_PATCH_START, - combine_class.ACTION_VERIFIER_CUSTOM_TRYBOTS, - combine_class.ACTION_PATCH_FAILED, - combine_class.ACTION_PATCH_FAILED, - combine_class.ACTION_PATCH_STOP, - ] - - event_basic = { - 'attempt_start_usec': attempt_start_usec, - 'cq_name': cq_name, - 'issue': issue, - 'patchset': patchset, - } - - events = [] - for i, action in enumerate(actions): - event = event_basic.copy() - event_failure_reason = (failure_reason if - action == combine_class.ACTION_PATCH_FAILED else None) - event.update({ - 'action': action, - 'timestamp_millis': 1000 * (i + 1), - 'failure_reason': event_failure_reason, - }) - if action == combine_class.ACTION_VERIFIER_CUSTOM_TRYBOTS: - event['contributing_buildbucket_ids'] = [11, 12] - events.append(event) - - return self.pipeline | beam.Create(events) - - def construct_bb_entries_pcol(self, builder_name, status): - bb_entries = [{'id':11, 'builder': builder_name, 'status':status}] - return self.pipeline | 'Construct BuildBucket entries' >> beam.Create( - bb_entries) - - # One CQ attempt, with matching BuildBucket entries that all pass. - def test_basic_pass(self): - cq_events_pcol = self.construct_cq_events_pcol() - bb_entries_pcol = self.construct_bb_entries_pcol('random_builder', - 'SUCCESS') - results = new_cq_attempts.process_input(cq_events_pcol, bb_entries_pcol) - - # There should be exactly 1 CQ attempt, with no fail type. - def expectation_checker(cq_attempts): - _beam_equals(len(cq_attempts), 1) - _beam_equals(cq_attempts[0]['fail_type'], None) - - util.assert_that(results, expectation_checker) - self.pipeline.run() - - def test_basic_failure_random_builder(self): - failure_reason = {'fail_type': 'FAILED_JOBS'} - cq_events_pcol = self.construct_cq_events_pcol(failure_reason) - bb_entries_pcol = self.construct_bb_entries_pcol('random_builder', - 'FAILURE') - results = new_cq_attempts.process_input(cq_events_pcol, bb_entries_pcol) - - # There should be exactly 1 CQ attempt, with no fail type. - def expectation_checker(cq_attempts): - _beam_equals(len(cq_attempts), 1) - _beam_equals(cq_attempts[0]['fail_type'], 'FAILED_JOBS') - - util.assert_that(results, expectation_checker) - self.pipeline.run() - - def test_basic_failure_chromium_presubmit(self): - failure_reason = {'fail_type': 'FAILED_JOBS'} - cq_events_pcol = self.construct_cq_events_pcol(failure_reason) - bb_entries_pcol = self.construct_bb_entries_pcol( - 'chromium_presubmit', 'FAILURE') - results = new_cq_attempts.process_input(cq_events_pcol, bb_entries_pcol) - - # There should be exactly 1 CQ attempt, with no fail type. - def expectation_checker(cq_attempts): - _beam_equals(len(cq_attempts), 1) - _beam_equals(cq_attempts[0]['fail_type'], 'FAILED_PRESUBMIT_BOT') - - util.assert_that(results, expectation_checker) - self.pipeline.run() - - def test_two_failures(self): - failure_reason = {'fail_type': 'FAILED_JOBS'} - cq_events_pcol = self.construct_cq_events_pcol(failure_reason) - bb_entries = [ - {'id':11, 'builder': 'chromium_presubmit', 'status':'FAILURE'}, - {'id':12, 'builder': 'win7-rel', 'status':'INFRA_FAILURE'}, - ] - bb_entries_pcol = (self.pipeline | 'Construct BuildBucket entries' >> - beam.Create(bb_entries)) - results = new_cq_attempts.process_input(cq_events_pcol, bb_entries_pcol) - - # There should be exactly 1 CQ attempt, with no fail type. - def expectation_checker(cq_attempts): - _beam_equals(len(cq_attempts), 1) - _beam_equals(cq_attempts[0]['fail_type'], 'FAILED_JOBS') - - util.assert_that(results, expectation_checker) - self.pipeline.run() - - def test_cq_attempt_no_bb_entries(self): - cq_events_pcol = self.construct_cq_events_pcol() - bb_entries_pcol = ( - self.pipeline | 'Construct 0 BuildBucket entries' >> beam.Create([])) - - results = new_cq_attempts.process_input(cq_events_pcol, bb_entries_pcol) - - # There should be exactly 1 CQ attempt, with no fail type. - def expectation_checker(cq_attempts): - _beam_equals(len(cq_attempts), 1) - cq_attempt = cq_attempts[0] - _beam_equals(cq_attempt['fail_type'], None) - _beam_equals(cq_attempt['issue'], '1') - - util.assert_that(results, expectation_checker) - self.pipeline.run() - - -if __name__ == '__main__': - unittest.main()