-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Integrate portfolio processing into runner
This PR integrates the runner with the PACTA portfolio processing code, integrating with Azure blob storage to read + write portfolios and reports. I've tested the `process_portfolio` piece against local Docker (i.e. `dockertask`), but I haven't tested the `create_report` flow yet in it's current incarnation. This PR is already large enough that I'll add the relevant async task stuff in a follow-up PR.
- Loading branch information
Showing
44 changed files
with
1,482 additions
and
406 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,5 @@ | ||
| # creation rules are evaluated sequentially, the first match wins | ||
| creation_rules: | ||
| - path_regex: secrets/local\.enc\.json$ | ||
| azure_keyvault: https://rmipactalocalsops.vault.azure.net/keys/sops/d670bcbc510f456d821306913b4c55c6 | ||
|
|
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,16 @@ | ||
| load("@io_bazel_rules_go//go:def.bzl", "go_library") | ||
|
|
||
| go_library( | ||
| name = "azblob", | ||
| srcs = ["azblob.go"], | ||
| importpath = "github.com/RMI/pacta/azure/azblob", | ||
| visibility = ["//visibility:public"], | ||
| deps = [ | ||
| "//blob", | ||
| "@com_github_azure_azure_sdk_for_go_sdk_azcore//:azcore", | ||
| "@com_github_azure_azure_sdk_for_go_sdk_azcore//to", | ||
| "@com_github_azure_azure_sdk_for_go_sdk_storage_azblob//:azblob", | ||
| "@com_github_azure_azure_sdk_for_go_sdk_storage_azblob//sas", | ||
| "@com_github_azure_azure_sdk_for_go_sdk_storage_azblob//service", | ||
| ], | ||
| ) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,185 @@ | ||
| // Package azblob wraps the existing Azure blob library to provide basic upload, | ||
| // download, and URL signing functionality against a standardized interface. | ||
| package azblob | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "io" | ||
| "sync" | ||
| "time" | ||
|
|
||
| "github.com/Azure/azure-sdk-for-go/sdk/azcore" | ||
| "github.com/Azure/azure-sdk-for-go/sdk/azcore/to" | ||
| "github.com/Azure/azure-sdk-for-go/sdk/storage/azblob" | ||
| "github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/sas" | ||
| azservice "github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/service" | ||
| "github.com/RMI/pacta/blob" | ||
| ) | ||
|
|
||
| const ( | ||
| Scheme = blob.Scheme("az") | ||
| ) | ||
|
|
||
| type Client struct { | ||
| storageAccount string | ||
| now func() time.Time | ||
|
|
||
| client *azblob.Client | ||
| svcClient *azservice.Client | ||
|
|
||
| cachedUDCMu *sync.Mutex | ||
| cachedUDC *azservice.UserDelegationCredential | ||
| cachedUDCExpiry time.Time | ||
| } | ||
|
|
||
| func NewClient(creds azcore.TokenCredential, storageAccount string) (*Client, error) { | ||
| serviceURL := fmt.Sprintf("https://%s.blob.core.windows.net/", storageAccount) | ||
|
|
||
| client, err := azblob.NewClient(serviceURL, creds, nil) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("failed to init Azure blob client: %w", err) | ||
| } | ||
|
|
||
| svcClient, err := azservice.NewClient(serviceURL, creds, nil) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("failed to init Azure blob service client: %w", err) | ||
| } | ||
|
|
||
| return &Client{ | ||
| storageAccount: storageAccount, | ||
| now: func() time.Time { return time.Now().UTC() }, | ||
|
|
||
| client: client, | ||
| svcClient: svcClient, | ||
|
|
||
| cachedUDCMu: &sync.Mutex{}, | ||
| }, nil | ||
| } | ||
|
|
||
| func (c *Client) Scheme() blob.Scheme { | ||
| return Scheme | ||
| } | ||
|
|
||
| func (c *Client) WriteBlob(ctx context.Context, uri string, r io.Reader) error { | ||
| ctr, blb, ok := blob.SplitURI(Scheme, uri) | ||
| if !ok { | ||
| return fmt.Errorf("malformed URI %q is not for Azure", uri) | ||
| } | ||
|
|
||
| if _, err := c.client.UploadStream(ctx, ctr, blb, r, nil); err != nil { | ||
| return fmt.Errorf("failed to upload blob: %w", err) | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| func (c *Client) ReadBlob(ctx context.Context, uri string) (io.ReadCloser, error) { | ||
| ctr, blb, ok := blob.SplitURI(Scheme, uri) | ||
| if !ok { | ||
| return nil, fmt.Errorf("malformed URI %q is not for Azure", uri) | ||
| } | ||
|
|
||
| resp, err := c.client.DownloadStream(ctx, ctr, blb, nil) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("failed to read blob: %w", err) | ||
| } | ||
|
|
||
| return resp.Body, 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/[email protected]/sas#example-package-UserDelegationSAS | ||
| func (c *Client) SignedUploadURL(ctx context.Context, uri string) (string, error) { | ||
| return c.signBlob(ctx, uri, &sas.BlobPermissions{Create: true, Write: true}) | ||
| } | ||
|
|
||
| // SignedDownloadURL returns a URL that is allowed to download the file at the given URI. | ||
| // See https://pkg.go.dev/github.com/Azure/azure-sdk-for-go/sdk/storage/[email protected]/sas#example-package-UserDelegationSAS | ||
| func (c *Client) SignedDownloadURL(ctx context.Context, uri string) (string, error) { | ||
| return c.signBlob(ctx, uri, &sas.BlobPermissions{Read: true}) | ||
| } | ||
|
|
||
| func (c *Client) signBlob(ctx context.Context, uri string, perms *sas.BlobPermissions) (string, error) { | ||
| ctr, blb, ok := blob.SplitURI(Scheme, uri) | ||
| if !ok { | ||
| return "", fmt.Errorf("malformed URI %q is not for Azure", uri) | ||
| } | ||
|
|
||
| // The blob component is important, otherwise the signed URL is applicable to the whole container. | ||
| if blb == "" { | ||
| return "", fmt.Errorf("uri %q did not contain a blob component", uri) | ||
| } | ||
|
|
||
| now := c.now().UTC().Add(-10 * time.Second) | ||
| udc, err := c.getUserDelegationCredential(ctx, now) | ||
| if err != nil { | ||
| return "", fmt.Errorf("failed to get udc: %w", err) | ||
| } | ||
|
|
||
| // Create Blob Signature Values with desired permissions and sign with user delegation credential | ||
| sasQueryParams, err := sas.BlobSignatureValues{ | ||
| Protocol: sas.ProtocolHTTPS, | ||
| StartTime: now, | ||
| ExpiryTime: now.Add(15 * time.Minute), | ||
| Permissions: perms.String(), | ||
| ContainerName: ctr, | ||
| BlobName: blb, | ||
| }.SignWithUserDelegation(udc) | ||
| if err != nil { | ||
| return "", fmt.Errorf("failed to sign blob: %w", err) | ||
| } | ||
|
|
||
| return fmt.Sprintf("https://%s.blob.core.windows.net/%s/%s?%s", c.storageAccount, ctr, blb, sasQueryParams.Encode()), nil | ||
| } | ||
|
|
||
| func (c *Client) ListBlobs(ctx context.Context, uriPrefix string) ([]string, error) { | ||
| ctr, blobPrefix, ok := blob.SplitURI(Scheme, uriPrefix) | ||
| if !ok { | ||
| return nil, fmt.Errorf("malformed URI prefix %q is not for Azure", uriPrefix) | ||
| } | ||
|
|
||
| if blobPrefix == "" { | ||
| return nil, fmt.Errorf("uri prefix %q did not contain a blob component", uriPrefix) | ||
| } | ||
|
|
||
| pager := c.client.NewListBlobsFlatPager(ctr, &azblob.ListBlobsFlatOptions{ | ||
| Prefix: &blobPrefix, | ||
| }) | ||
|
|
||
| var blobs []string | ||
| for pager.More() { | ||
| resp, err := pager.NextPage(ctx) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("failed to load page of blobs: %w", err) | ||
| } | ||
| for _, bi := range resp.Segment.BlobItems { | ||
| blobs = append(blobs, blob.Join(Scheme, ctr, *bi.Name)) | ||
| } | ||
| } | ||
|
|
||
| return blobs, nil | ||
| } | ||
|
|
||
| func (c *Client) getUserDelegationCredential(ctx context.Context, now time.Time) (*azservice.UserDelegationCredential, error) { | ||
| c.cachedUDCMu.Lock() | ||
| defer c.cachedUDCMu.Unlock() | ||
|
|
||
| expiry := now.Add(48 * time.Hour) | ||
| info := azservice.KeyInfo{ | ||
| Start: to.Ptr(now.UTC().Format(sas.TimeFormat)), | ||
| Expiry: to.Ptr(expiry.UTC().Format(sas.TimeFormat)), | ||
| } | ||
|
|
||
| if !c.cachedUDCExpiry.IsZero() && c.cachedUDCExpiry.Sub(now) > 1*time.Minute { | ||
| return c.cachedUDC, nil | ||
| } | ||
|
|
||
| udc, err := c.svcClient.GetUserDelegationCredential(ctx, info, nil) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("failed to get delegated credentials: %w", err) | ||
| } | ||
| c.cachedUDC = udc | ||
| c.cachedUDCExpiry = expiry | ||
|
|
||
| return udc, nil | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.