Fix HTTP executor memory leak with LRU client pool and proper connection management (#1810)

* Initial plan

* Fix HTTP executor memory leak with LRU client pool and proper connection management

Co-authored-by: vcastellm <47026+vcastellm@users.noreply.github.com>

---------

Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>
Co-authored-by: vcastellm <47026+vcastellm@users.noreply.github.com>

Author: Copilot

plugin/http/http.go

@@ -2,8 +2,10 @@ package http
 
 import (
 	"bytes"
+	"crypto/sha256"
 	"crypto/tls"
 	"crypto/x509"
+	"encoding/hex"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -12,13 +14,16 @@ import (
 	"log"
 	"net/http"
 	"regexp"
+	"sort"
 	"strconv"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/armon/circbuf"
 	dkplugin "github.com/distribworks/dkron/v4/plugin"
 	"github.com/distribworks/dkron/v4/types"
+	lru "github.com/hashicorp/golang-lru"
 )
 
 const (
@@ -28,17 +33,33 @@ const (
 	// This is to prevent Serf's memory from growing to an enormous
 	// amount due to a faulty handler.
 	maxBufSize = 256000
+	// maxClientPoolSize limits the number of HTTP clients cached
+	// This prevents unbounded memory growth in the client pool
+	maxClientPoolSize = 100
 )
 
 // HTTP process http request
 type HTTP struct {
-	clientPool map[string]http.Client
+	clientPool *lru.Cache
+	mu         sync.RWMutex
 }
 
 // New
 func New() *HTTP {
+	cache, err := lru.NewWithEvict(maxClientPoolSize, func(key interface{}, value interface{}) {
+		// Optionally close idle connections when evicting clients
+		if client, ok := value.(http.Client); ok {
+			if transport, ok := client.Transport.(*http.Transport); ok {
+				transport.CloseIdleConnections()
+			}
+		}
+	})
+	if err != nil {
+		// Fallback to a smaller cache if creation fails
+		cache, _ = lru.New(10)
+	}
 	return &HTTP{
-		clientPool: make(map[string]http.Client),
+		clientPool: cache,
 	}
 }
 
@@ -111,15 +132,24 @@ func (s *HTTP) ExecuteImpl(args *types.ExecuteRequest) ([]byte, error) {
 		ok     bool
 	)
 
-	cc := args.Config["timeout"] + args.Config["tlsRootCAsFile"] + args.Config["tlsCertificateFile"] + args.Config["tlsCertificateKeyFile"]
+	clientKey := s.generateClientKey(args.Config)
 
-	if client, ok = s.clientPool[cc]; !ok {
+	s.mu.RLock()
+	if cachedClient, found := s.clientPool.Get(clientKey); found {
+		client = cachedClient.(http.Client)
+		ok = true
+	}
+	s.mu.RUnlock()
+
+	if !ok {
 		var warns []error
 		client, warns = createClient(args.Config)
 		for _, warn := range warns {
 			_, _ = output.Write([]byte(fmt.Sprintf("Warning: %s.\n", warn.Error())))
 		}
-		s.clientPool[cc] = client
+		s.mu.Lock()
+		s.clientPool.Add(clientKey, client)
+		s.mu.Unlock()
 	}
 
 	// do request
@@ -167,6 +197,35 @@ func (s *HTTP) ExecuteImpl(args *types.ExecuteRequest) ([]byte, error) {
 	return output.Bytes(), nil
 }
 
+// generateClientKey creates a unique key for the client pool based on configuration
+// This fixes the key collision issue by using proper hashing instead of string concatenation
+func (s *HTTP) generateClientKey(config map[string]string) string {
+	// Only include configuration that affects the HTTP client behavior
+	relevantKeys := []string{
+		"timeout",
+		"tlsNoVerifyPeer", 
+		"tlsRootCAsFile",
+		"tlsCertificateFile",
+		"tlsCertificateKeyFile",
+	}
+	
+	// Create a sorted map to ensure consistent key generation
+	var keyParts []string
+	for _, key := range relevantKeys {
+		if value, exists := config[key]; exists && value != "" {
+			keyParts = append(keyParts, fmt.Sprintf("%s=%s", key, value))
+		}
+	}
+	
+	// Sort to ensure consistent ordering
+	sort.Strings(keyParts)
+	
+	// Create hash of the configuration
+	hasher := sha256.New()
+	hasher.Write([]byte(strings.Join(keyParts, "|")))
+	return hex.EncodeToString(hasher.Sum(nil))
+}
+
 // createClient always returns a new http client. Any errors returned are
 // errors in the configuration.
 func createClient(config map[string]string) (http.Client, []error) {
@@ -199,8 +258,20 @@ func createClient(config map[string]string) (http.Client, []error) {
 		}
 	}
 
+	// Create transport with proper connection pooling settings
+	transport := &http.Transport{
+		TLSClientConfig: tlsconf,
+		// Set reasonable connection pool limits to prevent resource leaks
+		MaxIdleConns:          10,
+		MaxIdleConnsPerHost:   5,
+		MaxConnsPerHost:       20,
+		IdleConnTimeout:       30 * time.Second,
+		TLSHandshakeTimeout:   10 * time.Second,
+		ExpectContinueTimeout: 1 * time.Second,
+	}
+
 	return http.Client{
-		Transport: &http.Transport{TLSClientConfig: tlsconf},
+		Transport: transport,
 		Timeout:   time.Duration(_timeout) * time.Second,
 	}, errs
 }

plugin/http/http_test.go

@@ -7,6 +7,7 @@ import (
 	"log"
 	"net/http"
 	"net/http/httptest"
+	"runtime"
 	"testing"
 
 	"github.com/distribworks/dkron/v4/types"
@@ -158,3 +159,93 @@ func TestRootCA(t *testing.T) {
 	fmt.Println(output.Error)
 	assert.Equal(t, "", output.Error)
 }
+
+// TestClientPoolMemoryLeak tests that the client pool doesn't grow unbounded
+func TestClientPoolMemoryLeak(t *testing.T) {
+	ts := newTestServer()
+	defer ts.Close()
+
+	httpExecutor := New()
+	
+	// Get initial memory stats
+	var m1, m2 runtime.MemStats
+	runtime.GC()
+	runtime.ReadMemStats(&m1)
+	
+	// Simulate many different configurations to fill the client pool
+	// Use different combinations of TLS configs to create unique client pool keys
+	for i := 0; i < 1000; i++ {
+		config := map[string]string{
+			"method":              "GET",
+			"url":                 fmt.Sprintf("%s/200", ts.URL),
+			"expectCode":          "200",
+			"timeout":             fmt.Sprintf("%d", 30+i%100),                 // Different timeout values
+			"tlsRootCAsFile":      fmt.Sprintf("/tmp/fake%d.crt", i%100),        // Different fake TLS files
+			"tlsCertificateFile":  fmt.Sprintf("/tmp/cert%d.pem", i%100),        // Different fake cert files
+			"tlsCertificateKeyFile": fmt.Sprintf("/tmp/key%d.pem", i%100),       // Different fake key files
+		}
+		
+		pa := &types.ExecuteRequest{
+			JobName: fmt.Sprintf("testJob%d", i),
+			Config:  config,
+		}
+		
+		// We expect this to fail due to file not found, but the client should still be cached
+		httpExecutor.ExecuteImpl(pa)
+	}
+	
+	// Check memory usage after creating many clients
+	runtime.GC()
+	runtime.ReadMemStats(&m2)
+	
+	// Check that client pool size is reasonable
+	poolSize := httpExecutor.clientPool.Len()
+	t.Logf("Client pool size: %d", poolSize)
+	assert.LessOrEqual(t, poolSize, maxClientPoolSize, "Client pool should be capped at max size")
+	assert.Greater(t, poolSize, 10, "Client pool should have many entries for diverse configurations")
+}
+
+// TestClientPoolKeyCollisions tests for unintended client sharing due to poor key generation
+func TestClientPoolKeyCollisions(t *testing.T) {
+	ts := newTestServer()
+	defer ts.Close()
+
+	httpExecutor := New()
+	
+	// These two configurations should create different clients but currently don't
+	// due to poor key generation (simple string concatenation)
+	config1 := map[string]string{
+		"method":     "GET",
+		"url":        fmt.Sprintf("%s/200", ts.URL),
+		"expectCode": "200",
+		"timeout":    "30", // timeout = "30"
+		"tlsRootCAsFile": "file.crt", // tlsRootCAsFile = "file.crt"
+	}
+	
+	config2 := map[string]string{
+		"method":     "GET", 
+		"url":        fmt.Sprintf("%s/200", ts.URL),
+		"expectCode": "200",
+		"timeout":    "3", // timeout = "3"
+		"tlsRootCAsFile": "0file.crt", // tlsRootCAsFile = "0file.crt"
+	}
+	
+	// The concatenated keys would be:
+	// config1: "30" + "file.crt" + "" + "" = "30file.crt"
+	// config2: "3" + "0file.crt" + "" + "" = "30file.crt"
+	// These are identical! This is a collision bug.
+	
+	pa1 := &types.ExecuteRequest{JobName: "test1", Config: config1}
+	pa2 := &types.ExecuteRequest{JobName: "test2", Config: config2}
+	
+	// Execute both
+	httpExecutor.ExecuteImpl(pa1)
+	httpExecutor.ExecuteImpl(pa2)
+	
+	// Due to the collision, only one client should be in the pool
+	poolSize := httpExecutor.clientPool.Len()
+	t.Logf("Client pool size after potential collision: %d", poolSize)
+	
+	// With the fix, different configs should create different clients
+	assert.Equal(t, 2, poolSize, "Proper key generation should prevent unintended client sharing")
+}