[core] Fix integration plugin loading

Author: teo

core/config.go

@@ -93,8 +93,9 @@ func setDefaults() error {
 	viper.SetDefault("verbose", false)
 	viper.SetDefault("veryVerbose", false)
 	viper.SetDefault("dumpWorkflows", false)
-	viper.SetDefault("configServiceUri", "apricot://localhost:47101")
-	viper.SetDefault("dcsServiceEndpoint", "localhost:50051")
+	viper.SetDefault("configServiceUri", "apricot://127.0.0.1:47101")
+	viper.SetDefault("dcsServiceEndpoint", "127.0.0.1:50051")
+	viper.SetDefault("ddSchedulerEndpoint", "127.0.0.1:50052")
 	viper.SetDefault("integrationPlugins", []string{})
 	viper.SetDefault("coreConfigEntry", "settings")
 	viper.SetDefault("fmqPlugin", "OCClite")
@@ -139,6 +140,7 @@ func setFlags() error {
 	pflag.Bool("dumpWorkflows", viper.GetBool("dumpWorkflows"), "Dump unprocessed and processed workflow files (`$PWD/wf-{,un}processed-<timestamp>.json`)")
 	pflag.String("configServiceUri", viper.GetString("configServiceUri"), "URI of the Apricot instance (`apricot://host:port`), Consul server (`consul://`) or YAML configuration file, entry point for all configuration")
 	pflag.String("dcsServiceEndpoint", viper.GetString("dcsServiceEndpoint"), "Endpoint of the DCS gRPC service (`host:port`)")
+	pflag.String("ddSchedulerEndpoint", viper.GetString("ddSchedulerEndpoint"), "Endpoint of the DD scheduler gRPC service (`host:port`)")
 	pflag.StringSlice("integrationPlugins", viper.GetStringSlice("integrationPlugins"), "List of integration plugins to load (default: empty)")
 	pflag.String("coreConfigEntry", viper.GetString("coreConfigEntry"), "key for AliECS core configuration within the `aliecs` component [EXPERT SETTING]")
 	pflag.String("fmqPlugin", viper.GetString("fmqPlugin"), "Name of the plugin for FairMQ tasks")

core/integration/ddsched/plugin.go

@@ -42,7 +42,7 @@ import (
 	"google.golang.org/grpc/connectivity"
 )
 
-const DCS_CALL_TIMEOUT = 10 * time.Second
+const DDSCHED_CALL_TIMEOUT = 10 * time.Second
 
 
 type Plugin struct {
@@ -81,7 +81,7 @@ func (p *Plugin) GetName() string {
 
 func (p *Plugin) Init(_ string) error {
 	if p.ddSchedClient == nil {
-		callTimeout := DCS_CALL_TIMEOUT
+		callTimeout := DDSCHED_CALL_TIMEOUT
 		cxt, cancel := context.WithTimeout(context.Background(), callTimeout)
 		p.ddSchedClient = NewClient(cxt, cancel, viper.GetString("ddSchedulerEndpoint"))
 		if p.ddSchedClient == nil {

core/integration/plugin.go

@@ -34,10 +34,12 @@ import (
 
 var log = logger.New(logrus.StandardLogger(), "integration")
 
-var(
-	once     sync.Once
+var (
+	once          sync.Once
+	instance      Plugins
+
+	loaderOnce    sync.Once
 	pluginLoaders map[string]func() Plugin
-	instance Plugins
 )
 
 type Plugins []Plugin
@@ -52,7 +54,7 @@ type Plugin interface {
 type NewFunc func(endpoint string) Plugin
 
 func RegisterPlugin(pluginName string, endpointArgumentName string, newFunc NewFunc) {
-	once.Do(func() {
+	loaderOnce.Do(func() {
 		pluginLoaders = make(map[string]func() Plugin)
 	})
 	pluginLoaders[pluginName] = func() Plugin {

core/workflow/callable/call.go

@@ -104,6 +104,7 @@ func (c *Call) Call() error {
 			template.WrapPointer(&returnVar),
 		}
 	c.VarStack["run_number"] = strconv.FormatUint(uint64(c.parentRole.GetCurrentRunNumber()), 10 )
+	c.VarStack["environment_id"] = c.parentRole.GetEnvironmentId().String()
 	objStack := integration.PluginsInstance().ObjectStack(c)
 
 	err := fields.Execute(apricot.Instance(), c.GetName(), c.VarStack, objStack, make(map[string]texttemplate.Template))

core/workflow/roleutils.go

@@ -69,6 +69,11 @@ func Walk(root Role, do func(role Role)) {
 		for _, child := range typed.Roles {
 			Walk(child, do)
 		}
+	case *iteratorRole:
+		do(typed)
+		for _, child := range typed.Roles {
+			LeafWalk(child, do)
+		}
 	case *taskRole:
 		do(typed)
 	case *callRole:
@@ -80,7 +85,11 @@ func LeafWalk(root Role, do func(role Role)) {
 	switch typed := root.(type) {
 	case *aggregatorRole:
 		for _, child := range typed.Roles {
-			Walk(child, do)
+			LeafWalk(child, do)
+		}
+	case *iteratorRole:
+		for _, child := range typed.Roles {
+			LeafWalk(child, do)
 		}
 	case *taskRole:
 		do(typed)