build finished handler for revision result creation

Summary: This implements the algorithm for selective testing result propagation as a build.finished listener.

Test Plan: unit tests

Reviewers: anupc

Reviewed By: anupc

Subscribers: changesbot, kylec, wwu, anupc

Differential Revision: https://tails.corp.dropbox.com/D232009

changes/config.py

@@ -255,6 +255,7 @@ def create_app(_read_config=True, **config):
         ('changes.listeners.phabricator_listener.build_finished_handler', 'build.finished'),
         ('changes.listeners.analytics_notifier.build_finished_handler', 'build.finished'),
         ('changes.listeners.analytics_notifier.job_finished_handler', 'job.finished'),
+        ('changes.listeners.revision_result.revision_result_build_finished_handler', 'build.finished'),
         ('changes.listeners.stats_notifier.build_finished_handler', 'build.finished'),
         ('changes.listeners.snapshot_build.build_finished_handler', 'build.finished'),
     )
@@ -459,6 +460,8 @@ def create_app(_read_config=True, **config):
         r'^--define=[A-Za-z0-9=]+',
     ]
 
+    app.config['SELECTIVE_TESTING_PROPAGATION_LIMIT'] = 30
+
     # Jobsteps go from 'pending_allocation' to 'allocated' once an external scheduler claims them, and
     # once they begin running they're updated to 'in_progress'. If the scheduler somehow fails or drops
     # the task, this value is used to time out the 'allocated' status and revert back to 'pending_allocation'.

changes/lib/revision_lib.py

@@ -0,0 +1,30 @@
+from typing import List  # NOQA
+from uuid import UUID  # NOQA
+
+from changes.constants import Status
+from changes.lib.build_type import get_any_commit_build_filters
+from changes.models.build import Build
+from changes.models.revision import Revision
+from changes.models.source import Source
+
+
+def get_latest_finished_build_for_revision(revision_sha, project_id):
+    # type: (str, UUID) -> Build
+    return Build.query.join(
+        Source, Build.source_id == Source.id,
+    ).filter(
+        Build.project_id == project_id,
+        Build.status == Status.finished,
+        Source.revision_sha == revision_sha,
+        *get_any_commit_build_filters()
+    ).order_by(
+        Build.date_created.desc(),
+    ).first()
+
+
+def get_child_revisions(revision):
+    # type: (Revision) -> List[Revision]
+    return Revision.query.filter(
+        Revision.repository_id == revision.repository_id,
+        Revision.parents.any(revision.sha),
+    ).all()

changes/listeners/revision_result.py

@@ -0,0 +1,130 @@
+import logging
+
+from collections import defaultdict
+from flask import current_app
+from uuid import UUID  # NOQA
+
+from changes.config import db
+from changes.constants import ResultSource
+from changes.db.utils import create_or_update
+from changes.lib.build_type import is_any_commit_build
+from changes.lib.revision_lib import get_child_revisions, get_latest_finished_build_for_revision
+from changes.models.bazeltarget import BazelTarget
+from changes.models.build import Build
+from changes.models.job import Job
+from changes.models.jobplan import JobPlan
+from changes.models.project import Project
+from changes.models.revision import Revision
+from changes.models.revisionresult import RevisionResult
+from changes.utils.agg import aggregate_result
+
+
+logger = logging.getLogger(__name__)
+
+
+def revision_result_build_finished_handler(build_id, **kwargs):
+    build = Build.query.get(build_id)
+    if build is None:
+        return
+    if not is_any_commit_build(build):
+        return
+    create_or_update_revision_result(
+        revision_sha=build.source.revision_sha,
+        project_id=build.project_id,
+        propagation_limit=current_app.config[
+            'SELECTIVE_TESTING_PROPAGATION_LIMIT'],
+    )
+
+
+def create_or_update_revision_result(revision_sha, project_id, propagation_limit):
+    """Given a revision sha and project ID, try to update the revision result
+    for it. This involves copying results for unaffected Bazel targets from
+    the latest parent build.
+
+    `propagation_limit` is used to control how many times this function will
+    be called recursively on the revision's children. If it is 0, then this
+    function only updates the current revision's revision result and does
+    not do any recursion.
+    """
+    # type: (str, UUID, int) -> None
+    project = Project.query.get(project_id)
+    revision = Revision.query.filter(
+        Revision.sha == revision_sha,
+        Revision.repository_id == project.repository_id,
+    ).first()
+    last_finished_build = get_latest_finished_build_for_revision(
+        revision_sha, project_id)
+    if not last_finished_build:
+        return
+
+    unaffected_targets = BazelTarget.query.join(
+        Job, BazelTarget.job_id == Job.id,
+    ).filter(
+        BazelTarget.result_source == ResultSource.from_parent,
+        Job.build_id == last_finished_build.id,
+    ).all()
+
+    if len(unaffected_targets) > 0 and len(revision.parents) > 0:
+        # TODO(naphat) there's probably a better way to select parent,
+        # but that happens rarely enough that it can be punted for now
+        parent_revision_sha = revision.parents[0]
+
+        # TODO(naphat) we should find a better way to select parent builds.
+        # Even if a parent build is not finished, we can already start to
+        # take a look at target results, as it may already have results
+        # for all of our unaffected_targets
+        # perhaps an optimization is to take the latest build, instead
+        # of the latest finished build. Finished build are more likely
+        # to have the complete set of targets we need though. But if
+        # a finished build is not the latest build, then maybe that
+        # finished build had an infra failure. Anyways, for simplicity,
+        # let's stick to finished build for now.
+        parent_build = get_latest_finished_build_for_revision(
+            parent_revision_sha, project_id)
+        if parent_build:
+            # group unaffected targets by jobs
+            unaffected_targets_groups = defaultdict(lambda: {})
+            for target in unaffected_targets:
+                unaffected_targets_groups[target.job_id][target.name] = target
+
+            # process targets in batch, grouped by job id
+            # almost always, this is going to be a single job - there is
+            # usually only one autogenerated plan per project.
+            for job_id, targets_dict in unaffected_targets_groups.iteritems():
+                jobplan = JobPlan.query.filter(
+                    JobPlan.project_id == project_id,
+                    JobPlan.build_id == last_finished_build.id,
+                    JobPlan.job_id == job_id,
+                ).first()
+                if not jobplan:
+                    continue
+                parent_targets = BazelTarget.query.join(
+                    Job, BazelTarget.job_id == Job.id,
+                ).join(
+                    JobPlan, BazelTarget.job_id == JobPlan.job_id,
+                ).filter(
+                    Job.build_id == parent_build.id,
+                    BazelTarget.name.in_(targets_dict),
+                    JobPlan.plan_id == jobplan.plan_id,
+                )
+                for parent_target in parent_targets:
+                    targets_dict[parent_target.name].result = parent_target.result
+                    db.session.add(targets_dict[parent_target.name])
+        else:
+            logger.info("Revision %s could not find a parent build for parent revision %s.", revision_sha, parent_revision_sha)
+
+    create_or_update(RevisionResult, where={
+        'revision_sha': revision_sha,
+        'project_id': project_id,
+    }, values={
+        'build_id': last_finished_build.id,
+        'result': aggregate_result([last_finished_build.result] + [t.result for t in unaffected_targets]),
+    })
+
+    db.session.commit()
+
+    if propagation_limit > 0:
+        # TODO stop the propagation if nothing changed
+        for child_revision in get_child_revisions(revision):
+            create_or_update_revision_result(
+                child_revision.sha, project_id, propagation_limit=propagation_limit - 1)

changes/testutils/fixtures.py

@@ -28,6 +28,7 @@
 from changes.models.project import Project, ProjectOption
 from changes.models.repository import Repository, RepositoryStatus
 from changes.models.revision import Revision
+from changes.models.revisionresult import RevisionResult
 from changes.models.snapshot import Snapshot, SnapshotImage
 from changes.models.source import Source
 from changes.models.step import Step
@@ -290,6 +291,12 @@ def create_project(self, **kwargs):
 
         return project
 
+    def create_revision_result(self, **kwargs):
+        revision_result = RevisionResult(**kwargs)
+        db.session.add(revision_result)
+        db.session.commit()
+        return revision_result
+
     def create_change(self, project, **kwargs):
         kwargs.setdefault('label', 'Sample')
 
@@ -319,7 +326,7 @@ def create_test(self, job, **kwargs):
 
         return case
 
-    def create_target(self, job, jobstep, **kwargs):
+    def create_target(self, job, jobstep=None, **kwargs):
         kwargs.setdefault('name', '//' + uuid4().hex + ':test')
 
         target = BazelTarget(

conftest.py

@@ -52,6 +52,7 @@ def app(request, session_config):
         SNAPSHOT_S3_BUCKET='snapshot-bucket',
         MAX_EXECUTORS=10,
         BAZEL_ARTIFACT_SUFFIX='.bazel',
+        SELECTIVE_TESTING_PROPAGATION_LIMIT=1,
     )
     app_context = app.test_request_context()
     context = app_context.push()

tests/changes/lib/test_revision_lib.py

@@ -0,0 +1,31 @@
+from changes.constants import Status
+from changes.lib.revision_lib import get_latest_finished_build_for_revision, get_child_revisions
+from changes.testutils.cases import TestCase
+
+
+class GetLatestFinishedCommitBuildTestCase(TestCase):
+    def test_correct(self):
+        project = self.create_project()
+        source = self.create_source(project)
+        diff_source = self.create_source(project, revision_sha=source.revision_sha, patch=self.create_patch())
+        self.create_build(source=source, status=Status.finished, project=project)
+        self.create_build(source=source, status=Status.finished, project=project)
+        latest_build = self.create_build(source=source, project=project)
+        self.create_build(source=source, status=Status.in_progress, project=project)  # in progress
+        self.create_build(source=diff_source, status=Status.finished, project=project)  # diff build
+        self.create_build(status=Status.finished, project=project)  # different commit
+
+        assert get_latest_finished_build_for_revision(source.revision_sha, project.id)
+
+
+class GetChildRevisions(TestCase):
+    def test_correct(self):
+        repository = self.create_repo()
+        parent1 = self.create_revision(repository=repository)
+        parent2 = self.create_revision(repository=repository)
+        child1 = self.create_revision(parents=[parent2.sha], repository=repository)
+        child2 = self.create_revision(parents=[parent1.sha, parent2.sha], repository=repository)
+        self.create_revision(parents=[parent1.sha])  # different repo
+
+        assert get_child_revisions(parent1) == [child2]
+        assert set(get_child_revisions(parent2)) == set([child1, child2])