Fix issues with PubSub worker (#553)

Co-authored-by: Azeem Shaikh <[email protected]>

cron/data/blob.go

@@ -33,6 +33,20 @@ const (
 	filePrefixFormat = "2006.01.02/150405/"
 )
 
+func BlobExists(ctx context.Context, bucketURL, key string) (bool, error) {
+	bucket, err := blob.OpenBucket(ctx, bucketURL)
+	if err != nil {
+		return false, fmt.Errorf("error from blob.OpenBucket: %w", err)
+	}
+	defer bucket.Close()
+
+	ret, err := bucket.Exists(ctx, key)
+	if err != nil {
+		return ret, fmt.Errorf("error during bucket.Exists: %w", err)
+	}
+	return ret, nil
+}
+
 func WriteToBlobStore(ctx context.Context, bucketURL, filename string, data []byte) error {
 	bucket, err := blob.OpenBucket(ctx, bucketURL)
 	if err != nil {

cron/k8s/worker.yaml

@@ -42,6 +42,8 @@ spec:
           value: "gs://ossf-scorecards-cache"
         resources:
           requests:
-            memory: 1Gi
+            cpu: 5
+            memory: 5Gi
           limits:
-            memory: 1.5Gi
+            cpu: 12
+            memory: 12Gi

cron/worker/main.go

@@ -42,6 +42,20 @@ import (
 func processRequest(ctx context.Context,
 	batchRequest *data.ScorecardBatchRequest, checksToRun checker.CheckNameToFnMap, bucketURL string,
 	httpClient *http.Client, githubClient *github.Client, graphClient *githubv4.Client) error {
+	filename := data.GetBlobFilename(
+		fmt.Sprintf("shard-%05d", batchRequest.GetShardNum()),
+		batchRequest.GetJobTime().AsTime())
+	// Sanity check - make sure we are not re-processing an already processed request.
+	exists, err := data.BlobExists(ctx, bucketURL, filename)
+	if err != nil {
+		return fmt.Errorf("error during BlobExists: %w", err)
+	}
+	if exists {
+		log.Printf("Already processed shard %s. Nothing to do.", filename)
+		// We have already processed this request, nothing to do.
+		return nil
+	}
+
 	repoURLs := make([]repos.RepoURL, 0, len(batchRequest.GetRepos()))
 	for _, repo := range batchRequest.GetRepos() {
 		repoURL := repos.RepoURL{}
@@ -66,9 +80,6 @@ func processRequest(ctx context.Context,
 		}
 	}
 
-	filename := data.GetBlobFilename(
-		fmt.Sprintf("shard-%05d", batchRequest.GetShardNum()),
-		batchRequest.GetJobTime().AsTime())
 	if err := data.WriteToBlobStore(ctx, bucketURL, filename, buffer.Bytes()); err != nil {
 		return fmt.Errorf("error during WriteToBlobStore: %w", err)
 	}