k8s dataclient: run without polling data

.clusterfuzzlite/Dockerfile

@@ -1,4 +1,4 @@
-FROM gcr.io/oss-fuzz-base/base-builder-go@sha256:9e293085f163de4a9d9c02f0c17663b3d1e082db3355c142144bc2483d69b5ef
+FROM gcr.io/oss-fuzz-base/base-builder-go@sha256:9e76082f0d5e86f349c9f87f4e9bc994f905ba865160886ed121fe661e59d4aa
 
 COPY . $SRC/skipper
 COPY ./.clusterfuzzlite/build.sh $SRC/

.github/labeler.yml

@@ -0,0 +1,7 @@
+# Add 'enhancement' label to any PR where the head branch name starts with `feature/` in the name
+enhancement:
+ - head-branch: ['^feature/']
+
+# Add 'codefreeze' label to any PR that is opened against the `cw2025` branch
+codefreeze:
+ - base-branch: 'cw2025'

.github/workflows/gh-packages.yaml

@@ -47,7 +47,7 @@ jobs:
           make build.linux
 
       - name: Set up QEMU
-        uses: docker/setup-qemu-action@29109295f81e9208d7d86ff1c6c12d2833863392
+        uses: docker/setup-qemu-action@c7c53464625b32c7a7e944ae62b3e17d2b600130
 
       - name: Set up Docker Buildx
         uses: docker/setup-buildx-action@e468171a9de216ec08956ac3ada2f0791b6bd435
@@ -61,7 +61,7 @@ jobs:
           password: ${{ secrets.GITHUB_TOKEN }}
 
       - name: Docker meta
-        uses: docker/metadata-action@c1e51972afc2121e065aed6d45c65596fe445f3f
+        uses: docker/metadata-action@318604b99e75e41977312d83839a89be02ca4893
         id: meta
         with:
           images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}

config/config.go

@@ -195,6 +195,7 @@ type Config struct {
 	KubernetesBackendTrafficAlgorithm                    kubernetes.BackendTrafficAlgorithm `yaml:"-"`
 	KubernetesDefaultLoadBalancerAlgorithm               string                             `yaml:"kubernetes-default-lb-algorithm"`
 	KubernetesForceService                               bool                               `yaml:"kubernetes-force-service"`
+	KubernetesNoPoll                                     bool                               `yaml:"kubernetes-no-poll"`
 
 	// Default filters
 	DefaultFiltersDir string `yaml:"default-filters-dir"`
@@ -519,6 +520,7 @@ func NewConfig() *Config {
 	flag.StringVar(&cfg.KubernetesBackendTrafficAlgorithmString, "kubernetes-backend-traffic-algorithm", kubernetes.TrafficPredicateAlgorithm.String(), "sets the algorithm to be used for traffic splitting between backends: traffic-predicate or traffic-segment-predicate")
 	flag.StringVar(&cfg.KubernetesDefaultLoadBalancerAlgorithm, "kubernetes-default-lb-algorithm", kubernetes.DefaultLoadBalancerAlgorithm, "sets the default algorithm to be used for load balancing between backend endpoints, available options: roundRobin, consistentHash, random, powerOfRandomNChoices")
 	flag.BoolVar(&cfg.KubernetesForceService, "kubernetes-force-service", false, "overrides default Skipper functionality and routes traffic using Kubernetes Services instead of Endpoints")
+	flag.BoolVar(&cfg.KubernetesNoPoll, "kubernetes-no-poll", false, "force kubernetes dataclient to not to create/update routing table")
 
 	// Auth:
 	flag.BoolVar(&cfg.EnableOAuth2GrantFlow, "enable-oauth2-grant-flow", false, "enables OAuth2 Grant Flow filter")
@@ -935,6 +937,7 @@ func (c *Config) ToOptions() skipper.Options {
 		KubernetesBackendTrafficAlgorithm:              c.KubernetesBackendTrafficAlgorithm,
 		KubernetesDefaultLoadBalancerAlgorithm:         c.KubernetesDefaultLoadBalancerAlgorithm,
 		KubernetesForceService:                         c.KubernetesForceService,
+		KubernetesNoPoll:                               c.KubernetesNoPoll,
 
 		// API Monitoring:
 		ApiUsageMonitoringEnable:                c.ApiUsageMonitoringEnable,

config/config_test.go

@@ -114,6 +114,7 @@ func defaultConfig(with func(*Config)) *Config {
 		KubernetesRedisServicePort:              6379,
 		KubernetesBackendTrafficAlgorithmString: "traffic-predicate",
 		KubernetesDefaultLoadBalancerAlgorithm:  "roundRobin",
+		KubernetesNoPoll:                        false,
 		Oauth2TokeninfoTimeout:                  2 * time.Second,
 		Oauth2TokenintrospectionTimeout:         2 * time.Second,
 		Oauth2TokeninfoSubjectKey:               "uid",

dataclients/kubernetes/kube.go

@@ -257,6 +257,9 @@ type Options struct {
 
 	// ForwardBackendURL allows to use <forward> backend via kubernetes, for example routgroup backend `type: forward`.
 	ForwardBackendURL string
+
+	// KubernetesNoPoll force kubernetes dataclient to not create routing table
+	KubernetesNoPoll bool
 }
 
 // Client is a Skipper DataClient implementation used to create routes based on Kubernetes Ingress settings.
@@ -276,6 +279,7 @@ type Client struct {
 	state                  *clusterState
 	loggingInterval        time.Duration
 	loggingLastEnabled     time.Time
+	noPoll                 bool
 }
 
 // New creates and initializes a Kubernetes DataClient.
@@ -355,6 +359,7 @@ func New(o Options) (*Client, error) {
 		defaultFiltersDir:      o.DefaultFiltersDir,
 		forwardBackendURL:      o.ForwardBackendURL,
 		loggingInterval:        1 * time.Minute,
+		noPoll:                 o.KubernetesNoPoll,
 	}, nil
 }
 
@@ -420,6 +425,11 @@ func mapRoutes(routes []*eskip.Route) (map[string]*eskip.Route, []*eskip.Route)
 }
 
 func (c *Client) loadAndConvert() ([]*eskip.Route, error) {
+	if c.noPoll {
+		log.Debug("kubernetes dataclient is set to not poll")
+		return []*eskip.Route{}, nil
+	}
+
 	c.mu.Lock()
 	state, err := c.ClusterClient.fetchClusterState()
 	if err != nil {
@@ -503,7 +513,7 @@ func healthcheckRoutes(reverseSourcePredicate bool) []*eskip.Route {
 		if i > 0 {
 			cidrs.WriteString(", ")
 		}
-		cidrs.WriteString(fmt.Sprintf("%q", ip))
+		fmt.Fprintf(cidrs, "%q", ip)
 	}
 	params.SourceCIDRs = cidrs.String()
 
@@ -533,7 +543,7 @@ func (c *Client) LoadAll() ([]*eskip.Route, error) {
 //
 // TODO: implement a force reset after some time.
 func (c *Client) LoadUpdate() ([]*eskip.Route, []string, error) {
-	log.Debugf("polling for updates")
+	log.Debug("polling for updates")
 	r, err := c.loadAndConvert()
 	if err != nil {
 		log.Errorf("polling for updates failed: %v", err)
@@ -549,8 +559,7 @@ func (c *Client) LoadUpdate() ([]*eskip.Route, []string, error) {
 	)
 
 	for id := range c.current {
-		// TODO: use eskip.Eq()
-		if r, ok := next[id]; ok && r.String() != c.current[id].String() {
+		if r, ok := next[id]; ok && eskip.Eq(r, c.current[id]) {
 			updatedRoutes = append(updatedRoutes, r)
 		} else if !ok {
 			deletedIDs = append(deletedIDs, id)

dataclients/kubernetes/kube_test.go

@@ -2969,3 +2969,16 @@ func (mockSecretProvider) Add(string) error {
 }
 
 func (mockSecretProvider) Close() {}
+
+func TestNoPoll(t *testing.T) {
+	api := newTestAPI(t, nil, &definitions.IngressV1List{})
+	defer api.Close()
+
+	dc, err := New(Options{KubernetesURL: api.server.URL, KubernetesNoPoll: true})
+	require.NoError(t, err)
+
+	routes, err := dc.LoadAll()
+	require.NoError(t, err)
+
+	assert.Len(t, routes, 0)
+}

filters/builtin/builtin.go

@@ -8,6 +8,7 @@ import (
 	"github.com/zalando/skipper/filters/accesslog"
 	"github.com/zalando/skipper/filters/annotate"
 	"github.com/zalando/skipper/filters/auth"
+	"github.com/zalando/skipper/filters/awssigner/awssigv4"
 	"github.com/zalando/skipper/filters/circuit"
 	"github.com/zalando/skipper/filters/consistenthash"
 	"github.com/zalando/skipper/filters/cookie"
@@ -170,6 +171,7 @@ func Filters() []filters.Spec {
 		NewSetDynamicBackendScheme(),
 		NewSetDynamicBackendUrl(),
 		NewOriginMarkerSpec(),
+		awssigv4.New(),
 		diag.NewRandom(),
 		diag.NewRepeat(),
 		diag.NewRepeatHex(),
@@ -202,6 +204,7 @@ func Filters() []filters.Spec {
 		sed.NewDelimitedRequest(),
 		auth.NewBasicAuth(),
 		cookie.NewDropRequestCookie(),
+		cookie.NewDropResponseCookie(),
 		cookie.NewRequestCookie(),
 		cookie.NewResponseCookie(),
 		cookie.NewJSCookie(),

filters/shedder/admission.go

@@ -3,7 +3,7 @@ package shedder
 import (
 	"context"
 	"math"
-	"math/rand"
+	"math/rand/v2"
 	"net/http"
 	"sync"
 	"sync/atomic"
@@ -94,7 +94,8 @@ const (
 )
 
 type Options struct {
-	Tracer opentracing.Tracer
+	Tracer   opentracing.Tracer
+	testRand bool
 }
 
 type admissionControlPre struct{}
@@ -164,7 +165,8 @@ func (spec *admissionControlPost) Do(routes []*routing.Route) []*routing.Route {
 }
 
 type AdmissionControlSpec struct {
-	tracer opentracing.Tracer
+	tracer   opentracing.Tracer
+	testRand bool
 }
 
 type admissionControl struct {
@@ -190,6 +192,13 @@ type admissionControl struct {
 	success          []int64
 	counter          *atomic.Int64
 	successCounter   *atomic.Int64
+
+	muRand sync.Mutex
+	rand   func() float64
+}
+
+func randWithSeed() func() float64 {
+	return rand.New(rand.NewPCG(2, 3)).Float64
 }
 
 func NewAdmissionControl(o Options) filters.Spec {
@@ -198,7 +207,8 @@ func NewAdmissionControl(o Options) filters.Spec {
 		tracer = &opentracing.NoopTracer{}
 	}
 	return &AdmissionControlSpec{
-		tracer: tracer,
+		tracer:   tracer,
+		testRand: o.testRand,
 	}
 }
 
@@ -298,6 +308,11 @@ func (spec *AdmissionControlSpec) CreateFilter(args []interface{}) (filters.Filt
 
 	averageRpsFactor := float64(time.Second) / (float64(d) * float64(windowSize))
 
+	r := rand.Float64
+	if spec.testRand {
+		r = randWithSeed()
+	}
+
 	ac := &admissionControl{
 		once: sync.Once{},
 
@@ -319,6 +334,7 @@ func (spec *AdmissionControlSpec) CreateFilter(args []interface{}) (filters.Filt
 		success:          make([]int64, windowSize),
 		counter:          new(atomic.Int64),
 		successCounter:   new(atomic.Int64),
+		rand:             r,
 	}
 	go ac.tickWindows(d)
 	return ac, nil
@@ -406,8 +422,11 @@ func (ac *admissionControl) pReject() float64 {
 
 func (ac *admissionControl) shouldReject() bool {
 	p := ac.pReject() // [0, ac.maxRejectProbability] and -1 to disable
+	var r float64
+	ac.muRand.Lock()
 	/* #nosec */
-	r := rand.Float64() // [0,1)
+	r = ac.rand() // [0,1)
+	ac.muRand.Unlock()
 
 	if ac.mode == logInactive {
 		log.Infof("%s: p: %0.2f, r: %0.2f", filters.AdmissionControlName, p, r)

filters/shedder/admission_test.go

@@ -3,7 +3,6 @@ package shedder
 import (
 	"fmt"
 	"io"
-	"math/rand"
 	"net/http"
 	"net/http/httptest"
 	"net/url"
@@ -151,20 +150,19 @@ func TestAdmissionControl(t *testing.T) {
 		pExpectedAdmissionShedding: 0.0,
 	}} {
 		t.Run(ti.msg, func(t *testing.T) {
-			var mu sync.Mutex
+			randFunc := randWithSeed()
 			backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-				mu.Lock()
-				p := rand.Float64()
-				mu.Unlock()
-				if p < ti.pBackendErr {
+				if randFunc() < ti.pBackendErr {
 					w.WriteHeader(http.StatusInternalServerError)
 				} else {
 					w.WriteHeader(http.StatusOK)
 				}
 			}))
 			defer backend.Close()
 
-			spec := NewAdmissionControl(Options{}).(*AdmissionControlSpec)
+			spec := NewAdmissionControl(Options{
+				testRand: true,
+			}).(*AdmissionControlSpec)
 
 			args := make([]interface{}, 0, 6)
 			args = append(args, "testmetric", ti.mode, ti.d.String(), ti.windowsize, ti.minRequests, ti.successThreshold, ti.maxrejectprobability, ti.exponent)
@@ -262,7 +260,9 @@ func TestAdmissionControlChainOnlyBackendErrorPass(t *testing.T) {
 	}))
 	defer backend.Close()
 
-	spec := NewAdmissionControl(Options{}).(*AdmissionControlSpec)
+	spec := NewAdmissionControl(Options{
+		testRand: true,
+	}).(*AdmissionControlSpec)
 
 	argsLeaf := make([]interface{}, 0, 6)
 	argsLeaf = append(argsLeaf, "testmetric-leaf", "active", "5ms", 5, 5, 0.9, 0.95, 1.0)
@@ -393,7 +393,9 @@ func TestAdmissionControlCleanupModes(t *testing.T) {
 			}))
 			defer backend2.Close()
 
-			fspec := NewAdmissionControl(Options{})
+			fspec := NewAdmissionControl(Options{
+				testRand: true,
+			})
 			spec, ok := fspec.(*AdmissionControlSpec)
 			if !ok {
 				t.Fatal("FilterSpec is not a AdmissionControlSpec")
@@ -564,7 +566,9 @@ func TestAdmissionControlCleanupMultipleFilters(t *testing.T) {
 			}))
 			defer backend.Close()
 
-			fspec := NewAdmissionControl(Options{})
+			fspec := NewAdmissionControl(Options{
+				testRand: true,
+			})
 			spec, ok := fspec.(*AdmissionControlSpec)
 			if !ok {
 				t.Fatal("FilterSpec is not a AdmissionControlSpec")
@@ -605,7 +609,7 @@ func TestAdmissionControlSetsSpansTags(t *testing.T) {
 	}))
 	defer backend.Close()
 
-	fspec := NewAdmissionControl(Options{Tracer: tracer})
+	fspec := NewAdmissionControl(Options{Tracer: tracer, testRand: true})
 	spec, ok := fspec.(*AdmissionControlSpec)
 	if !ok {
 		t.Fatal("FilterSpec is not a AdmissionControlSpec")

filters/tee/teeloopback_test.go

@@ -7,6 +7,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/assert"
 	"github.com/zalando/skipper/eskip"
 	"github.com/zalando/skipper/filters"
 	"github.com/zalando/skipper/predicates/source"
@@ -59,10 +60,12 @@ func TestLoopbackAndMatchPredicate(t *testing.T) {
 	if err != nil {
 		t.Error("teeloopback: failed to execute the request.", err)
 	}
+	assert.Eventually(t, func() bool {
+		return matchRequestsCount(split, 1) && matchRequestsCount(shadow, 1)
+	}, 100*time.Millisecond, 10*time.Millisecond,
+		"teeloopback: expected to receive 1 requests in split and shadow backend but got Split: %d, Shadow: %d",
+		len(split.GetRequests()), len(shadow.GetRequests()))
 	waitForAll(split, original, shadow)
-	if !matchRequestsCount(shadow, 1) || !matchRequestsCount(split, 1) {
-		t.Errorf("teeloopback: expected to receive 1 requests in split and shadow backend but got Split: %d, Shadow: %d", len(split.GetRequests()), len(shadow.GetRequests()))
-	}
 	if !matchRequestsCount(original, 0) {
 		t.Errorf("teeloopback: backend of original route should not receive requests but got %d", len(original.GetRequests()))
 	}

fuzz/Dockerfile

@@ -1,4 +1,4 @@
-FROM amazonlinux:2023@sha256:8c22ec81612b78284b5db864e0fad3c8df7688a8fd5baa861d69fe3baf2c7377
+FROM amazonlinux:2023@sha256:5ea333708360add6cc16ecec2569b8b75b6ee862528217bac65ad80752f4129b
 
 WORKDIR /workspace