eventing: [Support Request | Help] Backend function (sink) was triggered multiple times within one function invocation

Expected Behavior

Generally, I expected that my backend function would be triggered once only per function invocation.

Any help would be much appreciated. Thanks.

Actual Behavior

The backend function was triggered 10 times per function invocation.

Logs from the imc-dispatcher

{"level":"error","ts":"2023-12-18T13:57:05.059Z","logger":"inmemorychannel-dispatcher","caller":"fanout/fanout_event_handler.go:303","msg":"Fanout had an error","commit":"747a9f5","knative.dev/pod":"imc-dispatcher-69d78445c5-prbhr","knative.dev/controller":"knative.dev.eventing.pkg.reconciler.inmemorychannel.dispatcher.Reconciler","knative.dev/kind":"messaging.knative.dev.InMemoryChannel","knative.dev/traceid":"a6acd47a-55b3-4ace-a0fd-ee45f816629d","knative.dev/key":"staging/default-kne-trigger","error":"unable to complete request to http://broker-filter.knative-eventing.svc.cluster.local/triggers/staging/helloworld-cloudevent-function/87db3ff9-b234-46d5-b7dc-92cb788c1bdc: unexpected HTTP response, expected 2xx, got 502","stacktrace":"knative.dev/eventing/pkg/channel/fanout.(*FanoutEventHandler).dispatch\n\tknative.dev/eventing/pkg/channel/fanout/fanout_event_handler.go:303\nknative.dev/eventing/pkg/channel/fanout.NewFanoutEventHandler.createEventReceiverFunction.func1.1\n\tknative.dev/eventing/pkg/channel/fanout/fanout_event_handler.go:223"}
{"level":"error","ts":"2023-12-18T14:00:30.641Z","logger":"inmemorychannel-dispatcher","caller":"fanout/fanout_event_handler.go:303","msg":"Fanout had an error","commit":"747a9f5","knative.dev/pod":"imc-dispatcher-69d78445c5-prbhr","knative.dev/controller":"knative.dev.eventing.pkg.reconciler.inmemorychannel.dispatcher.Reconciler","knative.dev/kind":"messaging.knative.dev.InMemoryChannel","knative.dev/traceid":"a6acd47a-55b3-4ace-a0fd-ee45f816629d","knative.dev/key":"staging/default-kne-trigger","error":"unable to complete request to http://broker-filter.knative-eventing.svc.cluster.local/triggers/staging/helloworld-cloudevent-function/87db3ff9-b234-46d5-b7dc-92cb788c1bdc: unexpected HTTP response, expected 2xx, got 502","stacktrace":"knative.dev/eventing/pkg/channel/fanout.(*FanoutEventHandler).dispatch\n\tknative.dev/eventing/pkg/channel/fanout/fanout_event_handler.go:303\nknative.dev/eventing/pkg/channel/fanout.NewFanoutEventHandler.createEventReceiverFunction.func1.1\n\tknative.dev/eventing/pkg/channel/fanout/fanout_event_handler.go:223"}

Steps to Reproduce the Problem

Create a default broker in the staging namespace

---
apiVersion: eventing.knative.dev/v1
kind: Broker
metadata:
  name: default
  namespace: staging
  annotations:
    eventing.knative.dev/broker.class: MTChannelBasedBroker

Create trigger that maps to the backend function

---
apiVersion: eventing.knative.dev/v1
kind: Trigger
metadata:
  name: helloworld-cloudevent-function
  namespace: staging
spec:
  broker: default
  filter:
    attributes:
      type: dev.knative.staging.helloworld-cloudevent-function
      source: dev.knative.staging/eventing
  subscriber:
    ref:
      apiVersion: v1
      kind: Service
      name: helloworld-cloudevent-function

Invoke the function with curl command or API Client Tool like Insomnia

curl --request POST \
  --url http://broker-ingress.knative-eventing.svc.cluster.local/staging/default  \
  --header 'Ce-Id: 730bb8d8-0b5a-4184-9c69-5840078ef7d6' \
  --header 'Ce-Source: dev.knative.staging/eventing' \
  --header 'Ce-Type: dev.knative.staging.helloworld-cloudevent-function' \
  --header 'Ce-specversion: 1.0' \
  --header 'Content-Type: application/json' \
  --header 'User-Agent: insomnia/8.4.5' \
  --data '{"msg": "ping"}'

Additional Info

Installation

Knative Operator -> Knative-Eventing

Logs from the backend

Source Code: https://github.com/rocats/helloworld-cloudevent-function

...
10.0.0.47 - - [18/Dec/2023 14:05:17] "POST / HTTP/1.1" 200 -
2023-12-18 14:05:43 [debug    ] Function loaded
2023-12-18 14:05:43 [debug    ] Received event with ID: 730bb8d8-0b5a-4184-9c69-5840078ef7d6 and data {'msg': 'ping'}
2023-12-18 14:05:43 [debug    ] Cloudevent: {'attributes': {'specversion': '1.0', 'id': '730bb8d8-0b5a-4184-9c69-5840078ef7d6', 'source': 'dev.knative.staging/eventing', 'type': 'dev.knative.staging.helloworld-cloudevent-function', 'datacontenttype': 'application/json', 'knativearrivaltime': '2023-12-18T14:04:52.220290693Z', 'time': '2023-12-18T14:05:43.305629+00:00'}, 'data': {'msg': 'ping'}}
2023-12-18 14:05:43 [info     ] OK
10.0.0.47 - - [18/Dec/2023 14:05:43] "POST / HTTP/1.1" 200 -
2023-12-18 14:06:21 [debug    ] Function loaded
2023-12-18 14:06:21 [debug    ] Received event with ID: 536808d3-88be-4077-9d7a-a3f162705f79 and data {'msg': 'ping'}
2023-12-18 14:06:21 [debug    ] Cloudevent: {'attributes': {'specversion': '1.0', 'id': '536808d3-88be-4077-9d7a-a3f162705f79', 'source': 'dev.knative.staging/eventing', 'type': 'dev.knative.staging.helloworld-cloudevent-function', 'datacontenttype': 'application/json', 'knativearrivaltime': '2023-12-18T14:02:56.76725493Z', 'time': '2023-12-18T14:06:21.463494+00:00'}, 'data': {'msg': 'ping'}}
2023-12-18 14:06:21 [info     ] OK
10.0.0.47 - - [18/Dec/2023 14:06:21] "POST / HTTP/1.1" 200 -
2023-12-18 14:06:34 [debug    ] Function loaded
2023-12-18 14:06:34 [debug    ] Received event with ID: 730bb8d8-0b5a-4184-9c69-5840078ef7d6 and data {'msg': 'ping'}
2023-12-18 14:06:34 [debug    ] Cloudevent: {'attributes': {'specversion': '1.0', 'id': '730bb8d8-0b5a-4184-9c69-5840078ef7d6', 'source': 'dev.knative.staging/eventing', 'type': 'dev.knative.staging.helloworld-cloudevent-function', 'datacontenttype': 'application/json', 'knativearrivaltime': '2023-12-18T14:04:52.220290693Z', 'time': '2023-12-18T14:06:34.515597+00:00'}, 'data': {'msg': 'ping'}}
2023-12-18 14:06:34 [info     ] OK
10.0.0.47 - - [18/Dec/2023 14:06:34] "POST / HTTP/1.1" 200 -
2023-12-18 14:08:16 [debug    ] Function loaded
2023-12-18 14:08:16 [debug    ] Received event with ID: 730bb8d8-0b5a-4184-9c69-5840078ef7d6 and data {'msg': 'ping'}
2023-12-18 14:08:16 [debug    ] Cloudevent: {'attributes': {'specversion': '1.0', 'id': '730bb8d8-0b5a-4184-9c69-5840078ef7d6', 'source': 'dev.knative.staging/eventing', 'type': 'dev.knative.staging.helloworld-cloudevent-function', 'datacontenttype': 'application/json', 'knativearrivaltime': '2023-12-18T14:04:52.220290693Z', 'time': '2023-12-18T14:08:16.925878+00:00'}, 'data': {'msg': 'ping'}}
2023-12-18 14:08:16 [info     ] OK
10.0.0.47 - - [18/Dec/2023 14:08:16] "POST / HTTP/1.1" 200 -
...

About this issue

Original URL
State: closed
Created 6 months ago
Comments: 19 (10 by maintainers)

Most upvoted comments

Issue resolved. Thanks all the kind help.

yqlbu on Dec 19, 2023

Thanks for your help! If you need any assistance in contributing to the knative docs, please feel free to ping me on github or on the CNCF slack instance (there are many knative channels there, all under the naming convention #knative-*)

Cali0707 on Dec 19, 2023

Sorry if I’m not understanding your question properly. If you want to return the response from your backend function you will need to use a cloudevent. Cloudevents can have arbitrary data in the data field, but need some specific headers to be set. You can read more in the links I’ve sent above for how that will work. Knative brokers will not accept anything except for valid cloudevents, so if you want to return data from your function you will need to use a cloudevent to do so. The good news is they are a lightweight set of headers in your http request/response.

According to the Knative spec the 202 status code indicates that the event was delivered successfully, and the 200 status code indicates that the event was delivered successfully and the event was in the reply. However in practice I believe that every 2xx response code is treated as a success

Cali0707 on Dec 19, 2023

Also, I wonder if the Knative Broker is capable of returning a raw response from the function alone with custom status_code.

e.g
{
   "message": "sent from my function"
}
200

For this, you need to look at the cloudevents format. Knative brokers only accept http cloudevents. To create one, you can either read the spec and set the appropriate http headers yourself, or use the cloudevents python sdk (or similar for other languages)

Cali0707 on Dec 19, 2023

@yqlbu the broker expects a cloud event which corresponds to the spec. Maybe you can use the python SDK to make a cloud event? Or alternatively, return '', 200 may work, as Pierangelo described here. However, I am not very familiar with flask.

For some sample nodejs functions, feel free to look around at the code samples in https://github.com/Cali0707/knative-eda-demos

Cali0707 on Dec 18, 2023

See the conversation here for more

Cali0707 on Dec 18, 2023