rpk profile: switch auth on profile switch, and notify

src/go/rpk/pkg/cli/container/common/common.go

@@ -499,11 +499,12 @@ func CreateProfile(fs afero.Fs, c Client, y *config.RpkYaml) error {
 		},
 	}
 
-	y.CurrentProfile = y.PushProfile(profile)
+	priorAuth, currentAuth := y.PushProfile(profile)
 	err = y.Write(fs)
 	if err != nil {
 		return err
 	}
+	config.MaybePrintAuthSwitchMessage(priorAuth, currentAuth)
 	return nil
 }
 

src/go/rpk/pkg/cli/profile/create.go

@@ -195,6 +195,7 @@ func CreateFlow(
 			return err
 		}
 		p = &o.Profile
+
 		if name == "" {
 			name = RpkCloudProfileName
 			if old := yAct.Profile(name); old != nil {
@@ -275,20 +276,22 @@ func CreateFlow(
 		// * The user's prior profile was NOT the rpk-cloud profile
 		// * We are switching to the existing rpk-cloud profile and updating the values
 		fmt.Printf("Switched to existing %q profile and updated it to talk to cluster %q.\n", RpkCloudProfileName, o.ClusterName)
-		yAct.MoveProfileToFront(p)
-		yAct.CurrentProfile = p.Name
+		priorAuth, currentAuth := yAct.MoveProfileToFront(p)
+		config.MaybePrintAuthSwitchMessage(priorAuth, currentAuth)
 
 	case fromCloud != "" && name == RpkCloudProfileName:
 		// * We are creating a NEW rpk-cloud profile and switching to it
 		fmt.Printf("Created and switched to a new profile %q to talk to cloud cluster %q.\n", RpkCloudProfileName, o.ClusterName)
-		yAct.CurrentProfile = yAct.PushProfile(*p)
+		priorAuth, currentAuth := yAct.PushProfile(*p)
+		config.MaybePrintAuthSwitchMessage(priorAuth, currentAuth)
 
 	default:
 		// * This is not a cloud nor container profile, this is a
 		//   regularly manually created profile, or duplicating an
 		//   existing one. We print standard messaging.
 		fmt.Printf("Created and switched to new profile %q.\n", p.Name)
-		yAct.CurrentProfile = yAct.PushProfile(*p)
+		priorAuth, currentAuth := yAct.PushProfile(*p)
+		config.MaybePrintAuthSwitchMessage(priorAuth, currentAuth)
 	}
 
 	if err := yAct.Write(fs); err != nil {
@@ -329,7 +332,7 @@ func createCloudProfile(ctx context.Context, y *config.RpkYaml, cfg *config.Conf
 	cl := cloudapi.NewClient(overrides.CloudAPIURL, a.AuthToken, httpapi.ReqTimeout(10*time.Second))
 
 	if clusterIDOrName == "prompt" {
-		return PromptCloudClusterProfile(ctx, cl)
+		return PromptCloudClusterProfile(ctx, a, cl)
 	}
 
 	var (
@@ -355,6 +358,11 @@ nameLookup:
 		clusterID = clusterIDOrName
 	}
 
+	// When we create a cloud profile, we want the namespace name so that
+	// we can print some nicer looking messages. When we finally know the
+	// cluster ID, we do a final namespace lookup and map the cluster's
+	// namespace UUID to the namespace name.
+
 	vc, err := cl.VirtualCluster(ctx, clusterID)
 	if err != nil { // if we fail for a vcluster, we try again for a normal cluster
 		c, err := cl.Cluster(ctx, clusterID)
@@ -370,9 +378,17 @@ nameLookup:
 			}
 			return CloudClusterOutputs{}, fmt.Errorf("unable to request details for cluster %q: %w", clusterID, err)
 		}
-		return fromCloudCluster(c), nil
+		ns, err := cl.NamespaceForID(ctx, c.NamespaceUUID)
+		if err != nil {
+			return CloudClusterOutputs{}, err
+		}
+		return fromCloudCluster(a, ns, c), nil
+	}
+	ns, err := cl.NamespaceForID(ctx, vc.NamespaceUUID)
+	if err != nil {
+		return CloudClusterOutputs{}, err
 	}
-	return fromVirtualCluster(vc), nil
+	return fromVirtualCluster(a, ns, vc), nil
 }
 
 func clusterNameToID(ctx context.Context, cl *cloudapi.Client, name string) (string, error) {
@@ -440,10 +456,17 @@ func findNamedCluster(name string, nss []cloudapi.Namespace, vcs []cloudapi.Virt
 
 // fromCloudCluster returns an rpk profile from a cloud cluster, as well
 // as if the cluster requires mtls or sasl.
-func fromCloudCluster(c cloudapi.Cluster) CloudClusterOutputs {
+func fromCloudCluster(yAuth *config.RpkCloudAuth, ns cloudapi.Namespace, c cloudapi.Cluster) CloudClusterOutputs {
 	p := config.RpkProfile{
 		Name:      c.Name,
 		FromCloud: true,
+		CloudCluster: config.RpkCloudCluster{
+			Namespace:   ns.Name,
+			ClusterID:   c.ID,
+			ClusterName: c.Name,
+			AuthOrgID:   yAuth.OrgID,
+			AuthKind:    string(yAuth.AnyKind()),
+		},
 	}
 	p.KafkaAPI.Brokers = c.Status.Listeners.Kafka.Default.URLs
 	var isMTLS, isSASL bool
@@ -463,7 +486,7 @@ func fromCloudCluster(c cloudapi.Cluster) CloudClusterOutputs {
 	}
 }
 
-func fromVirtualCluster(vc cloudapi.VirtualCluster) CloudClusterOutputs {
+func fromVirtualCluster(yAuth *config.RpkCloudAuth, ns cloudapi.Namespace, vc cloudapi.VirtualCluster) CloudClusterOutputs {
 	p := config.RpkProfile{
 		Name:      vc.Name,
 		FromCloud: true,
@@ -478,6 +501,13 @@ func fromVirtualCluster(vc cloudapi.VirtualCluster) CloudClusterOutputs {
 			Addresses: []string{vc.Status.Listeners.ConsoleURL},
 			TLS:       new(config.TLS),
 		},
+		CloudCluster: config.RpkCloudCluster{
+			Namespace:   ns.Name,
+			ClusterID:   vc.ID,
+			ClusterName: vc.Name,
+			AuthOrgID:   yAuth.OrgID,
+			AuthKind:    string(yAuth.AnyKind()),
+		},
 	}
 
 	return CloudClusterOutputs{
@@ -550,7 +580,7 @@ type CloudClusterOutputs struct {
 // user. If their cloud account has only one cluster, a profile is created for
 // it automatically. This returns ErrNoCloudClusters if the user has no cloud
 // clusters.
-func PromptCloudClusterProfile(ctx context.Context, cl *cloudapi.Client) (CloudClusterOutputs, error) {
+func PromptCloudClusterProfile(ctx context.Context, yAuth *config.RpkCloudAuth, cl *cloudapi.Client) (CloudClusterOutputs, error) {
 	org, nss, vcs, cs, err := cl.OrgNamespacesClusters(ctx)
 	if err != nil {
 		return CloudClusterOutputs{}, err
@@ -590,13 +620,21 @@ func PromptCloudClusterProfile(ctx context.Context, cl *cloudapi.Client) (CloudC
 		if err != nil {
 			return CloudClusterOutputs{}, fmt.Errorf("unable to get cluster %q information: %w", c.ID, err)
 		}
-		o = fromCloudCluster(c)
+		ns, err := cl.NamespaceForID(ctx, c.NamespaceUUID)
+		if err != nil {
+			return CloudClusterOutputs{}, err
+		}
+		o = fromCloudCluster(yAuth, ns, c)
 	} else {
 		c, err := cl.VirtualCluster(ctx, selected.vc.ID)
 		if err != nil {
 			return CloudClusterOutputs{}, fmt.Errorf("unable to get cluster %q information: %w", c.ID, err)
 		}
-		o = fromVirtualCluster(c)
+		ns, err := cl.NamespaceForID(ctx, c.NamespaceUUID)
+		if err != nil {
+			return CloudClusterOutputs{}, err
+		}
+		o = fromVirtualCluster(yAuth, ns, c)
 	}
 	o.Profile.Description = fmt.Sprintf("%s %q", org.Name, selected.name)
 	return o, nil

src/go/rpk/pkg/cli/profile/edit.go

@@ -43,13 +43,24 @@ exist, this command creates it and switches to it.
 			name := args[0]
 			p := y.Profile(name)
 			if p == nil {
-				y.CurrentProfile = y.PushProfile(config.RpkProfile{Name: name})
+				priorAuth, currentAuth := y.PushProfile(config.RpkProfile{Name: name})
+				// Defer, so that if we out.Die, we don't print the switch message
+				// (and we want to print this last anyway).
+				defer config.MaybePrintAuthSwitchMessage(priorAuth, currentAuth)
 				p = y.Profile(name)
 			}
 
+			preFromCloud := p.FromCloud
+			preCloudDetails := p.CloudCluster
 			update, err := rpkos.EditTmpYAMLFile(fs, *p)
 			out.MaybeDieErr(err)
 
+			if preFromCloud {
+				if !update.FromCloud || preCloudDetails != update.CloudCluster {
+					out.Die("cannot change a cloud profile to a non-cloud profile, and cannot change cloud cluster details")
+				}
+			}
+
 			// If a user clears the name by accident, we keep the old name.
 			if update.Name == "" {
 				update.Name = name

src/go/rpk/pkg/cli/profile/rename.go

@@ -44,10 +44,11 @@ func newRenameToCommand(fs afero.Fs, p *config.Params) *cobra.Command {
 			}
 			p.Name = to
 			y.CurrentProfile = to
-			y.MoveProfileToFront(p)
+			priorAuth, currentAuth := y.MoveProfileToFront(p)
 			err = y.Write(fs)
 			out.MaybeDieErr(err)
 			fmt.Printf("Renamed current profile to %q.\n", to)
+			config.MaybePrintAuthSwitchMessage(priorAuth, currentAuth)
 		},
 	}
 }

src/go/rpk/pkg/cli/profile/use.go

@@ -38,12 +38,12 @@ func newUseCommand(fs afero.Fs, p *config.Params) *cobra.Command {
 			if p == nil {
 				out.Die("profile %q does not exist", name)
 			}
-			y.CurrentProfile = name
-			y.MoveProfileToFront(p)
+			priorAuth, currentAuth := y.MoveProfileToFront(p)
 
 			err = y.Write(fs)
 			out.MaybeDieErr(err)
 			fmt.Printf("Set current profile to %q.\n", name)
+			config.MaybePrintAuthSwitchMessage(priorAuth, currentAuth)
 		},
 	}
 

src/go/rpk/pkg/cloudapi/cloudapi.go

@@ -134,3 +134,12 @@ func FindNamespace(nsID string, nss Namespaces) Namespace {
 	}
 	return Namespace{}
 }
+
+// NamespaceForID lists all namespaces to find the namespace for the given ID.
+func (cl *Client) NamespaceForID(ctx context.Context, nsid string) (Namespace, error) {
+	nss, err := cl.Namespaces(ctx)
+	if err != nil {
+		return Namespace{}, fmt.Errorf("unable to request namespaces: %w", err)
+	}
+	return FindNamespace(nsid, nss), nil
+}

src/go/rpk/pkg/config/params.go

@@ -1177,10 +1177,10 @@ func (c *Config) promptDeleteOldRpkYaml(fs afero.Fs) error {
 	}
 
 	if !isatty.IsTerminal(os.Stdin.Fd()) && !isatty.IsCygwinTerminal(os.Stdin.Fd()) {
-		return fmt.Errorf("rpk found invalid cloud auths or profiles, but is not running in a terminal, we cannot prompt if it is ok to delete the old auths or profiles")
+		return nil
 	}
 	if !isatty.IsTerminal(os.Stdout.Fd()) && !isatty.IsCygwinTerminal(os.Stdout.Fd()) {
-		return fmt.Errorf("rpk found invalid cloud auths or profiles, but is not running in a terminal, we cannot prompt if it is ok to delete the old auths or profiles")
+		return nil
 	}
 
 	if len(deleteAuthNames) > 0 {

src/go/rpk/pkg/config/rpk_yaml.go

@@ -185,14 +185,23 @@ func (y *RpkYaml) Profile(name string) *RpkProfile {
 	return nil
 }
 
-// PushProfile pushes a profile to the front and returns the profile's name.
-func (y *RpkYaml) PushProfile(p RpkProfile) string {
+// PushProfile pushes a profile to the front, updates the current profile, and
+// returns the prior profile's auth and the current profile's auth.
+func (y *RpkYaml) PushProfile(p RpkProfile) (priorAuth, currentAuth *RpkCloudAuth) {
+	priorAuth = y.CurrentAuth()
 	y.Profiles = append([]RpkProfile{p}, y.Profiles...)
-	return p.Name
+	if p.FromCloud {
+		y.CurrentCloudAuthOrgID = p.CloudCluster.AuthOrgID
+		y.CurrentCloudAuthKind = p.CloudCluster.AuthKind
+	}
+	currentAuth = y.CurrentAuth()
+	y.CurrentProfile = p.Name
+	return priorAuth, currentAuth
 }
 
 // MoveProfileToFront moves the given profile to the front of the list.
-func (y *RpkYaml) MoveProfileToFront(p *RpkProfile) {
+func (y *RpkYaml) MoveProfileToFront(p *RpkProfile) (priorAuth, currentAuth *RpkCloudAuth) {
+	priorAuth = y.CurrentAuth()
 	reordered := []RpkProfile{*p}
 	for i := range y.Profiles {
 		if &y.Profiles[i] == p {
@@ -201,6 +210,15 @@ func (y *RpkYaml) MoveProfileToFront(p *RpkProfile) {
 		reordered = append(reordered, y.Profiles[i])
 	}
 	y.Profiles = reordered
+	y.CurrentProfile = p.Name
+
+	// If this is a cloud profile, we switch the auth as well.
+	if p.FromCloud {
+		y.CurrentCloudAuthOrgID = p.CloudCluster.AuthOrgID
+		y.CurrentCloudAuthKind = p.CloudCluster.AuthKind
+	}
+	currentAuth = y.CurrentAuth()
+	return priorAuth, currentAuth
 }
 
 // LookupAuth returns an RpkCloudAuth based on the org and kind.
@@ -456,3 +474,26 @@ func (g *RpkGlobals) GetCommandTimeout() time.Duration {
 	}
 	return g.CommandTimeout.Duration
 }
+
+//////////
+// MISC //
+//////////
+
+// MaybePrintAuthSwitchMessage prints a message if the prior and current
+// auths are different.
+func MaybePrintAuthSwitchMessage(priorAuth *RpkCloudAuth, currentAuth *RpkCloudAuth) {
+	if priorAuth == nil {
+		if currentAuth == nil {
+			return
+		}
+		fmt.Println("rpk cloud commands are now talking to organization %q (%s).", currentAuth.Organization, currentAuth.OrgID)
+		return
+	}
+	if currentAuth == nil {
+		fmt.Println("rpk cloud commands are no longer talking to organization %q (%s) and are now talking to a self hosted cluster.", priorAuth.Organization, priorAuth.OrgID)
+		return
+	}
+	if priorAuth.Name != currentAuth.Name {
+		fmt.Println("rpk switched from talking to organization %q (%s) to %q (%s).", priorAuth.Organization, priorAuth.OrgID, currentAuth.Organization, currentAuth.OrgID)
+	}
+}