Merge branch 'master' into patch-1

Author: GameRoMan

configs/test/k8s/schedule-fuzz.yaml

@@ -18,10 +18,10 @@ metadata:
   name: schedule-fuzz
 spec:
   schedule: "*/10 * * * *"
-  concurrencyPolicy: Forbid
+  concurrencyPolicy: Allow
   jobTemplate:
     spec:
-      activeDeadlineSeconds: 900  # 15 minutes.
+      activeDeadlineSeconds: 1800  # 30 minutes.
       template:
         spec:
           containers:

configs/test/project.yaml

@@ -57,9 +57,8 @@ firebase:
   - google.com
   # - github.com
 
-uworker_tasks:
-  - analyze
-
+full_utask_model:
+  enabled: false
 
 stacktrace:
   # Stack frames to ignore when determining the crash signature.

infra/terraform/main.tf

@@ -132,3 +132,25 @@ resource "google_compute_router_nat" "nat_config" {
     filter = "ALL"
   }
 }
+
+resource "google_pubsub_topic" "crash_replication" {
+  name    = "crash-replication"
+  project = var.project_id
+}
+
+resource "google_pubsub_subscription" "crash_replication_push_to_appengine" {
+  ack_deadline_seconds = 120
+
+  name                       = "crash-replication-push-to-appengine"
+  project                    =  var.project_id
+
+  push_config {
+    oidc_token {
+      service_account_email = var.appengine_service_account
+    }
+
+    push_endpoint = var.testcase_replication_push_endpoint
+  }
+
+  topic = google_pubsub_topic.crash_replication.name
+}

infra/terraform/monitoring.tf

@@ -1626,4 +1626,4 @@ resource "google_monitoring_dashboard" "clusterfuzz_sli_dashboard" {
   "labels": {}
 }
 JSON
-}
+}

infra/terraform/variables.tf

@@ -54,3 +54,13 @@ variable "network_description" {
   default = ""
   description = "The network description"
 }
+
+variable "appengine_service_account" {
+  default = ""
+  description = "AppEngine service account for the clusterfuzz deployment"
+}
+
+variable "testcase_replication_push_endpoint" {
+  default = ""
+  description = "Endpoint to push testcase replication messages to"
+}

src/appengine/handlers/testcase_detail/create_issue.py

@@ -16,6 +16,7 @@
 from flask import request
 
 from clusterfuzz._internal.issue_management import issue_filer
+from clusterfuzz._internal.metrics import events
 from handlers import base_handler
 from handlers.testcase_detail import show
 from libs import handler
@@ -46,6 +47,13 @@ def create_issue(testcase, severity, cc_me):
         user_email=user_email,
         additional_ccs=additional_ccs)
 
+    events.emit(
+        events.IssueFilingEvent(
+            testcase=testcase,
+            issue_tracker_project=issue_tracker.project,
+            issue_id=str(issue_id) if issue_id else None,
+            issue_created=bool(issue_id)))
+
     if not issue_id:
       raise helpers.EarlyExitError('Unable to create new issue.', 500)
 

src/appengine/handlers/upload_testcase.py

@@ -35,6 +35,9 @@
 from clusterfuzz._internal.google_cloud_utils import blobs
 from clusterfuzz._internal.google_cloud_utils import storage
 from clusterfuzz._internal.issue_management import issue_tracker_utils
+from clusterfuzz._internal.metrics import events
+from clusterfuzz._internal.metrics import monitor
+from clusterfuzz._internal.metrics import monitoring_metrics
 from clusterfuzz._internal.system import archive
 from clusterfuzz._internal.system import environment
 from handlers import base_handler
@@ -303,6 +306,8 @@ def get_upload(self):
 
   def do_post(self):
     """Upload a testcase."""
+    # Set artifical task id env to be used by tracing.
+    environment.set_task_id_vars(task_name='upload_testcase')
     email = helpers.get_user_email()
     testcase_id = request.get('testcaseId')
     uploaded_file = self.get_upload()
@@ -393,8 +398,8 @@ def do_post(self):
         'trustedAgreement') == TRUSTED_AGREEMENT_TEXT.strip()
 
     # Chrome is the only ClusterFuzz deployment where there are trusted bots
-    # running utasks.
-    # This check also fails on oss-fuzz because of the way it abuses platform.
+    # running utasks. This check also fails on oss-fuzz because of the way it
+    # abuses platform.
     if (not trusted_agreement_signed and utils.is_chromium() and
         task_utils.is_remotely_executing_utasks() and
         ((platform_id and platform_id != 'Linux') or
@@ -606,8 +611,14 @@ def do_post(self):
         additional_metadata=testcase_metadata,
         crash_data=crash_data)
 
+    testcase = data_handler.get_testcase_by_id(testcase_id)
+    events.emit(
+        events.TestcaseCreationEvent(
+            testcase=testcase,
+            creation_origin=events.TestcaseOrigin.MANUAL_UPLOAD,
+            uploader=email))
+
     if not quiet_flag:
-      testcase = data_handler.get_testcase_by_id(testcase_id)
       issue = issue_tracker_utils.get_issue_for_testcase(testcase)
       if issue:
         report_url = data_handler.TESTCASE_REPORT_URL.format(
@@ -668,3 +679,14 @@ def get_upload(self):
   @handler.oauth
   def post(self, *args):
     return self.do_post()
+
+
+class CrashReplicationUploadHandler(base_handler.Handler):
+  """Handler that picks up the pubsub notification."""
+
+  @handler.pubsub_push
+  def post(self, message):
+    helpers.log(message.data.decode(), helpers.VIEW_OPERATION)
+    with monitor.wrap_with_monitoring():
+      monitoring_metrics.UPLOAD_TESTCASE_COUNT.increment()
+    return 'ok'

src/appengine/server.py

@@ -194,6 +194,8 @@ def register_routes(flask_app, routes):
     ('/upload-testcase/get-url-oauth', upload_testcase.UploadUrlHandlerOAuth),
     ('/upload-testcase/prepare', upload_testcase.PrepareUploadHandler),
     ('/upload-testcase/load', upload_testcase.JsonHandler),
+    ('/upload-testcase/crash-replication',
+     upload_testcase.CrashReplicationUploadHandler),
     ('/upload-testcase/upload', upload_testcase.UploadHandler),
     ('/upload-testcase/upload-oauth', upload_testcase.UploadHandlerOAuth),
     ('/update-job', jobs.UpdateJob),

src/clusterfuzz/_internal/base/concurrency.py

@@ -21,30 +21,16 @@
 POOL_SIZE = multiprocessing.cpu_count()
 
 
-class SingleThreadPool:
-  """Single thread pool for when it's not worth using Python's thread
-  implementation."""
-
-  def __init__(self, size):
-    del size
-
-  def map(self, f, l):
-    return list(map(f, l))
-
-
 @contextlib.contextmanager
-def make_pool(pool_size=POOL_SIZE, cpu_bound=False, max_pool_size=None):
+def make_pool(pool_size=POOL_SIZE, max_pool_size=None):
   """Returns a pool that can (usually) execute tasks concurrently."""
   if max_pool_size is not None:
     pool_size = min(pool_size, max_pool_size)
 
   # Don't use processes on Windows and unittests to avoid hangs.
   if (environment.get_value('PY_UNITTESTS') or
       environment.platform() == 'WINDOWS'):
-    if cpu_bound:
-      yield SingleThreadPool(pool_size)
-    else:
-      yield futures.ThreadPoolExecutor(pool_size)
+    yield futures.ThreadPoolExecutor(pool_size)
   else:
     yield futures.ProcessPoolExecutor(pool_size)
 

src/clusterfuzz/_internal/base/errors.py

@@ -27,6 +27,7 @@
     'interrupted function call',
     'out of memory',
     'systemexit:',
+    'Expired token, failed to download uworker_input',
 ]