mirrors/velero
1
0
Fork 0
mirror of https://github.com/vmware-tanzu/velero.git synced 2026-08-18 21:26:08 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

switch built-in cloud providers to run as plugins

Browse Source 
Signed-off-by: Steve Kriss <steve@heptio.com>
This commit is contained in:
Steve Kriss
2017-11-14 09:47:36 -08:00
parent 3975187d57
commit 24ce316788
19 changed files with 1350 additions and 368 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/apis/ark/v1/config.go
+3 -29
View File
@@ -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"`
}
pkg/apis/ark/v1/zz_generated.deepcopy.go
+5 -86
View File
@@ -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
pkg/cloudprovider/aws/block_store.go
+31 -24
View File
@@ -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)
}
pkg/cloudprovider/aws/object_store.go
+48 -22
View File
@@ -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),
})
pkg/cloudprovider/azure/block_store.go
+60 -42
View File
@@ -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
pkg/cloudprovider/azure/object_store.go
+21 -19
View File
@@ -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
}
pkg/cloudprovider/gcp/block_store.go
+33 -25
View File
@@ -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)
}
pkg/cloudprovider/gcp/object_store.go
+50 -28
View File
@@ -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),
})
pkg/cloudprovider/storage_interfaces.go
+10
View File
@@ -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)
pkg/cmd/ark/ark.go
+2
View File
@@ -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
pkg/cmd/server/plugin/plugin.go
+86
View File
@@ -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
}
pkg/cmd/server/server.go
+24 -93
View File
@@ -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
}
pkg/plugin/block_store.go
+247
View File
@@ -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
}
pkg/plugin/logrus_adapter.go
+158
View File
@@ -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")
}
pkg/plugin/logrus_adapter_test.go
+55
View File
@@ -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...))
})
}
}
pkg/plugin/manager.go
+185
View File
@@ -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
}
pkg/plugin/object_store.go
+291
View File
@@ -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
}
pkg/plugin/shared.go
+27
View File
@@ -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",
}
pkg/util/test/object_store.go
+14
View File
@@ -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)