Add insecureSkipTLS flag for grpc client/server connection

Signed-off-by: Ali Aqel <[email protected]>

cmd/adapter/main.go

@@ -56,12 +56,13 @@ type Adapter struct {
 var logger = klogr.New().WithName("keda_metrics_adapter")
 
 var (
-	adapterClientRequestQPS   float32
-	adapterClientRequestBurst int
-	metricsAPIServerPort      int
-	disableCompression        bool
-	metricsServiceAddr        string
-	profilingAddr             string
+	adapterClientRequestQPS             float32
+	adapterClientRequestBurst           int
+	metricsAPIServerPort                int
+	disableCompression                  bool
+	metricsServiceAddr                  string
+	profilingAddr                       string
+	insecureMetricsServiceSkipTLSVerify bool
 )
 
 func (a *Adapter) makeProvider(ctx context.Context) (provider.ExternalMetricsProvider, <-chan struct{}, error) {
@@ -123,7 +124,7 @@ func (a *Adapter) makeProvider(ctx context.Context) (provider.ExternalMetricsPro
 	}
 
 	logger.Info("Connecting Metrics Service gRPC client to the server", "address", metricsServiceAddr)
-	grpcClient, err := metricsservice.NewGrpcClient(metricsServiceAddr, a.SecureServing.ServerCert.CertDirectory)
+	grpcClient, err := metricsservice.NewGrpcClient(metricsServiceAddr, a.SecureServing.ServerCert.CertDirectory, metricsServiceGRPCAuthority, insecureMetricsServiceSkipTLSVerify)
 	if err != nil {
 		logger.Error(err, "error connecting Metrics Service gRPC client to the server", "address", metricsServiceAddr)
 		return nil, nil, err
@@ -237,6 +238,7 @@ func main() {
 	cmd.Flags().Float32Var(&adapterClientRequestQPS, "kube-api-qps", 20.0, "Set the QPS rate for throttling requests sent to the apiserver")
 	cmd.Flags().IntVar(&adapterClientRequestBurst, "kube-api-burst", 30, "Set the burst for throttling requests sent to the apiserver")
 	cmd.Flags().BoolVar(&disableCompression, "disable-compression", true, "Disable response compression for k8s restAPI in client-go. ")
+	cmd.Flags().BoolVar(&insecureMetricsServiceSkipTLSVerify, "insecure-metrics-service-skip-tls-verify", false, "Skip TLS verification on the GRPC connection to the metrics service")
 
 	if err := cmd.Flags().Parse(os.Args); err != nil {
 		return

cmd/operator/main.go

@@ -85,6 +85,7 @@ func main() {
 	var k8sClusterDomain string
 	var enableCertRotation bool
 	var validatingWebhookName string
+	var insecureMetricsServiceSkipTLSVerify bool
 	pflag.BoolVar(&enablePrometheusMetrics, "enable-prometheus-metrics", true, "Enable the prometheus metric of keda-operator.")
 	pflag.BoolVar(&enableOpenTelemetryMetrics, "enable-opentelemetry-metrics", false, "Enable the opentelemetry metric of keda-operator.")
 	pflag.StringVar(&metricsAddr, "metrics-bind-address", ":8080", "The address the prometheus metric endpoint binds to.")
@@ -106,6 +107,7 @@ func main() {
 	pflag.StringVar(&k8sClusterDomain, "k8s-cluster-domain", "cluster.local", "Kubernetes cluster domain. Defaults to cluster.local")
 	pflag.BoolVar(&enableCertRotation, "enable-cert-rotation", false, "enable automatic generation and rotation of TLS certificates/keys")
 	pflag.StringVar(&validatingWebhookName, "validating-webhook-name", "keda-admission", "ValidatingWebhookConfiguration name. Defaults to keda-admission")
+	pflag.BoolVar(&insecureMetricsServiceSkipTLSVerify, "insecure-metrics-service-skip-tls-verify", false, "Skip TLS verification on the GRPC server")
 	opts := zap.Options{}
 	opts.BindFlags(flag.CommandLine)
 	pflag.CommandLine.AddGoFlagSet(flag.CommandLine)
@@ -158,13 +160,14 @@ func main() {
 		Cache: ctrlcache.Options{
 			DefaultNamespaces: namespaces,
 		},
-		HealthProbeBindAddress: probeAddr,
-		PprofBindAddress:       profilingAddr,
-		LeaderElection:         enableLeaderElection,
-		LeaderElectionID:       "operator.keda.sh",
-		LeaseDuration:          leaseDuration,
-		RenewDeadline:          renewDeadline,
-		RetryPeriod:            retryPeriod,
+		HealthProbeBindAddress:  probeAddr,
+		PprofBindAddress:        profilingAddr,
+		LeaderElection:          enableLeaderElection,
+		LeaderElectionID:        "operator.keda.sh",
+		LeaderElectionNamespace: kedautil.GetPodNamespace(),
+		LeaseDuration:           leaseDuration,
+		RenewDeadline:           renewDeadline,
+		RetryPeriod:             retryPeriod,
 	})
 	if err != nil {
 		setupLog.Error(err, "unable to start manager")
@@ -299,7 +302,7 @@ func main() {
 		close(certReady)
 	}
 
-	grpcServer := metricsservice.NewGrpcServer(&scaledHandler, metricsServiceAddr, certDir, certReady)
+	grpcServer := metricsservice.NewGrpcServer(&scaledHandler, metricsServiceAddr, certDir, insecureMetricsServiceSkipTLSVerify, certReady)
 	if err := mgr.Add(&grpcServer); err != nil {
 		setupLog.Error(err, "unable to set up Metrics Service gRPC server")
 		os.Exit(1)

pkg/metricsservice/client.go

@@ -24,6 +24,7 @@ import (
 	"github.com/go-logr/logr"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/connectivity"
+	"google.golang.org/grpc/credentials/insecure"
 	"k8s.io/metrics/pkg/apis/external_metrics"
 	"k8s.io/metrics/pkg/apis/external_metrics/v1beta1"
 
@@ -36,7 +37,7 @@ type GrpcClient struct {
 	connection *grpc.ClientConn
 }
 
-func NewGrpcClient(url, certDir string) (*GrpcClient, error) {
+func NewGrpcClient(url, certDir string, skipTLSverify bool) (*GrpcClient, error) {
 	defaultConfig := `{
 		"methodConfig": [{
 		  "timeout": "3s",
@@ -49,14 +50,20 @@ func NewGrpcClient(url, certDir string) (*GrpcClient, error) {
 		  }
 		}]}`
 
-	creds, err := utils.LoadGrpcTLSCredentials(certDir, false)
-	if err != nil {
-		return nil, err
-	}
 	opts := []grpc.DialOption{
-		grpc.WithTransportCredentials(creds),
 		grpc.WithDefaultServiceConfig(defaultConfig),
 	}
+
+	if skipTLSverify {
+		opts = append(opts, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	} else {
+		creds, err := utils.LoadGrpcTLSCredentials(certDir, false)
+		if err != nil {
+			return nil, err
+		}
+		opts = append(opts, grpc.WithTransportCredentials(creds))
+	}
+
 	conn, err := grpc.Dial(url, opts...)
 	if err != nil {
 		return nil, err

pkg/metricsservice/server.go

@@ -33,11 +33,12 @@ import (
 var log = logf.Log.WithName("grpc_server")
 
 type GrpcServer struct {
-	server        *grpc.Server
-	address       string
-	certDir       string
-	certsReady    chan struct{}
-	scalerHandler *scaling.ScaleHandler
+	server                *grpc.Server
+	address               string
+	certDir               string
+	insecureSkipTLSVerify bool
+	certsReady            chan struct{}
+	scalerHandler         *scaling.ScaleHandler
 	api.UnimplementedMetricsServiceServer
 }
 
@@ -60,12 +61,13 @@ func (s *GrpcServer) GetMetrics(ctx context.Context, in *api.ScaledObjectRef) (*
 }
 
 // NewGrpcServer creates a new instance of GrpcServer
-func NewGrpcServer(scaleHandler *scaling.ScaleHandler, address, certDir string, certsReady chan struct{}) GrpcServer {
+func NewGrpcServer(scaleHandler *scaling.ScaleHandler, address, certDir string, insecureSkipTLSVerify bool, certsReady chan struct{}) GrpcServer {
 	return GrpcServer{
-		address:       address,
-		scalerHandler: scaleHandler,
-		certDir:       certDir,
-		certsReady:    certsReady,
+		address:               address,
+		scalerHandler:         scaleHandler,
+		certDir:               certDir,
+		insecureSkipTLSVerify: insecureSkipTLSVerify,
+		certsReady:            certsReady,
 	}
 }
 
@@ -87,11 +89,17 @@ func (s *GrpcServer) startServer() error {
 func (s *GrpcServer) Start(ctx context.Context) error {
 	<-s.certsReady
 	if s.server == nil {
-		creds, err := utils.LoadGrpcTLSCredentials(s.certDir, true)
-		if err != nil {
-			return err
+		opts := []grpc.ServerOption{}
+
+		if !s.insecureSkipTLSVerify {
+			creds, err := utils.LoadGrpcTLSCredentials(s.certDir, true)
+			if err != nil {
+				return err
+			}
+			opts = append(opts, grpc.Creds(creds))
 		}
-		s.server = grpc.NewServer(grpc.Creds(creds))
+
+		s.server = grpc.NewServer(opts...)
 		api.RegisterMetricsServiceServer(s.server, s)
 	}