fix #703: jumpIf not working with global filter

pkg/object/globalfilter/globalfilter.go

@@ -157,41 +157,20 @@ func (gf *GlobalFilter) Inherit(superSpec *supervisor.Spec, previousGeneration s
 
 // Handle `beforePipeline` and `afterPipeline` before and after the handler is executed.
 func (gf *GlobalFilter) Handle(ctx *context.Context, handler context.Handler) {
-	result := gf.beforeHandle(ctx)
-	if result == pipeline.BuiltInFilterEnd {
-		return
-	}
-	result = handler.Handle(ctx)
-	if result == pipeline.BuiltInFilterEnd {
-		return
-	}
-	gf.afterHandle(ctx)
-}
-
-// BeforeHandle before handler logic for beforePipeline spec.
-func (gf *GlobalFilter) beforeHandle(ctx *context.Context) string {
-	value := gf.beforePipeline.Load()
-	if value == nil {
-		return ""
-	}
-	handler, ok := value.(*pipeline.Pipeline)
+	p, ok := handler.(*pipeline.Pipeline)
 	if !ok {
-		return ""
+		panic("handler is not a pipeline")
 	}
-	return handler.Handle(ctx)
-}
 
-// AfterHandle after handler logic for afterPipeline spec.
-func (gf *GlobalFilter) afterHandle(ctx *context.Context) string {
-	value := gf.afterPipeline.Load()
-	if value == nil {
-		return ""
+	var before, after *pipeline.Pipeline
+	if v := gf.beforePipeline.Load(); v != nil {
+		before, _ = v.(*pipeline.Pipeline)
 	}
-	handler, ok := value.(*pipeline.Pipeline)
-	if !ok {
-		return ""
+	if v := gf.afterPipeline.Load(); v != nil {
+		after, _ = v.(*pipeline.Pipeline)
 	}
-	return handler.Handle(ctx)
+
+	p.HandleWithBeforeAfter(ctx, before, after)
 }
 
 // Close closes GlobalFilter itself.

pkg/object/pipeline/pipeline.go

@@ -300,20 +300,60 @@ func (p *Pipeline) getFilter(name string) filters.Filter {
 	return p.filters[name]
 }
 
+// HandleWithBeforeAfter handles the request, with additional flow defined by
+// the before/after pipeline.
+func (p *Pipeline) HandleWithBeforeAfter(ctx *context.Context, before, after *Pipeline) string {
+	result, sawEnd := "", false
+	flowLen := len(p.flow)
+	if before != nil {
+		flowLen += len(before.flow)
+	}
+	if after != nil {
+		flowLen += len(after.flow)
+	}
+	stats := make([]FilterStat, 0, flowLen)
+
+	if before != nil {
+		result, stats, sawEnd = p.doHandle(ctx, before.flow, stats)
+	}
+
+	if !sawEnd {
+		result, stats, sawEnd = p.doHandle(ctx, p.flow, stats)
+	}
+
+	if !sawEnd && after != nil {
+		result, stats, sawEnd = p.doHandle(ctx, after.flow, stats)
+	}
+
+	ctx.LazyAddTag(func() string {
+		return serializeStats(stats)
+	})
+	return result
+}
+
 // Handle is the handler to deal with the request.
 func (p *Pipeline) Handle(ctx *context.Context) string {
-	result, next := "", ""
 	stats := make([]FilterStat, 0, len(p.flow))
+	result, stats, _ := p.doHandle(ctx, p.flow, stats)
+	ctx.LazyAddTag(func() string {
+		return serializeStats(stats)
+	})
+	return result
+}
+
+func (p *Pipeline) doHandle(ctx *context.Context, flow []FlowNode, stats []FilterStat) (string, []FilterStat, bool) {
+	result, next, sawEnd := "", "", false
 
-	for i := range p.flow {
-		node := &p.flow[i]
+	for i := range flow {
+		node := &flow[i]
 		alias := node.filterAlias()
 
 		if next != "" && next != alias {
 			continue
 		}
 
 		if node.FilterName == BuiltInFilterEnd {
+			sawEnd = true
 			break
 		}
 
@@ -335,14 +375,12 @@ func (p *Pipeline) Handle(ctx *context.Context) string {
 
 		next = node.JumpIf[result]
 		if next == "" || next == BuiltInFilterEnd {
+			sawEnd = true
 			break
 		}
 	}
 
-	ctx.LazyAddTag(func() string {
-		return serializeStats(stats)
-	})
-	return result
+	return result, stats, sawEnd
 }
 
 // Status returns Status generated by Runtime.

pkg/object/pipeline/pipeline_test.go

@@ -401,3 +401,90 @@ filters:
 	assert.Equal(2, len(status.Filters))
 	assert.Empty(status.ToMetrics("123"), "no metrics")
 }
+
+func TestHandleWithBeforeAfter(t *testing.T) {
+	assert := assert.New(t)
+
+	stdReq, err := http.NewRequest(http.MethodGet, "http://localhost:9095", nil)
+	assert.Nil(err)
+	req, err := httpprot.NewRequest(stdReq)
+	assert.Nil(err)
+
+	filters.Register(MockFilterKind("Filter1", nil))
+	defer cleanup()
+
+	yamlSpec := `
+name: http-pipeline-test
+kind: Pipeline
+flow:
+  - filter: filter2
+filters:
+  - name: filter2 
+    kind: Filter1 
+`
+	spec, err := supervisor.NewSpec(yamlSpec)
+	assert.Nil(err)
+
+	pipeline := &Pipeline{}
+	pipeline.Init(spec, nil)
+	defer pipeline.Close()
+
+	ctx := context.New(tracing.NoopSpan)
+	ctx.SetRequest(context.DefaultNamespace, req)
+
+	pipeline.HandleWithBeforeAfter(ctx, nil, nil)
+	tags := ctx.Tags()
+	assert.NotContains(tags, "filter1")
+	assert.Contains(tags, "filter2")
+	assert.NotContains(tags, "filter3")
+
+	yamlSpec = `
+name: http-pipeline-after
+kind: Pipeline
+flow:
+  - filter: filter3
+filters:
+  - name: filter3 
+    kind: Filter1 
+`
+
+	spec, err = supervisor.NewSpec(yamlSpec)
+	assert.Nil(err)
+
+	after := &Pipeline{}
+	after.Init(spec, nil)
+	defer after.Close()
+
+	ctx = context.New(tracing.NoopSpan)
+	ctx.SetRequest(context.DefaultNamespace, req)
+	pipeline.HandleWithBeforeAfter(ctx, nil, after)
+	tags = ctx.Tags()
+	assert.NotContains(tags, "filter1")
+	assert.Contains(tags, "filter2")
+	assert.Contains(tags, "filter3")
+
+	yamlSpec = `
+name: http-pipeline-before
+kind: Pipeline
+flow:
+  - filter: filter1
+  - filter: END
+filters:
+  - name: filter1 
+    kind: Filter1 
+`
+	spec, err = supervisor.NewSpec(yamlSpec)
+	assert.Nil(err)
+
+	before := &Pipeline{}
+	before.Init(spec, nil)
+	defer before.Close()
+
+	ctx = context.New(tracing.NoopSpan)
+	ctx.SetRequest(context.DefaultNamespace, req)
+	pipeline.HandleWithBeforeAfter(ctx, before, after)
+	tags = ctx.Tags()
+	assert.Contains(tags, "filter1")
+	assert.NotContains(tags, "filter2")
+	assert.NotContains(tags, "filter3")
+}