instrumenting k8s client and cache

Signed-off-by: Daniel Rammer <[email protected]>
flyteorg · Nov 6, 2023 · 236da85 · 236da85
1 parent 77cda4b
commit 236da85
Show file tree

Hide file tree

Showing 2 changed files with 28 additions and 4 deletions.
diff --git a/cmd/single/start.go b/cmd/single/start.go
@@ -17,7 +17,6 @@ import (
 	adminScheduler "github.com/flyteorg/flyte/flyteadmin/scheduler"
 	propellerEntrypoint "github.com/flyteorg/flyte/flytepropeller/pkg/controller"
 	propellerConfig "github.com/flyteorg/flyte/flytepropeller/pkg/controller/config"
-	"github.com/flyteorg/flyte/flytepropeller/pkg/controller/executors"
 	"github.com/flyteorg/flyte/flytepropeller/pkg/signals"
 	webhookEntrypoint "github.com/flyteorg/flyte/flytepropeller/pkg/webhook"
 	webhookConfig "github.com/flyteorg/flyte/flytepropeller/pkg/webhook/config"
@@ -121,8 +120,21 @@ func startPropeller(ctx context.Context, cfg Propeller) error {
 			SyncPeriod:        &propellerCfg.DownstreamEval.Duration,
 			DefaultNamespaces: namespaceConfigs,
 		},
+		NewCache: func (config *rest.Config, options cache.Options) (cache.Cache, error) {
+			k8sCache, err := cache.New(config, options)
+			if err != nil {
+				return k8sCache, err
+			}
+
+			return otelutils.WrapK8sCache(k8sCache), nil
+		},
 		NewClient: func(config *rest.Config, options client.Options) (client.Client, error) {
-			return executors.NewFallbackClientBuilder(propellerScope.NewSubScope("kube")).Build(nil, config, options)
+			k8sClient, err := client.New(config, options)
+			if err != nil {
+				return k8sClient, err
+			}
+
+			return otelutils.WrapK8sClient(k8sClient), nil
 		},
 		Metrics: metricsserver.Options{
 			// Disable metrics serving

diff --git a/flytepropeller/cmd/controller/cmd/root.go b/flytepropeller/cmd/controller/cmd/root.go
@@ -22,7 +22,6 @@ import (
 
 	"github.com/flyteorg/flyte/flytepropeller/pkg/controller"
 	config2 "github.com/flyteorg/flyte/flytepropeller/pkg/controller/config"
-	"github.com/flyteorg/flyte/flytepropeller/pkg/controller/executors"
 	"github.com/flyteorg/flyte/flytepropeller/pkg/signals"
 	"github.com/flyteorg/flyte/flytestdlib/config"
 	"github.com/flyteorg/flyte/flytestdlib/config/viper"
@@ -145,8 +144,21 @@ func executeRootCmd(baseCtx context.Context, cfg *config2.Config) error {
 			SyncPeriod:        &cfg.DownstreamEval.Duration,
 			DefaultNamespaces: namespaceConfigs,
 		},
+		NewCache: func(config *rest.Config, opts cache.Options) (cache.Cache, error) {
+			k8sCache, err := cache.New(config, options)
+			if err != nil {
+				return k8sCache, err
+			}
+
+			return otelutils.WrapK8sCache(k8sCache)
+		},
 		NewClient: func(config *rest.Config, options client.Options) (client.Client, error) {
-			return executors.NewFallbackClientBuilder(propellerScope.NewSubScope("kube")).Build(nil, config, options)
+			k8sClient, err := client.New(config, options)
+			if err != nil {
+				return k8sClient, err
+			}
+
+			return otelutils.WrapK8sClient(k8sClient)
 		},
 		Metrics: metricsserver.Options{
 			// Disable metrics serving