feature: filter fifoWithBody

filters/builtin/builtin.go

@@ -216,6 +216,7 @@ func Filters() []filters.Spec {
 		auth.NewForwardToken(),
 		auth.NewForwardTokenField(),
 		scheduler.NewFifo(),
+		scheduler.NewFifoWithBody(),
 		scheduler.NewLIFO(),
 		scheduler.NewLIFOGroup(),
 		rfc.NewPath(),

filters/filters.go

@@ -35,6 +35,9 @@ const (
 
 	// BackendRatelimit is the key used in the state bag to configure backend ratelimit in proxy
 	BackendRatelimit = "backend:ratelimit"
+
+	// FifoWithBody
+	FifoWithBody = "fifo:body:func"
 )
 
 // FilterContext object providing state and information that is unique to a request.
@@ -327,6 +330,7 @@ const (
 	SetDynamicBackendUrl                       = "setDynamicBackendUrl"
 	ApiUsageMonitoringName                     = "apiUsageMonitoring"
 	FifoName                                   = "fifo"
+	FifoWithBodyName                           = "fifoWithBody"
 	LifoName                                   = "lifo"
 	LifoGroupName                              = "lifoGroup"
 	RfcPathName                                = "rfcPath"

filters/scheduler/cleanup_test.go

@@ -0,0 +1,85 @@
+package scheduler
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/zalando/skipper/filters"
+	"github.com/zalando/skipper/proxy"
+	"github.com/zalando/skipper/routing"
+	"github.com/zalando/skipper/routing/testdataclient"
+	"github.com/zalando/skipper/scheduler"
+)
+
+func TestCleanupOnBackendErrors(t *testing.T) {
+	doc := `
+		aroute: *
+		-> lifo(1, 1, "100ms")
+		-> lifoGroup("foo", 1, 1, "100ms")
+		-> lifo(2, 2, "200ms")
+		-> lifoGroup("bar", 1, 1, "100ms")
+		-> fifo(1, 1, "200ms")
+		-> "http://test.invalid"
+	`
+
+	dc, err := testdataclient.NewDoc(doc)
+	require.NoError(t, err)
+	defer dc.Close()
+
+	reg := scheduler.RegistryWith(scheduler.Options{})
+	defer reg.Close()
+
+	fr := make(filters.Registry)
+	fr.Register(NewLIFO())
+	fr.Register(NewLIFOGroup())
+	fr.Register(NewFifo())
+
+	ro := routing.Options{
+		SignalFirstLoad: true,
+		FilterRegistry:  fr,
+		DataClients:     []routing.DataClient{dc},
+		PostProcessors:  []routing.PostProcessor{reg},
+	}
+
+	rt := routing.New(ro)
+	defer rt.Close()
+
+	<-rt.FirstLoad()
+
+	pr := proxy.WithParams(proxy.Params{
+		Routing: rt,
+	})
+	defer pr.Close()
+
+	ts := httptest.NewServer(pr)
+	defer ts.Close()
+
+	rsp, err := http.Get(ts.URL)
+	require.NoError(t, err)
+	rsp.Body.Close()
+
+	var route *routing.Route
+	{
+		req, err := http.NewRequest("GET", ts.URL, nil)
+		require.NoError(t, err)
+
+		route, _ = rt.Get().Do(req)
+		require.NotNil(t, route, "failed to lookup route")
+	}
+
+	for _, f := range route.Filters {
+		if qf, ok := f.Filter.(interface{ GetQueue() *scheduler.Queue }); ok {
+			status := qf.GetQueue().Status()
+			assert.Equal(t, scheduler.QueueStatus{ActiveRequests: 0, QueuedRequests: 0, Closed: false}, status)
+		} else if qf, ok := f.Filter.(interface{ GetQueue() *scheduler.FifoQueue }); ok {
+			status := qf.GetQueue().Status()
+			assert.Equal(t, scheduler.QueueStatus{ActiveRequests: 0, QueuedRequests: 0, Closed: false}, status)
+		} else {
+			t.Fatal("filter does not implement GetQueue()")
+		}
+	}
+}

filters/scheduler/fifo.go

@@ -15,27 +15,51 @@
 	fifoKey string = "fifo"
 )
 
+type fifoType int
+
+const (
+	fifo fifoType = iota + 1
+	fifoWithBody
+)
+
 type (
-	fifoSpec   struct{}
+	fifoSpec struct {
+		typ fifoType
+	}
 	fifoFilter struct {
 		config scheduler.Config
 		queue  *scheduler.FifoQueue
+		typ    fifoType
 	}
 )
 
 func NewFifo() filters.Spec {
-	return &fifoSpec{}
+	return &fifoSpec{
+		typ: fifo,
+	}
+}
+
+func NewFifoWithBody() filters.Spec {
+	return &fifoSpec{
+		typ: fifoWithBody,
+	}
 }
 
-func (*fifoSpec) Name() string {
-	return filters.FifoName
+func (f *fifoSpec) Name() string {
+	switch f.typ {
+	case fifo:
+		return filters.FifoName
+	case fifoWithBody:
+		return filters.FifoWithBodyName
+	}
+	return "unknown"
 }
 
 // CreateFilter creates a fifoFilter, that will use a semaphore based
 // queue for handling requests to limit concurrency of a route. The first
 // parameter is maxConcurrency the second maxQueueSize and the third
 // timeout.
 func (s *fifoSpec) CreateFilter(args []interface{}) (filters.Filter, error) {
 	if len(args) != 3 {
 		return nil, filters.ErrInvalidFilterParameters
 	}
@@ -65,6 +89,7 @@
 	}
 
 	return &fifoFilter{
+		typ: s.typ,
 		config: scheduler.Config{
 			MaxConcurrency: cc,
 			MaxQueueSize:   qs,
@@ -139,14 +164,32 @@
 // Response will decrease the number of inflight requests to release
 // the concurrency reservation for the request.
 func (f *fifoFilter) Response(ctx filters.FilterContext) {
-	pending, ok := ctx.StateBag()[fifoKey].([]func())
-	if !ok {
-		return
+	g := f.createResponse(ctx)
+	switch f.typ {
+	case fifo:
+		g()
+	case fifoWithBody:
+		ctx.StateBag()[filters.FifoWithBody] = g
 	}
-	last := len(pending) - 1
-	if last < 0 {
-		return
+}
+
+func (f *fifoFilter) createResponse(ctx filters.FilterContext) func() {
+	return func() {
+		pending, ok := ctx.StateBag()[fifoKey].([]func())
+		if !ok {
+			return
+		}
+		last := len(pending) - 1
+		if last < 0 {
+			return
+		}
+		pending[last]()
+		ctx.StateBag()[fifoKey] = pending[:last]
 	}
-	pending[last]()
-	ctx.StateBag()[fifoKey] = pending[:last]
+}
+
+// HandleErrorResponse is to opt-in for filters to get called
+// Response(ctx) in case of errors via proxy. It has to return true to opt-in.
+func (f *fifoFilter) HandleErrorResponse() bool {
+	return true
 }