fix(revert): service not found error caused during configuration of externalNodes (#26)

Author: AlinsRan

pkg/kube/apisix/apis/config/v2/types.go

@@ -505,7 +505,8 @@ type ApisixUpstreamSpec struct {
 // ApisixUpstreamConfig contains rich features on APISIX Upstream, for instance
 // load balancer, health check, etc.
 type ApisixUpstreamConfig struct {
-	Granularity string `json:"granularity,omitempty" yaml:"scheme,omitempty"`
+	// TODO: Improve or abandon, the next step is to improve or abandon it
+	Granularity string `json:"granularity,omitempty" yaml:"granularity,omitempty"`
 	// LoadBalancer represents the load balancer configuration for Kubernetes Service.
 	// The default strategy is round robin.
 	// +optional

pkg/providers/apisix/apisix_upstream.go

@@ -16,7 +16,6 @@ package apisix
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"reflect"
 	"strconv"
@@ -192,13 +191,6 @@ func (c *apisixUpstreamController) sync(ctx context.Context, ev *types.Event) er
 		if au.Spec == nil {
 			return nil
 		}
-		svc, err := c.SvcLister.Services(namespace).Get(name)
-		if err != nil {
-			log.Errorf("failed to get service %s: %s", key, err)
-			errRecord = err
-			goto updateStatus
-		}
-
 		// We will prioritize ExternalNodes and Discovery.
 		if len(au.Spec.ExternalNodes) != 0 || au.Spec.Discovery != nil {
 			var newUps *apisixv1.Upstream
@@ -228,7 +220,7 @@ func (c *apisixUpstreamController) sync(ctx context.Context, ev *types.Event) er
 			}
 			// updateUpstream for real
 			upsName := apisixv1.ComposeExternalUpstreamName(au.Namespace, au.Name)
-			errRecord = c.updateUpstream(ctx, upsName, &au.Spec.ApisixUpstreamConfig, ev.Type.IsSyncEvent(), svc.Spec.ClusterIP)
+			errRecord = c.updateUpstream(ctx, upsName, &au.Spec.ApisixUpstreamConfig, ev.Type.IsSyncEvent())
 			if err == apisix.ErrNotFound {
 				errRecord = fmt.Errorf("%s", "upstream doesn't exist. It will be created after ApisixRoute is created referencing it.")
 			}
@@ -242,6 +234,14 @@ func (c *apisixUpstreamController) sync(ctx context.Context, ev *types.Event) er
 				portLevelSettings[port.Port] = port.ApisixUpstreamConfig
 			}
 		}
+
+		svc, err := c.SvcLister.Services(namespace).Get(name)
+		if err != nil {
+			log.Errorf("failed to get service %s: %s", key, err)
+			errRecord = err
+			goto updateStatus
+		}
+
 		var subsets []configv2.ApisixUpstreamSubset
 		subsets = append(subsets, configv2.ApisixUpstreamSubset{})
 		if len(au.Spec.Subsets) > 0 {
@@ -257,7 +257,7 @@ func (c *apisixUpstreamController) sync(ctx context.Context, ev *types.Event) er
 						cfg = au.Spec.ApisixUpstreamConfig
 					}
 				}
-				err := c.updateUpstream(ctx, apisixv1.ComposeUpstreamName(namespace, name, subset.Name, port.Port), &cfg, ev.Type.IsSyncEvent(), svc.Spec.ClusterIP)
+				err := c.updateUpstream(ctx, apisixv1.ComposeUpstreamName(namespace, name, subset.Name, port.Port), &cfg, ev.Type.IsSyncEvent())
 				if err != nil {
 					if err == apisix.ErrNotFound {
 						errRecord = fmt.Errorf("%s", "upstream doesn't exist. It will be created after ApisixRoute is created referencing it.")
@@ -325,7 +325,7 @@ func (c *apisixUpstreamController) updateStatus(obj kube.ApisixUpstream, statusE
 	}
 }
 
-func (c *apisixUpstreamController) updateUpstream(ctx context.Context, upsName string, cfg *configv2.ApisixUpstreamConfig, shouldCompare bool, svcClusterIP string) error {
+func (c *apisixUpstreamController) updateUpstream(ctx context.Context, upsName string, cfg *configv2.ApisixUpstreamConfig, shouldCompare bool) error {
 	// TODO: multi cluster
 	clusterName := c.Config.APISIX.DefaultClusterName
 
@@ -348,24 +348,8 @@ func (c *apisixUpstreamController) updateUpstream(ctx context.Context, upsName s
 	}
 
 	newUps.Metadata = ups.Metadata
+	newUps.Nodes = ups.Nodes
 
-	if cfg.Granularity == types.ResolveGranularity.Service {
-		if svcClusterIP == "" {
-			log.Errorw("ApisixRoute refers to a headless service but want to use the service level resolve granularity",
-				zap.String("ApisixUpstream name", upsName),
-			)
-			return errors.New("conflict headless service and backend resolve granularity")
-		}
-		for _, node := range ups.Nodes {
-			newUps.Nodes = append(newUps.Nodes, apisixv1.UpstreamNode{
-				Host:   svcClusterIP,
-				Port:   node.Port,
-				Weight: node.Weight,
-			})
-		}
-	} else {
-		newUps.Nodes = ups.Nodes
-	}
 	log.Debugw("updating upstream since ApisixUpstream changed",
 		zap.Any("upstream", newUps),
 		zap.String("ApisixUpstream name", upsName),

test/e2e/scaffold/k8s.go

@@ -783,6 +783,25 @@ func (s *Scaffold) GetKubernetesClient() *kubernetes.Clientset {
 	return client
 }
 
+func (s *Scaffold) GetEndpoints(endpointsName string) (*corev1.Endpoints, error) {
+	e := s.GetKubernetesClient()
+	return e.CoreV1().Endpoints(s.Namespace()).Get(context.Background(), endpointsName, metav1.GetOptions{})
+}
+
+func (s *Scaffold) GetEndpointIPs(endpointsName string) ([]string, error) {
+	endpoints, err := s.GetEndpoints(endpointsName)
+	if err != nil {
+		return nil, err
+	}
+	ips := make([]string, 0)
+	for _, subset := range endpoints.Subsets {
+		for _, addr := range subset.Addresses {
+			ips = append(ips, addr.IP)
+		}
+	}
+	return ips, nil
+}
+
 func (s *Scaffold) RunKubectlAndGetOutput(args ...string) (string, error) {
 	return k8s.RunKubectlAndGetOutputE(ginkgo.GinkgoT(), s.kubectlOptions, args...)
 }

test/e2e/suite-features/external-service.go

@@ -115,19 +115,19 @@ spec:
 `, name, nodeType, nodeName)
 		assert.Nil(ginkgo.GinkgoT(), s.CreateVersionedApisixResource(au))
 	}
-
-	PhaseCreateApisixUpstreamWithGranularity := func(s *scaffold.Scaffold, name string, nodeType v2.ApisixUpstreamExternalType, nodeName, granularity string) {
-		au := fmt.Sprintf(`
-apiVersion: apisix.apache.org/v2
-kind: ApisixUpstream
-metadata:
-  name: %s
-spec:
-  granularity: %s
-`, name, granularity)
-		assert.Nil(ginkgo.GinkgoT(), s.CreateVersionedApisixResource(au))
-	}
-
+	/*
+	   	PhaseCreateApisixUpstreamWithGranularity := func(s *scaffold.Scaffold, name string, nodeType v2.ApisixUpstreamExternalType, nodeName, granularity string) {
+	   		au := fmt.Sprintf(`
+	   apiVersion: apisix.apache.org/v2
+	   kind: ApisixUpstream
+	   metadata:
+	     name: %s
+	   spec:
+	     granularity: %s
+	   `, name, granularity)
+	   		assert.Nil(ginkgo.GinkgoT(), s.CreateVersionedApisixResource(au))
+	   	}
+	*/
 	PhaseValidateNoUpstreams := func(s *scaffold.Scaffold) {
 		ups, err := s.ListApisixUpstreams()
 		assert.Nil(ginkgo.GinkgoT(), err)
@@ -475,19 +475,19 @@ spec:
 			PhaseCreateApisixUpstream(s, "httpbin-upstream", v2.ExternalTypeDomain, "postman-echo.com")
 			PhaseCreateApisixRouteWithHostRewriteAndBackend(s, "httpbin-route", "httpbin-upstream", "postman-echo.com", "httpbin-temp", 80)
 			//configure the created upstream with granularity service
-			PhaseCreateApisixUpstreamWithGranularity(s, "httpbin-temp", v2.ExternalTypeService, "httpbin-temp", "service")
+			//PhaseCreateApisixUpstreamWithGranularity(s, "httpbin-temp", v2.ExternalTypeService, "httpbin-temp", "service")
 			time.Sleep(time.Second * 6)
 
-			svc, err := s.GetServiceByName("httpbin-temp")
-			assert.Nil(ginkgo.GinkgoT(), err, "get httpbin service")
-			ip := svc.Spec.ClusterIP
+			ips, err := s.GetEndpointIPs("httpbin-temp")
+			assert.Nil(ginkgo.GinkgoT(), err, "get httpbin service endpoint")
+			assert.Len(ginkgo.GinkgoT(), ips, 1, "httpbin service endpoint count")
 
 			upName := apisixv1.ComposeUpstreamName(s.Namespace(), "httpbin-temp", "", 80)
 			upID := id.GenID(upName)
 
 			// -- validation --
 			PhaseValidateTrafficSplit(s, 2, upID, map[string]*validateFactor{
-				ip: {
+				ips[0]: {
 					port:   80,
 					weight: translation.DefaultWeight,
 				},

test/e2e/suite-plugins/suite-plugins-traffic/api_breaker.go

@@ -30,15 +30,6 @@ var _ = ginkgo.Describe("suite-plugins-traffic: api-breaker plugin", func() {
 		s := scaffoldFunc()
 		ginkgo.It("sanity", func() {
 			backendSvc, backendPorts := s.DefaultHTTPBackend()
-			au := fmt.Sprintf(`
-apiVersion: apisix.apache.org/v2
-kind: ApisixUpstream
-metadata:
-  name: %s
-spec:
-  granularity: service
-`, backendSvc)
-			assert.Nil(ginkgo.GinkgoT(), s.CreateVersionedApisixResource(au))
 			ar := fmt.Sprintf(`
 apiVersion: apisix.apache.org/v2
 kind: ApisixRoute
@@ -95,15 +86,6 @@ spec:
 		})
 		ginkgo.It("disable plugin", func() {
 			backendSvc, backendPorts := s.DefaultHTTPBackend()
-			au := fmt.Sprintf(`
-apiVersion: apisix.apache.org/v2
-kind: ApisixUpstream
-metadata:
-  name: %s
-spec:
-  granularity: service
-`, backendSvc)
-			assert.Nil(ginkgo.GinkgoT(), s.CreateVersionedApisixResource(au))
 			ar := fmt.Sprintf(`
 apiVersion: apisix.apache.org/v2
 kind: ApisixRoute