Finish basic implmeentation of 'upgrade cloudprovider' and fix image version checking for 'migration providerid'

Author: detiber

cmd/helper/base/tool.go

@@ -32,13 +32,10 @@ func ClusterToKey(c *clusterv1.Cluster) string {
 }
 
 type Tool struct {
-	// TODO: make this private
-	Clusters []*clusterv1.Cluster
-
-	kubeconfig string
-	MgmtClient client.Client
-
-	mu              sync.Mutex
+	kubeconfig      string
+	MgmtClient      client.Client
+	baseMutex       sync.Mutex
+	clusters        []*clusterv1.Cluster
 	workloadClients map[string]client.Client
 	errors          map[string]error
 	outputBuffers   map[string]*bytes.Buffer
@@ -48,11 +45,16 @@ type Tool struct {
 var ErrMissingKubeConfig = errors.New("kubeconfig was nil")
 
 func (t *Tool) GetClusters() []*clusterv1.Cluster {
-	// TODO: should this lock???
-	return t.Clusters
+	t.baseMutex.Lock()
+	defer t.baseMutex.Unlock()
+
+	return t.clusters
 }
 
 func (t *Tool) Initialize(ctx context.Context, kubeconfig *string) error {
+	t.baseMutex.Lock()
+	defer t.baseMutex.Unlock()
+
 	if kubeconfig == nil {
 		return ErrMissingKubeConfig
 	}
@@ -79,7 +81,7 @@ func (t *Tool) Initialize(ctx context.Context, kubeconfig *string) error {
 		clusters = append(clusters, cluster)
 	}
 
-	t.Clusters = clusters
+	t.clusters = clusters
 	t.workloadClients = make(map[string]client.Client, size)
 	t.errors = make(map[string]error, size)
 	t.outputBuffers = make(map[string]*bytes.Buffer, size)
@@ -93,8 +95,8 @@ func (t *Tool) HasError(c *clusterv1.Cluster) bool {
 }
 
 func (t *Tool) GetErrorFor(c *clusterv1.Cluster) error {
-	t.mu.Lock()
-	defer t.mu.Unlock()
+	t.baseMutex.Lock()
+	defer t.baseMutex.Unlock()
 
 	if t.errors == nil {
 		return nil
@@ -104,17 +106,17 @@ func (t *Tool) GetErrorFor(c *clusterv1.Cluster) error {
 }
 
 func (t *Tool) GetOutputFor(c *clusterv1.Cluster) string {
-	t.mu.Lock()
-	defer t.mu.Unlock()
+	t.baseMutex.Lock()
+	defer t.baseMutex.Unlock()
 
 	t.flushBuffers()
 
 	return t.outputContents[ClusterToKey(c)]
 }
 
 func (t *Tool) AddErrorFor(c *clusterv1.Cluster, err error) {
-	t.mu.Lock()
-	defer t.mu.Unlock()
+	t.baseMutex.Lock()
+	defer t.baseMutex.Unlock()
 
 	if t.errors == nil {
 		t.errors = make(map[string]error)
@@ -124,8 +126,8 @@ func (t *Tool) AddErrorFor(c *clusterv1.Cluster, err error) {
 }
 
 func (t *Tool) GetBufferFor(c *clusterv1.Cluster) *bytes.Buffer {
-	t.mu.Lock()
-	defer t.mu.Unlock()
+	t.baseMutex.Lock()
+	defer t.baseMutex.Unlock()
 
 	if t.outputBuffers == nil {
 		t.outputBuffers = make(map[string]*bytes.Buffer)
@@ -163,6 +165,9 @@ func (t *Tool) GetWorkloadClient(
 	ctx context.Context,
 	cluster *clusterv1.Cluster,
 ) (client.Client, error) {
+	t.baseMutex.Lock()
+	defer t.baseMutex.Unlock()
+
 	if t.workloadClients == nil {
 		t.workloadClients = make(map[string]client.Client)
 	}

cmd/helper/migrate/providerid/migrator/migrator.go

@@ -21,14 +21,14 @@ import (
 	"sync"
 
 	"github.com/blang/semver"
+	"github.com/docker/distribution/reference"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/client-go/util/retry"
 	"sigs.k8s.io/cluster-api-provider-packet/cmd/helper/base"
 	clusterv1 "sigs.k8s.io/cluster-api/api/v1alpha3"
 	capiutil "sigs.k8s.io/cluster-api/util"
-	containerutil "sigs.k8s.io/cluster-api/util/container"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
@@ -53,9 +53,13 @@ func (m *Migrator) Initialize(ctx context.Context, kubeconfig *string) error {
 		return err
 	}
 
-	m.nodeStatus = make(map[string]map[string]bool, len(m.Clusters))
+	// Initialize the node status
+	m.mu.Lock()
+	clusters := m.GetClusters()
+	m.nodeStatus = make(map[string]map[string]bool, len(clusters))
+	m.mu.Unlock()
 
-	for _, c := range m.Clusters {
+	for _, c := range clusters {
 		nodes, err := m.getNodes(ctx, c)
 		if err != nil {
 			m.AddErrorFor(c, err)
@@ -74,8 +78,9 @@ func (m *Migrator) Initialize(ctx context.Context, kubeconfig *string) error {
 func (m *Migrator) CheckPrerequisites(ctx context.Context) error {
 	wg := new(sync.WaitGroup)
 
-	for i := range m.Clusters {
-		c := m.Clusters[i]
+	clusters := m.GetClusters()
+	for i := range clusters {
+		c := clusters[i]
 
 		wg.Add(1)
 
@@ -90,18 +95,20 @@ func (m *Migrator) CheckPrerequisites(ctx context.Context) error {
 
 	wg.Wait()
 
-	cappDeployment, err := getDeployment(
+	cappDeployment := new(appsv1.Deployment)
+	if err := m.MgmtClient.Get(
 		ctx,
-		m.MgmtClient,
-		"cluster-api-provider-packet-system",
-		"cluster-api-provider-packet-controller-manager",
-	)
-	if err != nil {
-		return fmt.Errorf("failed to get CAPP deployment: %w", err)
-	}
+		client.ObjectKey{
+			Namespace: "cluster-api-provider-packet-system",
+			Name:      "cluster-api-provider-packet-controller-manager",
+		},
+		cappDeployment,
+	); err != nil {
+		if apierrors.IsNotFound(err) {
+			return ErrMissingCAPPDeployment
+		}
 
-	if cappDeployment == nil {
-		return ErrMissingCAPPDeployment
+		return fmt.Errorf("failed to get CAPP deployment: %w", err)
 	}
 
 	ok, err := containerImageGTE(cappDeployment.Spec.Template.Spec.Containers[0], semver.MustParse("0.4.0"))
@@ -117,12 +124,16 @@ func (m *Migrator) CheckPrerequisites(ctx context.Context) error {
 }
 
 func containerImageGTE(container corev1.Container, version semver.Version) (bool, error) {
-	image, err := containerutil.ImageFromString(container.Image)
+	ref, err := reference.ParseNormalizedNamed(container.Image)
 	if err != nil {
-		return false, fmt.Errorf("failed to get image from container: %w", err)
+		return false, fmt.Errorf("failed to parse container reference %s: %w", container.Image, err)
 	}
 
-	imageVersion, err := capiutil.ParseMajorMinorPatch(image.Tag)
+	ref = reference.TagNameOnly(ref)
+	tagged, _ := ref.(reference.Tagged)
+	tag := tagged.Tag()
+
+	imageVersion, err := capiutil.ParseMajorMinorPatch(tag)
 	if err != nil {
 		return false, fmt.Errorf("failed to get version from image: %w", err)
 	}
@@ -141,22 +152,36 @@ func (m *Migrator) validateCloudProviderForCluster(ctx context.Context, c *clust
 		return err
 	}
 
-	packetCCMDeployment, err := getDeployment(ctx, workloadClient, "kube-system", "packet-cloud-controller-manager")
-	if err != nil {
-		return err
-	}
-
-	if packetCCMDeployment != nil {
+	packetCCMDeployment := new(appsv1.Deployment)
+	if err := workloadClient.Get(
+		ctx,
+		client.ObjectKey{Namespace: "kube-system", Name: "packet-cloud-controller-manager"},
+		packetCCMDeployment,
+	); err != nil {
+		if !apierrors.IsNotFound(err) {
+			// Ignore IsNotFound errors, since this is what we want to proceed
+			// We hit an unexpected error
+			return fmt.Errorf("failed to get deployment: %w", err)
+		}
+	} else {
+		// If we successfully retrieved the deployment, that means that the prerequisite step
+		// for upgrading the Cloud Provider has not been run yet
 		return ErrPacketCloudProviderFound
 	}
 
-	cpemDeployment, err := getDeployment(ctx, workloadClient, "kube-system", "cloud-provider-equinix-metal")
-	if err != nil {
-		return err
-	}
+	cpemDeployment := new(appsv1.Deployment)
+	if err := workloadClient.Get(
+		ctx,
+		client.ObjectKey{Namespace: "kube-system", Name: "cloud-provider-equinix-metal"},
+		cpemDeployment,
+	); err != nil {
+		if apierrors.IsNotFound(err) {
+			// Missing expected CPEM deployment
+			return ErrMissingCPEMDeployment
+		}
 
-	if cpemDeployment == nil {
-		return ErrMissingCPEMDeployment
+		// We hit an unexpected error
+		return fmt.Errorf("failed to get deployment: %w", err)
 	}
 
 	ok, err := containerImageGTE(cpemDeployment.Spec.Template.Spec.Containers[0], semver.MustParse("3.1.0"))
@@ -171,25 +196,6 @@ func (m *Migrator) validateCloudProviderForCluster(ctx context.Context, c *clust
 	return nil
 }
 
-func getDeployment(
-	ctx context.Context,
-	workloadClient client.Client,
-	namespace, name string,
-) (*appsv1.Deployment, error) {
-	deployment := new(appsv1.Deployment)
-	key := client.ObjectKey{Namespace: namespace, Name: name}
-
-	if err := workloadClient.Get(ctx, key, deployment); err != nil {
-		if apierrors.IsNotFound(err) {
-			return nil, nil
-		}
-
-		return nil, fmt.Errorf("failed to get deployment: %w", err)
-	}
-
-	return deployment, nil
-}
-
 func (m *Migrator) CalculatePercentage() float64 {
 	m.mu.Lock()
 	defer m.mu.Unlock()
@@ -200,7 +206,7 @@ func (m *Migrator) CalculatePercentage() float64 {
 
 	var totalNodes, doneNodes int
 
-	for _, c := range m.Clusters {
+	for _, c := range m.GetClusters() {
 		clusterKey := base.ClusterToKey(c)
 
 		if m.nodeStatus[clusterKey] == nil {
@@ -226,8 +232,9 @@ func (m *Migrator) CalculatePercentage() float64 {
 func (m *Migrator) Run(ctx context.Context) {
 	wg := new(sync.WaitGroup)
 
-	for i := range m.Clusters {
-		c := m.Clusters[i]
+	clusters := m.GetClusters()
+	for i := range clusters {
+		c := clusters[i]
 
 		wg.Add(1)
 

cmd/helper/ui/base.go

@@ -21,6 +21,7 @@ import (
 	"github.com/charmbracelet/bubbles/progress"
 	tea "github.com/charmbracelet/bubbletea"
 	"github.com/charmbracelet/lipgloss"
+	"github.com/muesli/reflow/wordwrap"
 	clusterv1 "sigs.k8s.io/cluster-api/api/v1alpha3"
 )
 
@@ -58,6 +59,8 @@ type Model struct {
 	progress *progress.Model
 	percent  float64
 	err      error
+	height   int
+	width    int
 }
 
 type TickMsg time.Time
@@ -92,6 +95,8 @@ func (m Model) Update(msg tea.Msg) (tea.Model, tea.Cmd) {
 			cmd = cleanQuit
 		}
 	case tea.WindowSizeMsg:
+		m.width = msg.Width
+		m.height = msg.Height
 		m.progress.Width = msg.Width - padding*2 - 4 //nolint: gomnd
 		if m.progress.Width > maxWidth {
 			m.progress.Width = maxWidth
@@ -145,7 +150,7 @@ func (m Model) View() string {
 		}
 	}
 
-	return s
+	return wordwrap.String(s, m.width)
 }
 
 func cleanQuit() tea.Msg {