using dedicated HTTP clients (kedacore#1251)

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

CHANGELOG.md

@@ -37,6 +37,7 @@
 
 ### New
 
+- KEDA uses a dedicated [HTTP client](https://pkg.go.dev/net/http#Client), connection pool, and (optional) TLS certificate for each configured scaler
 - KEDA scales any CustomResource that implements Scale subresource ([#703](https://github.com/kedacore/keda/issues/703))
 - Provide KEDA go-client ([#494](https://github.com/kedacore/keda/issues/494))
 - Define KEDA readiness and liveness probes ([#788](https://github.com/kedacore/keda/issues/788))

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,44 +41,44 @@ var (
  prometheusMetricsPath string
 )
 
-func (a *Adapter) makeProviderOrDie() provider.MetricsProvider {
+func (a *Adapter) makeProvider(globalHTTPTimeout time.Duration) (provider.MetricsProvider, error) {
  // Get a config to talk to the apiserver
  cfg, err := config.GetConfig()
  if err != nil {
  logger.Error(err, "failed to get the config")
- os.Exit(1)
+ return nil, fmt.Errorf("failed to get the config (%s)", err)
  }
 
  scheme := scheme.Scheme
  if err := appsv1.SchemeBuilder.AddToScheme(scheme); err != nil {
  logger.Error(err, "failed to add apps/v1 scheme to runtime scheme")
- os.Exit(1)
+ return nil, fmt.Errorf("failed to add apps/v1 scheme to runtime scheme (%s)", err)
  }
  if err := kedav1alpha1.SchemeBuilder.AddToScheme(scheme); err != nil {
  logger.Error(err, "failed to add keda scheme to runtime scheme")
- os.Exit(1)
+ return nil, fmt.Errorf("failed to add keda scheme to runtime scheme (%s)", err)
  }
 
  kubeclient, err := client.New(cfg, client.Options{
  Scheme: scheme,
  })
  if err != nil {
  logger.Error(err, "unable to construct new client")
- os.Exit(1)
+ return nil, fmt.Errorf("unable to construct new client (%s)", err)
  }
 
- handler := scaling.NewScaleHandler(kubeclient, nil, scheme)
+ handler := scaling.NewScaleHandler(kubeclient, nil, scheme, globalHTTPTimeout)
 
  namespace, err := getWatchNamespace()
  if err != nil {
  logger.Error(err, "failed to get watch namespace")
- os.Exit(1)
+ return nil, fmt.Errorf("failed to get watch namespace (%s)", err)
  }
 
  prometheusServer := &prommetrics.PrometheusMetricServer{}
  go func() { prometheusServer.NewServer(fmt.Sprintf(":%v", prometheusMetricsPort), prometheusMetricsPath) }()
 
- return kedaprovider.NewProvider(logger, handler, kubeclient, namespace)
+ return kedaprovider.NewProvider(logger, handler, kubeclient, namespace), nil
 }
 
 func printVersion() {
@@ -117,7 +119,23 @@ func main() {
  return
  }
 
- 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")
+ return
+ }
+
+ kedaProvider, err := cmd.makeProvider(time.Duration(globalHTTPTimeoutMS) * time.Millisecond)
+ if err != nil {
+ logger.Error(err, "making provider")
+ return
+ }
  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).