WIP: Add publisher confirms to ingress #543

ikavgo · 2021-12-08T12:51:38Z

Changes

/kind

Fixes #334

Release Note

Docs

knative-prow-robot · 2021-12-08T12:51:40Z

@ikvmw: The label(s) kind/<kind> cannot be applied, because the repository doesn't have them.

In response to this:

Changes

/kind

Fixes #334

Release Note
Docs

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.

knative-prow-robot · 2021-12-08T12:51:44Z

[APPROVALNOTIFIER] This PR is NOT APPROVED

This pull-request has been approved by: ikvmw
To complete the pull request process, please assign nak3 after the PR has been reviewed.
You can assign the PR to them by writing /assign @nak3 in a comment when ready.

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

Needs approval from an approver in each of these files:

OWNERS

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

codecov · 2021-12-08T12:58:20Z

Codecov Report

Merging #543 (4d67b20) into main (23a3926) will not change coverage.
The diff coverage is n/a.

❗ Current head 4d67b20 differs from pull request most recent head a9ef4ed. Consider uploading reports for the commit a9ef4ed to get more accurate results

@@           Coverage Diff           @@
##             main     #543   +/-   ##
=======================================
  Coverage   75.78%   75.78%           
=======================================
  Files          47       47           
  Lines        2783     2783           
=======================================
  Hits         2109     2109           
  Misses        549      549           
  Partials      125      125

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 23a3926...a9ef4ed. Read the comment docs.

knative-prow-robot · 2021-12-08T21:32:38Z

@ikvmw: PR needs rebase.

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.

benmoss

I think we need to step back and improve the way the amqp library handles confirms rather than working around it like this. It's pretty hard to reason about this solution with all the levels of concurrency that we're working with here.

I am thinking something like if channel.Publish returned a promise-type object that we would be able to push a lot of this complexity into the library and hide it from users.

benmoss · 2021-12-16T15:48:19Z

cmd/ingress/main.go

@@ -1,27 +1,29 @@
 /*
-Copyright 2020 The Knative Authors
+   Copyright 2020 The Knative Authors


Let's remove these whitespace changes

benmoss · 2021-12-16T16:56:56Z

cmd/ingress/main.go

+	channels []*Channel
+	logger   *zap.SugaredLogger
+
+	rc uint64


These variable names could be a little more clear

benmoss · 2021-12-16T17:06:09Z

cmd/ingress/main.go

+	crc := atomic.AddUint64(&env.rc, 1)
+
+	channel := env.channels[crc%uint64(env.cc)]


We should use a Go channel and have all these Channels select on it rather than trying to do this load balancing logic

benmoss · 2021-12-16T17:14:25Z

cmd/ingress/channel.go

+}
+
+func (channel *Channel) confirmsHandler() func() error {
+	return func() error {


I don't think you need to make this an anonymous function, just have this be the signature of confirmsHandler

benmoss · 2021-12-16T17:15:51Z

cmd/ingress/channel.go

+
+	if !present {
+		// really die?
+		log.Fatalf("Received confirm for unknown send. DeliveryTag: %d", confirm.DeliveryTag)


I would Warn but not Fatal here, this shouldn't be possible unless there's a bug in RabbitMQ

embano1

Overall I found the card hard to review for correctness due to the concurrency and non-transactional locking at play. I also don't see how this code is more performant than the proposed sync.Pool take/return Channel proposed by @bmoss.

Can you please explain why this complexity is justified? Perhaps I'm not understanding the optimizations at play.

embano1 · 2021-12-17T10:46:55Z

cmd/ingress/main.go

@@ -36,34 +38,44 @@ import (
 const (
 	defaultMaxIdleConnections        = 1000
 	defaultMaxIdleConnectionsPerHost = 1000
+	defaultMaxAmqpChannels           = 10
 )

 type envConfig struct {


nit: can we create a separate ingress server object and use envConfig only for what it actually does? It's a bit confusing in the code and method signatures

embano1 · 2021-12-17T10:48:07Z

cmd/ingress/main.go

 }

 func main() {
 	var env envConfig
+
+	var err error


If envConfig is separated from the server implementation, this line is not needed anymore (and addresses err shadowing in line 62)

embano1 · 2021-12-17T10:50:09Z

cmd/ingress/main.go

 	if err != nil {
 		log.Fatalf("failed to connect to RabbitMQ: %s", err)
 	}
-	defer conn.Close()
+	defer env.conn.Close()


I'd at least log the error here or use the run pattern where main only does initialization logic, e.g. env vars, and then calls a blocking run(context.Context) error on the ingress server. defer can then be handled inside run and any error easily returned to main if needed.

embano1 · 2021-12-17T10:52:10Z

cmd/ingress/main.go

+
+	env.openAmqpChannels()
+
+	defer env.closeAmqpChannels()

 	env.logger = logging.FromContext(context.Background())


we could use https://pkg.go.dev/knative.dev/pkg/signals to handle SIGTERM etc

embano1 · 2021-12-17T10:53:57Z

cmd/ingress/main.go

 	if err := envconfig.Process("", &env); err != nil {
 		log.Fatal("Failed to process env var", zap.Error(err))
 	}

-	conn, err := amqp.Dial(env.BrokerURL)
+	env.conn, err = amqp.Dial(env.BrokerURL)
+
 	if err != nil {
 		log.Fatalf("failed to connect to RabbitMQ: %s", err)


I'd use the pkg logger (move its construction up so you can use it) and thus allow for structured logging (see also other places where we still use log.)

embano1 · 2021-12-17T11:05:08Z

cmd/ingress/channel.go

+		log.Fatalf("failed to open a channel: %s", err)
+	}
+
+	channel := &Channel{channel: a_channel, counter: 1, pconfirms_chan: make(chan amqp.Confirmation)}


we use unbuffered channel for confirmation. docs for confirmation channel in NotifyPublish state it should be buffered to accomodate for multiple in-flights, otherwise risking deadlocks. Could we run into this scenario with an unbuffered channel?

embano1 · 2021-12-17T11:06:03Z

cmd/ingress/channel.go

+		for confirm := range channel.pconfirms_chan {
+			channel.handleConfirm(confirm)
+		}
+		return nil


a log debug line could not harm here to understand that the chan was closed, e.g. due to underlying RMQ channel closed

embano1 · 2021-12-17T11:12:50Z

cmd/ingress/main.go

+}
+
+func (env *envConfig) sendMessage(headers amqp.Table, bytes []byte) (chan amqp.Confirmation, error) {
+	crc := atomic.AddUint64(&env.rc, 1)


looks like this means the first message will bump counter to 1 and hence env.channels[0] would never be used? if so, you might change your slice indexing in line 175 accordingly

embano1 · 2021-12-17T11:13:48Z

cmd/ingress/main.go

-			Body:        bytes,
-		}); err != nil {
+
+	ch, err := env.sendMessage(headers, bytes)


nit: confirmCh

embano1 · 2021-12-17T11:24:17Z

cmd/ingress/channel.go

+}
+
+func (channel *Channel) publish(exchange string, headers amqp.Table, bytes []byte) (chan amqp.Confirmation, error) {
+	channel.mutex.Lock()


This basically means that every message published will lock the particular channel, i.e. serialize the logic and constrain parallelism to the number of defined channels, correct?

ikavgo added 2 commits December 8, 2021 13:14

update amqp091-go to 1.2

43df496

Reuse Ben's code and approach with mutex

4d67b20

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 Dec 8, 2021

knative-prow-robot added the size/L Denotes a PR that changes 100-499 lines, ignoring generated files. label Dec 8, 2021

knative-prow-robot added the needs-rebase Indicates a PR cannot be merged because it has merge conflicts with HEAD. label Dec 8, 2021

FIXUP: scratch the pool

a9ef4ed

benmoss reviewed Dec 16, 2021

View reviewed changes

embano1 reviewed Dec 17, 2021

View reviewed changes

deadtrickster mentioned this pull request Jan 12, 2022

Expose a method to enable out-of-order Publisher Confirms rabbitmq/amqp091-go#33

Merged

ikavgo closed this Jan 13, 2022

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

WIP: Add publisher confirms to ingress #543

WIP: Add publisher confirms to ingress #543

ikavgo commented Dec 8, 2021

knative-prow-robot commented Dec 8, 2021

Changes

knative-prow-robot commented Dec 8, 2021

codecov bot commented Dec 8, 2021 •

edited

Loading

knative-prow-robot commented Dec 8, 2021

benmoss left a comment

benmoss Dec 16, 2021

benmoss Dec 16, 2021

benmoss Dec 16, 2021

benmoss Dec 16, 2021

benmoss Dec 16, 2021

embano1 left a comment

embano1 Dec 17, 2021

embano1 Dec 17, 2021

embano1 Dec 17, 2021

embano1 Dec 17, 2021

embano1 Dec 17, 2021

embano1 Dec 17, 2021

embano1 Dec 17, 2021

embano1 Dec 17, 2021

embano1 Dec 17, 2021

embano1 Dec 17, 2021

		crc := atomic.AddUint64(&env.rc, 1)

		channel := env.channels[crc%uint64(env.cc)]

WIP: Add publisher confirms to ingress #543

WIP: Add publisher confirms to ingress #543

Conversation

ikavgo commented Dec 8, 2021

Changes

knative-prow-robot commented Dec 8, 2021

Changes

knative-prow-robot commented Dec 8, 2021

codecov bot commented Dec 8, 2021 • edited Loading

Codecov Report

knative-prow-robot commented Dec 8, 2021

benmoss left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

embano1 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codecov bot commented Dec 8, 2021 •

edited

Loading