package taskplugin import ( "bytes" "encoding/json" "fmt" "reflect" "github.com/juju/errors" "github.com/ovh/utask/pkg/jsonschema" "github.com/ovh/utask/pkg/utils" ) // ConfigFunc is a type of function to validate the contents of a configuration payload type ConfigFunc func(interface{}) error // ExecFunc is a type of function to be implemented by a plugin to perform an action in a task type ExecFunc func(string, interface{}, interface{}) (interface{}, interface{}, error) // PluginExecutor is a structure to generate action executors from different implementations // builtin or loaded as custom extensions type PluginExecutor struct { configfunc ConfigFunc execfunc ExecFunc resourcesFunc func(interface{}) []string configFactory func() interface{} pluginName string pluginVersion string contextFactory func(string) interface{} metadataSchema json.RawMessage tagsFunc tagsFunc } // Context generates a context payload to pass to Exec() func (r PluginExecutor) Context(stepName string) interface{} { if r.contextFactory != nil { return r.contextFactory(stepName) } return nil } // Resources returns a list of resources to be used by uTask engine for this plugin func (r PluginExecutor) Resources(baseConfig json.RawMessage, config json.RawMessage) []string { if r.resourcesFunc == nil { return []string{} } var cfg interface{} if r.configFactory != nil { cfg = r.configFactory() if len(baseConfig) > 0 { err := utils.JSONnumberUnmarshal(bytes.NewReader(baseConfig), cfg) if err != nil { return []string{} } } err := utils.JSONnumberUnmarshal(bytes.NewReader(config), cfg) if err != nil { return []string{} } } return r.resourcesFunc(cfg) } // ValidConfig asserts that a given configuration payload complies with the executor's definition func (r PluginExecutor) ValidConfig(baseConfig json.RawMessage, config json.RawMessage) error { if r.configFactory != nil { cfg := r.configFactory() if len(baseConfig) > 0 { err := utils.JSONnumberUnmarshal(bytes.NewReader(baseConfig), cfg) if err != nil { return errors.Annotate(err, "failed to unmarshal base configuration") } } err := utils.JSONnumberUnmarshal(bytes.NewReader(config), cfg) if err != nil { return errors.Annotate(err, "failed to unmarshal configuration") } return r.configfunc(cfg) } return nil } // Exec performs the action implemented by the executor func (r PluginExecutor) Exec(stepName string, baseConfig json.RawMessage, config json.RawMessage, ctx interface{}) (interface{}, interface{}, map[string]string, error) { var cfg interface{} if r.configFactory != nil { cfg = r.configFactory() if len(baseConfig) > 0 { err := utils.JSONnumberUnmarshal(bytes.NewReader(baseConfig), cfg) if err != nil { return nil, nil, nil, errors.Annotate(err, "failed to unmarshal base configuration") } } err := utils.JSONnumberUnmarshal(bytes.NewReader(config), cfg) if err != nil { return nil, nil, nil, errors.Annotate(err, "failed to unmarshal configuration") } } output, metadata, err := r.execfunc(stepName, cfg, ctx) var tags map[string]string if r.tagsFunc != nil { tags = r.tagsFunc(cfg, ctx, output, metadata, err) } return output, metadata, tags, err } // PluginName returns a plugin's name func (r PluginExecutor) PluginName() string { return r.pluginName } // PluginVersion returns a plugin's version func (r PluginExecutor) PluginVersion() string { return r.pluginVersion } // MetadataSchema returns json schema to validate the metadata returned on execution func (r PluginExecutor) MetadataSchema() json.RawMessage { return r.metadataSchema } type tagsFunc func(config, ctx, output, metadata interface{}, err error) map[string]string // PluginOpt is a helper struct to customize an action executor type PluginOpt struct { configCheckFunc ConfigFunc configObj interface{} contextObj interface{} contextFunc func(string) interface{} resourcesFunc func(interface{}) []string metadataFunc func() string tagsFunc tagsFunc } // WithConfig defines the configuration struct and validation function // for a plugin func WithConfig(configCheckFunc ConfigFunc, configObj interface{}) func(*PluginOpt) { return func(o *PluginOpt) { o.configCheckFunc = configCheckFunc o.configObj = configObj } } // WithContext defines the context object expected by the plugin func WithContext(contextObj interface{}) func(*PluginOpt) { return func(o *PluginOpt) { o.contextObj = contextObj } } // WithContextFunc defines a context-generating function func WithContextFunc(contextFunc func(string) interface{}) func(*PluginOpt) { return func(o *PluginOpt) { o.contextFunc = contextFunc } } // WithExecutorMetadata defines a jsonschema-generating function func WithExecutorMetadata(metadataFunc func() string) func(*PluginOpt) { return func(o *PluginOpt) { o.metadataFunc = metadataFunc } } // WithTags defines a function to manipulate the tags of a task. func WithTags(fn tagsFunc) func(*PluginOpt) { return func(o *PluginOpt) { o.tagsFunc = fn } } // WithResources defines a function indicating what resources will be needed by the plugin func WithResources(resourcesFunc func(interface{}) []string) func(*PluginOpt) { return func(o *PluginOpt) { o.resourcesFunc = resourcesFunc } } // New generates a step action executor from a given plugin func New(pluginName string, pluginVersion string, execfunc ExecFunc, opts ...func(*PluginOpt)) PluginExecutor { pOpt := &PluginOpt{} for _, o := range opts { o(pOpt) } if pluginName == "" { panic("registering plugin without name") } if execfunc == nil { panic(fmt.Sprintf("plugin executor '%s': nil exec function", pluginName)) } if pOpt.configObj != nil && pOpt.configCheckFunc == nil { panic(fmt.Sprintf("plugin executor '%s': nil config check function", pluginName)) } if pOpt.contextObj != nil && pOpt.contextFunc != nil { panic(fmt.Sprintf("plugin executor '%s': conflicting context object + factory", pluginName)) } var schema json.RawMessage if pOpt.metadataFunc != nil { metadata := pOpt.metadataFunc() s, err := jsonschema.NormalizeAndCompile(pluginName, []byte(metadata)) if err != nil { panic(fmt.Sprintf("plugin executor %q: %s", pluginName, err.Error())) } schema = s } var contextFactory func(string) interface{} if pOpt.contextFunc != nil { contextFactory = pOpt.contextFunc } else if pOpt.contextObj != nil { v := reflect.ValueOf(pOpt.contextObj) for v.Kind() == reflect.Ptr { v = v.Elem() } marshaled, err := utils.JSONMarshal(pOpt.contextObj) if err != nil { panic(fmt.Sprintf("plugin executor '%s': failed to marshal context object: %s", pluginName, err)) } contextFactory = func(stepName string) interface{} { i := reflect.New(v.Type()).Interface() utils.JSONnumberUnmarshal(bytes.NewReader(marshaled), i) return i } } var configFactory func() interface{} if pOpt.configObj != nil { v := reflect.ValueOf(pOpt.configObj) for v.Kind() == reflect.Ptr { v = v.Elem() } configFactory = func() interface{} { return reflect.New(v.Type()).Interface() } } return PluginExecutor{ pluginName: pluginName, pluginVersion: pluginVersion, configfunc: pOpt.configCheckFunc, execfunc: execfunc, resourcesFunc: pOpt.resourcesFunc, configFactory: configFactory, contextFactory: contextFactory, metadataSchema: schema, tagsFunc: pOpt.tagsFunc, } }