Support reusing existing storage containers across task providers (#687)

* Support for existing storage containers in aws and gcp. (#651)

* Update config schema.

* AWS support for existing S3 buckets.

* Existing bucket support for gcp.

* Add mocks and tests to existing bucket resource in aws.

* Update docs with new pre-allocated container fields.

* Support using pre-allocated blob containers in azure. (#660)

* AWS support for existing S3 buckets.

* Existing bucket support for gcp.

* Fix subdirectory in rclone remote.

* Blob container generates the rclone connection string.

* Introduce a type for generating rclone connection strings.

* Azure support for reusable blob containers.

* Update docs.

* Fix path prefix.

* Initialize s3 existing bucket with RemoteStorage struct.

* Update gcp and aws existing bucket data sources to align with the azure data source.

Use common.RemoteStorage to initialize the data sources.
Using rclone to verify storage during Read.
Remove aws s3 client mocks and tests that rely on them.

* Fix comment.

* K8s support for specifying an existing persistent volume claim (#661)

* K8s support for specifying an existing persistent volume claim.

Co-authored-by: Helio Machado <0x2b3bfa0+git@googlemail.com>

* Update use of Identifier struct.

* Combine container and container_path config keys.

* Update docs.

* Use split function from rclone.

Co-authored-by: Helio Machado <0x2b3bfa0+git@googlemail.com>

Author: Domas Monkus

docs/resources/task.md

@@ -65,6 +65,8 @@ resource "iterative_task" "example" {
 - `storage.workdir` - (Optional) Local working directory to upload and use as the `script` working directory.
 - `storage.output` - (Optional) Results directory (**relative to `workdir`**) to download (default: no download).
 - `storage.exclude` - (Optional) List of files and globs to exclude from transfering. Excluded files are neither uploaded to cloud storage nor downloaded from it. Exclusions are defined relative to `storage.workdir`.
+- `storage.container` - (Optional) Pre-allocated container to use for storage of task data, results and status.
+- `storage.container_opts` - (Optional) Block of cloud-specific container settings.
 - `environment` - (Optional) Map of environment variable names and values for the task script. Empty string values are replaced with local environment values. Empty values may also be combined with a [glob](<https://en.wikipedia.org/wiki/Glob_(programming)>) name to import all matching variables.
 - `timeout` - (Optional) Maximum number of seconds to run before instances are force-terminated. The countdown is reset each time TPI auto-respawns a spot instance.
 - `tags` - (Optional) Map of tags for the created cloud resources.
@@ -281,25 +283,86 @@ spec:
 
 ## Permission Set
 
+### Generic
+
 A set of "permissions" assigned to the `task` instance, format depends on the cloud provider
 
-#### Amazon Web Services
+### Cloud-specific
+
+#### Kubernetes
+
+The name of a service account in the current namespace.
+
+### Amazon Web Services
 
 An [instance profile `arn`](https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/iam-roles-for-amazon-ec2.html), e.g.:
 `permission_set = "arn:aws:iam:1234567890:instance-profile/rolename"`
 
-#### Google Cloud Platform
+### Google Cloud Platform
 
 A service account email and a [list of scopes](https://cloud.google.com/sdk/gcloud/reference/alpha/compute/instances/set-scopes#--scopes), e.g.:
 `permission_set = "sa-name@project_id.iam.gserviceaccount.com,scopes=storage-rw"`
 
-#### Microsoft Azure
+### Microsoft Azure
 A comma-separated list of [user-assigned identity](https://docs.microsoft.com/en-us/azure/active-directory/managed-identities-azure-resources/overview) ARM resource ids, e.g.:
 `permission_set = "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}/providers/Microsoft.ManagedIdentity/userAssignedIdentities/{identityName}"`
 
+## Pre-allocated blob container
+
+### Generic
+
+To use a pre-allocated container for storing task data, specify the `container` key in the `storage` section
+of the config:
+
+```hcl
+resource "iterative_task" "example" {
+  (...)
+  storage {
+    container = "container-name/path/path"
+  }
+  (...)
+}
+```
+
+The container name may include a path component, in this case the specified subdirectory will be used
+to store task execution results. Otherwise, a subdirectory will be created with a name matchin the
+task's randomly generated id.
+
+If the container name is suffixed with a forward slash, (`container-name/`), the root of the container
+will be used for storage.
+
+### Cloud-specific
+
+#### Amazon Web Services
+
+The container name is the name of the S3 container. It should be in the same region as the task deployment.
+
+#### Google Cloud Platform
+
+The container name is the name of the google cloud storage container.
+
 #### Kubernetes
 
-The name of a service account in the current namespace.
+The container name is the name of a predefined persistent volume claim.
+
+#### Microsoft Azure
+
+To use a pre-allocated azure blob container, the storage account name and access key need to be specified in
+the `storage` section:
+
+```
+resource "iterative_task" "example" {
+    (...)
+    storage {
+      container = "container-name"
+      container_opts = {
+        account = "storage-account-name"
+        key = "storage-account-key"
+      }
+    }
+    (...)
+}
+```
 
 ## Known Issues
 

go.mod

@@ -23,6 +23,7 @@ require (
 	github.com/docker/go-units v0.4.0
 	github.com/dustinkirkland/golang-petname v0.0.0-20191129215211-8e5a1ed0cff0
 	github.com/gobwas/glob v0.2.3
+	github.com/golang/mock v1.6.0 // indirect
 	github.com/google/go-github/v42 v42.0.0
 	github.com/google/go-github/v45 v45.2.0
 	github.com/google/uuid v1.3.0

go.sum

@@ -443,6 +443,7 @@ github.com/golang/mock v1.4.1/go.mod h1:UOMv5ysSaYNkG+OFQykRIcU/QvvxJf3p21QfJ2Bt
 github.com/golang/mock v1.4.3/go.mod h1:UOMv5ysSaYNkG+OFQykRIcU/QvvxJf3p21QfJ2Bt3cw=
 github.com/golang/mock v1.4.4/go.mod h1:l3mdAwkq5BuhzHwde/uurv3sEJeZMXNpwsxVWU71h+4=
 github.com/golang/mock v1.5.0/go.mod h1:CWnOUgYIOo4TcNZ0wHX3YZCqsaM1I1Jvs6v3mP3KVu8=
+github.com/golang/mock v1.6.0 h1:ErTB+efbowRARo13NNdxyJji2egdxLGQhRaY+DUumQc=
 github.com/golang/mock v1.6.0/go.mod h1:p6yTPP+5HYm5mzsMV8JkE6ZKdX+/wYM6Hr+LicevLPs=
 github.com/golang/protobuf v1.1.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
 github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=

iterative/resource_task.go

@@ -9,10 +9,10 @@ import (
 	"strings"
 	"time"
 
-	"github.com/sirupsen/logrus"
-
 	"github.com/hashicorp/terraform-plugin-sdk/v2/diag"
 	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/schema"
+	"github.com/rclone/rclone/lib/bucket"
+	"github.com/sirupsen/logrus"
 
 	"terraform-provider-iterative/iterative/utils"
 	"terraform-provider-iterative/task"
@@ -139,6 +139,20 @@ func resourceTask() *schema.Resource {
 							Optional: true,
 							Default:  "",
 						},
+						"container": {
+							Type:     schema.TypeString,
+							ForceNew: true,
+							Optional: true,
+							Default:  "",
+						},
+						"container_opts": {
+							Type:     schema.TypeMap,
+							ForceNew: true,
+							Optional: true,
+							Elem: &schema.Schema{
+								Type: schema.TypeString,
+							},
+						},
 						"exclude": {
 							Type:     schema.TypeList,
 							ForceNew: false,
@@ -349,19 +363,38 @@ func resourceTaskBuild(ctx context.Context, d *schema.ResourceData, m interface{
 	var directory string
 	var directoryOut string
 	var excludeList []string
+	var remoteStorage *common.RemoteStorage
 	if d.Get("storage").(*schema.Set).Len() > 0 {
 		storage := d.Get("storage").(*schema.Set).List()[0].(map[string]interface{})
 		directory = storage["workdir"].(string)
 		directoryOut = storage["output"].(string)
-		directoryOut = filepath.Clean(directoryOut)
 		if filepath.IsAbs(directoryOut) || strings.HasPrefix(directoryOut, "../") {
 			return nil, errors.New("storage.output must be inside storage.workdir")
 		}
-
 		excludes := storage["exclude"].([]interface{})
 		for _, exclude := range excludes {
 			excludeList = append(excludeList, exclude.(string))
 		}
+
+		// Propagate configuration for pre-allocated storage container.
+		containerRaw := storage["container"].(string)
+		if containerRaw != "" {
+			container, containerPath := bucket.Split(containerRaw)
+			remoteStorage = &common.RemoteStorage{
+				Container: container,
+				Path:      containerPath,
+				Config:    map[string]string{},
+			}
+			if storage["container_opts"] != nil {
+				remoteConfig := storage["container_opts"].(map[string]interface{})
+				var ok bool
+				for key, value := range remoteConfig {
+					if remoteStorage.Config[key], ok = value.(string); !ok {
+						return nil, fmt.Errorf("invalid value for remote config key %q: %v", key, value)
+					}
+				}
+			}
+		}
 	}
 
 	t := common.Task{
@@ -384,6 +417,7 @@ func resourceTaskBuild(ctx context.Context, d *schema.ResourceData, m interface{
 			},
 			// Egress is open on every port
 		},
+		RemoteStorage: remoteStorage,
 		Spot:          common.Spot(d.Get("spot").(float64)),
 		Parallelism:   uint16(d.Get("parallelism").(int)),
 		PermissionSet: d.Get("permission_set").(string),

task/aws/client/client.go

@@ -33,13 +33,18 @@ func New(ctx context.Context, cloud common.Cloud, tags map[string]string) (*Clie
 	if err != nil {
 		return nil, err
 	}
-
+	credentials, err := config.Credentials.Retrieve(ctx)
+	if err != nil {
+		return nil, err
+	}
 	c := &Client{
-		Cloud:  cloud,
-		Region: region,
-		Tags:   cloud.Tags,
-		Config: config,
+		Cloud:       cloud,
+		Region:      region,
+		Tags:        cloud.Tags,
+		Config:      config,
+		credentials: credentials,
 	}
+
 	c.Services.EC2 = ec2.NewFromConfig(config)
 	c.Services.S3 = s3.NewFromConfig(config)
 	c.Services.STS = sts.NewFromConfig(config)
@@ -52,8 +57,9 @@ type Client struct {
 	Region string
 	Tags   map[string]string
 
-	Config   aws.Config
-	Services struct {
+	Config      aws.Config
+	credentials aws.Credentials
+	Services    struct {
 		EC2         *ec2.Client
 		S3          *s3.Client
 		STS         *sts.Client
@@ -93,3 +99,8 @@ func (c *Client) DecodeError(ctx context.Context, encoded error) error {
 
 	return fmt.Errorf("unable to decode: %s", encoded.Error())
 }
+
+// Credentials returns the AWS credentials the client is currently using.
+func (c *Client) Credentials() aws.Credentials {
+	return c.credentials
+}

task/aws/resources/data_source_bucket.go

@@ -0,0 +1,62 @@
+package resources
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+
+	"terraform-provider-iterative/task/common"
+	"terraform-provider-iterative/task/common/machine"
+)
+
+// NewExistingS3Bucket returns a new data source refering to a pre-allocated
+// S3 bucket.
+func NewExistingS3Bucket(credentials aws.Credentials, storageParams common.RemoteStorage) *ExistingS3Bucket {
+	return &ExistingS3Bucket{
+		credentials: credentials,
+		params:      storageParams,
+	}
+}
+
+// ExistingS3Bucket identifies an existing S3 bucket.
+type ExistingS3Bucket struct {
+	credentials aws.Credentials
+
+	params common.RemoteStorage
+}
+
+// Read verifies the specified S3 bucket is accessible.
+func (b *ExistingS3Bucket) Read(ctx context.Context) error {
+	err := machine.CheckStorage(ctx, b.connection())
+	if err != nil {
+		return fmt.Errorf("failed to verify existing s3 bucket: %w", err)
+	}
+	return nil
+}
+
+func (b *ExistingS3Bucket) connection() machine.RcloneConnection {
+	region := b.params.Config["region"]
+	return machine.RcloneConnection{
+		Backend:   machine.RcloneBackendS3,
+		Container: b.params.Container,
+		Path:      b.params.Path,
+		Config: map[string]string{
+			"provider":          "AWS",
+			"region":            region,
+			"access_key_id":     b.credentials.AccessKeyID,
+			"secret_access_key": b.credentials.SecretAccessKey,
+			"session_token":     b.credentials.SessionToken,
+		},
+	}
+}
+
+// ConnectionString implements common.StorageCredentials.
+// The method returns the rclone connection string for the specific bucket.
+func (b *ExistingS3Bucket) ConnectionString(ctx context.Context) (string, error) {
+	connection := b.connection()
+	return connection.String(), nil
+}
+
+// build-time check to ensure Bucket implements BucketCredentials.
+var _ common.StorageCredentials = (*ExistingS3Bucket)(nil)

task/aws/resources/data_source_bucket_test.go

@@ -0,0 +1,28 @@
+package resources_test
+
+import (
+	"context"
+	"testing"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/stretchr/testify/require"
+
+	"terraform-provider-iterative/task/aws/resources"
+	"terraform-provider-iterative/task/common"
+)
+
+func TestExistingBucketConnectionString(t *testing.T) {
+	ctx := context.Background()
+	creds := aws.Credentials{
+		AccessKeyID:     "access-key-id",
+		SecretAccessKey: "secret-access-key",
+		SessionToken:    "session-token",
+	}
+	b := resources.NewExistingS3Bucket(creds, common.RemoteStorage{
+		Container: "pre-created-bucket",
+		Config:    map[string]string{"region": "us-east-1"},
+		Path:      "subdirectory"})
+	connStr, err := b.ConnectionString(ctx)
+	require.NoError(t, err)
+	require.Equal(t, ":s3,access_key_id='access-key-id',provider='AWS',region='us-east-1',secret_access_key='secret-access-key',session_token='session-token':pre-created-bucket/subdirectory", connStr)
+}

task/aws/resources/data_source_credentials.go

@@ -2,13 +2,12 @@ package resources
 
 import (
 	"context"
-	"fmt"
 
 	"terraform-provider-iterative/task/aws/client"
 	"terraform-provider-iterative/task/common"
 )
 
-func NewCredentials(client *client.Client, identifier common.Identifier, bucket *Bucket) *Credentials {
+func NewCredentials(client *client.Client, identifier common.Identifier, bucket common.StorageCredentials) *Credentials {
 	c := &Credentials{
 		client:     client,
 		Identifier: identifier.Long(),
@@ -21,7 +20,7 @@ type Credentials struct {
 	client       *client.Client
 	Identifier   string
 	Dependencies struct {
-		Bucket *Bucket
+		Bucket common.StorageCredentials
 	}
 	Resource map[string]string
 }
@@ -32,20 +31,16 @@ func (c *Credentials) Read(ctx context.Context) error {
 		return err
 	}
 
-	connectionString := fmt.Sprintf(
-		":s3,provider=AWS,region=%s,access_key_id=%s,secret_access_key=%s,session_token=%s:%s",
-		c.client.Region,
-		credentials.AccessKeyID,
-		credentials.SecretAccessKey,
-		credentials.SessionToken,
-		c.Dependencies.Bucket.Identifier,
-	)
+	bucketConnStr, err := c.Dependencies.Bucket.ConnectionString(ctx)
+	if err != nil {
+		return err
+	}
 
 	c.Resource = map[string]string{
 		"AWS_ACCESS_KEY_ID":       credentials.AccessKeyID,
 		"AWS_SECRET_ACCESS_KEY":   credentials.SecretAccessKey,
 		"AWS_SESSION_TOKEN":       credentials.SessionToken,
-		"RCLONE_REMOTE":           connectionString,
+		"RCLONE_REMOTE":           bucketConnStr,
 		"TPI_TASK_CLOUD_PROVIDER": string(c.client.Cloud.Provider),
 		"TPI_TASK_CLOUD_REGION":   c.client.Region,
 		"TPI_TASK_IDENTIFIER":     c.Identifier,