Start executing rollout steps in the ingress reconcile

pkg/reconciler/route/reconcile_resources.go

@@ -18,9 +18,11 @@ package route
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
 	"fmt"
 
+	"github.com/davecgh/go-spew/spew"
 	"go.uber.org/zap"
 	"golang.org/x/sync/errgroup"
 	corev1 "k8s.io/api/core/v1"
@@ -30,20 +32,30 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/sets"
 
+	"knative.dev/networking/pkg/apis/networking"
 	netv1alpha1 "knative.dev/networking/pkg/apis/networking/v1alpha1"
 	"knative.dev/pkg/apis/duck"
 	"knative.dev/pkg/controller"
+	"knative.dev/pkg/kmeta"
 	"knative.dev/pkg/logging"
 	v1 "knative.dev/serving/pkg/apis/serving/v1"
 	"knative.dev/serving/pkg/reconciler/route/config"
 	"knative.dev/serving/pkg/reconciler/route/resources"
 	"knative.dev/serving/pkg/reconciler/route/traffic"
 )
 
-func (c *Reconciler) reconcileIngress(ctx context.Context, r *v1.Route, desired *netv1alpha1.Ingress) (*netv1alpha1.Ingress, error) {
+func (c *Reconciler) reconcileIngress(ctx context.Context, r *v1.Route, desired *netv1alpha1.Ingress, tc *traffic.Config) (*netv1alpha1.Ingress, error) {
 	recorder := controller.GetEventRecorder(ctx)
 	ingress, err := c.ingressLister.Ingresses(desired.Namespace).Get(desired.Name)
+
+	// Get the current rollout state as described by the traffic.
+	curRO := tc.BuildRollout()
+
 	if apierrs.IsNotFound(err) {
+		// If there is no exisiting Ingress, then current rollout is _the_ rollout.
+		desired.Annotations = kmeta.UnionMaps(desired.Annotations, map[string]string{
+			networking.RolloutAnnotationKey: serializeRollout(ctx, curRO),
+		})
 		ingress, err = c.netclient.NetworkingV1alpha1().Ingresses(desired.Namespace).Create(ctx, desired, metav1.CreateOptions{})
 		if err != nil {
 			recorder.Eventf(r, corev1.EventTypeWarning, "CreationFailed", "Failed to create Ingress: %v", err)
@@ -54,25 +66,34 @@ func (c *Reconciler) reconcileIngress(ctx context.Context, r *v1.Route, desired
 		return ingress, nil
 	} else if err != nil {
 		return nil, err
-	} else if !equality.Semantic.DeepEqual(ingress.Spec, desired.Spec) ||
-		!equality.Semantic.DeepEqual(ingress.Annotations, desired.Annotations) ||
-		!equality.Semantic.DeepEqual(ingress.Labels, desired.Labels) {
-		// It is notable that one reason for differences here may be defaulting.
-		// When that is the case, the Update will end up being a nop because the
-		// webhook will bring them into alignment and no new reconciliation will occur.
-		// Also, compare annotation and label in case ingress.Class or parent route's labels
-		// is updated.
-
-		// Don't modify the informers copy
-		origin := ingress.DeepCopy()
-		origin.Spec = desired.Spec
-		origin.Annotations = desired.Annotations
-		origin.Labels = desired.Labels
-		updated, err := c.netclient.NetworkingV1alpha1().Ingresses(origin.Namespace).Update(ctx, origin, metav1.UpdateOptions{})
-		if err != nil {
-			return nil, fmt.Errorf("failed to update Ingress: %w", err)
+	} else {
+		// Ingress exists. We need to compute the rollout spec diff.
+		prevRO := deserializeRollout(ctx, ingress.Annotations[networking.RolloutAnnotationKey])
+		effectiveRO := curRO.Step(prevRO)
+		// Update the annotation.
+		desired.Annotations[networking.RolloutAnnotationKey] = serializeRollout(ctx, effectiveRO)
+		// TODO(vagababov): apply the Rollout to the ingress spec here.
+		if !equality.Semantic.DeepEqual(ingress.Spec, desired.Spec) ||
+			!equality.Semantic.DeepEqual(ingress.Annotations, desired.Annotations) ||
+			!equality.Semantic.DeepEqual(ingress.Labels, desired.Labels) {
+			// It is notable that one reason for differences here may be defaulting.
+			// When that is the case, the Update will end up being a nop because the
+			// webhook will bring them into alignment and no new reconciliation will occur.
+			// Also, compare annotation and label in case ingress.Class or parent route's labels
+			// is updated.
+
+			// Don't modify the informers copy
+			origin := ingress.DeepCopy()
+			origin.Spec = desired.Spec
+			origin.Annotations = desired.Annotations
+			origin.Labels = desired.Labels
+			updated, err := c.netclient.NetworkingV1alpha1().Ingresses(origin.Namespace).Update(
+				ctx, origin, metav1.UpdateOptions{})
+			if err != nil {
+				return nil, fmt.Errorf("failed to update Ingress: %w", err)
+			}
+			return updated, nil
 		}
-		return updated, nil
 	}
 
 	return ingress, err
@@ -227,3 +248,30 @@ func (c *Reconciler) reconcileTargetRevisions(ctx context.Context, t *traffic.Co
 	}
 	return eg.Wait()
 }
+
+func serializeRollout(ctx context.Context, r *traffic.Rollout) string {
+	sr, err := json.Marshal(r)
+	if err != nil {
+		// This must never happen in the normal course of things.
+		logging.FromContext(ctx).Warnw("Error serializing Rollout: "+spew.Sprint(r),
+			zap.Error(err))
+		return ""
+	}
+	return string(sr)
+}
+
+func deserializeRollout(ctx context.Context, ro string) *traffic.Rollout {
+	if ro == "" {
+		return nil
+	}
+	r := &traffic.Rollout{}
+	// Failure can happen if users manually tweaked the
+	// annotation or there's etcd corruption. Just log, rollouts
+	// are not mission critical.
+	if err := json.Unmarshal([]byte(ro), r); err != nil {
+		logging.FromContext(ctx).Warnw("Error deserializing Rollout: "+ro,
+			zap.Error(err))
+		return nil
+	}
+	return r
+}

pkg/reconciler/route/reconcile_resources_test.go

@@ -49,9 +49,9 @@ func TestReconcileIngressInsert(t *testing.T) {
 	defer cancel()
 
 	r := Route("test-ns", "test-route")
-	ci := newTestIngress(t, r)
+	ci, tc := newTestIngress(t, r)
 
-	if _, err := reconciler.reconcileIngress(ctx, r, ci); err != nil {
+	if _, err := reconciler.reconcileIngress(ctx, r, ci, tc); err != nil {
 		t.Error("Unexpected error:", err)
 	}
 }
@@ -65,15 +65,15 @@ func TestReconcileIngressUpdate(t *testing.T) {
 
 	r := Route("test-ns", "test-route")
 
-	ci := newTestIngress(t, r)
-	if _, err := reconciler.reconcileIngress(ctx, r, ci); err != nil {
+	ci, tc := newTestIngress(t, r)
+	if _, err := reconciler.reconcileIngress(ctx, r, ci, tc); err != nil {
 		t.Error("Unexpected error:", err)
 	}
 
 	updated := getRouteIngressFromClient(ctx, t, r)
 	fakeciinformer.Get(ctx).Informer().GetIndexer().Add(updated)
 
-	ci2 := newTestIngress(t, r, func(tc *traffic.Config) {
+	ci2, tc := newTestIngress(t, r, func(tc *traffic.Config) {
 		tc.Targets[traffic.DefaultTarget][0].TrafficTarget.Percent = ptr.Int64(50)
 		tc.Targets[traffic.DefaultTarget] = append(tc.Targets[traffic.DefaultTarget], traffic.RevisionTarget{
 			TrafficTarget: v1.TrafficTarget{
@@ -82,13 +82,13 @@ func TestReconcileIngressUpdate(t *testing.T) {
 			},
 		})
 	})
-	if _, err := reconciler.reconcileIngress(ctx, r, ci2); err != nil {
+	if _, err := reconciler.reconcileIngress(ctx, r, ci2, tc); err != nil {
 		t.Error("Unexpected error:", err)
 	}
 
 	updated = getRouteIngressFromClient(ctx, t, r)
 	if diff := cmp.Diff(ci2, updated); diff != "" {
-		t.Error("Unexpected diff (-want +got):", diff)
+		t.Errorf("Unexpected diff (-want +got):\n%s", diff)
 	}
 	if diff := cmp.Diff(ci, updated); diff == "" {
 		t.Error("Expected difference, but found none")
@@ -197,7 +197,7 @@ func getLastPinnedTimestamp(t *testing.T, rev *v1.Revision) (string, error) {
 	return lastPinnedTime, nil
 }
 
-func newTestIngress(t *testing.T, r *v1.Route, trafficOpts ...func(tc *traffic.Config)) *netv1alpha1.Ingress {
+func newTestIngress(t *testing.T, r *v1.Route, trafficOpts ...func(tc *traffic.Config)) (*netv1alpha1.Ingress, *traffic.Config) {
 	tc := &traffic.Config{Targets: map[string]traffic.RevisionTargets{
 		traffic.DefaultTarget: {{
 			TrafficTarget: v1.TrafficTarget{
@@ -219,7 +219,7 @@ func newTestIngress(t *testing.T, r *v1.Route, trafficOpts ...func(tc *traffic.C
 	if err != nil {
 		t.Error("Unexpected error:", err)
 	}
-	return ingress
+	return ingress, tc
 }
 
 func TestReconcileIngressClassAnnotation(t *testing.T) {
@@ -232,19 +232,19 @@ func TestReconcileIngressClassAnnotation(t *testing.T) {
 	const expClass = "foo.ingress.networking.knative.dev"
 
 	r := Route("test-ns", "test-route")
-	ci := newTestIngress(t, r)
-	if _, err := reconciler.reconcileIngress(ctx, r, ci); err != nil {
+	ci, tc := newTestIngress(t, r)
+	if _, err := reconciler.reconcileIngress(ctx, r, ci, tc); err != nil {
 		t.Error("Unexpected error:", err)
 	}
 
 	updated := getRouteIngressFromClient(ctx, t, r)
 	fakeciinformer.Get(ctx).Informer().GetIndexer().Add(updated)
 
-	ci2 := newTestIngress(t, r)
+	ci2, tc := newTestIngress(t, r)
 	// Add ingress.class annotation.
 	ci2.Annotations[networking.IngressClassAnnotationKey] = expClass
 
-	if _, err := reconciler.reconcileIngress(ctx, r, ci2); err != nil {
+	if _, err := reconciler.reconcileIngress(ctx, r, ci2, tc); err != nil {
 		t.Error("Unexpected error:", err)
 	}
 

pkg/reconciler/route/resources/ingress.go

@@ -18,11 +18,8 @@ package resources
 
 import (
 	"context"
-	"encoding/json"
 	"sort"
 
-	"github.com/davecgh/go-spew/spew"
-	"go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/intstr"
@@ -33,7 +30,6 @@ import (
 	netv1alpha1 "knative.dev/networking/pkg/apis/networking/v1alpha1"
 	ingress "knative.dev/networking/pkg/ingress"
 	"knative.dev/pkg/kmeta"
-	"knative.dev/pkg/logging"
 	"knative.dev/serving/pkg/activator"
 	apicfg "knative.dev/serving/pkg/apis/config"
 	"knative.dev/serving/pkg/apis/serving"
@@ -78,7 +74,6 @@ func MakeIngress(
 			}),
 			Annotations: kmeta.FilterMap(kmeta.UnionMaps(map[string]string{
 				networking.IngressClassAnnotationKey: ingressClass,
-				networking.RolloutAnnotationKey:      serializeRollout(ctx, tc.BuildRollout()),
 			}, r.GetAnnotations()), func(key string) bool {
 				return key == corev1.LastAppliedConfigAnnotation
 			}),
@@ -88,17 +83,6 @@ func MakeIngress(
 	}, nil
 }
 
-func serializeRollout(ctx context.Context, r *traffic.Rollout) string {
-	sr, err := json.Marshal(r)
-	if err != nil {
-		// This must not never happen in the normal course of things.
-		logging.FromContext(ctx).Warnw("Error serializing Rollout: "+spew.Sprint(r),
-			zap.Error(err))
-		return ""
-	}
-	return string(sr)
-}
-
 // makeIngressSpec builds a new IngressSpec from inputs.
 func makeIngressSpec(
 	ctx context.Context,

pkg/reconciler/route/resources/ingress_test.go

@@ -51,8 +51,6 @@ const (
 	testRouteName       = "test-route"
 	testAnnotationValue = "test-annotation-value"
 	testIngressClass    = "test-ingress"
-
-	emptyRollout = "{}"
 )
 
 func TestMakeIngressCorrectMetadata(t *testing.T) {
@@ -81,7 +79,6 @@ func TestMakeIngressCorrectMetadata(t *testing.T) {
 			// Make sure to get passdownIngressClass instead of ingressClass
 			networking.IngressClassAnnotationKey: passdownIngressClass,
 			"test-annotation":                    "bar",
-			networking.RolloutAnnotationKey:      emptyRollout,
 		},
 		OwnerReferences: []metav1.OwnerReference{*kmeta.NewControllerRef(r)},
 	}
@@ -133,7 +130,6 @@ func TestMakeIngressWithRollout(t *testing.T) {
 			// Make sure to get passdownIngressClass instead of ingressClass
 			networking.IngressClassAnnotationKey: passdownIngressClass,
 			"test-annotation":                    "bar",
-			networking.RolloutAnnotationKey:      serializeRollout(context.Background(), cfg.BuildRollout()),
 		},
 		OwnerReferences: []metav1.OwnerReference{*kmeta.NewControllerRef(r)},
 	}
@@ -922,7 +918,6 @@ func TestMakeIngressWithTLS(t *testing.T) {
 			Namespace: ns,
 			Annotations: map[string]string{
 				networking.IngressClassAnnotationKey: ingressClass,
-				networking.RolloutAnnotationKey:      emptyRollout,
 			},
 			Labels: map[string]string{
 				serving.RouteLabelKey:          "test-route",

pkg/reconciler/route/route.go

@@ -183,7 +183,7 @@ func (c *Reconciler) reconcileIngressResources(ctx context.Context, r *v1.Route,
 		return nil, err
 	}
 
-	ingress, err := c.reconcileIngress(ctx, r, desired)
+	ingress, err := c.reconcileIngress(ctx, r, desired, tc)
 	if err != nil {
 		return nil, err
 	}

pkg/reconciler/route/table_test.go

@@ -713,7 +713,11 @@ func TestReconcile(t *testing.T) {
 						}},
 					},
 				},
-			),
+				simpleRollout("config", []traffic.RevisionRollout{{
+					RevisionName: "config-00001", Percent: 99,
+				}, {
+					RevisionName: "config-00002", Percent: 1,
+				}})),
 		}},
 		WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
 			Object: Route("default", "new-latest-ready", WithConfigTarget("config"),
@@ -913,7 +917,11 @@ func TestReconcile(t *testing.T) {
 						}},
 					},
 				},
-			),
+				simpleRollout("config", []traffic.RevisionRollout{{
+					RevisionName: "config-00001", Percent: 99,
+				}, {
+					RevisionName: "config-00002", Percent: 1,
+				}})),
 		}},
 		WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
 			Object: Route("default", "update-ci-failure", WithConfigTarget("config"),
@@ -2555,6 +2563,12 @@ func ingressWithClass(r *v1.Route, tc *traffic.Config, class string, io ...Ingre
 func baseIngressWithClass(r *v1.Route, tc *traffic.Config, class string, io ...IngressOption) *netv1alpha1.Ingress {
 	ingress, _ := resources.MakeIngress(getContext(), r, tc, nil, class)
 
+	// By default attach current rollout.
+	ro := tc.BuildRollout()
+	ingress.Annotations = kmeta.UnionMaps(ingress.Annotations, map[string]string{
+		networking.RolloutAnnotationKey: serializeRollout(context.Background(), ro),
+	})
+
 	for _, opt := range io {
 		opt(ingress)
 	}
@@ -2569,6 +2583,12 @@ func ingressWithTLS(r *v1.Route, tc *traffic.Config, tls []netv1alpha1.IngressTL
 func baseIngressWithTLS(r *v1.Route, tc *traffic.Config, tls []netv1alpha1.IngressTLS, challenges []netv1alpha1.HTTP01Challenge, io ...IngressOption) *netv1alpha1.Ingress {
 	ingress, _ := resources.MakeIngress(getContext(), r, tc, tls, TestIngressClass, challenges...)
 
+	// By default attach current rollout.
+	ro := tc.BuildRollout()
+	ingress.Annotations = kmeta.UnionMaps(ingress.Annotations, map[string]string{
+		networking.RolloutAnnotationKey: serializeRollout(context.Background(), ro),
+	})
+
 	for _, opt := range io {
 		opt(ingress)
 	}
@@ -2724,3 +2744,17 @@ func setResponsiveGCFeature(ctx context.Context, flag cfgmap.Flag) context.Conte
 	c.Features.ResponsiveRevisionGC = flag
 	return cfgmap.ToContext(ctx, c)
 }
+
+func simpleRollout(cfg string, revs []traffic.RevisionRollout) IngressOption {
+	return func(i *netv1alpha1.Ingress) {
+		r := &traffic.Rollout{
+			Configurations: []traffic.ConfigurationRollout{{
+				ConfigurationName: cfg,
+				Percent:           100,
+				Revisions:         revs,
+			}},
+		}
+		ro := serializeRollout(context.Background(), r)
+		i.Annotations[networking.RolloutAnnotationKey] = ro
+	}
+}