Setup PubSub framework code.

Author: azeemsgoogle

cron/config/config.go

@@ -0,0 +1,23 @@
+// Copyright 2021 Security Scorecard Authors
+//
+// Licensed under the Apache License, Vershandlern 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 permisshandlerns and
+// limitathandlerns under the License.
+
+package config
+
+const (
+	ResultDataBucketURL string = "gs://ossf-scorecard-data"
+	RequestTopicURL            = "gcppubsub://projects/openssf/topics/scorecard-batch"
+	InputReposFile             = "../projects.txt"
+	ShardNumFilename           = ".shard_num"
+	ShardSize           int    = 250
+)

cron/cron

@@ -0,0 +1,63 @@
+// Copyright 2021 Security Scorecard Authors
+//
+// Licensed under the Apache License, Vershandlern 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 permisshandlerns and
+// limitathandlerns under the License.
+
+package data
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/ossf/scorecard/cron/config"
+	"gocloud.dev/blob"
+)
+
+const (
+	filePrefixFormat = "2006.01.02/150405/"
+)
+
+var (
+	errorWrite      = errors.New("blobWriter.Write")
+	errorWriteClose = errors.New("blobWriter.Close")
+	errorOpenBucket = errors.New("blob.OpenBucket")
+)
+
+func WriteToBlobStore(ctx context.Context, bucketURL, filename string, data []byte) error {
+	bucket, err := blob.OpenBucket(ctx, bucketURL)
+	if err != nil {
+		return fmt.Errorf("%w: %v", errorOpenBucket, err)
+	}
+	defer bucket.Close()
+
+	blobWriter, err := bucket.NewWriter(ctx, filename, nil)
+	if err != nil {
+		return fmt.Errorf("%w", err)
+	}
+	if _, err = blobWriter.Write(data); err != nil {
+		return fmt.Errorf("%w: %v", errorWrite, err)
+	}
+	if err := blobWriter.Close(); err != nil {
+		return fmt.Errorf("%w: %v", errorWriteClose, err)
+	}
+	return nil
+}
+
+func GetBlobFilename(filename string, datetime time.Time) string {
+	return datetime.Format(filePrefixFormat) + filename
+}
+
+func GetShardNumFilename(datetime time.Time) string {
+	return GetBlobFilename(config.ShardNumFilename, datetime)
+}

cron/data/blob.go

@@ -0,0 +1,55 @@
+// Copyright 2021 Security Scorecard Authors
+//
+// Licensed under the Apache License, Vershandlern 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 permisshandlerns and
+// limitathandlerns under the License.
+
+package data
+
+import (
+	"testing"
+	"time"
+)
+
+const (
+	inputTimeFormat string = "2006-01-02T15:04:05"
+)
+
+func TestGetBlobFilename(t *testing.T) {
+	t.Parallel()
+	testcases := []struct {
+		name             string
+		inputTime        string
+		inputFilename    string
+		expectedFilename string
+	}{
+		{
+			name:             "Basic",
+			inputTime:        "2021-04-23T15:06:43",
+			inputFilename:    "file-000",
+			expectedFilename: "2021.04.23/150643/file-000",
+		},
+	}
+	for _, testcase := range testcases {
+		testcase := testcase
+		t.Run(testcase.name, func(t *testing.T) {
+			t.Parallel()
+			datetime, err := time.Parse(inputTimeFormat, testcase.inputTime)
+			if err != nil {
+				t.Errorf("Failed to parse testcase.inputTime: %v", err)
+			}
+			gotFilename := GetBlobFilename(testcase.inputFilename, datetime)
+			if gotFilename != testcase.expectedFilename {
+				t.Errorf("Test failed. Expected: %s, Got: %s", testcase.expectedFilename, gotFilename)
+			}
+		})
+	}
+}

cron/data/blob_test.go

@@ -0,0 +1,92 @@
+// Copyright 2021 Security Scorecard Authors
+//
+// Licensed under the Apache License, Vershandlern 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 permisshandlerns and
+// limitathandlerns under the License.
+
+package data
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log"
+	"sync"
+	"sync/atomic"
+
+	"gocloud.dev/pubsub"
+	_ "gocloud.dev/pubsub/gcppubsub" // Needed to link in GCP drivers.
+	"google.golang.org/protobuf/encoding/protojson"
+)
+
+var (
+	errorOpenTopic        = errors.New("pubsub.OpenTopic")
+	errorProtoJSONMarshal = errors.New("protojson.Marshal")
+	errorPublish          = errors.New("total errors when publishing")
+)
+
+type Publisher interface {
+	Publish(request *ScorecardBatchRequest) error
+	Close() error
+}
+
+func CreatePublisher(ctx context.Context, topicURL string) (Publisher, error) {
+	ret := publisherImpl{}
+	topic, err := pubsub.OpenTopic(ctx, topicURL)
+	if err != nil {
+		return &ret, fmt.Errorf("%w: %v", errorOpenTopic, err)
+	}
+	return &publisherImpl{
+		ctx:   ctx,
+		topic: topic,
+	}, nil
+}
+
+type topicInterface interface {
+	Send(ctx context.Context, msg *pubsub.Message) error
+}
+
+type publisherImpl struct {
+	ctx         context.Context
+	topic       topicInterface
+	wg          sync.WaitGroup
+	totalErrors uint64
+}
+
+func (publisher *publisherImpl) Publish(request *ScorecardBatchRequest) error {
+	msg, err := protojson.Marshal(request)
+	if err != nil {
+		return fmt.Errorf("%w: %v", errorProtoJSONMarshal, err)
+	}
+
+	publisher.wg.Add(1)
+	go func() {
+		defer publisher.wg.Done()
+		err := publisher.topic.Send(publisher.ctx, &pubsub.Message{
+			Body: msg,
+		})
+		if err != nil {
+			log.Printf("Error when publishing message %s: %v", msg, err)
+			atomic.AddUint64(&publisher.totalErrors, 1)
+			return
+		}
+		log.Print("Successfully published message")
+	}()
+	return nil
+}
+
+func (publisher *publisherImpl) Close() error {
+	publisher.wg.Wait()
+	if publisher.totalErrors > 0 {
+		return fmt.Errorf("%w: %d", errorPublish, publisher.totalErrors)
+	}
+	return nil
+}

cron/data/publisher.go

@@ -0,0 +1,85 @@
+// Copyright 2021 Security Scorecard Authors
+//
+// Licensed under the Apache License, Vershandlern 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 permisshandlerns and
+// limitathandlerns under the License.
+
+package data
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	"gocloud.dev/pubsub"
+)
+
+type mockSucceedTopic struct{}
+
+func (topic *mockSucceedTopic) Send(ctx context.Context, msg *pubsub.Message) error {
+	return nil
+}
+
+type mockFailTopic struct{}
+
+func (topic *mockFailTopic) Send(ctx context.Context, msg *pubsub.Message) error {
+	// nolint: goerr113
+	return fmt.Errorf("mockFailTopic failed to send")
+}
+
+func TestPublish(t *testing.T) {
+	t.Parallel()
+	// nolint: govet
+	testcases := []struct {
+		numErrors uint64
+		name      string
+		errorMsg  string
+		hasError  bool
+		topic     topicInterface
+	}{
+		{
+			name:      "SendFails",
+			topic:     &mockFailTopic{},
+			hasError:  true,
+			numErrors: 1,
+			errorMsg:  "",
+		},
+		{
+			name:     "SendSucceeds",
+			topic:    &mockSucceedTopic{},
+			hasError: false,
+		},
+	}
+	for _, testcase := range testcases {
+		testcase := testcase
+		t.Run(testcase.name, func(t *testing.T) {
+			t.Parallel()
+			ctx := context.Background()
+			publisher := publisherImpl{
+				ctx:   ctx,
+				topic: testcase.topic,
+			}
+			request := ScorecardBatchRequest{
+				Repos: []string{"repo1"},
+			}
+			if err := publisher.Publish(&request); err != nil {
+				t.Errorf("Failed to parse message: %v", err)
+			}
+			err := publisher.Close()
+			if (err == nil) == testcase.hasError {
+				t.Errorf("Test failed. Expected: %t got: %v", testcase.hasError, err)
+			}
+			if testcase.hasError && testcase.numErrors != publisher.totalErrors {
+				t.Errorf("Test failed. Expected numErrors: %d, got: %d", testcase.numErrors, publisher.totalErrors)
+			}
+		})
+	}
+}

cron/data/publisher_test.go

@@ -0,0 +1,69 @@
+// Copyright 2021 Security Scorecard Authors
+//
+// Licensed under the Apache License, Vershandlern 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 permisshandlerns and
+// limitathandlerns under the License.
+
+package data
+
+import (
+	"encoding/csv"
+	"errors"
+	"fmt"
+	"io"
+
+	"github.com/jszwec/csvutil"
+	"github.com/ossf/scorecard/repos"
+)
+
+var errorCSVDecoder = errors.New("could not create CSV decoder")
+
+type Reader interface {
+	HasNext() bool
+	Next() (repos.RepoURL, error)
+}
+
+func MakeReader(reader io.Reader) (Reader, error) {
+	dec, err := csvutil.NewDecoder(csv.NewReader(reader))
+	if err != nil {
+		return &csvReader{}, fmt.Errorf("%w: %v", errorCSVDecoder, err)
+	}
+	return &csvReader{decoder: dec}, nil
+}
+
+type inputRepo struct {
+	Repo     string `csv:"repo"`
+	Metadata string `csv:"metadata"`
+}
+
+type csvReader struct {
+	decoder *csvutil.Decoder
+	err     error
+	next    inputRepo
+}
+
+func (reader *csvReader) HasNext() bool {
+	reader.err = reader.decoder.Decode(&reader.next)
+	return reader.err != io.EOF
+}
+
+func (reader *csvReader) Next() (repos.RepoURL, error) {
+	if reader.err != nil {
+		return repos.RepoURL{}, reader.err
+	}
+	ret := repos.RepoURL{}
+	var err error
+	err = ret.Set(reader.next.Repo)
+	if err == nil {
+		err = ret.ValidGitHubURL()
+	}
+	return ret, errors.Unwrap(err)
+}