diff --git a/cmd/controller/cmd/root.go b/cmd/controller/cmd/root.go index 315df6533..24bf51818 100644 --- a/cmd/controller/cmd/root.go +++ b/cmd/controller/cmd/root.go @@ -215,6 +215,13 @@ func executeRootCmd(cfg *config2.Config) { logger.Fatalf(ctx, "Failed to initialize controller run-time manager. Error: %v", err) } + go func() { + err = mgr.Start(ctx.Done()) + if err != nil { + logger.Fatalf(ctx, "Failed to start manager. Error: %v", err) + } + }() + c, err := controller.New(ctx, cfg, kubeClient, flyteworkflowClient, flyteworkflowInformerFactory, mgr, propellerScope) if err != nil { diff --git a/config.yaml b/config.yaml index b086508ed..5a9c4e3e9 100644 --- a/config.yaml +++ b/config.yaml @@ -31,6 +31,7 @@ tasks: - container - K8S-ARRAY - qubole-hive-executor + - spark # Sample plugins config plugins: # All k8s plugins default configuration @@ -62,7 +63,7 @@ storage: endpoint: http://localhost:30084 region: us-east-1 secret-key: miniostorage - type: minio + type: mem container: "my-s3-bucket" event: type: admin diff --git a/pkg/controller/nodes/task/k8s/plugin_manager.go b/pkg/controller/nodes/task/k8s/plugin_manager.go index 42f6556a2..11722a7bc 100644 --- a/pkg/controller/nodes/task/k8s/plugin_manager.go +++ b/pkg/controller/nodes/task/k8s/plugin_manager.go @@ -3,9 +3,12 @@ package k8s import ( "context" "fmt" + "reflect" "strings" "time" + "k8s.io/client-go/tools/cache" + "github.com/lyft/flytepropeller/pkg/controller/nodes/task/backoff" v1 "k8s.io/api/core/v1" @@ -417,8 +420,31 @@ func NewPluginManager(ctx context.Context, iCtx pluginsCore.SetupContext, entry handler.Funcs{ CreateFunc: func(evt event.CreateEvent, q2 workqueue.RateLimitingInterface) { logger.Debugf(context.Background(), "Create received for %s, ignoring.", evt.Meta.GetName()) + i, err := iCtx.KubeClient().GetCache().GetInformer(entry.ResourceToWatch) + if err != nil { + panic(err) + } + + si, casted := i.(cache.SharedIndexInformer) + if !casted { + panic(fmt.Errorf("wrong type. Actual: %v", reflect.TypeOf(i))) + } + + logger.Infof(ctx, "Found items in store [%v]", len(si.GetStore().List())) }, UpdateFunc: func(evt event.UpdateEvent, q2 workqueue.RateLimitingInterface) { + i, err := iCtx.KubeClient().GetCache().GetInformer(entry.ResourceToWatch) + if err != nil { + panic(err) + } + + si, casted := i.(cache.SharedIndexInformer) + if !casted { + panic(fmt.Errorf("wrong type. Actual: %v", reflect.TypeOf(i))) + } + + logger.Infof(ctx, "Found items in store [%v] for kind [%v]", len(si.GetStore().List()), entry.ResourceToWatch) + if evt.MetaNew == nil { logger.Warn(context.Background(), "Received an Update event with nil MetaNew.") } else if evt.MetaOld == nil || evt.MetaOld.GetResourceVersion() != evt.MetaNew.GetResourceVersion() { @@ -466,6 +492,18 @@ func NewPluginManager(ctx context.Context, iCtx pluginsCore.SetupContext, entry return nil, err } + i, err := iCtx.KubeClient().GetCache().GetInformer(entry.ResourceToWatch) + if err != nil { + panic(err) + } + + si, casted := i.(cache.SharedIndexInformer) + if !casted { + panic(fmt.Errorf("wrong type. Actual: %v", reflect.TypeOf(i))) + } + + logger.Infof(ctx, "Found items in store [%v]", len(si.GetStore().List())) + return &PluginManager{ id: entry.ID, plugin: entry.Plugin, diff --git a/pkg/controller/workers.go b/pkg/controller/workers.go index 7b6303c81..06347e0c0 100644 --- a/pkg/controller/workers.go +++ b/pkg/controller/workers.go @@ -3,7 +3,6 @@ package controller import ( "context" "fmt" - "runtime/pprof" "time" "github.com/lyft/flytestdlib/contextutils" @@ -139,17 +138,17 @@ func (w *WorkerPool) Run(ctx context.Context, threadiness int, synced ...cache.I } logger.Infof(ctx, "Starting workers [%d]", threadiness) - // Launch workers to process FlyteWorkflow resources - for i := 0; i < threadiness; i++ { - w.metrics.FreeWorkers.Inc() - logger.Infof(ctx, "Starting worker [%d]", i) - workerLabel := fmt.Sprintf("worker-%v", i) - go func() { - workerCtx := contextutils.WithGoroutineLabel(ctx, workerLabel) - pprof.SetGoroutineLabels(workerCtx) - w.runWorker(workerCtx) - }() - } + //// Launch workers to process FlyteWorkflow resources + //for i := 0; i < threadiness; i++ { + // w.metrics.FreeWorkers.Inc() + // logger.Infof(ctx, "Starting worker [%d]", i) + // workerLabel := fmt.Sprintf("worker-%v", i) + // go func() { + // workerCtx := contextutils.WithGoroutineLabel(ctx, workerLabel) + // pprof.SetGoroutineLabels(workerCtx) + // w.runWorker(workerCtx) + // }() + //} w.workQueue.Start(ctx) logger.Info(ctx, "Started workers")