diff --git a/pkg/apis/ark/v1/config.go b/pkg/apis/ark/v1/config.go
index fa0cec374..097c39dee 100644
--- a/pkg/apis/ark/v1/config.go
+++ b/pkg/apis/ark/v1/config.go
@@ -69,17 +69,11 @@ type Config struct {
 }
 
 // CloudProviderConfig is configuration information about how to connect
-// to a particular cloud. Only one of the members (AWS, GCP, Azure) may
-// be present.
+// to a particular cloud.
 type CloudProviderConfig struct {
-	// AWS is configuration information for connecting to AWS.
-	AWS *AWSConfig `json:"aws"`
+	Name string `json:"name"`
 
-	// GCP is configuration information for connecting to GCP.
-	GCP *GCPConfig `json:"gcp"`
-
-	// Azure is configuration information for connecting to Azure.
-	Azure *AzureConfig `json:"azure"`
+	Config map[string]string `json:"config"`
 }
 
 // ObjectStorageProviderConfig is configuration information for connecting to
@@ -93,23 +87,3 @@ type ObjectStorageProviderConfig struct {
 	// are stored.
 	Bucket string `json:"bucket"`
 }
-
-// AWSConfig is configuration information for connecting to AWS.
-type AWSConfig struct {
-	Region           string `json:"region"`
-	DisableSSL       bool   `json:"disableSSL"`
-	S3ForcePathStyle bool   `json:"s3ForcePathStyle"`
-	S3Url            string `json:"s3Url"`
-	KMSKeyID         string `json:"kmsKeyId"`
-}
-
-// GCPConfig is configuration information for connecting to GCP.
-type GCPConfig struct {
-	Project string `json:"project"`
-}
-
-// AzureConfig is configuration information for connecting to Azure.
-type AzureConfig struct {
-	Location   string          `json:"location"`
-	APITimeout metav1.Duration `json:"apiTimeout"`
-}
diff --git a/pkg/apis/ark/v1/zz_generated.deepcopy.go b/pkg/apis/ark/v1/zz_generated.deepcopy.go
index f31a42244..e73389538 100644
--- a/pkg/apis/ark/v1/zz_generated.deepcopy.go
+++ b/pkg/apis/ark/v1/zz_generated.deepcopy.go
@@ -32,14 +32,6 @@ import (
 // Deprecated: deepcopy registration will go away when static deepcopy is fully implemented.
 func GetGeneratedDeepCopyFuncs() []conversion.GeneratedDeepCopyFunc {
 	return []conversion.GeneratedDeepCopyFunc{
-		{Fn: func(in interface{}, out interface{}, c *conversion.Cloner) error {
-			in.(*AWSConfig).DeepCopyInto(out.(*AWSConfig))
-			return nil
-		}, InType: reflect.TypeOf(&AWSConfig{})},
-		{Fn: func(in interface{}, out interface{}, c *conversion.Cloner) error {
-			in.(*AzureConfig).DeepCopyInto(out.(*AzureConfig))
-			return nil
-		}, InType: reflect.TypeOf(&AzureConfig{})},
 		{Fn: func(in interface{}, out interface{}, c *conversion.Cloner) error {
 			in.(*Backup).DeepCopyInto(out.(*Backup))
 			return nil
@@ -104,10 +96,6 @@ func GetGeneratedDeepCopyFuncs() []conversion.GeneratedDeepCopyFunc {
 			in.(*ExecHook).DeepCopyInto(out.(*ExecHook))
 			return nil
 		}, InType: reflect.TypeOf(&ExecHook{})},
-		{Fn: func(in interface{}, out interface{}, c *conversion.Cloner) error {
-			in.(*GCPConfig).DeepCopyInto(out.(*GCPConfig))
-			return nil
-		}, InType: reflect.TypeOf(&GCPConfig{})},
 		{Fn: func(in interface{}, out interface{}, c *conversion.Cloner) error {
 			in.(*ObjectStorageProviderConfig).DeepCopyInto(out.(*ObjectStorageProviderConfig))
 			return nil
@@ -155,39 +143,6 @@ func GetGeneratedDeepCopyFuncs() []conversion.GeneratedDeepCopyFunc {
 	}
 }
 
-// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
-func (in *AWSConfig) DeepCopyInto(out *AWSConfig) {
-	*out = *in
-	return
-}
-
-// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new AWSConfig.
-func (in *AWSConfig) DeepCopy() *AWSConfig {
-	if in == nil {
-		return nil
-	}
-	out := new(AWSConfig)
-	in.DeepCopyInto(out)
-	return out
-}
-
-// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
-func (in *AzureConfig) DeepCopyInto(out *AzureConfig) {
-	*out = *in
-	out.APITimeout = in.APITimeout
-	return
-}
-
-// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new AzureConfig.
-func (in *AzureConfig) DeepCopy() *AzureConfig {
-	if in == nil {
-		return nil
-	}
-	out := new(AzureConfig)
-	in.DeepCopyInto(out)
-	return out
-}
-
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *Backup) DeepCopyInto(out *Backup) {
 	*out = *in
@@ -453,31 +408,11 @@ func (in *BackupStatus) DeepCopy() *BackupStatus {
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *CloudProviderConfig) DeepCopyInto(out *CloudProviderConfig) {
 	*out = *in
-	if in.AWS != nil {
-		in, out := &in.AWS, &out.AWS
-		if *in == nil {
-			*out = nil
-		} else {
-			*out = new(AWSConfig)
-			**out = **in
-		}
-	}
-	if in.GCP != nil {
-		in, out := &in.GCP, &out.GCP
-		if *in == nil {
-			*out = nil
-		} else {
-			*out = new(GCPConfig)
-			**out = **in
-		}
-	}
-	if in.Azure != nil {
-		in, out := &in.Azure, &out.Azure
-		if *in == nil {
-			*out = nil
-		} else {
-			*out = new(AzureConfig)
-			**out = **in
+	if in.Config != nil {
+		in, out := &in.Config, &out.Config
+		*out = make(map[string]string, len(*in))
+		for key, val := range *in {
+			(*out)[key] = val
 		}
 	}
 	return
@@ -707,22 +642,6 @@ func (in *ExecHook) DeepCopy() *ExecHook {
 	return out
 }
 
-// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
-func (in *GCPConfig) DeepCopyInto(out *GCPConfig) {
-	*out = *in
-	return
-}
-
-// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new GCPConfig.
-func (in *GCPConfig) DeepCopy() *GCPConfig {
-	if in == nil {
-		return nil
-	}
-	out := new(GCPConfig)
-	in.DeepCopyInto(out)
-	return out
-}
-
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *ObjectStorageProviderConfig) DeepCopyInto(out *ObjectStorageProviderConfig) {
 	*out = *in
diff --git a/pkg/cloudprovider/aws/block_store.go b/pkg/cloudprovider/aws/block_store.go
index 1a680f88b..5c41ea6f3 100644
--- a/pkg/cloudprovider/aws/block_store.go
+++ b/pkg/cloudprovider/aws/block_store.go
@@ -27,6 +27,13 @@ import (
 	"github.com/heptio/ark/pkg/cloudprovider"
 )
 
+const regionKey = "region"
+
+// iopsVolumeTypes is a set of AWS EBS volume types for which IOPS should
+// be captured during snapshot and provided when creating a new volume
+// from snapshot.
+var iopsVolumeTypes = sets.NewString("io1")
+
 type blockStore struct {
 	ec2 *ec2.EC2
 }
@@ -44,29 +51,29 @@ func getSession(config *aws.Config) (*session.Session, error) {
 	return sess, nil
 }
 
-func NewBlockStore(region string) (cloudprovider.BlockStore, error) {
+func NewBlockStore() cloudprovider.BlockStore {
+	return &blockStore{}
+}
+
+func (b *blockStore) Init(config map[string]string) error {
+	region := config[regionKey]
 	if region == "" {
-		return nil, errors.New("missing region in aws configuration in config file")
+		return errors.Errorf("missing %s in aws configuration", regionKey)
 	}
 
 	awsConfig := aws.NewConfig().WithRegion(region)
 
 	sess, err := getSession(awsConfig)
 	if err != nil {
-		return nil, err
+		return err
 	}
 
-	return &blockStore{
-		ec2: ec2.New(sess),
-	}, nil
+	b.ec2 = ec2.New(sess)
+
+	return nil
 }
 
-// iopsVolumeTypes is a set of AWS EBS volume types for which IOPS should
-// be captured during snapshot and provided when creating a new volume
-// from snapshot.
-var iopsVolumeTypes = sets.NewString("io1")
-
-func (op *blockStore) CreateVolumeFromSnapshot(snapshotID, volumeType, volumeAZ string, iops *int64) (volumeID string, err error) {
+func (b *blockStore) CreateVolumeFromSnapshot(snapshotID, volumeType, volumeAZ string, iops *int64) (volumeID string, err error) {
 	req := &ec2.CreateVolumeInput{
 		SnapshotId:       &snapshotID,
 		AvailabilityZone: &volumeAZ,
@@ -77,7 +84,7 @@ func (op *blockStore) CreateVolumeFromSnapshot(snapshotID, volumeType, volumeAZ
 		req.Iops = iops
 	}
 
-	res, err := op.ec2.CreateVolume(req)
+	res, err := b.ec2.CreateVolume(req)
 	if err != nil {
 		return "", errors.WithStack(err)
 	}
@@ -85,12 +92,12 @@ func (op *blockStore) CreateVolumeFromSnapshot(snapshotID, volumeType, volumeAZ
 	return *res.VolumeId, nil
 }
 
-func (op *blockStore) GetVolumeInfo(volumeID, volumeAZ string) (string, *int64, error) {
+func (b *blockStore) GetVolumeInfo(volumeID, volumeAZ string) (string, *int64, error) {
 	req := &ec2.DescribeVolumesInput{
 		VolumeIds: []*string{&volumeID},
 	}
 
-	res, err := op.ec2.DescribeVolumes(req)
+	res, err := b.ec2.DescribeVolumes(req)
 	if err != nil {
 		return "", nil, errors.WithStack(err)
 	}
@@ -117,12 +124,12 @@ func (op *blockStore) GetVolumeInfo(volumeID, volumeAZ string) (string, *int64,
 	return volumeType, iops, nil
 }
 
-func (op *blockStore) IsVolumeReady(volumeID, volumeAZ string) (ready bool, err error) {
+func (b *blockStore) IsVolumeReady(volumeID, volumeAZ string) (ready bool, err error) {
 	req := &ec2.DescribeVolumesInput{
 		VolumeIds: []*string{&volumeID},
 	}
 
-	res, err := op.ec2.DescribeVolumes(req)
+	res, err := b.ec2.DescribeVolumes(req)
 	if err != nil {
 		return false, errors.WithStack(err)
 	}
@@ -133,7 +140,7 @@ func (op *blockStore) IsVolumeReady(volumeID, volumeAZ string) (ready bool, err
 	return *res.Volumes[0].State == ec2.VolumeStateAvailable, nil
 }
 
-func (op *blockStore) ListSnapshots(tagFilters map[string]string) ([]string, error) {
+func (b *blockStore) ListSnapshots(tagFilters map[string]string) ([]string, error) {
 	req := &ec2.DescribeSnapshotsInput{}
 
 	for k, v := range tagFilters {
@@ -145,7 +152,7 @@ func (op *blockStore) ListSnapshots(tagFilters map[string]string) ([]string, err
 	}
 
 	var ret []string
-	err := op.ec2.DescribeSnapshotsPages(req, func(res *ec2.DescribeSnapshotsOutput, lastPage bool) bool {
+	err := b.ec2.DescribeSnapshotsPages(req, func(res *ec2.DescribeSnapshotsOutput, lastPage bool) bool {
 		for _, snapshot := range res.Snapshots {
 			ret = append(ret, *snapshot.SnapshotId)
 		}
@@ -159,12 +166,12 @@ func (op *blockStore) ListSnapshots(tagFilters map[string]string) ([]string, err
 	return ret, nil
 }
 
-func (op *blockStore) CreateSnapshot(volumeID, volumeAZ string, tags map[string]string) (string, error) {
+func (b *blockStore) CreateSnapshot(volumeID, volumeAZ string, tags map[string]string) (string, error) {
 	req := &ec2.CreateSnapshotInput{
 		VolumeId: &volumeID,
 	}
 
-	res, err := op.ec2.CreateSnapshot(req)
+	res, err := b.ec2.CreateSnapshot(req)
 	if err != nil {
 		return "", errors.WithStack(err)
 	}
@@ -184,17 +191,17 @@ func (op *blockStore) CreateSnapshot(volumeID, volumeAZ string, tags map[string]
 
 	tagsReq.SetTags(ec2Tags)
 
-	_, err = op.ec2.CreateTags(tagsReq)
+	_, err = b.ec2.CreateTags(tagsReq)
 
 	return *res.SnapshotId, errors.WithStack(err)
 }
 
-func (op *blockStore) DeleteSnapshot(snapshotID string) error {
+func (b *blockStore) DeleteSnapshot(snapshotID string) error {
 	req := &ec2.DeleteSnapshotInput{
 		SnapshotId: &snapshotID,
 	}
 
-	_, err := op.ec2.DeleteSnapshot(req)
+	_, err := b.ec2.DeleteSnapshot(req)
 
 	return errors.WithStack(err)
 }
diff --git a/pkg/cloudprovider/aws/object_store.go b/pkg/cloudprovider/aws/object_store.go
index 6e8a530cf..6aa1874a3 100644
--- a/pkg/cloudprovider/aws/object_store.go
+++ b/pkg/cloudprovider/aws/object_store.go
@@ -18,6 +18,7 @@ package aws
 
 import (
 	"io"
+	"strconv"
 	"time"
 
 	"github.com/aws/aws-sdk-go/aws"
@@ -29,15 +30,40 @@ import (
 	"github.com/heptio/ark/pkg/cloudprovider"
 )
 
+const (
+	s3URLKey            = "s3Url"
+	kmsKeyIDKey         = "kmsKeyId"
+	s3ForcePathStyleKey = "s3ForcePathStyle"
+)
+
 type objectStore struct {
 	s3         *s3.S3
 	s3Uploader *s3manager.Uploader
 	kmsKeyID   string
 }
 
-func NewObjectStore(region, s3URL, kmsKeyID string, s3ForcePathStyle bool) (cloudprovider.ObjectStore, error) {
+func NewObjectStore() cloudprovider.ObjectStore {
+	return &objectStore{}
+}
+
+func (o *objectStore) Init(config map[string]string) error {
+	var (
+		region              = config[regionKey]
+		s3URL               = config[s3URLKey]
+		kmsKeyID            = config[kmsKeyIDKey]
+		s3ForcePathStyleVal = config[s3ForcePathStyleKey]
+		s3ForcePathStyle    bool
+		err                 error
+	)
+
 	if region == "" {
-		return nil, errors.New("missing region in aws configuration in config file")
+		return errors.Errorf("missing %s in aws configuration", regionKey)
+	}
+
+	if s3ForcePathStyleVal != "" {
+		if s3ForcePathStyle, err = strconv.ParseBool(s3ForcePathStyleVal); err != nil {
+			return errors.Wrapf(err, "could not parse %s (expected bool)", s3ForcePathStyleKey)
+		}
 	}
 
 	awsConfig := aws.NewConfig().
@@ -60,17 +86,17 @@ func NewObjectStore(region, s3URL, kmsKeyID string, s3ForcePathStyle bool) (clou
 
 	sess, err := getSession(awsConfig)
 	if err != nil {
-		return nil, err
+		return err
 	}
 
-	return &objectStore{
-		s3:         s3.New(sess),
-		s3Uploader: s3manager.NewUploader(sess),
-		kmsKeyID:   kmsKeyID,
-	}, nil
+	o.s3 = s3.New(sess)
+	o.s3Uploader = s3manager.NewUploader(sess)
+	o.kmsKeyID = kmsKeyID
+
+	return nil
 }
 
-func (op *objectStore) PutObject(bucket string, key string, body io.Reader) error {
+func (o *objectStore) PutObject(bucket string, key string, body io.Reader) error {
 	req := &s3manager.UploadInput{
 		Bucket: &bucket,
 		Key:    &key,
@@ -78,23 +104,23 @@ func (op *objectStore) PutObject(bucket string, key string, body io.Reader) erro
 	}
 
 	// if kmsKeyID is not empty, enable "aws:kms" encryption
-	if op.kmsKeyID != "" {
+	if o.kmsKeyID != "" {
 		req.ServerSideEncryption = aws.String("aws:kms")
-		req.SSEKMSKeyId = &op.kmsKeyID
+		req.SSEKMSKeyId = &o.kmsKeyID
 	}
 
-	_, err := op.s3Uploader.Upload(req)
+	_, err := o.s3Uploader.Upload(req)
 
 	return errors.Wrapf(err, "error putting object %s", key)
 }
 
-func (op *objectStore) GetObject(bucket string, key string) (io.ReadCloser, error) {
+func (o *objectStore) GetObject(bucket string, key string) (io.ReadCloser, error) {
 	req := &s3.GetObjectInput{
 		Bucket: &bucket,
 		Key:    &key,
 	}
 
-	res, err := op.s3.GetObject(req)
+	res, err := o.s3.GetObject(req)
 	if err != nil {
 		return nil, errors.Wrapf(err, "error getting object %s", key)
 	}
@@ -102,14 +128,14 @@ func (op *objectStore) GetObject(bucket string, key string) (io.ReadCloser, erro
 	return res.Body, nil
 }
 
-func (op *objectStore) ListCommonPrefixes(bucket string, delimiter string) ([]string, error) {
+func (o *objectStore) ListCommonPrefixes(bucket string, delimiter string) ([]string, error) {
 	req := &s3.ListObjectsV2Input{
 		Bucket:    &bucket,
 		Delimiter: &delimiter,
 	}
 
 	var ret []string
-	err := op.s3.ListObjectsV2Pages(req, func(page *s3.ListObjectsV2Output, lastPage bool) bool {
+	err := o.s3.ListObjectsV2Pages(req, func(page *s3.ListObjectsV2Output, lastPage bool) bool {
 		for _, prefix := range page.CommonPrefixes {
 			ret = append(ret, *prefix.Prefix)
 		}
@@ -123,14 +149,14 @@ func (op *objectStore) ListCommonPrefixes(bucket string, delimiter string) ([]st
 	return ret, nil
 }
 
-func (op *objectStore) ListObjects(bucket, prefix string) ([]string, error) {
+func (o *objectStore) ListObjects(bucket, prefix string) ([]string, error) {
 	req := &s3.ListObjectsV2Input{
 		Bucket: &bucket,
 		Prefix: &prefix,
 	}
 
 	var ret []string
-	err := op.s3.ListObjectsV2Pages(req, func(page *s3.ListObjectsV2Output, lastPage bool) bool {
+	err := o.s3.ListObjectsV2Pages(req, func(page *s3.ListObjectsV2Output, lastPage bool) bool {
 		for _, obj := range page.Contents {
 			ret = append(ret, *obj.Key)
 		}
@@ -144,19 +170,19 @@ func (op *objectStore) ListObjects(bucket, prefix string) ([]string, error) {
 	return ret, nil
 }
 
-func (op *objectStore) DeleteObject(bucket string, key string) error {
+func (o *objectStore) DeleteObject(bucket string, key string) error {
 	req := &s3.DeleteObjectInput{
 		Bucket: &bucket,
 		Key:    &key,
 	}
 
-	_, err := op.s3.DeleteObject(req)
+	_, err := o.s3.DeleteObject(req)
 
 	return errors.Wrapf(err, "error deleting object %s", key)
 }
 
-func (op *objectStore) CreateSignedURL(bucket, key string, ttl time.Duration) (string, error) {
-	req, _ := op.s3.GetObjectRequest(&s3.GetObjectInput{
+func (o *objectStore) CreateSignedURL(bucket, key string, ttl time.Duration) (string, error) {
+	req, _ := o.s3.GetObjectRequest(&s3.GetObjectInput{
 		Bucket: aws.String(bucket),
 		Key:    aws.String(key),
 	})
diff --git a/pkg/cloudprovider/azure/block_store.go b/pkg/cloudprovider/azure/block_store.go
index 368e803ff..8824dedc6 100644
--- a/pkg/cloudprovider/azure/block_store.go
+++ b/pkg/cloudprovider/azure/block_store.go
@@ -33,15 +33,6 @@ import (
 	"github.com/heptio/ark/pkg/cloudprovider"
 )
 
-type blockStore struct {
-	disks         *disk.DisksClient
-	snaps         *disk.SnapshotsClient
-	subscription  string
-	resourceGroup string
-	location      string
-	apiTimeout    time.Duration
-}
-
 const (
 	azureClientIDKey         string = "AZURE_CLIENT_ID"
 	azureClientSecretKey     string = "AZURE_CLIENT_SECRET"
@@ -50,8 +41,20 @@ const (
 	azureStorageAccountIDKey string = "AZURE_STORAGE_ACCOUNT_ID"
 	azureStorageKeyKey       string = "AZURE_STORAGE_KEY"
 	azureResourceGroupKey    string = "AZURE_RESOURCE_GROUP"
+
+	locationKey   = "location"
+	apiTimeoutKey = "apiTimeout"
 )
 
+type blockStore struct {
+	disks         *disk.DisksClient
+	snaps         *disk.SnapshotsClient
+	subscription  string
+	resourceGroup string
+	location      string
+	apiTimeout    time.Duration
+}
+
 func getConfig() map[string]string {
 	cfg := map[string]string{
 		azureClientIDKey:         "",
@@ -70,9 +73,24 @@ func getConfig() map[string]string {
 	return cfg
 }
 
-func NewBlockStore(location string, apiTimeout time.Duration) (cloudprovider.BlockStore, error) {
+func NewBlockStore() cloudprovider.BlockStore {
+	return &blockStore{}
+}
+
+func (b *blockStore) Init(config map[string]string) error {
+	var (
+		location      = config[locationKey]
+		apiTimeoutVal = config[apiTimeoutKey]
+		apiTimeout    time.Duration
+		err           error
+	)
+
 	if location == "" {
-		return nil, errors.New("missing location in azure configuration in config file")
+		return errors.Errorf("missing %s in azure configuration", locationKey)
+	}
+
+	if apiTimeout, err = time.ParseDuration(apiTimeoutVal); err != nil {
+		return errors.Wrapf(err, "could not parse %s (expected time.Duration)", apiTimeoutKey)
 	}
 
 	if apiTimeout == 0 {
@@ -83,7 +101,7 @@ func NewBlockStore(location string, apiTimeout time.Duration) (cloudprovider.Blo
 
 	spt, err := helpers.NewServicePrincipalTokenFromCredentials(cfg, azure.PublicCloud.ResourceManagerEndpoint)
 	if err != nil {
-		return nil, errors.Wrap(err, "error creating new service principal token")
+		return errors.Wrap(err, "error creating new service principal token")
 	}
 
 	disksClient := disk.NewDisksClient(cfg[azureSubscriptionIDKey])
@@ -99,11 +117,11 @@ func NewBlockStore(location string, apiTimeout time.Duration) (cloudprovider.Blo
 
 	locs, err := groupClient.ListLocations(cfg[azureSubscriptionIDKey])
 	if err != nil {
-		return nil, errors.WithStack(err)
+		return errors.WithStack(err)
 	}
 
 	if locs.Value == nil {
-		return nil, errors.New("no locations returned from Azure API")
+		return errors.New("no locations returned from Azure API")
 	}
 
 	locationExists := false
@@ -115,26 +133,26 @@ func NewBlockStore(location string, apiTimeout time.Duration) (cloudprovider.Blo
 	}
 
 	if !locationExists {
-		return nil, errors.Errorf("location %q not found", location)
+		return errors.Errorf("location %q not found", location)
 	}
 
-	return &blockStore{
-		disks:         &disksClient,
-		snaps:         &snapsClient,
-		subscription:  cfg[azureSubscriptionIDKey],
-		resourceGroup: cfg[azureResourceGroupKey],
-		location:      location,
-		apiTimeout:    apiTimeout,
-	}, nil
+	b.disks = &disksClient
+	b.snaps = &snapsClient
+	b.subscription = cfg[azureSubscriptionIDKey]
+	b.resourceGroup = cfg[azureResourceGroupKey]
+	b.location = location
+	b.apiTimeout = apiTimeout
+
+	return nil
 }
 
-func (op *blockStore) CreateVolumeFromSnapshot(snapshotID, volumeType, volumeAZ string, iops *int64) (string, error) {
-	fullSnapshotName := getFullSnapshotName(op.subscription, op.resourceGroup, snapshotID)
+func (b *blockStore) CreateVolumeFromSnapshot(snapshotID, volumeType, volumeAZ string, iops *int64) (string, error) {
+	fullSnapshotName := getFullSnapshotName(b.subscription, b.resourceGroup, snapshotID)
 	diskName := "restore-" + uuid.NewV4().String()
 
 	disk := disk.Model{
 		Name:     &diskName,
-		Location: &op.location,
+		Location: &b.location,
 		Properties: &disk.Properties{
 			CreationData: &disk.CreationData{
 				CreateOption:     disk.Copy,
@@ -144,10 +162,10 @@ func (op *blockStore) CreateVolumeFromSnapshot(snapshotID, volumeType, volumeAZ
 		},
 	}
 
-	ctx, cancel := context.WithTimeout(context.Background(), op.apiTimeout)
+	ctx, cancel := context.WithTimeout(context.Background(), b.apiTimeout)
 	defer cancel()
 
-	_, errChan := op.disks.CreateOrUpdate(op.resourceGroup, *disk.Name, disk, ctx.Done())
+	_, errChan := b.disks.CreateOrUpdate(b.resourceGroup, *disk.Name, disk, ctx.Done())
 
 	err := <-errChan
 
@@ -157,8 +175,8 @@ func (op *blockStore) CreateVolumeFromSnapshot(snapshotID, volumeType, volumeAZ
 	return diskName, nil
 }
 
-func (op *blockStore) GetVolumeInfo(volumeID, volumeAZ string) (string, *int64, error) {
-	res, err := op.disks.Get(op.resourceGroup, volumeID)
+func (b *blockStore) GetVolumeInfo(volumeID, volumeAZ string) (string, *int64, error) {
+	res, err := b.disks.Get(b.resourceGroup, volumeID)
 	if err != nil {
 		return "", nil, errors.WithStack(err)
 	}
@@ -166,8 +184,8 @@ func (op *blockStore) GetVolumeInfo(volumeID, volumeAZ string) (string, *int64,
 	return string(res.AccountType), nil, nil
 }
 
-func (op *blockStore) IsVolumeReady(volumeID, volumeAZ string) (ready bool, err error) {
-	res, err := op.disks.Get(op.resourceGroup, volumeID)
+func (b *blockStore) IsVolumeReady(volumeID, volumeAZ string) (ready bool, err error) {
+	res, err := b.disks.Get(b.resourceGroup, volumeID)
 	if err != nil {
 		return false, errors.WithStack(err)
 	}
@@ -179,8 +197,8 @@ func (op *blockStore) IsVolumeReady(volumeID, volumeAZ string) (ready bool, err
 	return *res.ProvisioningState == "Succeeded", nil
 }
 
-func (op *blockStore) ListSnapshots(tagFilters map[string]string) ([]string, error) {
-	res, err := op.snaps.ListByResourceGroup(op.resourceGroup)
+func (b *blockStore) ListSnapshots(tagFilters map[string]string) ([]string, error) {
+	res, err := b.snaps.ListByResourceGroup(b.resourceGroup)
 	if err != nil {
 		return nil, errors.WithStack(err)
 	}
@@ -213,8 +231,8 @@ Snapshot:
 	return ret, nil
 }
 
-func (op *blockStore) CreateSnapshot(volumeID, volumeAZ string, tags map[string]string) (string, error) {
-	fullDiskName := getFullDiskName(op.subscription, op.resourceGroup, volumeID)
+func (b *blockStore) CreateSnapshot(volumeID, volumeAZ string, tags map[string]string) (string, error) {
+	fullDiskName := getFullDiskName(b.subscription, b.resourceGroup, volumeID)
 	// snapshot names must be <= 80 characters long
 	var snapshotName string
 	suffix := "-" + uuid.NewV4().String()
@@ -234,7 +252,7 @@ func (op *blockStore) CreateSnapshot(volumeID, volumeAZ string, tags map[string]
 			},
 		},
 		Tags:     &map[string]*string{},
-		Location: &op.location,
+		Location: &b.location,
 	}
 
 	for k, v := range tags {
@@ -242,10 +260,10 @@ func (op *blockStore) CreateSnapshot(volumeID, volumeAZ string, tags map[string]
 		(*snap.Tags)[k] = &val
 	}
 
-	ctx, cancel := context.WithTimeout(context.Background(), op.apiTimeout)
+	ctx, cancel := context.WithTimeout(context.Background(), b.apiTimeout)
 	defer cancel()
 
-	_, errChan := op.snaps.CreateOrUpdate(op.resourceGroup, *snap.Name, snap, ctx.Done())
+	_, errChan := b.snaps.CreateOrUpdate(b.resourceGroup, *snap.Name, snap, ctx.Done())
 
 	err := <-errChan
 
@@ -256,11 +274,11 @@ func (op *blockStore) CreateSnapshot(volumeID, volumeAZ string, tags map[string]
 	return snapshotName, nil
 }
 
-func (op *blockStore) DeleteSnapshot(snapshotID string) error {
-	ctx, cancel := context.WithTimeout(context.Background(), op.apiTimeout)
+func (b *blockStore) DeleteSnapshot(snapshotID string) error {
+	ctx, cancel := context.WithTimeout(context.Background(), b.apiTimeout)
 	defer cancel()
 
-	_, errChan := op.snaps.Delete(op.resourceGroup, snapshotID, ctx.Done())
+	_, errChan := b.snaps.Delete(b.resourceGroup, snapshotID, ctx.Done())
 
 	err := <-errChan
 
diff --git a/pkg/cloudprovider/azure/object_store.go b/pkg/cloudprovider/azure/object_store.go
index cd0954a65..24f24ca41 100644
--- a/pkg/cloudprovider/azure/object_store.go
+++ b/pkg/cloudprovider/azure/object_store.go
@@ -27,29 +27,31 @@ import (
 	"github.com/heptio/ark/pkg/cloudprovider"
 )
 
-// ref. https://github.com/Azure-Samples/storage-blob-go-getting-started/blob/master/storageExample.go
-
 type objectStore struct {
 	blobClient *storage.BlobStorageClient
 }
 
-func NewObjectStore() (cloudprovider.ObjectStore, error) {
+func NewObjectStore() cloudprovider.ObjectStore {
+	return &objectStore{}
+}
+
+func (o *objectStore) Init(config map[string]string) error {
 	cfg := getConfig()
 
 	storageClient, err := storage.NewBasicClient(cfg[azureStorageAccountIDKey], cfg[azureStorageKeyKey])
 	if err != nil {
-		return nil, errors.WithStack(err)
+		return errors.WithStack(err)
 	}
 
 	blobClient := storageClient.GetBlobService()
 
-	return &objectStore{
-		blobClient: &blobClient,
-	}, nil
+	o.blobClient = &blobClient
+
+	return nil
 }
 
-func (op *objectStore) PutObject(bucket string, key string, body io.Reader) error {
-	container, err := getContainerReference(op.blobClient, bucket)
+func (o *objectStore) PutObject(bucket string, key string, body io.Reader) error {
+	container, err := getContainerReference(o.blobClient, bucket)
 	if err != nil {
 		return err
 	}
@@ -62,8 +64,8 @@ func (op *objectStore) PutObject(bucket string, key string, body io.Reader) erro
 	return errors.WithStack(blob.CreateBlockBlobFromReader(body, nil))
 }
 
-func (op *objectStore) GetObject(bucket string, key string) (io.ReadCloser, error) {
-	container, err := getContainerReference(op.blobClient, bucket)
+func (o *objectStore) GetObject(bucket string, key string) (io.ReadCloser, error) {
+	container, err := getContainerReference(o.blobClient, bucket)
 	if err != nil {
 		return nil, err
 	}
@@ -81,8 +83,8 @@ func (op *objectStore) GetObject(bucket string, key string) (io.ReadCloser, erro
 	return res, nil
 }
 
-func (op *objectStore) ListCommonPrefixes(bucket string, delimiter string) ([]string, error) {
-	container, err := getContainerReference(op.blobClient, bucket)
+func (o *objectStore) ListCommonPrefixes(bucket string, delimiter string) ([]string, error) {
+	container, err := getContainerReference(o.blobClient, bucket)
 	if err != nil {
 		return nil, err
 	}
@@ -106,8 +108,8 @@ func (op *objectStore) ListCommonPrefixes(bucket string, delimiter string) ([]st
 	return ret, nil
 }
 
-func (op *objectStore) ListObjects(bucket, prefix string) ([]string, error) {
-	container, err := getContainerReference(op.blobClient, bucket)
+func (o *objectStore) ListObjects(bucket, prefix string) ([]string, error) {
+	container, err := getContainerReference(o.blobClient, bucket)
 	if err != nil {
 		return nil, err
 	}
@@ -129,8 +131,8 @@ func (op *objectStore) ListObjects(bucket, prefix string) ([]string, error) {
 	return ret, nil
 }
 
-func (op *objectStore) DeleteObject(bucket string, key string) error {
-	container, err := getContainerReference(op.blobClient, bucket)
+func (o *objectStore) DeleteObject(bucket string, key string) error {
+	container, err := getContainerReference(o.blobClient, bucket)
 	if err != nil {
 		return err
 	}
@@ -145,8 +147,8 @@ func (op *objectStore) DeleteObject(bucket string, key string) error {
 
 const sasURIReadPermission = "r"
 
-func (op *objectStore) CreateSignedURL(bucket, key string, ttl time.Duration) (string, error) {
-	container, err := getContainerReference(op.blobClient, bucket)
+func (o *objectStore) CreateSignedURL(bucket, key string, ttl time.Duration) (string, error) {
+	container, err := getContainerReference(o.blobClient, bucket)
 	if err != nil {
 		return "", err
 	}
diff --git a/pkg/cloudprovider/gcp/block_store.go b/pkg/cloudprovider/gcp/block_store.go
index 18d756bef..f012a82f0 100644
--- a/pkg/cloudprovider/gcp/block_store.go
+++ b/pkg/cloudprovider/gcp/block_store.go
@@ -31,44 +31,52 @@ import (
 	"github.com/heptio/ark/pkg/cloudprovider"
 )
 
+const projectKey = "project"
+
 type blockStore struct {
 	gce     *compute.Service
 	project string
 }
 
-func NewBlockStore(project string) (cloudprovider.BlockStore, error) {
+func NewBlockStore() cloudprovider.BlockStore {
+	return &blockStore{}
+}
+
+func (b *blockStore) Init(config map[string]string) error {
+	project := config[projectKey]
+
 	if project == "" {
-		return nil, errors.New("missing project in gcp configuration in config file")
+		return errors.Errorf("missing %s in gcp configuration", projectKey)
 	}
 
 	client, err := google.DefaultClient(oauth2.NoContext, compute.ComputeScope)
 	if err != nil {
-		return nil, errors.WithStack(err)
+		return errors.WithStack(err)
 	}
 
 	gce, err := compute.New(client)
 	if err != nil {
-		return nil, errors.WithStack(err)
+		return errors.WithStack(err)
 	}
 
 	// validate project
 	res, err := gce.Projects.Get(project).Do()
 	if err != nil {
-		return nil, errors.WithStack(err)
+		return errors.WithStack(err)
 	}
 
 	if res == nil {
-		return nil, errors.Errorf("error getting project %q", project)
+		return errors.Errorf("error getting project %q", project)
 	}
 
-	return &blockStore{
-		gce:     gce,
-		project: project,
-	}, nil
+	b.gce = gce
+	b.project = project
+
+	return nil
 }
 
-func (op *blockStore) CreateVolumeFromSnapshot(snapshotID, volumeType, volumeAZ string, iops *int64) (volumeID string, err error) {
-	res, err := op.gce.Snapshots.Get(op.project, snapshotID).Do()
+func (b *blockStore) CreateVolumeFromSnapshot(snapshotID, volumeType, volumeAZ string, iops *int64) (volumeID string, err error) {
+	res, err := b.gce.Snapshots.Get(b.project, snapshotID).Do()
 	if err != nil {
 		return "", errors.WithStack(err)
 	}
@@ -79,15 +87,15 @@ func (op *blockStore) CreateVolumeFromSnapshot(snapshotID, volumeType, volumeAZ
 		Type:           volumeType,
 	}
 
-	if _, err = op.gce.Disks.Insert(op.project, volumeAZ, disk).Do(); err != nil {
+	if _, err = b.gce.Disks.Insert(b.project, volumeAZ, disk).Do(); err != nil {
 		return "", errors.WithStack(err)
 	}
 
 	return disk.Name, nil
 }
 
-func (op *blockStore) GetVolumeInfo(volumeID, volumeAZ string) (string, *int64, error) {
-	res, err := op.gce.Disks.Get(op.project, volumeAZ, volumeID).Do()
+func (b *blockStore) GetVolumeInfo(volumeID, volumeAZ string) (string, *int64, error) {
+	res, err := b.gce.Disks.Get(b.project, volumeAZ, volumeID).Do()
 	if err != nil {
 		return "", nil, errors.WithStack(err)
 	}
@@ -95,8 +103,8 @@ func (op *blockStore) GetVolumeInfo(volumeID, volumeAZ string) (string, *int64,
 	return res.Type, nil, nil
 }
 
-func (op *blockStore) IsVolumeReady(volumeID, volumeAZ string) (ready bool, err error) {
-	disk, err := op.gce.Disks.Get(op.project, volumeAZ, volumeID).Do()
+func (b *blockStore) IsVolumeReady(volumeID, volumeAZ string) (ready bool, err error) {
+	disk, err := b.gce.Disks.Get(b.project, volumeAZ, volumeID).Do()
 	if err != nil {
 		return false, errors.WithStack(err)
 	}
@@ -105,7 +113,7 @@ func (op *blockStore) IsVolumeReady(volumeID, volumeAZ string) (ready bool, err
 	return disk.Status == "READY", nil
 }
 
-func (op *blockStore) ListSnapshots(tagFilters map[string]string) ([]string, error) {
+func (b *blockStore) ListSnapshots(tagFilters map[string]string) ([]string, error) {
 	useParentheses := len(tagFilters) > 1
 	subFilters := make([]string, 0, len(tagFilters))
 
@@ -119,7 +127,7 @@ func (op *blockStore) ListSnapshots(tagFilters map[string]string) ([]string, err
 
 	filter := strings.Join(subFilters, " ")
 
-	res, err := op.gce.Snapshots.List(op.project).Filter(filter).Do()
+	res, err := b.gce.Snapshots.List(b.project).Filter(filter).Do()
 	if err != nil {
 		return nil, errors.WithStack(err)
 	}
@@ -132,7 +140,7 @@ func (op *blockStore) ListSnapshots(tagFilters map[string]string) ([]string, err
 	return ret, nil
 }
 
-func (op *blockStore) CreateSnapshot(volumeID, volumeAZ string, tags map[string]string) (string, error) {
+func (b *blockStore) CreateSnapshot(volumeID, volumeAZ string, tags map[string]string) (string, error) {
 	// snapshot names must adhere to RFC1035 and be 1-63 characters
 	// long
 	var snapshotName string
@@ -148,7 +156,7 @@ func (op *blockStore) CreateSnapshot(volumeID, volumeAZ string, tags map[string]
 		Name: snapshotName,
 	}
 
-	_, err := op.gce.Disks.CreateSnapshot(op.project, volumeAZ, volumeID, &gceSnap).Do()
+	_, err := b.gce.Disks.CreateSnapshot(b.project, volumeAZ, volumeID, &gceSnap).Do()
 	if err != nil {
 		return "", errors.WithStack(err)
 	}
@@ -156,7 +164,7 @@ func (op *blockStore) CreateSnapshot(volumeID, volumeAZ string, tags map[string]
 	// the snapshot is not immediately available after creation for putting labels
 	// on it. poll for a period of time.
 	if pollErr := wait.Poll(1*time.Second, 30*time.Second, func() (bool, error) {
-		if res, err := op.gce.Snapshots.Get(op.project, gceSnap.Name).Do(); err == nil {
+		if res, err := b.gce.Snapshots.Get(b.project, gceSnap.Name).Do(); err == nil {
 			gceSnap = *res
 			return true, nil
 		}
@@ -170,7 +178,7 @@ func (op *blockStore) CreateSnapshot(volumeID, volumeAZ string, tags map[string]
 		LabelFingerprint: gceSnap.LabelFingerprint,
 	}
 
-	_, err = op.gce.Snapshots.SetLabels(op.project, gceSnap.Name, labels).Do()
+	_, err = b.gce.Snapshots.SetLabels(b.project, gceSnap.Name, labels).Do()
 	if err != nil {
 		return "", errors.WithStack(err)
 	}
@@ -178,8 +186,8 @@ func (op *blockStore) CreateSnapshot(volumeID, volumeAZ string, tags map[string]
 	return gceSnap.Name, nil
 }
 
-func (op *blockStore) DeleteSnapshot(snapshotID string) error {
-	_, err := op.gce.Snapshots.Delete(op.project, snapshotID).Do()
+func (b *blockStore) DeleteSnapshot(snapshotID string) error {
+	_, err := b.gce.Snapshots.Delete(b.project, snapshotID).Do()
 
 	return errors.WithStack(err)
 }
diff --git a/pkg/cloudprovider/gcp/object_store.go b/pkg/cloudprovider/gcp/object_store.go
index b26b42780..34fac5372 100644
--- a/pkg/cloudprovider/gcp/object_store.go
+++ b/pkg/cloudprovider/gcp/object_store.go
@@ -18,6 +18,8 @@ package gcp
 
 import (
 	"io"
+	"io/ioutil"
+	"os"
 	"strings"
 	"time"
 
@@ -31,42 +33,69 @@ import (
 	"github.com/heptio/ark/pkg/cloudprovider"
 )
 
+const credentialsEnvVar = "GOOGLE_APPLICATION_CREDENTIALS"
+
 type objectStore struct {
 	gcs            *storage.Service
 	googleAccessID string
 	privateKey     []byte
 }
 
-func NewObjectStore(googleAccessID string, privateKey []byte) (cloudprovider.ObjectStore, error) {
+func NewObjectStore() cloudprovider.ObjectStore {
+	return &objectStore{}
+}
+
+func (o *objectStore) Init(config map[string]string) error {
+	credentialsFile := os.Getenv(credentialsEnvVar)
+	if credentialsFile == "" {
+		return errors.Errorf("%s is undefined", credentialsEnvVar)
+	}
+
+	// Get the email and private key from the credentials file so we can pre-sign download URLs
+	creds, err := ioutil.ReadFile(credentialsFile)
+	if err != nil {
+		return errors.WithStack(err)
+	}
+	jwtConfig, err := google.JWTConfigFromJSON(creds)
+	if err != nil {
+		return errors.WithStack(err)
+	}
+	if jwtConfig.Email == "" {
+		return errors.Errorf("credentials file pointed to by %s does not contain an email", credentialsEnvVar)
+	}
+	if len(jwtConfig.PrivateKey) == 0 {
+		return errors.Errorf("credentials file pointed to by %s does not contain a private key", credentialsEnvVar)
+	}
+
 	client, err := google.DefaultClient(oauth2.NoContext, storage.DevstorageReadWriteScope)
 	if err != nil {
-		return nil, errors.WithStack(err)
+		return errors.WithStack(err)
 	}
 
 	gcs, err := storage.New(client)
 	if err != nil {
-		return nil, errors.WithStack(err)
+		return errors.WithStack(err)
 	}
 
-	return &objectStore{
-		gcs:            gcs,
-		googleAccessID: googleAccessID,
-		privateKey:     privateKey,
-	}, nil
+	o.gcs = gcs
+	o.googleAccessID = jwtConfig.Email
+	o.privateKey = jwtConfig.PrivateKey
+
+	return nil
 }
 
-func (op *objectStore) PutObject(bucket string, key string, body io.Reader) error {
+func (o *objectStore) PutObject(bucket string, key string, body io.Reader) error {
 	obj := &storage.Object{
 		Name: key,
 	}
 
-	_, err := op.gcs.Objects.Insert(bucket, obj).Media(body).Do()
+	_, err := o.gcs.Objects.Insert(bucket, obj).Media(body).Do()
 
 	return errors.WithStack(err)
 }
 
-func (op *objectStore) GetObject(bucket string, key string) (io.ReadCloser, error) {
-	res, err := op.gcs.Objects.Get(bucket, key).Download()
+func (o *objectStore) GetObject(bucket string, key string) (io.ReadCloser, error) {
+	res, err := o.gcs.Objects.Get(bucket, key).Download()
 	if err != nil {
 		return nil, errors.WithStack(err)
 	}
@@ -74,8 +103,8 @@ func (op *objectStore) GetObject(bucket string, key string) (io.ReadCloser, erro
 	return res.Body, nil
 }
 
-func (op *objectStore) ListCommonPrefixes(bucket string, delimiter string) ([]string, error) {
-	res, err := op.gcs.Objects.List(bucket).Delimiter(delimiter).Do()
+func (o *objectStore) ListCommonPrefixes(bucket string, delimiter string) ([]string, error) {
+	res, err := o.gcs.Objects.List(bucket).Delimiter(delimiter).Do()
 	if err != nil {
 		return nil, errors.WithStack(err)
 	}
@@ -90,8 +119,8 @@ func (op *objectStore) ListCommonPrefixes(bucket string, delimiter string) ([]st
 	return ret, nil
 }
 
-func (op *objectStore) ListObjects(bucket, prefix string) ([]string, error) {
-	res, err := op.gcs.Objects.List(bucket).Prefix(prefix).Do()
+func (o *objectStore) ListObjects(bucket, prefix string) ([]string, error) {
+	res, err := o.gcs.Objects.List(bucket).Prefix(prefix).Do()
 	if err != nil {
 		return nil, errors.WithStack(err)
 	}
@@ -104,21 +133,14 @@ func (op *objectStore) ListObjects(bucket, prefix string) ([]string, error) {
 	return ret, nil
 }
 
-func (op *objectStore) DeleteObject(bucket string, key string) error {
-	return errors.Wrapf(op.gcs.Objects.Delete(bucket, key).Do(), "error deleting object %s", key)
+func (o *objectStore) DeleteObject(bucket string, key string) error {
+	return errors.Wrapf(o.gcs.Objects.Delete(bucket, key).Do(), "error deleting object %s", key)
 }
 
-func (op *objectStore) CreateSignedURL(bucket, key string, ttl time.Duration) (string, error) {
-	if op.googleAccessID == "" {
-		return "", errors.New("unable to create a pre-signed URL - make sure GOOGLE_APPLICATION_CREDENTIALS points to a valid GCE service account file (missing email address)")
-	}
-	if len(op.privateKey) == 0 {
-		return "", errors.New("unable to create a pre-signed URL - make sure GOOGLE_APPLICATION_CREDENTIALS points to a valid GCE service account file (missing private key)")
-	}
-
+func (o *objectStore) CreateSignedURL(bucket, key string, ttl time.Duration) (string, error) {
 	return newstorage.SignedURL(bucket, key, &newstorage.SignedURLOptions{
-		GoogleAccessID: op.googleAccessID,
-		PrivateKey:     op.privateKey,
+		GoogleAccessID: o.googleAccessID,
+		PrivateKey:     o.privateKey,
 		Method:         "GET",
 		Expires:        time.Now().Add(ttl),
 	})
diff --git a/pkg/cloudprovider/storage_interfaces.go b/pkg/cloudprovider/storage_interfaces.go
index 0a78166f8..7f4a7bb38 100644
--- a/pkg/cloudprovider/storage_interfaces.go
+++ b/pkg/cloudprovider/storage_interfaces.go
@@ -24,6 +24,11 @@ import (
 // ObjectStore exposes basic object-storage operations required
 // by Ark.
 type ObjectStore interface {
+	// Init prepares the ObjectStore for usage using the provided map of
+	// configuration key-value pairs. It returns an error if the ObjectStore
+	// cannot be initialized from the provided config.
+	Init(config map[string]string) error
+
 	// PutObject creates a new object using the data in body within the specified
 	// object storage bucket with the given key.
 	PutObject(bucket string, key string, body io.Reader) error
@@ -51,6 +56,11 @@ type ObjectStore interface {
 // BlockStore exposes basic block-storage operations required
 // by Ark.
 type BlockStore interface {
+	// Init prepares the BlockStore for usage using the provided map of
+	// configuration key-value pairs. It returns an error if the BlockStore
+	// cannot be initialized from the provided config.
+	Init(config map[string]string) error
+
 	// CreateVolumeFromSnapshot creates a new block volume, initialized from the provided snapshot,
 	// and with the specified type and IOPS (if using provisioned IOPS).
 	CreateVolumeFromSnapshot(snapshotID, volumeType, volumeAZ string, iops *int64) (volumeID string, err error)
diff --git a/pkg/cmd/ark/ark.go b/pkg/cmd/ark/ark.go
index 0dffd21be..7c3137c65 100644
--- a/pkg/cmd/ark/ark.go
+++ b/pkg/cmd/ark/ark.go
@@ -29,6 +29,7 @@ import (
 	"github.com/heptio/ark/pkg/cmd/cli/restore"
 	"github.com/heptio/ark/pkg/cmd/cli/schedule"
 	"github.com/heptio/ark/pkg/cmd/server"
+	"github.com/heptio/ark/pkg/cmd/server/plugin"
 	"github.com/heptio/ark/pkg/cmd/version"
 )
 
@@ -57,6 +58,7 @@ operations can also be performed as 'ark backup get' and 'ark schedule create'.`
 		get.NewCommand(f),
 		describe.NewCommand(f),
 		create.NewCommand(f),
+		plugin.NewCommand(),
 	)
 
 	// add the glog flags
diff --git a/pkg/cmd/server/plugin/plugin.go b/pkg/cmd/server/plugin/plugin.go
new file mode 100644
index 000000000..3272f9ce2
--- /dev/null
+++ b/pkg/cmd/server/plugin/plugin.go
@@ -0,0 +1,86 @@
+/*
+Copyright 2017 Heptio Inc.
+
+Licensed under the Apache License, Version 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 permissions and
+limitations under the License.
+*/
+
+package plugin
+
+import (
+	"log"
+
+	plugin "github.com/hashicorp/go-plugin"
+	"github.com/spf13/cobra"
+
+	"github.com/heptio/ark/pkg/cloudprovider"
+	"github.com/heptio/ark/pkg/cloudprovider/aws"
+	"github.com/heptio/ark/pkg/cloudprovider/azure"
+	"github.com/heptio/ark/pkg/cloudprovider/gcp"
+	arkplugin "github.com/heptio/ark/pkg/plugin"
+)
+
+func NewCommand() *cobra.Command {
+	objectStores := map[string]cloudprovider.ObjectStore{
+		"aws":   aws.NewObjectStore(),
+		"gcp":   gcp.NewObjectStore(),
+		"azure": azure.NewObjectStore(),
+	}
+
+	blockStores := map[string]cloudprovider.BlockStore{
+		"aws":   aws.NewBlockStore(),
+		"gcp":   gcp.NewBlockStore(),
+		"azure": azure.NewBlockStore(),
+	}
+
+	c := &cobra.Command{
+		Use:    "plugin [KIND] [NAME]",
+		Hidden: true,
+		Short:  "INTERNAL COMMAND ONLY - not intended to be run directly by users",
+		Run: func(c *cobra.Command, args []string) {
+			if len(args) != 2 {
+				log.Fatalf("You must specify exactly two arguments, the plugin kind and the plugin name")
+			}
+
+			kind := args[0]
+			name := args[1]
+
+			log.Printf("Running plugin command for kind=%s, name=%s", kind, name)
+
+			switch kind {
+			case "cloudprovider":
+				objectStore, found := objectStores[name]
+				if !found {
+					log.Fatalf("Unrecognized plugin name %q", name)
+				}
+
+				blockStore, found := blockStores[name]
+				if !found {
+					log.Fatalf("Unrecognized plugin name %q", name)
+				}
+
+				plugin.Serve(&plugin.ServeConfig{
+					HandshakeConfig: arkplugin.Handshake,
+					Plugins: map[string]plugin.Plugin{
+						string(arkplugin.PluginKindObjectStore): arkplugin.NewObjectStorePlugin(objectStore),
+						string(arkplugin.PluginKindBlockStore):  arkplugin.NewBlockStorePlugin(blockStore),
+					},
+					GRPCServer: plugin.DefaultGRPCServer,
+				})
+			default:
+				log.Fatalf("Unsupported plugin kind %q", kind)
+			}
+		},
+	}
+
+	return c
+}
diff --git a/pkg/cmd/server/server.go b/pkg/cmd/server/server.go
index 1448e218b..f0235ceaf 100644
--- a/pkg/cmd/server/server.go
+++ b/pkg/cmd/server/server.go
@@ -19,8 +19,6 @@ package server
 import (
 	"context"
 	"fmt"
-	"io/ioutil"
-	"os"
 	"reflect"
 	"sort"
 	"strings"
@@ -30,7 +28,6 @@ import (
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
-	"golang.org/x/oauth2/google"
 
 	"k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
@@ -47,9 +44,6 @@ import (
 	"github.com/heptio/ark/pkg/backup"
 	"github.com/heptio/ark/pkg/client"
 	"github.com/heptio/ark/pkg/cloudprovider"
-	arkaws "github.com/heptio/ark/pkg/cloudprovider/aws"
-	"github.com/heptio/ark/pkg/cloudprovider/azure"
-	"github.com/heptio/ark/pkg/cloudprovider/gcp"
 	"github.com/heptio/ark/pkg/cmd"
 	"github.com/heptio/ark/pkg/cmd/util/flag"
 	"github.com/heptio/ark/pkg/controller"
@@ -57,6 +51,7 @@ import (
 	clientset "github.com/heptio/ark/pkg/generated/clientset/versioned"
 	arkv1client "github.com/heptio/ark/pkg/generated/clientset/versioned/typed/ark/v1"
 	informers "github.com/heptio/ark/pkg/generated/informers/externalversions"
+	"github.com/heptio/ark/pkg/plugin"
 	"github.com/heptio/ark/pkg/restore"
 	"github.com/heptio/ark/pkg/restore/restorers"
 	"github.com/heptio/ark/pkg/util/kube"
@@ -146,6 +141,7 @@ type server struct {
 	ctx                   context.Context
 	cancelFunc            context.CancelFunc
 	logger                *logrus.Logger
+	pluginManager         plugin.Manager
 }
 
 func newServer(kubeconfig, baseName string, logger *logrus.Logger) (*server, error) {
@@ -173,9 +169,10 @@ func newServer(kubeconfig, baseName string, logger *logrus.Logger) (*server, err
 		discoveryClient:       arkClient.Discovery(),
 		clientPool:            dynamic.NewDynamicClientPool(clientConfig),
 		sharedInformerFactory: informers.NewSharedInformerFactory(arkClient, 0),
-		ctx:        ctx,
-		cancelFunc: cancelFunc,
-		logger:     logger,
+		ctx:           ctx,
+		cancelFunc:    cancelFunc,
+		logger:        logger,
+		pluginManager: plugin.NewManager(logger, logger.Level),
 	}
 
 	return s, nil
@@ -325,7 +322,7 @@ func (s *server) watchConfig(config *api.Config) {
 
 func (s *server) initBackupService(config *api.Config) error {
 	s.logger.Info("Configuring cloud provider for backup service")
-	objectStore, err := getObjectStore(config.BackupStorageProvider.CloudProviderConfig, "backupStorageProvider", s.logger)
+	objectStore, err := getObjectStore(config.BackupStorageProvider.CloudProviderConfig, s.pluginManager)
 	if err != nil {
 		return err
 	}
@@ -341,7 +338,7 @@ func (s *server) initSnapshotService(config *api.Config) error {
 	}
 
 	s.logger.Info("Configuring cloud provider for snapshot service")
-	blockStore, err := getBlockStore(*config.PersistentVolumeProvider, "persistentVolumeProvider")
+	blockStore, err := getBlockStore(*config.PersistentVolumeProvider, s.pluginManager)
 	if err != nil {
 		return err
 	}
@@ -349,103 +346,37 @@ func (s *server) initSnapshotService(config *api.Config) error {
 	return nil
 }
 
-func hasOneCloudProvider(cloudConfig api.CloudProviderConfig) bool {
-	found := false
-
-	if cloudConfig.AWS != nil {
-		found = true
-	}
-
-	if cloudConfig.GCP != nil {
-		if found {
-			return false
-		}
-		found = true
-	}
-
-	if cloudConfig.Azure != nil {
-		if found {
-			return false
-		}
-		found = true
-	}
-
-	return found
-}
-
-func getObjectStore(cloudConfig api.CloudProviderConfig, field string, logger *logrus.Logger) (cloudprovider.ObjectStore, error) {
-	var (
-		objectStore cloudprovider.ObjectStore
-		err         error
-	)
-
-	if !hasOneCloudProvider(cloudConfig) {
-		return nil, errors.Errorf("you must specify exactly one of aws, gcp, or azure for %s", field)
-	}
-
-	switch {
-	case cloudConfig.AWS != nil:
-		objectStore, err = arkaws.NewObjectStore(
-			cloudConfig.AWS.Region,
-			cloudConfig.AWS.S3Url,
-			cloudConfig.AWS.KMSKeyID,
-			cloudConfig.AWS.S3ForcePathStyle)
-	case cloudConfig.GCP != nil:
-		var email string
-		var privateKey []byte
-
-		credentialsFile := os.Getenv("GOOGLE_APPLICATION_CREDENTIALS")
-		if credentialsFile != "" {
-			// Get the email and private key from the credentials file so we can pre-sign download URLs
-			creds, err := ioutil.ReadFile(credentialsFile)
-			if err != nil {
-				return nil, errors.WithStack(err)
-			}
-			jwtConfig, err := google.JWTConfigFromJSON(creds)
-			if err != nil {
-				return nil, errors.WithStack(err)
-			}
-			email = jwtConfig.Email
-			privateKey = jwtConfig.PrivateKey
-		} else {
-			logger.Warning("GOOGLE_APPLICATION_CREDENTIALS is undefined; some features such as downloading log files will not work")
-		}
-
-		objectStore, err = gcp.NewObjectStore(email, privateKey)
-	case cloudConfig.Azure != nil:
-		objectStore, err = azure.NewObjectStore()
+func getObjectStore(cloudConfig api.CloudProviderConfig, manager plugin.Manager) (cloudprovider.ObjectStore, error) {
+	if cloudConfig.Name == "" {
+		return nil, errors.New("object storage provider name must not be empty")
 	}
 
+	objectStore, err := manager.GetObjectStore(cloudConfig.Name)
 	if err != nil {
 		return nil, err
 	}
 
+	if err := objectStore.Init(cloudConfig.Config); err != nil {
+		return nil, err
+	}
+
 	return objectStore, nil
 }
 
-func getBlockStore(cloudConfig api.CloudProviderConfig, field string) (cloudprovider.BlockStore, error) {
-	var (
-		blockStore cloudprovider.BlockStore
-		err        error
-	)
-
-	if !hasOneCloudProvider(cloudConfig) {
-		return nil, errors.Errorf("you must specify exactly one of aws, gcp, or azure for %s", field)
-	}
-
-	switch {
-	case cloudConfig.AWS != nil:
-		blockStore, err = arkaws.NewBlockStore(cloudConfig.AWS.Region)
-	case cloudConfig.GCP != nil:
-		blockStore, err = gcp.NewBlockStore(cloudConfig.GCP.Project)
-	case cloudConfig.Azure != nil:
-		blockStore, err = azure.NewBlockStore(cloudConfig.Azure.Location, cloudConfig.Azure.APITimeout.Duration)
+func getBlockStore(cloudConfig api.CloudProviderConfig, manager plugin.Manager) (cloudprovider.BlockStore, error) {
+	if cloudConfig.Name == "" {
+		return nil, errors.New("block storage provider name must not be empty")
 	}
 
+	blockStore, err := manager.GetBlockStore(cloudConfig.Name)
 	if err != nil {
 		return nil, err
 	}
 
+	if err := blockStore.Init(cloudConfig.Config); err != nil {
+		return nil, err
+	}
+
 	return blockStore, nil
 }
 
diff --git a/pkg/plugin/block_store.go b/pkg/plugin/block_store.go
new file mode 100644
index 000000000..401f8c264
--- /dev/null
+++ b/pkg/plugin/block_store.go
@@ -0,0 +1,247 @@
+/*
+Copyright 2017 the Heptio Ark contributors.
+
+Licensed under the Apache License, Version 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 permissions and
+limitations under the License.
+*/
+
+package plugin
+
+import (
+	"github.com/hashicorp/go-plugin"
+	"golang.org/x/net/context"
+	"google.golang.org/grpc"
+
+	"github.com/heptio/ark/pkg/cloudprovider"
+	proto "github.com/heptio/ark/pkg/plugin/generated"
+)
+
+// BlockStorePlugin is an implementation of go-plugin's Plugin
+// interface with support for gRPC for the cloudprovider/BlockStore
+// interface.
+type BlockStorePlugin struct {
+	plugin.NetRPCUnsupportedPlugin
+
+	impl cloudprovider.BlockStore
+}
+
+// NewBlockStorePlugin constructs a BlockStorePlugin.
+func NewBlockStorePlugin(blockStore cloudprovider.BlockStore) *BlockStorePlugin {
+	return &BlockStorePlugin{
+		impl: blockStore,
+	}
+}
+
+// GRPCServer registers a BlockStore gRPC server.
+func (p *BlockStorePlugin) GRPCServer(s *grpc.Server) error {
+	proto.RegisterBlockStoreServer(s, &BlockStoreGRPCServer{impl: p.impl})
+	return nil
+}
+
+// GRPCClient returns a BlockStore gRPC client.
+func (p *BlockStorePlugin) GRPCClient(c *grpc.ClientConn) (interface{}, error) {
+	return &BlockStoreGRPCClient{grpcClient: proto.NewBlockStoreClient(c)}, nil
+}
+
+// BlockStoreGRPCClient implements the cloudprovider.BlockStore interface and uses a
+// gRPC client to make calls to the plugin server.
+type BlockStoreGRPCClient struct {
+	grpcClient proto.BlockStoreClient
+}
+
+// Init prepares the BlockStore for usage using the provided map of
+// configuration key-value pairs. It returns an error if the BlockStore
+// cannot be initialized from the provided config.
+func (c *BlockStoreGRPCClient) Init(config map[string]string) error {
+	_, err := c.grpcClient.Init(context.Background(), &proto.InitRequest{Config: config})
+
+	return err
+}
+
+// CreateVolumeFromSnapshot creates a new block volume, initialized from the provided snapshot,
+// and with the specified type and IOPS (if using provisioned IOPS).
+func (c *BlockStoreGRPCClient) CreateVolumeFromSnapshot(snapshotID, volumeType, volumeAZ string, iops *int64) (string, error) {
+	req := &proto.CreateVolumeRequest{
+		SnapshotID: snapshotID,
+		VolumeType: volumeType,
+		VolumeAZ:   volumeAZ,
+	}
+
+	if iops == nil {
+		req.Iops = 0
+	} else {
+		req.Iops = *iops
+	}
+
+	res, err := c.grpcClient.CreateVolumeFromSnapshot(context.Background(), req)
+	if err != nil {
+		return "", err
+	}
+
+	return res.VolumeID, nil
+}
+
+// GetVolumeInfo returns the type and IOPS (if using provisioned IOPS) for a specified block
+// volume.
+func (c *BlockStoreGRPCClient) GetVolumeInfo(volumeID, volumeAZ string) (string, *int64, error) {
+	res, err := c.grpcClient.GetVolumeInfo(context.Background(), &proto.GetVolumeInfoRequest{VolumeID: volumeID, VolumeAZ: volumeAZ})
+	if err != nil {
+		return "", nil, err
+	}
+
+	var iops *int64
+	if res.Iops != 0 {
+		iops = &res.Iops
+	}
+
+	return res.VolumeType, iops, nil
+}
+
+// IsVolumeReady returns whether the specified volume is ready to be used.
+func (c *BlockStoreGRPCClient) IsVolumeReady(volumeID, volumeAZ string) (bool, error) {
+	res, err := c.grpcClient.IsVolumeReady(context.Background(), &proto.IsVolumeReadyRequest{VolumeID: volumeID, VolumeAZ: volumeAZ})
+	if err != nil {
+		return false, err
+	}
+
+	return res.Ready, nil
+}
+
+// ListSnapshots returns a list of all snapshots matching the specified set of tag key/values.
+func (c *BlockStoreGRPCClient) ListSnapshots(tagFilters map[string]string) ([]string, error) {
+	res, err := c.grpcClient.ListSnapshots(context.Background(), &proto.ListSnapshotsRequest{TagFilters: tagFilters})
+	if err != nil {
+		return nil, err
+	}
+
+	return res.SnapshotIDs, nil
+}
+
+// CreateSnapshot creates a snapshot of the specified block volume, and applies the provided
+// set of tags to the snapshot.
+func (c *BlockStoreGRPCClient) CreateSnapshot(volumeID, volumeAZ string, tags map[string]string) (string, error) {
+	req := &proto.CreateSnapshotRequest{
+		VolumeID: volumeID,
+		VolumeAZ: volumeAZ,
+		Tags:     tags,
+	}
+
+	res, err := c.grpcClient.CreateSnapshot(context.Background(), req)
+	if err != nil {
+		return "", err
+	}
+
+	return res.SnapshotID, nil
+}
+
+// DeleteSnapshot deletes the specified volume snapshot.
+func (c *BlockStoreGRPCClient) DeleteSnapshot(snapshotID string) error {
+	_, err := c.grpcClient.DeleteSnapshot(context.Background(), &proto.DeleteSnapshotRequest{SnapshotID: snapshotID})
+
+	return err
+}
+
+// BlockStoreGRPCServer implements the proto-generated BlockStoreServer interface, and accepts
+// gRPC calls and forwards them to an implementation of the pluggable interface.
+type BlockStoreGRPCServer struct {
+	impl cloudprovider.BlockStore
+}
+
+// Init prepares the BlockStore for usage using the provided map of
+// configuration key-value pairs. It returns an error if the BlockStore
+// cannot be initialized from the provided config.
+func (s *BlockStoreGRPCServer) Init(ctx context.Context, req *proto.InitRequest) (*proto.Empty, error) {
+	if err := s.impl.Init(req.Config); err != nil {
+		return nil, err
+	}
+
+	return &proto.Empty{}, nil
+}
+
+// CreateVolumeFromSnapshot creates a new block volume, initialized from the provided snapshot,
+// and with the specified type and IOPS (if using provisioned IOPS).
+func (s *BlockStoreGRPCServer) CreateVolumeFromSnapshot(ctx context.Context, req *proto.CreateVolumeRequest) (*proto.CreateVolumeResponse, error) {
+	snapshotID := req.SnapshotID
+	volumeType := req.VolumeType
+	volumeAZ := req.VolumeAZ
+	var iops *int64
+
+	if req.Iops != 0 {
+		iops = &req.Iops
+	}
+
+	volumeID, err := s.impl.CreateVolumeFromSnapshot(snapshotID, volumeType, volumeAZ, iops)
+	if err != nil {
+		return nil, err
+	}
+
+	return &proto.CreateVolumeResponse{VolumeID: volumeID}, nil
+}
+
+// GetVolumeInfo returns the type and IOPS (if using provisioned IOPS) for a specified block
+// volume.
+func (s *BlockStoreGRPCServer) GetVolumeInfo(ctx context.Context, req *proto.GetVolumeInfoRequest) (*proto.GetVolumeInfoResponse, error) {
+	volumeType, iops, err := s.impl.GetVolumeInfo(req.VolumeID, req.VolumeAZ)
+	if err != nil {
+		return nil, err
+	}
+
+	res := &proto.GetVolumeInfoResponse{
+		VolumeType: volumeType,
+	}
+
+	if iops != nil {
+		res.Iops = *iops
+	}
+
+	return res, nil
+}
+
+// IsVolumeReady returns whether the specified volume is ready to be used.
+func (s *BlockStoreGRPCServer) IsVolumeReady(ctx context.Context, req *proto.IsVolumeReadyRequest) (*proto.IsVolumeReadyResponse, error) {
+	ready, err := s.impl.IsVolumeReady(req.VolumeID, req.VolumeAZ)
+	if err != nil {
+		return nil, err
+	}
+
+	return &proto.IsVolumeReadyResponse{Ready: ready}, nil
+}
+
+// ListSnapshots returns a list of all snapshots matching the specified set of tag key/values.
+func (s *BlockStoreGRPCServer) ListSnapshots(ctx context.Context, req *proto.ListSnapshotsRequest) (*proto.ListSnapshotsResponse, error) {
+	snapshotIDs, err := s.impl.ListSnapshots(req.TagFilters)
+	if err != nil {
+		return nil, err
+	}
+
+	return &proto.ListSnapshotsResponse{SnapshotIDs: snapshotIDs}, nil
+}
+
+// CreateSnapshot creates a snapshot of the specified block volume, and applies the provided
+// set of tags to the snapshot.
+func (s *BlockStoreGRPCServer) CreateSnapshot(ctx context.Context, req *proto.CreateSnapshotRequest) (*proto.CreateSnapshotResponse, error) {
+	snapshotID, err := s.impl.CreateSnapshot(req.VolumeID, req.VolumeAZ, req.Tags)
+	if err != nil {
+		return nil, err
+	}
+
+	return &proto.CreateSnapshotResponse{SnapshotID: snapshotID}, nil
+}
+
+// DeleteSnapshot deletes the specified volume snapshot.
+func (s *BlockStoreGRPCServer) DeleteSnapshot(ctx context.Context, req *proto.DeleteSnapshotRequest) (*proto.Empty, error) {
+	if err := s.impl.DeleteSnapshot(req.SnapshotID); err != nil {
+		return nil, err
+	}
+
+	return &proto.Empty{}, nil
+}
diff --git a/pkg/plugin/logrus_adapter.go b/pkg/plugin/logrus_adapter.go
new file mode 100644
index 000000000..a8b3bc256
--- /dev/null
+++ b/pkg/plugin/logrus_adapter.go
@@ -0,0 +1,158 @@
+/*
+Copyright 2017 the Heptio Ark contributors.
+
+Licensed under the Apache License, Version 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 permissions and
+limitations under the License.
+*/
+
+package plugin
+
+import (
+	"fmt"
+	"log"
+
+	hclog "github.com/hashicorp/go-hclog"
+	"github.com/sirupsen/logrus"
+)
+
+const pluginNameField = "pluginName"
+
+// logrusAdapter implements the hclog.Logger interface and
+// delegates all calls to a logrus logger.
+type logrusAdapter struct {
+	impl  logrus.FieldLogger
+	level logrus.Level
+	name  string
+}
+
+// args are alternating key, value pairs, where the keys
+// are expected to be strings, and values can be any type.
+func argsToFields(args ...interface{}) logrus.Fields {
+	fields := make(map[string]interface{})
+
+	for i := 0; i < len(args); i += 2 {
+		switch args[i] {
+		case "time", "timestamp", "level":
+			// remove `time` & `timestamp` because this info will be added
+			// by the Ark logger and we don't want to have duplicated
+			// fields.
+			//
+			// remove `level` because it'll be added by the Ark logger based
+			// on the call we make (and go-plugin is determining which level
+			// to log at based on the hclog-compatible `@level` field which
+			// we're adding via HcLogLevelHook).
+		default:
+			var val interface{}
+			if i+1 < len(args) {
+				val = args[i+1]
+			}
+
+			fields[fmt.Sprintf("%v", args[i])] = val
+		}
+	}
+
+	return logrus.Fields(fields)
+}
+
+// Trace emits a message and key/value pairs at the DEBUG level
+// (logrus doesn't have a TRACE level)
+func (l *logrusAdapter) Trace(msg string, args ...interface{}) {
+	l.Debug(msg, args...)
+}
+
+// Debug emits a message and key/value pairs at the DEBUG level
+func (l *logrusAdapter) Debug(msg string, args ...interface{}) {
+	l.impl.WithFields(argsToFields(args...)).Debug(msg)
+}
+
+// Info emits a message and key/value pairs at the INFO level
+func (l *logrusAdapter) Info(msg string, args ...interface{}) {
+	l.impl.WithFields(argsToFields(args...)).Info(msg)
+}
+
+// Warn emits a message and key/value pairs at the WARN level
+func (l *logrusAdapter) Warn(msg string, args ...interface{}) {
+	l.impl.WithFields(argsToFields(args...)).Warn(msg)
+}
+
+// Error emits a message and key/value pairs at the ERROR level
+func (l *logrusAdapter) Error(msg string, args ...interface{}) {
+	l.impl.WithFields(argsToFields(args...)).Error(msg)
+}
+
+// IsTrace indicates if TRACE logs would be emitted. This and the other Is* guards
+// are used to elide expensive logging code based on the current level.
+func (l *logrusAdapter) IsTrace() bool {
+	return l.IsDebug()
+}
+
+// IsDebug indicates if DEBUG logs would be emitted. This and the other Is* guards
+// are used to elide expensive logging code based on the current level.
+func (l *logrusAdapter) IsDebug() bool {
+	return l.level <= logrus.DebugLevel
+}
+
+// IsInfo indicates if INFO logs would be emitted. This and the other Is* guards
+// are used to elide expensive logging code based on the current level.
+func (l *logrusAdapter) IsInfo() bool {
+	return l.level <= logrus.InfoLevel
+}
+
+// IsWarn indicates if WARN logs would be emitted. This and the other Is* guards
+// are used to elide expensive logging code based on the current level.
+func (l *logrusAdapter) IsWarn() bool {
+	return l.level <= logrus.WarnLevel
+}
+
+// IsError indicates if ERROR logs would be emitted. This and the other Is* guards
+// are used to elide expensive logging code based on the current level.
+func (l *logrusAdapter) IsError() bool {
+	return l.level <= logrus.ErrorLevel
+}
+
+// With creates a sublogger that will always have the given key/value pairs
+func (l *logrusAdapter) With(args ...interface{}) hclog.Logger {
+	return &logrusAdapter{
+		impl:  l.impl.WithFields(argsToFields(args...)),
+		level: l.level,
+	}
+}
+
+// Named creates a logger that will add a `pluginName` field with the name string
+// as the value. If the logger already has a name, the new value will be appended
+// to the current name.
+func (l *logrusAdapter) Named(name string) hclog.Logger {
+	var newName string
+	if l.name == "" {
+		newName = name
+	} else {
+		newName = l.name + "." + name
+	}
+
+	return l.ResetNamed(newName)
+}
+
+// ResetNamed creates a logger that will add a `pluginName` field with the name string
+// as the value. This sets the name of the logger to the value directly, unlike `Named`
+// which appends the given value to the current name.
+func (l *logrusAdapter) ResetNamed(name string) hclog.Logger {
+	return &logrusAdapter{
+		impl:  l.impl.WithField(pluginNameField, name),
+		level: l.level,
+		name:  name,
+	}
+}
+
+// StandardLogger returns a value that conforms to the stdlib log.Logger interface
+func (l *logrusAdapter) StandardLogger(opts *hclog.StandardLoggerOptions) *log.Logger {
+	panic("not implemented")
+}
diff --git a/pkg/plugin/logrus_adapter_test.go b/pkg/plugin/logrus_adapter_test.go
new file mode 100644
index 000000000..98e2872ef
--- /dev/null
+++ b/pkg/plugin/logrus_adapter_test.go
@@ -0,0 +1,55 @@
+package plugin
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/sirupsen/logrus"
+)
+
+func TestArgsToFields(t *testing.T) {
+	tests := []struct {
+		name           string
+		args           []interface{}
+		expectedFields logrus.Fields
+	}{
+		{
+			name:           "empty args results in empty map of fields",
+			args:           []interface{}{},
+			expectedFields: logrus.Fields(map[string]interface{}{}),
+		},
+		{
+			name: "matching string keys/values are correctly set as fields",
+			args: []interface{}{"key-1", "value-1", "key-2", "value-2"},
+			expectedFields: logrus.Fields(map[string]interface{}{
+				"key-1": "value-1",
+				"key-2": "value-2",
+			}),
+		},
+		{
+			name: "time/timestamp/level entries are removed",
+			args: []interface{}{"time", time.Now(), "key-1", "value-1", "timestamp", time.Now(), "key-2", "value-2", "level", "WARN"},
+			expectedFields: logrus.Fields(map[string]interface{}{
+				"key-1": "value-1",
+				"key-2": "value-2",
+			}),
+		},
+		{
+			name: "odd number of args adds the last arg as a field with a nil value",
+			args: []interface{}{"key-1", "value-1", "key-2", "value-2", "key-3"},
+			expectedFields: logrus.Fields(map[string]interface{}{
+				"key-1": "value-1",
+				"key-2": "value-2",
+				"key-3": nil,
+			}),
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			assert.Equal(t, test.expectedFields, argsToFields(test.args...))
+		})
+	}
+}
diff --git a/pkg/plugin/manager.go b/pkg/plugin/manager.go
new file mode 100644
index 000000000..cff434467
--- /dev/null
+++ b/pkg/plugin/manager.go
@@ -0,0 +1,185 @@
+/*
+Copyright 2017 the Heptio Ark contributors.
+
+Licensed under the Apache License, Version 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 permissions and
+limitations under the License.
+*/
+
+package plugin
+
+import (
+	"fmt"
+	"os"
+	"os/exec"
+	"path/filepath"
+
+	"github.com/hashicorp/go-hclog"
+	plugin "github.com/hashicorp/go-plugin"
+	"github.com/pkg/errors"
+	"github.com/sirupsen/logrus"
+
+	"github.com/heptio/ark/pkg/cloudprovider"
+)
+
+// PluginKind is a type alias for a string that describes
+// the kind of an Ark-supported plugin.
+type PluginKind string
+
+func (k PluginKind) String() string {
+	return string(k)
+}
+
+const (
+	// PluginKindObjectStore is the Kind string for
+	// an Object Store plugin.
+	PluginKindObjectStore PluginKind = "objectstore"
+
+	// PluginKindBlockStore is the Kind string for
+	// a Block Store plugin.
+	PluginKindBlockStore PluginKind = "blockstore"
+
+	pluginDir = "/plugins"
+)
+
+type pluginInfo struct {
+	kind PluginKind
+	name string
+}
+
+// Manager exposes functions for getting implementations of the pluggable
+// Ark interfaces.
+type Manager interface {
+	// GetObjectStore returns the plugin implementation of the
+	// cloudprovider.ObjectStore interface with the specified name.
+	GetObjectStore(name string) (cloudprovider.ObjectStore, error)
+
+	// GetBlockStore returns the plugin implementation of the
+	// cloudprovider.BlockStore interface with the specified name.
+	GetBlockStore(name string) (cloudprovider.BlockStore, error)
+}
+
+type manager struct {
+	logger          hclog.Logger
+	clients         map[pluginInfo]*plugin.Client
+	internalPlugins map[pluginInfo]interface{}
+}
+
+// NewManager constructs a manager for getting plugin implementations.
+func NewManager(logger logrus.FieldLogger, level logrus.Level) Manager {
+	return &manager{
+		logger:  (&logrusAdapter{impl: logger, level: level}),
+		clients: make(map[pluginInfo]*plugin.Client),
+		internalPlugins: map[pluginInfo]interface{}{
+			{kind: PluginKindObjectStore, name: "aws"}: struct{}{},
+			{kind: PluginKindBlockStore, name: "aws"}:  struct{}{},
+
+			{kind: PluginKindObjectStore, name: "gcp"}: struct{}{},
+			{kind: PluginKindBlockStore, name: "gcp"}:  struct{}{},
+
+			{kind: PluginKindObjectStore, name: "azure"}: struct{}{},
+			{kind: PluginKindBlockStore, name: "azure"}:  struct{}{},
+		},
+	}
+}
+
+func addPlugins(config *plugin.ClientConfig, kinds ...PluginKind) {
+	for _, kind := range kinds {
+		if kind == PluginKindObjectStore {
+			config.Plugins[kind.String()] = &ObjectStorePlugin{}
+		} else if kind == PluginKindBlockStore {
+			config.Plugins[kind.String()] = &BlockStorePlugin{}
+		}
+	}
+}
+
+func (m *manager) getPlugin(descriptor pluginInfo, logger hclog.Logger) (interface{}, error) {
+	client, found := m.clients[descriptor]
+	if !found {
+		var (
+			externalPath = filepath.Join(pluginDir, fmt.Sprintf("ark-%s-%s", descriptor.kind, descriptor.name))
+			config       = &plugin.ClientConfig{
+				HandshakeConfig:  Handshake,
+				AllowedProtocols: []plugin.Protocol{plugin.ProtocolGRPC},
+				Plugins:          make(map[string]plugin.Plugin),
+				Logger:           logger,
+			}
+		)
+
+		// First check to see if there's an external plugin for this kind and name. this
+		// is so users can override the built-in plugins if they want. If it doesn't exist,
+		// see if there's an internal one.
+		if _, err := os.Stat(externalPath); err == nil {
+			addPlugins(config, descriptor.kind)
+			config.Cmd = exec.Command(externalPath)
+
+			client = plugin.NewClient(config)
+
+			m.clients[descriptor] = client
+		} else if _, found := m.internalPlugins[descriptor]; found {
+			addPlugins(config, PluginKindObjectStore, PluginKindBlockStore)
+			config.Cmd = exec.Command("/ark", "plugin", "cloudprovider", descriptor.name)
+
+			client = plugin.NewClient(config)
+
+			// since a single sub-process will serve both an object and block store
+			// for a given cloud-provider, record this client as being valid for both
+			m.clients[pluginInfo{PluginKindObjectStore, descriptor.name}] = client
+			m.clients[pluginInfo{PluginKindBlockStore, descriptor.name}] = client
+		} else {
+			return nil, errors.Errorf("plugin not found for kind=%s, name=%s", descriptor.kind, descriptor.name)
+		}
+	}
+
+	protocolClient, err := client.Client()
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
+
+	plugin, err := protocolClient.Dispense(descriptor.kind.String())
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
+
+	return plugin, nil
+}
+
+// GetObjectStore returns the plugin implementation of the cloudprovider.ObjectStore
+// interface with the specified name.
+func (m *manager) GetObjectStore(name string) (cloudprovider.ObjectStore, error) {
+	pluginObj, err := m.getPlugin(pluginInfo{PluginKindObjectStore, name}, m.logger)
+	if err != nil {
+		return nil, err
+	}
+
+	objStore, ok := pluginObj.(cloudprovider.ObjectStore)
+	if !ok {
+		return nil, errors.New("could not convert gRPC client to cloudprovider.ObjectStore")
+	}
+
+	return objStore, nil
+}
+
+// GetBlockStore returns the plugin implementation of the cloudprovider.BlockStore
+// interface with the specified name.
+func (m *manager) GetBlockStore(name string) (cloudprovider.BlockStore, error) {
+	pluginObj, err := m.getPlugin(pluginInfo{PluginKindBlockStore, name}, m.logger)
+	if err != nil {
+		return nil, err
+	}
+
+	blockStore, ok := pluginObj.(cloudprovider.BlockStore)
+	if !ok {
+		return nil, errors.New("could not convert gRPC client to cloudprovider.BlockStore")
+	}
+
+	return blockStore, nil
+}
diff --git a/pkg/plugin/object_store.go b/pkg/plugin/object_store.go
new file mode 100644
index 000000000..17108bedc
--- /dev/null
+++ b/pkg/plugin/object_store.go
@@ -0,0 +1,291 @@
+/*
+Copyright 2017 the Heptio Ark contributors.
+
+Licensed under the Apache License, Version 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 permissions and
+limitations under the License.
+*/
+
+package plugin
+
+import (
+	"io"
+	"time"
+
+	"github.com/hashicorp/go-plugin"
+	"golang.org/x/net/context"
+	"google.golang.org/grpc"
+
+	"github.com/heptio/ark/pkg/cloudprovider"
+	proto "github.com/heptio/ark/pkg/plugin/generated"
+)
+
+const byteChunkSize = 16384
+
+// ObjectStorePlugin is an implementation of go-plugin's Plugin
+// interface with support for gRPC for the cloudprovider/ObjectStore
+// interface.
+type ObjectStorePlugin struct {
+	plugin.NetRPCUnsupportedPlugin
+
+	impl cloudprovider.ObjectStore
+}
+
+// NewObjectStorePlugin construct an ObjectStorePlugin.
+func NewObjectStorePlugin(objectStore cloudprovider.ObjectStore) *ObjectStorePlugin {
+	return &ObjectStorePlugin{
+		impl: objectStore,
+	}
+}
+
+// GRPCServer registers an ObjectStore gRPC server.
+func (p *ObjectStorePlugin) GRPCServer(s *grpc.Server) error {
+	proto.RegisterObjectStoreServer(s, &ObjectStoreGRPCServer{impl: p.impl})
+	return nil
+}
+
+// GRPCClient returns an ObjectStore gRPC client.
+func (p *ObjectStorePlugin) GRPCClient(c *grpc.ClientConn) (interface{}, error) {
+	return &ObjectStoreGRPCClient{grpcClient: proto.NewObjectStoreClient(c)}, nil
+}
+
+// ObjectStoreGRPCClient implements the cloudprovider.ObjectStore interface and uses a
+// gRPC client to make calls to the plugin server.
+type ObjectStoreGRPCClient struct {
+	grpcClient proto.ObjectStoreClient
+}
+
+// Init prepares the ObjectStore for usage using the provided map of
+// configuration key-value pairs. It returns an error if the ObjectStore
+// cannot be initialized from the provided config.
+func (c *ObjectStoreGRPCClient) Init(config map[string]string) error {
+	_, err := c.grpcClient.Init(context.Background(), &proto.InitRequest{Config: config})
+
+	return err
+}
+
+// PutObject creates a new object using the data in body within the specified
+// object storage bucket with the given key.
+func (c *ObjectStoreGRPCClient) PutObject(bucket, key string, body io.Reader) error {
+	stream, err := c.grpcClient.PutObject(context.Background())
+	if err != nil {
+		return err
+	}
+
+	// read from the provider io.Reader into chunks, and send each one over
+	// the gRPC stream
+	chunk := make([]byte, byteChunkSize)
+	for {
+		n, err := body.Read(chunk)
+		if err == io.EOF {
+			_, resErr := stream.CloseAndRecv()
+			return resErr
+		}
+		if err != nil {
+			stream.CloseSend()
+			return err
+		}
+
+		if err := stream.Send(&proto.PutObjectRequest{Bucket: bucket, Key: key, Body: chunk[0:n]}); err != nil {
+			return err
+		}
+	}
+}
+
+// GetObject retrieves the object with the given key from the specified
+// bucket in object storage.
+func (c *ObjectStoreGRPCClient) GetObject(bucket, key string) (io.ReadCloser, error) {
+	stream, err := c.grpcClient.GetObject(context.Background(), &proto.GetObjectRequest{Bucket: bucket, Key: key})
+	if err != nil {
+		return nil, err
+	}
+
+	receive := func() ([]byte, error) {
+		data, err := stream.Recv()
+		if err != nil {
+			return nil, err
+		}
+
+		return data.Data, nil
+	}
+
+	close := func() error {
+		return stream.CloseSend()
+	}
+
+	return &StreamReadCloser{receive: receive, close: close}, nil
+}
+
+// ListCommonPrefixes gets a list of all object key prefixes that come
+// before the provided delimiter (this is often used to simulate a directory
+// hierarchy in object storage).
+func (c *ObjectStoreGRPCClient) ListCommonPrefixes(bucket, delimiter string) ([]string, error) {
+	res, err := c.grpcClient.ListCommonPrefixes(context.Background(), &proto.ListCommonPrefixesRequest{Bucket: bucket, Delimiter: delimiter})
+	if err != nil {
+		return nil, err
+	}
+
+	return res.Prefixes, nil
+}
+
+// ListObjects gets a list of all objects in bucket that have the same prefix.
+func (c *ObjectStoreGRPCClient) ListObjects(bucket, prefix string) ([]string, error) {
+	res, err := c.grpcClient.ListObjects(context.Background(), &proto.ListObjectsRequest{Bucket: bucket, Prefix: prefix})
+	if err != nil {
+		return nil, err
+	}
+
+	return res.Keys, nil
+}
+
+// DeleteObject removes object with the specified key from the given
+// bucket.
+func (c *ObjectStoreGRPCClient) DeleteObject(bucket, key string) error {
+	_, err := c.grpcClient.DeleteObject(context.Background(), &proto.DeleteObjectRequest{Bucket: bucket, Key: key})
+
+	return err
+}
+
+// CreateSignedURL creates a pre-signed URL for the given bucket and key that expires after ttl.
+func (c *ObjectStoreGRPCClient) CreateSignedURL(bucket, key string, ttl time.Duration) (string, error) {
+	res, err := c.grpcClient.CreateSignedURL(context.Background(), &proto.CreateSignedURLRequest{
+		Bucket: bucket,
+		Key:    key,
+		Ttl:    int64(ttl),
+	})
+	if err != nil {
+		return "", nil
+	}
+
+	return res.Url, nil
+}
+
+// ObjectStoreGRPCServer implements the proto-generated ObjectStoreServer interface, and accepts
+// gRPC calls and forwards them to an implementation of the pluggable interface.
+type ObjectStoreGRPCServer struct {
+	impl cloudprovider.ObjectStore
+}
+
+// Init prepares the ObjectStore for usage using the provided map of
+// configuration key-value pairs. It returns an error if the ObjectStore
+// cannot be initialized from the provided config.
+func (s *ObjectStoreGRPCServer) Init(ctx context.Context, req *proto.InitRequest) (*proto.Empty, error) {
+	if err := s.impl.Init(req.Config); err != nil {
+		return nil, err
+	}
+
+	return &proto.Empty{}, nil
+}
+
+// PutObject creates a new object using the data in body within the specified
+// object storage bucket with the given key.
+func (s *ObjectStoreGRPCServer) PutObject(stream proto.ObjectStore_PutObjectServer) error {
+	// we need to read the first chunk ahead of time to get the bucket and key;
+	// in our receive method, we'll use `first` on the first call
+	firstChunk, err := stream.Recv()
+	if err != nil {
+		return err
+	}
+
+	bucket := firstChunk.Bucket
+	key := firstChunk.Key
+
+	receive := func() ([]byte, error) {
+		if firstChunk != nil {
+			res := firstChunk.Body
+			firstChunk = nil
+			return res, nil
+		}
+
+		data, err := stream.Recv()
+		if err != nil {
+			return nil, err
+		}
+		return data.Body, nil
+	}
+
+	close := func() error {
+		return nil
+	}
+
+	if err := s.impl.PutObject(bucket, key, &StreamReadCloser{receive: receive, close: close}); err != nil {
+		return err
+	}
+
+	return stream.SendAndClose(&proto.Empty{})
+}
+
+// GetObject retrieves the object with the given key from the specified
+// bucket in object storage.
+func (s *ObjectStoreGRPCServer) GetObject(req *proto.GetObjectRequest, stream proto.ObjectStore_GetObjectServer) error {
+	rdr, err := s.impl.GetObject(req.Bucket, req.Key)
+	if err != nil {
+		return err
+	}
+
+	chunk := make([]byte, byteChunkSize)
+	for {
+		n, err := rdr.Read(chunk)
+		if err != nil && err != io.EOF {
+			return err
+		}
+		if n == 0 {
+			return nil
+		}
+
+		if err := stream.Send(&proto.Bytes{Data: chunk[0:n]}); err != nil {
+			return err
+		}
+	}
+}
+
+// ListCommonPrefixes gets a list of all object key prefixes that come
+// before the provided delimiter (this is often used to simulate a directory
+// hierarchy in object storage).
+func (s *ObjectStoreGRPCServer) ListCommonPrefixes(ctx context.Context, req *proto.ListCommonPrefixesRequest) (*proto.ListCommonPrefixesResponse, error) {
+	prefixes, err := s.impl.ListCommonPrefixes(req.Bucket, req.Delimiter)
+	if err != nil {
+		return nil, err
+	}
+
+	return &proto.ListCommonPrefixesResponse{Prefixes: prefixes}, nil
+}
+
+// ListObjects gets a list of all objects in bucket that have the same prefix.
+func (s *ObjectStoreGRPCServer) ListObjects(ctx context.Context, req *proto.ListObjectsRequest) (*proto.ListObjectsResponse, error) {
+	keys, err := s.impl.ListObjects(req.Bucket, req.Prefix)
+	if err != nil {
+		return nil, err
+	}
+
+	return &proto.ListObjectsResponse{Keys: keys}, nil
+}
+
+// DeleteObject removes object with the specified key from the given
+// bucket.
+func (s *ObjectStoreGRPCServer) DeleteObject(ctx context.Context, req *proto.DeleteObjectRequest) (*proto.Empty, error) {
+	err := s.impl.DeleteObject(req.Bucket, req.Key)
+	if err != nil {
+		return nil, err
+	}
+
+	return &proto.Empty{}, nil
+}
+
+// CreateSignedURL creates a pre-signed URL for the given bucket and key that expires after ttl.
+func (s *ObjectStoreGRPCServer) CreateSignedURL(ctx context.Context, req *proto.CreateSignedURLRequest) (*proto.CreateSignedURLResponse, error) {
+	url, err := s.impl.CreateSignedURL(req.Bucket, req.Key, time.Duration(req.Ttl))
+	if err != nil {
+		return nil, err
+	}
+
+	return &proto.CreateSignedURLResponse{Url: url}, nil
+}
diff --git a/pkg/plugin/shared.go b/pkg/plugin/shared.go
new file mode 100644
index 000000000..0b92621cb
--- /dev/null
+++ b/pkg/plugin/shared.go
@@ -0,0 +1,27 @@
+/*
+Copyright 2017 the Heptio Ark contributors.
+
+Licensed under the Apache License, Version 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 permissions and
+limitations under the License.
+*/
+
+package plugin
+
+import plugin "github.com/hashicorp/go-plugin"
+
+// Handshake is configuration information that allows go-plugin
+// clients and servers to perform a handshake.
+var Handshake = plugin.HandshakeConfig{
+	ProtocolVersion:  1,
+	MagicCookieKey:   "ARK_PLUGIN",
+	MagicCookieValue: "hello",
+}
diff --git a/pkg/util/test/object_store.go b/pkg/util/test/object_store.go
index 6a41df60b..57616167e 100644
--- a/pkg/util/test/object_store.go
+++ b/pkg/util/test/object_store.go
@@ -84,6 +84,20 @@ func (_m *ObjectStore) GetObject(bucket string, key string) (io.ReadCloser, erro
 	return r0, r1
 }
 
+// Init provides a mock function with given fields: config
+func (_m *ObjectStore) Init(config map[string]string) error {
+	ret := _m.Called(config)
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(map[string]string) error); ok {
+		r0 = rf(config)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
 // ListCommonPrefixes provides a mock function with given fields: bucket, delimiter
 func (_m *ObjectStore) ListCommonPrefixes(bucket string, delimiter string) ([]string, error) {
 	ret := _m.Called(bucket, delimiter)