diff --git a/pkg/datapath/linux/ipsec/cell_test.go b/pkg/datapath/linux/ipsec/cell_test.go index 00df198a94369..2faeba984f963 100644 --- a/pkg/datapath/linux/ipsec/cell_test.go +++ b/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, diff --git a/pkg/dial/resolver.go b/pkg/dial/resolver.go index b5815b0c3d5c4..57f4d5777aef2 100644 --- a/pkg/dial/resolver.go +++ b/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 { diff --git a/pkg/loadbalancer/reflectors/k8s.go b/pkg/loadbalancer/reflectors/k8s.go index 6de989d83a3fd..032da4f219cb6 100644 --- a/pkg/loadbalancer/reflectors/k8s.go +++ b/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. diff --git a/pkg/wireguard/agent/cell_test.go b/pkg/wireguard/agent/cell_test.go index 49fbeefcf1f62..77de1de4d8602 100644 --- a/pkg/wireguard/agent/cell_test.go +++ b/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,