Use client generator to generate all k8s clients and add respective client-go metrics (#2668)

* Define client generator to generate all k8s clients
* Increase QPS and burst values
* Capture client-go metrics
* Support for controller runtime metrics

Signed-off-by: Sanket Sudake <sanketsudake@gmail.com>

pkg/mqtrigger/metrics.go

@@ -18,26 +18,27 @@ package mqtrigger
 
 import (
 	"github.com/prometheus/client_golang/prometheus"
-	"github.com/prometheus/client_golang/prometheus/promauto"
+
+	"github.com/fission/fission/pkg/utils/metrics"
 )
 
 var (
 	labels            = []string{"trigger_name", "trigger_namespace"}
-	subscriptionCount = promauto.NewGaugeVec(
+	subscriptionCount = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
 			Name: "fission_mqt_subscriptions",
 			Help: "Total number of subscriptions to mq currently",
 		},
 		[]string{},
 	)
-	messageCount = promauto.NewCounterVec(
+	messageCount = prometheus.NewCounterVec(
 		prometheus.CounterOpts{
 			Name: "fission_mqt_messages_processed_total",
 			Help: "Total number of messages processed",
 		},
 		labels,
 	)
-	messageLagCount = promauto.NewGaugeVec(
+	messageLagCount = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
 			Name: "fission_mqt_message_lag",
 			Help: "Total number of messages lag per topic and partition",
@@ -61,3 +62,10 @@ func IncreaseMessageCount(trigname, trignamespace string) {
 func SetMessageLagCount(trigname, trignamespace, topic, partition string, lag int64) {
 	messageLagCount.WithLabelValues(trigname, trignamespace, topic, partition).Set(float64(lag))
 }
+
+func init() {
+	registry := metrics.Registry
+	registry.MustRegister(subscriptionCount)
+	registry.MustRegister(messageCount)
+	registry.MustRegister(messageLagCount)
+}

pkg/mqtrigger/scalermanager.go

@@ -50,7 +50,8 @@ var (
 )
 
 func getScaledObjectClient(namespace string) (dynamic.ResourceInterface, error) {
-	dynamicClient, err := crd.GetDynamicClient()
+	clientGen := crd.NewClientGenerator()
+	dynamicClient, err := clientGen.GetDynamicClient()
 	if err != nil {
 		return nil, err
 	}
@@ -58,7 +59,8 @@ func getScaledObjectClient(namespace string) (dynamic.ResourceInterface, error)
 }
 
 func getAuthTriggerClient(namespace string) (dynamic.ResourceInterface, error) {
-	dynamicClient, err := crd.GetDynamicClient()
+	clientGen := crd.NewClientGenerator()
+	dynamicClient, err := clientGen.GetDynamicClient()
 	if err != nil {
 		return nil, err
 	}
@@ -149,10 +151,16 @@ func mqTriggerEventHandlers(ctx context.Context, logger *zap.Logger, kubeClient
 // StartScalerManager watches for changes in MessageQueueTrigger and,
 // Based on changes, it Creates, Updates and Deletes Objects of Kind ScaledObjects, AuthenticationTriggers and Deployments
 func StartScalerManager(ctx context.Context, logger *zap.Logger, routerURL string) error {
-	fissionClient, kubeClient, _, _, err := crd.MakeFissionClient()
+	clientGen := crd.NewClientGenerator()
+	fissionClient, err := clientGen.GetFissionClient()
 	if err != nil {
-		return err
+		return errors.Wrap(err, "failed to get fission client")
 	}
+	kubeClient, err := clientGen.GetKubernetesClient()
+	if err != nil {
+		return errors.Wrap(err, "failed to get kubernetes client")
+	}
+
 	err = crd.WaitForCRDs(ctx, logger, fissionClient)
 	if err != nil {
 		return errors.Wrap(err, "error waiting for CRDs")