WIP: Add publisher confirms to ingress

cmd/ingress/channel.go

@@ -0,0 +1,107 @@
+/*
+   Copyright 2021 The Knative Authors
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package main
+
+import (
+	"log"
+	"sync"
+
+	amqp "github.com/rabbitmq/amqp091-go"
+)
+
+type Channel struct {
+	channel        *amqp.Channel
+	counter        uint64
+	mutex          sync.Mutex
+	confirmsMap    sync.Map
+	pconfirms_chan chan amqp.Confirmation
+}
+
+func openChannel(conn *amqp.Connection) *Channel {
+	a_channel, err := conn.Channel()
+	if err != nil {
+		log.Fatalf("failed to open a channel: %s", err)
+	}
+
+	channel := &Channel{channel: a_channel, counter: 1, pconfirms_chan: make(chan amqp.Confirmation)}
+
+	go channel.confirmsHandler()
+
+	a_channel.NotifyPublish(channel.pconfirms_chan)
+	// noWait is false
+	if err := a_channel.Confirm(false); err != nil {
+		log.Fatalf("faild to switch connection channel to confirm mode: %s", err)
+	}
+
+	return channel
+}
+
+func (channel *Channel) publish(exchange string, headers amqp.Table, bytes []byte) (chan amqp.Confirmation, error) {
+	channel.mutex.Lock()
+	defer channel.mutex.Unlock()
+
+	cchan := make(chan amqp.Confirmation, 1)
+
+	channel.confirmsMap.Store(channel.counter, cchan)
+
+	if err := channel.channel.Publish(
+		exchange,
+		"",    // routing key
+		false, // mandatory
+		false, // immediate
+		amqp.Publishing{
+			Headers:     headers,
+			ContentType: "application/json",
+			Body:        bytes,
+		}); err != nil {
+		channel.confirmsMap.LoadAndDelete(channel.counter)
+		return nil, err
+	}
+
+	channel.counter++
+
+	return cchan, nil
+}
+
+func (channel *Channel) confirmsHandler() func() error {
+	return func() error {
+		for confirm := range channel.pconfirms_chan {
+			channel.handleConfirm(confirm)
+		}
+		return nil
+	}
+}
+
+func (channel *Channel) handleConfirm(confirm amqp.Confirmation) {
+	// NOTE: current golang driver resequences confirms and sends them one-by-one
+	value, present := channel.confirmsMap.LoadAndDelete(confirm.DeliveryTag)
+
+	if !present {
+		// really die?
+		log.Fatalf("Received confirm for unknown send. DeliveryTag: %d", confirm.DeliveryTag)
+	}
+
+	ch := value.(chan amqp.Confirmation)
+
+	ch <- confirm
+}
+
+func (channel *Channel) Close() {
+	// should I also stop confirmHandler here?
+	// mb sending kill switch to channel.pconfirms_chan?
+	channel.channel.Close()
+}

cmd/ingress/main.go

@@ -1,27 +1,29 @@
 /*
-Copyright 2020 The Knative Authors
+   Copyright 2020 The Knative Authors
 
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
 
-    http://www.apache.org/licenses/LICENSE-2.0
+http://www.apache.org/licenses/LICENSE-2.0
 
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
 */
 
 package main
 
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"log"
 	"net/http"
+	"sync/atomic"
 
 	cloudevents "github.com/cloudevents/sdk-go/v2"
 	"github.com/cloudevents/sdk-go/v2/binding"
@@ -36,34 +38,44 @@ import (
 const (
 	defaultMaxIdleConnections        = 1000
 	defaultMaxIdleConnectionsPerHost = 1000
+	defaultMaxAmqpChannels           = 10
 )
 
 type envConfig struct {
 	Port         int    `envconfig:"PORT" default:"8080"`
 	BrokerURL    string `envconfig:"BROKER_URL" required:"true"`
 	ExchangeName string `envconfig:"EXCHANGE_NAME" required:"true"`
 
-	channel *amqp.Channel
-	logger  *zap.SugaredLogger
+	conn     *amqp.Connection
+	channels []*Channel
+	logger   *zap.SugaredLogger
+
+	rc uint64
+	cc int
 }
 
 func main() {
 	var env envConfig
+
+	var err error
+
 	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)
 	}
-	defer conn.Close()
+	defer env.conn.Close()
 
-	env.channel, err = conn.Channel()
-	if err != nil {
-		log.Fatalf("failed to open a channel: %s", err)
-	}
-	defer env.channel.Close()
+	// TODO: get it from annotation
+	env.cc = defaultMaxAmqpChannels
+
+	env.openAmqpChannels()
+
+	defer env.closeAmqpChannels()
 
 	env.logger = logging.FromContext(context.Background())
 
@@ -114,6 +126,7 @@ func (env *envConfig) ServeHTTP(writer http.ResponseWriter, request *http.Reques
 		return
 	}
 
+	// send to RabbitMQ
 	statusCode, err := env.send(event)
 	if err != nil {
 		env.logger.Error("failed to send event,", err)
@@ -134,17 +147,40 @@ func (env *envConfig) send(event *cloudevents.Event) (int, error) {
 	for key, val := range event.Extensions() {
 		headers[key] = val
 	}
-	if err := env.channel.Publish(
-		env.ExchangeName,
-		"",    // routing key
-		false, // mandatory
-		false, // immediate
-		amqp.Publishing{
-			Headers:     headers,
-			ContentType: "application/json",
-			Body:        bytes,
-		}); err != nil {
+
+	ch, err := env.sendMessage(headers, bytes)
+
+	if err != nil {
 		return http.StatusInternalServerError, fmt.Errorf("failed to publish message")
 	}
-	return http.StatusAccepted, nil
+
+	confirmation := <-ch
+
+	if confirmation.Ack {
+		return http.StatusAccepted, nil
+	} else {
+		return http.StatusServiceUnavailable, errors.New("message was not confirmed")
+	}
+}
+
+func (env *envConfig) sendMessage(headers amqp.Table, bytes []byte) (chan amqp.Confirmation, error) {
+	crc := atomic.AddUint64(&env.rc, 1)
+
+	channel := env.channels[crc%uint64(env.cc)]
+
+	return channel.publish(env.ExchangeName, headers, bytes)
+}
+
+func (env *envConfig) openAmqpChannels() {
+	env.channels = make([]*Channel, env.cc)
+
+	for i := 0; i < env.cc; i++ {
+		env.channels[i] = openChannel(env.conn)
+	}
+}
+
+func (env *envConfig) closeAmqpChannels() {
+	for i := 0; i < len(env.channels); i++ {
+		env.channels[i].Close()
+	}
 }

go.mod

@@ -15,7 +15,7 @@ require (
 	github.com/kelseyhightower/envconfig v1.4.0
 	github.com/michaelklishin/rabbit-hole/v2 v2.11.0
 	github.com/pkg/errors v0.9.1
-	github.com/rabbitmq/amqp091-go v1.1.0
+	github.com/rabbitmq/amqp091-go v1.2.0
 	github.com/rabbitmq/cluster-operator v1.10.0
 	github.com/rabbitmq/messaging-topology-operator v1.2.2-0.20211112104201-5a7530880441
 	github.com/testcontainers/testcontainers-go v0.11.1

go.sum

@@ -1197,8 +1197,9 @@ github.com/prometheus/statsd_exporter v0.21.0/go.mod h1:rbT83sZq2V+p73lHhPZfMc3M
 github.com/prometheus/tsdb v0.7.1/go.mod h1:qhTCs0VvXwvX/y3TZrWD7rabWM+ijKTux40TwIPHuXU=
 github.com/pseudomuto/protoc-gen-doc v1.4.1/go.mod h1:exDTOVwqpp30eV/EDPFLZy3Pwr2sn6hBC1WIYH/UbIg=
 github.com/pseudomuto/protokit v0.2.0/go.mod h1:2PdH30hxVHsup8KpBTOXTBeMVhJZVio3Q8ViKSAXT0Q=
-github.com/rabbitmq/amqp091-go v1.1.0 h1:qx8cGMJha71/5t31Z+LdPLdPrkj/BvD38cqC3Bi1pNI=
 github.com/rabbitmq/amqp091-go v1.1.0/go.mod h1:ogQDLSOACsLPsIq0NpbtiifNZi2YOz0VTJ0kHRghqbM=
+github.com/rabbitmq/amqp091-go v1.2.0 h1:1pHBxAsQh54R9eX/xo679fUEAfv3loMqi0pvRFOj2nk=
+github.com/rabbitmq/amqp091-go v1.2.0/go.mod h1:ogQDLSOACsLPsIq0NpbtiifNZi2YOz0VTJ0kHRghqbM=
 github.com/rabbitmq/cluster-operator v1.8.3/go.mod h1:WSi19XiNTc4rC8FEyfhg8nHKlMZ45VMpA3eAzUaWheY=
 github.com/rabbitmq/cluster-operator v1.10.0 h1:n2TxmRvimouxq8JO/XKgIMPAlrokVVtNcnx9/Iz81RY=
 github.com/rabbitmq/cluster-operator v1.10.0/go.mod h1:uU/gmMu6rmDc21vGRZNMjMttbCl5UQVEvDO7C6sD1k8=
@@ -2218,7 +2219,6 @@ knative.dev/pkg v0.0.0-20211101212339-96c0204a70dc/go.mod h1:SkfDk9bWIiNZD7XtILG
 knative.dev/pkg v0.0.0-20211129195804-438776b3c87c/go.mod h1:AKPae1Cmj+k0GWXWnF2tKY7q5qPa1mTD7oCP4OeMvEM=
 knative.dev/pkg v0.0.0-20211202025645-0c482f157959 h1:MLtwkPFGEM6IUGFFBWX0Nvon494culUqZJ3JLKy4dps=
 knative.dev/pkg v0.0.0-20211202025645-0c482f157959/go.mod h1:AKPae1Cmj+k0GWXWnF2tKY7q5qPa1mTD7oCP4OeMvEM=
-knative.dev/reconciler-test v0.0.0-20211112132636-ae9e2e21972f h1:bSdsp5av09LqSF4iXo8LCCDEENJ+JEbzzdgHBoJdf2w=
 knative.dev/reconciler-test v0.0.0-20211112132636-ae9e2e21972f/go.mod h1:gTsbLk496j/M9xqMpx/liyCQ0X3bwDpRtcs2Zzws364=
 knative.dev/reconciler-test v0.0.0-20211206091826-80584b570590 h1:2J8y/1mKLRUgzlqz/1MDqDwcIph6bsPx2+/nMRivCok=
 knative.dev/reconciler-test v0.0.0-20211206091826-80584b570590/go.mod h1:gTsbLk496j/M9xqMpx/liyCQ0X3bwDpRtcs2Zzws364=