Merge branch 'main' into grady/workinguploads

Author: gbdubs

azure/azblob/azblob.go

@@ -101,16 +101,6 @@ func (c *Client) DeleteBlob(ctx context.Context, uri string) error {
 	return nil
 }
 
-func (c *Client) DeleteBlobs(ctx context.Context, uris []string) error {
-	// TODO: Implement parallel delete if slow.
-	for _, uri := range uris {
-		if err := c.DeleteBlob(ctx, uri); err != nil {
-			return err
-		}
-	}
-	return nil
-}
-
 // SignedUploadURL returns a URL that is allowed to upload to the given URI.
 // See https://pkg.go.dev/github.com/Azure/azure-sdk-for-go/sdk/storage/azblob@v1.0.0/sas#example-package-UserDelegationSAS
 func (c *Client) SignedUploadURL(ctx context.Context, uri string) (string, error) {

cmd/runner/BUILD.bazel

@@ -25,12 +25,6 @@ go_library(
     ],
 )
 
-go_binary(
-    name = "runner",
-    embed = [":runner_lib"],
-    visibility = ["//visibility:public"],
-)
-
 pkg_tar(
     name = "runner_tar",
     srcs = [":runner"],
@@ -72,3 +66,9 @@ oci_tarball(
     image = ":image",
     repo_tags = [],
 )
+
+go_binary(
+    name = "runner",
+    embed = [":runner_lib"],
+    visibility = ["//visibility:public"],
+)

cmd/runner/README.md

@@ -12,3 +12,14 @@ If you do want to actually run the full `runner` image on Azure, you can use:
 # Run the backend, tell it to create tasks as real Azure Container Apps Jobs.
 bazel run //scripts:run_apiserver -- --use_azure_runner
 ```
+
+### Creating a new docker image to run locally
+
+```bash
+# Build the runner binary
+bazel build --@io_bazel_rules_go//go/config:pure //cmd/runner:image_tarball
+# Load the new image into docker, which will output a SHA256 value
+docker load < bazel-bin/cmd/runner/image_tarball/tarball.tar
+# Tag the runner image in order for it to be picked up locally. Don't push this to the registry!
+docker tag <SHA from previous step> rmisa.azurecr.io/runner
+```

cmd/runner/main.go

@@ -256,7 +256,6 @@ func (h *handler) parsePortfolio(ctx context.Context, taskID task.ID, req *task.
 		id := uuid.New().String()
 		// TODO: Probably set the CSV extension in the signed upload URL instead.
 		destPath := filepath.Join("/", "mnt", "raw_portfolios", fmt.Sprintf("%s.csv", id))
-
 		if err := h.downloadBlob(ctx, string(srcURI), destPath); err != nil {
 			return fmt.Errorf("failed to download raw portfolio blob: %w", err)
 		}
@@ -292,7 +291,7 @@ func (h *handler) parsePortfolio(ctx context.Context, taskID task.ID, req *task.
 	// NOTE: This code could benefit from some concurrency, but I'm opting not to prematurely optimize.
 	var out []*task.ParsePortfolioResponseItem
 	for _, p := range paths {
-		lineCount, err := countLines(p)
+		lineCount, err := countCSVLines(p)
 		if err != nil {
 			return fmt.Errorf("failed to count lines in file %q: %w", p, err)
 		}
@@ -342,7 +341,7 @@ func (h *handler) parsePortfolio(ctx context.Context, taskID task.ID, req *task.
 	return nil
 }
 
-func countLines(path string) (int, error) {
+func countCSVLines(path string) (int, error) {
 	file, err := os.Open(path)
 	if err != nil {
 		return 0, fmt.Errorf("opening file failed: %w", err)
@@ -356,7 +355,7 @@ func countLines(path string) (int, error) {
 	if err := scanner.Err(); err != nil {
 		return 0, fmt.Errorf("scanner.error returned: %w", err)
 	}
-	return lineCount, nil
+	return lineCount - 1, nil // Subtract 1 for the header row
 }
 
 func createReportReq() (*task.CreateReportRequest, error) {

cmd/runner/taskrunner/taskrunner.go

@@ -83,19 +83,22 @@ func New(cfg *Config) (*TaskRunner, error) {
 }
 
 func (tr *TaskRunner) ParsePortfolio(ctx context.Context, req *task.ParsePortfolioRequest) (task.ID, task.RunnerID, error) {
-	var taskBuffer bytes.Buffer
-	if err := json.NewEncoder(&taskBuffer).Encode(req); err != nil {
+	var buf bytes.Buffer
+	if err := json.NewEncoder(&buf).Encode(req); err != nil {
 		return "", "", fmt.Errorf("failed to encode ParsePortfolioRequest: %w", err)
 	}
-	tr.logger.Info("triggering parse portfolio task", zap.Any("req", req))
+	value := buf.String()
+	if len(value) > 128*1024 {
+		return "", "", fmt.Errorf("ParsePortfolioRequest is too large: %d bytes > 128 kb", len(value))
+	}
 	return tr.run(ctx, []task.EnvVar{
 		{
 			Key:   "TASK_TYPE",
 			Value: string(task.ParsePortfolio),
 		},
 		{
 			Key:   "PARSE_PORTFOLIO_REQUEST",
-			Value: taskBuffer.String(),
+			Value: value,
 		},
 	})
 }

cmd/server/pactasrv/BUILD.bazel

@@ -9,6 +9,7 @@ go_library(
         "initiative_user_relationship.go",
         "pacta_version.go",
         "pactasrv.go",
+        "parallel.go",
         "portfolio.go",
         "upload.go",
         "user.go",

cmd/server/pactasrv/incomplete_upload.go

@@ -43,7 +43,7 @@ func (s *Server) DeleteIncompleteUpload(ctx context.Context, request api.DeleteI
 	if err != nil {
 		return nil, oapierr.Internal("failed to delete incomplete upload", zap.Error(err))
 	}
-	if err := s.Blob.DeleteBlobs(ctx, []string{string(blobURI)}); err != nil {
+	if err := s.deleteBlobs(ctx, []string{string(blobURI)}); err != nil {
 		return nil, oapierr.Internal("failed to delete blob", zap.Error(err))
 	}
 	return api.DeleteIncompleteUpload204Response{}, nil

cmd/server/pactasrv/pactasrv.go

@@ -93,11 +93,9 @@ type DB interface {
 type Blob interface {
 	Scheme() blob.Scheme
 
-	// For uploading portfolios
 	SignedUploadURL(ctx context.Context, uri string) (string, error)
-	// For downloading reports
 	SignedDownloadURL(ctx context.Context, uri string) (string, error)
-	DeleteBlobs(ctx context.Context, uris []string) error
+	DeleteBlob(ctx context.Context, uri string) error
 }
 
 type Server struct {

cmd/server/pactasrv/parallel.go

@@ -0,0 +1,24 @@
+package pactasrv
+
+import (
+	"context"
+	"fmt"
+)
+
+type blobDeleter interface {
+	DeleteBlob(ctx context.Context, uri string) error
+}
+
+func deleteBlobs(ctx context.Context, bd blobDeleter, uris []string) error {
+	// Implement parallel delete if slow - not prematurely optimizing.
+	for i, uri := range uris {
+		if err := bd.DeleteBlob(ctx, uri); err != nil {
+			return fmt.Errorf("deleting blob %d/%d: %w", i, len(uris), err)
+		}
+	}
+	return nil
+}
+
+func (s *Server) deleteBlobs(ctx context.Context, uris []string) error {
+	return deleteBlobs(ctx, s.Blob, uris)
+}

cmd/server/pactasrv/portfolio.go

@@ -42,7 +42,7 @@ func (s *Server) DeletePortfolio(ctx context.Context, request api.DeletePortfoli
 	if err != nil {
 		return nil, oapierr.Internal("failed to delete portfolio", zap.Error(err))
 	}
-	if err := s.Blob.DeleteBlobs(ctx, asStrs(blobURIs)); err != nil {
+	if err := s.deleteBlobs(ctx, asStrs(blobURIs)); err != nil {
 		return nil, oapierr.Internal("failed to delete blob", zap.Error(err))
 	}
 	return api.DeletePortfolio204Response{}, nil

db/sqldb/migrations/0005_json_blob_type.down.sql

@@ -16,4 +16,4 @@ ALTER TABLE blob
     ALTER file_type TYPE file_type 
         USING file_type::file_type;
 
-COMMIT;
+COMMIT;

db/sqldb/sqldb_test.go

@@ -86,6 +86,7 @@ func TestSchemaHistory(t *testing.T) {
 		{ID: 2, Version: 2}, // 0002_create_user_table
 		{ID: 3, Version: 3}, // 0003_domain_types
 		{ID: 4, Version: 4}, // 0004_audit_log_tweaks
+		{ID: 5, Version: 5}, // 0005_json_blob_type
 	}
 
 	if diff := cmp.Diff(want, got); diff != "" {

pacta/enum_test.go

@@ -1,6 +1,9 @@
 package pacta
 
-import "testing"
+import (
+	"path/filepath"
+	"testing"
+)
 
 func TestParseAuthMechanism(t *testing.T) {
 	testParseEnum(t, AuthnMechanismValues, ParseAuthnMechanism)
@@ -12,6 +15,20 @@ func TestParseLanguage(t *testing.T) {
 
 func TestParseFileType(t *testing.T) {
 	testParseEnum(t, FileTypeValues, ParseFileType)
+	otherCases := []string{
+		"hello/world.json",
+		"hello/world/hithere.JsOn",
+		"  hello/world/hithere.json   ",
+	}
+	for _, c := range otherCases {
+		ft, err := ParseFileType(filepath.Ext(c))
+		if err != nil {
+			t.Errorf("expected successful parse, got %w", err)
+		}
+		if ft != FileType_JSON {
+			t.Errorf("expected JSON, got %v", ft)
+		}
+	}
 }
 
 // need

pacta/pacta.go

@@ -204,8 +204,8 @@ const (
 	FileType_CSV  = "csv"
 	FileType_YAML = "yaml"
 	FileType_ZIP  = "zip"
-	FileType_JSON = "json"
 	FileType_HTML = "html"
+	FileType_JSON = "json"
 )
 
 var FileTypeValues = []FileType{
@@ -214,6 +214,7 @@ var FileTypeValues = []FileType{
 	FileType_ZIP,
 	FileType_JSON,
 	FileType_HTML,
+	FileType_JSON,
 }
 
 func ParseFileType(s string) (FileType, error) {
@@ -228,12 +229,12 @@ func ParseFileType(s string) (FileType, error) {
 		return FileType_YAML, nil
 	case "zip":
 		return FileType_ZIP, nil
-	case "json":
-		return FileType_JSON, nil
 	case "html":
 		return FileType_HTML, nil
+	case "json":
+		return FileType_JSON, nil
 	}
-	return "", fmt.Errorf("unknown FileType: %q", s)
+	return "", fmt.Errorf("unknown pacta.FileType: %q", s)
 }
 
 type BlobURI string