Merge pull request #24841 from chaodaiG/crier-gerrit-multiple-reporters

Crier gerrit reporter: worker can be more than 1

Author: k8s-ci-robot

prow/cmd/crier/main.go

@@ -78,14 +78,6 @@ type options struct {
 }
 
 func (o *options) validate() error {
-
-	// TODO(krzyzacy): gerrit && github report are actually stateful..
-	// Need a better design to re-enable parallel reporting
-	if o.gerritWorkers > 1 {
-		logrus.Warn("gerrit reporter only supports one worker")
-		o.gerritWorkers = 1
-	}
-
 	if o.gerritWorkers+o.pubsubWorkers+o.githubWorkers+o.slackWorkers+o.gcsWorkers+o.k8sGCSWorkers+o.blobStorageWorkers+o.k8sBlobStorageWorkers <= 0 {
 		return errors.New("crier need to have at least one report worker to start")
 	}
@@ -253,7 +245,7 @@ func main() {
 	}
 
 	if o.gerritWorkers > 0 {
-		gerritReporter, err := gerritreporter.NewReporter(cfg, o.cookiefilePath, o.gerritProjects, mgr.GetCache())
+		gerritReporter, err := gerritreporter.NewReporter(cfg, o.cookiefilePath, o.gerritProjects, mgr.GetClient())
 		if err != nil {
 			logrus.WithError(err).Fatal("Error starting gerrit reporter")
 		}

prow/cmd/crier/main_test.go

@@ -51,10 +51,10 @@ func TestOptions(t *testing.T) {
 		},
 		//Gerrit Reporter
 		{
-			name: "gerrit only support one worker",
+			name: "gerrit supports multiple workers",
 			args: []string{"--gerrit-workers=99", "--gerrit-projects=foo=bar", "--cookiefile=foobar", "--config-path=foo"},
 			expected: &options{
-				gerritWorkers:  1,
+				gerritWorkers:  99,
 				cookiefilePath: "foobar",
 				gerritProjects: map[string][]string{
 					"foo": {"bar"},
@@ -74,7 +74,7 @@ func TestOptions(t *testing.T) {
 			name: "gerrit missing --cookiefile",
 			args: []string{"--gerrit-workers=5", "--gerrit-projects=foo=bar", "--config-path=foo"},
 			expected: &options{
-				gerritWorkers: 1,
+				gerritWorkers: 5,
 				gerritProjects: map[string][]string{
 					"foo": {"bar"},
 				},

prow/crier/BUILD.bazel

@@ -10,12 +10,10 @@ go_library(
     visibility = ["//visibility:public"],
     deps = [
         "//prow/apis/prowjobs/v1:go_default_library",
+        "//prow/crier/reporters/criercommonlib:go_default_library",
         "@com_github_prometheus_client_golang//prometheus:go_default_library",
         "@com_github_sirupsen_logrus//:go_default_library",
         "@io_k8s_apimachinery//pkg/api/errors:go_default_library",
-        "@io_k8s_apimachinery//pkg/types:go_default_library",
-        "@io_k8s_apimachinery//pkg/util/wait:go_default_library",
-        "@io_k8s_client_go//util/retry:go_default_library",
         "@io_k8s_sigs_controller_runtime//pkg/builder:go_default_library",
         "@io_k8s_sigs_controller_runtime//pkg/client:go_default_library",
         "@io_k8s_sigs_controller_runtime//pkg/controller:go_default_library",
@@ -35,6 +33,7 @@ filegroup(
     name = "all-srcs",
     srcs = [
         ":package-srcs",
+        "//prow/crier/reporters/criercommonlib:all-srcs",
         "//prow/crier/reporters/gcs:all-srcs",
         "//prow/crier/reporters/gerrit:all-srcs",
         "//prow/crier/reporters/github:all-srcs",

prow/crier/controller.go

@@ -24,16 +24,14 @@ import (
 
 	"github.com/sirupsen/logrus"
 	"k8s.io/apimachinery/pkg/api/errors"
-	"k8s.io/apimachinery/pkg/types"
-	"k8s.io/apimachinery/pkg/util/wait"
-	"k8s.io/client-go/util/retry"
 	"sigs.k8s.io/controller-runtime/pkg/builder"
 	ctrlruntimeclient "sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 
 	prowv1 "k8s.io/test-infra/prow/apis/prowjobs/v1"
+	"k8s.io/test-infra/prow/crier/reporters/criercommonlib"
 )
 
 type ReportClient interface {
@@ -80,66 +78,6 @@ func New(
 	return nil
 }
 
-func (r *reconciler) updateReportState(ctx context.Context, pj *prowv1.ProwJob, log *logrus.Entry, reportedState prowv1.ProwJobState) error {
-	// update pj report status
-	newpj := pj.DeepCopy()
-	// we set omitempty on PrevReportStates, so here we need to init it if is nil
-	if newpj.Status.PrevReportStates == nil {
-		newpj.Status.PrevReportStates = map[string]prowv1.ProwJobState{}
-	}
-	newpj.Status.PrevReportStates[r.reporter.GetName()] = reportedState
-
-	if err := r.pjclientset.Patch(ctx, newpj, ctrlruntimeclient.MergeFrom(pj)); err != nil {
-		return fmt.Errorf("failed to patch: %w", err)
-	}
-
-	// Block until the update is in the lister to make sure that events from another controller
-	// that also does reporting dont trigger another report because our lister doesn't yet contain
-	// the updated Status
-	name := types.NamespacedName{Namespace: pj.Namespace, Name: pj.Name}
-	if err := wait.Poll(100*time.Millisecond, 10*time.Second, func() (bool, error) {
-		if err := r.pjclientset.Get(ctx, name, pj); err != nil {
-			return false, err
-		}
-		if pj.Status.PrevReportStates != nil &&
-			pj.Status.PrevReportStates[r.reporter.GetName()] == reportedState {
-			return true, nil
-		}
-		return false, nil
-	}); err != nil {
-		return fmt.Errorf("failed to wait for updated report status to be in lister: %w", err)
-	}
-	return nil
-}
-
-func (r *reconciler) updateReportStateWithRetries(ctx context.Context, pj *prowv1.ProwJob, log *logrus.Entry) error {
-	reportState := pj.Status.State
-	log = log.WithFields(logrus.Fields{
-		"prowjob":   pj.Name,
-		"jobName":   pj.Spec.Job,
-		"jobStatus": reportState,
-	})
-	// We have to retry here, if we return we lose the information that we already reported this job.
-	if err := retry.RetryOnConflict(retry.DefaultBackoff, func() error {
-		// Get it first, this is very cheap
-		name := types.NamespacedName{Namespace: pj.Namespace, Name: pj.Name}
-		if err := r.pjclientset.Get(ctx, name, pj); err != nil {
-			return err
-		}
-		// Must not wrap until we have kube 1.19, otherwise the RetryOnConflict won't recognize conflicts
-		// correctly
-		return r.updateReportState(ctx, pj, log, reportState)
-	}); err != nil {
-		// Very subpar, we will report again. But even if we didn't do that now, we would do so
-		// latest when crier gets restarted. In an ideal world, all reporters are idempotent and
-		// reporting has no cost.
-		return fmt.Errorf("failed to update report state on prowjob: %w", err)
-	}
-
-	log.Info("Successfully updated report state on prowjob")
-	return nil
-}
-
 // Reconcile retrieves each queued item and takes the necessary handler action based off of if
 // the item was created or deleted.
 func (r *reconciler) Reconcile(ctx context.Context, req reconcile.Request) (reconcile.Result, error) {
@@ -207,7 +145,7 @@ func (r *reconciler) reconcile(ctx context.Context, log *logrus.Entry, req recon
 	log.WithField("job-count", len(pjs)).Info("Reported job(s), now will update pj(s).")
 	var lastErr error
 	for _, pjob := range pjs {
-		if err := r.updateReportStateWithRetries(ctx, pjob, log); err != nil {
+		if err := criercommonlib.UpdateReportStateWithRetries(ctx, pjob, log, r.pjclientset, r.reporter.GetName()); err != nil {
 			log.WithError(err).Error("Failed to update report state on prowjob")
 			// The error above is alreay logged, so it would be duplicated
 			// effort to combine all errors to return, only capture the last

prow/crier/reporters/criercommonlib/BUILD.bazel

@@ -0,0 +1,42 @@
+load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "go_default_library",
+    srcs = [
+        "shardedlock.go",
+        "updatereportstatus.go",
+    ],
+    importpath = "k8s.io/test-infra/prow/crier/reporters/criercommonlib",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//prow/apis/prowjobs/v1:go_default_library",
+        "@com_github_sirupsen_logrus//:go_default_library",
+        "@io_k8s_apimachinery//pkg/types:go_default_library",
+        "@io_k8s_apimachinery//pkg/util/wait:go_default_library",
+        "@io_k8s_client_go//util/retry:go_default_library",
+        "@io_k8s_sigs_controller_runtime//pkg/client:go_default_library",
+        "@org_golang_x_sync//semaphore:go_default_library",
+    ],
+)
+
+filegroup(
+    name = "package-srcs",
+    srcs = glob(["**"]),
+    tags = ["automanaged"],
+    visibility = ["//visibility:private"],
+)
+
+filegroup(
+    name = "all-srcs",
+    srcs = [":package-srcs"],
+    tags = ["automanaged"],
+    visibility = ["//visibility:public"],
+)
+
+go_test(
+    name = "go_default_test",
+    srcs = ["shardedlock_test.go"],
+    embed = [":go_default_library"],
+    tags = ["manual"],
+    deps = ["@org_golang_x_sync//semaphore:go_default_library"],
+)

prow/crier/reporters/criercommonlib/shardedlock.go

@@ -0,0 +1,102 @@
+/*
+Copyright 2022 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Package criercommonlib contains shared lib used by reporters
+package criercommonlib
+
+import (
+	"context"
+	"time"
+
+	"github.com/sirupsen/logrus"
+	"golang.org/x/sync/semaphore"
+)
+
+// SimplePull contains info for identifying a shard
+type SimplePull struct {
+	org, repo string
+	number    int
+}
+
+// NewSimplePull creates SimplePull
+func NewSimplePull(org, repo string, number int) *SimplePull {
+	return &SimplePull{org: org, repo: repo, number: number}
+}
+
+// ShardedLock contains sharding information based on PRs
+type ShardedLock struct {
+	// semaphore is chosed over mutex, as Acquire from semaphore respects
+	// context timeout while mutex doesn't
+	mapLock *semaphore.Weighted
+	locks   map[SimplePull]*semaphore.Weighted
+}
+
+// NewShardedLock creates ShardedLock
+func NewShardedLock() *ShardedLock {
+	return &ShardedLock{
+		mapLock: semaphore.NewWeighted(1),
+		locks:   map[SimplePull]*semaphore.Weighted{},
+	}
+}
+
+// GetLock aquires the lock for a PR
+func (s *ShardedLock) GetLock(ctx context.Context, key SimplePull) (*semaphore.Weighted, error) {
+	if err := s.mapLock.Acquire(ctx, 1); err != nil {
+		return nil, err
+	}
+	defer s.mapLock.Release(1)
+	if _, exists := s.locks[key]; !exists {
+		s.locks[key] = semaphore.NewWeighted(1)
+	}
+	return s.locks[key], nil
+}
+
+// Cleanup deletes all locks by acquiring first
+// the mapLock and then each individual lock before
+// deleting it. The individual lock must be acquired
+// because otherwise it may be held, we delete it from
+// the map, it gets recreated and acquired and two
+// routines report in parallel for the same job.
+// Note that while this function is running, no new
+// presubmit reporting can happen, as we hold the mapLock.
+func (s *ShardedLock) Cleanup() {
+	ctx := context.Background()
+	s.mapLock.Acquire(ctx, 1)
+	defer s.mapLock.Release(1)
+
+	for key, lock := range s.locks {
+		// There is a very low chance of race condition, that two threads got
+		// different locks from the same PR, which would end up with duplicated
+		// report once. Since this is very complicated to fix and the impact is
+		// really low, would just keep it as is.
+		// For details see: https://github.com/kubernetes/test-infra/pull/20343
+		lock.Acquire(ctx, 1)
+		delete(s.locks, key)
+		lock.Release(1)
+	}
+}
+
+// RunCleanup asynchronously runs the cleanup once per hour.
+func (s *ShardedLock) RunCleanup() {
+	go func() {
+		for range time.Tick(time.Hour) {
+			logrus.Debug("Starting to clean up presubmit locks")
+			startTime := time.Now()
+			s.Cleanup()
+			logrus.WithField("duration", time.Since(startTime).String()).Debug("Finished cleaning up presubmit locks")
+		}
+	}()
+}

prow/crier/reporters/criercommonlib/shardedlock_test.go

@@ -0,0 +1,36 @@
+/*
+Copyright 2022 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Package criercommonlib contains shared lib used by reporters
+package criercommonlib
+
+import (
+	"testing"
+
+	"golang.org/x/sync/semaphore"
+)
+
+func TestShardedLockCleanup(t *testing.T) {
+	t.Parallel()
+	sl := &ShardedLock{mapLock: semaphore.NewWeighted(1), locks: map[SimplePull]*semaphore.Weighted{}}
+	key := SimplePull{"org", "repo", 1}
+	sl.locks[key] = semaphore.NewWeighted(1)
+	sl.Cleanup()
+	if _, exists := sl.locks[key]; exists {
+		t.Error("lock didn't get cleaned up")
+	}
+
+}