Start konnectivity BEFORE api-server

cmd/controller/controller.go

@@ -216,10 +216,32 @@ func (c *command) start(ctx context.Context) error {
 	logrus.Infof("using storage backend %s", nodeConfig.Spec.Storage.Type)
 	nodeComponents.Add(ctx, storageBackend)
 
+	controllerLeaseCounter := &controller.K0sControllersLeaseCounter{
+		ClusterConfig:     nodeConfig,
+		KubeClientFactory: adminClientFactory,
+	}
+
+	if !c.SingleNode {
+		nodeComponents.Add(ctx, controllerLeaseCounter)
+	}
+
 	enableKonnectivity := !c.SingleNode && !slices.Contains(c.DisableComponents, constant.KonnectivityServerComponentName)
 	disableEndpointReconciler := !slices.Contains(c.DisableComponents, constant.APIEndpointReconcilerComponentName) &&
 		nodeConfig.Spec.API.ExternalAddress != ""
 
+	if enableKonnectivity {
+		nodeComponents.Add(ctx, &controller.Konnectivity{
+			SingleNode:                 c.SingleNode,
+			LogLevel:                   c.Logging[constant.KonnectivityServerComponentName],
+			K0sVars:                    c.K0sVars,
+			KubeClientFactory:          adminClientFactory,
+			NodeConfig:                 nodeConfig,
+			EventEmitter:               prober.NewEventEmitter(),
+			K0sControllersLeaseCounter: controllerLeaseCounter,
+		})
+
+	}
+
 	nodeComponents.Add(ctx, &controller.APIServer{
 		ClusterConfig:             nodeConfig,
 		K0sVars:                   c.K0sVars,
@@ -229,13 +251,6 @@ func (c *command) start(ctx context.Context) error {
 		DisableEndpointReconciler: disableEndpointReconciler,
 	})
 
-	if !c.SingleNode {
-		nodeComponents.Add(ctx, &controller.K0sControllersLeaseCounter{
-			ClusterConfig:     nodeConfig,
-			KubeClientFactory: adminClientFactory,
-		})
-	}
-
 	var leaderElector interface {
 		leaderelector.Interface
 		manager.Component
@@ -467,13 +482,14 @@ func (c *command) start(ctx context.Context) error {
 	}
 
 	if enableKonnectivity {
-		clusterComponents.Add(ctx, &controller.Konnectivity{
-			SingleNode:        c.SingleNode,
-			LogLevel:          c.Logging[constant.KonnectivityServerComponentName],
-			K0sVars:           c.K0sVars,
-			KubeClientFactory: adminClientFactory,
-			NodeConfig:        nodeConfig,
-			EventEmitter:      prober.NewEventEmitter(),
+		clusterComponents.Add(ctx, &controller.KonnectivityAgent{
+			SingleNode:                 c.SingleNode,
+			LogLevel:                   c.Logging[constant.KonnectivityServerComponentName],
+			K0sVars:                    c.K0sVars,
+			KubeClientFactory:          adminClientFactory,
+			NodeConfig:                 nodeConfig,
+			EventEmitter:               prober.NewEventEmitter(),
+			K0sControllersLeaseCounter: controllerLeaseCounter,
 		})
 	}
 

inttest/upgrade/upgrade_test.go

@@ -166,6 +166,9 @@ func (s *UpgradeSuite) TestK0sGetsUp() {
 
 	err = s.WaitForNodeReady(s.WorkerNode(1), kc)
 	s.NoError(err)
+
+	s.Require().NoError(common.WaitForPodLogs(s.Context(), kc, "kube-system"))
+
 }
 
 func TestUpgradeSuite(t *testing.T) {

pkg/component/controller/controllersleasecounter.go

@@ -19,6 +19,7 @@ package controller
 import (
 	"context"
 	"fmt"
+	"time"
 
 	"github.com/k0sproject/k0s/pkg/apis/k0s/v1beta1"
 	"github.com/k0sproject/k0s/pkg/component/manager"
@@ -37,12 +38,16 @@ type K0sControllersLeaseCounter struct {
 
 	cancelFunc  context.CancelFunc
 	leaseCancel context.CancelFunc
+
+	subscribers []chan int
 }
 
 var _ manager.Component = (*K0sControllersLeaseCounter)(nil)
 
 // Init initializes the component needs
 func (l *K0sControllersLeaseCounter) Init(_ context.Context) error {
+	l.subscribers = make([]chan int, 0)
+
 	return nil
 }
 
@@ -88,6 +93,9 @@ func (l *K0sControllersLeaseCounter) Start(ctx context.Context) error {
 			}
 		}
 	}()
+
+	go l.runLeaseCounter(ctx)
+
 	return nil
 }
 
@@ -102,3 +110,54 @@ func (l *K0sControllersLeaseCounter) Stop() error {
 	}
 	return nil
 }
+
+// Check the numbers of controller every 10 secs and notify the subscribers
+func (l *K0sControllersLeaseCounter) runLeaseCounter(ctx context.Context) {
+	log := logrus.WithFields(logrus.Fields{"component": "controllerlease"})
+	log.Debug("starting controller lease counter every 10 secs")
+	ticker := time.NewTicker(10 * time.Second)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			log.Info("stopping controller lease counter")
+			return
+		case <-ticker.C:
+			log.Debug("counting controller lease holders")
+			count, err := l.countLeaseHolders(ctx)
+			if err != nil {
+				log.Errorf("failed to count controller leases: %s", err)
+			}
+			l.notifySubscribers(count)
+		}
+	}
+}
+
+func (l *K0sControllersLeaseCounter) countLeaseHolders(ctx context.Context) (int, error) {
+	client, err := l.KubeClientFactory.GetClient()
+	if err != nil {
+		return 0, err
+	}
+
+	return kubeutil.GetControlPlaneNodeCount(ctx, client)
+}
+
+// Notify the subscribers about the current controller count
+func (l *K0sControllersLeaseCounter) notifySubscribers(count int) {
+	log := logrus.WithFields(logrus.Fields{"component": "controllerlease"})
+	log.Debugf("notifying subscribers (%d) about controller count: %d", len(l.subscribers), count)
+	for _, ch := range l.subscribers {
+		// Use non-blocking send to avoid blocking the loop
+		select {
+		case ch <- count:
+		case <-time.After(5 * time.Second):
+			log.Warn("timeout when sending count to subsrciber")
+		}
+	}
+}
+
+func (l *K0sControllersLeaseCounter) Subscribe() <-chan int {
+	ch := make(chan int, 1)
+	l.subscribers = append(l.subscribers, ch)
+	return ch
+}