From 70634312c71caae98efcfe77b07c4f2ba69cb1f8 Mon Sep 17 00:00:00 2001 From: WorrierKhushal Date: Wed, 19 Aug 2026 15:48:00 +0530 Subject: [PATCH 1/2] fix(raven): fix GatewayPublicService update on failover or multiple replicas --- .../gateway_public_service_controller.go | 14 ++- .../gateway_public_service_controller_test.go | 116 +++++++++++++++++- 2 files changed, 120 insertions(+), 10 deletions(-) diff --git a/pkg/yurtmanager/controller/raven/gatewaypublicservice/gateway_public_service_controller.go b/pkg/yurtmanager/controller/raven/gatewaypublicservice/gateway_public_service_controller.go index e3e2084c2f9..94034c40592 100644 --- a/pkg/yurtmanager/controller/raven/gatewaypublicservice/gateway_public_service_controller.go +++ b/pkg/yurtmanager/controller/raven/gatewaypublicservice/gateway_public_service_controller.go @@ -280,8 +280,9 @@ func (r *ReconcileService) manageService(ctx context.Context, gateway *ravenv1be for i := 0; i < len(addSvc); i++ { if err := r.Create(ctx, addSvc[i]); err != nil { if apierrs.IsAlreadyExists(err) { - klog.V(2).Info(Format("service %s/%s has already exist, ignore creating it", addSvc[i].GetNamespace(), addSvc[i].GetName())) - return nil + klog.V(2).Info(Format("service %s/%s has already exist, reconcile it as update", addSvc[i].GetNamespace(), addSvc[i].GetName())) + updateSvc = append(updateSvc, addSvc[i]) + continue } return fmt.Errorf("failed create service for gateway %s type %s , error %s", gateway.GetName(), gatewayType, err.Error()) } @@ -309,8 +310,9 @@ func (r *ReconcileService) manageEndpoints(ctx context.Context, gateway *ravenv1 for i := 0; i < len(addEps); i++ { if err := r.Create(ctx, addEps[i]); err != nil { if apierrs.IsAlreadyExists(err) { - klog.V(2).Info(Format("endpoints %s/%s has already exist, ignore creating it", addEps[i].GetNamespace(), addEps[i].GetName())) - return nil + klog.V(2).Info(Format("endpoints %s/%s has already exist, reconcile it as update", addEps[i].GetNamespace(), addEps[i].GetName())) + updateEps = append(updateEps, addEps[i]) + continue } return fmt.Errorf("failed create endpoints for gateway %s type %s , error %s", gateway.GetName(), gatewayType, err.Error()) } @@ -486,7 +488,7 @@ func acquiredSpecService(gateway *ravenv1beta1.Gateway, gatewayType string, prox case ravenv1beta1.Proxy: services = append(services, corev1.Service{ ObjectMeta: metav1.ObjectMeta{ - Name: util.FormatName(fmt.Sprintf("%s-%s", util.GatewayProxyServiceNamePrefix, gateway.GetName())), + Name: util.FormatName(fmt.Sprintf("%s-%s-%s", util.GatewayProxyServiceNamePrefix, gateway.GetName(), aep.NodeName)), Namespace: util.WorkingNamespace, Labels: map[string]string{ raven.LabelCurrentGateway: gateway.GetName(), @@ -513,7 +515,7 @@ func acquiredSpecService(gateway *ravenv1beta1.Gateway, gatewayType string, prox case ravenv1beta1.Tunnel: services = append(services, corev1.Service{ ObjectMeta: metav1.ObjectMeta{ - Name: util.FormatName(fmt.Sprintf("%s-%s", util.GatewayTunnelServiceNamePrefix, gateway.GetName())), + Name: util.FormatName(fmt.Sprintf("%s-%s-%s", util.GatewayTunnelServiceNamePrefix, gateway.GetName(), aep.NodeName)), Namespace: util.WorkingNamespace, Labels: map[string]string{ raven.LabelCurrentGateway: gateway.GetName(), diff --git a/pkg/yurtmanager/controller/raven/gatewaypublicservice/gateway_public_service_controller_test.go b/pkg/yurtmanager/controller/raven/gatewaypublicservice/gateway_public_service_controller_test.go index 97b15530b51..9eaee3d8861 100644 --- a/pkg/yurtmanager/controller/raven/gatewaypublicservice/gateway_public_service_controller_test.go +++ b/pkg/yurtmanager/controller/raven/gatewaypublicservice/gateway_public_service_controller_test.go @@ -14,6 +14,7 @@ See the License for the specific language governing permissions and limitations under the License. */ +//nolint:staticcheck // SA1019: corev1.Endpoints is deprecated but still supported for backward compatibility package gatewaypublicservice import ( @@ -206,12 +207,119 @@ func MockReconcile() *ReconcileService { } } -func TestReconcileService_Reconcile(t *testing.T) { +func servicesByEndpointNode(t *testing.T, r *ReconcileService) map[string]string { + t.Helper() + svcList := &corev1.ServiceList{} + if err := r.List(context.Background(), svcList); err != nil { + t.Fatalf("failed to list services: %v", err) + } + byNode := make(map[string]string) + for _, svc := range svcList.Items { + byNode[svc.Labels[util.LabelCurrentGatewayEndpoints]] = svc.Name + } + return byNode +} + +func assertEndpointsForService(t *testing.T, r *ReconcileService, name, expectedIP string) { + t.Helper() + eps := &corev1.Endpoints{} + if err := r.Get(context.Background(), types.NamespacedName{Namespace: util.WorkingNamespace, Name: name}, eps); err != nil { + t.Errorf("expected endpoints %q to exist, got error: %v", name, err) + return + } + if len(eps.Subsets) != 1 || len(eps.Subsets[0].Addresses) != 1 || eps.Subsets[0].Addresses[0].IP != expectedIP { + t.Errorf("expected endpoints %q to contain IP %q, got %v", name, expectedIP, eps.Subsets) + } +} + +func TestReconcileService_TwoReplicas(t *testing.T) { r := MockReconcile() - _, err := r.Reconcile(context.Background(), reconcile.Request{NamespacedName: types.NamespacedName{Name: MockGateway}}) - if err != nil { - t.Errorf("failed to reconcile service %s", MockGateway) + if _, err := r.Reconcile(context.Background(), reconcile.Request{NamespacedName: types.NamespacedName{Name: MockGateway}}); err != nil { + t.Fatalf("failed to reconcile service %s: %v", MockGateway, err) + } + + byNode := servicesByEndpointNode(t, r) + if len(byNode) != 2 { + t.Errorf("expected 2 LoadBalancer services for 2 active proxy endpoints, got %d", len(byNode)) + } + for _, nodeName := range []string{Node1Name, Node2Name} { + name, ok := byNode[nodeName] + if !ok { + t.Errorf("expected a service for active node %q, none found", nodeName) + continue + } + svc := &corev1.Service{} + if err := r.Get(context.Background(), types.NamespacedName{Namespace: util.WorkingNamespace, Name: name}, svc); err != nil { + t.Errorf("failed to get service %q: %v", name, err) + continue + } + if svc.Spec.Type != corev1.ServiceTypeLoadBalancer { + t.Errorf("expected service %q to be of type LoadBalancer, got %s", name, svc.Spec.Type) + } + assertEndpointsForService(t, r, name, map[string]string{Node1Name: Node1Address, Node2Name: Node2Address}[nodeName]) + } +} + +func TestReconcileService_Failover(t *testing.T) { + r := MockReconcile() + + gateway := &ravenv1beta1.Gateway{} + if err := r.Get(context.Background(), types.NamespacedName{Name: MockGateway}, gateway); err != nil { + t.Fatalf("failed to get gateway: %v", err) + } + gateway.Status.ActiveEndpoints = []*ravenv1beta1.Endpoint{ + { + NodeName: Node1Name, + Type: ravenv1beta1.Proxy, + Port: ravenv1beta1.DefaultProxyServerExposedPort, + UnderNAT: false, + }, + } + if err := r.Update(context.Background(), gateway); err != nil { + t.Fatalf("failed to update gateway: %v", err) + } + + if _, err := r.Reconcile(context.Background(), reconcile.Request{NamespacedName: types.NamespacedName{Name: MockGateway}}); err != nil { + t.Fatalf("failed to reconcile service %s: %v", MockGateway, err) + } + + byNode := servicesByEndpointNode(t, r) + if len(byNode) != 1 || byNode[Node1Name] == "" { + t.Errorf("expected a service only for active node %q, got %v", Node1Name, byNode) + } + + gateway = &ravenv1beta1.Gateway{} + if err := r.Get(context.Background(), types.NamespacedName{Name: MockGateway}, gateway); err != nil { + t.Fatalf("failed to get gateway: %v", err) + } + gateway.Status.ActiveEndpoints = []*ravenv1beta1.Endpoint{ + { + NodeName: Node2Name, + Type: ravenv1beta1.Proxy, + Port: ravenv1beta1.DefaultProxyServerExposedPort, + UnderNAT: false, + }, + } + if err := r.Update(context.Background(), gateway); err != nil { + t.Fatalf("failed to update gateway: %v", err) + } + + if _, err := r.Reconcile(context.Background(), reconcile.Request{NamespacedName: types.NamespacedName{Name: MockGateway}}); err != nil { + t.Fatalf("failed to reconcile service %s: %v", MockGateway, err) + } + + byNode = servicesByEndpointNode(t, r) + if len(byNode) != 1 { + t.Fatalf("expected exactly 1 service after failover, got %v", byNode) + } + if byNode[Node1Name] != "" { + t.Errorf("expected service for old active node %q to be deleted after failover", Node1Name) + } + name, ok := byNode[Node2Name] + if !ok { + t.Fatalf("expected service for new active node %q to be created, none found", Node2Name) } + assertEndpointsForService(t, r, name, Node2Address) } func TestClassifyService_PreservesImmutableFields(t *testing.T) { From f8bf3a70c63b8ceab253df926c8227925bb9d175 Mon Sep 17 00:00:00 2001 From: WorrierKhushal Date: Fri, 21 Aug 2026 04:27:17 +0530 Subject: [PATCH 2/2] test: add coverage for AlreadyExists reconcile-as-update, Create error, and Tunnel endpoint paths --- .../gateway_public_service_controller_test.go | 231 ++++++++++++++++++ 1 file changed, 231 insertions(+) diff --git a/pkg/yurtmanager/controller/raven/gatewaypublicservice/gateway_public_service_controller_test.go b/pkg/yurtmanager/controller/raven/gatewaypublicservice/gateway_public_service_controller_test.go index 9eaee3d8861..b6e5480cc5b 100644 --- a/pkg/yurtmanager/controller/raven/gatewaypublicservice/gateway_public_service_controller_test.go +++ b/pkg/yurtmanager/controller/raven/gatewaypublicservice/gateway_public_service_controller_test.go @@ -19,14 +19,18 @@ package gatewaypublicservice import ( "context" + "strings" "testing" corev1 "k8s.io/api/core/v1" + apierrs "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/schema" "k8s.io/apimachinery/pkg/types" clientgoscheme "k8s.io/client-go/kubernetes/scheme" "k8s.io/client-go/tools/record" + "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/client/fake" "sigs.k8s.io/controller-runtime/pkg/reconcile" @@ -407,3 +411,230 @@ func TestClassifyService_PreservesImmutableFields(t *testing.T) { t.Errorf("expected ExternalTrafficPolicy to be Local, got %s", svc.Spec.ExternalTrafficPolicy) } } + +type createHookClient struct { + client.Client + createHook func(ctx context.Context, base client.Client, obj client.Object, opts ...client.CreateOption) error +} + +func (c *createHookClient) Create(ctx context.Context, obj client.Object, opts ...client.CreateOption) error { + if c.createHook != nil { + return c.createHook(ctx, c.Client, obj, opts...) + } + return c.Client.Create(ctx, obj, opts...) +} + +func TestReconcileService_AlreadyExistsReconciledAsUpdate(t *testing.T) { + testCases := []struct { + name string + resource string + stalePort int32 + }{ + {name: "service create collides", resource: "services", stalePort: 9000}, + {name: "endpoints create collides", resource: "endpoints", stalePort: 9001}, + } + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + r := MockReconcile() + + gateway := &ravenv1beta1.Gateway{} + if err := r.Get(context.Background(), types.NamespacedName{Name: MockGateway}, gateway); err != nil { + t.Fatalf("failed to get gateway: %v", err) + } + gateway.Status.ActiveEndpoints = []*ravenv1beta1.Endpoint{ + { + NodeName: Node1Name, + Type: ravenv1beta1.Proxy, + Port: ravenv1beta1.DefaultProxyServerExposedPort, + UnderNAT: false, + }, + } + if err := r.Update(context.Background(), gateway); err != nil { + t.Fatalf("failed to update gateway: %v", err) + } + + base := r.Client + r.Client = &createHookClient{ + Client: base, + createHook: func(ctx context.Context, c client.Client, obj client.Object, opts ...client.CreateOption) error { + isTarget := false + switch tc.resource { + case "services": + _, isTarget = obj.(*corev1.Service) + case "endpoints": + _, isTarget = obj.(*corev1.Endpoints) + } + if !isTarget { + return c.Create(ctx, obj, opts...) + } + // simulate a stale object that already exists in the API server, + // then let Create collide with it. + stale := obj.DeepCopyObject().(client.Object) + switch tc.resource { + case "services": + stale.(*corev1.Service).Spec.Ports[0].Port = tc.stalePort + case "endpoints": + stale.(*corev1.Endpoints).Subsets[0].Ports[0].Port = tc.stalePort + } + if err := c.Create(ctx, stale); err != nil { + return err + } + // match the resourceVersion of the collided object so the + // follow-up update in the reconcile is accepted by the fake client + obj.SetResourceVersion(stale.GetResourceVersion()) + return apierrs.NewAlreadyExists(schema.GroupResource{Group: "", Resource: tc.resource}, obj.GetName()) + }, + } + + if _, err := r.Reconcile(context.Background(), reconcile.Request{NamespacedName: types.NamespacedName{Name: MockGateway}}); err != nil { + t.Fatalf("failed to reconcile service %s: %v", MockGateway, err) + } + + // the colliding object must be reconciled as an update, not duplicated + byNode := servicesByEndpointNode(t, r) + if len(byNode) != 1 || byNode[Node1Name] == "" { + t.Fatalf("expected exactly 1 service for active node %q, got %v", Node1Name, byNode) + } + svcName := byNode[Node1Name] + + svc := &corev1.Service{} + if err := r.Get(context.Background(), types.NamespacedName{Namespace: util.WorkingNamespace, Name: svcName}, svc); err != nil { + t.Fatalf("failed to get service %q: %v", svcName, err) + } + if len(svc.Spec.Ports) != 1 || svc.Spec.Ports[0].Port != ravenv1beta1.DefaultProxyServerExposedPort { + t.Errorf("expected service port to be reconciled to %d, got %v", ravenv1beta1.DefaultProxyServerExposedPort, svc.Spec.Ports) + } + + epsList := &corev1.EndpointsList{} + if err := r.List(context.Background(), epsList); err != nil { + t.Fatalf("failed to list endpoints: %v", err) + } + if len(epsList.Items) != 1 { + t.Fatalf("expected exactly 1 endpoints for active node %q, got %d", Node1Name, len(epsList.Items)) + } + eps := epsList.Items[0] + if eps.GetName() != svcName { + t.Errorf("expected endpoints %q to share the service name, got %q", svcName, eps.GetName()) + } + assertEndpointsForService(t, r, svcName, Node1Address) + if len(eps.Subsets) != 1 || len(eps.Subsets[0].Ports) != 1 || eps.Subsets[0].Ports[0].Port != ravenv1beta1.DefaultProxyServerExposedPort { + t.Errorf("expected endpoints to be reconciled to port %d, got %v", ravenv1beta1.DefaultProxyServerExposedPort, eps.Subsets) + } + }) + } +} + +func TestReconcileService_CreateErrorReturned(t *testing.T) { + testCases := []struct { + name string + resource string + wantErrSub string + }{ + {name: "service create forbidden", resource: "services", wantErrSub: "failed create service for gateway gw-mock"}, + {name: "endpoints create forbidden", resource: "endpoints", wantErrSub: "failed create endpoints for gateway gw-mock"}, + } + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + r := MockReconcile() + base := r.Client + r.Client = &createHookClient{ + Client: base, + createHook: func(ctx context.Context, c client.Client, obj client.Object, opts ...client.CreateOption) error { + isTarget := false + switch tc.resource { + case "services": + _, isTarget = obj.(*corev1.Service) + case "endpoints": + _, isTarget = obj.(*corev1.Endpoints) + } + if isTarget { + return apierrs.NewForbidden(schema.GroupResource{Group: "", Resource: tc.resource}, obj.GetName(), nil) + } + return c.Create(ctx, obj, opts...) + }, + } + + _, err := r.Reconcile(context.Background(), reconcile.Request{NamespacedName: types.NamespacedName{Name: MockGateway}}) + if err == nil { + t.Fatalf("expected reconcile to fail when %s create is forbidden, got nil error", tc.resource) + } + if !strings.Contains(err.Error(), tc.wantErrSub) { + t.Errorf("expected error to contain %q, got %v", tc.wantErrSub, err) + } + }) + } +} + +func TestReconcileService_TunnelActiveEndpoint(t *testing.T) { + r := MockReconcile() + + gateway := &ravenv1beta1.Gateway{} + if err := r.Get(context.Background(), types.NamespacedName{Name: MockGateway}, gateway); err != nil { + t.Fatalf("failed to get gateway: %v", err) + } + gateway.Status.ActiveEndpoints = []*ravenv1beta1.Endpoint{ + { + NodeName: Node1Name, + Type: ravenv1beta1.Proxy, + Port: ravenv1beta1.DefaultProxyServerExposedPort, + UnderNAT: false, + }, + { + NodeName: Node1Name, + Type: ravenv1beta1.Tunnel, + Port: ravenv1beta1.DefaultTunnelServerExposedPort, + UnderNAT: false, + }, + } + if err := r.Update(context.Background(), gateway); err != nil { + t.Fatalf("failed to update gateway: %v", err) + } + + if _, err := r.Reconcile(context.Background(), reconcile.Request{NamespacedName: types.NamespacedName{Name: MockGateway}}); err != nil { + t.Fatalf("failed to reconcile service %s: %v", MockGateway, err) + } + + svcList := &corev1.ServiceList{} + if err := r.List(context.Background(), svcList); err != nil { + t.Fatalf("failed to list services: %v", err) + } + proxySvcName, tunnelSvcName := "", "" + for _, svc := range svcList.Items { + switch svc.Labels[raven.LabelCurrentGatewayType] { + case ravenv1beta1.Proxy: + if svc.Labels[util.LabelCurrentGatewayEndpoints] == Node1Name { + proxySvcName = svc.Name + } + case ravenv1beta1.Tunnel: + if svc.Labels[util.LabelCurrentGatewayEndpoints] == Node1Name { + tunnelSvcName = svc.Name + } + } + } + if proxySvcName == "" { + t.Fatalf("expected a proxy service for node %q, none found", Node1Name) + } + if tunnelSvcName == "" { + t.Fatalf("expected a tunnel service for node %q, none found", Node1Name) + } + if !strings.HasPrefix(tunnelSvcName, util.GatewayTunnelServiceNamePrefix+"-"+MockGateway+"-"+Node1Name+"-") { + t.Errorf("expected tunnel service name to have prefix %q, got %q", util.GatewayTunnelServiceNamePrefix+"-"+MockGateway+"-"+Node1Name+"-", tunnelSvcName) + } + if !strings.HasPrefix(proxySvcName, util.GatewayProxyServiceNamePrefix+"-"+MockGateway+"-"+Node1Name+"-") { + t.Errorf("expected proxy service name to have prefix %q, got %q", util.GatewayProxyServiceNamePrefix+"-"+MockGateway+"-"+Node1Name+"-", proxySvcName) + } + + tunnelSvc := &corev1.Service{} + if err := r.Get(context.Background(), types.NamespacedName{Namespace: util.WorkingNamespace, Name: tunnelSvcName}, tunnelSvc); err != nil { + t.Fatalf("failed to get tunnel service %q: %v", tunnelSvcName, err) + } + if tunnelSvc.Spec.Type != corev1.ServiceTypeLoadBalancer { + t.Errorf("expected tunnel service to be of type LoadBalancer, got %s", tunnelSvc.Spec.Type) + } + if len(tunnelSvc.Spec.Ports) != 1 || tunnelSvc.Spec.Ports[0].Protocol != corev1.ProtocolUDP || tunnelSvc.Spec.Ports[0].Port != ravenv1beta1.DefaultTunnelServerExposedPort { + t.Errorf("expected tunnel service to expose UDP port %d, got %v", ravenv1beta1.DefaultTunnelServerExposedPort, tunnelSvc.Spec.Ports) + } + + assertEndpointsForService(t, r, tunnelSvcName, Node1Address) + assertEndpointsForService(t, r, proxySvcName, Node1Address) +}