Merge pull request #655 from DataDog/alm/backport-pr44527

[CMPT-3823] Backport fix for KVStore service resolution

Co-authored-by: 41ks <alex.melhem@datadoghq.com>

Author: gh-worker-dd-mergequeue-cf854d[bot]

pkg/datapath/linux/ipsec/cell_test.go

@@ -40,6 +40,7 @@ import (
 	"github.com/cilium/cilium/pkg/kvstore"
 	"github.com/cilium/cilium/pkg/kvstore/store"
 	"github.com/cilium/cilium/pkg/loadbalancer"
+	"github.com/cilium/cilium/pkg/loadbalancer/reflectors"
 	"github.com/cilium/cilium/pkg/loadbalancer/writer"
 	"github.com/cilium/cilium/pkg/maps/encrypt"
 	"github.com/cilium/cilium/pkg/mtu"
@@ -133,15 +134,18 @@ func TestPrivileged_TestIPSecCell(t *testing.T) {
 			source.Cell,
 			watchers.Cell,
 			dial.ServiceResolverCell,
+			reflectors.K8sReflectorCell,
 			clustermesh.Cell,
 			writer.Cell,
 			ipset.Cell,
 			k8s.ResourcesCell,
+			k8s.PodTableCell,
 			node.LocalNodeStoreTestCell,
 			k8sClient.FakeClientCell(),
 			kvstore.Cell(kvstore.DisabledBackendName),
 
 			cell.Provide(
+				reflectors.NetnsCookieSupportFunc,
 				newIPsecAgent,
 				newIPsecConfig,
 

pkg/dial/resolver.go

@@ -4,9 +4,11 @@
 package dial
 
 import (
+	"cmp"
 	"context"
 	"fmt"
 	"iter"
+	"log/slog"
 	"math/rand/v2"
 	"net/netip"
 	"slices"
@@ -17,14 +19,17 @@ import (
 	"github.com/cilium/hive/cell"
 	"github.com/cilium/hive/job"
 	"github.com/cilium/statedb"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 
 	cmtypes "github.com/cilium/cilium/pkg/clustermesh/types"
+	k8sClient "github.com/cilium/cilium/pkg/k8s/client"
 	"github.com/cilium/cilium/pkg/k8s/resource"
 	slim_corev1 "github.com/cilium/cilium/pkg/k8s/slim/k8s/api/core/v1"
 	"github.com/cilium/cilium/pkg/loadbalancer"
 	"github.com/cilium/cilium/pkg/loadbalancer/reflectors"
 	"github.com/cilium/cilium/pkg/lock"
+	"github.com/cilium/cilium/pkg/logging/logfields"
 	"github.com/cilium/cilium/pkg/time"
 )
 
@@ -126,15 +131,23 @@ var _ Resolver = (*lbServiceResolver)(nil)
 
 // lbServiceResolver maps DNS names matching Kubernetes services to the
 // corresponding ClusterIP address using Table[*Frontend].
+// If the frontend table lookup fails, it falls back to fetching the service
+// directly from the kube-apiserver.
 type lbServiceResolver struct {
 	db        *statedb.DB
 	frontends statedb.Table[*loadbalancer.Frontend]
+	cs        k8sClient.Clientset
+	log       *slog.Logger
 }
 
-func newLBServiceResolver(jg job.Group, db *statedb.DB, frontends statedb.Table[*loadbalancer.Frontend]) Resolver {
+// A dependency on reflector.K8sReflectorRegistered is used to ensure the order of initialization.
+// Otherwise, the resolver could be used before the frontends table is initialized.
+func newLBServiceResolver(_ reflectors.K8sReflectorRegistered, jg job.Group, db *statedb.DB, frontends statedb.Table[*loadbalancer.Frontend], cs k8sClient.Clientset, log *slog.Logger) Resolver {
 	return &lbServiceResolver{
 		db:        db,
 		frontends: frontends,
+		cs:        cs,
+		log:       log,
 	}
 }
 
@@ -152,8 +165,11 @@ func (sr *lbServiceResolver) resolve(ctx context.Context, host string) string {
 	// Wait for the frontends table to be initialized from k8s. We can't check that
 	// the table has been initialized by all initializers since at least ClusterMesh
 	// uses [Resolve] to look up KVStore address.
+	// If the frontends table is not initialized, fallback to kube-apiserver.
 	txn := sr.db.ReadTxn()
 	init, waitInit := sr.frontends.Initialized(txn)
+	// We give the frontends table 5 seconds to be initialized to avoid a deadlock.
+	initTimeout := time.After(5 * time.Second)
 	for !init {
 		pending := sr.frontends.PendingInitializers(txn)
 		if !slices.ContainsFunc(pending, func(s string) bool { return strings.HasPrefix(s, reflectors.K8sInitializerPrefix) }) {
@@ -164,6 +180,13 @@ func (sr *lbServiceResolver) resolve(ctx context.Context, host string) string {
 			return host
 		case <-waitInit:
 			init = true
+		case <-initTimeout:
+			sr.log.Warn(
+				"Frontends table not initialized, falling back to kube-apiserver",
+				logfields.K8sNamespace, nsname.Namespace,
+				logfields.K8sSvcName, nsname.Name,
+			)
+			return cmp.Or(sr.resolveFromAPIServer(ctx, nsname), host)
 		case <-time.After(100 * time.Millisecond):
 		}
 		txn = sr.db.ReadTxn()
@@ -183,6 +206,22 @@ func (sr *lbServiceResolver) resolve(ctx context.Context, host string) string {
 	return host
 }
 
+// resolveFromAPIServer fetches the service directly from the kube-apiserver
+// as a fallback when the frontends table takes too long to be initialized.
+func (sr *lbServiceResolver) resolveFromAPIServer(ctx context.Context, nsname types.NamespacedName) string {
+	svc, err := sr.cs.Slim().CoreV1().Services(nsname.Namespace).Get(ctx, nsname.Name, metav1.GetOptions{})
+	if err != nil {
+		return ""
+	}
+
+	if _, err := netip.ParseAddr(svc.Spec.ClusterIP); err != nil {
+		// The ClusterIP is not a valid IP address (e.g., headless service)
+		return ""
+	}
+
+	return svc.Spec.ClusterIP
+}
+
 func ServiceURLToNamespacedName(host string) (types.NamespacedName, error) {
 	tokens := strings.Split(host, ".")
 	if len(tokens) < 2 {

pkg/loadbalancer/reflectors/k8s.go

@@ -65,8 +65,9 @@ var K8sReflectorCell = cell.Module(
 	"k8s-reflector",
 	"Reflects load-balancing state from Kubernetes",
 
+	cell.Provide(provideK8sReflector),
 	cell.ProvidePrivate(newEventStream),
-	cell.Invoke(RegisterK8sReflector),
+	cell.Invoke(func(_ K8sReflectorRegistered) {}),
 )
 
 type reflectorParams struct {
@@ -88,6 +89,13 @@ type reflectorParams struct {
 	SVCMetrics             SVCMetrics `optional:"true"`
 }
 
+type K8sReflectorRegistered struct{}
+
+func provideK8sReflector(p reflectorParams) K8sReflectorRegistered {
+	RegisterK8sReflector(p)
+	return K8sReflectorRegistered{}
+}
+
 func (p reflectorParams) waitTime() time.Duration {
 	if p.TestConfig != nil {
 		// Use a much lower wait time in tests to trigger more edge cases and make them faster.

pkg/wireguard/agent/cell_test.go

@@ -40,6 +40,7 @@ import (
 	"github.com/cilium/cilium/pkg/kvstore"
 	"github.com/cilium/cilium/pkg/kvstore/store"
 	"github.com/cilium/cilium/pkg/loadbalancer"
+	"github.com/cilium/cilium/pkg/loadbalancer/reflectors"
 	"github.com/cilium/cilium/pkg/loadbalancer/writer"
 	"github.com/cilium/cilium/pkg/mtu"
 	"github.com/cilium/cilium/pkg/node"
@@ -119,16 +120,19 @@ func TestPrivileged_TestWireGuardCell(t *testing.T) {
 			source.Cell,
 			watchers.Cell,
 			dial.ServiceResolverCell,
+			reflectors.K8sReflectorCell,
 			clustermesh.Cell,
 			writer.Cell,
 			ipset.Cell,
 			k8s.ResourcesCell,
+			k8s.PodTableCell,
 			cell.Config(envoyCfg.SecretSyncConfig{}),
 			k8sClient.FakeClientCell(),
 			kvstore.Cell(kvstore.DisabledBackendName),
 			node.LocalNodeStoreTestCell,
 
 			cell.Provide(
+				reflectors.NetnsCookieSupportFunc,
 				newWireguardAgent,
 				newWireguardConfig,