using dedicated HTTP clients

fixes kedacore#1133

Signed-off-by: Aaron Schlesinger <aaron@ecomaz.net>

reading timeout from env var, and storing HTTP client in ScalerConfig

Signed-off-by: Aaron Schlesinger <aaron@ecomaz.net>

fixing undeclared name client

Signed-off-by: Aaron Schlesinger <aaron@ecomaz.net>

adapter/main.go

@@ -5,6 +5,8 @@ import (
  "fmt"
  "os"
  "runtime"
+ "strconv"
+ "time"
 
  appsv1 "k8s.io/api/apps/v1"
  "k8s.io/apimachinery/pkg/util/wait"
@@ -39,7 +41,7 @@ var (
  prometheusMetricsPath string
 )
 
-func (a *Adapter) makeProviderOrDie() provider.MetricsProvider {
+func (a *Adapter) makeProviderOrDie(globalHTTPTimeout time.Duration) provider.MetricsProvider {
  // Get a config to talk to the apiserver
  cfg, err := config.GetConfig()
  if err != nil {
@@ -65,7 +67,7 @@ func (a *Adapter) makeProviderOrDie() provider.MetricsProvider {
  os.Exit(1)
  }
 
- handler := scaling.NewScaleHandler(kubeclient, nil, scheme)
+ handler := scaling.NewScaleHandler(kubeclient, nil, scheme, globalHTTPTimeout)
 
  namespace, err := getWatchNamespace()
  if err != nil {
@@ -113,9 +115,22 @@ func main() {
  cmd.Flags().AddGoFlagSet(flag.CommandLine) // make sure we get the klog flags
  cmd.Flags().IntVar(&prometheusMetricsPort, "metrics-port", 9022, "Set the port to expose prometheus metrics")
  cmd.Flags().StringVar(&prometheusMetricsPath, "metrics-path", "/metrics", "Set the path for the prometheus metrics endpoint")
+
  cmd.Flags().Parse(os.Args)
 
- kedaProvider := cmd.makeProviderOrDie()
+ globalHTTPTimeoutStr := os.Getenv("KEDA_HTTP_DEFAULT_TIMEOUT")
+ if globalHTTPTimeoutStr == "" {
+ // default to 3 seconds if they don't pass the env var
+ globalHTTPTimeoutStr = "3000"
+ }
+
+ globalHTTPTimeoutMS, err := strconv.Atoi(globalHTTPTimeoutStr)
+ if err != nil {
+ logger.Error(err, "Invalid KEDA_HTTP_DEFAULT_TIMEOUT")
+ os.Exit(1)
+ }
+
+ kedaProvider := cmd.makeProviderOrDie(time.Duration(globalHTTPTimeoutMS) * time.Millisecond)
  cmd.WithExternalMetrics(kedaProvider)
 
  logger.Info(cmd.Message)

controllers/scaledjob_controller.go

@@ -3,6 +3,7 @@ package controllers
 import (
  "context"
  "fmt"
+ "time"
 
  "github.com/go-logr/logr"
  batchv1 "k8s.io/api/batch/v1"
@@ -25,14 +26,15 @@ import (
 // ScaledJobReconciler reconciles a ScaledJob object
 type ScaledJobReconciler struct {
  client.Client
- Log logr.Logger
- Scheme *runtime.Scheme
- scaleHandler scaling.ScaleHandler
+ Log logr.Logger
+ Scheme *runtime.Scheme
+ scaleHandler scaling.ScaleHandler
+ globalHTTPTimeout time.Duration
 }
 
 // SetupWithManager initializes the ScaledJobReconciler instance and starts a new controller managed by the passed Manager instance.
 func (r *ScaledJobReconciler) SetupWithManager(mgr ctrl.Manager) error {
- r.scaleHandler = scaling.NewScaleHandler(mgr.GetClient(), nil, mgr.GetScheme())
+ r.scaleHandler = scaling.NewScaleHandler(mgr.GetClient(), nil, mgr.GetScheme(), r.globalHTTPTimeout)
 
  return ctrl.NewControllerManagedBy(mgr).
  // Ignore updates to ScaledJob Status (in this case metadata.Generation does not change)

controllers/scaledobject_controller.go

@@ -4,6 +4,7 @@ import (
  "context"
  "fmt"
  "sync"
+ "time"
 
  "github.com/go-logr/logr"
  autoscalingv2beta2 "k8s.io/api/autoscaling/v2beta2"
@@ -48,6 +49,8 @@ type ScaledObjectReconciler struct {
  scaledObjectsGenerations *sync.Map
  scaleHandler scaling.ScaleHandler
  kubeVersion kedautil.K8sVersion
+
+ globalHTTPTimeout time.Duration
 }
 
 // SetupWithManager initializes the ScaledObjectReconciler instance and starts a new controller managed by the passed Manager instance.
@@ -75,7 +78,7 @@ func (r *ScaledObjectReconciler) SetupWithManager(mgr ctrl.Manager) error {
  // Init the rest of ScaledObjectReconciler
  r.restMapper = mgr.GetRESTMapper()
  r.scaledObjectsGenerations = &sync.Map{}
- r.scaleHandler = scaling.NewScaleHandler(mgr.GetClient(), r.scaleClient, mgr.GetScheme())
+ r.scaleHandler = scaling.NewScaleHandler(mgr.GetClient(), r.scaleClient, mgr.GetScheme(), r.globalHTTPTimeout)
 
  // Start controller
  return ctrl.NewControllerManagedBy(mgr).

pkg/scalers/artemis_scaler.go

@@ -21,7 +21,8 @@ import (
 )
 
 type artemisScaler struct {
- metadata *artemisMetadata
+ metadata *artemisMetadata
+ httpClient *http.Client
 }
 
 //revive:disable:var-naming breaking change on restApiTemplate, wouldn't bring any benefit to users
@@ -55,13 +56,21 @@ var artemisLog = logf.Log.WithName("artemis_queue_scaler")
 
 // NewArtemisQueueScaler creates a new artemis queue Scaler
 func NewArtemisQueueScaler(config *ScalerConfig) (Scaler, error) {
+ // do we need to guarantee this timeout for a specific
+ // reason? if not, we can have buildScaler pass in
+ // the global client
+ httpClient := &http.Client{
+ Timeout: 3 * time.Second,
+ }
+
  artemisMetadata, err := parseArtemisMetadata(config)
  if err != nil {
  return nil, fmt.Errorf("error parsing artemis metadata: %s", err)
  }
 
  return &artemisScaler{
- metadata: artemisMetadata,
+ metadata: artemisMetadata,
+ httpClient: httpClient,
  }, nil
 }
 
@@ -165,9 +174,7 @@ func (s *artemisScaler) getQueueMessageCount() (int, error) {
  var monitoringInfo *artemisMonitoring
  messageCount := 0
 
- client := &http.Client{
- Timeout: time.Second * 3,
- }
+ client := s.httpClient
  url := s.getMonitoringEndpoint()
 
  req, err := http.NewRequest("GET", url, nil)

pkg/scalers/artemis_scaler_test.go

@@ -1,6 +1,7 @@
 package scalers
 
 import (
+ "net/http"
  "testing"
 )
 
@@ -117,7 +118,10 @@ func TestArtemisGetMetricSpecForScaling(t *testing.T) {
  if err != nil {
  t.Fatal("Could not parse metadata:", err)
  }
- mockArtemisScaler := artemisScaler{meta}
+ mockArtemisScaler := artemisScaler{
+ metadata: meta,
+ httpClient: http.DefaultClient,
+ }
 
  metricSpec := mockArtemisScaler.GetMetricSpecForScaling()
  metricName := metricSpec[0].External.Metric.Name

pkg/scalers/azure/azure_aad_podidentity.go

@@ -14,10 +14,15 @@ const (
 )
 
 // GetAzureADPodIdentityToken returns the AADToken for resource
-func GetAzureADPodIdentityToken(audience string) (AADToken, error) {
+func GetAzureADPodIdentityToken(httpClient *http.Client, audience string) (AADToken, error) {
  var token AADToken
 
- resp, err := http.Get(fmt.Sprintf(msiURL, url.QueryEscape(audience)))
+ urlStr := fmt.Sprintf(msiURL, url.QueryEscape(audience))
+ req, err := http.NewRequest("GET", urlStr, nil)
+ if err != nil {
+ return token, err
+ }
+ resp, err := httpClient.Do(req)
  if err != nil {
  return token, err
  }

pkg/scalers/azure/azure_blob.go

@@ -2,15 +2,16 @@ package azure
 
 import (
  "context"
+ "net/http"
 
  "github.com/Azure/azure-storage-blob-go/azblob"
 
  kedav1alpha1 "github.com/kedacore/keda/v2/api/v1alpha1"
 )
 
 // GetAzureBlobListLength returns the count of the blobs in blob container in int
-func GetAzureBlobListLength(ctx context.Context, podIdentity kedav1alpha1.PodIdentityProvider, connectionString, blobContainerName string, accountName string, blobDelimiter string, blobPrefix string) (int, error) {
- credential, endpoint, err := ParseAzureStorageBlobConnection(podIdentity, connectionString, accountName)
+func GetAzureBlobListLength(ctx context.Context, httpClient *http.Client, podIdentity kedav1alpha1.PodIdentityProvider, connectionString, blobContainerName string, accountName string, blobDelimiter string, blobPrefix string) (int, error) {
+ credential, endpoint, err := ParseAzureStorageBlobConnection(httpClient, podIdentity, connectionString, accountName)
  if err != nil {
  return -1, err
  }

pkg/scalers/azure/azure_blob_test.go

@@ -2,12 +2,14 @@ package azure
 
 import (
  "context"
+ "net/http"
  "strings"
  "testing"
 )
 
 func TestGetBlobLength(t *testing.T) {
- length, err := GetAzureBlobListLength(context.TODO(), "", "", "blobContainerName", "", "", "")
+ httpClient := http.DefaultClient
+ length, err := GetAzureBlobListLength(context.TODO(), httpClient, "", "", "blobContainerName", "", "", "")
  if length != -1 {
  t.Error("Expected length to be -1, but got", length)
  }
@@ -20,7 +22,7 @@ func TestGetBlobLength(t *testing.T) {
  t.Error("Expected error to contain parsing error message, but got", err.Error())
  }
 
- length, err = GetAzureBlobListLength(context.TODO(), "", "DefaultEndpointsProtocol=https;AccountName=name;AccountKey=key==;EndpointSuffix=core.windows.net", "blobContainerName", "", "", "")
+ length, err = GetAzureBlobListLength(context.TODO(), httpClient, "", "DefaultEndpointsProtocol=https;AccountName=name;AccountKey=key==;EndpointSuffix=core.windows.net", "blobContainerName", "", "", "")
 
  if length != -1 {
  t.Error("Expected length to be -1, but got", length)

pkg/scalers/azure/azure_eventhub.go

@@ -6,6 +6,7 @@ import (
  "encoding/json"
  "errors"
  "fmt"
+ "net/http"
  "net/url"
  "strings"
 
@@ -80,8 +81,8 @@ func GetEventHubClient(info EventHubInfo) (*eventhub.Hub, error) {
 }
 
 // GetCheckpointFromBlobStorage accesses Blob storage and gets checkpoint information of a partition
-func GetCheckpointFromBlobStorage(ctx context.Context, info EventHubInfo, partitionID string) (Checkpoint, error) {
- blobCreds, storageEndpoint, err := ParseAzureStorageBlobConnection(kedav1alpha1.PodIdentityProviderNone, info.StorageConnection, "")
+func GetCheckpointFromBlobStorage(ctx context.Context, httpClient *http.Client, info EventHubInfo, partitionID string) (Checkpoint, error) {
+ blobCreds, storageEndpoint, err := ParseAzureStorageBlobConnection(httpClient, kedav1alpha1.PodIdentityProviderNone, info.StorageConnection, "")
  if err != nil {
  return Checkpoint{}, err
  }

pkg/scalers/azure/azure_queue.go

@@ -2,15 +2,16 @@ package azure
 
 import (
  "context"
+ "net/http"
 
  "github.com/Azure/azure-storage-queue-go/azqueue"
 
  kedav1alpha1 "github.com/kedacore/keda/v2/api/v1alpha1"
 )
 
 // GetAzureQueueLength returns the length of a queue in int
-func GetAzureQueueLength(ctx context.Context, podIdentity kedav1alpha1.PodIdentityProvider, connectionString, queueName string, accountName string) (int32, error) {
- credential, endpoint, err := ParseAzureStorageQueueConnection(podIdentity, connectionString, accountName)
+func GetAzureQueueLength(ctx context.Context, httpClient *http.Client, podIdentity kedav1alpha1.PodIdentityProvider, connectionString, queueName string, accountName string) (int32, error) {
+ credential, endpoint, err := ParseAzureStorageQueueConnection(httpClient, podIdentity, connectionString, accountName)
  if err != nil {
  return -1, err
  }

pkg/scalers/azure/azure_queue_test.go

@@ -2,12 +2,13 @@ package azure
 
 import (
  "context"
+ "net/http"
  "strings"
  "testing"
 )
 
 func TestGetQueueLength(t *testing.T) {
- length, err := GetAzureQueueLength(context.TODO(), "", "", "queueName", "")
+ length, err := GetAzureQueueLength(context.TODO(), http.DefaultClient, "", "", "queueName", "")
  if length != -1 {
  t.Error("Expected length to be -1, but got", length)
  }
@@ -20,7 +21,7 @@ func TestGetQueueLength(t *testing.T) {
  t.Error("Expected error to contain parsing error message, but got", err.Error())
  }
 
- length, err = GetAzureQueueLength(context.TODO(), "", "DefaultEndpointsProtocol=https;AccountName=name;AccountKey=key==;EndpointSuffix=core.windows.net", "queueName", "")
+ length, err = GetAzureQueueLength(context.TODO(), http.DefaultClient, "", "DefaultEndpointsProtocol=https;AccountName=name;AccountKey=key==;EndpointSuffix=core.windows.net", "queueName", "")
 
  if length != -1 {
  t.Error("Expected length to be -1, but got", length)

pkg/scalers/azure/azure_storage.go

@@ -3,6 +3,7 @@ package azure
 import (
  "errors"
  "fmt"
+ "net/http"
  "net/url"
  "strings"
 
@@ -42,10 +43,10 @@ func (e StorageEndpointType) Name() string {
 }
 
 // ParseAzureStorageQueueConnection parses queue connection string and returns credential and resource url
-func ParseAzureStorageQueueConnection(podIdentity kedav1alpha1.PodIdentityProvider, connectionString, accountName string) (azqueue.Credential, *url.URL, error) {
+func ParseAzureStorageQueueConnection(httpClient *http.Client, podIdentity kedav1alpha1.PodIdentityProvider, connectionString, accountName string) (azqueue.Credential, *url.URL, error) {
  switch podIdentity {
  case kedav1alpha1.PodIdentityProviderAzure:
- token, err := GetAzureADPodIdentityToken("https://storage.azure.com/")
+ token, err := GetAzureADPodIdentityToken(httpClient, "https://storage.azure.com/")
  if err != nil {
  return nil, nil, err
  }
@@ -75,10 +76,10 @@ func ParseAzureStorageQueueConnection(podIdentity kedav1alpha1.PodIdentityProvid
 }
 
 // ParseAzureStorageBlobConnection parses blob connection string and returns credential and resource url
-func ParseAzureStorageBlobConnection(podIdentity kedav1alpha1.PodIdentityProvider, connectionString, accountName string) (azblob.Credential, *url.URL, error) {
+func ParseAzureStorageBlobConnection(httpClient *http.Client, podIdentity kedav1alpha1.PodIdentityProvider, connectionString, accountName string) (azblob.Credential, *url.URL, error) {
  switch podIdentity {
  case kedav1alpha1.PodIdentityProviderAzure:
- token, err := GetAzureADPodIdentityToken("https://storage.azure.com/")
+ token, err := GetAzureADPodIdentityToken(httpClient, "https://storage.azure.com/")
  if err != nil {
  return nil, nil, err
  }

pkg/scalers/azure_blob_scaler.go

@@ -3,6 +3,7 @@ package scalers
 import (
  "context"
  "fmt"
+ "net/http"
  "strconv"
 
  "github.com/kedacore/keda/v2/pkg/scalers/azure"
@@ -28,6 +29,7 @@ const (
 type azureBlobScaler struct {
  metadata *azureBlobMetadata
  podIdentity kedav1alpha1.PodIdentityProvider
+ httpClient *http.Client
 }
 
 type azureBlobMetadata struct {
@@ -51,6 +53,7 @@ func NewAzureBlobScaler(config *ScalerConfig) (Scaler, error) {
  return &azureBlobScaler{
  metadata: meta,
  podIdentity: podIdentity,
+ httpClient: config.HTTPClient,
  }, nil
 }
 
@@ -122,6 +125,7 @@ func parseAzureBlobMetadata(config *ScalerConfig) (*azureBlobMetadata, kedav1alp
 func (s *azureBlobScaler) IsActive(ctx context.Context) (bool, error) {
  length, err := azure.GetAzureBlobListLength(
  ctx,
+ s.httpClient,
  s.podIdentity,
  s.metadata.connection,
  s.metadata.blobContainerName,
@@ -161,6 +165,7 @@ func (s *azureBlobScaler) GetMetricSpecForScaling() []v2beta2.MetricSpec {
 func (s *azureBlobScaler) GetMetrics(ctx context.Context, metricName string, metricSelector labels.Selector) ([]external_metrics.ExternalMetricValue, error) {
  bloblen, err := azure.GetAzureBlobListLength(
  ctx,
+ s.httpClient,
  s.podIdentity,
  s.metadata.connection,
  s.metadata.blobContainerName,

pkg/scalers/azure_blob_scaler_test.go

@@ -1,6 +1,7 @@
 package scalers
 
 import (
+ "net/http"
  "testing"
 
  kedav1alpha1 "github.com/kedacore/keda/v2/api/v1alpha1"
@@ -68,7 +69,11 @@ func TestAzBlobGetMetricSpecForScaling(t *testing.T) {
  if err != nil {
  t.Fatal("Could not parse metadata:", err)
  }
- mockAzBlobScaler := azureBlobScaler{meta, podIdentity}
+ mockAzBlobScaler := azureBlobScaler{
+ metadata: meta,
+ podIdentity: podIdentity,
+ httpClient: http.DefaultClient,
+ }
 
  metricSpec := mockAzBlobScaler.GetMetricSpecForScaling()
  metricName := metricSpec[0].External.Metric.Name