Feature #18 add x flow

filters/filters.go

@@ -3,6 +3,7 @@ package filters
 import (
 	// import filter packages here:
 
+	"errors"
 	"github.com/zalando/skipper/filters/healthcheck"
 	"github.com/zalando/skipper/filters/humanstxt"
 	"github.com/zalando/skipper/filters/pathrewrite"
@@ -14,6 +15,10 @@ import (
 	"github.com/zalando/skipper/skipper"
 )
 
+var (
+	ErrInvalidFilterParameters = errors.New("Invalid filter parameters")
+)
+
 // takes a registry object and registers the filter spec in the package
 func Register(registry skipper.FilterRegistry) {
 	registry.Add(

filters/flowid/filter.go

@@ -0,0 +1,64 @@
+package flowid
+
+import (
+	"github.com/zalando/skipper/filters"
+	"github.com/zalando/skipper/skipper"
+)
+
+const (
+	filterName       = "FlowId"
+	flowIdHeaderName = "X-Flow-Id"
+)
+
+type flowId struct {
+	id            string
+	reuseExisting bool
+	flowIdLength  uint8
+}
+
+func New(id string, allowOverride bool, len uint8) skipper.Filter {
+	return &flowId{id, allowOverride, len}
+}
+
+func (this *flowId) Id() string { return this.id }
+
+func (this *flowId) Name() string { return filterName }
+
+func (this *flowId) Request(fc skipper.FilterContext) {
+	r := fc.Request()
+	var flowId string
+
+	if this.reuseExisting {
+		flowId = r.Header.Get(flowIdHeaderName)
+		if isValid(flowId) {
+			return
+		}
+	}
+
+	flowId, err := newFlowId(this.flowIdLength)
+	if err == nil {
+		fc.Request().Header.Set(flowIdHeaderName, flowId)
+	}
+}
+
+func (this *flowId) Response(skipper.FilterContext) {}
+
+func (this *flowId) MakeFilter(id string, fc skipper.FilterConfig) (skipper.Filter, error) {
+	var reuseExisting bool
+	if len(fc) > 0 {
+		if r, ok := fc[0].(bool); ok {
+			reuseExisting = r
+		} else {
+			return nil, filters.ErrInvalidFilterParameters
+		}
+	}
+	var flowIdLength uint8 = defaultLen
+	if len(fc) > 1 {
+		if l, ok := fc[1].(float64); ok && l >= minLength && l <= maxLength {
+			flowIdLength = uint8(l)
+		} else {
+			return nil, filters.ErrInvalidFilterParameters
+		}
+	}
+	return New(id, reuseExisting, flowIdLength), nil
+}

filters/flowid/filter_test.go

@@ -0,0 +1,99 @@
+package flowid
+
+import (
+	"github.com/zalando/skipper/mock"
+	"github.com/zalando/skipper/skipper"
+	"net/http"
+	"testing"
+)
+
+const (
+	testFlowId    = "FLOW-ID-FOR-TESTING"
+	invalidFlowId = "[<>] (o) [<>]"
+)
+
+var (
+	testFlowIdSpec           = &flowId{}
+	filterConfigWithReuse    = skipper.FilterConfig{true}
+	filterConfigWithoutReuse = skipper.FilterConfig{false}
+)
+
+func TestNewFlowIdGeneration(t *testing.T) {
+	f, _ := testFlowIdSpec.MakeFilter(filterName, filterConfigWithReuse)
+	fc := buildfilterContext()
+	f.Request(fc)
+
+	flowId := fc.Request().Header.Get(flowIdHeaderName)
+	if !isValid(flowId) {
+		t.Errorf("'%s' is not a valid flow id", flowId)
+	}
+}
+
+func TestFlowIdReuseExisting(t *testing.T) {
+	f, _ := testFlowIdSpec.MakeFilter(filterName, filterConfigWithReuse)
+	fc := buildfilterContext(flowIdHeaderName, testFlowId)
+	f.Request(fc)
+
+	flowId := fc.Request().Header.Get(flowIdHeaderName)
+	if flowId != testFlowId {
+		t.Errorf("Got wrong flow id. Expected '%s' got '%s'", testFlowId, flowId)
+	}
+}
+
+func TestFlowIdIgnoreReuseExisting(t *testing.T) {
+	f, _ := testFlowIdSpec.MakeFilter(filterName, filterConfigWithoutReuse)
+	fc := buildfilterContext(flowIdHeaderName, testFlowId)
+	f.Request(fc)
+
+	flowId := fc.Request().Header.Get(flowIdHeaderName)
+	if flowId == testFlowId {
+		t.Errorf("Got wrong flow id. Expected a newly generated flowid but got the test flow id '%s'", flowId)
+	}
+}
+
+func TestFlowIdRejectInvalidReusedFlowId(t *testing.T) {
+	f, _ := testFlowIdSpec.MakeFilter(filterName, filterConfigWithReuse)
+	fc := buildfilterContext(flowIdHeaderName, invalidFlowId)
+	f.Request(fc)
+
+	flowId := fc.Request().Header.Get(flowIdHeaderName)
+	if flowId == invalidFlowId {
+		t.Errorf("Got wrong flow id. Expected a newly generated flowid but got the test flow id '%s'", flowId)
+	}
+}
+
+func TestFlowIdWithSpecificLen(t *testing.T) {
+	fc := skipper.FilterConfig{true, float64(42.0)}
+	f, _ := testFlowIdSpec.MakeFilter(filterName, fc)
+	fctx := buildfilterContext()
+	f.Request(fctx)
+
+	flowId := fctx.Request().Header.Get(flowIdHeaderName)
+
+	l := len(flowId)
+	if l != 42 {
+		t.Errorf("Wrong flowId len. Expected %d, got %d", 42, l)
+	}
+}
+
+func TestFlowIdWithInvalidParameters(t *testing.T) {
+	fc := skipper.FilterConfig{"wrong-parameter-type"}
+	_, err := testFlowIdSpec.MakeFilter(filterName, fc)
+	if err != ErrInvalidFilterParameters {
+		t.Errorf("Expected an invalid parameters error, got %s", err)
+	}
+
+	fc = skipper.FilterConfig{true, float64(minLength - 1)}
+	_, err = testFlowIdSpec.MakeFilter(filterName, fc)
+	if err != ErrInvalidFilterParameters {
+		t.Errorf("Expected an invalid parameters error, got %s", err)
+	}
+}
+
+func buildfilterContext(headers ...string) skipper.FilterContext {
+	r, _ := http.NewRequest("GET", "http://example.org", nil)
+	for i := 0; i < len(headers); i += 2 {
+		r.Header.Set(headers[i], headers[i+1])
+	}
+	return &mock.FilterContext{FRequest: r}
+}

filters/flowid/hashing.go

@@ -0,0 +1,37 @@
+package flowid
+
+import (
+	"crypto/rand"
+	"encoding/hex"
+	"errors"
+	"fmt"
+	"regexp"
+)
+
+const (
+	defaultLen = 16
+	maxLength  = 254
+	minLength  = 8
+)
+
+var (
+	ErrInvalidLen  = errors.New(fmt.Sprintf("Invalid length. len must be >= %d and < %d", minLength, maxLength))
+	flowIdRegex, _ = regexp.Compile(`^[\w+/=\-]+$`)
+)
+
+func newFlowId(len uint8) (string, error) {
+	if len < minLength || len > maxLength || len%2 != 0 {
+		return "", ErrInvalidLen
+	}
+
+	u := make([]byte, hex.DecodedLen(int(len)))
+	buf := make([]byte, len)
+
+	rand.Read(u)
+	hex.Encode(buf, u)
+	return string(buf), nil
+}
+
+func isValid(flowId string) bool {
+	return len(flowId) >= minLength && len(flowId) <= maxLength && flowIdRegex.MatchString(flowId)
+}

filters/flowid/hashing_test.go

@@ -0,0 +1,64 @@
+package flowid
+
+import "testing"
+
+func TestFlowIdInvalidLength(t *testing.T) {
+	_, err := newFlowId(0)
+	if err == nil {
+		t.Errorf("Request for an invalid flow id length (0) succeeded and it shouldn't")
+	}
+
+	_, err = newFlowId(15)
+	if err != ErrInvalidLen {
+		t.Errorf("Request for an invalid flow id length (odd number) succeeded and it shouldn't")
+	}
+}
+
+func TestFlowIdLength(t *testing.T) {
+	for expected := minLength; expected <= maxLength; expected += 2 {
+		flowId, err := newFlowId(uint8(expected))
+		if err != nil {
+			t.Errorf("Failed to generate flowId with len %d", expected)
+		}
+
+		l := len(flowId)
+		if l != expected {
+			t.Errorf("Got wrong flowId len. Requested %d, got %d (%s)", expected, l, flowId)
+		}
+	}
+}
+
+func BenchmarkFlowIdLen8(b *testing.B) {
+	testFlowIdWithLen(b.N, 8)
+}
+
+func BenchmarkFlowIdLen10(b *testing.B) {
+	testFlowIdWithLen(b.N, 10)
+}
+
+func BenchmarkFlowIdLen12(b *testing.B) {
+	testFlowIdWithLen(b.N, 12)
+}
+
+func BenchmarkFlowIdLen14(b *testing.B) {
+	testFlowIdWithLen(b.N, 14)
+}
+
+func BenchmarkFlowIdLen16(b *testing.B) {
+	testFlowIdWithLen(b.N, 16)
+}
+
+func BenchmarkFlowIdLen32(b *testing.B) {
+	testFlowIdWithLen(b.N, 32)
+}
+
+func BenchmarkFlowIdLen64(b *testing.B) {
+	testFlowIdWithLen(b.N, 64)
+}
+
+func testFlowIdWithLen(times int, len uint8) {
+	for i := 0; i < times; i++ {
+		newFlowId(len)
+	}
+
+}

filters/flowid/readme.md

@@ -0,0 +1,67 @@
+# Flow ID Filter
+
+Flow Ids let you correlate router logs for a given request against the upstream application logs for that same request.
+If your upstream application makes other requests to other services it can provide the same Flow Id value so that all
+of those logs can be correlated.
+
+## How it works
+
+Skipper generates a unique Flow ID for every HTTP request that it receives. The Flow ID is then passed to your 
+upstream application as an HTTP header called `X-Flow-Id`.
+
+The filter takes 2 optional parameters:
+
+1. Accept existing `X-Flow-Id` header
+2. Flow Id length
+
+The first parameter is a boolean parameter that, when set to true, will make the filter skip the generation of
+a new flow id. If the existing header value is not a valid flow id it is ignored and a new flow id is also generated.
+
+The second parameter is a number that defines the length of the generated flow ids. Valid options are any even number
+between 8 and 254.
+
+## Usage
+
+The filter can be used with many different combinations of parameters. It can also be used without any parameter, since 
+both are options.
+
+### Default parameters
+	FlowId()
+Without any parameters, the filter doesn't reuse existing `X-Flow-Id` headers and generates new ones with 16 bytes.
+
+### Reuse existing flow id
+	FlowId(true)
+With only the first parameter with the boolean value `true` the filter will accept existing `X-Flow-Id` headers, if 
+they're present in the request.
+
+### Generate bigger flow ids
+	FlowId(false, 64)
+This example doesn't accept existing `X-Flow-Id` headers and will always generate new flow ids with 64 bytes.
+
+
+## Some benchmarks
+
+To decide upon which hashing mechanism to use we tested some versions of UUID v1 - v4 and some other implementations.
+The results are as follow:
+
+	Benchmark_uuidv1-4	            5000000	       281 ns/op
+	Benchmark_uuidv2-4	            5000000	       284 ns/op
+	Benchmark_uuidv3-4	            2000000	       605 ns/op
+	Benchmark_uuidv4-4	            1000000	      1903 ns/op
+	BenchmarkRndAndSprintf-4  	    500000	      3312 ns/op
+	BenchmarkSha1-4 	            1000000	      2188 ns/op
+	BenchmarkMd5-4  	            1000000	      2076 ns/op
+	BenchmarkFnv-4  	            500000	      2223 ns/op
+
+The current implementation just gets len / 2 (hex.DecodedLen) bytes from the PRNG and hex encodes them. 
+Its performance is only dependent on the length of the generated FlowId, according to the following benchmarks:
+
+	BenchmarkFlowIdLen8-4 	 1000000	      1157 ns/op
+	BenchmarkFlowIdLen10-4	 1000000	      1162 ns/op
+	BenchmarkFlowIdLen12-4	 1000000	      1163 ns/op
+	BenchmarkFlowIdLen14-4	 1000000	      1171 ns/op
+	BenchmarkFlowIdLen16-4	 1000000	      1180 ns/op
+	BenchmarkFlowIdLen32-4	 1000000	      1957 ns/op
+	BenchmarkFlowIdLen64-4	  300000	      3520 ns/op
+
+As you can see, starting at len = 32 (16 random bytes) the performance starts dropping dramatically. 

proxy/proxy.go

@@ -272,4 +272,4 @@ func (p *proxy) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 func addBranding(rs *http.Response) {
 	rs.Header.Set("X-Powered-By", "Skipper")
 	rs.Header.Set("Server", "Skipper")
-}
+}