Extract providers (#1985)

* Remove cloud providers and reorg code

Signed-off-by: Carlisia <carlisia@vmware.com>

* Update dependencies

Signed-off-by: Carlisia <carlisia@vmware.com>

* Fix tests

Signed-off-by: Carlisia <carlisia@vmware.com>

* fix dependency issues

Signed-off-by: Carlisia <carlisia@vmware.com>

* Delete dup test

Signed-off-by: Carlisia <carlisia@vmware.com>

* Add back spaces to file

Signed-off-by: Carlisia <carlisia@vmware.com>

* Remove and update docs

Signed-off-by: Carlisia <carlisia@vmware.com>

* Make the plugins flag required

Signed-off-by: Carlisia <carlisia@vmware.com>

* Add changelog

Signed-off-by: Carlisia <carlisia@vmware.com>

* Make the plugins flag conditional

Signed-off-by: Carlisia <carlisia@vmware.com>
This commit is contained in:
KubeKween
2019-10-22 15:31:27 -07:00
committed by Adnan Abdulhussein
parent 69f993aebd
commit d26bf05b33
214 changed files with 441 additions and 250600 deletions
-66
View File
@@ -1,66 +0,0 @@
/*
Copyright 2018 the Velero 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 aws
import (
"context"
"net/url"
"github.com/aws/aws-sdk-go/aws/endpoints"
"github.com/aws/aws-sdk-go/aws/session"
"github.com/aws/aws-sdk-go/service/s3/s3manager"
"github.com/pkg/errors"
)
// GetBucketRegion returns the AWS region that a bucket is in, or an error
// if the region cannot be determined.
func GetBucketRegion(bucket string) (string, error) {
var region string
session, err := session.NewSession()
if err != nil {
return "", errors.WithStack(err)
}
for _, partition := range endpoints.DefaultPartitions() {
for regionHint := range partition.Regions() {
region, _ = s3manager.GetBucketRegion(context.Background(), session, bucket, regionHint)
// we only need to try a single region hint per partition, so break after the first
break
}
if region != "" {
return region, nil
}
}
return "", errors.New("unable to determine bucket's region")
}
// IsValidS3URLScheme returns true if the scheme is http:// or https://
// and the url parses correctly, otherwise, return false
func IsValidS3URLScheme(s3URL string) bool {
u, err := url.Parse(s3URL)
if err != nil {
return false
}
if u.Scheme != "http" && u.Scheme != "https" {
return false
}
return true
}
-29
View File
@@ -1,29 +0,0 @@
/*
Copyright 2018 the Velero 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 aws
import (
"testing"
"github.com/stretchr/testify/assert"
)
func TestS3URL(t *testing.T) {
assert.True(t, IsValidS3URLScheme("http://foo"))
assert.True(t, IsValidS3URLScheme("https://foo"))
assert.False(t, IsValidS3URLScheme("httpd://foo"))
assert.False(t, IsValidS3URLScheme(""))
}
-363
View File
@@ -1,363 +0,0 @@
/*
Copyright 2017, 2019 the Velero 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 aws
import (
"crypto/tls"
"io"
"net/http"
"sort"
"strconv"
"time"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/aws/awserr"
"github.com/aws/aws-sdk-go/aws/endpoints"
"github.com/aws/aws-sdk-go/aws/request"
v4 "github.com/aws/aws-sdk-go/aws/signer/v4"
"github.com/aws/aws-sdk-go/service/s3"
"github.com/aws/aws-sdk-go/service/s3/s3manager"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
"github.com/vmware-tanzu/velero/pkg/plugin/framework"
)
const (
s3URLKey = "s3Url"
publicURLKey = "publicUrl"
kmsKeyIDKey = "kmsKeyId"
s3ForcePathStyleKey = "s3ForcePathStyle"
bucketKey = "bucket"
signatureVersionKey = "signatureVersion"
credentialProfileKey = "profile"
serverSideEncryptionKey = "serverSideEncryption"
insecureSkipTLSVerifyKey = "insecureSkipTLSVerify"
)
type s3Interface interface {
HeadObject(input *s3.HeadObjectInput) (*s3.HeadObjectOutput, error)
GetObject(input *s3.GetObjectInput) (*s3.GetObjectOutput, error)
ListObjectsV2Pages(input *s3.ListObjectsV2Input, fn func(*s3.ListObjectsV2Output, bool) bool) error
DeleteObject(input *s3.DeleteObjectInput) (*s3.DeleteObjectOutput, error)
GetObjectRequest(input *s3.GetObjectInput) (req *request.Request, output *s3.GetObjectOutput)
}
type ObjectStore struct {
log logrus.FieldLogger
s3 s3Interface
preSignS3 s3Interface
s3Uploader *s3manager.Uploader
kmsKeyID string
signatureVersion string
serverSideEncryption string
}
func NewObjectStore(logger logrus.FieldLogger) *ObjectStore {
return &ObjectStore{log: logger}
}
func isValidSignatureVersion(signatureVersion string) bool {
switch signatureVersion {
case "1", "4":
return true
}
return false
}
func (o *ObjectStore) Init(config map[string]string) error {
if err := framework.ValidateObjectStoreConfigKeys(config,
regionKey,
s3URLKey,
publicURLKey,
kmsKeyIDKey,
s3ForcePathStyleKey,
signatureVersionKey,
credentialProfileKey,
serverSideEncryptionKey,
insecureSkipTLSVerifyKey,
); err != nil {
return err
}
var (
region = config[regionKey]
s3URL = config[s3URLKey]
publicURL = config[publicURLKey]
kmsKeyID = config[kmsKeyIDKey]
s3ForcePathStyleVal = config[s3ForcePathStyleKey]
signatureVersion = config[signatureVersionKey]
credentialProfile = config[credentialProfileKey]
serverSideEncryption = config[serverSideEncryptionKey]
insecureSkipTLSVerifyVal = config[insecureSkipTLSVerifyKey]
// note that bucket is automatically added to the config map
// by the server from the ObjectStorageProviderConfig so
// doesn't need to be explicitly set by the user within
// config.
bucket = config[bucketKey]
s3ForcePathStyle bool
insecureSkipTLSVerify bool
err error
)
if s3ForcePathStyleVal != "" {
if s3ForcePathStyle, err = strconv.ParseBool(s3ForcePathStyleVal); err != nil {
return errors.Wrapf(err, "could not parse %s (expected bool)", s3ForcePathStyleKey)
}
}
// AWS (not an alternate S3-compatible API) and region not
// explicitly specified: determine the bucket's region
if s3URL == "" && region == "" {
var err error
region, err = GetBucketRegion(bucket)
if err != nil {
return err
}
}
serverConfig, err := newAWSConfig(s3URL, region, s3ForcePathStyle)
if err != nil {
return err
}
if insecureSkipTLSVerifyVal != "" {
if insecureSkipTLSVerify, err = strconv.ParseBool(insecureSkipTLSVerifyVal); err != nil {
return errors.Wrapf(err, "could not parse %s (expected bool)", insecureSkipTLSVerifyKey)
}
}
if insecureSkipTLSVerify {
serverConfig.HTTPClient = &http.Client{
Transport: &http.Transport{
TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
},
}
}
serverSession, err := getSession(serverConfig, credentialProfile)
if err != nil {
return err
}
o.s3 = s3.New(serverSession)
o.s3Uploader = s3manager.NewUploader(serverSession)
o.kmsKeyID = kmsKeyID
o.serverSideEncryption = serverSideEncryption
if signatureVersion != "" {
if !isValidSignatureVersion(signatureVersion) {
return errors.Errorf("invalid signature version: %s", signatureVersion)
}
o.signatureVersion = signatureVersion
}
if publicURL != "" {
publicConfig, err := newAWSConfig(publicURL, region, s3ForcePathStyle)
if err != nil {
return err
}
publicSession, err := getSession(publicConfig, credentialProfile)
if err != nil {
return err
}
o.preSignS3 = s3.New(publicSession)
} else {
o.preSignS3 = o.s3
}
return nil
}
func newAWSConfig(url, region string, forcePathStyle bool) (*aws.Config, error) {
awsConfig := aws.NewConfig().
WithRegion(region).
WithS3ForcePathStyle(forcePathStyle)
if url != "" {
if !IsValidS3URLScheme(url) {
return nil, errors.Errorf("Invalid s3 url %s, URL must be valid according to https://golang.org/pkg/net/url/#Parse and start with http:// or https://", url)
}
awsConfig = awsConfig.WithEndpointResolver(
endpoints.ResolverFunc(func(service, region string, optFns ...func(*endpoints.Options)) (endpoints.ResolvedEndpoint, error) {
if service == endpoints.S3ServiceID {
return endpoints.ResolvedEndpoint{
URL: url,
}, nil
}
return endpoints.DefaultResolver().EndpointFor(service, region, optFns...)
}),
)
}
return awsConfig, nil
}
func (o *ObjectStore) PutObject(bucket, key string, body io.Reader) error {
req := &s3manager.UploadInput{
Bucket: &bucket,
Key: &key,
Body: body,
}
switch {
// if kmsKeyID is not empty, assume a server-side encryption (SSE)
// algorithm of "aws:kms"
case o.kmsKeyID != "":
req.ServerSideEncryption = aws.String("aws:kms")
req.SSEKMSKeyId = &o.kmsKeyID
// otherwise, use the SSE algorithm specified, if any
case o.serverSideEncryption != "":
req.ServerSideEncryption = aws.String(o.serverSideEncryption)
}
_, err := o.s3Uploader.Upload(req)
return errors.Wrapf(err, "error putting object %s", key)
}
const notFoundCode = "NotFound"
// ObjectExists checks if there is an object with the given key in the object storage bucket.
func (o *ObjectStore) ObjectExists(bucket, key string) (bool, error) {
log := o.log.WithFields(
logrus.Fields{
"bucket": bucket,
"key": key,
},
)
req := &s3.HeadObjectInput{
Bucket: aws.String(bucket),
Key: aws.String(key),
}
log.Debug("Checking if object exists")
if _, err := o.s3.HeadObject(req); err != nil {
log.Debug("Checking for AWS specific error information")
if aerr, ok := err.(awserr.Error); ok {
log.WithFields(
logrus.Fields{
"code": aerr.Code(),
"message": aerr.Message(),
},
).Debugf("awserr.Error contents (origErr=%v)", aerr.OrigErr())
// The code will be NotFound if the key doesn't exist.
// See https://github.com/aws/aws-sdk-go/issues/1208 and https://github.com/aws/aws-sdk-go/pull/1213.
log.Debugf("Checking for code=%s", notFoundCode)
if aerr.Code() == notFoundCode {
log.Debug("Object doesn't exist - got not found")
return false, nil
}
}
return false, errors.WithStack(err)
}
log.Debug("Object exists")
return true, nil
}
func (o *ObjectStore) GetObject(bucket, key string) (io.ReadCloser, error) {
req := &s3.GetObjectInput{
Bucket: &bucket,
Key: &key,
}
res, err := o.s3.GetObject(req)
if err != nil {
return nil, errors.Wrapf(err, "error getting object %s", key)
}
return res.Body, nil
}
func (o *ObjectStore) ListCommonPrefixes(bucket, prefix, delimiter string) ([]string, error) {
req := &s3.ListObjectsV2Input{
Bucket: &bucket,
Prefix: &prefix,
Delimiter: &delimiter,
}
var ret []string
err := o.s3.ListObjectsV2Pages(req, func(page *s3.ListObjectsV2Output, lastPage bool) bool {
for _, prefix := range page.CommonPrefixes {
ret = append(ret, *prefix.Prefix)
}
return !lastPage
})
if err != nil {
return nil, errors.WithStack(err)
}
return ret, nil
}
func (o *ObjectStore) ListObjects(bucket, prefix string) ([]string, error) {
req := &s3.ListObjectsV2Input{
Bucket: &bucket,
Prefix: &prefix,
}
var ret []string
err := o.s3.ListObjectsV2Pages(req, func(page *s3.ListObjectsV2Output, lastPage bool) bool {
for _, obj := range page.Contents {
ret = append(ret, *obj.Key)
}
return !lastPage
})
if err != nil {
return nil, errors.WithStack(err)
}
// ensure that returned objects are in a consistent order so that the deletion logic deletes the objects before
// the pseudo-folder prefix object for s3 providers (such as Quobyte) that return the pseudo-folder as an object.
// See https://github.com/vmware-tanzu/velero/pull/999
sort.Sort(sort.Reverse(sort.StringSlice(ret)))
return ret, nil
}
func (o *ObjectStore) DeleteObject(bucket, key string) error {
req := &s3.DeleteObjectInput{
Bucket: &bucket,
Key: &key,
}
_, err := o.s3.DeleteObject(req)
return errors.Wrapf(err, "error deleting object %s", key)
}
func (o *ObjectStore) CreateSignedURL(bucket, key string, ttl time.Duration) (string, error) {
req, _ := o.preSignS3.GetObjectRequest(&s3.GetObjectInput{
Bucket: aws.String(bucket),
Key: aws.String(key),
})
if o.signatureVersion == "1" {
req.Handlers.Sign.Remove(v4.SignRequestHandler)
req.Handlers.Sign.PushBackNamed(v1SignRequestHandler)
}
return req.Presign(ttl)
}
-125
View File
@@ -1,125 +0,0 @@
/*
Copyright 2018, 2019 the Velero 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 aws
import (
"testing"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/aws/awserr"
"github.com/aws/aws-sdk-go/aws/request"
"github.com/aws/aws-sdk-go/service/s3"
"github.com/pkg/errors"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/mock"
"github.com/stretchr/testify/require"
"github.com/vmware-tanzu/velero/pkg/test"
)
func TestIsValidSignatureVersion(t *testing.T) {
assert.True(t, isValidSignatureVersion("1"))
assert.True(t, isValidSignatureVersion("4"))
assert.False(t, isValidSignatureVersion("3"))
}
type mockS3 struct {
mock.Mock
}
func (m *mockS3) HeadObject(input *s3.HeadObjectInput) (*s3.HeadObjectOutput, error) {
args := m.Called(input)
return args.Get(0).(*s3.HeadObjectOutput), args.Error(1)
}
func (m *mockS3) GetObject(input *s3.GetObjectInput) (*s3.GetObjectOutput, error) {
args := m.Called(input)
return args.Get(0).(*s3.GetObjectOutput), args.Error(1)
}
func (m *mockS3) ListObjectsV2Pages(input *s3.ListObjectsV2Input, fn func(*s3.ListObjectsV2Output, bool) bool) error {
args := m.Called(input, fn)
return args.Error(0)
}
func (m *mockS3) DeleteObject(input *s3.DeleteObjectInput) (*s3.DeleteObjectOutput, error) {
args := m.Called(input)
return args.Get(0).(*s3.DeleteObjectOutput), args.Error(1)
}
func (m *mockS3) GetObjectRequest(input *s3.GetObjectInput) (req *request.Request, output *s3.GetObjectOutput) {
args := m.Called(input)
return args.Get(0).(*request.Request), args.Get(1).(*s3.GetObjectOutput)
}
func TestObjectExists(t *testing.T) {
tests := []struct {
name string
errorResponse error
expectedExists bool
expectedError string
}{
{
name: "exists",
errorResponse: nil,
expectedExists: true,
},
{
name: "doesn't exist",
errorResponse: awserr.New(s3.ErrCodeNoSuchKey, "no such key", nil),
expectedExists: false,
expectedError: "NoSuchKey: no such key",
},
{
name: "error checking for existence",
errorResponse: errors.Errorf("bad"),
expectedExists: false,
expectedError: "bad",
},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
s := new(mockS3)
defer s.AssertExpectations(t)
o := &ObjectStore{
log: test.NewLogger(),
s3: s,
}
bucket := "b"
key := "k"
req := &s3.HeadObjectInput{
Bucket: aws.String(bucket),
Key: aws.String(key),
}
s.On("HeadObject", req).Return(&s3.HeadObjectOutput{}, tc.errorResponse)
exists, err := o.ObjectExists(bucket, key)
if tc.expectedError != "" {
assert.EqualError(t, err, tc.expectedError)
return
}
require.NoError(t, err)
assert.Equal(t, tc.expectedExists, exists)
})
}
}
@@ -1,147 +0,0 @@
/*
Copyright 2018 the Velero 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 aws
import (
"crypto/hmac"
"crypto/sha1"
"encoding/base64"
"fmt"
"net/url"
"strconv"
"strings"
"time"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/aws/credentials"
"github.com/aws/aws-sdk-go/aws/request"
"github.com/pkg/errors"
)
var (
errInvalidMethod = errors.New("v1 signer only handles HTTP GET")
)
type signer struct {
// Values that must be populated from the request
request *request.Request
time time.Time
credentials *credentials.Credentials
debug aws.LogLevelType
logger aws.Logger
query url.Values
stringToSign string
signature string
}
// SignRequestHandler is a named request handler the SDK will use to sign
// service client request with using the V4 signature.
var v1SignRequestHandler = request.NamedHandler{
Name: "v1.SignRequestHandler", Fn: signSDKRequest,
}
func signSDKRequest(req *request.Request) {
// If the request does not need to be signed ignore the signing of the
// request if the AnonymousCredentials object is used.
if req.Config.Credentials == credentials.AnonymousCredentials {
return
}
if req.HTTPRequest.Method != "GET" {
// The V1 signer only supports GET
req.Error = errInvalidMethod
return
}
v1 := signer{
request: req,
time: req.Time,
credentials: req.Config.Credentials,
debug: req.Config.LogLevel.Value(),
logger: req.Config.Logger,
}
req.Error = v1.sign()
if req.Error != nil {
return
}
req.HTTPRequest.URL.RawQuery = v1.query.Encode()
}
func (v1 *signer) sign() error {
credentialsValue, err := v1.credentials.Get()
if err != nil {
return errors.Wrap(err, "error getting credentials")
}
httpRequest := v1.request.HTTPRequest
v1.query = httpRequest.URL.Query()
// Set new query parameters
v1.query.Set("AWSAccessKeyId", credentialsValue.AccessKeyID)
if credentialsValue.SessionToken != "" {
v1.query.Set("SecurityToken", credentialsValue.SessionToken)
}
// in case this is a retry, ensure no signature present
v1.query.Del("Signature")
method := httpRequest.Method
path := httpRequest.URL.Path
if path == "" {
path = "/"
}
duration := int64(v1.request.ExpireTime / time.Second)
expires := strconv.FormatInt(duration, 10)
// build the canonical string for the v1 signature
v1.stringToSign = strings.Join([]string{
method,
"",
"",
expires,
path,
}, "\n")
hash := hmac.New(sha1.New, []byte(credentialsValue.SecretAccessKey))
hash.Write([]byte(v1.stringToSign))
v1.signature = base64.StdEncoding.EncodeToString(hash.Sum(nil))
v1.query.Set("Signature", v1.signature)
v1.query.Set("Expires", expires)
if v1.debug.Matches(aws.LogDebugWithSigning) {
v1.logSigningInfo()
}
return nil
}
const logSignInfoMsg = `DEBUG: Request Signature:
---[ STRING TO SIGN ]--------------------------------
%s
---[ SIGNATURE ]-------------------------------------
%s
-----------------------------------------------------`
func (v1 *signer) logSigningInfo() {
msg := fmt.Sprintf(logSignInfoMsg, v1.stringToSign, v1.query.Get("Signature"))
v1.logger.Log(msg)
}
-308
View File
@@ -1,308 +0,0 @@
/*
Copyright 2017, 2019 the Velero 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 aws
import (
"fmt"
"os"
"regexp"
"strings"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/aws/awserr"
"github.com/aws/aws-sdk-go/aws/session"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
v1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/util/sets"
"github.com/vmware-tanzu/velero/pkg/plugin/framework"
)
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 VolumeSnapshotter struct {
log logrus.FieldLogger
ec2 *ec2.EC2
}
// takes AWS credential config & a profile to create a new session
func getSession(config *aws.Config, profile string) (*session.Session, error) {
sessionOptions := session.Options{Config: *config, Profile: profile}
sess, err := session.NewSessionWithOptions(sessionOptions)
if err != nil {
return nil, errors.WithStack(err)
}
if _, err := sess.Config.Credentials.Get(); err != nil {
return nil, errors.WithStack(err)
}
return sess, nil
}
func NewVolumeSnapshotter(logger logrus.FieldLogger) *VolumeSnapshotter {
return &VolumeSnapshotter{log: logger}
}
func (b *VolumeSnapshotter) Init(config map[string]string) error {
if err := framework.ValidateVolumeSnapshotterConfigKeys(config, regionKey, credentialProfileKey); err != nil {
return err
}
region := config[regionKey]
credentialProfile := config[credentialProfileKey]
if region == "" {
return errors.Errorf("missing %s in aws configuration", regionKey)
}
awsConfig := aws.NewConfig().WithRegion(region)
sess, err := getSession(awsConfig, credentialProfile)
if err != nil {
return err
}
b.ec2 = ec2.New(sess)
return nil
}
func (b *VolumeSnapshotter) CreateVolumeFromSnapshot(snapshotID, volumeType, volumeAZ string, iops *int64) (volumeID string, err error) {
// describe the snapshot so we can apply its tags to the volume
snapReq := &ec2.DescribeSnapshotsInput{
SnapshotIds: []*string{&snapshotID},
}
snapRes, err := b.ec2.DescribeSnapshots(snapReq)
if err != nil {
return "", errors.WithStack(err)
}
if count := len(snapRes.Snapshots); count != 1 {
return "", errors.Errorf("expected 1 snapshot from DescribeSnapshots for %s, got %v", snapshotID, count)
}
// filter tags through getTagsForCluster() function in order to apply
// proper ownership tags to restored volumes
req := &ec2.CreateVolumeInput{
SnapshotId: &snapshotID,
AvailabilityZone: &volumeAZ,
VolumeType: &volumeType,
Encrypted: snapRes.Snapshots[0].Encrypted,
TagSpecifications: []*ec2.TagSpecification{
{
ResourceType: aws.String(ec2.ResourceTypeVolume),
Tags: getTagsForCluster(snapRes.Snapshots[0].Tags),
},
},
}
if iopsVolumeTypes.Has(volumeType) && iops != nil {
req.Iops = iops
}
res, err := b.ec2.CreateVolume(req)
if err != nil {
return "", errors.WithStack(err)
}
return *res.VolumeId, nil
}
func (b *VolumeSnapshotter) GetVolumeInfo(volumeID, volumeAZ string) (string, *int64, error) {
volumeInfo, err := b.describeVolume(volumeID)
if err != nil {
return "", nil, err
}
var (
volumeType string
iops *int64
)
if volumeInfo.VolumeType != nil {
volumeType = *volumeInfo.VolumeType
}
if iopsVolumeTypes.Has(volumeType) && volumeInfo.Iops != nil {
iops = volumeInfo.Iops
}
return volumeType, iops, nil
}
func (b *VolumeSnapshotter) describeVolume(volumeID string) (*ec2.Volume, error) {
req := &ec2.DescribeVolumesInput{
VolumeIds: []*string{&volumeID},
}
res, err := b.ec2.DescribeVolumes(req)
if err != nil {
return nil, errors.WithStack(err)
}
if count := len(res.Volumes); count != 1 {
return nil, errors.Errorf("Expected one volume from DescribeVolumes for volume ID %v, got %v", volumeID, count)
}
return res.Volumes[0], nil
}
func (b *VolumeSnapshotter) CreateSnapshot(volumeID, volumeAZ string, tags map[string]string) (string, error) {
// describe the volume so we can copy its tags to the snapshot
volumeInfo, err := b.describeVolume(volumeID)
if err != nil {
return "", err
}
res, err := b.ec2.CreateSnapshot(&ec2.CreateSnapshotInput{
VolumeId: &volumeID,
TagSpecifications: []*ec2.TagSpecification{
{
ResourceType: aws.String(ec2.ResourceTypeSnapshot),
Tags: getTags(tags, volumeInfo.Tags),
},
},
})
if err != nil {
return "", errors.WithStack(err)
}
return *res.SnapshotId, nil
}
func getTagsForCluster(snapshotTags []*ec2.Tag) []*ec2.Tag {
var result []*ec2.Tag
clusterName, haveAWSClusterNameEnvVar := os.LookupEnv("AWS_CLUSTER_NAME")
if haveAWSClusterNameEnvVar {
result = append(result, ec2Tag("kubernetes.io/cluster/"+clusterName, "owned"))
result = append(result, ec2Tag("KubernetesCluster", clusterName))
}
for _, tag := range snapshotTags {
if haveAWSClusterNameEnvVar && (strings.HasPrefix(*tag.Key, "kubernetes.io/cluster/") || *tag.Key == "KubernetesCluster") {
// if the AWS_CLUSTER_NAME variable is found we want current cluster
// to overwrite the old ownership on volumes
continue
}
result = append(result, ec2Tag(*tag.Key, *tag.Value))
}
return result
}
func getTags(veleroTags map[string]string, volumeTags []*ec2.Tag) []*ec2.Tag {
var result []*ec2.Tag
// set Velero-assigned tags
for k, v := range veleroTags {
result = append(result, ec2Tag(k, v))
}
// copy tags from volume to snapshot
for _, tag := range volumeTags {
// we want current Velero-assigned tags to overwrite any older versions
// of them that may exist due to prior snapshots/restores
if _, found := veleroTags[*tag.Key]; found {
continue
}
result = append(result, ec2Tag(*tag.Key, *tag.Value))
}
return result
}
func ec2Tag(key, val string) *ec2.Tag {
return &ec2.Tag{Key: &key, Value: &val}
}
func (b *VolumeSnapshotter) DeleteSnapshot(snapshotID string) error {
req := &ec2.DeleteSnapshotInput{
SnapshotId: &snapshotID,
}
_, err := b.ec2.DeleteSnapshot(req)
// if it's a NotFound error, we don't need to return an error
// since the snapshot is not there.
// see https://docs.aws.amazon.com/AWSEC2/latest/APIReference/errors-overview.html
if awsErr, ok := err.(awserr.Error); ok && awsErr.Code() == "InvalidSnapshot.NotFound" {
return nil
}
if err != nil {
return errors.WithStack(err)
}
return nil
}
var ebsVolumeIDRegex = regexp.MustCompile("vol-.*")
func (b *VolumeSnapshotter) GetVolumeID(unstructuredPV runtime.Unstructured) (string, error) {
pv := new(v1.PersistentVolume)
if err := runtime.DefaultUnstructuredConverter.FromUnstructured(unstructuredPV.UnstructuredContent(), pv); err != nil {
return "", errors.WithStack(err)
}
if pv.Spec.AWSElasticBlockStore == nil {
return "", nil
}
if pv.Spec.AWSElasticBlockStore.VolumeID == "" {
return "", errors.New("spec.awsElasticBlockStore.volumeID not found")
}
return ebsVolumeIDRegex.FindString(pv.Spec.AWSElasticBlockStore.VolumeID), nil
}
func (b *VolumeSnapshotter) SetVolumeID(unstructuredPV runtime.Unstructured, volumeID string) (runtime.Unstructured, error) {
pv := new(v1.PersistentVolume)
if err := runtime.DefaultUnstructuredConverter.FromUnstructured(unstructuredPV.UnstructuredContent(), pv); err != nil {
return nil, errors.WithStack(err)
}
if pv.Spec.AWSElasticBlockStore == nil {
return nil, errors.New("spec.awsElasticBlockStore not found")
}
pvFailureDomainZone := pv.Labels["failure-domain.beta.kubernetes.io/zone"]
if len(pvFailureDomainZone) > 0 {
pv.Spec.AWSElasticBlockStore.VolumeID = fmt.Sprintf("aws://%s/%s", pvFailureDomainZone, volumeID)
} else {
pv.Spec.AWSElasticBlockStore.VolumeID = volumeID
}
res, err := runtime.DefaultUnstructuredConverter.ToUnstructured(pv)
if err != nil {
return nil, errors.WithStack(err)
}
return &unstructured.Unstructured{Object: res}, nil
}
@@ -1,298 +0,0 @@
/*
Copyright 2017 the Velero 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 aws
import (
"os"
"sort"
"testing"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
v1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
"k8s.io/apimachinery/pkg/runtime"
)
func TestGetVolumeID(t *testing.T) {
b := &VolumeSnapshotter{}
pv := &unstructured.Unstructured{
Object: map[string]interface{}{},
}
// missing spec.awsElasticBlockStore -> no error
volumeID, err := b.GetVolumeID(pv)
require.NoError(t, err)
assert.Equal(t, "", volumeID)
// missing spec.awsElasticBlockStore.volumeID -> error
aws := map[string]interface{}{}
pv.Object["spec"] = map[string]interface{}{
"awsElasticBlockStore": aws,
}
volumeID, err = b.GetVolumeID(pv)
assert.Error(t, err)
assert.Equal(t, "", volumeID)
// regex miss
aws["volumeID"] = "foo"
volumeID, err = b.GetVolumeID(pv)
assert.NoError(t, err)
assert.Equal(t, "", volumeID)
// regex match 1
aws["volumeID"] = "aws://us-east-1c/vol-abc123"
volumeID, err = b.GetVolumeID(pv)
assert.NoError(t, err)
assert.Equal(t, "vol-abc123", volumeID)
// regex match 2
aws["volumeID"] = "vol-abc123"
volumeID, err = b.GetVolumeID(pv)
assert.NoError(t, err)
assert.Equal(t, "vol-abc123", volumeID)
}
func TestSetVolumeID(t *testing.T) {
b := &VolumeSnapshotter{}
pv := &unstructured.Unstructured{
Object: map[string]interface{}{},
}
// missing spec.awsElasticBlockStore -> error
updatedPV, err := b.SetVolumeID(pv, "vol-updated")
require.Error(t, err)
// happy path
aws := map[string]interface{}{}
pv.Object["spec"] = map[string]interface{}{
"awsElasticBlockStore": aws,
}
labels := map[string]interface{}{
"failure-domain.beta.kubernetes.io/zone": "us-east-1a",
}
pv.Object["metadata"] = map[string]interface{}{
"labels": labels,
}
updatedPV, err = b.SetVolumeID(pv, "vol-updated")
require.NoError(t, err)
res := new(v1.PersistentVolume)
require.NoError(t, runtime.DefaultUnstructuredConverter.FromUnstructured(updatedPV.UnstructuredContent(), res))
require.NotNil(t, res.Spec.AWSElasticBlockStore)
assert.Equal(t, "aws://us-east-1a/vol-updated", res.Spec.AWSElasticBlockStore.VolumeID)
}
func TestSetVolumeIDNoZone(t *testing.T) {
b := &VolumeSnapshotter{}
pv := &unstructured.Unstructured{
Object: map[string]interface{}{},
}
// missing spec.awsElasticBlockStore -> error
updatedPV, err := b.SetVolumeID(pv, "vol-updated")
require.Error(t, err)
// happy path
aws := map[string]interface{}{}
pv.Object["spec"] = map[string]interface{}{
"awsElasticBlockStore": aws,
}
updatedPV, err = b.SetVolumeID(pv, "vol-updated")
require.NoError(t, err)
res := new(v1.PersistentVolume)
require.NoError(t, runtime.DefaultUnstructuredConverter.FromUnstructured(updatedPV.UnstructuredContent(), res))
require.NotNil(t, res.Spec.AWSElasticBlockStore)
assert.Equal(t, "vol-updated", res.Spec.AWSElasticBlockStore.VolumeID)
}
func TestGetTagsForCluster(t *testing.T) {
tests := []struct {
name string
isNameSet bool
snapshotTags []*ec2.Tag
expected []*ec2.Tag
}{
{
name: "degenerate case (no tags)",
isNameSet: false,
snapshotTags: nil,
expected: nil,
},
{
name: "cluster tags exist and remain set",
isNameSet: false,
snapshotTags: []*ec2.Tag{
ec2Tag("KubernetesCluster", "old-cluster"),
ec2Tag("kubernetes.io/cluster/old-cluster", "owned"),
ec2Tag("aws-key", "aws-val"),
},
expected: []*ec2.Tag{
ec2Tag("KubernetesCluster", "old-cluster"),
ec2Tag("kubernetes.io/cluster/old-cluster", "owned"),
ec2Tag("aws-key", "aws-val"),
},
},
{
name: "cluster tags only get applied",
isNameSet: true,
snapshotTags: nil,
expected: []*ec2.Tag{
ec2Tag("KubernetesCluster", "current-cluster"),
ec2Tag("kubernetes.io/cluster/current-cluster", "owned"),
},
},
{
name: "non-overlaping cluster and snapshot tags both get applied",
isNameSet: true,
snapshotTags: []*ec2.Tag{ec2Tag("aws-key", "aws-val")},
expected: []*ec2.Tag{
ec2Tag("KubernetesCluster", "current-cluster"),
ec2Tag("kubernetes.io/cluster/current-cluster", "owned"),
ec2Tag("aws-key", "aws-val"),
},
},
{name: "overlaping cluster tags, current cluster tags take precedence",
isNameSet: true,
snapshotTags: []*ec2.Tag{
ec2Tag("KubernetesCluster", "old-name"),
ec2Tag("kubernetes.io/cluster/old-name", "owned"),
ec2Tag("aws-key", "aws-val"),
},
expected: []*ec2.Tag{
ec2Tag("KubernetesCluster", "current-cluster"),
ec2Tag("kubernetes.io/cluster/current-cluster", "owned"),
ec2Tag("aws-key", "aws-val"),
},
},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
if test.isNameSet {
os.Setenv("AWS_CLUSTER_NAME", "current-cluster")
}
res := getTagsForCluster(test.snapshotTags)
sort.Slice(res, func(i, j int) bool {
return *res[i].Key < *res[j].Key
})
sort.Slice(test.expected, func(i, j int) bool {
return *test.expected[i].Key < *test.expected[j].Key
})
assert.Equal(t, test.expected, res)
if test.isNameSet {
os.Unsetenv("AWS_CLUSTER_NAME")
}
})
}
}
func TestGetTags(t *testing.T) {
tests := []struct {
name string
veleroTags map[string]string
volumeTags []*ec2.Tag
expected []*ec2.Tag
}{
{
name: "degenerate case (no tags)",
veleroTags: nil,
volumeTags: nil,
expected: nil,
},
{
name: "velero tags only get applied",
veleroTags: map[string]string{
"velero-key1": "velero-val1",
"velero-key2": "velero-val2",
},
volumeTags: nil,
expected: []*ec2.Tag{
ec2Tag("velero-key1", "velero-val1"),
ec2Tag("velero-key2", "velero-val2"),
},
},
{
name: "volume tags only get applied",
veleroTags: nil,
volumeTags: []*ec2.Tag{
ec2Tag("aws-key1", "aws-val1"),
ec2Tag("aws-key2", "aws-val2"),
},
expected: []*ec2.Tag{
ec2Tag("aws-key1", "aws-val1"),
ec2Tag("aws-key2", "aws-val2"),
},
},
{
name: "non-overlapping velero and volume tags both get applied",
veleroTags: map[string]string{"velero-key": "velero-val"},
volumeTags: []*ec2.Tag{ec2Tag("aws-key", "aws-val")},
expected: []*ec2.Tag{
ec2Tag("velero-key", "velero-val"),
ec2Tag("aws-key", "aws-val"),
},
},
{
name: "when tags overlap, velero tags take precedence",
veleroTags: map[string]string{
"velero-key": "velero-val",
"overlapping-key": "velero-val",
},
volumeTags: []*ec2.Tag{
ec2Tag("aws-key", "aws-val"),
ec2Tag("overlapping-key", "aws-val"),
},
expected: []*ec2.Tag{
ec2Tag("velero-key", "velero-val"),
ec2Tag("overlapping-key", "velero-val"),
ec2Tag("aws-key", "aws-val"),
},
},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
res := getTags(test.veleroTags, test.volumeTags)
sort.Slice(res, func(i, j int) bool {
return *res[i].Key < *res[j].Key
})
sort.Slice(test.expected, func(i, j int) bool {
return *test.expected[i].Key < *test.expected[j].Key
})
assert.Equal(t, test.expected, res)
})
}
}
-104
View File
@@ -1,104 +0,0 @@
/*
Copyright 2018 the Velero 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 azure
import (
"os"
"strings"
"github.com/Azure/go-autorest/autorest/adal"
"github.com/Azure/go-autorest/autorest/azure"
"github.com/joho/godotenv"
"github.com/pkg/errors"
)
const (
tenantIDEnvVar = "AZURE_TENANT_ID"
subscriptionIDEnvVar = "AZURE_SUBSCRIPTION_ID"
clientIDEnvVar = "AZURE_CLIENT_ID"
clientSecretEnvVar = "AZURE_CLIENT_SECRET"
cloudNameEnvVar = "AZURE_CLOUD_NAME"
resourceGroupConfigKey = "resourceGroup"
)
// GetResticEnvVars gets the environment variables that restic
// relies on (AZURE_ACCOUNT_NAME and AZURE_ACCOUNT_KEY) based
// on info in the provided object storage location config map.
func GetResticEnvVars(config map[string]string) (map[string]string, error) {
storageAccountKey, _, err := getStorageAccountKey(config)
if err != nil {
return nil, err
}
return map[string]string{
"AZURE_ACCOUNT_NAME": config[storageAccountConfigKey],
"AZURE_ACCOUNT_KEY": storageAccountKey,
}, nil
}
func loadEnv() error {
envFile := os.Getenv("AZURE_CREDENTIALS_FILE")
if envFile == "" {
return nil
}
if err := godotenv.Overload(envFile); err != nil {
return errors.Wrapf(err, "error loading environment from AZURE_CREDENTIALS_FILE (%s)", envFile)
}
return nil
}
// ParseAzureEnvironment returns an azure.Environment for the given cloud
// name, or azure.PublicCloud if cloudName is empty.
func parseAzureEnvironment(cloudName string) (*azure.Environment, error) {
if cloudName == "" {
return &azure.PublicCloud, nil
}
env, err := azure.EnvironmentFromName(cloudName)
return &env, errors.WithStack(err)
}
func newServicePrincipalToken(tenantID, clientID, clientSecret string, env *azure.Environment) (*adal.ServicePrincipalToken, error) {
oauthConfig, err := adal.NewOAuthConfig(env.ActiveDirectoryEndpoint, tenantID)
if err != nil {
return nil, errors.Wrap(err, "error getting OAuthConfig")
}
return adal.NewServicePrincipalToken(*oauthConfig, clientID, clientSecret, env.ResourceManagerEndpoint)
}
func getRequiredValues(getValue func(string) string, keys ...string) (map[string]string, error) {
missing := []string{}
results := map[string]string{}
for _, key := range keys {
if val := getValue(key); val == "" {
missing = append(missing, key)
} else {
results[key] = val
}
}
if len(missing) > 0 {
return nil, errors.Errorf("the following keys do not have values: %s", strings.Join(missing, ", "))
}
return results, nil
}
-345
View File
@@ -1,345 +0,0 @@
/*
Copyright 2017, 2019 the Velero 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 azure
import (
"context"
"io"
"os"
"strings"
"time"
storagemgmt "github.com/Azure/azure-sdk-for-go/services/storage/mgmt/2018-02-01/storage"
"github.com/Azure/azure-sdk-for-go/storage"
"github.com/Azure/go-autorest/autorest"
"github.com/Azure/go-autorest/autorest/azure"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
"github.com/vmware-tanzu/velero/pkg/plugin/framework"
)
const (
storageAccountConfigKey = "storageAccount"
subscriptionIdConfigKey = "subscriptionId"
)
type containerGetter interface {
getContainer(bucket string) (container, error)
}
type azureContainerGetter struct {
blobService *storage.BlobStorageClient
}
func (cg *azureContainerGetter) getContainer(bucket string) (container, error) {
container := cg.blobService.GetContainerReference(bucket)
if container == nil {
return nil, errors.Errorf("unable to get container reference for bucket %v", bucket)
}
return &azureContainer{
container: container,
}, nil
}
type container interface {
ListBlobs(params storage.ListBlobsParameters) (storage.BlobListResponse, error)
}
type azureContainer struct {
container *storage.Container
}
func (c *azureContainer) ListBlobs(params storage.ListBlobsParameters) (storage.BlobListResponse, error) {
return c.container.ListBlobs(params)
}
type blobGetter interface {
getBlob(bucket, key string) (blob, error)
}
type azureBlobGetter struct {
blobService *storage.BlobStorageClient
}
func (bg *azureBlobGetter) getBlob(bucket, key string) (blob, error) {
container := bg.blobService.GetContainerReference(bucket)
if container == nil {
return nil, errors.Errorf("unable to get container reference for bucket %v", bucket)
}
blob := container.GetBlobReference(key)
if blob == nil {
return nil, errors.Errorf("unable to get blob reference for key %v", key)
}
return &azureBlob{
blob: blob,
}, nil
}
type blob interface {
CreateBlockBlobFromReader(blob io.Reader, options *storage.PutBlobOptions) error
Exists() (bool, error)
Get(options *storage.GetBlobOptions) (io.ReadCloser, error)
Delete(options *storage.DeleteBlobOptions) error
GetSASURI(options *storage.BlobSASOptions) (string, error)
}
type azureBlob struct {
blob *storage.Blob
}
func (b *azureBlob) CreateBlockBlobFromReader(blob io.Reader, options *storage.PutBlobOptions) error {
return b.blob.CreateBlockBlobFromReader(blob, options)
}
func (b *azureBlob) Exists() (bool, error) {
return b.blob.Exists()
}
func (b *azureBlob) Get(options *storage.GetBlobOptions) (io.ReadCloser, error) {
return b.blob.Get(options)
}
func (b *azureBlob) Delete(options *storage.DeleteBlobOptions) error {
return b.blob.Delete(options)
}
func (b *azureBlob) GetSASURI(options *storage.BlobSASOptions) (string, error) {
return b.blob.GetSASURI(*options)
}
type ObjectStore struct {
containerGetter containerGetter
blobGetter blobGetter
log logrus.FieldLogger
}
func NewObjectStore(logger logrus.FieldLogger) *ObjectStore {
return &ObjectStore{log: logger}
}
func getStorageAccountKey(config map[string]string) (string, *azure.Environment, error) {
// load environment vars from $AZURE_CREDENTIALS_FILE, if it exists
if err := loadEnv(); err != nil {
return "", nil, err
}
// 1. we need AZURE_TENANT_ID, AZURE_CLIENT_ID, AZURE_CLIENT_SECRET, AZURE_SUBSCRIPTION_ID
envVars, err := getRequiredValues(os.Getenv, tenantIDEnvVar, clientIDEnvVar, clientSecretEnvVar, subscriptionIDEnvVar)
if err != nil {
return "", nil, errors.Wrap(err, "unable to get all required environment variables")
}
// 2. Get Azure cloud from AZURE_CLOUD_NAME, if it exists. If the env var does not
// exist, parseAzureEnvironment will return azure.PublicCloud.
env, err := parseAzureEnvironment(os.Getenv(cloudNameEnvVar))
if err != nil {
return "", nil, errors.Wrap(err, "unable to parse azure cloud name environment variable")
}
// 3. check whether a different subscription ID was set for backups in config["subscriptionId"]
subscriptionId := envVars[subscriptionIDEnvVar]
if val := config[subscriptionIdConfigKey]; val != "" {
subscriptionId = val
}
// 4. we need config["resourceGroup"], config["storageAccount"]
if _, err := getRequiredValues(mapLookup(config), resourceGroupConfigKey, storageAccountConfigKey); err != nil {
return "", env, errors.Wrap(err, "unable to get all required config values")
}
// 5. get SPT
spt, err := newServicePrincipalToken(envVars[tenantIDEnvVar], envVars[clientIDEnvVar], envVars[clientSecretEnvVar], env)
if err != nil {
return "", env, errors.Wrap(err, "error getting service principal token")
}
// 6. get storageAccountsClient
storageAccountsClient := storagemgmt.NewAccountsClientWithBaseURI(env.ResourceManagerEndpoint, subscriptionId)
storageAccountsClient.Authorizer = autorest.NewBearerAuthorizer(spt)
// 7. get storage key
res, err := storageAccountsClient.ListKeys(context.TODO(), config[resourceGroupConfigKey], config[storageAccountConfigKey])
if err != nil {
return "", env, errors.WithStack(err)
}
if res.Keys == nil || len(*res.Keys) == 0 {
return "", env, errors.New("No storage keys found")
}
var storageKey string
for _, key := range *res.Keys {
// uppercase both strings for comparison because the ListKeys call returns e.g. "FULL" but
// the storagemgmt.Full constant in the SDK is defined as "Full".
if strings.ToUpper(string(key.Permissions)) == strings.ToUpper(string(storagemgmt.Full)) {
storageKey = *key.Value
break
}
}
if storageKey == "" {
return "", env, errors.New("No storage key with Full permissions found")
}
return storageKey, env, nil
}
func mapLookup(data map[string]string) func(string) string {
return func(key string) string {
return data[key]
}
}
func (o *ObjectStore) Init(config map[string]string) error {
if err := framework.ValidateObjectStoreConfigKeys(config,
resourceGroupConfigKey,
storageAccountConfigKey,
subscriptionIdConfigKey,
); err != nil {
return err
}
storageAccountKey, env, err := getStorageAccountKey(config)
if err != nil {
return err
}
// 6. get storageClient and blobClient
storageClient, err := storage.NewBasicClientOnSovereignCloud(config[storageAccountConfigKey], storageAccountKey, *env)
if err != nil {
return errors.Wrap(err, "error getting storage client")
}
blobClient := storageClient.GetBlobService()
o.containerGetter = &azureContainerGetter{
blobService: &blobClient,
}
o.blobGetter = &azureBlobGetter{
blobService: &blobClient,
}
return nil
}
func (o *ObjectStore) PutObject(bucket, key string, body io.Reader) error {
blob, err := o.blobGetter.getBlob(bucket, key)
if err != nil {
return err
}
return errors.WithStack(blob.CreateBlockBlobFromReader(body, nil))
}
func (o *ObjectStore) ObjectExists(bucket, key string) (bool, error) {
blob, err := o.blobGetter.getBlob(bucket, key)
if err != nil {
return false, err
}
exists, err := blob.Exists()
if err != nil {
return false, errors.WithStack(err)
}
return exists, nil
}
func (o *ObjectStore) GetObject(bucket, key string) (io.ReadCloser, error) {
blob, err := o.blobGetter.getBlob(bucket, key)
if err != nil {
return nil, err
}
res, err := blob.Get(nil)
if err != nil {
return nil, errors.WithStack(err)
}
return res, nil
}
func (o *ObjectStore) ListCommonPrefixes(bucket, prefix, delimiter string) ([]string, error) {
container, err := o.containerGetter.getContainer(bucket)
if err != nil {
return nil, err
}
params := storage.ListBlobsParameters{
Prefix: prefix,
Delimiter: delimiter,
}
res, err := container.ListBlobs(params)
if err != nil {
return nil, errors.WithStack(err)
}
return res.BlobPrefixes, nil
}
func (o *ObjectStore) ListObjects(bucket, prefix string) ([]string, error) {
container, err := o.containerGetter.getContainer(bucket)
if err != nil {
return nil, err
}
params := storage.ListBlobsParameters{
Prefix: prefix,
}
res, err := container.ListBlobs(params)
if err != nil {
return nil, errors.WithStack(err)
}
ret := make([]string, 0, len(res.Blobs))
for _, blob := range res.Blobs {
ret = append(ret, blob.Name)
}
return ret, nil
}
func (o *ObjectStore) DeleteObject(bucket string, key string) error {
blob, err := o.blobGetter.getBlob(bucket, key)
if err != nil {
return err
}
return errors.WithStack(blob.Delete(nil))
}
func (o *ObjectStore) CreateSignedURL(bucket, key string, ttl time.Duration) (string, error) {
blob, err := o.blobGetter.getBlob(bucket, key)
if err != nil {
return "", err
}
opts := storage.BlobSASOptions{
SASOptions: storage.SASOptions{
Expiry: time.Now().Add(ttl),
},
BlobServiceSASPermissions: storage.BlobServiceSASPermissions{
Read: true,
},
}
return blob.GetSASURI(&opts)
}
@@ -1,152 +0,0 @@
/*
Copyright 2018, 2019 the Velero 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 azure
import (
"io"
"testing"
"github.com/Azure/azure-sdk-for-go/storage"
"github.com/pkg/errors"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/mock"
"github.com/stretchr/testify/require"
)
func TestObjectExists(t *testing.T) {
tests := []struct {
name string
getBlobError error
exists bool
errorResponse error
expectedExists bool
expectedError string
}{
{
name: "getBlob error",
exists: false,
errorResponse: errors.New("getBlob"),
expectedExists: false,
expectedError: "getBlob",
},
{
name: "exists",
exists: true,
errorResponse: nil,
expectedExists: true,
},
{
name: "doesn't exist",
exists: false,
errorResponse: nil,
expectedExists: false,
},
{
name: "error checking for existence",
exists: false,
errorResponse: errors.New("bad"),
expectedExists: false,
expectedError: "bad",
},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
blobGetter := new(mockBlobGetter)
defer blobGetter.AssertExpectations(t)
o := &ObjectStore{
blobGetter: blobGetter,
}
bucket := "b"
key := "k"
blob := new(mockBlob)
defer blob.AssertExpectations(t)
blobGetter.On("getBlob", bucket, key).Return(blob, tc.getBlobError)
blob.On("Exists").Return(tc.exists, tc.errorResponse)
exists, err := o.ObjectExists(bucket, key)
if tc.expectedError != "" {
assert.EqualError(t, err, tc.expectedError)
return
}
require.NoError(t, err)
assert.Equal(t, tc.expectedExists, exists)
})
}
}
type mockBlobGetter struct {
mock.Mock
}
func (m *mockBlobGetter) getBlob(bucket string, key string) (blob, error) {
args := m.Called(bucket, key)
return args.Get(0).(blob), args.Error(1)
}
type mockBlob struct {
mock.Mock
}
func (m *mockBlob) CreateBlockBlobFromReader(blob io.Reader, options *storage.PutBlobOptions) error {
args := m.Called(blob, options)
return args.Error(0)
}
func (m *mockBlob) Exists() (bool, error) {
args := m.Called()
return args.Bool(0), args.Error(1)
}
func (m *mockBlob) Get(options *storage.GetBlobOptions) (io.ReadCloser, error) {
args := m.Called(options)
return args.Get(0).(io.ReadCloser), args.Error(1)
}
func (m *mockBlob) Delete(options *storage.DeleteBlobOptions) error {
args := m.Called(options)
return args.Error(0)
}
func (m *mockBlob) GetSASURI(options *storage.BlobSASOptions) (string, error) {
args := m.Called(options)
return args.String(0), args.Error(1)
}
type mockContainerGetter struct {
mock.Mock
}
func (m *mockContainerGetter) getContainer(bucket string) (container, error) {
args := m.Called(bucket)
return args.Get(0).(container), args.Error(1)
}
type mockContainer struct {
mock.Mock
}
func (m *mockContainer) ListBlobs(params storage.ListBlobsParameters) (storage.BlobListResponse, error) {
args := m.Called(params)
return args.Get(0).(storage.BlobListResponse), args.Error(1)
}
@@ -1,403 +0,0 @@
/*
Copyright 2017, 2019 the Velero 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 azure
import (
"context"
"fmt"
"net/http"
"os"
"regexp"
"strings"
"time"
disk "github.com/Azure/azure-sdk-for-go/services/compute/mgmt/2018-04-01/compute"
"github.com/Azure/go-autorest/autorest"
"github.com/pkg/errors"
uuid "github.com/satori/go.uuid"
"github.com/sirupsen/logrus"
v1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
"k8s.io/apimachinery/pkg/runtime"
"github.com/vmware-tanzu/velero/pkg/plugin/framework"
)
const (
resourceGroupEnvVar = "AZURE_RESOURCE_GROUP"
apiTimeoutConfigKey = "apiTimeout"
snapshotsResource = "snapshots"
disksResource = "disks"
)
type VolumeSnapshotter struct {
log logrus.FieldLogger
disks *disk.DisksClient
snaps *disk.SnapshotsClient
disksSubscription string
snapsSubscription string
disksResourceGroup string
snapsResourceGroup string
apiTimeout time.Duration
}
type snapshotIdentifier struct {
subscription string
resourceGroup string
name string
}
func (si *snapshotIdentifier) String() string {
return getComputeResourceName(si.subscription, si.resourceGroup, snapshotsResource, si.name)
}
func NewVolumeSnapshotter(logger logrus.FieldLogger) *VolumeSnapshotter {
return &VolumeSnapshotter{log: logger}
}
func (b *VolumeSnapshotter) Init(config map[string]string) error {
if err := framework.ValidateVolumeSnapshotterConfigKeys(config, resourceGroupConfigKey, apiTimeoutConfigKey, subscriptionIdConfigKey); err != nil {
return err
}
// load environment vars from $AZURE_CREDENTIALS_FILE, if it exists
if err := loadEnv(); err != nil {
return err
}
// 1. we need AZURE_TENANT_ID, AZURE_CLIENT_ID, AZURE_CLIENT_SECRET, AZURE_SUBSCRIPTION_ID, AZURE_RESOURCE_GROUP
envVars, err := getRequiredValues(os.Getenv, tenantIDEnvVar, clientIDEnvVar, clientSecretEnvVar, subscriptionIDEnvVar, resourceGroupEnvVar)
if err != nil {
return errors.Wrap(err, "unable to get all required environment variables")
}
// 2. set a different subscriptionId for snapshots if specified
snapshotsSubscriptionId := envVars[subscriptionIDEnvVar]
if val := config[subscriptionIdConfigKey]; val != "" {
// if subscription was set in config, it is required to also set the resource group
if _, err := getRequiredValues(mapLookup(config), resourceGroupConfigKey); err != nil {
return errors.Wrap(err, "resourceGroup not specified, but is a requirement when backing up to a different subscription")
}
snapshotsSubscriptionId = val
}
// 3. Get Azure cloud from AZURE_CLOUD_NAME, if it exists. If the env var does not
// exist, parseAzureEnvironment will return azure.PublicCloud.
env, err := parseAzureEnvironment(os.Getenv(cloudNameEnvVar))
if err != nil {
return errors.Wrap(err, "unable to parse azure cloud name environment variable")
}
// 4. if config["apiTimeout"] is empty, default to 2m; otherwise, parse it
var apiTimeout time.Duration
if val := config[apiTimeoutConfigKey]; val == "" {
apiTimeout = 2 * time.Minute
} else {
apiTimeout, err = time.ParseDuration(val)
if err != nil {
return errors.Wrapf(err, "unable to parse value %q for config key %q (expected a duration string)", val, apiTimeoutConfigKey)
}
}
// 5. get SPT
spt, err := newServicePrincipalToken(envVars[tenantIDEnvVar], envVars[clientIDEnvVar], envVars[clientSecretEnvVar], env)
if err != nil {
return errors.Wrap(err, "error getting service principal token")
}
// 6. set up clients
disksClient := disk.NewDisksClientWithBaseURI(env.ResourceManagerEndpoint, envVars[subscriptionIDEnvVar])
snapsClient := disk.NewSnapshotsClientWithBaseURI(env.ResourceManagerEndpoint, snapshotsSubscriptionId)
disksClient.PollingDelay = 5 * time.Second
snapsClient.PollingDelay = 5 * time.Second
authorizer := autorest.NewBearerAuthorizer(spt)
disksClient.Authorizer = authorizer
snapsClient.Authorizer = authorizer
b.disks = &disksClient
b.snaps = &snapsClient
b.disksSubscription = envVars[subscriptionIDEnvVar]
b.snapsSubscription = snapshotsSubscriptionId
b.disksResourceGroup = envVars[resourceGroupEnvVar]
b.snapsResourceGroup = config[resourceGroupConfigKey]
// if no resource group was explicitly specified in 'config',
// use the value from the env var (i.e. the same one as where
// the cluster & disks are)
if b.snapsResourceGroup == "" {
b.snapsResourceGroup = envVars[resourceGroupEnvVar]
}
b.apiTimeout = apiTimeout
return nil
}
func (b *VolumeSnapshotter) CreateVolumeFromSnapshot(snapshotID, volumeType, volumeAZ string, iops *int64) (string, error) {
snapshotIdentifier, err := parseFullSnapshotName(snapshotID)
if err != nil {
return "", err
}
// Lookup snapshot info for its Location & Tags so we can apply them to the volume
snapshotInfo, err := b.snaps.Get(context.TODO(), snapshotIdentifier.resourceGroup, snapshotIdentifier.name)
if err != nil {
return "", errors.WithStack(err)
}
diskName := "restore-" + uuid.NewV4().String()
disk := disk.Disk{
Name: &diskName,
Location: snapshotInfo.Location,
DiskProperties: &disk.DiskProperties{
CreationData: &disk.CreationData{
CreateOption: disk.Copy,
SourceResourceID: stringPtr(snapshotIdentifier.String()),
},
},
Sku: &disk.DiskSku{
Name: disk.StorageAccountTypes(volumeType),
},
Tags: snapshotInfo.Tags,
}
// Restore the disk in the correct zone
regionParts := strings.Split(volumeAZ, "-")
if len(regionParts) >= 2 {
disk.Zones = &[]string{regionParts[len(regionParts)-1]}
}
ctx, cancel := context.WithTimeout(context.Background(), b.apiTimeout)
defer cancel()
future, err := b.disks.CreateOrUpdate(ctx, b.disksResourceGroup, *disk.Name, disk)
if err != nil {
return "", errors.WithStack(err)
}
if err = future.WaitForCompletionRef(ctx, b.disks.Client); err != nil {
return "", errors.WithStack(err)
}
if _, err = future.Result(*b.disks); err != nil {
return "", errors.WithStack(err)
}
return diskName, nil
}
func (b *VolumeSnapshotter) GetVolumeInfo(volumeID, volumeAZ string) (string, *int64, error) {
res, err := b.disks.Get(context.TODO(), b.disksResourceGroup, volumeID)
if err != nil {
return "", nil, errors.WithStack(err)
}
if res.Sku == nil {
return "", nil, errors.New("disk has a nil SKU")
}
return string(res.Sku.Name), nil, nil
}
func (b *VolumeSnapshotter) CreateSnapshot(volumeID, volumeAZ string, tags map[string]string) (string, error) {
// Lookup disk info for its Location
diskInfo, err := b.disks.Get(context.TODO(), b.disksResourceGroup, volumeID)
if err != nil {
return "", errors.WithStack(err)
}
fullDiskName := getComputeResourceName(b.disksSubscription, b.disksResourceGroup, disksResource, volumeID)
// snapshot names must be <= 80 characters long
var snapshotName string
suffix := "-" + uuid.NewV4().String()
if len(volumeID) <= (80 - len(suffix)) {
snapshotName = volumeID + suffix
} else {
snapshotName = volumeID[0:80-len(suffix)] + suffix
}
snap := disk.Snapshot{
Name: &snapshotName,
DiskProperties: &disk.DiskProperties{
CreationData: &disk.CreationData{
CreateOption: disk.Copy,
SourceResourceID: &fullDiskName,
},
},
Tags: getSnapshotTags(tags, diskInfo.Tags),
Location: diskInfo.Location,
}
ctx, cancel := context.WithTimeout(context.Background(), b.apiTimeout)
defer cancel()
future, err := b.snaps.CreateOrUpdate(ctx, b.snapsResourceGroup, *snap.Name, snap)
if err != nil {
return "", errors.WithStack(err)
}
if err = future.WaitForCompletionRef(ctx, b.snaps.Client); err != nil {
return "", errors.WithStack(err)
}
if _, err = future.Result(*b.snaps); err != nil {
return "", errors.WithStack(err)
}
return getComputeResourceName(b.snapsSubscription, b.snapsResourceGroup, snapshotsResource, snapshotName), nil
}
func getSnapshotTags(veleroTags map[string]string, diskTags map[string]*string) map[string]*string {
if diskTags == nil && len(veleroTags) == 0 {
return nil
}
snapshotTags := make(map[string]*string)
// copy tags from disk to snapshot
if diskTags != nil {
for k, v := range diskTags {
snapshotTags[k] = stringPtr(*v)
}
}
// merge Velero-assigned tags with the disk's tags (note that we want current
// Velero-assigned tags to overwrite any older versions of them that may exist
// due to prior snapshots/restores)
for k, v := range veleroTags {
// Azure does not allow slashes in tag keys, so replace
// with dash (inline with what Kubernetes does)
key := strings.Replace(k, "/", "-", -1)
snapshotTags[key] = stringPtr(v)
}
return snapshotTags
}
func stringPtr(s string) *string {
return &s
}
func (b *VolumeSnapshotter) DeleteSnapshot(snapshotID string) error {
snapshotInfo, err := parseFullSnapshotName(snapshotID)
if err != nil {
return err
}
ctx, cancel := context.WithTimeout(context.Background(), b.apiTimeout)
defer cancel()
// we don't want to return an error if the snapshot doesn't exist, and
// the Delete(..) call does not return a clear error if that's the case,
// so first try to get it and return early if we get a 404.
_, err = b.snaps.Get(ctx, snapshotInfo.resourceGroup, snapshotInfo.name)
if azureErr, ok := err.(autorest.DetailedError); ok && azureErr.StatusCode == http.StatusNotFound {
b.log.WithField("snapshotID", snapshotID).Debug("Snapshot not found")
return nil
}
future, err := b.snaps.Delete(ctx, snapshotInfo.resourceGroup, snapshotInfo.name)
if err != nil {
return errors.WithStack(err)
}
if err = future.WaitForCompletionRef(ctx, b.snaps.Client); err != nil {
b.log.WithError(err).Errorf("Error waiting for completion ref")
return errors.WithStack(err)
}
_, err = future.Result(*b.snaps)
if err != nil {
return errors.WithStack(err)
}
return nil
}
func getComputeResourceName(subscription, resourceGroup, resource, name string) string {
return fmt.Sprintf("/subscriptions/%s/resourceGroups/%s/providers/Microsoft.Compute/%s/%s", subscription, resourceGroup, resource, name)
}
var snapshotURIRegexp = regexp.MustCompile(
`^\/subscriptions\/(?P<subscription>.*)\/resourceGroups\/(?P<resourceGroup>.*)\/providers\/Microsoft.Compute\/snapshots\/(?P<snapshotName>.*)$`)
// parseFullSnapshotName takes a fully-qualified snapshot name and returns
// a snapshot identifier or an error if the snapshot name does not match the
// regexp.
func parseFullSnapshotName(name string) (*snapshotIdentifier, error) {
submatches := snapshotURIRegexp.FindStringSubmatch(name)
if len(submatches) != len(snapshotURIRegexp.SubexpNames()) {
return nil, errors.New("snapshot URI could not be parsed")
}
snapshotID := &snapshotIdentifier{}
// capture names start at index 1 to line up with the corresponding indexes
// of submatches (see godoc on SubexpNames())
for i, names := 1, snapshotURIRegexp.SubexpNames(); i < len(names); i++ {
switch names[i] {
case "subscription":
snapshotID.subscription = submatches[i]
case "resourceGroup":
snapshotID.resourceGroup = submatches[i]
case "snapshotName":
snapshotID.name = submatches[i]
default:
return nil, errors.New("unexpected named capture from snapshot URI regex")
}
}
return snapshotID, nil
}
func (b *VolumeSnapshotter) GetVolumeID(unstructuredPV runtime.Unstructured) (string, error) {
pv := new(v1.PersistentVolume)
if err := runtime.DefaultUnstructuredConverter.FromUnstructured(unstructuredPV.UnstructuredContent(), pv); err != nil {
return "", errors.WithStack(err)
}
if pv.Spec.AzureDisk == nil {
return "", nil
}
if pv.Spec.AzureDisk.DiskName == "" {
return "", errors.New("spec.azureDisk.diskName not found")
}
return pv.Spec.AzureDisk.DiskName, nil
}
func (b *VolumeSnapshotter) SetVolumeID(unstructuredPV runtime.Unstructured, volumeID string) (runtime.Unstructured, error) {
pv := new(v1.PersistentVolume)
if err := runtime.DefaultUnstructuredConverter.FromUnstructured(unstructuredPV.UnstructuredContent(), pv); err != nil {
return nil, errors.WithStack(err)
}
if pv.Spec.AzureDisk == nil {
return nil, errors.New("spec.azureDisk not found")
}
pv.Spec.AzureDisk.DiskName = volumeID
pv.Spec.AzureDisk.DataDiskURI = getComputeResourceName(b.disksSubscription, b.disksResourceGroup, disksResource, volumeID)
res, err := runtime.DefaultUnstructuredConverter.ToUnstructured(pv)
if err != nil {
return nil, errors.WithStack(err)
}
return &unstructured.Unstructured{Object: res}, nil
}
@@ -1,210 +0,0 @@
/*
Copyright 2017 the Velero 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 azure
import (
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
v1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
"k8s.io/apimachinery/pkg/runtime"
)
func TestGetVolumeID(t *testing.T) {
b := &VolumeSnapshotter{}
pv := &unstructured.Unstructured{
Object: map[string]interface{}{},
}
// missing spec.azureDisk -> no error
volumeID, err := b.GetVolumeID(pv)
require.NoError(t, err)
assert.Equal(t, "", volumeID)
// missing spec.azureDisk.diskName -> error
azure := map[string]interface{}{}
pv.Object["spec"] = map[string]interface{}{
"azureDisk": azure,
}
volumeID, err = b.GetVolumeID(pv)
assert.Error(t, err)
assert.Equal(t, "", volumeID)
// valid
azure["diskName"] = "foo"
volumeID, err = b.GetVolumeID(pv)
assert.NoError(t, err)
assert.Equal(t, "foo", volumeID)
}
func TestSetVolumeID(t *testing.T) {
b := &VolumeSnapshotter{
disksResourceGroup: "rg",
disksSubscription: "sub",
}
pv := &unstructured.Unstructured{
Object: map[string]interface{}{},
}
// missing spec.azureDisk -> error
updatedPV, err := b.SetVolumeID(pv, "updated")
require.Error(t, err)
// happy path, no diskURI
azure := map[string]interface{}{}
pv.Object["spec"] = map[string]interface{}{
"azureDisk": azure,
}
updatedPV, err = b.SetVolumeID(pv, "updated")
require.NoError(t, err)
res := new(v1.PersistentVolume)
require.NoError(t, runtime.DefaultUnstructuredConverter.FromUnstructured(updatedPV.UnstructuredContent(), res))
require.NotNil(t, res.Spec.AzureDisk)
assert.Equal(t, "updated", res.Spec.AzureDisk.DiskName)
assert.Equal(t, "/subscriptions/sub/resourceGroups/rg/providers/Microsoft.Compute/disks/updated", res.Spec.AzureDisk.DataDiskURI)
// with diskURI
azure["diskURI"] = "/foo/bar/updated/blarg"
updatedPV, err = b.SetVolumeID(pv, "revised")
require.NoError(t, err)
res = new(v1.PersistentVolume)
require.NoError(t, runtime.DefaultUnstructuredConverter.FromUnstructured(updatedPV.UnstructuredContent(), res))
require.NotNil(t, res.Spec.AzureDisk)
assert.Equal(t, "revised", res.Spec.AzureDisk.DiskName)
assert.Equal(t, "/subscriptions/sub/resourceGroups/rg/providers/Microsoft.Compute/disks/revised", res.Spec.AzureDisk.DataDiskURI)
}
func TestParseFullSnapshotName(t *testing.T) {
// invalid name
fullName := "foo/bar"
_, err := parseFullSnapshotName(fullName)
assert.Error(t, err)
// valid name (current format)
fullName = "/subscriptions/sub-1/resourceGroups/rg-1/providers/Microsoft.Compute/snapshots/snap-1"
snap, err := parseFullSnapshotName(fullName)
require.NoError(t, err)
assert.Equal(t, "sub-1", snap.subscription)
assert.Equal(t, "rg-1", snap.resourceGroup)
assert.Equal(t, "snap-1", snap.name)
}
func TestGetComputeResourceName(t *testing.T) {
assert.Equal(t, "/subscriptions/sub-1/resourceGroups/rg-1/providers/Microsoft.Compute/disks/disk-1", getComputeResourceName("sub-1", "rg-1", disksResource, "disk-1"))
assert.Equal(t, "/subscriptions/sub-1/resourceGroups/rg-1/providers/Microsoft.Compute/snapshots/snap-1", getComputeResourceName("sub-1", "rg-1", snapshotsResource, "snap-1"))
}
func TestGetSnapshotTags(t *testing.T) {
tests := []struct {
name string
veleroTags map[string]string
diskTags map[string]*string
expected map[string]*string
}{
{
name: "degenerate case (no tags)",
veleroTags: nil,
diskTags: nil,
expected: nil,
},
{
name: "velero tags only get applied",
veleroTags: map[string]string{
"velero-key1": "velero-val1",
"velero-key2": "velero-val2",
},
diskTags: nil,
expected: map[string]*string{
"velero-key1": stringPtr("velero-val1"),
"velero-key2": stringPtr("velero-val2"),
},
},
{
name: "slashes in velero tag keys get replaces with dashes",
veleroTags: map[string]string{
"velero/key1": "velero-val1",
"velero/key/2": "velero-val2",
},
diskTags: nil,
expected: map[string]*string{
"velero-key1": stringPtr("velero-val1"),
"velero-key-2": stringPtr("velero-val2"),
},
},
{
name: "volume tags only get applied",
veleroTags: nil,
diskTags: map[string]*string{
"azure-key1": stringPtr("azure-val1"),
"azure-key2": stringPtr("azure-val2"),
},
expected: map[string]*string{
"azure-key1": stringPtr("azure-val1"),
"azure-key2": stringPtr("azure-val2"),
},
},
{
name: "non-overlapping velero and volume tags both get applied",
veleroTags: map[string]string{"velero-key": "velero-val"},
diskTags: map[string]*string{"azure-key": stringPtr("azure-val")},
expected: map[string]*string{
"velero-key": stringPtr("velero-val"),
"azure-key": stringPtr("azure-val"),
},
},
{
name: "when tags overlap, velero tags take precedence",
veleroTags: map[string]string{
"velero-key": "velero-val",
"overlapping-key": "velero-val",
},
diskTags: map[string]*string{
"azure-key": stringPtr("azure-val"),
"overlapping-key": stringPtr("azure-val"),
},
expected: map[string]*string{
"velero-key": stringPtr("velero-val"),
"azure-key": stringPtr("azure-val"),
"overlapping-key": stringPtr("velero-val"),
},
},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
res := getSnapshotTags(test.veleroTags, test.diskTags)
if test.expected == nil {
assert.Nil(t, res)
return
}
assert.Equal(t, len(test.expected), len(res))
for k, v := range test.expected {
assert.Equal(t, v, res[k])
}
})
}
}
-261
View File
@@ -1,261 +0,0 @@
/*
Copyright 2017, 2019 the Velero 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 gcp
import (
"context"
"encoding/base64"
"io"
"time"
"cloud.google.com/go/storage"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
"golang.org/x/oauth2/google"
"google.golang.org/api/iamcredentials/v1"
"google.golang.org/api/iterator"
"google.golang.org/api/option"
"github.com/vmware-tanzu/velero/pkg/plugin/framework"
)
const (
credentialsEnvVar = "GOOGLE_APPLICATION_CREDENTIALS"
kmsKeyNameConfigKey = "kmsKeyName"
serviceAccountConfig = "serviceAccount"
)
// bucketWriter wraps the GCP SDK functions for accessing object store so they can be faked for testing.
type bucketWriter interface {
// getWriteCloser returns an io.WriteCloser that can be used to upload data to the specified bucket for the specified key.
getWriteCloser(bucket, key string) io.WriteCloser
getAttrs(bucket, key string) (*storage.ObjectAttrs, error)
}
type writer struct {
client *storage.Client
kmsKeyName string
}
func (w *writer) getWriteCloser(bucket, key string) io.WriteCloser {
writer := w.client.Bucket(bucket).Object(key).NewWriter(context.Background())
writer.KMSKeyName = w.kmsKeyName
return writer
}
func (w *writer) getAttrs(bucket, key string) (*storage.ObjectAttrs, error) {
return w.client.Bucket(bucket).Object(key).Attrs(context.Background())
}
type ObjectStore struct {
log logrus.FieldLogger
client *storage.Client
googleAccessID string
privateKey []byte
bucketWriter bucketWriter
iamSvc *iamcredentials.Service
}
func NewObjectStore(logger logrus.FieldLogger) *ObjectStore {
return &ObjectStore{log: logger}
}
func (o *ObjectStore) Init(config map[string]string) error {
if err := framework.ValidateObjectStoreConfigKeys(config, kmsKeyNameConfigKey, serviceAccountConfig); err != nil {
return err
}
// Find default token source to extract the GoogleAccessID
ctx := context.Background()
creds, err := google.FindDefaultCredentials(ctx)
if err != nil {
return errors.WithStack(err)
}
if creds.JSON != nil {
// Using Credentials File
err = o.initFromKeyFile(creds)
} else {
// Using compute engine credentials. Use this if workload identity is enabled.
err = o.initFromComputeEngine(config)
}
if err != nil {
return errors.WithStack(err)
}
client, err := storage.NewClient(ctx, option.WithScopes(storage.ScopeReadWrite))
if err != nil {
return errors.WithStack(err)
}
o.client = client
o.bucketWriter = &writer{
client: o.client,
kmsKeyName: config[kmsKeyNameConfigKey],
}
return nil
}
func (o *ObjectStore) initFromKeyFile(creds *google.Credentials) error {
jwtConfig, err := google.JWTConfigFromJSON(creds.JSON)
if err != nil {
return errors.Wrap(err, "error parsing credentials file; should be JSON")
}
if jwtConfig.Email == "" {
return errors.Errorf("credentials file pointed to by %s does not contain an email", "GOOGLE_APPLICATION_CREDENTIALS")
}
if len(jwtConfig.PrivateKey) == 0 {
return errors.Errorf("credentials file pointed to by %s does not contain a private key", "GOOGLE_APPLICATION_CREDENTIALS")
}
o.googleAccessID = jwtConfig.Email
o.privateKey = jwtConfig.PrivateKey
return nil
}
func (o *ObjectStore) initFromComputeEngine(config map[string]string) error {
var err error
var ok bool
o.googleAccessID, ok = config["serviceAccount"]
if !ok {
return errors.Errorf("serviceAccount is expected to be provided as an item in BackupStorageLocation's config")
}
o.iamSvc, err = iamcredentials.NewService(context.Background())
return err
}
func (o *ObjectStore) PutObject(bucket, key string, body io.Reader) error {
w := o.bucketWriter.getWriteCloser(bucket, key)
// The writer returned by NewWriter is asynchronous, so errors aren't guaranteed
// until Close() is called
_, copyErr := io.Copy(w, body)
// Ensure we close w and report errors properly
closeErr := w.Close()
if copyErr != nil {
return copyErr
}
return closeErr
}
func (o *ObjectStore) ObjectExists(bucket, key string) (bool, error) {
if _, err := o.bucketWriter.getAttrs(bucket, key); err != nil {
if err == storage.ErrObjectNotExist {
return false, nil
}
return false, errors.WithStack(err)
}
return true, nil
}
func (o *ObjectStore) GetObject(bucket, key string) (io.ReadCloser, error) {
r, err := o.client.Bucket(bucket).Object(key).NewReader(context.Background())
if err != nil {
return nil, errors.WithStack(err)
}
return r, nil
}
func (o *ObjectStore) ListCommonPrefixes(bucket, prefix, delimiter string) ([]string, error) {
q := &storage.Query{
Prefix: prefix,
Delimiter: delimiter,
}
iter := o.client.Bucket(bucket).Objects(context.Background(), q)
var res []string
for {
obj, err := iter.Next()
if err != nil && err != iterator.Done {
return nil, errors.WithStack(err)
}
if err == iterator.Done {
break
}
if obj.Prefix != "" {
res = append(res, obj.Prefix)
}
}
return res, nil
}
func (o *ObjectStore) ListObjects(bucket, prefix string) ([]string, error) {
q := &storage.Query{
Prefix: prefix,
}
var res []string
iter := o.client.Bucket(bucket).Objects(context.Background(), q)
for {
obj, err := iter.Next()
if err == iterator.Done {
return res, nil
}
if err != nil {
return nil, errors.WithStack(err)
}
res = append(res, obj.Name)
}
}
func (o *ObjectStore) DeleteObject(bucket, key string) error {
return errors.Wrapf(o.client.Bucket(bucket).Object(key).Delete(context.Background()), "error deleting object %s", key)
}
/*
* Use the iamSignBlob api call to sign the url if there is no credentials file to get the key from.
* https://cloud.google.com/iam/credentials/reference/rest/v1/projects.serviceAccounts/signBlob
*/
func (o *ObjectStore) SignBytes(bytes []byte) ([]byte, error) {
name := "projects/-/serviceAccounts/" + o.googleAccessID
resp, err := o.iamSvc.Projects.ServiceAccounts.SignBlob(name, &iamcredentials.SignBlobRequest{
Payload: base64.StdEncoding.EncodeToString(bytes),
}).Context(context.Background()).Do()
if err != nil {
return nil, err
}
return base64.StdEncoding.DecodeString(resp.SignedBlob)
}
func (o *ObjectStore) CreateSignedURL(bucket, key string, ttl time.Duration) (string, error) {
options := storage.SignedURLOptions{
GoogleAccessID: o.googleAccessID,
Method: "GET",
Expires: time.Now().Add(ttl),
}
if o.privateKey == nil {
options.SignBytes = o.SignBytes
} else {
options.PrivateKey = o.privateKey
}
return storage.SignedURL(bucket, key, &options)
}
-154
View File
@@ -1,154 +0,0 @@
/*
Copyright 2018 the Velero 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 gcp
import (
"errors"
"io"
"strings"
"testing"
"cloud.google.com/go/storage"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
velerotest "github.com/vmware-tanzu/velero/pkg/test"
)
type mockWriteCloser struct {
closeErr error
writeErr error
}
func (m *mockWriteCloser) Close() error {
return m.closeErr
}
func (m *mockWriteCloser) Write(b []byte) (int, error) {
return len(b), m.writeErr
}
func newMockWriteCloser(writeErr, closeErr error) *mockWriteCloser {
return &mockWriteCloser{writeErr: writeErr, closeErr: closeErr}
}
type fakeWriter struct {
wc *mockWriteCloser
attrsErr error
}
func newFakeWriter(wc *mockWriteCloser) *fakeWriter {
return &fakeWriter{wc: wc}
}
func (fw *fakeWriter) getWriteCloser(bucket, name string) io.WriteCloser {
return fw.wc
}
func (fw *fakeWriter) getAttrs(bucket, key string) (*storage.ObjectAttrs, error) {
return new(storage.ObjectAttrs), fw.attrsErr
}
func TestPutObject(t *testing.T) {
tests := []struct {
name string
writeErr error
closeErr error
expectedErr error
}{
{
name: "No errors returns nil",
closeErr: nil,
writeErr: nil,
expectedErr: nil,
},
{
name: "Close() errors are returned",
closeErr: errors.New("error closing"),
expectedErr: errors.New("error closing"),
},
{
name: "Write() errors are returned",
writeErr: errors.New("error writing"),
expectedErr: errors.New("error writing"),
},
{
name: "Write errors supercede close errors",
writeErr: errors.New("error writing"),
closeErr: errors.New("error closing"),
expectedErr: errors.New("error writing"),
},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
wc := newMockWriteCloser(test.writeErr, test.closeErr)
o := NewObjectStore(velerotest.NewLogger())
o.bucketWriter = newFakeWriter(wc)
err := o.PutObject("bucket", "key", strings.NewReader("contents"))
assert.Equal(t, test.expectedErr, err)
})
}
}
func TestObjectExists(t *testing.T) {
tests := []struct {
name string
errorResponse error
expectedExists bool
expectedError string
}{
{
name: "exists",
errorResponse: nil,
expectedExists: true,
},
{
name: "doesn't exist",
errorResponse: storage.ErrObjectNotExist,
expectedExists: false,
},
{
name: "error checking for existence",
errorResponse: errors.New("bad"),
expectedExists: false,
expectedError: "bad",
},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
o := NewObjectStore(velerotest.NewLogger())
w := newFakeWriter(nil)
o.bucketWriter = w
w.attrsErr = tc.errorResponse
bucket := "b"
key := "k"
exists, err := o.ObjectExists(bucket, key)
if tc.expectedError != "" {
assert.EqualError(t, err, tc.expectedError)
return
}
require.NoError(t, err)
assert.Equal(t, tc.expectedExists, exists)
})
}
}
-356
View File
@@ -1,356 +0,0 @@
/*
Copyright 2017, 2019 the Velero 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 gcp
import (
"encoding/json"
"net/http"
"strings"
"github.com/pkg/errors"
uuid "github.com/satori/go.uuid"
"github.com/sirupsen/logrus"
"golang.org/x/oauth2"
"golang.org/x/oauth2/google"
"google.golang.org/api/compute/v1"
"google.golang.org/api/googleapi"
v1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
"k8s.io/apimachinery/pkg/runtime"
"github.com/vmware-tanzu/velero/pkg/plugin/framework"
)
const (
zoneSeparator = "__"
projectKey = "project"
snapshotLocationKey = "snapshotLocation"
)
type VolumeSnapshotter struct {
log logrus.FieldLogger
gce *compute.Service
snapshotLocation string
volumeProject string
snapshotProject string
}
func NewVolumeSnapshotter(logger logrus.FieldLogger) *VolumeSnapshotter {
return &VolumeSnapshotter{log: logger}
}
func (b *VolumeSnapshotter) Init(config map[string]string) error {
if err := framework.ValidateVolumeSnapshotterConfigKeys(config, snapshotLocationKey, projectKey); err != nil {
return err
}
/* Works with both credential files and the default compute engine service account */
creds, err := google.FindDefaultCredentials(oauth2.NoContext, compute.ComputeScope)
if err != nil {
return errors.WithStack(err)
}
b.snapshotLocation = config[snapshotLocationKey]
b.volumeProject = creds.ProjectID
// get snapshot project from 'project' config key if specified,
// otherwise from the credentials file
b.snapshotProject = config[projectKey]
if b.snapshotProject == "" {
b.snapshotProject = b.volumeProject
}
client := oauth2.NewClient(oauth2.NoContext, creds.TokenSource)
gce, err := compute.New(client)
if err != nil {
return errors.WithStack(err)
}
b.gce = gce
return nil
}
// isMultiZone returns true if the failure-domain tag contains
// double underscore, which is the separator used
// by GKE when a storage class spans multiple availablity
// zones.
func isMultiZone(volumeAZ string) bool {
return strings.Contains(volumeAZ, zoneSeparator)
}
// parseRegion parses a failure-domain tag with multiple zones
// and returns a single region. Zones are sperated by double underscores (__).
// For example
// input: us-central1-a__us-central1-b
// return: us-central1
// When a custom storage class spans multiple geographical zones,
// such as us-central1 and us-west1 only the zone matching the cluster is used
// in the failure-domain tag.
// For example
// Cluster nodes in us-central1-c, us-central1-f
// Storage class zones us-central1-a, us-central1-f, us-east1-a, us-east1-d
// The failure-domain tag would be: us-central1-a__us-central1-f
func parseRegion(volumeAZ string) (string, error) {
zones := strings.Split(volumeAZ, zoneSeparator)
zone := zones[0]
parts := strings.SplitAfterN(zone, "-", 3)
if len(parts) < 2 {
return "", errors.Errorf("failed to parse region from zone: %q", volumeAZ)
}
return parts[0] + strings.TrimSuffix(parts[1], "-"), nil
}
// Retrieve the URLs for zones via the GCP API.
func (b *VolumeSnapshotter) getZoneURLs(volumeAZ string) ([]string, error) {
zones := strings.Split(volumeAZ, zoneSeparator)
var zoneURLs []string
for _, z := range zones {
zone, err := b.gce.Zones.Get(b.volumeProject, z).Do()
if err != nil {
return nil, errors.WithStack(err)
}
zoneURLs = append(zoneURLs, zone.SelfLink)
}
return zoneURLs, nil
}
func (b *VolumeSnapshotter) CreateVolumeFromSnapshot(snapshotID, volumeType, volumeAZ string, iops *int64) (volumeID string, err error) {
// get the snapshot so we can apply its tags to the volume
res, err := b.gce.Snapshots.Get(b.snapshotProject, snapshotID).Do()
if err != nil {
return "", errors.WithStack(err)
}
// Kubernetes uses the description field of GCP disks to store a JSON doc containing
// tags.
//
// use the snapshot's description (which contains tags from the snapshotted disk
// plus Velero-specific tags) to set the new disk's description.
disk := &compute.Disk{
Name: "restore-" + uuid.NewV4().String(),
SourceSnapshot: res.SelfLink,
Type: volumeType,
Description: res.Description,
}
if isMultiZone(volumeAZ) {
volumeRegion, err := parseRegion(volumeAZ)
if err != nil {
return "", err
}
// URLs for zones that the volume is replicated to within GCP
zoneURLs, err := b.getZoneURLs(volumeAZ)
if err != nil {
return "", err
}
disk.ReplicaZones = zoneURLs
if _, err = b.gce.RegionDisks.Insert(b.volumeProject, volumeRegion, disk).Do(); err != nil {
return "", errors.WithStack(err)
}
} else {
if _, err = b.gce.Disks.Insert(b.volumeProject, volumeAZ, disk).Do(); err != nil {
return "", errors.WithStack(err)
}
}
return disk.Name, nil
}
func (b *VolumeSnapshotter) GetVolumeInfo(volumeID, volumeAZ string) (string, *int64, error) {
var (
res *compute.Disk
err error
)
if isMultiZone(volumeAZ) {
volumeRegion, err := parseRegion(volumeAZ)
if err != nil {
return "", nil, errors.WithStack(err)
}
res, err = b.gce.RegionDisks.Get(b.volumeProject, volumeRegion, volumeID).Do()
if err != nil {
return "", nil, errors.WithStack(err)
}
} else {
res, err = b.gce.Disks.Get(b.volumeProject, volumeAZ, volumeID).Do()
if err != nil {
return "", nil, errors.WithStack(err)
}
}
return res.Type, nil, nil
}
func (b *VolumeSnapshotter) 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
suffix := "-" + uuid.NewV4().String()
if len(volumeID) <= (63 - len(suffix)) {
snapshotName = volumeID + suffix
} else {
snapshotName = volumeID[0:63-len(suffix)] + suffix
}
if isMultiZone(volumeAZ) {
volumeRegion, err := parseRegion(volumeAZ)
if err != nil {
return "", errors.WithStack(err)
}
return b.createRegionSnapshot(snapshotName, volumeID, volumeRegion, tags)
} else {
return b.createSnapshot(snapshotName, volumeID, volumeAZ, tags)
}
}
func (b *VolumeSnapshotter) createSnapshot(snapshotName, volumeID, volumeAZ string, tags map[string]string) (string, error) {
disk, err := b.gce.Disks.Get(b.volumeProject, volumeAZ, volumeID).Do()
if err != nil {
return "", errors.WithStack(err)
}
gceSnap := compute.Snapshot{
Name: snapshotName,
Description: getSnapshotTags(tags, disk.Description, b.log),
}
if b.snapshotLocation != "" {
gceSnap.StorageLocations = []string{b.snapshotLocation}
}
_, err = b.gce.Disks.CreateSnapshot(b.snapshotProject, volumeAZ, volumeID, &gceSnap).Do()
if err != nil {
return "", errors.WithStack(err)
}
return gceSnap.Name, nil
}
func (b *VolumeSnapshotter) createRegionSnapshot(snapshotName, volumeID, volumeRegion string, tags map[string]string) (string, error) {
disk, err := b.gce.RegionDisks.Get(b.volumeProject, volumeRegion, volumeID).Do()
if err != nil {
return "", errors.WithStack(err)
}
gceSnap := compute.Snapshot{
Name: snapshotName,
Description: getSnapshotTags(tags, disk.Description, b.log),
}
if b.snapshotLocation != "" {
gceSnap.StorageLocations = []string{b.snapshotLocation}
}
_, err = b.gce.RegionDisks.CreateSnapshot(b.snapshotProject, volumeRegion, volumeID, &gceSnap).Do()
if err != nil {
return "", errors.WithStack(err)
}
return gceSnap.Name, nil
}
func getSnapshotTags(veleroTags map[string]string, diskDescription string, log logrus.FieldLogger) string {
// Kubernetes uses the description field of GCP disks to store a JSON doc containing
// tags.
//
// use the tags in the disk's description (if a valid JSON doc) plus the tags arg
// to set the snapshot's description.
var snapshotTags map[string]string
if err := json.Unmarshal([]byte(diskDescription), &snapshotTags); err != nil {
// error decoding the disk's description, so just use the Velero-assigned tags
log.WithError(err).
Error("unable to decode disk's description as JSON, so only applying Velero-assigned tags to snapshot")
snapshotTags = veleroTags
} else {
// merge Velero-assigned tags with the disk's tags (note that we want current
// Velero-assigned tags to overwrite any older versions of them that may exist
// due to prior snapshots/restores)
for k, v := range veleroTags {
snapshotTags[k] = v
}
}
if len(snapshotTags) == 0 {
return ""
}
tagsJSON, err := json.Marshal(snapshotTags)
if err != nil {
log.WithError(err).Error("unable to encode snapshot's tags to JSON, so not tagging snapshot")
return ""
}
return string(tagsJSON)
}
func (b *VolumeSnapshotter) DeleteSnapshot(snapshotID string) error {
_, err := b.gce.Snapshots.Delete(b.snapshotProject, snapshotID).Do()
// if it's a 404 (not found) error, we don't need to return an error
// since the snapshot is not there.
if gcpErr, ok := err.(*googleapi.Error); ok && gcpErr.Code == http.StatusNotFound {
return nil
}
if err != nil {
return errors.WithStack(err)
}
return nil
}
func (b *VolumeSnapshotter) GetVolumeID(unstructuredPV runtime.Unstructured) (string, error) {
pv := new(v1.PersistentVolume)
if err := runtime.DefaultUnstructuredConverter.FromUnstructured(unstructuredPV.UnstructuredContent(), pv); err != nil {
return "", errors.WithStack(err)
}
if pv.Spec.GCEPersistentDisk == nil {
return "", nil
}
if pv.Spec.GCEPersistentDisk.PDName == "" {
return "", errors.New("spec.gcePersistentDisk.pdName not found")
}
return pv.Spec.GCEPersistentDisk.PDName, nil
}
func (b *VolumeSnapshotter) SetVolumeID(unstructuredPV runtime.Unstructured, volumeID string) (runtime.Unstructured, error) {
pv := new(v1.PersistentVolume)
if err := runtime.DefaultUnstructuredConverter.FromUnstructured(unstructuredPV.UnstructuredContent(), pv); err != nil {
return nil, errors.WithStack(err)
}
if pv.Spec.GCEPersistentDisk == nil {
return nil, errors.New("spec.gcePersistentDisk not found")
}
pv.Spec.GCEPersistentDisk.PDName = volumeID
res, err := runtime.DefaultUnstructuredConverter.ToUnstructured(pv)
if err != nil {
return nil, errors.WithStack(err)
}
return &unstructured.Unstructured{Object: res}, nil
}
@@ -1,217 +0,0 @@
/*
Copyright 2017 the Velero 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 gcp
import (
"encoding/json"
"testing"
"github.com/pkg/errors"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
v1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
"k8s.io/apimachinery/pkg/runtime"
velerotest "github.com/vmware-tanzu/velero/pkg/test"
)
func TestGetVolumeID(t *testing.T) {
b := &VolumeSnapshotter{}
pv := &unstructured.Unstructured{
Object: map[string]interface{}{},
}
// missing spec.gcePersistentDisk -> no error
volumeID, err := b.GetVolumeID(pv)
require.NoError(t, err)
assert.Equal(t, "", volumeID)
// missing spec.gcePersistentDisk.pdName -> error
gce := map[string]interface{}{}
pv.Object["spec"] = map[string]interface{}{
"gcePersistentDisk": gce,
}
volumeID, err = b.GetVolumeID(pv)
assert.Error(t, err)
assert.Equal(t, "", volumeID)
// valid
gce["pdName"] = "abc123"
volumeID, err = b.GetVolumeID(pv)
assert.NoError(t, err)
assert.Equal(t, "abc123", volumeID)
}
func TestSetVolumeID(t *testing.T) {
b := &VolumeSnapshotter{}
pv := &unstructured.Unstructured{
Object: map[string]interface{}{},
}
// missing spec.gcePersistentDisk -> error
updatedPV, err := b.SetVolumeID(pv, "abc123")
require.Error(t, err)
// happy path
gce := map[string]interface{}{}
pv.Object["spec"] = map[string]interface{}{
"gcePersistentDisk": gce,
}
updatedPV, err = b.SetVolumeID(pv, "123abc")
require.NoError(t, err)
res := new(v1.PersistentVolume)
require.NoError(t, runtime.DefaultUnstructuredConverter.FromUnstructured(updatedPV.UnstructuredContent(), res))
require.NotNil(t, res.Spec.GCEPersistentDisk)
assert.Equal(t, "123abc", res.Spec.GCEPersistentDisk.PDName)
}
func TestGetSnapshotTags(t *testing.T) {
tests := []struct {
name string
veleroTags map[string]string
diskDescription string
expected string
}{
{
name: "degenerate case (no tags)",
veleroTags: nil,
diskDescription: "",
expected: "",
},
{
name: "velero tags only get applied",
veleroTags: map[string]string{
"velero-key1": "velero-val1",
"velero-key2": "velero-val2",
},
diskDescription: "",
expected: `{"velero-key1":"velero-val1","velero-key2":"velero-val2"}`,
},
{
name: "disk tags only get applied",
veleroTags: nil,
diskDescription: `{"aws-key1":"aws-val1","aws-key2":"aws-val2"}`,
expected: `{"aws-key1":"aws-val1","aws-key2":"aws-val2"}`,
},
{
name: "non-overlapping velero and disk tags both get applied",
veleroTags: map[string]string{"velero-key": "velero-val"},
diskDescription: `{"aws-key":"aws-val"}`,
expected: `{"velero-key":"velero-val","aws-key":"aws-val"}`,
},
{
name: "when tags overlap, velero tags take precedence",
veleroTags: map[string]string{
"velero-key": "velero-val",
"overlapping-key": "velero-val",
},
diskDescription: `{"aws-key":"aws-val","overlapping-key":"aws-val"}`,
expected: `{"velero-key":"velero-val","aws-key":"aws-val","overlapping-key":"velero-val"}`,
},
{
name: "if disk description is invalid JSON, apply just velero tags",
veleroTags: map[string]string{"velero-key": "velero-val"},
diskDescription: `THIS IS INVALID JSON`,
expected: `{"velero-key":"velero-val"}`,
},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
res := getSnapshotTags(test.veleroTags, test.diskDescription, velerotest.NewLogger())
if test.expected == "" {
assert.Equal(t, test.expected, res)
return
}
var actualMap map[string]interface{}
require.NoError(t, json.Unmarshal([]byte(res), &actualMap))
var expectedMap map[string]interface{}
require.NoError(t, json.Unmarshal([]byte(test.expected), &expectedMap))
assert.Equal(t, len(expectedMap), len(actualMap))
for k, v := range expectedMap {
assert.Equal(t, v, actualMap[k])
}
})
}
}
func TestRegionHelpers(t *testing.T) {
tests := []struct {
name string
volumeAZ string
expectedRegion string
expectedIsMultiZone bool
expectedError error
}{
{
name: "valid multizone(2) tag",
volumeAZ: "us-central1-a__us-central1-b",
expectedRegion: "us-central1",
expectedIsMultiZone: true,
expectedError: nil,
},
{
name: "valid multizone(4) tag",
volumeAZ: "us-central1-a__us-central1-b__us-central1-f__us-central1-e",
expectedRegion: "us-central1",
expectedIsMultiZone: true,
expectedError: nil,
},
{
name: "valid single zone tag",
volumeAZ: "us-central1-a",
expectedRegion: "us-central1",
expectedIsMultiZone: false,
expectedError: nil,
},
{
name: "invalid single zone tag",
volumeAZ: "us^central1^a",
expectedRegion: "",
expectedIsMultiZone: false,
expectedError: errors.Errorf("failed to parse region from zone: %q", "us^central1^a"),
},
{
name: "invalid multizone tag",
volumeAZ: "us^central1^a__us^central1^b",
expectedRegion: "",
expectedIsMultiZone: true,
expectedError: errors.Errorf("failed to parse region from zone: %q", "us^central1^a__us^central1^b"),
},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
assert.Equal(t, test.expectedIsMultiZone, isMultiZone(test.volumeAZ))
region, err := parseRegion(test.volumeAZ)
if test.expectedError == nil {
assert.NoError(t, err)
} else {
assert.Equal(t, test.expectedError.Error(), err.Error())
}
assert.Equal(t, test.expectedRegion, region)
})
}
}
+5 -1
View File
@@ -95,7 +95,7 @@ func (o *InstallOptions) BindFlags(flags *pflag.FlagSet) {
flags.BoolVar(&o.UseRestic, "use-restic", o.UseRestic, "create restic deployment. Optional.")
flags.BoolVar(&o.Wait, "wait", o.Wait, "wait for Velero deployment to be ready. Optional.")
flags.DurationVar(&o.DefaultResticMaintenanceFrequency, "default-restic-prune-frequency", o.DefaultResticMaintenanceFrequency, "how often 'restic prune' is run for restic repositories by default. Optional.")
flags.Var(&o.Plugins, "plugins", "Plugin container images to install into the Velero Deployment. Optional.")
flags.Var(&o.Plugins, "plugins", "Plugin container images to install into the Velero Deployment")
}
// NewInstallOptions instantiates a new, default InstallOptions struct.
@@ -334,6 +334,10 @@ func (o *InstallOptions) Validate(c *cobra.Command, args []string, f client.Fact
if o.ProviderName != "" {
return errors.New("--provider must be empty when using --no-default-backup-location and --use-volume-snapshots=false")
}
} else {
if len(o.Plugins) == 0 {
return errors.New("--plugins flag is required")
}
}
switch {
-33
View File
@@ -22,9 +22,6 @@ import (
"github.com/vmware-tanzu/velero/pkg/backup"
"github.com/vmware-tanzu/velero/pkg/client"
"github.com/vmware-tanzu/velero/pkg/cloudprovider/aws"
"github.com/vmware-tanzu/velero/pkg/cloudprovider/azure"
"github.com/vmware-tanzu/velero/pkg/cloudprovider/gcp"
velerodiscovery "github.com/vmware-tanzu/velero/pkg/discovery"
veleroplugin "github.com/vmware-tanzu/velero/pkg/plugin/framework"
"github.com/vmware-tanzu/velero/pkg/restore"
@@ -38,12 +35,6 @@ func NewCommand(f client.Factory) *cobra.Command {
Short: "INTERNAL COMMAND ONLY - not intended to be run directly by users",
Run: func(c *cobra.Command, args []string) {
pluginServer.
RegisterObjectStore("velero.io/aws", newAwsObjectStore).
RegisterObjectStore("velero.io/azure", newAzureObjectStore).
RegisterObjectStore("velero.io/gcp", newGcpObjectStore).
RegisterVolumeSnapshotter("velero.io/aws", newAwsVolumeSnapshotter).
RegisterVolumeSnapshotter("velero.io/azure", newAzureVolumeSnapshotter).
RegisterVolumeSnapshotter("velero.io/gcp", newGcpVolumeSnapshotter).
RegisterBackupItemAction("velero.io/pv", newPVBackupItemAction).
RegisterBackupItemAction("velero.io/pod", newPodBackupItemAction).
RegisterBackupItemAction("velero.io/service-account", newServiceAccountBackupItemAction(f)).
@@ -66,30 +57,6 @@ func NewCommand(f client.Factory) *cobra.Command {
return c
}
func newAwsObjectStore(logger logrus.FieldLogger) (interface{}, error) {
return aws.NewObjectStore(logger), nil
}
func newAzureObjectStore(logger logrus.FieldLogger) (interface{}, error) {
return azure.NewObjectStore(logger), nil
}
func newGcpObjectStore(logger logrus.FieldLogger) (interface{}, error) {
return gcp.NewObjectStore(logger), nil
}
func newAwsVolumeSnapshotter(logger logrus.FieldLogger) (interface{}, error) {
return aws.NewVolumeSnapshotter(logger), nil
}
func newAzureVolumeSnapshotter(logger logrus.FieldLogger) (interface{}, error) {
return azure.NewVolumeSnapshotter(logger), nil
}
func newGcpVolumeSnapshotter(logger logrus.FieldLogger) (interface{}, error) {
return gcp.NewVolumeSnapshotter(logger), nil
}
func newPVBackupItemAction(logger logrus.FieldLogger) (interface{}, error) {
return backup.NewPVCAction(logger), nil
}
@@ -1,12 +1,12 @@
/*
Copyright 2018 the Velero contributors.
Copyright 2018, 2019 the Velero 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.
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
limitations under the License.
*/
package cloudprovider
package persistence
import (
"bytes"
@@ -27,15 +27,15 @@ import (
type BucketData map[string][]byte
// InMemoryObjectStore is a simple implementation of the ObjectStore interface
// inMemoryObjectStore is a simple implementation of the ObjectStore interface
// that stores its data in-memory/in-proc. This is mainly intended to be used
// as a test fake.
type InMemoryObjectStore struct {
type inMemoryObjectStore struct {
Data map[string]BucketData
}
func NewInMemoryObjectStore(buckets ...string) *InMemoryObjectStore {
o := &InMemoryObjectStore{
func newInMemoryObjectStore(buckets ...string) *inMemoryObjectStore {
o := &inMemoryObjectStore{
Data: make(map[string]BucketData),
}
@@ -50,11 +50,11 @@ func NewInMemoryObjectStore(buckets ...string) *InMemoryObjectStore {
// Interface Implementation
//
func (o *InMemoryObjectStore) Init(config map[string]string) error {
func (o *inMemoryObjectStore) Init(config map[string]string) error {
return nil
}
func (o *InMemoryObjectStore) PutObject(bucket, key string, body io.Reader) error {
func (o *inMemoryObjectStore) PutObject(bucket, key string, body io.Reader) error {
bucketData, ok := o.Data[bucket]
if !ok {
return errors.New("bucket not found")
@@ -70,7 +70,7 @@ func (o *InMemoryObjectStore) PutObject(bucket, key string, body io.Reader) erro
return nil
}
func (o *InMemoryObjectStore) ObjectExists(bucket, key string) (bool, error) {
func (o *inMemoryObjectStore) ObjectExists(bucket, key string) (bool, error) {
bucketData, ok := o.Data[bucket]
if !ok {
return false, errors.New("bucket not found")
@@ -80,7 +80,7 @@ func (o *InMemoryObjectStore) ObjectExists(bucket, key string) (bool, error) {
return ok, nil
}
func (o *InMemoryObjectStore) GetObject(bucket, key string) (io.ReadCloser, error) {
func (o *inMemoryObjectStore) GetObject(bucket, key string) (io.ReadCloser, error) {
bucketData, ok := o.Data[bucket]
if !ok {
return nil, errors.New("bucket not found")
@@ -94,7 +94,7 @@ func (o *InMemoryObjectStore) GetObject(bucket, key string) (io.ReadCloser, erro
return ioutil.NopCloser(bytes.NewReader(obj)), nil
}
func (o *InMemoryObjectStore) ListCommonPrefixes(bucket, prefix, delimiter string) ([]string, error) {
func (o *inMemoryObjectStore) ListCommonPrefixes(bucket, prefix, delimiter string) ([]string, error) {
keys, err := o.ListObjects(bucket, prefix)
if err != nil {
return nil, err
@@ -124,7 +124,7 @@ func (o *InMemoryObjectStore) ListCommonPrefixes(bucket, prefix, delimiter strin
return prefixes, nil
}
func (o *InMemoryObjectStore) ListObjects(bucket, prefix string) ([]string, error) {
func (o *inMemoryObjectStore) ListObjects(bucket, prefix string) ([]string, error) {
bucketData, ok := o.Data[bucket]
if !ok {
return nil, errors.New("bucket not found")
@@ -140,7 +140,7 @@ func (o *InMemoryObjectStore) ListObjects(bucket, prefix string) ([]string, erro
return objs, nil
}
func (o *InMemoryObjectStore) DeleteObject(bucket, key string) error {
func (o *inMemoryObjectStore) DeleteObject(bucket, key string) error {
bucketData, ok := o.Data[bucket]
if !ok {
return errors.New("bucket not found")
@@ -151,7 +151,7 @@ func (o *InMemoryObjectStore) DeleteObject(bucket, key string) error {
return nil
}
func (o *InMemoryObjectStore) CreateSignedURL(bucket, key string, ttl time.Duration) (string, error) {
func (o *inMemoryObjectStore) CreateSignedURL(bucket, key string, ttl time.Duration) (string, error) {
bucketData, ok := o.Data[bucket]
if !ok {
return "", errors.New("bucket not found")
@@ -169,7 +169,7 @@ func (o *InMemoryObjectStore) CreateSignedURL(bucket, key string, ttl time.Durat
// Test Helper Methods
//
func (o *InMemoryObjectStore) ClearBucket(bucket string) {
func (o *inMemoryObjectStore) ClearBucket(bucket string) {
if _, ok := o.Data[bucket]; !ok {
return
}
+9 -10
View File
@@ -34,9 +34,8 @@ import (
velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
"github.com/vmware-tanzu/velero/pkg/builder"
"github.com/vmware-tanzu/velero/pkg/cloudprovider"
cloudprovidermocks "github.com/vmware-tanzu/velero/pkg/cloudprovider/mocks"
"github.com/vmware-tanzu/velero/pkg/plugin/velero"
providermocks "github.com/vmware-tanzu/velero/pkg/plugin/velero/mocks"
velerotest "github.com/vmware-tanzu/velero/pkg/test"
"github.com/vmware-tanzu/velero/pkg/util/encode"
"github.com/vmware-tanzu/velero/pkg/volume"
@@ -46,12 +45,12 @@ type objectBackupStoreTestHarness struct {
// embedded to reduce verbosity when calling methods
*objectBackupStore
objectStore *cloudprovider.InMemoryObjectStore
objectStore *inMemoryObjectStore
bucket, prefix string
}
func newObjectBackupStoreTestHarness(bucket, prefix string) *objectBackupStoreTestHarness {
objectStore := cloudprovider.NewInMemoryObjectStore(bucket)
objectStore := newInMemoryObjectStore(bucket)
return &objectBackupStoreTestHarness{
objectBackupStore: &objectBackupStore{
@@ -70,7 +69,7 @@ func TestIsValid(t *testing.T) {
tests := []struct {
name string
prefix string
storageData cloudprovider.BucketData
storageData BucketData
expectErr bool
}{
{
@@ -163,7 +162,7 @@ func TestListBackups(t *testing.T) {
tests := []struct {
name string
prefix string
storageData cloudprovider.BucketData
storageData BucketData
expectedRes []string
expectedErr string
}{
@@ -456,7 +455,7 @@ func TestDeleteBackup(t *testing.T) {
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
objectStore := new(cloudprovidermocks.ObjectStore)
objectStore := new(providermocks.ObjectStore)
backupStore := &objectBackupStore{
objectStore: objectStore,
bucket: "test-bucket",
@@ -629,7 +628,7 @@ func TestNewObjectBackupStore(t *testing.T) {
name: "when Bucket has a leading and trailing slash, they are both stripped",
location: builder.ForBackupStorageLocation("", "").Provider("provider-1").Bucket("/bucket/").Result(),
objectStoreGetter: objectStoreGetter{
"provider-1": cloudprovider.NewInMemoryObjectStore("bucket"),
"provider-1": newInMemoryObjectStore("bucket"),
},
wantBucket: "bucket",
},
@@ -637,7 +636,7 @@ func TestNewObjectBackupStore(t *testing.T) {
name: "when Prefix has a leading and trailing slash, the leading slash is stripped and the trailing slash is left",
location: builder.ForBackupStorageLocation("", "").Provider("provider-1").Bucket("bucket").Prefix("/prefix/").Result(),
objectStoreGetter: objectStoreGetter{
"provider-1": cloudprovider.NewInMemoryObjectStore("bucket"),
"provider-1": newInMemoryObjectStore("bucket"),
},
wantBucket: "bucket",
wantPrefix: "prefix/",
@@ -646,7 +645,7 @@ func TestNewObjectBackupStore(t *testing.T) {
name: "when Prefix has no leading or trailing slash, a trailing slash is added",
location: builder.ForBackupStorageLocation("", "").Provider("provider-1").Bucket("bucket").Prefix("prefix").Result(),
objectStoreGetter: objectStoreGetter{
"provider-1": cloudprovider.NewInMemoryObjectStore("bucket"),
"provider-1": newInMemoryObjectStore("bucket"),
},
wantBucket: "bucket",
wantPrefix: "prefix/",
+1 -1
View File
@@ -21,7 +21,7 @@ import (
"os"
"os/exec"
"github.com/hashicorp/go-hclog"
hclog "github.com/hashicorp/go-hclog"
hcplugin "github.com/hashicorp/go-plugin"
"github.com/sirupsen/logrus"
@@ -26,8 +26,8 @@ import (
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
cloudprovidermocks "github.com/vmware-tanzu/velero/pkg/cloudprovider/mocks"
"github.com/vmware-tanzu/velero/pkg/plugin/framework"
providermocks "github.com/vmware-tanzu/velero/pkg/plugin/velero/mocks"
)
func TestRestartableGetObjectStore(t *testing.T) {
@@ -49,7 +49,7 @@ func TestRestartableGetObjectStore(t *testing.T) {
},
{
name: "happy path",
plugin: new(cloudprovidermocks.ObjectStore),
plugin: new(providermocks.ObjectStore),
},
}
@@ -97,7 +97,7 @@ func TestRestartableObjectStoreReinitialize(t *testing.T) {
err := r.reinitialize(3)
assert.EqualError(t, err, "int is not a ObjectStore!")
objectStore := new(cloudprovidermocks.ObjectStore)
objectStore := new(providermocks.ObjectStore)
objectStore.Test(t)
defer objectStore.AssertExpectations(t)
@@ -129,7 +129,7 @@ func TestRestartableObjectStoreGetDelegate(t *testing.T) {
// Happy path
p.On("resetIfNeeded").Return(nil)
objectStore := new(cloudprovidermocks.ObjectStore)
objectStore := new(providermocks.ObjectStore)
objectStore.Test(t)
defer objectStore.AssertExpectations(t)
p.On("getByKindAndName", key).Return(objectStore, nil)
@@ -160,7 +160,7 @@ func TestRestartableObjectStoreInit(t *testing.T) {
assert.EqualError(t, err, "getByKindAndName error")
// Delegate returns error
objectStore := new(cloudprovidermocks.ObjectStore)
objectStore := new(providermocks.ObjectStore)
objectStore.Test(t)
defer objectStore.AssertExpectations(t)
p.On("getByKindAndName", key).Return(objectStore, nil)
@@ -194,7 +194,7 @@ func TestRestartableObjectStoreDelegatedFunctions(t *testing.T) {
}
},
func() mockable {
return new(cloudprovidermocks.ObjectStore)
return new(providermocks.ObjectStore)
},
restartableDelegateTest{
function: "PutObject",
@@ -25,8 +25,8 @@ import (
"github.com/stretchr/testify/require"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
"github.com/vmware-tanzu/velero/pkg/cloudprovider/mocks"
"github.com/vmware-tanzu/velero/pkg/plugin/framework"
providermocks "github.com/vmware-tanzu/velero/pkg/plugin/velero/mocks"
)
func TestRestartableGetVolumeSnapshotter(t *testing.T) {
@@ -48,7 +48,7 @@ func TestRestartableGetVolumeSnapshotter(t *testing.T) {
},
{
name: "happy path",
plugin: new(mocks.VolumeSnapshotter),
plugin: new(providermocks.VolumeSnapshotter),
},
}
@@ -96,7 +96,7 @@ func TestRestartableVolumeSnapshotterReinitialize(t *testing.T) {
err := r.reinitialize(3)
assert.EqualError(t, err, "int is not a VolumeSnapshotter!")
volumeSnapshotter := new(mocks.VolumeSnapshotter)
volumeSnapshotter := new(providermocks.VolumeSnapshotter)
volumeSnapshotter.Test(t)
defer volumeSnapshotter.AssertExpectations(t)
@@ -128,7 +128,7 @@ func TestRestartableVolumeSnapshotterGetDelegate(t *testing.T) {
// Happy path
p.On("resetIfNeeded").Return(nil)
volumeSnapshotter := new(mocks.VolumeSnapshotter)
volumeSnapshotter := new(providermocks.VolumeSnapshotter)
volumeSnapshotter.Test(t)
defer volumeSnapshotter.AssertExpectations(t)
p.On("getByKindAndName", key).Return(volumeSnapshotter, nil)
@@ -159,7 +159,7 @@ func TestRestartableVolumeSnapshotterInit(t *testing.T) {
assert.EqualError(t, err, "getByKindAndName error")
// Delegate returns error
volumeSnapshotter := new(mocks.VolumeSnapshotter)
volumeSnapshotter := new(providermocks.VolumeSnapshotter)
volumeSnapshotter.Test(t)
defer volumeSnapshotter.AssertExpectations(t)
p.On("getByKindAndName", key).Return(volumeSnapshotter, nil)
@@ -205,7 +205,7 @@ func TestRestartableVolumeSnapshotterDelegatedFunctions(t *testing.T) {
}
},
func() mockable {
return new(mocks.VolumeSnapshotter)
return new(providermocks.VolumeSnapshotter)
},
restartableDelegateTest{
function: "CreateVolumeFromSnapshot",
+1 -1
View File
@@ -17,7 +17,7 @@ limitations under the License.
package framework
import (
"github.com/hashicorp/go-plugin"
plugin "github.com/hashicorp/go-plugin"
"golang.org/x/net/context"
"google.golang.org/grpc"
+1 -1
View File
@@ -17,7 +17,7 @@ limitations under the License.
package framework
import (
"github.com/hashicorp/go-plugin"
plugin "github.com/hashicorp/go-plugin"
"golang.org/x/net/context"
"google.golang.org/grpc"
+1 -1
View File
@@ -19,7 +19,7 @@ package framework
import (
"testing"
"github.com/hashicorp/go-plugin"
plugin "github.com/hashicorp/go-plugin"
"github.com/stretchr/testify/assert"
)
+1 -1
View File
@@ -17,7 +17,7 @@ limitations under the License.
package framework
import (
"github.com/hashicorp/go-plugin"
plugin "github.com/hashicorp/go-plugin"
"golang.org/x/net/context"
"google.golang.org/grpc"
+1 -1
View File
@@ -17,7 +17,7 @@ limitations under the License.
package framework
import (
"github.com/hashicorp/go-plugin"
plugin "github.com/hashicorp/go-plugin"
"golang.org/x/net/context"
"google.golang.org/grpc"
+165
View File
@@ -0,0 +1,165 @@
// Code generated by mockery v1.0.0. DO NOT EDIT.
package mocks
import io "io"
import mock "github.com/stretchr/testify/mock"
import time "time"
// ObjectStore is an autogenerated mock type for the ObjectStore type
type ObjectStore struct {
mock.Mock
}
// CreateSignedURL provides a mock function with given fields: bucket, key, ttl
func (_m *ObjectStore) CreateSignedURL(bucket string, key string, ttl time.Duration) (string, error) {
ret := _m.Called(bucket, key, ttl)
var r0 string
if rf, ok := ret.Get(0).(func(string, string, time.Duration) string); ok {
r0 = rf(bucket, key, ttl)
} else {
r0 = ret.Get(0).(string)
}
var r1 error
if rf, ok := ret.Get(1).(func(string, string, time.Duration) error); ok {
r1 = rf(bucket, key, ttl)
} else {
r1 = ret.Error(1)
}
return r0, r1
}
// DeleteObject provides a mock function with given fields: bucket, key
func (_m *ObjectStore) DeleteObject(bucket string, key string) error {
ret := _m.Called(bucket, key)
var r0 error
if rf, ok := ret.Get(0).(func(string, string) error); ok {
r0 = rf(bucket, key)
} else {
r0 = ret.Error(0)
}
return r0
}
// GetObject provides a mock function with given fields: bucket, key
func (_m *ObjectStore) GetObject(bucket string, key string) (io.ReadCloser, error) {
ret := _m.Called(bucket, key)
var r0 io.ReadCloser
if rf, ok := ret.Get(0).(func(string, string) io.ReadCloser); ok {
r0 = rf(bucket, key)
} else {
if ret.Get(0) != nil {
r0 = ret.Get(0).(io.ReadCloser)
}
}
var r1 error
if rf, ok := ret.Get(1).(func(string, string) error); ok {
r1 = rf(bucket, key)
} else {
r1 = ret.Error(1)
}
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, prefix, delimiter
func (_m *ObjectStore) ListCommonPrefixes(bucket string, prefix string, delimiter string) ([]string, error) {
ret := _m.Called(bucket, prefix, delimiter)
var r0 []string
if rf, ok := ret.Get(0).(func(string, string, string) []string); ok {
r0 = rf(bucket, prefix, delimiter)
} else {
if ret.Get(0) != nil {
r0 = ret.Get(0).([]string)
}
}
var r1 error
if rf, ok := ret.Get(1).(func(string, string, string) error); ok {
r1 = rf(bucket, prefix, delimiter)
} else {
r1 = ret.Error(1)
}
return r0, r1
}
// ListObjects provides a mock function with given fields: bucket, prefix
func (_m *ObjectStore) ListObjects(bucket string, prefix string) ([]string, error) {
ret := _m.Called(bucket, prefix)
var r0 []string
if rf, ok := ret.Get(0).(func(string, string) []string); ok {
r0 = rf(bucket, prefix)
} else {
if ret.Get(0) != nil {
r0 = ret.Get(0).([]string)
}
}
var r1 error
if rf, ok := ret.Get(1).(func(string, string) error); ok {
r1 = rf(bucket, prefix)
} else {
r1 = ret.Error(1)
}
return r0, r1
}
// ObjectExists provides a mock function with given fields: bucket, key
func (_m *ObjectStore) ObjectExists(bucket string, key string) (bool, error) {
ret := _m.Called(bucket, key)
var r0 bool
if rf, ok := ret.Get(0).(func(string, string) bool); ok {
r0 = rf(bucket, key)
} else {
r0 = ret.Get(0).(bool)
}
var r1 error
if rf, ok := ret.Get(1).(func(string, string) error); ok {
r1 = rf(bucket, key)
} else {
r1 = ret.Error(1)
}
return r0, r1
}
// PutObject provides a mock function with given fields: bucket, key, body
func (_m *ObjectStore) PutObject(bucket string, key string, body io.Reader) error {
ret := _m.Called(bucket, key, body)
var r0 error
if rf, ok := ret.Get(0).(func(string, string, io.Reader) error); ok {
r0 = rf(bucket, key, body)
} else {
r0 = ret.Error(0)
}
return r0
}
@@ -1,20 +1,5 @@
/*
Copyright 2018 the Velero 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.
*/
// Code generated by mockery v1.0.0. DO NOT EDIT.
package mocks
import mock "github.com/stretchr/testify/mock"
+182
View File
@@ -0,0 +1,182 @@
/*
Copyright 2017, 2019 the Velero 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 restic
import (
"context"
"os"
"strings"
storagemgmt "github.com/Azure/azure-sdk-for-go/services/storage/mgmt/2018-02-01/storage"
"github.com/Azure/go-autorest/autorest"
"github.com/Azure/go-autorest/autorest/adal"
"github.com/Azure/go-autorest/autorest/azure"
"github.com/joho/godotenv"
"github.com/pkg/errors"
)
const (
tenantIDEnvVar = "AZURE_TENANT_ID"
subscriptionIDEnvVar = "AZURE_SUBSCRIPTION_ID"
clientIDEnvVar = "AZURE_CLIENT_ID"
clientSecretEnvVar = "AZURE_CLIENT_SECRET"
cloudNameEnvVar = "AZURE_CLOUD_NAME"
resourceGroupConfigKey = "resourceGroup"
storageAccountConfigKey = "storageAccount"
subscriptionIdConfigKey = "subscriptionId"
)
func getStorageAccountKey(config map[string]string) (string, *azure.Environment, error) {
// load environment vars from $AZURE_CREDENTIALS_FILE, if it exists
if err := loadEnv(); err != nil {
return "", nil, err
}
// 1. we need AZURE_TENANT_ID, AZURE_CLIENT_ID, AZURE_CLIENT_SECRET, AZURE_SUBSCRIPTION_ID
envVars, err := getRequiredValues(os.Getenv, tenantIDEnvVar, clientIDEnvVar, clientSecretEnvVar, subscriptionIDEnvVar)
if err != nil {
return "", nil, errors.Wrap(err, "unable to get all required environment variables")
}
// 2. Get Azure cloud from AZURE_CLOUD_NAME, if it exists. If the env var does not
// exist, parseAzureEnvironment will return azure.PublicCloud.
env, err := parseAzureEnvironment(os.Getenv(cloudNameEnvVar))
if err != nil {
return "", nil, errors.Wrap(err, "unable to parse azure cloud name environment variable")
}
// 3. check whether a different subscription ID was set for backups in config["subscriptionId"]
subscriptionId := envVars[subscriptionIDEnvVar]
if val := config[subscriptionIdConfigKey]; val != "" {
subscriptionId = val
}
// 4. we need config["resourceGroup"], config["storageAccount"]
if _, err := getRequiredValues(mapLookup(config), resourceGroupConfigKey, storageAccountConfigKey); err != nil {
return "", env, errors.Wrap(err, "unable to get all required config values")
}
// 5. get SPT
spt, err := newServicePrincipalToken(envVars[tenantIDEnvVar], envVars[clientIDEnvVar], envVars[clientSecretEnvVar], env)
if err != nil {
return "", env, errors.Wrap(err, "error getting service principal token")
}
// 6. get storageAccountsClient
storageAccountsClient := storagemgmt.NewAccountsClientWithBaseURI(env.ResourceManagerEndpoint, subscriptionId)
storageAccountsClient.Authorizer = autorest.NewBearerAuthorizer(spt)
// 7. get storage key
res, err := storageAccountsClient.ListKeys(context.TODO(), config[resourceGroupConfigKey], config[storageAccountConfigKey])
if err != nil {
return "", env, errors.WithStack(err)
}
if res.Keys == nil || len(*res.Keys) == 0 {
return "", env, errors.New("No storage keys found")
}
var storageKey string
for _, key := range *res.Keys {
// uppercase both strings for comparison because the ListKeys call returns e.g. "FULL" but
// the storagemgmt.Full constant in the SDK is defined as "Full".
if strings.ToUpper(string(key.Permissions)) == strings.ToUpper(string(storagemgmt.Full)) {
storageKey = *key.Value
break
}
}
if storageKey == "" {
return "", env, errors.New("No storage key with Full permissions found")
}
return storageKey, env, nil
}
func mapLookup(data map[string]string) func(string) string {
return func(key string) string {
return data[key]
}
}
// getResticEnvVars gets the environment variables that restic
// relies on (AZURE_ACCOUNT_NAME and AZURE_ACCOUNT_KEY) based
// on info in the provided object storage location config map.
func getResticEnvVars(config map[string]string) (map[string]string, error) {
storageAccountKey, _, err := getStorageAccountKey(config)
if err != nil {
return nil, err
}
return map[string]string{
"AZURE_ACCOUNT_NAME": config[storageAccountConfigKey],
"AZURE_ACCOUNT_KEY": storageAccountKey,
}, nil
}
func loadEnv() error {
envFile := os.Getenv("AZURE_CREDENTIALS_FILE")
if envFile == "" {
return nil
}
if err := godotenv.Overload(envFile); err != nil {
return errors.Wrapf(err, "error loading environment from AZURE_CREDENTIALS_FILE (%s)", envFile)
}
return nil
}
// ParseAzureEnvironment returns an azure.Environment for the given cloud
// name, or azure.PublicCloud if cloudName is empty.
func parseAzureEnvironment(cloudName string) (*azure.Environment, error) {
if cloudName == "" {
return &azure.PublicCloud, nil
}
env, err := azure.EnvironmentFromName(cloudName)
return &env, errors.WithStack(err)
}
func newServicePrincipalToken(tenantID, clientID, clientSecret string, env *azure.Environment) (*adal.ServicePrincipalToken, error) {
oauthConfig, err := adal.NewOAuthConfig(env.ActiveDirectoryEndpoint, tenantID)
if err != nil {
return nil, errors.Wrap(err, "error getting OAuthConfig")
}
return adal.NewServicePrincipalToken(*oauthConfig, clientID, clientSecret, env.ResourceManagerEndpoint)
}
func getRequiredValues(getValue func(string) string, keys ...string) (map[string]string, error) {
missing := []string{}
results := map[string]string{}
for _, key := range keys {
if val := getValue(key); val == "" {
missing = append(missing, key)
} else {
results[key] = val
}
}
if len(missing) > 0 {
return nil, errors.Errorf("the following keys do not have values: %s", strings.Join(missing, ", "))
}
return results, nil
}
+1 -2
View File
@@ -28,7 +28,6 @@ import (
corev1listers "k8s.io/client-go/listers/core/v1"
velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
"github.com/vmware-tanzu/velero/pkg/cloudprovider/azure"
velerov1listers "github.com/vmware-tanzu/velero/pkg/generated/listers/velero/v1"
"github.com/vmware-tanzu/velero/pkg/label"
"github.com/vmware-tanzu/velero/pkg/util/filesystem"
@@ -221,7 +220,7 @@ func AzureCmdEnv(backupLocationLister velerov1listers.BackupStorageLocationListe
return nil, errors.Wrap(err, "error getting backup storage location")
}
azureVars, err := azure.GetResticEnvVars(loc.Spec.Config)
azureVars, err := getResticEnvVars(loc.Spec.Config)
if err != nil {
return nil, errors.Wrap(err, "error getting azure restic env vars")
}
+31 -2
View File
@@ -17,14 +17,17 @@ limitations under the License.
package restic
import (
"context"
"fmt"
"path"
"strings"
"github.com/aws/aws-sdk-go/aws/endpoints"
"github.com/aws/aws-sdk-go/aws/session"
"github.com/aws/aws-sdk-go/service/s3/s3manager"
"github.com/pkg/errors"
velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
"github.com/vmware-tanzu/velero/pkg/cloudprovider/aws"
"github.com/vmware-tanzu/velero/pkg/persistence"
)
@@ -38,7 +41,7 @@ const (
// this func is assigned to a package-level variable so it can be
// replaced when unit-testing
var getAWSBucketRegion = aws.GetBucketRegion
var getAWSBucketRegion = getBucketRegion
// getRepoPrefix returns the prefix of the value of the --repo flag for
// restic commands, i.e. everything except the "/<repo-name>".
@@ -98,3 +101,29 @@ func GetRepoIdentifier(location *velerov1api.BackupStorageLocation, name string)
return fmt.Sprintf("%s/%s", strings.TrimSuffix(prefix, "/"), name), nil
}
// getBucketRegion returns the AWS region that a bucket is in, or an error
// if the region cannot be determined.
func getBucketRegion(bucket string) (string, error) {
var region string
session, err := session.NewSession()
if err != nil {
return "", errors.WithStack(err)
}
for _, partition := range endpoints.DefaultPartitions() {
for regionHint := range partition.Regions() {
region, _ = s3manager.GetBucketRegion(context.Background(), session, bucket, regionHint)
// we only need to try a single region hint per partition, so break after the first
break
}
if region != "" {
return region, nil
}
}
return "", errors.New("unable to determine bucket's region")
}
+2 -2
View File
@@ -27,10 +27,10 @@ import (
api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
"github.com/vmware-tanzu/velero/pkg/builder"
cloudprovidermocks "github.com/vmware-tanzu/velero/pkg/cloudprovider/mocks"
"github.com/vmware-tanzu/velero/pkg/generated/clientset/versioned/fake"
informers "github.com/vmware-tanzu/velero/pkg/generated/informers/externalversions"
"github.com/vmware-tanzu/velero/pkg/plugin/velero"
providermocks "github.com/vmware-tanzu/velero/pkg/plugin/velero/mocks"
velerotest "github.com/vmware-tanzu/velero/pkg/test"
"github.com/vmware-tanzu/velero/pkg/volume"
)
@@ -199,7 +199,7 @@ func TestExecutePVAction_SnapshotRestores(t *testing.T) {
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
var (
volumeSnapshotter = new(cloudprovidermocks.VolumeSnapshotter)
volumeSnapshotter = new(providermocks.VolumeSnapshotter)
volumeSnapshotterGetter = providerToVolumeSnapshotterMap(map[string]velero.VolumeSnapshotter{
tc.expectedProvider: volumeSnapshotter,
})