feat(status): implement ingress v1 status syncing

mflendrich · mflendrich · commit 7ce67571e8be · 2020-09-11T03:01:05.000+02:00
This is merely a copy-paste of existing v1beta1 functionality;
resolution of TODOs is out of scope of this commit.
diff --git a/internal/ingress/status/status.go b/internal/ingress/status/status.go
@@ -312,37 +312,41 @@ func sliceToStatus(endpoints []string) []apiv1.LoadBalancerIngress {
 
 // updateStatus changes the status information of Ingress rules
 func (s *statusSync) updateStatus(ctx context.Context, newIngressPoint []apiv1.LoadBalancerIngress) {
-	ings := s.IngressLister.ListIngressesV1beta1()
-	tcpIngresses, err := s.IngressLister.ListTCPIngresses()
-	if err != nil {
-		s.Logger.Errorf("failed to list TPCIngresses: %v", err)
-	}
-	knativeIngresses, err := s.IngressLister.ListKnativeIngresses()
-	if err != nil {
-		s.Logger.Errorf("failed to list Knative Ingresses: %v", err)
-	}
-
 	p := pool.NewLimited(10)
 	defer p.Close()
 
 	batch := p.Batch()
 
-	for _, ing := range ings {
+	for _, ing := range s.IngressLister.ListIngressesV1beta1() {
 		batch.Queue(s.runUpdateIngressV1beta1(ctx, ing, newIngressPoint, s.CoreClient))
 	}
-	for _, ing := range tcpIngresses {
-		batch.Queue(s.runUpdateTCPIngress(ctx, ing, newIngressPoint, s.KongConfigClient))
+
+	for _, ing := range s.IngressLister.ListIngressesV1() {
+		batch.Queue(s.runUpdateIngressV1(ctx, ing, newIngressPoint, s.CoreClient))
+	}
+
+	if tcpIngresses, err := s.IngressLister.ListTCPIngresses(); err != nil {
+		s.Logger.Errorf("failed to list TPCIngresses: %v", err)
+	} else {
+		for _, ing := range tcpIngresses {
+			batch.Queue(s.runUpdateTCPIngress(ctx, ing, newIngressPoint, s.KongConfigClient))
+		}
 	}
-	for _, ing := range knativeIngresses {
-		batch.Queue(s.runUpdateKnativeIngress(ctx, ing, newIngressPoint, s.KnativeClient))
+
+	if knativeIngresses, err := s.IngressLister.ListKnativeIngresses(); err != nil {
+		s.Logger.Errorf("failed to list Knative Ingresses: %v", err)
+	} else {
+		for _, ing := range knativeIngresses {
+			batch.Queue(s.runUpdateKnativeIngress(ctx, ing, newIngressPoint, s.KnativeClient))
+		}
 	}
 
 	batch.QueueComplete()
 	batch.WaitAll()
 }
 
-func (s *statusSync) runUpdateIngressV1beta1(ctx context.Context, ing *networkingv1beta1.Ingress, status []apiv1.LoadBalancerIngress,
-	client clientset.Interface) pool.WorkFunc {
+func (s *statusSync) runUpdateIngressV1beta1(ctx context.Context, ing *networkingv1beta1.Ingress,
+	status []apiv1.LoadBalancerIngress, client clientset.Interface) pool.WorkFunc {
 	return func(wu pool.WorkUnit) (interface{}, error) {
 		if wu.IsCancelled() {
 			return nil, nil
@@ -352,7 +356,7 @@ func (s *statusSync) runUpdateIngressV1beta1(ctx context.Context, ing *networkin
 			"ingress_namespace": ing.Namespace,
 			"ingress_name":      ing.Name,
 		})
-		sort.SliceStable(status, lessLoadBalancerIngress(status))
+		sort.SliceStable(status, lessLoadBalancerIngress(status)) // BUG: data race - see issue #829
 
 		curIPs := ing.Status.LoadBalancer.Ingress
 		sort.SliceStable(curIPs, lessLoadBalancerIngress(curIPs))
@@ -363,6 +367,14 @@ func (s *statusSync) runUpdateIngressV1beta1(ctx context.Context, ing *networkin
 		}
 
 		switch s.IngressAPI {
+		case utils.NetworkingV1:
+			// I expect this case to never happen, because if s.IngressAPI == NetworkingV1, then I expect Store to have only
+			// v1 ingresses (and no v1beta1 ingresses). If Store happens to have a v1beta1 Ingress nonetheless, I'm choosing
+			// not to drop it, but to log a warning and talk networking.k8s.io/v1beta1 (as opposed to extensions/v1beta1)
+			// because a v1-supporting Kubernetes API is more likely to support the former than the latter.
+			logger.Warnf("statusSync got an unexpected v1beta1 Ingress when it expected v1")
+			fallthrough
+
 		case utils.NetworkingV1beta1:
 			ingClient := client.NetworkingV1beta1().Ingresses(ing.Namespace)
 
@@ -407,6 +419,49 @@ func (s *statusSync) runUpdateIngressV1beta1(ctx context.Context, ing *networkin
 	}
 }
 
+func (s *statusSync) runUpdateIngressV1(ctx context.Context, ing *networkingv1.Ingress,
+	status []apiv1.LoadBalancerIngress, client clientset.Interface) pool.WorkFunc {
+	return func(wu pool.WorkUnit) (interface{}, error) {
+		if wu.IsCancelled() {
+			return nil, nil
+		}
+
+		logger := s.Logger.WithFields(logrus.Fields{
+			"ingress_namespace": ing.Namespace,
+			"ingress_name":      ing.Name,
+		})
+		sort.SliceStable(status, lessLoadBalancerIngress(status)) // BUG: data race - see issue #829
+
+		curIPs := ing.Status.LoadBalancer.Ingress
+		sort.SliceStable(curIPs, lessLoadBalancerIngress(curIPs))
+
+		if ingressSliceEqual(status, curIPs) {
+			logger.Debugf("no change in status, update skipped")
+			return true, nil
+		}
+
+		ingClient := client.NetworkingV1().Ingresses(ing.Namespace)
+
+		currIng, err := ingClient.Get(ctx, ing.Name, metav1.GetOptions{})
+		if err != nil {
+			return nil, fmt.Errorf("failed to fetch Ingress %v/%v: %w", ing.Namespace, ing.Name, err)
+		}
+
+		logger.WithField("ingress_status", status).Debugf("attempting to update ingress status")
+		currIng.Status.LoadBalancer.Ingress = status
+		_, err = ingClient.UpdateStatus(ctx, currIng, metav1.UpdateOptions{})
+		if err != nil {
+			// TODO return this error?
+			logger.Errorf("failed to update ingress status: %v", err)
+		} else {
+			logger.WithField("ingress_status", status).Debugf("successfully updated ingress status")
+		}
+
+		return true, nil
+
+	}
+}
+
 func toCoreLBStatus(knativeLBStatus *knative.LoadBalancerStatus) []apiv1.LoadBalancerIngress {
 	var res []apiv1.LoadBalancerIngress
 	if knativeLBStatus == nil {
@@ -447,7 +502,7 @@ func (s *statusSync) runUpdateKnativeIngress(ctx context.Context,
 			"ingress_namespace": ing.Namespace,
 			"ingress_name":      ing.Name,
 		})
-		sort.SliceStable(status, lessLoadBalancerIngress(status))
+		sort.SliceStable(status, lessLoadBalancerIngress(status)) // BUG: data race - see issue #829
 		curIPs := toCoreLBStatus(ing.Status.PublicLoadBalancer)
 		sort.SliceStable(curIPs, lessLoadBalancerIngress(curIPs))
 
@@ -507,7 +562,7 @@ func (s *statusSync) runUpdateTCPIngress(ctx context.Context,
 			"ingress_namespace": ing.Namespace,
 			"ingress_name":      ing.Name,
 		})
-		sort.SliceStable(status, lessLoadBalancerIngress(status))
+		sort.SliceStable(status, lessLoadBalancerIngress(status)) // BUG: data race - see issue #829
 
 		curIPs := ing.Status.LoadBalancer.Ingress
 		sort.SliceStable(curIPs, lessLoadBalancerIngress(curIPs))
diff --git a/internal/ingress/status/status_test.go b/internal/ingress/status/status_test.go
@@ -183,7 +183,7 @@ func fakeSynFn(interface{}) error {
 	return nil
 }
 
-func buildExtensionsIngresses() []networkingv1beta1.Ingress {
+func buildIngressesV1beta1() []networkingv1beta1.Ingress {
 	return []networkingv1beta1.Ingress{
 		{
 			ObjectMeta: metav1.ObjectMeta{
@@ -234,18 +234,66 @@ func buildExtensionsIngresses() []networkingv1beta1.Ingress {
 	}
 }
 
-type testIngressLister struct {
+func buildIngressesV1() []networkingv1.Ingress {
+	return []networkingv1.Ingress{
+		{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "foo_ingress_1",
+				Namespace: apiv1.NamespaceDefault,
+			},
+			Status: networkingv1.IngressStatus{
+				LoadBalancer: apiv1.LoadBalancerStatus{
+					Ingress: []apiv1.LoadBalancerIngress{
+						{
+							IP:       "10.0.0.1",
+							Hostname: "foo1",
+						},
+					},
+				},
+			},
+		},
+		{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "foo_ingress_different_class",
+				Namespace: metav1.NamespaceDefault,
+				Annotations: map[string]string{
+					"kubernetes.io/ingress.class": "no-nginx",
+				},
+			},
+			Status: networkingv1.IngressStatus{
+				LoadBalancer: apiv1.LoadBalancerStatus{
+					Ingress: []apiv1.LoadBalancerIngress{
+						{
+							IP:       "0.0.0.0",
+							Hostname: "foo.bar.com",
+						},
+					},
+				},
+			},
+		},
+		{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "foo_ingress_2",
+				Namespace: apiv1.NamespaceDefault,
+			},
+			Status: networkingv1.IngressStatus{
+				LoadBalancer: apiv1.LoadBalancerStatus{
+					Ingress: []apiv1.LoadBalancerIngress{},
+				},
+			},
+		},
+	}
 }
 
-func (til *testIngressLister) ListIngressesV1beta1() []*networkingv1beta1.Ingress {
-	var ingresses []*networkingv1beta1.Ingress
-	ingresses = append(ingresses, &networkingv1beta1.Ingress{
+var sampleIngressesV1beta1 = []*networkingv1beta1.Ingress{
+	{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      "foo_ingress_non_01",
 			Namespace: apiv1.NamespaceDefault,
-		}})
+		},
+	},
 
-	ingresses = append(ingresses, &networkingv1beta1.Ingress{
+	{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      "foo_ingress_1",
 			Namespace: apiv1.NamespaceDefault,
@@ -255,13 +303,41 @@ func (til *testIngressLister) ListIngressesV1beta1() []*networkingv1beta1.Ingres
 				Ingress: buildLoadBalancerIngressByIP(),
 			},
 		},
-	})
+	},
+}
+
+var sampleIngressesV1 = []*networkingv1.Ingress{
+	{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "foo_ingress_non_01",
+			Namespace: apiv1.NamespaceDefault,
+		},
+	},
+
+	{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "foo_ingress_1",
+			Namespace: apiv1.NamespaceDefault,
+		},
+		Status: networkingv1.IngressStatus{
+			LoadBalancer: apiv1.LoadBalancerStatus{
+				Ingress: buildLoadBalancerIngressByIP(),
+			},
+		},
+	},
+}
+
+type testIngressLister struct {
+	ingressesV1beta1 []*networkingv1beta1.Ingress
+	ingressesV1      []*networkingv1.Ingress
+}
 
-	return ingresses
+func (til *testIngressLister) ListIngressesV1beta1() []*networkingv1beta1.Ingress {
+	return til.ingressesV1beta1
 }
 
 func (til *testIngressLister) ListIngressesV1() []*networkingv1.Ingress {
-	return nil
+	return til.ingressesV1
 }
 
 func (til *testIngressLister) ListTCPIngresses() ([]*configurationv1beta1.TCPIngress, error) {
@@ -272,10 +348,6 @@ func (til *testIngressLister) ListKnativeIngresses() ([]*knative.Ingress, error)
 	return nil, nil
 }
 
-func buildIngressLister() ingressLister {
-	return &testIngressLister{}
-}
-
 func buildStatusSync() statusSync {
 	return statusSync{
 		pod: &utils.PodInfo{
@@ -287,9 +359,9 @@ func buildStatusSync() statusSync {
 		},
 		syncQueue: task.NewTaskQueue(fakeSynFn, logrus.New()),
 		Config: Config{
-			CoreClient:     buildSimpleClientSet(&networkingv1beta1.IngressList{Items: buildExtensionsIngresses()}),
+			CoreClient:     buildSimpleClientSet(&networkingv1beta1.IngressList{Items: buildIngressesV1beta1()}),
 			PublishService: apiv1.NamespaceDefault + "/" + "foo",
-			IngressLister:  buildIngressLister(),
+			IngressLister:  &testIngressLister{ingressesV1beta1: sampleIngressesV1beta1},
 			IngressAPI:     utils.ExtensionsV1beta1,
 		},
 	}
@@ -301,9 +373,9 @@ func TestStatusActionsV1beta1(t *testing.T) {
 	os.Setenv("POD_NAME", "foo1")
 	os.Setenv("POD_NAMESPACE", apiv1.NamespaceDefault)
 	c := Config{
-		CoreClient:             buildSimpleClientSet(&networkingv1beta1.IngressList{Items: buildExtensionsIngresses()}),
+		CoreClient:             buildSimpleClientSet(&networkingv1beta1.IngressList{Items: buildIngressesV1beta1()}),
 		PublishService:         apiv1.NamespaceDefault + "/" + "foo",
-		IngressLister:          buildIngressLister(),
+		IngressLister:          &testIngressLister{ingressesV1beta1: sampleIngressesV1beta1},
 		UpdateStatusOnShutdown: true,
 		IngressAPI:             utils.NetworkingV1beta1,
 		Logger:                 logrus.New(),
@@ -365,6 +437,76 @@ func TestStatusActionsV1beta1(t *testing.T) {
 	}
 }
 
+func TestStatusActionsV1(t *testing.T) {
+	ctx := context.Background()
+	// make sure election can be created
+	os.Setenv("POD_NAME", "foo1")
+	os.Setenv("POD_NAMESPACE", apiv1.NamespaceDefault)
+	c := Config{
+		CoreClient:             buildSimpleClientSet(&networkingv1.IngressList{Items: buildIngressesV1()}),
+		PublishService:         apiv1.NamespaceDefault + "/" + "foo",
+		IngressLister:          &testIngressLister{ingressesV1: sampleIngressesV1},
+		UpdateStatusOnShutdown: true,
+		IngressAPI:             utils.NetworkingV1,
+		Logger:                 logrus.New(),
+	}
+	// create object
+	fkSync, err := NewStatusSyncer(ctx, c)
+	if fkSync == nil {
+		t.Fatalf("expected a valid Sync")
+	}
+
+	fk := fkSync.(statusSync)
+
+	// start it and wait for the election and syn actions
+	go fk.Run()
+	//  wait for the election
+	time.Sleep(100 * time.Millisecond)
+	// execute sync
+	err = fk.sync("just-test")
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+
+	newIPs := []apiv1.LoadBalancerIngress{{
+		IP: "11.0.0.2",
+	}}
+	fooIngress1, err1 := fk.CoreClient.NetworkingV1().Ingresses(
+		apiv1.NamespaceDefault).Get(ctx, "foo_ingress_1", metav1.GetOptions{})
+	if err1 != nil {
+		t.Fatalf("unexpected error")
+	}
+	fooIngress1CurIPs := fooIngress1.Status.LoadBalancer.Ingress
+	if !ingressSliceEqual(fooIngress1CurIPs, newIPs) {
+		t.Fatalf("returned %v but expected %v", fooIngress1CurIPs, newIPs)
+	}
+
+	time.Sleep(1 * time.Second)
+
+	// execute shutdown
+	fk.Shutdown(true)
+	// ingress should be empty
+	newIPs2 := []apiv1.LoadBalancerIngress{}
+	fooIngress2, err2 := fk.CoreClient.NetworkingV1().Ingresses(
+		apiv1.NamespaceDefault).Get(ctx, "foo_ingress_1", metav1.GetOptions{})
+	if err2 != nil {
+		t.Fatalf("unexpected error")
+	}
+	fooIngress2CurIPs := fooIngress2.Status.LoadBalancer.Ingress
+	if !ingressSliceEqual(fooIngress2CurIPs, newIPs2) {
+		t.Fatalf("returned %v but expected %v", fooIngress2CurIPs, newIPs2)
+	}
+
+	oic, err := fk.CoreClient.NetworkingV1().Ingresses(
+		metav1.NamespaceDefault).Get(ctx, "foo_ingress_different_class", metav1.GetOptions{})
+	if err != nil {
+		t.Fatalf("unexpected error")
+	}
+	if oic.Status.LoadBalancer.Ingress[0].IP != "0.0.0.0" && oic.Status.LoadBalancer.Ingress[0].Hostname != "foo.bar.com" {
+		t.Fatalf("invalid ingress status for rule with different class")
+	}
+}
+
 func TestCallback(t *testing.T) {
 	buildStatusSync()
 }
