Skip to content

Commit

Permalink
bugfix: even no endpoints left after filter, an empty object should b…
Browse files Browse the repository at this point in the history 
…e returned to clients
  • Loading branch information
@rambohe-ch
rambohe-ch committed Oct 11, 2022
1 parent cd00f26 commit 1a9753f
Show file tree
Hide file tree
Showing 2 changed files with 82 additions and 91 deletions.
99 changes: 29 additions & 70 deletions pkg/yurthub/filter/servicetopology/handler.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -81,33 +81,24 @@ func (fh *serviceTopologyFilterHandler) ObjectResponseFilter(b []byte) ([]byte,
// filter endpointSlice before k8s 1.21
var items []discoveryV1beta1.EndpointSlice
for i := range v.Items {
isNil, item := fh.serviceTopologyHandler(&v.Items[i])
if !isNil {
eps := item.(*discoveryV1beta1.EndpointSlice)
items = append(items, *eps)
}
eps := fh.serviceTopologyHandler(&v.Items[i]).(*discoveryV1beta1.EndpointSlice)
items = append(items, *eps)
}
v.Items = items
return fh.serializer.Encode(v)
case *discovery.EndpointSliceList:
var items []discovery.EndpointSlice
for i := range v.Items {
isNil, item := fh.serviceTopologyHandler(&v.Items[i])
if !isNil {
eps := item.(*discovery.EndpointSlice)
items = append(items, *eps)
}
eps := fh.serviceTopologyHandler(&v.Items[i]).(*discovery.EndpointSlice)
items = append(items, *eps)
}
v.Items = items
return fh.serializer.Encode(v)
case *v1.EndpointsList:
var items []v1.Endpoints
for i := range v.Items {
isNil, item := fh.serviceTopologyHandler(&v.Items[i])
if !isNil {
ep := item.(*v1.Endpoints)
items = append(items, *ep)
}
ep := fh.serviceTopologyHandler(&v.Items[i]).(*v1.Endpoints)
items = append(items, *ep)
}
v.Items = items
return fh.serializer.Encode(v)
Expand All @@ -134,20 +125,17 @@ func (fh *serviceTopologyFilterHandler) StreamResponseFilter(rc io.ReadCloser, c
return err
}

isNil, filteredObj := fh.serviceTopologyHandler(obj)
if !isNil {
ch <- watch.Event{
Type: watchType,
Object: filteredObj,
}
ch <- watch.Event{
Type: watchType,
Object: fh.serviceTopologyHandler(obj),
}
}
}

func (fh *serviceTopologyFilterHandler) serviceTopologyHandler(obj runtime.Object) (bool, runtime.Object) {
func (fh *serviceTopologyFilterHandler) serviceTopologyHandler(obj runtime.Object) runtime.Object {
needHandle, serviceTopologyType := fh.resolveServiceTopologyType(obj)
if !needHandle || len(serviceTopologyType) == 0 {
return false, obj
return obj
}

switch serviceTopologyType {
Expand All @@ -158,7 +146,7 @@ func (fh *serviceTopologyFilterHandler) serviceTopologyHandler(obj runtime.Objec
// close traffic on the same node pool
return fh.nodePoolTopologyHandler(obj)
default:
return false, obj
return obj
}
}

Expand Down Expand Up @@ -190,36 +178,24 @@ func (fh *serviceTopologyFilterHandler) resolveServiceTopologyType(obj runtime.O
return false, ""
}

func (fh *serviceTopologyFilterHandler) nodeTopologyHandler(obj runtime.Object) (bool, runtime.Object) {
func (fh *serviceTopologyFilterHandler) nodeTopologyHandler(obj runtime.Object) runtime.Object {
switch v := obj.(type) {
case *discoveryV1beta1.EndpointSlice:
newObj := reassembleV1beta1EndpointSlice(v, fh.nodeName, nil)
if newObj == nil {
return true, obj
}
return false, newObj
return reassembleV1beta1EndpointSlice(v, fh.nodeName, nil)
case *discovery.EndpointSlice:
newObj := reassembleEndpointSlice(v, fh.nodeName, nil)
if newObj == nil {
return true, obj
}
return false, newObj
return reassembleEndpointSlice(v, fh.nodeName, nil)
case *v1.Endpoints:
newObj := reassembleEndpoints(v, fh.nodeName, nil)
if newObj == nil {
return true, obj
}
return false, newObj
return reassembleEndpoints(v, fh.nodeName, nil)
default:
return false, obj
return obj
}
}

func (fh *serviceTopologyFilterHandler) nodePoolTopologyHandler(obj runtime.Object) (bool, runtime.Object) {
func (fh *serviceTopologyFilterHandler) nodePoolTopologyHandler(obj runtime.Object) runtime.Object {
currentNode, err := fh.nodeGetter(fh.nodeName)
if err != nil {
klog.Warningf("skip serviceTopologyFilterHandler, failed to get current node %s, err: %v", fh.nodeName, err)
return false, obj
return obj
}

nodePoolName, ok := currentNode.Labels[nodepoolv1alpha1.LabelCurrentNodePool]
Expand All @@ -231,30 +207,18 @@ func (fh *serviceTopologyFilterHandler) nodePoolTopologyHandler(obj runtime.Obje
nodePool, err := fh.nodePoolLister.Get(nodePoolName)
if err != nil {
klog.Warningf("serviceTopologyFilterHandler: failed to get nodepool %s, err: %v", nodePoolName, err)
return false, obj
return obj
}

switch v := obj.(type) {
case *discoveryV1beta1.EndpointSlice:
newObj := reassembleV1beta1EndpointSlice(v, "", nodePool)
if newObj == nil {
return true, obj
}
return false, newObj
return reassembleV1beta1EndpointSlice(v, "", nodePool)
case *discovery.EndpointSlice:
newObj := reassembleEndpointSlice(v, "", nodePool)
if newObj == nil {
return true, obj
}
return false, newObj
return reassembleEndpointSlice(v, "", nodePool)
case *v1.Endpoints:
newObj := reassembleEndpoints(v, "", nodePool)
if newObj == nil {
return true, obj
}
return false, newObj
return reassembleEndpoints(v, "", nodePool)
default:
return false, obj
return obj
}
}

Expand All @@ -279,9 +243,8 @@ func reassembleV1beta1EndpointSlice(endpointSlice *discoveryV1beta1.EndpointSlic
}
}
}
if len(newEps) == 0 {
return nil
}

// even no endpoints left, empty endpoints slice should be returned
endpointSlice.Endpoints = newEps
return endpointSlice
}
Expand All @@ -308,9 +271,7 @@ func reassembleEndpointSlice(endpointSlice *discovery.EndpointSlice, nodeName st
}
}

if len(newEps) == 0 {
return nil
}
// even no endpoints left, empty endpoints slice should be returned
endpointSlice.Endpoints = newEps
return endpointSlice
}
Expand Down Expand Up @@ -338,10 +299,8 @@ func reassembleEndpoints(endpoints *v1.Endpoints, nodeName string, nodePool *nod
newEpSubsets = append(newEpSubsets, endpoints.Subsets[i])
}
}
if len(newEpSubsets) == 0 {
// this endpoints has no valid addresses for ingress controller, return nil to ignore it
return nil
}

// even no subsets left, empty subset slice should be returned
endpoints.Subsets = newEpSubsets
return endpoints
}
Expand Down
74 changes: 53 additions & 21 deletions pkg/yurthub/filter/servicetopology/handler_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -842,7 +842,7 @@ func TestServiceTopologyHandler(t *testing.T) {
},
Status: nodepoolv1alpha1.NodePoolStatus{
Nodes: []string{
"node2",
currentNodeName,
"node3",
},
},
Expand All @@ -856,12 +856,20 @@ func TestServiceTopologyHandler(t *testing.T) {
},
Status: nodepoolv1alpha1.NodePoolStatus{
Nodes: []string{
currentNodeName,
"node2",
},
},
},
),
expectResult: nil,
expectResult: &discoveryV1beta1.EndpointSlice{
ObjectMeta: metav1.ObjectMeta{
Name: "svc1-np7sf",
Namespace: "default",
Labels: map[string]string{
discoveryV1beta1.LabelServiceName: "svc1",
},
},
},
},
"v1beta1.EndpointSlice: currentNode has no endpoints in nodepool": {
object: &discoveryV1beta1.EndpointSlice{
Expand Down Expand Up @@ -971,7 +979,15 @@ func TestServiceTopologyHandler(t *testing.T) {
},
},
),
expectResult: nil,
expectResult: &discoveryV1beta1.EndpointSlice{
ObjectMeta: metav1.ObjectMeta{
Name: "svc1-np7sf",
Namespace: "default",
Labels: map[string]string{
discoveryV1beta1.LabelServiceName: "svc1",
},
},
},
},
"v1.EndpointSlice: topologyKeys is kubernetes.io/hostname": {
object: &discovery.EndpointSlice{
Expand Down Expand Up @@ -1700,7 +1716,15 @@ func TestServiceTopologyHandler(t *testing.T) {
},
},
),
expectResult: nil,
expectResult: &discovery.EndpointSlice{
ObjectMeta: metav1.ObjectMeta{
Name: "svc1-np7sf",
Namespace: "default",
Labels: map[string]string{
discoveryV1beta1.LabelServiceName: "svc1",
},
},
},
},
"v1.EndpointSlice: currentNode has no endpoints in nodePool": {
object: &discovery.EndpointSlice{
Expand Down Expand Up @@ -1790,7 +1814,15 @@ func TestServiceTopologyHandler(t *testing.T) {
},
},
),
expectResult: nil,
expectResult: &discovery.EndpointSlice{
ObjectMeta: metav1.ObjectMeta{
Name: "svc1-np7sf",
Namespace: "default",
Labels: map[string]string{
discoveryV1beta1.LabelServiceName: "svc1",
},
},
},
},
"v1.Endpoints: topologyKeys is kubernetes.io/hostname": {
object: &corev1.Endpoints{
Expand Down Expand Up @@ -2461,7 +2493,12 @@ func TestServiceTopologyHandler(t *testing.T) {
},
},
),
expectResult: nil,
expectResult: &corev1.Endpoints{
ObjectMeta: metav1.ObjectMeta{
Name: "svc1",
Namespace: "default",
},
},
},
"v1.Endpoints: currentNode has no endpoints in nodepool": {
object: &corev1.Endpoints{
Expand Down Expand Up @@ -2548,7 +2585,12 @@ func TestServiceTopologyHandler(t *testing.T) {
},
},
),
expectResult: nil,
expectResult: &corev1.Endpoints{
ObjectMeta: metav1.ObjectMeta{
Name: "svc1",
Namespace: "default",
},
},
},
"v1.Endpoints: unknown openyurt.io/topologyKeys": {
object: &corev1.Endpoints{
Expand Down Expand Up @@ -2692,19 +2734,9 @@ func TestServiceTopologyHandler(t *testing.T) {
nodeGetter: nodeGetter,
}

isNil, handledObject := fh.serviceTopologyHandler(tt.object)
if tt.expectResult != nil {
if isNil {
t.Errorf("serviceTopologyHandler expect %v, but got nil", tt.expectResult)
}

if !reflect.DeepEqual(handledObject, tt.expectResult) {
t.Errorf("serviceTopologyHandler expect: \n%#+v\nbut got: \n%#+v\n", tt.expectResult, handledObject)
}
} else {
if !isNil {
t.Errorf("serviceTopologyHandler expect nil, but got %v", handledObject)
}
handledObject := fh.serviceTopologyHandler(tt.object)
if !reflect.DeepEqual(handledObject, tt.expectResult) {
t.Errorf("serviceTopologyHandler expect: \n%#+v\nbut got: \n%#+v\n", tt.expectResult, handledObject)
}
})
}
Expand Down

0 comments on commit 1a9753f

Please sign in to comment.