Allow rate limit changes for inCluster mode (#606)

* Allow rate limit changes for inCluster mode

---------

Co-authored-by: João Bologna <joao.bologna@wildlifestudios.com>

docs/reference/Kubernetes.md

@@ -70,6 +70,10 @@ Maestro provides two ways for configuring kubernetes cluster access.
 Set `adapters.runtime.kubernetes.inCluster` config value to true or use its env var equivalent, the kubernetes client will be configured 
 automatically using the same _service account_ of the maestro component running pod.
 
+The kubernetes client has a default rate limiter which implements a token bucket approach.
+The inCluster mode allows rate limit changes via `adapters.runtime.kubernetes.qps` 
+and `adapters.runtime.kubernetes.burst` configs.
+
 This mode is recommended to be used when running maestro components in the same cluster
 in which the schedulers and rooms will be managed.
 

internal/service/adapters.go

@@ -58,6 +58,7 @@ import (
 	"github.com/topfreegames/maestro/internal/core/services/schedulers"
 	semconv "go.opentelemetry.io/otel/semconv/v1.4.0"
 	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/clientcmd"
 )
 
@@ -67,6 +68,8 @@ const (
 	runtimeKubernetesMasterURLPath  = "adapters.runtime.kubernetes.masterUrl"
 	runtimeKubernetesKubeconfigPath = "adapters.runtime.kubernetes.kubeconfig"
 	runtimeKubernetesInClusterPath  = "adapters.runtime.kubernetes.inCluster"
+	runtimeKubernetesQPS            = "adapters.runtime.kubernetes.qps"
+	runtimeKubernetesBurst          = "adapters.runtime.kubernetes.burst"
 	// Redis operation storage
 	operationStorageRedisURLPath      = "adapters.operationStorage.redis.url"
 	operationLeaseStorageRedisURLPath = "adapters.operationLeaseStorage.redis.url"
@@ -115,14 +118,22 @@ func NewEventsForwarder(c config.Config) (ports.EventsForwarder, error) {
 func NewRuntimeKubernetes(c config.Config) (ports.Runtime, error) {
 	var masterURL string
 	var kubeConfigPath string
+	var qps int
+	var burst int
 
 	inCluster := c.GetBool(runtimeKubernetesInClusterPath)
-	if !inCluster {
+	if inCluster {
+		qps = c.GetInt(runtimeKubernetesQPS)
+		burst = c.GetInt(runtimeKubernetesBurst)
+	} else {
 		masterURL = c.GetString(runtimeKubernetesMasterURLPath)
 		kubeConfigPath = c.GetString(runtimeKubernetesKubeconfigPath)
 	}
 
-	clientSet, err := createKubernetesClient(masterURL, kubeConfigPath)
+	clientSet, err := createKubernetesClient(masterURL, kubeConfigPath, func(conf *rest.Config) {
+		conf.QPS = float32(qps)
+		conf.Burst = burst
+	})
 	if err != nil {
 		return nil, fmt.Errorf("failed to initialize Kubernetes runtime: %w", err)
 	}
@@ -275,14 +286,18 @@ func connectToPostgres(url string) (*pg.Options, error) {
 	return opts, nil
 }
 
-func createKubernetesClient(masterURL, kubeconfigPath string) (kubernetes.Interface, error) {
+func createKubernetesClient(masterURL, kubeconfigPath string, opts ...func(*rest.Config)) (kubernetes.Interface, error) {
 	// NOTE: if neither masterURL or kubeconfigPath are not passed, this will
 	// fallback to in cluster config.
 	kubeconfig, err := clientcmd.BuildConfigFromFlags(masterURL, kubeconfigPath)
 	if err != nil {
 		return nil, fmt.Errorf("failed to construct kubernetes config: %w", err)
 	}
 
+	for _, opt := range opts {
+		opt(kubeconfig)
+	}
+
 	client, err := kubernetes.NewForConfig(kubeconfig)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create kubernetes client: %w", err)