[Question]: Pipeline HandleWithBeforeAfter not work well

[cyrnicolase]

https://github.com/easegress-io/easegress/blob/main/pkg/object/pipeline/pipeline.go#L391

result = node.filter.Handle(ctx)
stats = append(stats, FilterStat{
	Name:     alias,
	Kind:     node.filter.Kind().Name,
	Duration: fasttime.Since(start),
	Result:   result,
})

var ok bool
if next, ok = node.JumpIf[result]; result != "" && !ok {
	next = BuiltInFilterEnd
}

if next == BuiltInFilterEnd {
	sawEnd = true
	break
}

https://github.com/easegress-io/easegress/blob/main/pkg/object/pipeline/pipeline.go#L338

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, _ = p.doHandle(ctx, after.flow, stats)
}

I did not configure the JumpIf In my pipelines. When the backend response a http code 500, the pipeline will jump to the END. After that, the globalFilter's afterPipeline will not be executed in the program. This design is remarkably implausible.

I opine that this logic can be omitted.

// var ok bool
// if next, ok = node.JumpIf[result]; result != "" && !ok {
// 	next = BuiltInFilterEnd
// }

next = node.JumpIf[result]

[xxx7xxxx]

Thank you for pointing out the consideration of such a core design. I'd like to conclude that the END should end all flows including afterPipeline or not. Currently, it does jump over the afterPipeline. It is designed and implemented by @localvar , we will decide whether to change/enhance the design or not after getting his thoughts.

[localvar]

hi @cyrnicolase , this design is for backward compatibility with v1.x: the flow terminate when a filter returns an unexpected result.

If we modify the code as your proposal, the behavior will change to: the flow fallthrough to next filter when a filter returns an unexpected result.

Could you please provide the pipeline spec? so that we can fully understand your requirement

[cyrnicolase]

HI @localvar , this is my pipeline spec.

name: joyparty-https-proxy
kind: HTTPServer

port: 443
https: true
http3: false                # not use http3

keepAlive: true
keepAliveTimeout: 60s       # 保持链接的时间
maxConnections: 10240       # 客户端最大链接数
clientMaxBodySize: 83886080       # Max request body size; 80M
xForwardedFor: true
globalFilter: joyparty-global-filter
accessLogFormat: "{\"time\":\"{{Time}}\",\"remoteAddr\":\"{{RemoteAddr}}\",\"realIP\":\"{{RealIP}}\",\"host\":\"{{Host}}\",\"method\":\"{{Method}}\",\"uri\":\"{{URI}}\",\"proto\":\"{{Proto}}\",\"statusCode\":{{StatusCode}},\"duration\":\"{{Duration}}\",\"reqSize\":{{ReqSize}},\"respSize\":{{RespSize}},\"tags\":\"{{Tags}}\"}"

autoCert: true              # 自动化证书管理

rules:
  # websocket 反向代理
  - hosts:
      - isRegexp: false
        value: doll-api.lite-test.example.org
    paths:
      - backend: joyparty-pipeline-wsproxy-k8s
        pathPrefix: /api/free/ws/
        clientMaxBodySize: -1       # 表示客户端上行数据为 stream
      - backend: joyparty-pipeline-wsproxy-k8s
        pathPrefix: /api/portal/ws/
        clientMaxBodySize: -1       # 表示客户端上行数据为 stream

  # http 反向代理
  - hosts:
      - isRegexp: false
        value: admin-doll.test.example.org
    paths:
      - backend: joyparty-pipeline-httpproxy-k8s

  # 其他的域名流量也导入到 k8s 集群
  # 这里是做了一个 default 的方案
  - paths:
      - backend: joyparty-pipeline-httpproxy-k8s

---

# 代理到K8S集群的 http-proxy
name: joyparty-pipeline-httpproxy-k8s
kind: Pipeline
filters:
  - name: http-proxy
    kind: Proxy
    serverMaxBodySize: 83886080   # 80M
    pools:
      - servers:
          - url: http:https://10.0.0.113
          - url: http:https://10.0.0.114
        loadBalance:
          policy: roundRobin

---

# 代理到K8S集群的 wsproxy
name: joyparty-pipeline-wsproxy-k8s
kind: Pipeline
filters:
  - name: wsproxy
    kind: WebSocketProxy
    serverMaxBodySize: -1         # 表示服务端响应数据是 stream
    pools:
      - servers:
          - url: ws:https://10.0.0.113
          - url: ws:https://10.0.0.114
        insecureSkipVerify: true    # 不做Header头的Host检查
        loadBalance:
          policy: roundRobin

---

# 全局过滤器，生成请求ID，如果业务端要使用自己的请求ID，可以在自己的Pipeline中增加 YPDRequestID 过滤器
name: joyparty-global-filter
kind: GlobalFilter
beforePipeline:
  flow:
    - filter: ypd-request-id
  filters:
    - name: ypd-request-id
      kind: YPDRequestID
      algo: xid

afterPipeline:
  flow:
    - filter: ypd-request-log
  filters:
    - name: ypd-request-log
      kind: YPDRequestLog

My goal is use the globalFilter to record a custom log after an API Request. I believe that all HTTP responses should be as expected, and the flow should not go directly to the 'END' stage unless there are fatal errors. It is better for the user to orchestrate the flow transitions using 'JumpIf' rather than relying on them implicitly.

[localvar]

I think we can add a new option to the pipeline spec to to tell the pipeline on how to handle unexpected filter results, like below (onUnexpectedResult is just ad hoc name for the example, we should find a better one):

name: joyparty-pipeline-wsproxy-k8s
kind: Pipeline
onUnexpectedResult: terminate # could be 'fallthrough' or 'terminate', default is 'terminate'
filters:
  ...

if onUnexpectedResult is set to terminate, the flow jumps to end, this keeps the original behavior; and if set to fallthrough, the flow jumps to the next filter, this is the behavior which @cyrnicolase asks for.

[cyrnicolase]

@localvar Thanks for your reply.

My point is that the return values set by all Filters should be as expected. The Pipeline system itself should not be concerned with which Filter results are considered onUnexpectedResult. This control should be left to the users to manage based on JumpIf. Would this be better?

[localvar]

@cyrnicolase ,I've thought about your proposal and think it could be another solution with more precise control than mine.

the cons of your proposal is it requires user to add JumpIf to all filters, because we need to keep the default behavior unchanged.

we can also add the option to both pipeline and JumpIf, maybe this is the best solution.

[cyrnicolase]

Got it, when can we expect this feature to be merged into the Main branch?

[localvar]

Got it, when can we expect this feature to be merged into the Main branch?

@xxx7xxxx @suchen-sci

[suchen-sci]

Well, I think current way to solve the flow in pipeline is correct, when meet unexpected result and no jumpif is set, then we should terminate the pipeline. If previous filter is failed, then run filters after it can't be right. If a user want to run filters after a failed filter, then use jumpif.

But what may cause the problem is the part of before pipeline and after pipeline logic. END should only end current pipeline. And for after pipeline, it is global, users may use it some special things. So maybe, if I to fix this issue, I would like to add a option to pipeline to runAfterPipelineEvenEnd to spec, maybe better name?

I think add options to pipeline or jumpif may over complicate this problem. Because it breaks the flow execution and may bring much more problems. If a filter failed, and still run filters after it, what would happen?
In this case, after pipeline is special, technically speaking, it is not part of the current pipeline, it defined by global filters.

[suchen-sci]

i even think we should add this option to GlobalFilter?

[xxx7xxxx]

IMO, this proposal adds too much complexity to the execution logic of Pipeline, which is already enough in a YAML format. END should be an end for the current pipeline itself without taking effect on other pipelines. But the backward compatibility surely matters, so I propose adding options in golang-style in GlobalFilter. E.g:

name: joyparty-global-filter
kind: GlobalFilter
beforePipeline:

  # New Options
  fallthrough: true # default is false

  flow:
    - filter: ypd-request-id
  filters:
    - name: ypd-request-id
      kind: YPDRequestID
      algo: xid

# New Options
pipeline:
  fallthrough: true # default is false

afterPipeline:
  flow:
    - filter: ypd-request-log
  filters:
    - name: ypd-request-log
      kind: YPDRequestLog

If fallthrough is set true, END means the next pipeline instead of the whole process.

[localvar]

From the implementation aspect, adding the option to pipeline would be easier than adding it to the global filter. But I think the decision should be made from the requirement aspect:: user want which compoenent to control the behavior?

I don't have a strong idea for the answer as both components seem reasonable, maybe @cyrnicolase could help answer the question or we may need to wait for more use cases.

[cyrnicolase]

I agree that the END should be an end for the current pipeline itself without taking effect on other pipelines.

My viewpoint is the fallthrough option can be in the Pipeline which means the END whether could fallthrough to the next Pipeline.

name: joyparty-pipeline-httpproxy-k8s
kind: Pipeline
fallthrough: false     # whether pass END to the next pipeline, default is true
filters:
  - name: http-proxy
    kind: Proxy
    serverMaxBodySize: 83886080   # 80M
    pools:
      - servers:
          - url: http:https://10.0.0.113
          - url: http:https://10.0.0.114
        loadBalance:
          policy: roundRobin

Some code maybe like this.

[cyrnicolase]

Hello @localvar @suchen-sci @xxx7xxxx

How do we deal with this issue in the future? Is there a conclusion yet?

[suchen-sci]

I am ok that add an option to pipeline spec. It can help to control more cases.
By setting fallthrough to true in before pipeline, then if before pipeline failed, pipeline still works.
By setting fallthrough to true in pipeline, after pipeline will run if pipeline fails.

Am i right?

[suchen-sci]

Emmm, I think there are several cases we should consider carefully:

1. if beforePipeline failed, it is end, pipeline, afterPipeline three case.
2. if pipeline failed, it is end, afterPipeline two case.

I think no matter what design we choose, we need to figure out all these cases. Otherwise, we may need to change code later.

[suchen-sci]

And i am also afraid that add an option to pipeline spec works fine for a small set of pipelines. Some of our user may use 1000 pipelines. Then if they what to use this feature, they may need to edit 1000 yaml files...

Generally speaking, user want same behavior for GlobalFilter? So, maybe all pipelines that under that GlobalFilter should have same behavior. If that is true, I think @xxx7xxxx design maybe more appropriate in this case?

What do you think @cyrnicolase

[suchen-sci]

By the way, sorry for the delay.

[xxx7xxxx]

@cyrnicolase
We think adding options in GlobalFilter would be a suitable solution since your original requirement is to schedule 3 pipelines at a higher level instead of within the pipeline. But another problem about the GlobalFilter solution is whether you have a case that needs failed requests in beforePipeline to fall through to afterPipeline?

[cyrnicolase]

@cyrnicolase We think adding options in GlobalFilter would be a suitable solution since your original requirement is to schedule 3 pipelines at a higher level instead of within the pipeline. But another problem about the GlobalFilter solution is whether you have a case that needs failed requests in beforePipeline to fall through to afterPipeline?

Not yet, at least not for now.

And i am also afraid that add an option to pipeline spec works fine for a small set of pipelines. Some of our user may use 1000 pipelines. Then if they what to use this feature, they may need to edit 1000 yaml files...

Generally speaking, user want same behavior for GlobalFilter? So, maybe all pipelines that under that GlobalFilter should have same behavior. If that is true, I think @xxx7xxxx design maybe more appropriate in this case?

What do you think @cyrnicolase

I am all ok :)

[xxx7xxxx]

Okay, We would like to implement the proposal. Thx for your feedback.

[xxx7xxxx]

@cyrnicolase Hey we have supported this feature, please check that out #1233

[cyrnicolase]

@cyrnicolase Hey we have supported this feature, please check that out #1233

ok, thanks !