Add Google Object Store Support to Kanister (#4923)

* Add Google Object Store Support to Kanister

* Added unit test

* Address review comments, add unit tests

* undo kando location push changes to get size

pkg/function/data_test.go

@@ -4,17 +4,19 @@ import (
 	"context"
 	"fmt"
 
+	. "gopkg.in/check.v1"
+	"k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+
 	kanister "github.com/kanisterio/kanister/pkg"
 	crv1alpha1 "github.com/kanisterio/kanister/pkg/apis/cr/v1alpha1"
 	"github.com/kanisterio/kanister/pkg/client/clientset/versioned"
 	"github.com/kanisterio/kanister/pkg/kube"
+	"github.com/kanisterio/kanister/pkg/objectstore"
 	"github.com/kanisterio/kanister/pkg/param"
 	"github.com/kanisterio/kanister/pkg/resource"
 	"github.com/kanisterio/kanister/pkg/testutil"
-	. "gopkg.in/check.v1"
-	"k8s.io/api/core/v1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/client-go/kubernetes"
 )
 
 type DataSuite struct {
@@ -165,7 +167,11 @@ func (s *DataSuite) TestBackupRestoreData(c *C) {
 	tp, err := param.New(ctx, s.cli, s.crCli, as)
 	c.Assert(err, IsNil)
 
-	tp.Profile = testutil.ObjectStoreProfileOrSkip(c)
+	location := crv1alpha1.Location{
+		Type:   crv1alpha1.LocationTypeS3Compliant,
+		Bucket: testutil.GetEnvOrSkip(c, testutil.TestS3BucketName),
+	}
+	tp.Profile = testutil.ObjectStoreProfileOrSkip(c, objectstore.ProviderTypeS3, location)
 
 	for _, bp := range []crv1alpha1.Blueprint{
 		*newBackupDataBlueprint(),
@@ -314,6 +320,11 @@ func (s *DataSuite) initPVCTemplateParams(c *C, pvc *v1.PersistentVolumeClaim, o
 	}
 	tp, err := param.New(context.Background(), s.cli, s.crCli, as)
 	c.Assert(err, IsNil)
-	tp.Profile = testutil.ObjectStoreProfileOrSkip(c)
+
+	location := crv1alpha1.Location{
+		Type:   crv1alpha1.LocationTypeS3Compliant,
+		Bucket: testutil.GetEnvOrSkip(c, testutil.TestS3BucketName),
+	}
+	tp.Profile = testutil.ObjectStoreProfileOrSkip(c, objectstore.ProviderTypeS3, location)
 	return tp
 }

pkg/kando/location_test.go

@@ -7,6 +7,8 @@ import (
 
 	. "gopkg.in/check.v1"
 
+	crv1alpha1 "github.com/kanisterio/kanister/pkg/apis/cr/v1alpha1"
+	"github.com/kanisterio/kanister/pkg/objectstore"
 	"github.com/kanisterio/kanister/pkg/testutil"
 )
 
@@ -17,7 +19,11 @@ var _ = Suite(&LocationSuite{})
 const testContent = "test-content"
 
 func (s *LocationSuite) TestLocationObjectStore(c *C) {
-	p := testutil.ObjectStoreProfileOrSkip(c)
+	location := crv1alpha1.Location{
+		Type:   crv1alpha1.LocationTypeS3Compliant,
+		Bucket: testutil.GetEnvOrSkip(c, testutil.TestS3BucketName),
+	}
+	p := testutil.ObjectStoreProfileOrSkip(c, objectstore.ProviderTypeS3, location)
 	ctx := context.Background()
 	dir := c.MkDir()
 	path := filepath.Join(dir, "test-object1.txt")

pkg/location/location.go

@@ -14,6 +14,7 @@ import (
 	log "github.com/sirupsen/logrus"
 
 	crv1alpha1 "github.com/kanisterio/kanister/pkg/apis/cr/v1alpha1"
+	"github.com/kanisterio/kanister/pkg/objectstore"
 	"github.com/kanisterio/kanister/pkg/param"
 )
 
@@ -25,6 +26,12 @@ func Write(ctx context.Context, in io.Reader, profile param.Profile, suffix stri
 		args := s3CompliantWriteArgs(profile, suffix)
 		env := s3CompliantEnv(profile)
 		return writeExec(ctx, in, bin, args, env)
+	case crv1alpha1.LocationTypeGCS:
+		path := filepath.Join(
+			profile.Location.Prefix,
+			suffix,
+		)
+		return writeData(ctx, objectstore.ProviderTypeGCS, profile, in, path)
 	}
 	return errors.Errorf("Unsupported Location type: %s", profile.Location.Type)
 }
@@ -37,6 +44,12 @@ func Read(ctx context.Context, out io.Writer, profile param.Profile, suffix stri
 		args := s3CompliantReadArgs(profile, suffix)
 		env := s3CompliantEnv(profile)
 		return readExec(ctx, out, bin, args, env)
+	case crv1alpha1.LocationTypeGCS:
+		path := filepath.Join(
+			profile.Location.Prefix,
+			suffix,
+		)
+		return readData(ctx, objectstore.ProviderTypeGCS, profile, out, path)
 	}
 	return errors.Errorf("Unsupported Location type: %s", profile.Location.Type)
 }
@@ -53,6 +66,12 @@ func Delete(ctx context.Context, profile param.Profile, suffix string) error {
 		args := s3CompliantDeleteArgs(profile, suffix, recursiveCmd)
 		env := s3CompliantEnv(profile)
 		return deleteExec(ctx, bin, args, env)
+	case crv1alpha1.LocationTypeGCS:
+		path := filepath.Join(
+			profile.Location.Prefix,
+			suffix,
+		)
+		return deleteData(ctx, objectstore.ProviderTypeGCS, profile, path)
 	}
 	return errors.Errorf("Unsupported Location type: %s", profile.Location.Type)
 }
@@ -222,3 +241,74 @@ func s3CompliantFlags(profile param.Profile) (cmd []string) {
 	}
 	return cmd
 }
+
+func readData(ctx context.Context, pType objectstore.ProviderType, profile param.Profile, out io.Writer, path string) error {
+	bucket, err := getBucket(ctx, pType, profile)
+	if err != nil {
+		return err
+	}
+
+	r, _, err := bucket.Get(ctx, path)
+	if err != nil {
+		return err
+	}
+	if _, err := io.Copy(out, r); err != nil {
+		return err
+	}
+	return nil
+}
+
+func writeData(ctx context.Context, pType objectstore.ProviderType, profile param.Profile, in io.Reader, path string) error {
+	bucket, err := getBucket(ctx, pType, profile)
+	if err != nil {
+		return err
+	}
+	if err := bucket.Put(ctx, path, in, 0, nil); err != nil {
+		return errors.Errorf("failed to write contents to bucket '%s'", profile.Location.Bucket)
+	}
+	return nil
+}
+
+func deleteData(ctx context.Context, pType objectstore.ProviderType, profile param.Profile, path string) error {
+	bucket, err := getBucket(ctx, pType, profile)
+	if err != nil {
+		return err
+	}
+	return bucket.Delete(ctx, path)
+}
+
+func getBucket(ctx context.Context, pType objectstore.ProviderType, profile param.Profile) (objectstore.Bucket, error) {
+	pc := objectstore.ProviderConfig{
+		Type: pType,
+	}
+	secret, err := getOSSecret(pType, profile.Credential)
+	if err != nil {
+		return nil, err
+	}
+	provider, err := objectstore.NewProvider(ctx, pc, secret)
+	if err != nil {
+		return nil, err
+	}
+	return provider.GetBucket(ctx, profile.Location.Bucket)
+}
+
+func getOSSecret(pType objectstore.ProviderType, cred param.Credential) (*objectstore.Secret, error) {
+	secret := &objectstore.Secret{}
+	switch pType {
+	case objectstore.ProviderTypeS3:
+		secret.Type = objectstore.SecretTypeAwsAccessKey
+		secret.Aws = &objectstore.SecretAws{
+			AccessKeyID:     cred.KeyPair.ID,
+			SecretAccessKey: cred.KeyPair.Secret,
+		}
+	case objectstore.ProviderTypeGCS:
+		secret.Type = objectstore.SecretTypeGcpServiceAccountKey
+		secret.Gcp = &objectstore.SecretGcp{
+			ProjectID:  cred.KeyPair.ID,
+			ServiceKey: cred.KeyPair.Secret,
+		}
+	default:
+		return nil, errors.Errorf("unknown or unsupported provider type '%s'", pType)
+	}
+	return secret, nil
+}

pkg/location/location_test.go

@@ -4,17 +4,82 @@ import (
 	"bytes"
 	"context"
 	"io"
+	"math/rand"
 	"testing"
+	"time"
 
 	. "gopkg.in/check.v1"
+
+	crv1alpha1 "github.com/kanisterio/kanister/pkg/apis/cr/v1alpha1"
+	"github.com/kanisterio/kanister/pkg/objectstore"
+	"github.com/kanisterio/kanister/pkg/param"
+	"github.com/kanisterio/kanister/pkg/testutil"
 )
 
 // Hook up gocheck into the "go test" runner.
 func Test(t *testing.T) { TestingT(t) }
 
-type LocationSuite struct{}
+type LocationSuite struct {
+	osType         objectstore.ProviderType
+	provider       objectstore.Provider
+	rand           *rand.Rand
+	root           objectstore.Bucket // root of the default test bucket
+	suiteDirPrefix string             // directory name prefix for all tests in this suite
+	testpath       string
+	region         string // bucket region
+	profile        param.Profile
+}
 
-var _ = Suite(&LocationSuite{})
+const (
+	testBucketName = "kanister-gcp-tests"
+	testRegionS3   = "us-west-2"
+)
+
+var _ = Suite(&LocationSuite{osType: objectstore.ProviderTypeGCS, region: ""})
+
+func (s *LocationSuite) SetUpSuite(c *C) {
+	switch s.osType {
+	case objectstore.ProviderTypeGCS:
+		testutil.GetEnvOrSkip(c, "GOOGLE_APPLICATION_CREDENTIALS")
+		location := crv1alpha1.Location{
+			Type:   crv1alpha1.LocationTypeGCS,
+			Bucket: testBucketName,
+		}
+		s.profile = *testutil.ObjectStoreProfileOrSkip(c, objectstore.ProviderTypeGCS, location)
+	default:
+		c.Fatalf("Unrecognized objectstore '%s'", s.osType)
+	}
+	var err error
+	ctx := context.Background()
+
+	s.rand = rand.New(rand.NewSource(time.Now().UnixNano()))
+	pc := objectstore.ProviderConfig{Type: s.osType}
+	secret, err := getOSSecret(s.osType, s.profile.Credential)
+	c.Check(err, IsNil)
+	s.provider, err = objectstore.NewProvider(ctx, pc, secret)
+	c.Check(err, IsNil)
+	c.Assert(s.provider, NotNil)
+
+	s.root, err = objectstore.GetOrCreateBucket(ctx, s.provider, testBucketName, s.region)
+	c.Check(err, IsNil)
+	c.Assert(s.root, NotNil)
+	s.suiteDirPrefix = time.Now().UTC().Format(time.RFC3339Nano)
+	s.testpath = s.suiteDirPrefix + "/testlocation.txt"
+}
+
+func (s *LocationSuite) TearDownTest(c *C) {
+	if s.testpath != "" {
+		c.Assert(s.root, NotNil)
+		ctx := context.Background()
+		err := s.root.Delete(ctx, s.testpath)
+		if err != nil {
+			c.Log("Cannot cleanup test directory: ", s.testpath)
+			return
+		}
+		err = s.provider.DeleteBucket(ctx, testBucketName)
+		c.Check(err, IsNil)
+	}
+}
 
 func (s *LocationSuite) TestWrite(c *C) {
 	ctx := context.Background()
@@ -97,3 +162,15 @@ func (s *LocationSuite) TestRead(c *C) {
 		c.Check(buf.String(), Equals, tc.out)
 	}
 }
+
+func (s *LocationSuite) TestWriteAndReadData(c *C) {
+	ctx := context.Background()
+	teststring := "test-content"
+	err := writeData(ctx, s.osType, s.profile, bytes.NewBufferString(teststring), s.testpath)
+	c.Check(err, IsNil)
+	buf := bytes.NewBuffer(nil)
+	err = readData(ctx, s.osType, s.profile, buf, s.testpath)
+	c.Check(err, IsNil)
+	c.Check(buf.String(), Equals, teststring)
+
+}

pkg/testutil/fixture.go

@@ -1,47 +1,54 @@
 package testutil
 
 import (
+	"context"
 	"fmt"
 	"os"
 
+	"golang.org/x/oauth2/google"
+	compute "google.golang.org/api/compute/v1"
 	"gopkg.in/check.v1"
 
 	crv1alpha1 "github.com/kanisterio/kanister/pkg/apis/cr/v1alpha1"
-	"github.com/kanisterio/kanister/pkg/location"
+	"github.com/kanisterio/kanister/pkg/blockstorage"
+	"github.com/kanisterio/kanister/pkg/blockstorage/awsebs"
+	"github.com/kanisterio/kanister/pkg/objectstore"
 	"github.com/kanisterio/kanister/pkg/param"
 )
 
-const testBucketName = "S3_TEST_BUCKET"
+const TestS3BucketName = "S3_TEST_BUCKET"
 
-var objectStoreTestEnvVars []string = []string{
-	location.AWSAccessKeyID,
-	location.AWSSecretAccessKey,
-	testBucketName,
-}
+func ObjectStoreProfileOrSkip(c *check.C, osType objectstore.ProviderType, location crv1alpha1.Location) *param.Profile {
+	var key, val string
 
-func ObjectStoreProfileOrSkip(c *check.C) *param.Profile {
-	skipIfEnvNotSet(c, objectStoreTestEnvVars)
+	switch osType {
+	case objectstore.ProviderTypeS3:
+		key = GetEnvOrSkip(c, awsebs.AccessKeyID)
+		val = GetEnvOrSkip(c, awsebs.SecretAccessKey)
+	case objectstore.ProviderTypeGCS:
+		GetEnvOrSkip(c, blockstorage.GoogleCloudCreds)
+		creds, err := google.FindDefaultCredentials(context.Background(), compute.ComputeScope)
+		c.Check(err, check.IsNil)
+		key = creds.ProjectID
+		val = string(creds.JSON)
+	}
 	return &param.Profile{
-		Location: crv1alpha1.Location{
-			Type:   crv1alpha1.LocationTypeS3Compliant,
-			Bucket: os.Getenv(testBucketName),
-			Prefix: c.TestName(),
-		},
+		Location: location,
 		Credential: param.Credential{
 			Type: param.CredentialTypeKeyPair,
 			KeyPair: &param.KeyPair{
-				ID:     os.Getenv(location.AWSAccessKeyID),
-				Secret: os.Getenv(location.AWSSecretAccessKey),
+				ID:     key,
+				Secret: val,
 			},
 		},
 	}
 }
 
-func skipIfEnvNotSet(c *check.C, envVars []string) {
-	for _, ev := range envVars {
-		if os.Getenv(ev) == "" {
-			reason := fmt.Sprintf("Test %s requires the environemnt variable '%s'", c.TestName(), ev)
-			c.Skip(reason)
-		}
+func GetEnvOrSkip(c *check.C, varName string) string {
+	v := os.Getenv(varName)
+	if v == "" {
+		reason := fmt.Sprintf("Test %s requires the environemnt variable '%s'", c.TestName(), varName)
+		c.Skip(reason)
 	}
+	return v
 }