IMC and Multi-Tenant Channel Based Broker retries #2932

pierDipi · 2020-04-07T18:57:13Z

Proposed Changes

MessageDispatcher retries sending events.
Added e2e tests for the redelivery feature.

Release Note

- 🎁 Add new feature 
In Memory Channel and Multi-Tenant Channel Based Broker retry sending events

knative-prow-robot · 2020-04-07T18:57:27Z

Hi @pierDipi. Thanks for your PR.

I'm waiting for a knative member to verify that this patch is reasonable to test. If it is, they should reply with /ok-to-test on its own line. Until that is done, I will not automatically test new commits in this PR, but the usual testing commands by org members will still work. Regular contributors should join the org to skip this step.

Once the patch is verified, the new status will be reflected by the ok-to-test label.

I understand the commands that are listed here.

Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes/test-infra repository.

pierDipi · 2020-04-07T18:57:49Z

@slinkydeveloper @lionelvillard

lionelvillard · 2020-04-08T13:17:00Z

/hold

let's wait for the next CloudEvent SDK preview release to land.

slinkydeveloper · 2020-04-14T08:08:42Z

Let's keep this on hold, we need to understand if we can use the protocol/http.Protocol from the sdk

vaikas · 2020-05-08T17:24:05Z

Just a ping on the status of this?

slinkydeveloper · 2020-05-08T17:28:01Z

@vaikas i have no updates, let's check this next week

slinkydeveloper · 2020-06-04T08:59:31Z

What I'm really worried about is that we're adding ourselves code for retrying http requests that already exists out there (which is error prone and we need to test). For example, i'm not really sure what happens in that code if you pass an unbuffered io.Reader in the request body, because after it's read for the first time it could be broken to read it a second time...
Have you considered using this library? https://github.com/hashicorp/go-retryablehttp
We can integrate it in zero time, we don't need any testing, and it just works (libraries from these org are also well mantained)

pierDipi · 2020-06-04T09:52:44Z

Have you considered using this library? https://github.com/hashicorp/go-retryablehttp
We can integrate it in zero time, we don't need any testing, and it just works (libraries from these org are also well mantained)

No, I didn't, but I'm fine with this

Let me know if I can go ahead using that library.

slinkydeveloper · 2020-06-04T09:56:07Z

I think we need somebody else on this topic, @n3wscott wdyt?

lionelvillard · 2020-06-04T12:46:54Z

really cool ideas in this library! Thanks for sharing @slinkydeveloper! It is very flexible (BYO backoff policy and retry policy) so from a functional PoV it does what we want.

matzew · 2020-06-25T07:30:33Z

Have you considered using this library? https://github.com/hashicorp/go-retryablehttp
We can integrate it in zero time, we don't need any testing, and it just works (libraries from these org are also well mantained)

I second what @slinkydeveloper said here! Let's use proven and maintained code, instead of hacking /cc @pierDipi

pierDipi · 2020-06-25T07:46:51Z

Ok, I'll continue to work on this.

slinkydeveloper · 2020-07-16T14:12:08Z

/assign

Signed-off-by: Pierangelo Di Pilato <[email protected]>

slinkydeveloper

Little hint on the signature of MessageDispatcher

pkg/channel/message_dispatcher.go

pkg/kncloudevents/message_sender.go

Signed-off-by: Pierangelo Di Pilato <[email protected]>

pkg/kncloudevents/message_sender.go

Signed-off-by: Pierangelo Di Pilato <[email protected]>

pkg/reconciler/inmemorychannel/dispatcher/inmemorychannel.go

Signed-off-by: Pierangelo Di Pilato <[email protected]>

knative-metrics-robot · 2020-07-21T07:36:24Z

The following is the coverage report on the affected files.
Say /test pull-knative-eventing-go-coverage to re-run this coverage report

File	Old Coverage	New Coverage	Delta
pkg/channel/fanout/fanout_message_handler.go	92.6%	89.1%	-3.5
pkg/channel/message_dispatcher.go	86.1%	87.3%	1.3
pkg/channel/multichannelfanout/multi_channel_fanout_message_handler.go	90.9%	84.6%	-6.3
pkg/reconciler/inmemorychannel/dispatcher/inmemorychannel.go	78.9%	77.1%	-1.8

pierDipi · 2020-07-21T08:27:02Z

/retest

knative-test-reporter-robot · 2020-07-21T09:14:33Z

The following jobs failed:

Test name	Triggers	Retries
pull-knative-eventing-integration-tests		0/3

Failed non-flaky tests preventing automatic retry of pull-knative-eventing-integration-tests:

test/e2e.TestEventTransformationForTriggerV1Beta1BrokerV1
test/e2e.TestEventTransformationForTriggerV1Beta1BrokerV1/InMemoryChannel-messaging.knative.dev/v1beta1

pierDipi · 2020-07-21T09:18:57Z

Failed to create v1 broker "inmemorychannel": conversion webhook for eventing.knative.dev/v1, Kind=Broker failed: Post https://eventing-webhook.knative-eventing-n2htblyaln.svc:443/resource-conversion?timeout=30s: EOF

Have you ever seen something similar?

vaikas · 2020-07-21T09:20:23Z

Yeah :(

knative/pkg#1509

pierDipi · 2020-07-21T09:21:47Z

Thanks!

/retest

slinkydeveloper · 2020-07-21T09:40:25Z

/unhold

vaikas · 2020-07-21T10:16:15Z

pkg/channel/fanout/fanout_message_handler.go

 	eventingduck "knative.dev/eventing/pkg/apis/duck/v1beta1"
 	"knative.dev/eventing/pkg/channel"
+	"knative.dev/eventing/pkg/kncloudevents"
 )

 const (


Not your doing, but 15 minute timeout? :) I'll look what this is for. Just a note :)

It seems too much :)

It's used here:

eventing/pkg/channel/fanout/fanout_message_handler.go

Lines 175 to 193 in d5ea26f

errorCh := make(chan error, subs)

for _, sub := range f.config.Subscriptions {

go func(s Subscription) {

errorCh <- f.makeFanoutRequest(ctx, bufferedMessage, additionalHeaders, s)

}(sub)

}

for range f.config.Subscriptions {

select {

case err := <-errorCh:

if err != nil {

f.logger.Error("Fanout had an error", zap.Error(err))

return err

}

case <-time.After(f.timeout):

f.logger.Error("Fanout timed out")

return errors.New("fanout timed out")

}

}

vaikas · 2020-07-21T10:16:59Z

pkg/channel/fanout/fanout_message_handler.go

 )

 const (
 	defaultTimeout = 15 * time.Minute
 )

+type Subscription struct {
+	eventingduck.SubscriberSpec
+	RetriesConfig kncloudevents.RetryConfig


nit: RetriesConfig => RetryConfig?

vaikas · 2020-07-21T10:28:20Z

Does this also fix / help with:
#2357

/lgtm
/approve

knative-prow-robot · 2020-07-21T10:28:32Z

[APPROVALNOTIFIER] This PR is APPROVED

This pull-request has been approved by: pierDipi, vaikas

The full list of commands accepted by this bot can be found here.

The pull request process is described here

Needs approval from an approver in each of these files:

~~OWNERS~~ [vaikas]

Approvers can indicate their approval by writing /approve in a comment
Approvers can cancel approval by writing /approve cancel in a comment

pierDipi · 2020-07-21T10:32:14Z

Does this also fix / help with:
#2357

It looks like we have the opposite behavior: https://prow.knative.dev/view/gcs/knative-prow/pr-logs/pull/knative_eventing/2932/pull-knative-eventing-upgrade-tests/1285477487198867458

event #* received 2 times, but should be received only once

slinkydeveloper · 2020-07-21T10:45:11Z

Thanks @pierDipi for looking in this long-standing issue!

pierDipi · 2020-07-21T10:46:15Z

Thanks for your reviews 😄

googlebot added the cla: yes Indicates the PR's author has signed the CLA. label Apr 7, 2020

knative-prow-robot added size/XL Denotes a PR that changes 500-999 lines, ignoring generated files. area/test-and-release Test infrastructure, tests or release labels Apr 7, 2020

knative-prow-robot requested review from chaodaiG, slinkydeveloper and yolocs April 7, 2020 18:57

knative-prow-robot added the needs-ok-to-test Indicates a PR that requires an org member to verify it is safe to test. label Apr 7, 2020

knative-prow-robot added do-not-merge/hold Indicates that a PR should not merge because someone has issued a /hold command. needs-rebase Indicates a PR cannot be merged because it has merge conflicts with HEAD. labels Apr 8, 2020

pierDipi force-pushed the broker-redelivery branch from 382ed6c to a2606e0 Compare June 4, 2020 08:59

knative-prow-robot removed the needs-rebase Indicates a PR cannot be merged because it has merge conflicts with HEAD. label Jun 4, 2020

pierDipi force-pushed the broker-redelivery branch from a2606e0 to c8d1375 Compare July 16, 2020 11:56

knative-prow-robot added size/XXL Denotes a PR that changes 1000+ lines, ignoring generated files. and removed size/XL Denotes a PR that changes 500-999 lines, ignoring generated files. labels Jul 16, 2020

pierDipi changed the title ~~Message sender retries~~ [WIP] Retries Jul 16, 2020

knative-prow-robot added the do-not-merge/work-in-progress Indicates that a PR should not merge because it is a work in progress. label Jul 16, 2020

pierDipi force-pushed the broker-redelivery branch from c8d1375 to 728436c Compare July 16, 2020 12:42

Create a new struct with SubscriberSpec and RetriesConfig

e029985

Signed-off-by: Pierangelo Di Pilato <[email protected]>

pierDipi requested a review from slinkydeveloper July 21, 2020 06:31

slinkydeveloper reviewed Jul 21, 2020

View reviewed changes

pkg/channel/message_dispatcher.go Outdated Show resolved Hide resolved

pkg/kncloudevents/message_sender.go Outdated Show resolved Hide resolved

Add DispatchMessageWithRetries to MessageDispatcher interface

5a73e1c

Signed-off-by: Pierangelo Di Pilato <[email protected]>

vaikas reviewed Jul 21, 2020

View reviewed changes

pkg/kncloudevents/message_sender.go Outdated Show resolved Hide resolved

Rename to RetryConfig

3a43578

Signed-off-by: Pierangelo Di Pilato <[email protected]>

vaikas reviewed Jul 21, 2020

View reviewed changes

pkg/reconciler/inmemorychannel/dispatcher/inmemorychannel.go Show resolved Hide resolved

Add UT for delivery spec

030c6c6

Signed-off-by: Pierangelo Di Pilato <[email protected]>

knative-prow-robot removed the do-not-merge/hold Indicates that a PR should not merge because someone has issued a /hold command. label Jul 21, 2020

pierDipi requested review from slinkydeveloper and vaikas July 21, 2020 10:14

vaikas reviewed Jul 21, 2020

View reviewed changes

knative-prow-robot assigned vaikas Jul 21, 2020

knative-prow-robot added the lgtm Indicates that a PR is ready to be merged. label Jul 21, 2020

knative-prow-robot added the approved Indicates a PR has been approved by an approver from all required OWNERS files. label Jul 21, 2020

knative-prow-robot merged commit d5ea26f into knative:master Jul 21, 2020

pierDipi mentioned this pull request Jul 21, 2020

Rename RetryConfig #3671

Merged

pierDipi mentioned this pull request Jul 21, 2020

[regression] Upgrade to latest breaks receiving events or produce event doubles #2665

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

IMC and Multi-Tenant Channel Based Broker retries #2932

IMC and Multi-Tenant Channel Based Broker retries #2932

pierDipi commented Apr 7, 2020 •

edited

Loading

knative-prow-robot commented Apr 7, 2020

pierDipi commented Apr 7, 2020

lionelvillard commented Apr 8, 2020

slinkydeveloper commented Apr 14, 2020

vaikas commented May 8, 2020

slinkydeveloper commented May 8, 2020

slinkydeveloper commented Jun 4, 2020 •

edited

Loading

pierDipi commented Jun 4, 2020

slinkydeveloper commented Jun 4, 2020

lionelvillard commented Jun 4, 2020

matzew commented Jun 25, 2020

pierDipi commented Jun 25, 2020

slinkydeveloper commented Jul 16, 2020

slinkydeveloper left a comment

knative-metrics-robot commented Jul 21, 2020

pierDipi commented Jul 21, 2020

knative-test-reporter-robot commented Jul 21, 2020

pierDipi commented Jul 21, 2020

vaikas commented Jul 21, 2020

pierDipi commented Jul 21, 2020

slinkydeveloper commented Jul 21, 2020

vaikas Jul 21, 2020

pierDipi Jul 21, 2020

vaikas Jul 21, 2020

pierDipi Jul 21, 2020

vaikas commented Jul 21, 2020

knative-prow-robot commented Jul 21, 2020

pierDipi commented Jul 21, 2020

slinkydeveloper commented Jul 21, 2020

pierDipi commented Jul 21, 2020

	errorCh := make(chan error, subs)
	for _, sub := range f.config.Subscriptions {
	go func(s Subscription) {
	errorCh <- f.makeFanoutRequest(ctx, bufferedMessage, additionalHeaders, s)
	}(sub)
	}

	for range f.config.Subscriptions {
	select {
	case err := <-errorCh:
	if err != nil {
	f.logger.Error("Fanout had an error", zap.Error(err))
	return err
	}
	case <-time.After(f.timeout):
	f.logger.Error("Fanout timed out")
	return errors.New("fanout timed out")
	}
	}

IMC and Multi-Tenant Channel Based Broker retries #2932

IMC and Multi-Tenant Channel Based Broker retries #2932

Conversation

pierDipi commented Apr 7, 2020 • edited Loading

Proposed Changes

knative-prow-robot commented Apr 7, 2020

pierDipi commented Apr 7, 2020

lionelvillard commented Apr 8, 2020

slinkydeveloper commented Apr 14, 2020

vaikas commented May 8, 2020

slinkydeveloper commented May 8, 2020

slinkydeveloper commented Jun 4, 2020 • edited Loading

pierDipi commented Jun 4, 2020

slinkydeveloper commented Jun 4, 2020

lionelvillard commented Jun 4, 2020

matzew commented Jun 25, 2020

pierDipi commented Jun 25, 2020

slinkydeveloper commented Jul 16, 2020

slinkydeveloper left a comment

Choose a reason for hiding this comment

knative-metrics-robot commented Jul 21, 2020

pierDipi commented Jul 21, 2020

knative-test-reporter-robot commented Jul 21, 2020

pierDipi commented Jul 21, 2020

vaikas commented Jul 21, 2020

pierDipi commented Jul 21, 2020

slinkydeveloper commented Jul 21, 2020

vaikas Jul 21, 2020

Choose a reason for hiding this comment

pierDipi Jul 21, 2020

Choose a reason for hiding this comment

vaikas Jul 21, 2020

Choose a reason for hiding this comment

pierDipi Jul 21, 2020

Choose a reason for hiding this comment

vaikas commented Jul 21, 2020

knative-prow-robot commented Jul 21, 2020

pierDipi commented Jul 21, 2020

slinkydeveloper commented Jul 21, 2020

pierDipi commented Jul 21, 2020

pierDipi commented Apr 7, 2020 •

edited

Loading

slinkydeveloper commented Jun 4, 2020 •

edited

Loading