SDKQE-3441: Add ability to add external links to Capella Columnar clu…

…sters (#69)

cmd/links-add-capella.go

@@ -0,0 +1,51 @@
+package cmd
+
+import (
+	"github.com/couchbaselabs/cbdinocluster/deployment/clouddeploy"
+	"github.com/spf13/cobra"
+	"go.uber.org/zap"
+)
+
+var linksCapellaCmd = &cobra.Command{
+	Use:   "capella",
+	Short: "Link a capella cluster to a columnar instance. Provide either a capella Cbdino id, or a capella cluster id (i.e. not created by cbdino) ",
+	Args:  cobra.MinimumNArgs(1),
+	Run: func(cmd *cobra.Command, args []string) {
+		helper := CmdHelper{}
+		logger := helper.GetLogger()
+		ctx := helper.GetContext()
+
+		linkName, _ := cmd.Flags().GetString("link-name")
+		capellaId, _ := cmd.Flags().GetString("cbd-id")
+		directId, _ := cmd.Flags().GetString("capella-id")
+
+		if linkName == "" {
+			logger.Fatal("you must specify a link name")
+		}
+
+		if capellaId == directId && directId == "" {
+			logger.Fatal("you must specify only one of a cbd-id or a direct capella-id ")
+		}
+
+		_, deployer, cluster := helper.IdentifyCluster(ctx, args[0])
+
+		cloudDeployer, ok := deployer.(*clouddeploy.Deployer)
+		if !ok {
+			logger.Fatal("links capella is only supported for cloud deployments")
+		}
+
+		err := cloudDeployer.CreateCapellaLink(ctx, cluster.GetID(), linkName, capellaId, directId)
+		if err != nil {
+			logger.Fatal("failed to link capella cluster", zap.Error(err))
+		}
+	},
+}
+
+func init() {
+	linksAddCmd.AddCommand(linksCapellaCmd)
+
+	linksCapellaCmd.Flags().String("link-name", "", "The name of the link to be created")
+	linksCapellaCmd.Flags().String("cbd-id", "", "The cbdino id of the capella cluster to link")
+	linksCapellaCmd.Flags().String("capella-id", "", "The direct capella cluster id, if created without cbdino")
+
+}

cmd/links-add-s3.go

@@ -0,0 +1,60 @@
+package cmd
+
+import (
+	"github.com/couchbaselabs/cbdinocluster/deployment/clouddeploy"
+	"github.com/spf13/cobra"
+	"go.uber.org/zap"
+)
+
+var linksS3Cmd = &cobra.Command{
+	Use:   "s3",
+	Short: "Link a S3 bucket to a columnar instance.",
+	Args:  cobra.MinimumNArgs(1),
+	Run: func(cmd *cobra.Command, args []string) {
+		helper := CmdHelper{}
+		logger := helper.GetLogger()
+		ctx := helper.GetContext()
+
+		linkName, _ := cmd.Flags().GetString("link-name")
+		region, _ := cmd.Flags().GetString("region")
+		// Optionals
+		endpoint, _ := cmd.Flags().GetString("endpoint")
+		accessKey, _ := cmd.Flags().GetString("access-key")
+		secretKey, _ := cmd.Flags().GetString("secret-key")
+
+		if linkName == "" {
+			logger.Fatal("you must give the link a name")
+		}
+
+		if region == "" {
+			logger.Fatal("you must specify an AWS region")
+		}
+		_, deployer, cluster := helper.IdentifyCluster(ctx, args[0])
+
+		cloudDeployer, ok := deployer.(*clouddeploy.Deployer)
+		if !ok {
+			logger.Fatal("links s3 is only supported for cloud deployments")
+		}
+
+		if accessKey == "" && secretKey == "" {
+			awsCreds := helper.GetAWSCredentials(ctx)
+			accessKey = awsCreds.AccessKeyID
+			secretKey = awsCreds.SecretAccessKey
+		}
+
+		err := cloudDeployer.CreateS3Link(ctx, cluster.GetID(), linkName, region, endpoint, accessKey, secretKey)
+		if err != nil {
+			logger.Fatal("failed to setup S3 link", zap.Error(err))
+		}
+	},
+}
+
+func init() {
+	linksAddCmd.AddCommand(linksS3Cmd)
+
+	linksS3Cmd.Flags().String("link-name", "", "The name of the link to be created")
+	linksS3Cmd.Flags().String("region", "", "The AWS region the S3 bucket is in.")
+	linksS3Cmd.Flags().String("endpoint", "", "The S3 endpoint. Optional.")
+	linksS3Cmd.Flags().String("access-key", "", "AWS AccessKeyId to use. Will use the cbdino config values if not flag not provided.")
+	linksS3Cmd.Flags().String("secret-key", "", "AWS SecretKey to use. Will use the cbdino config values if not flag not provided.")
+}

cmd/links-add.go

@@ -0,0 +1,15 @@
+package cmd
+
+import (
+	"github.com/spf13/cobra"
+)
+
+var linksAddCmd = &cobra.Command{
+	Use:   "add",
+	Short: "Add a link to a columnar cluster",
+	Run:   nil,
+}
+
+func init() {
+	linksCmd.AddCommand(linksAddCmd)
+}

cmd/links-drop.go

@@ -0,0 +1,34 @@
+package cmd
+
+import (
+	"github.com/couchbaselabs/cbdinocluster/deployment/clouddeploy"
+	"github.com/spf13/cobra"
+	"go.uber.org/zap"
+)
+
+var linksDropCmd = &cobra.Command{
+	Use:   "drop",
+	Short: "Drop a link on a columnar instance",
+	Args:  cobra.MinimumNArgs(2),
+	Run: func(cmd *cobra.Command, args []string) {
+		helper := CmdHelper{}
+		logger := helper.GetLogger()
+		ctx := helper.GetContext()
+		_, deployer, cluster := helper.IdentifyCluster(ctx, args[0])
+		linkName := args[1]
+
+		cloudDeployer, ok := deployer.(*clouddeploy.Deployer)
+		if !ok {
+			logger.Fatal("links is only supported for cloud deployments")
+		}
+
+		err := cloudDeployer.DropLink(ctx, cluster.GetID(), linkName)
+		if err != nil {
+			logger.Fatal("failed to drop link", zap.Error(err))
+		}
+	},
+}
+
+func init() {
+	linksCmd.AddCommand(linksDropCmd)
+}

cmd/links.go

@@ -0,0 +1,15 @@
+package cmd
+
+import (
+	"github.com/spf13/cobra"
+)
+
+var linksCmd = &cobra.Command{
+	Use:   "links",
+	Short: "Provides ability to link data sources to columnar instances",
+	Run:   nil,
+}
+
+func init() {
+	rootCmd.AddCommand(linksCmd)
+}

deployment/caodeploy/deployer.go

@@ -732,3 +732,15 @@ func (d *Deployer) UnpauseNode(ctx context.Context, clusterID string, nodeID str
 func (d *Deployer) RedeployCluster(ctx context.Context, clusterID string) error {
 	return errors.New("caodeploy does not support redeploy cluster")
 }
+
+func (d *Deployer) CreateCapellaLink(ctx context.Context, columnarID, linkName, clusterId, directID string) error {
+	return errors.New("caodeploy does not support create capella link")
+}
+
+func (d *Deployer) CreateS3Link(ctx context.Context, columnarID, linkName, region, endpoint, accessKey, secretKey string) error {
+	return errors.New("caodeploy does not support create S3 link")
+}
+
+func (d *Deployer) DropLink(ctx context.Context, columnarID, linkName string) error {
+	return errors.New("caodeploy does not support drop link")
+}

deployment/clouddeploy/deployer.go

@@ -1933,6 +1933,70 @@ func (d *Deployer) RedeployCluster(ctx context.Context, clusterID string) error
 	return nil
 }
 
+func (d *Deployer) CreateCapellaLink(ctx context.Context, columnarID, linkName, clusterId, directID string) error {
+	columnarInfo, err := d.getCluster(ctx, columnarID)
+	if err != nil {
+		return err
+	}
+	if columnarInfo.Columnar == nil {
+		return errors.Wrap(err, "this is not a columnar cluster")
+	}
+
+	resolvedClusterId := directID
+	if directID == "" {
+		clusterInfo, err := d.getCluster(ctx, clusterId)
+		if err != nil {
+			return err
+		}
+		if clusterInfo.Columnar != nil {
+			return errors.Wrap(err, "can not link to another columnar cluster")
+		}
+		resolvedClusterId = clusterInfo.Cluster.Id
+	}
+
+	req := &capellacontrol.CreateColumnarCapellaLinkRequest{
+		LinkName:           linkName,
+		ProvisionedCluster: capellacontrol.ProvisionedCluster{ClusterId: resolvedClusterId},
+	}
+	return d.mgr.Client.CreateColumnarCapellaLink(ctx, columnarInfo.Columnar.TenantID, columnarInfo.Columnar.ProjectID, columnarInfo.Columnar.ID, req)
+}
+
+func (d *Deployer) CreateS3Link(ctx context.Context, columnarID, linkName, region, endpoint, accessKey, secretKey string) error {
+	columnarInfo, err := d.getCluster(ctx, columnarID)
+	if err != nil {
+		return err
+	}
+	if columnarInfo.Columnar == nil {
+		return errors.Wrap(err, "this is not a columnar cluster")
+	}
+
+	req := &capellacontrol.CreateColumnarS3LinkRequest{
+		Region:          region,
+		AccessKeyId:     accessKey,
+		SecretAccessKey: secretKey,
+		SessionToken:    "",
+		Endpoint:        endpoint,
+		Type:            "s3",
+	}
+	return d.mgr.Client.CreateColumnarS3Link(ctx, columnarInfo.Columnar.TenantID, columnarInfo.Columnar.ProjectID, columnarInfo.Columnar.ID, linkName, req)
+}
+
+func (d *Deployer) DropLink(ctx context.Context, columnarID, linkName string) error {
+	columnarInfo, err := d.getCluster(ctx, columnarID)
+	if err != nil {
+		return err
+	}
+	if columnarInfo.Columnar == nil {
+		return errors.Wrap(err, "this is not a columnar cluster")
+	}
+
+	req := &capellacontrol.ColumnarQueryRequest{
+		Statement:   fmt.Sprintf("DROP LINK `%s`", linkName),
+		MaxWarnings: 25,
+	}
+	return d.mgr.Client.DoBasicColumnarQuery(ctx, columnarInfo.Columnar.TenantID, columnarInfo.Columnar.ProjectID, columnarInfo.Columnar.ID, req)
+}
+
 func (d *Deployer) GetGatewayCertificate(ctx context.Context, clusterID string) (string, error) {
 	return "", errors.New("clouddeploy does not support getting gateway certificates")
 }

deployment/deployer.go

@@ -122,4 +122,7 @@ type Deployer interface {
 	PauseNode(ctx context.Context, clusterID string, nodeID string) error
 	UnpauseNode(ctx context.Context, clusterID string, nodeID string) error
 	RedeployCluster(ctx context.Context, clusterID string) error
+	CreateCapellaLink(ctx context.Context, columnarID, linkName, clusterId, directID string) error
+	CreateS3Link(ctx context.Context, columnarID, linkName, region, endpoint, accessKey, secretKey string) error
+	DropLink(ctx context.Context, columnarID, linkName string) error
 }

deployment/dockerdeploy/deployer.go

@@ -1584,3 +1584,15 @@ func (d *Deployer) UnpauseNode(ctx context.Context, clusterID string, nodeID str
 func (d *Deployer) RedeployCluster(ctx context.Context, clusterID string) error {
 	return errors.New("docker deploy does not support redeploy cluster")
 }
+
+func (d *Deployer) CreateCapellaLink(ctx context.Context, columnarID, linkName, clusterId, directID string) error {
+	return errors.New("docker deploy does not support create capella link")
+}
+
+func (d *Deployer) CreateS3Link(ctx context.Context, columnarID, linkName, region, endpoint, accessKey, secretKey string) error {
+	return errors.New("docker deploy does not support create S3 link")
+}
+
+func (d *Deployer) DropLink(ctx context.Context, columnarID, linkName string) error {
+	return errors.New("docker deploy does not support drop link")
+}

deployment/localdeploy/deployer.go

@@ -211,3 +211,15 @@ func (d *Deployer) LoadSampleBucket(ctx context.Context, clusterID string, bucke
 func (d *Deployer) RedeployCluster(ctx context.Context, clusterID string) error {
 	return errors.New("localdeploy does not support redeploy cluster")
 }
+
+func (d *Deployer) CreateCapellaLink(ctx context.Context, columnarID, linkName, clusterId, directID string) error {
+	return errors.New("localdeploy does not support create capella link")
+}
+
+func (d *Deployer) CreateS3Link(ctx context.Context, columnarID, linkName, region, endpoint, accessKey, secretKey string) error {
+	return errors.New("localdeploy does not support create S3 link")
+}
+
+func (d *Deployer) DropLink(ctx context.Context, columnarID, linkName string) error {
+	return errors.New("localdeploy does not support drop link")
+}

utils/capellacontrol/controller.go

@@ -1614,10 +1614,6 @@ func (c *Controller) LoadColumnarSampleBucket(
 	form, _ := query.Values(req)
 	path := fmt.Sprintf("/v2/organizations/%s/projects/%s/instance/%s/proxy/api/v1/samples?%s", tenantID, projectID, clusterID, form.Encode())
 	err := c.doBasicReq(ctx, false, "POST", path, req, nil)
-	if err != nil {
-		return err
-	}
-
 	return err
 }
 
@@ -1632,9 +1628,62 @@ func (c *Controller) LoadClusterSampleBucket(
 ) error {
 	path := fmt.Sprintf("/v2/organizations/%s/projects/%s/clusters/%s/buckets/samples", tenantID, projectID, clusterID)
 	err := c.doBasicReq(ctx, false, "POST", path, req, nil)
-	if err != nil {
-		return err
-	}
+	return err
+}
+
+type ProvisionedCluster struct {
+	ClusterId string `json:"clusterId"`
+}
+
+type CreateColumnarCapellaLinkRequest struct {
+	LinkName           string             `json:"linkName"`
+	ProvisionedCluster ProvisionedCluster `json:"provisionedCluster"`
+}
+
+func (c *Controller) CreateColumnarCapellaLink(
+	ctx context.Context,
+	tenantID, projectID, columnarID string,
+	req *CreateColumnarCapellaLinkRequest,
+) error {
+	path := fmt.Sprintf("/v2/organizations/%s/projects/%s/instance/%s/links", tenantID, projectID, columnarID)
+	err := c.doBasicReq(ctx, false, "POST", path, req, nil)
+	return err
+}
 
+type CreateColumnarS3LinkRequest struct {
+	Region          string `url:"region"`
+	AccessKeyId     string `url:"accessKeyId"`
+	SecretAccessKey string `url:"secretAccessKey"`
+	SessionToken    string `url:"sessionToken"`
+	Endpoint        string `url:"endpoint"`
+	Type            string `url:"type"`
+}
+
+func (c *Controller) CreateColumnarS3Link(
+	ctx context.Context,
+	tenantID, projectID, columnarID, linkName string,
+	req *CreateColumnarS3LinkRequest,
+) error {
+	form, _ := query.Values(req)
+
+	path := fmt.Sprintf("/v2/organizations/%s/projects/%s/instance/%s/proxy/analytics/link/%s?%s",
+		tenantID, projectID, columnarID, linkName, form.Encode())
+	err := c.doBasicReq(ctx, false, "POST", path, req, nil)
+	return err
+}
+
+type ColumnarQueryRequest struct {
+	Statement   string `json:"statement"`
+	MaxWarnings int    `json:"max-warnings"`
+}
+
+// Expect no results
+func (c *Controller) DoBasicColumnarQuery(
+	ctx context.Context,
+	tenantID, projectID, columnarID string,
+	req *ColumnarQueryRequest,
+) error {
+	path := fmt.Sprintf("/v2/organizations/%s/projects/%s/instance/%s/proxy/analytics/service", tenantID, projectID, columnarID)
+	err := c.doBasicReq(ctx, false, "POST", path, req, nil)
 	return err
 }