support global filter fallthrough when meet error (#1233)

build/test/integration_test.go

@@ -1320,7 +1320,7 @@ func TestEgctlNamespace(t *testing.T) {
  mockNamespace := `
 name: mockNamespace
 kind: MockNamespacer
-namespace: mockNamespace 
+namespace: mockNamespace
 httpservers:
 - kind: HTTPServer
  name: mock-httpserver
@@ -1642,3 +1642,166 @@ filters:
  assert.Equal(tc.query, queryMap)
  }
 }
+
+func TestGlobalFilterFallthrough(t *testing.T) {
+ assert := assert.New(t)
+
+ gfYamlTmpl := `
+name: global-filter
+kind: GlobalFilter
+beforePipeline:
+ filters:
+ - name: validator
+ kind: Validator
+ headers:
+ Before-Pipeline:
+ values: ["valid"]
+afterPipeline:
+ filters:
+ - name: adaptor
+ kind: ResponseAdaptor
+ header:
+ add:
+ After-Pipeline: valid
+%s
+`
+ yaml := fmt.Sprintf(gfYamlTmpl, "")
+ err := createResource(yaml)
+ assert.Nil(err)
+ defer deleteResource("globalfilter", "global-filter")
+
+ yaml = `
+name: httpserver-gf
+kind: HTTPServer
+port: 10099
+globalFilter: global-filter
+rules:
+- paths:
+ - pathPrefix: /health
+ backend: pipeline-ok
+ - pathPrefix: /
+ backend: pipeline-gf
+`
+ err = createResource(yaml)
+ assert.Nil(err)
+ defer deleteResource("httpserver", "httpserver-gf")
+
+ makeReq := func() *http.Request {
+ req, err := http.NewRequest(http.MethodGet, "http:https://127.0.0.1:10099", nil)
+ assert.Nil(err)
+ return req
+ }
+
+ yaml = `
+name: pipeline-ok
+kind: Pipeline
+filters: 
+- name: responsebuilder
+ kind: ResponseBuilder
+ protocol: http
+ template: |
+ statusCode: 200
+`
+ err = createResource(yaml)
+ assert.Nil(err)
+ defer deleteResource("pipeline", "pipeline-ok")
+
+ // invalid url
+ yaml = `
+name: pipeline-gf
+kind: Pipeline
+filters: 
+- name: proxy
+ kind: Proxy
+ pools:
+ - servers:
+ - url: http:https://127.0.0.1:11111
+`
+ err = createResource(yaml)
+ assert.Nil(err)
+ defer deleteResource("pipeline", "pipeline-gf")
+
+ httpServerStarted := checkServerStart(func() *http.Request {
+ req, err := http.NewRequest(http.MethodGet, "http:https://127.0.0.1:10099/health", nil)
+ req.Header.Add("Before-Pipeline", "valid")
+ require.Nil(t, err)
+ return req
+ })
+ assert.True(httpServerStarted)
+ time.Sleep(3 * time.Second)
+
+ // not pass before, not exec pipeline and after
+ // bad request from before
+ // not header of after
+ req := makeReq()
+ resp, err := http.DefaultClient.Do(req)
+ resp.Body.Close()
+ assert.Nil(err)
+ assert.Equal(http.StatusBadRequest, resp.StatusCode, resp)
+ assert.Empty(resp.Header.Get("After-Pipeline"), resp)
+
+ // pass before, exec pipeline, meet error, not exec after
+ // status code of 503, error in filter
+ req = makeReq()
+ req.Header.Add("Before-Pipeline", "valid")
+ resp, err = http.DefaultClient.Do(req)
+ resp.Body.Close()
+ assert.Nil(err)
+ assert.Equal(http.StatusServiceUnavailable, resp.StatusCode, resp)
+ assert.Empty(resp.Header.Get("After-Pipeline"), resp)
+
+ // pass before, exec pipeline, exec after
+ req = makeReq()
+ req.URL.Path = "/health"
+ req.Header.Add("Before-Pipeline", "valid")
+ resp, err = http.DefaultClient.Do(req)
+ resp.Body.Close()
+ assert.Nil(err)
+ assert.Equal(http.StatusOK, resp.StatusCode, resp)
+ assert.Equal("valid", resp.Header.Get("After-Pipeline"), resp)
+
+ // update gfYaml to fallthrough before pipeline
+ yaml = fmt.Sprintf(gfYamlTmpl, `
+fallthrough:
+ beforePipeline: true
+`)
+ err = applyResource(yaml)
+ assert.Nil(err)
+ time.Sleep(3 * time.Second)
+
+ // pass before, exec pipeline, meet error, exec after
+ // not add header to before
+ req = makeReq()
+ resp, err = http.DefaultClient.Do(req)
+ resp.Body.Close()
+ assert.Nil(err)
+ assert.Equal(http.StatusServiceUnavailable, resp.StatusCode, resp)
+ assert.Empty(resp.Header.Get("After-Pipeline"), resp)
+
+ // pass before, exec pipeline, exec after
+ req = makeReq()
+ req.URL.Path = "/health"
+ resp, err = http.DefaultClient.Do(req)
+ resp.Body.Close()
+ assert.Nil(err)
+ assert.Equal(http.StatusOK, resp.StatusCode, resp)
+ assert.Equal("valid", resp.Header.Get("After-Pipeline"), resp)
+
+ // update gfYaml to fallthrough before pipeline
+ yaml = fmt.Sprintf(gfYamlTmpl, `
+fallthrough:
+ beforePipeline: true
+ pipeline: true
+`)
+ err = applyResource(yaml)
+ assert.Nil(err)
+ time.Sleep(3 * time.Second)
+
+ // fallthrough before, fallthrough pipeline, exec after
+ req = makeReq()
+ resp, err = http.DefaultClient.Do(req)
+ resp.Body.Close()
+ assert.Nil(err)
+ assert.Equal(http.StatusServiceUnavailable, resp.StatusCode, resp)
+ assert.Equal("valid", resp.Header.Get("After-Pipeline"), resp)
+}

docs/07.Reference/7.01.Controllers.md

@@ -391,6 +391,15 @@ beforePipeline:
  - name: validator
  kind: Validator
  ...
+
+# fallthrough controls the error handling behavior in different pipelines.
+# specifying whether subsequent stages should execute despite encountering errors in earlier stages. 
+# By default, encountering an error in either the beforePipeline or the pipeline stages halts the entire process.
+fallthrough:
+ # false indicates that if an error occurs in the beforePipeline, the pipeline will not execute.
+ beforePipeline: false
+ # false means that if an error occurs in the pipeline, the afterPipeline will not execute.
+ pipeline: false
 ---
 name: server-example
 kind: HTTPServer

pkg/object/globalfilter/globalfilter.go

@@ -66,6 +66,13 @@ type (
  Spec struct {
  BeforePipeline *pipeline.Spec `json:"beforePipeline,omitempty"`
  AfterPipeline *pipeline.Spec `json:"afterPipeline,omitempty"`
+ Fallthrough Fallthrough `json:"fallthrough,omitempty"`
+ }
+
+ // Fallthrough describes the fallthrough behavior.
+ Fallthrough struct {
+ BeforePipeline bool `json:"beforePipeline,omitempty"`
+ Pipeline bool `json:"pipeline,omitempty"`
  }
 
  // pipelineSpec defines pipeline spec to create an pipeline entity.
@@ -197,7 +204,11 @@ func (gf *GlobalFilter) Handle(ctx *context.Context, handler context.Handler) {
 
  before, _ := gf.beforePipeline.Load().(*pipeline.Pipeline)
  after, _ := gf.afterPipeline.Load().(*pipeline.Pipeline)
- p.HandleWithBeforeAfter(ctx, before, after)
+ option := pipeline.HandleWithBeforeAfterOption{
+ FallthroughBefore: gf.spec.Fallthrough.BeforePipeline,
+ FallthroughPipeline: gf.spec.Fallthrough.Pipeline,
+ }
+ p.HandleWithBeforeAfter(ctx, before, after, option)
 }
 
 // Close closes GlobalFilter itself.

pkg/object/pipeline/pipeline.go

@@ -310,9 +310,17 @@ func (p *Pipeline) getFilter(name string) filters.Filter {
  return p.filters[name]
 }
 
+// HandleWithBeforeAfterOption is the option of HandleWithBeforeAfter.
+// FallthroughBefore: if true, the pipeline will be executed even if the before pipeline ends.
+// FallthroughPipeline: if true, the after pipeline will be executed even if the pipeline ends.
+type HandleWithBeforeAfterOption struct {
+ FallthroughBefore bool
+ FallthroughPipeline bool
+}
+
 // HandleWithBeforeAfter handles the request, with additional flow defined by
 // the before/after pipeline.
-func (p *Pipeline) HandleWithBeforeAfter(ctx *context.Context, before, after *Pipeline) string {
+func (p *Pipeline) HandleWithBeforeAfter(ctx *context.Context, before, after *Pipeline, option HandleWithBeforeAfterOption) string {
  if len(p.spec.Data) > 0 {
  ctx.SetData("PIPELINE", p.spec.Data)
  }
@@ -331,11 +339,11 @@ func (p *Pipeline) HandleWithBeforeAfter(ctx *context.Context, before, after *Pi
  result, stats, sawEnd = p.doHandle(ctx, before.flow, stats)
  }
 
- if !sawEnd {
+ if !sawEnd || option.FallthroughBefore {
  result, stats, sawEnd = p.doHandle(ctx, p.flow, stats)
  }
 
- if !sawEnd && after != nil {
+ if (after != nil) && (!sawEnd || option.FallthroughPipeline) {
  result, stats, _ = p.doHandle(ctx, after.flow, stats)
  }
 

pkg/object/pipeline/pipeline_test.go

@@ -442,7 +442,7 @@ filters:
  ctx := context.New(tracing.NoopSpan)
  ctx.SetRequest(context.DefaultNamespace, req)
 
- pipeline.HandleWithBeforeAfter(ctx, nil, nil)
+ pipeline.HandleWithBeforeAfter(ctx, nil, nil, HandleWithBeforeAfterOption{})
  tags := ctx.Tags()
  assert.NotContains(tags, "filter1")
  assert.Contains(tags, "filter2")
@@ -467,7 +467,7 @@ filters:
 
  ctx = context.New(tracing.NoopSpan)
  ctx.SetRequest(context.DefaultNamespace, req)
- pipeline.HandleWithBeforeAfter(ctx, nil, after)
+ pipeline.HandleWithBeforeAfter(ctx, nil, after, HandleWithBeforeAfterOption{})
  tags = ctx.Tags()
  assert.NotContains(tags, "filter1")
  assert.Contains(tags, "filter2")
@@ -492,7 +492,7 @@ filters:
 
  ctx = context.New(tracing.NoopSpan)
  ctx.SetRequest(context.DefaultNamespace, req)
- pipeline.HandleWithBeforeAfter(ctx, before, after)
+ pipeline.HandleWithBeforeAfter(ctx, before, after, HandleWithBeforeAfterOption{})
  tags = ctx.Tags()
  assert.Contains(tags, "filter1")
  assert.NotContains(tags, "filter2")