Merge pull request #26 from jkyros/http-fix-memory-leak

OCPBUGS-30145: fix: Prevented memory leak generated by not correctly cleaned HTTP resources (kedacore#5293)
openshift · Mar 27, 2024 · dc4c94b · dc4c94b
2 parents b493640 + cd2197d
commit dc4c94b
Show file tree

Hide file tree

Showing 27 changed files with 91 additions and 27 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -91,6 +91,7 @@ New deprecation(s):
 - **General**: Fix CVE-2023-45142 in Opentelemetry ([#5089](https://github.com/kedacore/keda/issues/5089))
 - **General**: Fix logger in Opentelemetry collector ([#5094](https://github.com/kedacore/keda/issues/5094))
 - **General**: Fix otelgrpc DoS vulnerability ([#5208](https://github.com/kedacore/keda/issues/5208))
+- **General**: Prevented memory leak generated by not correctly cleaning http connections ([#5248](https://github.com/kedacore/keda/issues/5248))
 - **General**: Prevented stuck status due to timeouts during scalers generation ([#5083](https://github.com/kedacore/keda/issues/5083))
 - **Azure Pipelines**: No more HTTP 400 errors produced by poolName with spaces ([#5107](https://github.com/kedacore/keda/issues/5107))
 

diff --git a/pkg/scalers/activemq_scaler.go b/pkg/scalers/activemq_scaler.go
@@ -280,5 +280,8 @@ func (s *activeMQScaler) GetMetricsAndActivity(ctx context.Context, metricName s
 }
 
 func (s *activeMQScaler) Close(context.Context) error {
+	if s.httpClient != nil {
+		s.httpClient.CloseIdleConnections()
+	}
 	return nil
 }
diff --git a/pkg/scalers/artemis_scaler.go b/pkg/scalers/artemis_scaler.go
@@ -279,7 +279,9 @@ func (s *artemisScaler) GetMetricsAndActivity(ctx context.Context, metricName st
 	return []external_metrics.ExternalMetricValue{metric}, messages > s.metadata.activationQueueLength, nil
 }
 
-// Nothing to close here.
 func (s *artemisScaler) Close(context.Context) error {
+	if s.httpClient != nil {
+		s.httpClient.CloseIdleConnections()
+	}
 	return nil
 }
diff --git a/pkg/scalers/azure_blob_scaler.go b/pkg/scalers/azure_blob_scaler.go
@@ -174,6 +174,9 @@ func parseAzureBlobMetadata(config *ScalerConfig, logger logr.Logger) (*azure.Bl
 }
 
 func (s *azureBlobScaler) Close(context.Context) error {
+	if s.httpClient != nil {
+		s.httpClient.CloseIdleConnections()
+	}
 	return nil
 }
 

diff --git a/pkg/scalers/azure_data_explorer_scaler.go b/pkg/scalers/azure_data_explorer_scaler.go
@@ -200,5 +200,8 @@ func (s azureDataExplorerScaler) GetMetricSpecForScaling(context.Context) []v2.M
 }
 
 func (s azureDataExplorerScaler) Close(context.Context) error {
+	if s.client != nil && s.client.HttpClient() != nil {
+		s.client.HttpClient().CloseIdleConnections()
+	}
 	return nil
 }
diff --git a/pkg/scalers/azure_log_analytics_scaler.go b/pkg/scalers/azure_log_analytics_scaler.go
@@ -279,6 +279,9 @@ func (s *azureLogAnalyticsScaler) GetMetricsAndActivity(ctx context.Context, met
 }
 
 func (s *azureLogAnalyticsScaler) Close(context.Context) error {
+	if s.httpClient != nil {
+		s.httpClient.CloseIdleConnections()
+	}
 	return nil
 }
 

diff --git a/pkg/scalers/azure_pipelines_scaler.go b/pkg/scalers/azure_pipelines_scaler.go
@@ -458,5 +458,8 @@ func (s *azurePipelinesScaler) GetMetricsAndActivity(ctx context.Context, metric
 }
 
 func (s *azurePipelinesScaler) Close(context.Context) error {
+	if s.httpClient != nil {
+		s.httpClient.CloseIdleConnections()
+	}
 	return nil
 }
diff --git a/pkg/scalers/azure_queue_scaler.go b/pkg/scalers/azure_queue_scaler.go
@@ -159,6 +159,9 @@ func parseAzureQueueMetadata(config *ScalerConfig, logger logr.Logger) (*azureQu
 }
 
 func (s *azureQueueScaler) Close(context.Context) error {
+	if s.httpClient != nil {
+		s.httpClient.CloseIdleConnections()
+	}
 	return nil
 }
 

diff --git a/pkg/scalers/cassandra_scaler.go b/pkg/scalers/cassandra_scaler.go
@@ -222,7 +222,8 @@ func (s *cassandraScaler) GetQueryResult(ctx context.Context) (int64, error) {
 
 // Close closes the Cassandra session connection.
 func (s *cassandraScaler) Close(_ context.Context) error {
-	s.session.Close()
-
+	if s.session != nil {
+		s.session.Close()
+	}
 	return nil
 }
diff --git a/pkg/scalers/datadog_scaler.go b/pkg/scalers/datadog_scaler.go
@@ -263,6 +263,9 @@ func newDatadogConnection(ctx context.Context, meta *datadogMetadata, config *Sc
 
 // No need to close connections
 func (s *datadogScaler) Close(context.Context) error {
+	if s.apiClient != nil {
+		s.apiClient.GetConfig().HTTPClient.CloseIdleConnections()
+	}
 	return nil
 }
 

diff --git a/pkg/scalers/github_runner_scaler.go b/pkg/scalers/github_runner_scaler.go
@@ -670,5 +670,8 @@ func (s *githubRunnerScaler) GetMetricSpecForScaling(_ context.Context) []v2.Met
 }
 
 func (s *githubRunnerScaler) Close(_ context.Context) error {
+	if s.httpClient != nil {
+		s.httpClient.CloseIdleConnections()
+	}
 	return nil
 }
diff --git a/pkg/scalers/graphite_scaler.go b/pkg/scalers/graphite_scaler.go
@@ -141,6 +141,9 @@ func parseGraphiteMetadata(config *ScalerConfig) (*graphiteMetadata, error) {
 }
 
 func (s *graphiteScaler) Close(context.Context) error {
+	if s.httpClient != nil {
+		s.httpClient.CloseIdleConnections()
+	}
 	return nil
 }
 

diff --git a/pkg/scalers/loki_scaler.go b/pkg/scalers/loki_scaler.go
@@ -165,6 +165,9 @@ func parseLokiMetadata(config *ScalerConfig) (meta *lokiMetadata, err error) {
 
 // Close returns a nil error
 func (s *lokiScaler) Close(context.Context) error {
+	if s.httpClient != nil {
+		s.httpClient.CloseIdleConnections()
+	}
 	return nil
 }
 

diff --git a/pkg/scalers/metrics_api_scaler.go b/pkg/scalers/metrics_api_scaler.go
@@ -23,7 +23,7 @@ import (
 type metricsAPIScaler struct {
 	metricType v2.MetricTargetType
 	metadata   *metricsAPIScalerMetadata
-	client     *http.Client
+	httpClient *http.Client
 	logger     logr.Logger
 }
 
@@ -89,7 +89,7 @@ func NewMetricsAPIScaler(config *ScalerConfig) (Scaler, error) {
 	return &metricsAPIScaler{
 		metricType: metricType,
 		metadata:   meta,
-		client:     httpClient,
+		httpClient: httpClient,
 		logger:     InitializeLogger(config, "metrics_api_scaler"),
 	}, nil
 }
@@ -233,7 +233,7 @@ func (s *metricsAPIScaler) getMetricValue(ctx context.Context) (float64, error)
 		return 0, err
 	}
 
-	r, err := s.client.Do(request)
+	r, err := s.httpClient.Do(request)
 	if err != nil {
 		return 0, err
 	}
@@ -257,6 +257,9 @@ func (s *metricsAPIScaler) getMetricValue(ctx context.Context) (float64, error)
 
 // Close does nothing in case of metricsAPIScaler
 func (s *metricsAPIScaler) Close(context.Context) error {
+	if s.httpClient != nil {
+		s.httpClient.CloseIdleConnections()
+	}
 	return nil
 }
 

diff --git a/pkg/scalers/metrics_api_scaler_test.go b/pkg/scalers/metrics_api_scaler_test.go
@@ -243,8 +243,8 @@ func TestGetMetricValueErrorMessage(t *testing.T) {
 
 	httpClient := http.Client{Transport: &mockHTTPRoundTripper}
 	s := metricsAPIScaler{
-		metadata: &metricsAPIScalerMetadata{url: "http://dummy:1230/api/v1/"},
-		client:   &httpClient,
+		metadata:   &metricsAPIScalerMetadata{url: "http://dummy:1230/api/v1/"},
+		httpClient: &httpClient,
 	}
 
 	_, err := s.getMetricValue(context.TODO())

diff --git a/pkg/scalers/nats_jetstream_scaler.go b/pkg/scalers/nats_jetstream_scaler.go
@@ -486,5 +486,8 @@ func (s *natsJetStreamScaler) GetMetricsAndActivity(ctx context.Context, metricN
 }
 
 func (s *natsJetStreamScaler) Close(context.Context) error {
+	if s.httpClient != nil {
+		s.httpClient.CloseIdleConnections()
+	}
 	return nil
 }
diff --git a/pkg/scalers/openstack_metrics_scaler.go b/pkg/scalers/openstack_metrics_scaler.go
@@ -246,6 +246,7 @@ func (s *openstackMetricScaler) GetMetricsAndActivity(ctx context.Context, metri
 }
 
 func (s *openstackMetricScaler) Close(context.Context) error {
+	s.metricClient.HTTPClient.CloseIdleConnections()
 	return nil
 }
 

diff --git a/pkg/scalers/openstack_swift_scaler.go b/pkg/scalers/openstack_swift_scaler.go
@@ -54,7 +54,7 @@ type openstackSwiftAuthenticationMetadata struct {
 type openstackSwiftScaler struct {
 	metricType  v2.MetricTargetType
 	metadata    *openstackSwiftMetadata
-	swiftClient openstack.Client
+	swiftClient *openstack.Client
 	logger      logr.Logger
 }
 
@@ -241,7 +241,7 @@ func NewOpenstackSwiftScaler(ctx context.Context, config *ScalerConfig) (Scaler,
 	return &openstackSwiftScaler{
 		metricType:  metricType,
 		metadata:    openstackSwiftMetadata,
-		swiftClient: swiftClient,
+		swiftClient: &swiftClient,
 		logger:      logger,
 	}, nil
 }
@@ -369,6 +369,9 @@ func parseOpenstackSwiftAuthenticationMetadata(config *ScalerConfig) (*openstack
 }
 
 func (s *openstackSwiftScaler) Close(context.Context) error {
+	if s.swiftClient != nil {
+		s.swiftClient.HTTPClient.CloseIdleConnections()
+	}
 	return nil
 }
 

diff --git a/pkg/scalers/openstack_swift_scaler_test.go b/pkg/scalers/openstack_swift_scaler_test.go
@@ -113,7 +113,7 @@ func TestOpenstackSwiftGetMetricSpecForScaling(t *testing.T) {
 			t.Fatal("Could not parse auth metadata:", err)
 		}
 
-		mockSwiftScaler := openstackSwiftScaler{"", meta, openstack.Client{}, logr.Discard()}
+		mockSwiftScaler := openstackSwiftScaler{"", meta, &openstack.Client{}, logr.Discard()}
 
 		metricSpec := mockSwiftScaler.GetMetricSpecForScaling(context.Background())
 

diff --git a/pkg/scalers/prometheus_scaler.go b/pkg/scalers/prometheus_scaler.go
@@ -239,6 +239,9 @@ func parseAuthConfig(config *ScalerConfig, meta *prometheusMetadata) error {
 }
 
 func (s *prometheusScaler) Close(context.Context) error {
+	if s.httpClient != nil {
+		s.httpClient.CloseIdleConnections()
+	}
 	return nil
 }
 

diff --git a/pkg/scalers/pulsar_scaler.go b/pkg/scalers/pulsar_scaler.go
@@ -22,9 +22,9 @@ import (
 )
 
 type pulsarScaler struct {
-	metadata pulsarMetadata
-	client   *http.Client
-	logger   logr.Logger
+	metadata   pulsarMetadata
+	httpClient *http.Client
+	logger     logr.Logger
 }
 
 type pulsarMetadata struct {
@@ -126,9 +126,9 @@ func NewPulsarScaler(config *ScalerConfig) (Scaler, error) {
 	}
 
 	return &pulsarScaler{
-		client:   client,
-		metadata: pulsarMetadata,
-		logger:   logger,
+		httpClient: client,
+		metadata:   pulsarMetadata,
+		logger:     logger,
 	}, nil
 }
 
@@ -243,8 +243,8 @@ func (s *pulsarScaler) GetStats(ctx context.Context) (*pulsarStats, error) {
 		return nil, fmt.Errorf("error requesting stats from admin url: %w", err)
 	}
 
-	client := s.client
-	if s.metadata.pulsarAuth.EnableOAuth {
+	client := s.httpClient
+	if s.metadata.pulsarAuth != nil && s.metadata.pulsarAuth.EnableOAuth {
 		config := clientcredentials.Config{
 			ClientID:     s.metadata.pulsarAuth.ClientID,
 			ClientSecret: s.metadata.pulsarAuth.ClientSecret,
@@ -331,7 +331,9 @@ func (s *pulsarScaler) GetMetricSpecForScaling(context.Context) []v2.MetricSpec
 }
 
 func (s *pulsarScaler) Close(context.Context) error {
-	s.client = nil
+	if s.httpClient != nil {
+		s.httpClient.CloseIdleConnections()
+	}
 	return nil
 }
 

diff --git a/pkg/scalers/rabbitmq_scaler.go b/pkg/scalers/rabbitmq_scaler.go
@@ -469,6 +469,9 @@ func (s *rabbitMQScaler) Close(context.Context) error {
 			return err
 		}
 	}
+	if s.httpClient != nil {
+		s.httpClient.CloseIdleConnections()
+	}
 	return nil
 }
 

diff --git a/pkg/scalers/selenium_grid_scaler.go b/pkg/scalers/selenium_grid_scaler.go
@@ -22,7 +22,7 @@ import (
 type seleniumGridScaler struct {
 	metricType v2.MetricTargetType
 	metadata   *seleniumGridScalerMetadata
-	client     *http.Client
+	httpClient *http.Client
 	logger     logr.Logger
 }
 
@@ -93,7 +93,7 @@ func NewSeleniumGridScaler(config *ScalerConfig) (Scaler, error) {
 	return &seleniumGridScaler{
 		metricType: metricType,
 		metadata:   meta,
-		client:     httpClient,
+		httpClient: httpClient,
 		logger:     logger,
 	}, nil
 }
@@ -158,6 +158,9 @@ func parseSeleniumGridScalerMetadata(config *ScalerConfig) (*seleniumGridScalerM
 
 // No cleanup required for selenium grid scaler
 func (s *seleniumGridScaler) Close(context.Context) error {
+	if s.httpClient != nil {
+		s.httpClient.CloseIdleConnections()
+	}
 	return nil
 }
 
@@ -200,7 +203,7 @@ func (s *seleniumGridScaler) getSessionsCount(ctx context.Context, logger logr.L
 		return -1, err
 	}
 
-	res, err := s.client.Do(req)
+	res, err := s.httpClient.Do(req)
 	if err != nil {
 		return -1, err
 	}

diff --git a/pkg/scalers/solace_scaler.go b/pkg/scalers/solace_scaler.go
@@ -436,5 +436,8 @@ func (s *SolaceScaler) GetMetricsAndActivity(ctx context.Context, metricName str
 
 // Do Nothing - Satisfies Interface
 func (s *SolaceScaler) Close(context.Context) error {
+	if s.httpClient != nil {
+		s.httpClient.CloseIdleConnections()
+	}
 	return nil
 }
diff --git a/pkg/scalers/solr_scaler.go b/pkg/scalers/solr_scaler.go
@@ -186,5 +186,8 @@ func (s *solrScaler) GetMetricsAndActivity(ctx context.Context, metricName strin
 
 // Close closes the http client connection.
 func (s *solrScaler) Close(context.Context) error {
+	if s.httpClient != nil {
+		s.httpClient.CloseIdleConnections()
+	}
 	return nil
 }
diff --git a/pkg/scalers/stan_scaler.go b/pkg/scalers/stan_scaler.go
@@ -248,5 +248,8 @@ func (s *stanScaler) GetMetricsAndActivity(ctx context.Context, metricName strin
 
 // Nothing to close here.
 func (s *stanScaler) Close(context.Context) error {
+	if s.httpClient != nil {
+		s.httpClient.CloseIdleConnections()
+	}
 	return nil
 }
diff --git a/pkg/util/certificates.go b/pkg/util/certificates.go
@@ -36,7 +36,7 @@ var rootCAs *x509.CertPool
 
 func getRootCAs() *x509.CertPool {
 	if rootCAs != nil {
-		return rootCAs.Clone()
+		return rootCAs
 	}
 
 	rootCAs, _ = x509.SystemCertPool()
@@ -46,13 +46,13 @@ func getRootCAs() *x509.CertPool {
 
 	if _, err := os.Stat(customCAPath); errors.Is(err, fs.ErrNotExist) {
 		logger.V(1).Info(fmt.Sprintf("the path %s doesn't exist, skipping custom CA registrations", customCAPath))
-		return rootCAs.Clone()
+		return rootCAs
 	}
 
 	files, err := os.ReadDir(customCAPath)
 	if err != nil {
 		logger.Error(err, fmt.Sprintf("unable to read %s", customCAPath))
-		return rootCAs.Clone()
+		return rootCAs
 	}
 
 	for _, file := range files {
@@ -74,5 +74,5 @@ func getRootCAs() *x509.CertPool {
 		logger.V(1).Info(fmt.Sprintf("the certificate %s has been added to the pool", file.Name()))
 	}
 
-	return rootCAs.Clone()
+	return rootCAs
 }