kncloudevents now uses a singleton http client (#4465)

* Fix #4461

Now every new message sender always reuse the same underlying client, whenever possible

Signed-off-by: Francesco Guardiani <francescoguard@gmail.com>

* Increase coverage

Signed-off-by: Francesco Guardiani <francescoguard@gmail.com>

* Brought back the previous method to avoid breakage

Signed-off-by: Francesco Guardiani <francescoguard@gmail.com>

* Now we use nice language

Signed-off-by: Francesco Guardiani <francescoguard@gmail.com>

* Removed useless test

Signed-off-by: Francesco Guardiani <francescoguard@gmail.com>

* Suggestions

Signed-off-by: Francesco Guardiani <francescoguard@gmail.com>

* Imports job

Signed-off-by: Francesco Guardiani <francescoguard@gmail.com>

* nit

Signed-off-by: Francesco Guardiani <francescoguard@gmail.com>

* Fancy ut

Signed-off-by: Francesco Guardiani <francescoguard@gmail.com>

* Copyright

Signed-off-by: Francesco Guardiani <francescoguard@gmail.com>

cmd/mtbroker/ingress/main.go

@@ -128,7 +128,8 @@ func main() {
 		MaxIdleConns:        defaultMaxIdleConnections,
 		MaxIdleConnsPerHost: defaultMaxIdleConnectionsPerHost,
 	}
-	sender, err := kncloudevents.NewHTTPMessageSender(&connectionArgs, "")
+	kncloudevents.ConfigureConnectionArgs(&connectionArgs)
+	sender, err := kncloudevents.NewHTTPMessageSenderWithTarget("")
 	if err != nil {
 		logger.Fatal("Unable to create message sender", zap.Error(err))
 	}

pkg/adapter/v2/main_message.go

@@ -108,7 +108,7 @@ func MainMessageAdapterWithContext(ctx context.Context, component string, ector
 		logger.Error("Error setting up trace publishing", zap.Error(err))
 	}
 
-	httpBindingsSender, err := kncloudevents.NewHTTPMessageSender(nil, env.GetSink())
+	httpBindingsSender, err := kncloudevents.NewHTTPMessageSenderWithTarget(env.GetSink())
 	if err != nil {
 		logger.Fatal("error building cloud event client", zap.Error(err))
 	}

pkg/channel/message_dispatcher.go

@@ -68,15 +68,9 @@ type DispatchExecutionInfo struct {
 	ResponseCode int
 }
 
-// NewMessageDispatcher creates a new Message dispatcher that can dispatch
-// events to HTTP destinations.
-func NewMessageDispatcher(logger *zap.Logger) *MessageDispatcherImpl {
-	return NewMessageDispatcherFromConfig(logger, defaultEventDispatcherConfig)
-}
-
 // NewMessageDispatcherFromConfig creates a new Message dispatcher based on config.
-func NewMessageDispatcherFromConfig(logger *zap.Logger, config EventDispatcherConfig) *MessageDispatcherImpl {
-	sender, err := kncloudevents.NewHTTPMessageSender(&config.ConnectionArgs, "")
+func NewMessageDispatcher(logger *zap.Logger) *MessageDispatcherImpl {
+	sender, err := kncloudevents.NewHTTPMessageSenderWithTarget("")
 	if err != nil {
 		logger.Fatal("Unable to create cloudevents binding sender", zap.Error(err))
 	}

pkg/inmemorychannel/message_dispatcher_benchmark_test.go

@@ -76,7 +76,7 @@ func BenchmarkDispatcher_dispatch_ok_through_2_channels(b *testing.B) {
 	}
 
 	// Let's mock this stuff!
-	httpSender, err := kncloudevents.NewHTTPMessageSender(nil, channelAUrl.String())
+	httpSender, err := kncloudevents.NewHTTPMessageSenderWithTarget(channelAUrl.String())
 	if err != nil {
 		b.Fatal(err)
 	}

pkg/inmemorychannel/message_dispatcher_test.go

@@ -256,7 +256,7 @@ func TestDispatcher_dispatch(t *testing.T) {
 	}()
 
 	// Ok now everything should be ready to send the event
-	httpsender, err := kncloudevents.NewHTTPMessageSender(nil, channelAProxy.URL)
+	httpsender, err := kncloudevents.NewHTTPMessageSenderWithTarget(channelAProxy.URL)
 	if err != nil {
 		t.Fatal(err)
 	}

pkg/kncloudevents/http_client.go

@@ -0,0 +1,106 @@
+/*
+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
+
+    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 kncloudevents
+
+import (
+	nethttp "net/http"
+	"sync"
+	"time"
+
+	"go.opencensus.io/plugin/ochttp"
+	"knative.dev/pkg/tracing/propagation/tracecontextb3"
+)
+
+const (
+	defaultRetryWaitMin = 1 * time.Second
+	defaultRetryWaitMax = 30 * time.Second
+)
+
+type holder struct {
+	clientMutex    sync.Mutex
+	connectionArgs *ConnectionArgs
+	client         **nethttp.Client
+}
+
+var clientHolder = holder{}
+
+// The used HTTP client is a singleton, so the same http client is reused across all the application.
+// If connection args is modified, client is cleaned and a new one is created.
+func getClient() *nethttp.Client {
+	clientHolder.clientMutex.Lock()
+	defer clientHolder.clientMutex.Unlock()
+
+	if clientHolder.client == nil {
+		// Add connection options to the default transport.
+		var base = nethttp.DefaultTransport.(*nethttp.Transport).Clone()
+		clientHolder.connectionArgs.configureTransport(base)
+		c := &nethttp.Client{
+			// Add output tracing.
+			Transport: &ochttp.Transport{
+				Base:        base,
+				Propagation: tracecontextb3.TraceContextEgress,
+			},
+		}
+		clientHolder.client = &c
+	}
+
+	return *clientHolder.client
+}
+
+// ConfigureConnectionArgs configures the new connection args.
+// The existing client won't be affected, but a new one will be created.
+// Use sparingly, because it might lead to creating a lot of clients, none of them sharing their connection pool!
+func ConfigureConnectionArgs(ca *ConnectionArgs) {
+	clientHolder.clientMutex.Lock()
+	defer clientHolder.clientMutex.Unlock()
+
+	// Check if same config
+	if clientHolder.connectionArgs != nil &&
+		ca != nil &&
+		ca.MaxIdleConns == clientHolder.connectionArgs.MaxIdleConns &&
+		ca.MaxIdleConnsPerHost == clientHolder.connectionArgs.MaxIdleConnsPerHost {
+		return
+	}
+
+	if clientHolder.client != nil {
+		// Let's try to clean up a bit the existing client
+		// Note: this won't remove it nor close it
+		(*clientHolder.client).CloseIdleConnections()
+
+		// Setting client to nil
+		clientHolder.client = nil
+	}
+
+	clientHolder.connectionArgs = ca
+}
+
+// ConnectionArgs allow to configure connection parameters to the underlying
+// HTTP Client transport.
+type ConnectionArgs struct {
+	// MaxIdleConns refers to the max idle connections, as in net/http/transport.
+	MaxIdleConns int
+	// MaxIdleConnsPerHost refers to the max idle connections per host, as in net/http/transport.
+	MaxIdleConnsPerHost int
+}
+
+func (ca *ConnectionArgs) configureTransport(transport *nethttp.Transport) {
+	if ca == nil {
+		return
+	}
+	transport.MaxIdleConns = ca.MaxIdleConns
+	transport.MaxIdleConnsPerHost = ca.MaxIdleConnsPerHost
+}

pkg/kncloudevents/http_client_test.go

@@ -0,0 +1,72 @@
+/*
+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
+
+    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 kncloudevents
+
+import (
+	nethttp "net/http"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"go.opencensus.io/plugin/ochttp"
+)
+
+func TestConfigureConnectionArgs(t *testing.T) {
+	// Set connection args
+	ConfigureConnectionArgs(&ConnectionArgs{
+		MaxIdleConnsPerHost: 1000,
+		MaxIdleConns:        1000,
+	})
+	client1 := getClient()
+
+	require.Same(t, getClient(), client1)
+	require.Equal(t, 1000, castToTransport(client1).MaxIdleConns)
+	require.Equal(t, 1000, castToTransport(client1).MaxIdleConnsPerHost)
+
+	// Set other connection args
+	ConfigureConnectionArgs(&ConnectionArgs{
+		MaxIdleConnsPerHost: 2000,
+		MaxIdleConns:        2000,
+	})
+	client2 := getClient()
+
+	require.Same(t, getClient(), client2)
+	require.Equal(t, 2000, castToTransport(client2).MaxIdleConns)
+	require.Equal(t, 2000, castToTransport(client2).MaxIdleConnsPerHost)
+
+	// Try to set the same value and client should not be cleaned up
+	ConfigureConnectionArgs(&ConnectionArgs{
+		MaxIdleConnsPerHost: 2000,
+		MaxIdleConns:        2000,
+	})
+	require.Same(t, getClient(), client2)
+
+	// Set back to nil
+	ConfigureConnectionArgs(nil)
+	client3 := getClient()
+
+	require.Same(t, getClient(), client3)
+	require.Equal(t, nethttp.DefaultTransport.(*nethttp.Transport).MaxIdleConns, castToTransport(client3).MaxIdleConns)
+	require.Equal(t, nethttp.DefaultTransport.(*nethttp.Transport).MaxIdleConnsPerHost, castToTransport(client3).MaxIdleConnsPerHost)
+
+	require.NotSame(t, client1, client2)
+	require.NotSame(t, client1, client3)
+	require.NotSame(t, client2, client3)
+}
+
+func castToTransport(client *nethttp.Client) *nethttp.Transport {
+	return client.Transport.(*ochttp.Transport).Base.(*nethttp.Transport)
+}

pkg/kncloudevents/message_sender.go

@@ -25,17 +25,10 @@ import (
 
 	"github.com/hashicorp/go-retryablehttp"
 	"github.com/rickb777/date/period"
-	"go.opencensus.io/plugin/ochttp"
-	"knative.dev/pkg/tracing/propagation/tracecontextb3"
 
 	duckv1 "knative.dev/eventing/pkg/apis/duck/v1"
 )
 
-const (
-	defaultRetryWaitMin = 1 * time.Second
-	defaultRetryWaitMax = 30 * time.Second
-)
-
 var noRetries = RetryConfig{
 	RetryMax: 0,
 	CheckRetry: func(ctx context.Context, resp *nethttp.Response, err error) (bool, error) {
@@ -46,41 +39,19 @@ var noRetries = RetryConfig{
 	},
 }
 
-// ConnectionArgs allow to configure connection parameters to the underlying
-// HTTP Client transport.
-type ConnectionArgs struct {
-	// MaxIdleConns refers to the max idle connections, as in net/http/transport.
-	MaxIdleConns int
-	// MaxIdleConnsPerHost refers to the max idle connections per host, as in net/http/transport.
-	MaxIdleConnsPerHost int
-}
-
-func (ca *ConnectionArgs) ConfigureTransport(transport *nethttp.Transport) {
-	if ca == nil {
-		return
-	}
-	transport.MaxIdleConns = ca.MaxIdleConns
-	transport.MaxIdleConnsPerHost = ca.MaxIdleConnsPerHost
-}
-
 type HTTPMessageSender struct {
 	Client *nethttp.Client
 	Target string
 }
 
-func NewHTTPMessageSender(connectionArgs *ConnectionArgs, target string) (*HTTPMessageSender, error) {
-	// Add connection options to the default transport.
-	var base = nethttp.DefaultTransport.(*nethttp.Transport).Clone()
-	connectionArgs.ConfigureTransport(base)
-	// Add output tracing.
-	client := &nethttp.Client{
-		Transport: &ochttp.Transport{
-			Base:        base,
-			Propagation: tracecontextb3.TraceContextEgress,
-		},
-	}
+// Deprecated: Don't use this anymore, now it has the same effect of NewHTTPMessageSenderWithTarget
+// If you need to modify the connection args, use ConfigureConnectionArgs sparingly.
+func NewHTTPMessageSender(ca *ConnectionArgs, target string) (*HTTPMessageSender, error) {
+	return NewHTTPMessageSenderWithTarget(target)
+}
 
-	return &HTTPMessageSender{Client: client, Target: target}, nil
+func NewHTTPMessageSenderWithTarget(target string) (*HTTPMessageSender, error) {
+	return &HTTPMessageSender{Client: getClient(), Target: target}, nil
 }
 
 func (s *HTTPMessageSender) NewCloudEventRequest(ctx context.Context) (*nethttp.Request, error) {

pkg/kncloudevents/message_sender_test.go

@@ -248,7 +248,7 @@ func TestRetriesOnNetworkErrors(t *testing.T) {
 	req, err := http.NewRequest("POST", "http://"+target, nil)
 	assert.Nil(t, err)
 
-	sender, err := NewHTTPMessageSender(nil, "")
+	sender, err := NewHTTPMessageSenderWithTarget("")
 	assert.Nil(t, err)
 
 	_, err = sender.SendWithRetries(req, &r)

pkg/mtbroker/filter/filter_handler.go

@@ -75,13 +75,12 @@ type FilterResult string
 // NewHandler creates a new Handler and its associated MessageReceiver. The caller is responsible for
 // Start()ing the returned Handler.
 func NewHandler(logger *zap.Logger, triggerLister eventinglisters.TriggerLister, reporter StatsReporter, port int) (*Handler, error) {
-
-	connectionArgs := kncloudevents.ConnectionArgs{
+	kncloudevents.ConfigureConnectionArgs(&kncloudevents.ConnectionArgs{
 		MaxIdleConns:        defaultMaxIdleConnections,
 		MaxIdleConnsPerHost: defaultMaxIdleConnectionsPerHost,
-	}
+	})
 
-	sender, err := kncloudevents.NewHTTPMessageSender(&connectionArgs, "")
+	sender, err := kncloudevents.NewHTTPMessageSenderWithTarget("")
 	if err != nil {
 		return nil, fmt.Errorf("failed to create message sender: %w", err)
 	}

pkg/mtbroker/ingress/ingress_handler_test.go

@@ -24,17 +24,17 @@ import (
 	"testing"
 	"time"
 
-	"knative.dev/eventing/pkg/kncloudevents"
-
 	"github.com/cloudevents/sdk-go/v2/client"
 	"github.com/cloudevents/sdk-go/v2/event"
 	cehttp "github.com/cloudevents/sdk-go/v2/protocol/http"
 	"github.com/google/go-cmp/cmp"
 	"go.uber.org/zap"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
+
 	"knative.dev/eventing/pkg/apis/eventing"
 	eventingv1 "knative.dev/eventing/pkg/apis/eventing/v1"
+	"knative.dev/eventing/pkg/kncloudevents"
 	broker "knative.dev/eventing/pkg/mtbroker"
 	reconcilertestingv1 "knative.dev/eventing/pkg/reconciler/testing/v1"
 )
@@ -213,7 +213,7 @@ func TestHandler_ServeHTTP(t *testing.T) {
 				annotatedBrokers = append(annotatedBrokers, b)
 			}
 			listers := reconcilertestingv1.NewListers(annotatedBrokers)
-			sender, _ := kncloudevents.NewHTTPMessageSender(nil, "")
+			sender, _ := kncloudevents.NewHTTPMessageSenderWithTarget("")
 			h := &Handler{
 				Sender:       sender,
 				Defaulter:    tc.defaulter,

pkg/reconciler/inmemorychannel/dispatcher/controller.go

@@ -21,7 +21,10 @@ import (
 	"time"
 
 	"k8s.io/client-go/tools/cache"
+
 	"knative.dev/eventing/pkg/channel/multichannelfanout"
+	"knative.dev/eventing/pkg/kncloudevents"
+
 	"knative.dev/pkg/injection"
 
 	"github.com/google/uuid"
@@ -103,13 +106,15 @@ func NewController(
 		return controller.Options{SkipStatusUpdates: true, FinalizerName: finalizerName}
 	})
 
-	// Nothing to filter, enqueue all imcs if configmap updates.
-	noopFilter := func(interface{}) bool { return true }
-	resyncIMCs := configmap.TypeFilter(channel.EventDispatcherConfig{})(func(string, interface{}) {
-		impl.FilteredGlobalResync(noopFilter, informer)
+	globalSyncAfterDispatcherConfigUpdate := configmap.TypeFilter(channel.EventDispatcherConfig{})(func(key string, val interface{}) {
+		conf := val.(channel.EventDispatcherConfig)
+		kncloudevents.ConfigureConnectionArgs(&conf.ConnectionArgs)
+
+		// Nothing to filter, enqueue all imcs if configmap updates.
+		impl.FilteredGlobalResync(func(interface{}) bool { return true }, informer)
 	})
 	// Watch for configmap changes and trigger imc reconciliation by enqueuing imcs.
-	configStore := channel.NewEventDispatcherConfigStore(logging.FromContext(ctx), resyncIMCs)
+	configStore := channel.NewEventDispatcherConfigStore(logging.FromContext(ctx), globalSyncAfterDispatcherConfigUpdate)
 	configStore.WatchConfigs(cmw)
 	r.eventDispatcherConfigStore = configStore