Sync Plugin Interface

go.mod

@@ -135,3 +135,11 @@ require (
 )
 
 replace github.com/aws/amazon-sagemaker-operator-for-k8s => github.com/aws/amazon-sagemaker-operator-for-k8s v1.0.1-0.20210303003444-0fb33b1fd49d
+
+// replace github.com/flyteorg/flyteidl/gen/pb-go/flyteidl/admin => /mnt/c/code/dev/flyteidl/gen/pb-go/flyteidl/admin
+
+// replace github.com/flyteorg/flyteidl/gen/pb-go/flyteidl/admin => /mnt/c/code/dev/flyteidl/gen/pb-go/flyteidl/admin
+
+// replace github.com/flyteorg/flyteidl/gen/pb-go/flyteidl/service => /mnt/c/code/dev/flyteidl/gen/pb-go/flyteidl/service
+
+replace github.com/flyteorg/flyteidl v1.5.13 => /mnt/c/code/dev/flyteidl

go/tasks/config/config.go

@@ -1,7 +1,10 @@
 package config
 
 import (
+	"context"
+
 	"github.com/flyteorg/flytestdlib/config"
+	"github.com/flyteorg/flytestdlib/logger"
 )
 
 const configSectionKey = "plugins"
@@ -21,6 +24,12 @@ func GetConfig() *Config {
 	return rootSection.GetConfig().(*Config)
 }
 
+/*
+resourcemanager&{noop 1000 {[]    0}}
+admin-launcher&{100 10 10000 10}
+workflowStore&{ResourceVersionCache}
+*/
 func MustRegisterSubSection(subSectionKey string, section config.Config) config.Section {
+	logger.Error(context.TODO(), "@@@ MustRegisterSubSection subSectionKey->[%v], section->[%v]", subSectionKey, section)
 	return rootSection.MustRegisterSection(subSectionKey, section)
 }

go/tasks/pluginmachinery/internal/webapi/cache.go

@@ -2,6 +2,7 @@ package webapi
 
 import (
 	"context"
+	"runtime"
 
 	"github.com/flyteorg/flytestdlib/promutils"
 	"k8s.io/client-go/util/workqueue"
@@ -55,6 +56,9 @@ type CacheItem struct {
 func (q *ResourceCache) SyncResource(ctx context.Context, batch cache.Batch) (
 	updatedBatch []cache.ItemSyncResponse, err error) {
 
+	pc, file, line, _ := runtime.Caller(1)
+	funcName := runtime.FuncForPC(pc).Name()
+	logger.Infof(context.TODO(), "@@@ webapi cache.go SyncResource was called by file [%v] [%v]:[%v]", file, funcName, line)
 	resp := make([]cache.ItemSyncResponse, 0, len(batch))
 	for _, resource := range batch {
 		// Cast the item back to the thing we want to work with.
@@ -98,7 +102,7 @@ func (q *ResourceCache) SyncResource(ctx context.Context, batch cache.Batch) (
 		}
 
 		// Get an updated status
-		logger.Debugf(ctx, "Querying AsyncPlugin for %s", resource.GetID())
+		logger.Infof(ctx, "@@@ Querying AsyncPlugin for %s", resource.GetID())
 		newResource, err := q.client.Get(ctx, newPluginContext(cacheItem.ResourceMeta, cacheItem.Resource, "", nil))
 		if err != nil {
 			logger.Infof(ctx, "Error retrieving resource [%s]. Error: %v", resource.GetID(), err)

go/tasks/pluginmachinery/internal/webapi/core.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/gob"
 	"fmt"
+	"runtime"
 	"time"
 
 	"k8s.io/utils/clock"
@@ -36,6 +37,7 @@ const (
 type CorePlugin struct {
 	id             string
 	p              webapi.AsyncPlugin
+	sp             webapi.SyncPlugin
 	cache          cache.AutoRefresh
 	tokenAllocator tokenAllocator
 	metrics        Metrics
@@ -68,7 +70,45 @@ func (c CorePlugin) GetProperties() core.PluginProperties {
 	return core.PluginProperties{}
 }
 
+func (c CorePlugin) syncHandle(ctx context.Context, tCtx core.TaskExecutionContext) (core.Transition, error) {
+	/*
+		_, err := c.unmarshalState(ctx, tCtx.PluginStateReader())
+		if err != nil {
+			return core.UnknownTransition, err
+		}
+
+	*/
+
+	incomingState, err := c.unmarshalState(ctx, tCtx.PluginStateReader())
+	if err != nil {
+		return core.UnknownTransition, err
+	}
+
+	if incomingState.Phase == PhaseNotStarted {
+		// nextState, phaseInfo, err = c.sp.Do(ctx, tCtx, &incomingState)
+	} else {
+		return core.UnknownTransition, err
+	}
+
+	// write a function, syncLaunch in launcher.go
+	logger.Infof(ctx, "@@@ SyncHandle was called")
+	return core.UnknownTransition, nil
+}
+
+/*
+	1. call sp.do
+	2. check error
+	3. write state
+*/
 func (c CorePlugin) Handle(ctx context.Context, tCtx core.TaskExecutionContext) (core.Transition, error) {
+	if c.sp != nil {
+		return c.syncHandle(ctx, tCtx)
+	}
+
+	pc, file, line, _ := runtime.Caller(1)
+	funcName := runtime.FuncForPC(pc).Name()
+	logger.Infof(context.TODO(), "@@@ core.go Handle by [%v] [%v]:[%v]", file, funcName, line)
+
 	incomingState, err := c.unmarshalState(ctx, tCtx.PluginStateReader())
 	if err != nil {
 		return core.UnknownTransition, err
@@ -165,47 +205,92 @@ func createRemotePlugin(pluginEntry webapi.PluginEntry, c clock.Clock) core.Plug
 		RegisteredTaskTypes: pluginEntry.SupportedTaskTypes,
 		LoadPlugin: func(ctx context.Context, iCtx core.SetupContext) (
 			core.Plugin, error) {
-			p, err := pluginEntry.PluginLoader(ctx, iCtx)
+			p, sp, err := pluginEntry.PluginLoader(ctx, iCtx)
 			if err != nil {
 				return nil, err
 			}
 
-			err = validateConfig(p.GetConfig())
-			if err != nil {
-				return nil, fmt.Errorf("config validation failed. Error: %w", err)
-			}
+			if p != nil {
+				err = validateConfig(p.GetConfig())
+				if err != nil {
+					return nil, fmt.Errorf("config validation failed. Error: %w", err)
+				}
 
-			// If the plugin will use a custom state, register it to be able to
-			// serialize/deserialize interfaces later.
-			if customState := p.GetConfig().ResourceMeta; customState != nil {
-				gob.Register(customState)
-			}
+				// If the plugin will use a custom state, register it to be able to
+				// serialize/deserialize interfaces later.
+				if customState := p.GetConfig().ResourceMeta; customState != nil {
+					gob.Register(customState)
+				}
 
-			if quotas := p.GetConfig().ResourceQuotas; len(quotas) > 0 {
-				for ns, quota := range quotas {
-					err := iCtx.ResourceRegistrar().RegisterResourceQuota(ctx, ns, quota)
-					if err != nil {
-						return nil, err
+				if quotas := p.GetConfig().ResourceQuotas; len(quotas) > 0 {
+					for ns, quota := range quotas {
+						err := iCtx.ResourceRegistrar().RegisterResourceQuota(ctx, ns, quota)
+						if err != nil {
+							return nil, err
+						}
 					}
 				}
-			}
 
-			resourceCache, err := NewResourceCache(ctx, pluginEntry.ID, p, p.GetConfig().Caching,
-				iCtx.MetricsScope().NewSubScope("cache"))
+				resourceCache, err := NewResourceCache(ctx, pluginEntry.ID, p, p.GetConfig().Caching,
+					iCtx.MetricsScope().NewSubScope("cache"))
 
-			if err != nil {
-				return nil, err
-			}
+				if err != nil {
+					return nil, err
+				}
 
-			err = resourceCache.Start(ctx)
-			if err != nil {
-				return nil, err
+				err = resourceCache.Start(ctx)
+				if err != nil {
+					return nil, err
+				}
+
+				return CorePlugin{
+					id:             pluginEntry.ID,
+					p:              p,
+					sp:             sp,
+					cache:          resourceCache,
+					metrics:        newMetrics(iCtx.MetricsScope()),
+					tokenAllocator: newTokenAllocator(c),
+				}, nil
+
+			} else if sp != nil {
+				err = validateConfig(p.GetConfig())
+				if err != nil {
+					return nil, fmt.Errorf("config validation failed. Error: %w", err)
+				}
+
+				// If the plugin will use a custom state, register it to be able to
+				// serialize/deserialize interfaces later.
+				if customState := sp.GetConfig().ResourceMeta; customState != nil {
+					gob.Register(customState)
+				}
+
+				if quotas := sp.GetConfig().ResourceQuotas; len(quotas) > 0 {
+					for ns, quota := range quotas {
+						err := iCtx.ResourceRegistrar().RegisterResourceQuota(ctx, ns, quota)
+						if err != nil {
+							return nil, err
+						}
+					}
+				}
+
+				// resourceCache, err := NewResourceCache(ctx, pluginEntry.ID, p, sp.GetConfig().Caching,
+				// 	iCtx.MetricsScope().NewSubScope("cache"))
+
+				// if err != nil {
+				// 	return nil, err
+				// }
+
+				// err = resourceCache.Start(ctx)
+				// if err != nil {
+				// 	return nil, err
+				// }
 			}
 
 			return CorePlugin{
 				id:             pluginEntry.ID,
 				p:              p,
-				cache:          resourceCache,
+				sp:             sp,
+				cache:          nil,
 				metrics:        newMetrics(iCtx.MetricsScope()),
 				tokenAllocator: newTokenAllocator(c),
 			}, nil
@@ -214,5 +299,9 @@ func createRemotePlugin(pluginEntry webapi.PluginEntry, c clock.Clock) core.Plug
 }
 
 func CreateRemotePlugin(pluginEntry webapi.PluginEntry) core.PluginEntry {
+	pc, file, line, _ := runtime.Caller(1)
+	funcName := runtime.FuncForPC(pc).Name()
+	logger.Infof(context.TODO(), "@@@ core.go func CreateRemotePlugin was called by file [%v] [%v]:[%v]", file, funcName, line)
+
 	return createRemotePlugin(pluginEntry, clock.RealClock{})
 }

go/tasks/pluginmachinery/internal/webapi/core_test.go

@@ -83,12 +83,12 @@ func TestCreateRemotePlugin(t *testing.T) {
 	CreateRemotePlugin(webapi.PluginEntry{
 		ID:                 "MyTestPlugin",
 		SupportedTaskTypes: []core.TaskType{"test-task"},
-		PluginLoader: func(ctx context.Context, iCtx webapi.PluginSetupContext) (webapi.AsyncPlugin, error) {
+		PluginLoader: func(ctx context.Context, iCtx webapi.PluginSetupContext) (webapi.AsyncPlugin, webapi.SyncPlugin, error) {
 			return newPluginWithProperties(webapi.PluginConfig{
 				Caching: webapi.CachingConfig{
 					Size: 10,
 				},
-			}), nil
+			}), nil, nil
 		},
 		IsDefault:           false,
 		DefaultForTaskTypes: []core.TaskType{"test-task"},

go/tasks/pluginmachinery/internal/webapi/launcher.go

@@ -2,6 +2,7 @@ package webapi
 
 import (
 	"context"
+	"runtime"
 	"time"
 
 	"github.com/flyteorg/flytestdlib/cache"
@@ -13,6 +14,11 @@ import (
 
 func launch(ctx context.Context, p webapi.AsyncPlugin, tCtx core.TaskExecutionContext, cache cache.AutoRefresh,
 	state *State) (newState *State, phaseInfo core.PhaseInfo, err error) {
+
+	pc, file, line, _ := runtime.Caller(1)
+	funcName := runtime.FuncForPC(pc).Name()
+	logger.Infof(context.TODO(), "@@@ launcher.go launch by [%v] [%v]:[%v]", file, funcName, line)
+
 	rMeta, r, err := p.Create(ctx, tCtx)
 	if err != nil {
 		logger.Errorf(ctx, "Failed to create resource. Error: %v", err)

go/tasks/pluginmachinery/internal/webapi/state.go

@@ -19,6 +19,9 @@ const (
 	// remotely.
 	PhaseAllocationTokenAcquired
 
+	// PhaseResourcesCreating indicates the task is being created remotely.
+	PhaseResourcesCreating
+
 	// PhaseResourcesCreated indicates the task has been created remotely.
 	PhaseResourcesCreated
 

go/tasks/pluginmachinery/registry.go

@@ -2,6 +2,7 @@ package pluginmachinery
 
 import (
 	"context"
+	"runtime"
 	"sync"
 
 	internalRemote "github.com/flyteorg/flyteplugins/go/tasks/pluginmachinery/internal/webapi"
@@ -27,6 +28,14 @@ func PluginRegistry() TaskPluginRegistry {
 }
 
 func (p *taskPluginRegistry) RegisterRemotePlugin(info webapi.PluginEntry) {
+	pc, file, line, _ := runtime.Caller(1)
+	funcName := runtime.FuncForPC(pc).Name()
+	logger.Infof(context.TODO(), "@@@ RegisterRemotePlugin was called by file [%v] [%v]:[%v]", file, funcName, line)
+
+	logger.Infof(context.TODO(), "@@@ Registering Remote plugin Info ID [%v]", info.ID)
+	logger.Infof(context.TODO(), "@@@ Registering Remote plugin Info SupportedTaskTypes [%v]", info.SupportedTaskTypes)
+	logger.Infof(context.TODO(), "@@@ Registering Remote plugin Info PluginLoader [%v]", info.PluginLoader)
+	logger.Infof(context.TODO(), "@@@ Registering Remote plugin Info [%v]", info)
 	ctx := context.Background()
 	if info.ID == "" {
 		logger.Panicf(ctx, "ID is required attribute for k8s plugin")
@@ -43,6 +52,10 @@ func (p *taskPluginRegistry) RegisterRemotePlugin(info webapi.PluginEntry) {
 	p.m.Lock()
 	defer p.m.Unlock()
 	p.corePlugin = append(p.corePlugin, internalRemote.CreateRemotePlugin(info))
+	pluginEntry := internalRemote.CreateRemotePlugin(info)
+	logger.Infof(context.TODO(), "@@@ CreateRemotePlugin ID: %s", pluginEntry.ID)
+	logger.Infof(context.TODO(), "@@@ CreateRemotePlugin RegisteredTaskTypes: %v", pluginEntry.RegisteredTaskTypes)
+	// logger.Infof(context.TODO(), "@@@ internalRemote.CreateRemotePlugin(info) [%v]", internalRemote.CreateRemotePlugin(info))
 }
 
 func CreateRemotePlugin(pluginEntry webapi.PluginEntry) core.PluginEntry {
@@ -51,6 +64,14 @@ func CreateRemotePlugin(pluginEntry webapi.PluginEntry) core.PluginEntry {
 
 // Use this method to register Kubernetes Plugins
 func (p *taskPluginRegistry) RegisterK8sPlugin(info k8s.PluginEntry) {
+	pc, file, line, _ := runtime.Caller(1)
+	funcName := runtime.FuncForPC(pc).Name()
+	logger.Infof(context.TODO(), "@@@ RegisterK8sPlugin was called by file [%v] [%v]:[%v]", file, funcName, line)
+
+	logger.Infof(context.TODO(), "@@@ Registering K8s plugin Info ID [%v]", info.ID)
+	logger.Infof(context.TODO(), "@@@ Registering K8s plugin Info RegisteredTaskTypes [%v]", info.RegisteredTaskTypes)
+	logger.Infof(context.TODO(), "@@@ Registering K8s plugin Info ResourceToWatch [%v]", info.ResourceToWatch)
+	logger.Infof(context.TODO(), "@@@ Registering K8s plugin Info [%v]", info)
 	if info.ID == "" {
 		logger.Panicf(context.TODO(), "ID is required attribute for k8s plugin")
 	}
@@ -70,10 +91,19 @@ func (p *taskPluginRegistry) RegisterK8sPlugin(info k8s.PluginEntry) {
 	p.m.Lock()
 	defer p.m.Unlock()
 	p.k8sPlugin = append(p.k8sPlugin, info)
+	logger.Infof(context.TODO(), "@@@ p.k8sPlugin [%v]", p.corePlugin)
 }
 
 // Use this method to register core plugins
 func (p *taskPluginRegistry) RegisterCorePlugin(info core.PluginEntry) {
+	pc, file, line, _ := runtime.Caller(1)
+	funcName := runtime.FuncForPC(pc).Name()
+	logger.Infof(context.TODO(), "@@@ RegisterCorePlugin was called by file [%v] [%v]:[%v]", file, funcName, line)
+
+	logger.Infof(context.TODO(), "@@@ Registering core plugin Info ID [%v]", info.ID)
+	logger.Infof(context.TODO(), "@@@ Registering core plugin Info RegisteredTaskTypes [%v]", info.RegisteredTaskTypes)
+	logger.Infof(context.TODO(), "@@@ Registering core plugin Info LoadPlugin [%v]", info.LoadPlugin)
+	logger.Infof(context.TODO(), "@@@ Registering core plugin Info [%v]", info)
 	if info.ID == "" {
 		logger.Panicf(context.TODO(), "ID is required attribute for k8s plugin")
 	}
@@ -87,6 +117,7 @@ func (p *taskPluginRegistry) RegisterCorePlugin(info core.PluginEntry) {
 	p.m.Lock()
 	defer p.m.Unlock()
 	p.corePlugin = append(p.corePlugin, info)
+	logger.Infof(context.TODO(), "@@@ p.corePlugin [%v]", p.corePlugin)
 }
 
 // Returns a snapshot of all the registered core plugins.
@@ -107,6 +138,7 @@ type TaskPluginRegistry interface {
 	RegisterK8sPlugin(info k8s.PluginEntry)
 	RegisterCorePlugin(info core.PluginEntry)
 	RegisterRemotePlugin(info webapi.PluginEntry)
+	// RegisterRemoteSyncPlugin(info webapi.PluginEntry)
 	GetCorePlugins() []core.PluginEntry
 	GetK8sPlugins() []k8s.PluginEntry
 }