workers: revision worker implementation

WIP DO NOT MERGE
Commit message TBD

- add abstract Worker class (bug 1744327)
- add main worker flag and capacity/throttle flags
- add many to many fields + association to revisions/landing jobs
- add method to parse diff and list affected files
- add more test coverage for revision_worker.py
- add mots integration (bug 1740107)
- add new RevisionWorker that pre-processes revisions (bug 1788728)
- add new RevisionWorker that pre-processes revisions (bug 1788728)
- add new start/stop commands to manage workers
- add new flags to stop workers gracefully (*_WORKER_STOPPED)
- add patch caching on disk
- add proper loop/process functionality to workers
- add repo.use_revision_worker feature flag (bug 1788732)
- add mots hashes check
- improved edge search functionality
- implement stack hashes to detect changes in revisions (via get_stack_hashes)
- include new Lando revision info via API endpoint
- refactor dependency and stack fetching and parsing using networkx
- refactored revision worker and landing worker to use Worker class
- remove s3/boto/etc. dependencies (bug 1753728)
- rename old command lando-cli landing-worker to lando-cli start-landing-worker
- run pre/post mots query
- store mots output in revision model

Author: zzzeid

.flake8

@@ -1,7 +1,7 @@
 [flake8]
 max-line-length = 88
 select = C,E,F,W,B,B9
-ignore = E203, E501, W503, B006
+ignore = E203, E501, W503, B006, E712, E711
 exclude =
     .hg,
     .git,

Dockerfile

@@ -54,8 +54,13 @@ RUN cd / && pip install --no-cache /app
 ENV PYTHONPATH /app
 RUN chown -R app:app /app
 
-# Create repos directory for transplanting in landing-worker
+# Create repos directory for landing-worker and revision worker.
 RUN mkdir /repos
+RUN chown -R app:app /repos
+
+# Create patches directory to cache patches.
+RUN mkdir /patches
+RUN chown -R app:app /patches
 
 # Run as a non-privileged user
 USER app

Dockerfile-dev

@@ -20,6 +20,7 @@ ENV PYTHONUNBUFFERED=1
 ENV FLASK_RUN_PORT=9000
 ENV FLASK_RUN_HOST=0.0.0.0
 ENV FLASK_DEBUG=1
+ENV HTTP_ALLOWED=1
 
 ENTRYPOINT ["lando-cli"]
 CMD ["run"]
@@ -48,9 +49,14 @@ RUN cd / && pip install --no-cache /app
 ENV PYTHONPATH /app
 RUN chown -R app:app /app
 
+# Create repos directory for landing worker and revision worker.
 RUN mkdir /repos
 RUN chown -R app:app /repos
 
+# Create patches directory to store cached patches.
+RUN mkdir /patches
+RUN chown -R app:app /patches
+
 # Run as a non-privileged user
 USER app
 

docker-compose.yml

@@ -131,25 +131,24 @@ services:
       - smtp
   lando-api.landing-worker:
     image: lando-api
-    command: ["landing-worker"]
+    command: ["start-landing-worker"]
     environment:
-      - ENV=localdev
-      - DATABASE_URL=postgresql://postgres:[email protected]/lando_api_dev
-      - SENTRY_DSN=
-      # See http://docs.celeryproject.org/en/stable/getting-started/brokers/redis.html#configuration
-      # for the full URL format.
-      - CELERY_BROKER_URL=redis://redis.queue/0
-      - OIDC_IDENTIFIER=https://lando-api.test
-      - OIDC_DOMAIN=https://auth0.test
-      - LANDO_UI_URL=https://lando.test
-      - REPO_CLONES_PATH=/repos
-      - REPOS_TO_LAND=localdev
+      CELERY_BROKER_URL: "redis://redis.queue/0"
+      DATABASE_URL: "postgresql://postgres:[email protected]/lando_api_dev"
+      ENV: "localdev"
+      LANDO_UI_URL: "https://lando.test"
+      OIDC_DOMAIN: "https://auth0.test"
+      OIDC_IDENTIFIER: "https://lando-api.test"
+      REPOS_TO_LAND: "localdev"
+      REPO_CLONES_PATH: "/repos"
+      SENTRY_DSN: ""
     user: root
     volumes:
       - ./:/app
       - ./migrations/:/migrations/
       # Prevent writing python cache to the host.
       - caches_cache:/app/.cache/
+      - repos:/repos
     depends_on:
       - lando-api.db
       - redis.queue
@@ -177,3 +176,5 @@ volumes:
   caches_pycache:
   caches_cache:
   caches_pytest_cache:
+  repos:
+  patches:

landoapi/api/landing_jobs.py

@@ -62,8 +62,7 @@ def put(landing_job_id: str, data: dict):
         )
 
     if landing_job.status in (LandingJobStatus.SUBMITTED, LandingJobStatus.DEFERRED):
-        landing_job.transition_status(LandingJobAction.CANCEL)
-        db.session.commit()
+        landing_job.transition_status(LandingJobAction.CANCEL, commit=True, db=db)
         return {"id": landing_job.id}, 200
     else:
         raise ProblemException(

landoapi/api/revisions.py

@@ -10,6 +10,7 @@
 from landoapi.decorators import require_phabricator_api_key
 from landoapi.models import SecApprovalRequest
 from landoapi.phabricator import PhabricatorClient
+from landoapi.models.revisions import Revision
 from landoapi.projects import get_secure_project_phid
 from landoapi.revisions import revision_is_secure
 from landoapi.secapproval import send_sanitized_commit_message_for_review
@@ -88,3 +89,15 @@ def request_sec_approval(phab: PhabricatorClient, data: dict):
     db.session.commit()
 
     return {}, 200
+
+
+def get_stack_hashes(revision_id: int) -> tuple:
+    """
+    Given a revision, returns revision stack hashes.
+
+    A stack hash is used to detect a change in a revision.
+    """
+    revision = Revision.query.filter(Revision.id == revision_id).one_or_none()
+    if revision:
+        return revision.stack_hashes, 200
+    return {}, 404

landoapi/api/stacks.py

@@ -8,6 +8,7 @@
 from flask import current_app
 from landoapi.commit_message import format_commit_message
 from landoapi.decorators import require_phabricator_api_key
+from landoapi.models.revisions import Revision
 from landoapi.phabricator import PhabricatorClient
 from landoapi.projects import (
     get_release_managers,
@@ -116,19 +117,25 @@ def get(phab: PhabricatorClient, revision_id: str):
     }
 
     revisions_response = []
-    for revision_phid, revision in stack_data.revisions.items():
-        fields = PhabricatorClient.expect(revision, "fields")
+    for _phid, phab_revision in stack_data.revisions.items():
+        lando_revision = Revision.query.filter(
+            Revision.revision_id == phab_revision["id"]
+        ).one_or_none()
+        revision_phid = PhabricatorClient.expect(phab_revision, "phid")
+        fields = PhabricatorClient.expect(phab_revision, "fields")
         diff_phid = PhabricatorClient.expect(fields, "diffPHID")
         repo_phid = PhabricatorClient.expect(fields, "repositoryPHID")
         diff = stack_data.diffs[diff_phid]
-        human_revision_id = "D{}".format(PhabricatorClient.expect(revision, "id"))
+        human_revision_id = "D{}".format(PhabricatorClient.expect(phab_revision, "id"))
         revision_url = urllib.parse.urljoin(
             current_app.config["PHABRICATOR_URL"], human_revision_id
         )
-        secure = revision_is_secure(revision, secure_project_phid)
-        commit_description = find_title_and_summary_for_display(phab, revision, secure)
-        bug_id = get_bugzilla_bug(revision)
-        reviewers = get_collated_reviewers(revision)
+        secure = revision_is_secure(phab_revision, secure_project_phid)
+        commit_description = find_title_and_summary_for_display(
+            phab, phab_revision, secure
+        )
+        bug_id = get_bugzilla_bug(phab_revision)
+        reviewers = get_collated_reviewers(phab_revision)
         accepted_reviewers = reviewers_for_commit_message(
             reviewers, users, projects, sec_approval_project_phid
         )
@@ -163,16 +170,16 @@ def get(phab: PhabricatorClient, revision_id: str):
             {
                 "id": human_revision_id,
                 "phid": revision_phid,
-                "status": serialize_status(revision),
+                "status": serialize_status(phab_revision),
                 "blocked_reason": blocked.get(revision_phid, ""),
                 "bug_id": bug_id,
                 "title": commit_description.title,
                 "url": revision_url,
                 "date_created": PhabricatorClient.to_datetime(
-                    PhabricatorClient.expect(revision, "fields", "dateCreated")
+                    PhabricatorClient.expect(phab_revision, "fields", "dateCreated")
                 ).isoformat(),
                 "date_modified": PhabricatorClient.to_datetime(
-                    PhabricatorClient.expect(revision, "fields", "dateModified")
+                    PhabricatorClient.expect(phab_revision, "fields", "dateModified")
                 ).isoformat(),
                 "summary": commit_description.summary,
                 "commit_message_title": commit_message_title,
@@ -183,6 +190,9 @@ def get(phab: PhabricatorClient, revision_id: str):
                 "reviewers": serialize_reviewers(reviewers, users, projects, diff_phid),
                 "is_secure": secure,
                 "is_using_secure_commit_message": commit_description.sanitized,
+                "lando_revision": lando_revision.serialize()
+                if lando_revision
+                else None,
             }
         )