Skip to content

Commit

Permalink
Stream Sharing GA enhancements (#1457)
Browse files Browse the repository at this point in the history 
Co-authored-by: Brian Strauch <[email protected]>
  • Loading branch information
@stkr89 @brianstrauch
stkr89 and brianstrauch authored Oct 13, 2022
1 parent 92c3a43 commit 0996206
Show file tree
Hide file tree
Showing 30 changed files with 561 additions and 139 deletions.
5 changes: 4 additions & 1 deletion cmd/lint/main.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -96,6 +96,7 @@ var flagRules = []linter.FlagRule{
"message-send-max-retries",
"request-required-acks",
"schema-registry-cluster-id",
"schema-registry-subjects",
"skip-message-on-error",
"source-bootstrap-server",
),
Expand All @@ -105,6 +106,7 @@ var flagRules = []linter.FlagRule{
linter.ExcludeFlag(
"aws-account-id",
"azure-subscription-id",
"gcp-project-id",
"ca-cert-path",
"client-cert-path",
"client-key-path",
Expand All @@ -129,6 +131,7 @@ var flagRules = []linter.FlagRule{
"request-timeout-ms",
"retry-backoff-ms",
"schema-registry-cluster-id",
"schema-registry-subjects",
"skip-message-on-error",
"socket-buffer-size",
"source-api-key",
Expand Down Expand Up @@ -308,7 +311,7 @@ func main() {
for _, cfg := range configs {
cfg.IsTest = true
cfg.Version = new(pversion.Version)

cmd := pcmd.NewConfluentCommand(cfg)
if err := l.Lint(cmd); err != nil {
fmt.Printf(`For context "%s", %v`, cfg.CurrentContext, err)
Expand Down
2 changes: 1 addition & 1 deletion go.mod
View file
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@ require (
github.com/confluentinc/cc-structs/kafka/util v0.1096.0
github.com/confluentinc/ccloud-sdk-go-v1 v0.0.108
github.com/confluentinc/ccloud-sdk-go-v2/apikeys v0.4.0
github.com/confluentinc/ccloud-sdk-go-v2/cdx v0.0.2
github.com/confluentinc/ccloud-sdk-go-v2/cdx v0.0.4
github.com/confluentinc/ccloud-sdk-go-v2/cli v0.1.0
github.com/confluentinc/ccloud-sdk-go-v2/cmk v0.6.0
github.com/confluentinc/ccloud-sdk-go-v2/connect v0.3.0
Expand Down
4 changes: 2 additions & 2 deletions go.sum
View file
Original file line number Diff line number Diff line change
Expand Up @@ -637,8 +637,8 @@ github.com/confluentinc/ccloud-sdk-go-v2-internal/networking v0.0.5/go.mod h1:b0
github.com/confluentinc/ccloud-sdk-go-v2-internal/networking v0.0.7/go.mod h1:b0fKj4FlWseVcd8CsjXUdvvDtAGHu8nDUJjPykqWInQ=
github.com/confluentinc/ccloud-sdk-go-v2/apikeys v0.4.0 h1:8fWyLwMuy8ec0MVF5Avd54UvbIxhDFhZzanHBVwgxdw=
github.com/confluentinc/ccloud-sdk-go-v2/apikeys v0.4.0/go.mod h1:wNa9Qg2e2v/+PQsUyKh+qB22hhLkPR6Ahy6rP+1jmGI=
github.com/confluentinc/ccloud-sdk-go-v2/cdx v0.0.2 h1:i+wiKAGc7rF3bnwukPyxS0sfpuwtaoT9vBJ04cSCeuU=
github.com/confluentinc/ccloud-sdk-go-v2/cdx v0.0.2/go.mod h1:L8U9xs2duASJnjIYkwGrSbZNpApsbh+vlxsJlZMHJPA=
github.com/confluentinc/ccloud-sdk-go-v2/cdx v0.0.4 h1:9eTUPKELQQP81WBI1WKnb24U7DM6ludy81KnUGSZwMc=
github.com/confluentinc/ccloud-sdk-go-v2/cdx v0.0.4/go.mod h1:L8U9xs2duASJnjIYkwGrSbZNpApsbh+vlxsJlZMHJPA=
github.com/confluentinc/ccloud-sdk-go-v2/cli v0.1.0 h1:YQFYJXWU4J10F2v4LhJm5OFSDe2PfZIHTHkzsbl1JuU=
github.com/confluentinc/ccloud-sdk-go-v2/cli v0.1.0/go.mod h1:y9C4nB+n7+vy50G83aY19h/1iKr+z3MvhW1YPVdf+t4=
github.com/confluentinc/ccloud-sdk-go-v2/cmk v0.6.0 h1:qOBunYd2bWo/IikZ60xMDcp/KfZ/ZahCOoeTjxXOoRU=
Expand Down
137 changes: 110 additions & 27 deletions internal/cmd/stream-share/command_consumer_redeem.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,38 +2,79 @@ package streamshare

import (
"fmt"
"strings"

"github.com/spf13/cobra"

cdxv1 "github.com/confluentinc/ccloud-sdk-go-v2/cdx/v1"
pcmd "github.com/confluentinc/cli/internal/pkg/cmd"
"github.com/confluentinc/cli/internal/pkg/examples"
"github.com/confluentinc/cli/internal/pkg/output"
)

var (
redeemTokenFields = []string{"Id", "ApiKey", "Secret", "KafkaBootstrapUrl", "Resources"}
redeemTokenFields = []string{"Id", "ApiKey", "ApiSecret", "KafkaBootstrapUrl", "SchemaRegistryApiKey", "SchemaRegistrySecret", "SchemaRegistryUrl", "Resources"}
redeemTokenPrivateLinkFields = []string{"NetworkDnsDomain", "NetworkZones", "NetworkZonalSubdomains", "NetworkKind",
"NetworkPrivateLinkDataType", "NetworkPrivateLinkData"}
redeemTokenHumanLabelMap = map[string]string{
"Id": "ID",
"ApiKey": "API Key",
"Secret": "Secret",
"KafkaBootstrapUrl": "Kafka Bootstrap URL",
"Resources": "Resources",
"Id": "ID",
"ApiKey": "API Key",
"ApiSecret": "API Secret",
"KafkaBootstrapUrl": "Kafka Bootstrap URL",
"SchemaRegistryApiKey": "Schema Registry API Key",
"SchemaRegistrySecret": "Schema Registry Secret",
"SchemaRegistryUrl": "Schema Registry URL",
"Resources": "Resources",
}
redeemTokenPrivateLinkHumanLabelMap = map[string]string{
"NetworkDnsDomain": "Network DNS Domain",
"NetworkZones": "Network Zones",
"NetworkZonalSubdomains": "Network Zonal Subdomains",
"NetworkKind": "Network Kind",
"NetworkPrivateLinkDataType": "Network Private Link Data Type",
"NetworkPrivateLinkData": "Network Private Link Data",
}
redeemTokenStructuredLabelMap = map[string]string{
"Id": "id",
"ApiKey": "api_key",
"Secret": "secret",
"KafkaBootstrapUrl": "kafka_bootstrap_url",
"Resources": "resources",
"Id": "id",
"ApiKey": "api_key",
"ApiSecret": "secret",
"KafkaBootstrapUrl": "kafka_bootstrap_url",
"SchemaRegistryApiKey": "schema_registry_api_key",
"SchemaRegistrySecret": "schema_registry_secret",
"SchemaRegistryUrl": "schema_registry_url",
"Resources": "resources",
}
redeemTokenPrivateLinkStructuredLabelMap = map[string]string{
"NetworkDnsDomain": "network_dns_domain",
"NetworkZones": "network_zones",
"NetworkZonalSubdomains": "network_zonal_subdomains",
"NetworkKind": "network_kind",
"NetworkPrivateLinkDataType": "network_private_link_data_type",
"NetworkPrivateLinkData": "network_private_link_data",
}
)

type privateLinkNetworkDetails struct {
networkKind string
privateLinkDataType string
privateLinkData interface{}
}

type redeemToken struct {
Id string
ApiKey string
Secret string
KafkaBootstrapUrl string
Resources []string
Id string
ApiKey string
ApiSecret string
KafkaBootstrapUrl string
SchemaRegistryApiKey string
SchemaRegistrySecret string
SchemaRegistryUrl string
Resources []string
NetworkDnsDomain string
NetworkZones string
NetworkZonalSubdomains []string
NetworkKind string
NetworkPrivateLinkDataType string
NetworkPrivateLinkData interface{}
}

func (c *command) newRedeemCommand() *cobra.Command {
Expand All @@ -50,8 +91,9 @@ func (c *command) newRedeemCommand() *cobra.Command {
),
}

cmd.Flags().String("aws-account-id", "", "The AWS account ID for the consumer network.")
cmd.Flags().String("azure-subscription-id", "", "The Azure subscription ID for the consumer network.")
cmd.Flags().String("aws-account-id", "", "Consumer's AWS account ID for PrivateLink access.")
cmd.Flags().String("azure-subscription-id", "", "Consumer's Azure subscription ID for PrivateLink access.")
cmd.Flags().String("gcp-project-id", "", "Consumer's GCP project ID for Private Service Connect access.")

pcmd.AddOutputFlag(cmd)

Expand All @@ -71,29 +113,70 @@ func (c *command) redeemShare(cmd *cobra.Command, args []string) error {
return err
}

redeemResponse, err := c.V2Client.RedeemSharedToken(token, awsAccountId, azureSubscriptionId)
gcpProjectId, err := cmd.Flags().GetString("gcp-project-id")
if err != nil {
return err
}

redeemResponse, err := c.V2Client.RedeemSharedToken(token, awsAccountId, azureSubscriptionId, gcpProjectId)
if err != nil {
return err
}

var resources []string
for _, resource := range redeemResponse.GetResources() {
if resource.CdxV1SharedTopic != nil {
resources = append(resources, fmt.Sprintf("%s:%s", resource.CdxV1SharedTopic.GetKind(), resource.CdxV1SharedTopic.GetTopic()))
resources = append(resources, fmt.Sprintf(`%s="%s"`, resource.CdxV1SharedTopic.GetKind(), resource.CdxV1SharedTopic.GetTopic()))
}

if resource.CdxV1SharedGroup != nil {
resources = append(resources, fmt.Sprintf("%s:%s", resource.CdxV1SharedGroup.GetKind(), resource.CdxV1SharedGroup.GetGroupPrefix()))
resources = append(resources, fmt.Sprintf(`Group_Prefix="%s"`, resource.CdxV1SharedGroup.GetGroupPrefix()))
}
}

tokenObj := &redeemToken{
Id: redeemResponse.GetId(),
ApiKey: redeemResponse.GetApikey(),
Secret: redeemResponse.GetSecret(),
KafkaBootstrapUrl: redeemResponse.GetKafkaBootstrapUrl(),
Resources: resources,
Id: redeemResponse.GetId(),
ApiKey: redeemResponse.GetApiKey(),
ApiSecret: redeemResponse.GetSecret(),
KafkaBootstrapUrl: redeemResponse.GetKafkaBootstrapUrl(),
SchemaRegistryApiKey: redeemResponse.GetSchemaRegistryApiKey(),
SchemaRegistrySecret: redeemResponse.GetSchemaRegistrySecret(),
SchemaRegistryUrl: redeemResponse.GetSchemaRegistryUrl(),
Resources: resources,
}

// non private link cluster
if awsAccountId == "" && azureSubscriptionId == "" && gcpProjectId == "" {
return output.DescribeObject(cmd, tokenObj, redeemTokenFields, redeemTokenHumanLabelMap, redeemTokenStructuredLabelMap)
}

return c.handlePrivateLinkClusterRedeem(cmd, redeemResponse, tokenObj)
}

func (c *command) handlePrivateLinkClusterRedeem(cmd *cobra.Command, redeemResponse cdxv1.CdxV1RedeemTokenResponse, tokenObj *redeemToken) error {
consumerSharedResources, err := c.V2Client.ListConsumerSharedResources(redeemResponse.GetId())
if err != nil {
return err
}

var network cdxv1.CdxV1Network
if len(consumerSharedResources) != 0 {
privateNetwork, err := c.V2Client.GetPrivateLinkNetworkConfig(consumerSharedResources[0].GetId())
if err != nil {
return err
}
network = privateNetwork
}

return output.DescribeObject(cmd, tokenObj, redeemTokenFields, redeemTokenHumanLabelMap, redeemTokenStructuredLabelMap)
networkDetails := getPrivateLinkNetworkDetails(network)
tokenObj.NetworkDnsDomain = network.GetDnsDomain()
tokenObj.NetworkZones = strings.Join(network.GetZones(), ",")
tokenObj.NetworkZonalSubdomains = mapSubdomainsToList(network.GetZonalSubdomains())
tokenObj.NetworkKind = networkDetails.networkKind
tokenObj.NetworkPrivateLinkDataType = networkDetails.privateLinkDataType
tokenObj.NetworkPrivateLinkData = networkDetails.privateLinkData

return output.DescribeObject(cmd, tokenObj, append(redeemTokenFields, redeemTokenPrivateLinkFields...),
combineMaps(redeemTokenHumanLabelMap, redeemTokenPrivateLinkHumanLabelMap),
combineMaps(redeemTokenStructuredLabelMap, redeemTokenPrivateLinkStructuredLabelMap))
}
54 changes: 30 additions & 24 deletions internal/cmd/stream-share/command_consumer_share.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -9,34 +9,40 @@ import (
)

var (
consumerShareListFields = []string{"Id", "ProviderName", "Status", "SharedResourceId", "InviteExpiresAt"}
consumerShareListHumanLabels = []string{"ID", "Provider Name", "Status", "Shared Resource ID", "Invite Expires At"}
consumerShareListStructuredLabels = []string{"id", "provider_name", "status", "shared_resource_id", "invite_expires_at"}
consumerShareListFields = []string{"Id", "ProviderName", "ProviderOrganizationName", "Status", "InviteExpiresAt"}
consumerShareListHumanLabels = []string{"ID", "Provider Name", "Provider Organization Name", "Status", "Invite Expires At"}
consumerShareListStructuredLabels = []string{"id", "provider_user_name", "provider_organization_name", "status", "invite_expires_at"}
)

var (
consumerHumanLabelMap = map[string]string{
"Id": "ID",
"ProviderName": "Provider Name",
"Status": "Status",
"SharedResourceId": "Shared Resource ID",
"InviteExpiresAt": "Invite Expires At",
"Id": "ID",
"ProviderName": "Provider Name",
"ProviderOrganizationName": "Provider Organization Name",
"Status": "Status",
"InviteExpiresAt": "Invite Expires At",
}
consumerStructuredLabelMap = map[string]string{
"Id": "id",
"ProviderName": "provider_name",
"Status": "status",
"SharedResourceId": "shared_resource_id",
"InviteExpiresAt": "invite_expires_at",
"Id": "id",
"ProviderName": "provider_user_name",
"ProviderOrganizationName": "provider_organization_name",
"Status": "status",
"InviteExpiresAt": "invite_expires_at",
}
)

type consumerShare struct {
Id string
ProviderName string
Status string
SharedResourceId string
InviteExpiresAt time.Time
Id string
ProviderName string
ProviderOrganizationName string
Status string
InviteExpiresAt time.Time
NetworkDnsDomain string
NetworkZones string
NetworkZonalSubdomains []string
NetworkKind string
NetworkPrivateLinkDataType string
NetworkPrivateLinkData interface{}
}

func (c *command) newConsumerShareCommand() *cobra.Command {
Expand Down Expand Up @@ -78,12 +84,12 @@ func (c *command) autocompleteConsumerShares() []string {
}

func (c *command) buildConsumerShare(share cdxv1.CdxV1ConsumerShare) *consumerShare {
sharedResource := share.GetSharedResource()
status := share.GetStatus()
return &consumerShare{
Id: share.GetId(),
ProviderName: share.GetProviderUserName(),
Status: share.GetStatus(),
SharedResourceId: sharedResource.GetId(),
InviteExpiresAt: share.GetInviteExpiresAt(),
Id: share.GetId(),
ProviderName: share.GetProviderUserName(),
ProviderOrganizationName: share.GetProviderOrganizationName(),
Status: status.GetPhase(),
InviteExpiresAt: share.GetInviteExpiresAt(),
}
}
39 changes: 38 additions & 1 deletion internal/cmd/stream-share/command_consumer_share_describe.go
View file
Original file line number Diff line number Diff line change
@@ -1,8 +1,11 @@
package streamshare

import (
"strings"

"github.com/spf13/cobra"

cdxv1 "github.com/confluentinc/ccloud-sdk-go-v2/cdx/v1"
pcmd "github.com/confluentinc/cli/internal/pkg/cmd"
"github.com/confluentinc/cli/internal/pkg/examples"
"github.com/confluentinc/cli/internal/pkg/output"
Expand Down Expand Up @@ -36,5 +39,39 @@ func (c *command) describeConsumerShare(cmd *cobra.Command, args []string) error
return err
}

return output.DescribeObject(cmd, c.buildConsumerShare(consumerShare), consumerShareListFields, consumerHumanLabelMap, consumerStructuredLabelMap)
consumerSharedResources, err := c.V2Client.ListConsumerSharedResources(shareId)
if err != nil {
return err
}

var network cdxv1.CdxV1Network
if len(consumerSharedResources) > 0 {
privateNetwork, err := c.V2Client.GetPrivateLinkNetworkConfig(consumerSharedResources[0].GetId())
if err != nil {
return err
}
network = privateNetwork
}

consumerShareObj := c.buildConsumerShare(consumerShare)
cloud := network.GetCloud()
if cloud.CdxV1AwsNetwork == nil && cloud.CdxV1AzureNetwork == nil && cloud.CdxV1GcpNetwork == nil {
return output.DescribeObject(cmd, consumerShareObj, consumerShareListFields, consumerHumanLabelMap, consumerStructuredLabelMap)
}

return c.handlePrivateLinkClusterConsumerShare(cmd, network, consumerShareObj)
}

func (c *command) handlePrivateLinkClusterConsumerShare(cmd *cobra.Command, network cdxv1.CdxV1Network, consumerShareObj *consumerShare) error {
networkDetails := getPrivateLinkNetworkDetails(network)
consumerShareObj.NetworkDnsDomain = network.GetDnsDomain()
consumerShareObj.NetworkZones = strings.Join(network.GetZones(), ",")
consumerShareObj.NetworkZonalSubdomains = mapSubdomainsToList(network.GetZonalSubdomains())
consumerShareObj.NetworkKind = networkDetails.networkKind
consumerShareObj.NetworkPrivateLinkDataType = networkDetails.privateLinkDataType
consumerShareObj.NetworkPrivateLinkData = networkDetails.privateLinkData

return output.DescribeObject(cmd, consumerShareObj, append(consumerShareListFields, redeemTokenPrivateLinkFields...),
combineMaps(consumerHumanLabelMap, redeemTokenPrivateLinkHumanLabelMap),
combineMaps(consumerStructuredLabelMap, redeemTokenPrivateLinkStructuredLabelMap))
}
Loading

0 comments on commit 0996206

Please sign in to comment.