Merge branch 'KUBE-1216--go-mod' into KUBE-1163-delete-with-provider-id

Author: ValyaB

e2e/client/api.gen.go

@@ -24,12 +24,15 @@ import (
 )
 
 const (
-	approveCSRTimeout              = 4 * time.Minute
-	groupSystemNodesName           = "system:nodes"
+	approveCSRTimeout    = 4 * time.Minute
+	groupSystemNodesName = "system:nodes"
+	// kubeletBootstrapRequestingUser is the observed requestor for node's CSRs from real GKE clusters.
 	kubeletBootstrapRequestingUser = "kubelet-bootstrap"
-	clusterControllerSAName        = "system:serviceaccount:castai-agent:castai-cluster-controller"
-	approvedMessage                = "This CSR was approved by CAST AI"
-	csrOutdatedAfter               = time.Hour
+	// kubeletNodepoolBootstrapRequestingUser is the observed requestor for node's CSRs for GKE clusters 1.33+.
+	kubeletNodepoolBootstrapRequestingUser = "kubelet-nodepool-bootstrap"
+	clusterControllerSAName                = "system:serviceaccount:castai-agent:castai-cluster-controller"
+	approvedMessage                        = "This CSR was approved by CAST AI"
+	csrOutdatedAfter                       = time.Hour
 )
 
 func NewApprovalManager(log logrus.FieldLogger, clientset kubernetes.Interface) *ApprovalManager {
@@ -148,8 +151,8 @@ func (m *ApprovalManager) runAutoApproveForCastAINodes(ctx context.Context, c <-
 					"signer": csr.SignerName(),
 					"csr":    csr.Name(),
 				})
-				if shouldSkip(log, csr) {
-					log.Debug("skipping csr")
+				if skip, reason := shouldSkip(csr); skip {
+					log.Infof("skipping csr due to reason: %s", reason)
 					return
 				}
 				log.Info("auto approving csr")
@@ -221,32 +224,26 @@ func (m *ApprovalManager) handle(ctx context.Context, log logrus.FieldLogger, cs
 	return errors.New("certificate signing request was not approved")
 }
 
-func shouldSkip(log logrus.FieldLogger, csr *wrapper.CSR) bool {
+func shouldSkip(csr *wrapper.CSR) (bool, string) {
 	if csr.Approved() {
-		log.Debug("csr already approved")
-		return true
+		return true, "csr already approved"
 	}
 	if time.Since(csr.CreatedAt()) > csrOutdatedAfter {
-		log.Debug("csr is outdated")
-		return true
+		return true, fmt.Sprintf("csr is outdated, %v is larger than %v", time.Since(csr.CreatedAt()), csrOutdatedAfter)
 	}
 	if !managedSigner(csr.SignerName()) {
-		log.Debug("csr unknown signer")
-		return true
+		return true, fmt.Sprintf("csr unknown signer: %s", csr.SignerName())
 	}
 	if !managedCSRNamePrefix(csr.Name()) {
-		log.Debug("csr name not managed by CAST AI: ", csr.Name())
-		return true
+		return true, fmt.Sprintf("csr name not managed by CAST AI: %s", csr.Name())
 	}
 	if !managedCSRRequestingUser(csr.RequestingUser()) {
-		log.Debug("csr requesting user is not managed by CAST AI")
-		return true
+		return true, fmt.Sprintf("csr requesting user is not managed by CAST AI: %s", csr.RequestingUser())
 	}
 	if !managerSubjectCommonName(csr.ParsedCertificateRequest().Subject.CommonName) {
-		log.Debug("csr common name is not managed by CAST AI")
-		return true
+		return true, fmt.Sprintf("csr common name is not managed by CAST AI %s", csr.ParsedCertificateRequest().Subject.CommonName)
 	}
-	return false
+	return false, ""
 }
 
 func managedSigner(signerName string) bool {
@@ -259,7 +256,10 @@ func managedCSRNamePrefix(n string) bool {
 }
 
 func managedCSRRequestingUser(s string) bool {
-	return s == kubeletBootstrapRequestingUser || s == clusterControllerSAName || strings.HasPrefix(s, "system:node:")
+	return s == kubeletBootstrapRequestingUser || // kubelet bootstrap user variation
+		s == kubeletNodepoolBootstrapRequestingUser || // kubelet bootstrap user variation (observed initially on GKE 1.33+)
+		s == clusterControllerSAName || // if created by CC
+		strings.HasPrefix(s, "system:node:") // Post-bootstrap node user; usually for serving certificate
 }
 
 func managerSubjectCommonName(commonName string) bool {

e2e/client/client.gen.go

@@ -78,6 +78,7 @@ func TestCSRApprove(t *testing.T) {
 
 		csrResult, err := client.CertificatesV1().CertificateSigningRequests().Get(ctx, csrName, metav1.GetOptions{})
 		r.NoError(err)
+		r.GreaterOrEqual(len(csrResult.Status.Conditions), 1)
 
 		r.Equal(csrResult.Status.Conditions[0].Type, certv1.CertificateApproved)
 	})
@@ -115,6 +116,112 @@ func TestCSRApprove(t *testing.T) {
 		r.NoError(err)
 		r.Len(csrResult.Status.Conditions, 0)
 	})
+
+	t.Run("approves for kubelet-bootstrap user", func(t *testing.T) {
+		r := require.New(t)
+		t.Parallel()
+
+		csrName := "node-csr-123"
+		userName := "kubelet-bootstrap"
+		client := fake.NewClientset(getCSRv1(csrName, userName))
+		s := NewApprovalManager(log, client)
+		watcher := watch.NewFake()
+		client.PrependWatchReactor("certificatesigningrequests", ktest.DefaultWatchReactor(watcher, nil))
+
+		ctx := context.Background()
+		var wg sync.WaitGroup
+		wg.Add(2)
+		go func() {
+			defer wg.Done()
+			if err := s.Start(ctx); err != nil {
+				t.Logf("failed to start approval manager: %s", err.Error())
+			}
+		}()
+		go func() {
+			defer wg.Done()
+			watcher.Add(getCSRv1(csrName, userName))
+			time.Sleep(100 * time.Millisecond)
+			s.Stop()
+		}()
+
+		wg.Wait()
+
+		csrResult, err := client.CertificatesV1().CertificateSigningRequests().Get(ctx, csrName, metav1.GetOptions{})
+		r.NoError(err)
+		r.GreaterOrEqual(len(csrResult.Status.Conditions), 1)
+
+		r.Equal(csrResult.Status.Conditions[0].Type, certv1.CertificateApproved)
+	})
+
+	t.Run("approves for kubelet-nodepool-bootstrap user", func(t *testing.T) {
+		r := require.New(t)
+		t.Parallel()
+
+		csrName := "node-csr-123"
+		userName := "kubelet-nodepool-bootstrap"
+		client := fake.NewClientset(getCSRv1(csrName, userName))
+		s := NewApprovalManager(log, client)
+		watcher := watch.NewFake()
+		client.PrependWatchReactor("certificatesigningrequests", ktest.DefaultWatchReactor(watcher, nil))
+
+		ctx := context.Background()
+		var wg sync.WaitGroup
+		wg.Add(2)
+		go func() {
+			defer wg.Done()
+			if err := s.Start(ctx); err != nil {
+				t.Logf("failed to start approval manager: %s", err.Error())
+			}
+		}()
+		go func() {
+			defer wg.Done()
+			watcher.Add(getCSRv1(csrName, userName))
+			time.Sleep(100 * time.Millisecond)
+			s.Stop()
+		}()
+
+		wg.Wait()
+
+		csrResult, err := client.CertificatesV1().CertificateSigningRequests().Get(ctx, csrName, metav1.GetOptions{})
+		r.NoError(err)
+		r.GreaterOrEqual(len(csrResult.Status.Conditions), 1)
+
+		r.Equal(csrResult.Status.Conditions[0].Type, certv1.CertificateApproved)
+	})
+
+	t.Run("skips for unknown user", func(t *testing.T) {
+		r := require.New(t)
+		t.Parallel()
+
+		csrName := "node-csr-123"
+		userName := "some-unknown-user"
+		client := fake.NewClientset(getCSRv1(csrName, userName))
+		s := NewApprovalManager(log, client)
+		watcher := watch.NewFake()
+		client.PrependWatchReactor("certificatesigningrequests", ktest.DefaultWatchReactor(watcher, nil))
+
+		ctx := context.Background()
+		var wg sync.WaitGroup
+		wg.Add(2)
+		go func() {
+			defer wg.Done()
+			if err := s.Start(ctx); err != nil {
+				t.Logf("failed to start approval manager: %s", err.Error())
+			}
+		}()
+		go func() {
+			defer wg.Done()
+			watcher.Add(getCSRv1(csrName, userName))
+			time.Sleep(100 * time.Millisecond)
+			s.Stop()
+		}()
+
+		wg.Wait()
+
+		csrResult, err := client.CertificatesV1().CertificateSigningRequests().Get(ctx, csrName, metav1.GetOptions{})
+		r.NoError(err)
+		r.Len(csrResult.Status.Conditions, 0)
+	})
 }
 
 func TestApproveCSRExponentialBackoff(t *testing.T) {

internal/actions/csr/svc.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"reflect"
+	"time"
 
 	"github.com/sirupsen/logrus"
 	v1 "k8s.io/api/core/v1"
@@ -22,14 +23,19 @@ import (
 
 func NewEvictPodHandler(log logrus.FieldLogger, clientset kubernetes.Interface) ActionHandler {
 	return &EvictPodHandler{
-		log:       log,
-		clientset: clientset,
+		log:                           log,
+		clientset:                     clientset,
+		podEvictRetryDelay:            5 * time.Second,
+		podsTerminationWaitRetryDelay: 10 * time.Second,
 	}
 }
 
 type EvictPodHandler struct {
 	log       logrus.FieldLogger
 	clientset kubernetes.Interface
+
+	podEvictRetryDelay            time.Duration
+	podsTerminationWaitRetryDelay time.Duration
 }
 
 func (h *EvictPodHandler) Handle(ctx context.Context, action *castai.ClusterAction) error {
@@ -93,9 +99,10 @@ func (h *EvictPodHandler) evictPod(ctx context.Context, log logrus.FieldLogger,
 		return fmt.Errorf("unsupported eviction version: %s", groupVersion.String())
 	}
 
+	backoff := waitext.NewConstantBackoff(h.podEvictRetryDelay)
 	return waitext.Retry(
 		ctx,
-		defaultBackoff(),
+		backoff,
 		waitext.Forever,
 		func(ctx context.Context) (bool, error) {
 			err := submit(ctx)
@@ -119,9 +126,10 @@ func (h *EvictPodHandler) evictPod(ctx context.Context, log logrus.FieldLogger,
 }
 
 func (h *EvictPodHandler) waitForPodToBeDeleted(ctx context.Context, log logrus.FieldLogger, namespace, name string) error {
+	backoff := waitext.NewConstantBackoff(h.podsTerminationWaitRetryDelay)
 	return waitext.Retry(
 		ctx, // controls how long we might wait at most.
-		defaultBackoff(),
+		backoff,
 		waitext.Forever,
 		func(ctx context.Context) (bool, error) {
 			deleted, phase, err := h.isPodDeleted(ctx, namespace, name)