[core] DD scheduler integration plugin

Author: teo

Makefile

@@ -38,6 +38,7 @@ BUILD_FLAGS=$(CGO_LDFLAGS) $(BUILD_ENV_FLAGS)
 endif
 REPOPATH = github.com/AliceO2Group/Control
 ODC_PROTO="https://raw.githubusercontent.com/FairRootGroup/ODC/master/grpc-proto/odc.proto"
+DD_PROTO="https://raw.githubusercontent.com/AliceO2Group/DataDistribution/master/src/DataDistControl/DataDistControl.proto"
 
 VERBOSE_1 := -v
 VERBOSE_2 := -v -x
@@ -54,7 +55,7 @@ WHAT_o2-apricot_BUILD_FLAGS=$(BUILD_ENV_FLAGS)
 INSTALL_WHAT:=$(patsubst %, install_%, $(WHAT))
 
 
-GENERATE_DIRS := ./apricot ./coconut/cmd ./core ./executor ./core/integration/dcs ./odcshim ./walnut
+GENERATE_DIRS := ./apricot ./coconut/cmd ./core ./core/integration/dcs ./core/integration/ddsched ./executor ./odcshim ./walnut
 SRC_DIRS := ./apricot ./cmd/* ./core ./coconut ./executor ./common ./configuration ./occ/peanut ./odcshim ./walnut
 
 # Use linker flags to provide version/build settings to the target
@@ -126,10 +127,15 @@ vendor:
 	@echo -e "\033[1;33mgo mod vendor\033[0m"
 	@go mod vendor
 
+# FIXME: these two protofiles should be committed in order to be available offline
 	@echo -e "\033[1;33mcurl odc.proto\033[0m"
 	@mkdir -p odcshim/odcprotos
 	@curl -s -L $(ODC_PROTO) -o odcshim/odcprotos/odc.proto
 
+	@echo -e "\033[1;33mcurl ddsched.proto\033[0m"
+	@mkdir -p core/integration/ddsched/protos
+	@curl -s -L $(DD_PROTO) -o core/integration/ddsched/protos/ddsched.proto
+
 # WORKAROUND: In order to avoid the following issues:
 # https://github.com/golang/protobuf/issues/992
 # https://github.com/golang/protobuf/issues/1158

cmd/o2-aliecs-core/main.go

@@ -29,11 +29,24 @@ import (
 
 	"github.com/AliceO2Group/Control/common/logger/infologger"
 	"github.com/AliceO2Group/Control/core"
+	"github.com/AliceO2Group/Control/core/integration"
+	"github.com/AliceO2Group/Control/core/integration/dcs"
+	"github.com/AliceO2Group/Control/core/integration/ddsched"
 	log "github.com/sirupsen/logrus"
 	"github.com/teo/logrus-prefixed-formatter"
 )
 
 func init() {
+	// TODO: enable/disable switches for plugins
+	integration.RegisterPlugin(
+		"dcs",
+		"dcsServiceEndpoint",
+		dcs.NewPlugin)
+	integration.RegisterPlugin(
+		"ddsched",
+		"ddSchedulerEndpoint",
+		ddsched.NewPlugin)
+
 	log.SetFormatter(&prefixed.TextFormatter{
 		FullTimestamp:   true,
 		SpacePadding:    20,

core/config.go

@@ -139,7 +139,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("integrationPlugins", viper.GetString("integrationPlugins"), "List of integration plugins to load (default: empty)")
+	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")
 	pflag.String("fmqPluginSearchPath", viper.GetString("fmqPluginSearchPath"), "Path to the directory where the FairMQ plugins are found on controlled nodes")

core/integration/dcs/plugin.go

@@ -35,7 +35,9 @@ import (
 	"strconv"
 	"time"
 
+	"github.com/AliceO2Group/Control/core/integration"
 	dcspb "github.com/AliceO2Group/Control/core/integration/dcs/protos"
+	"github.com/AliceO2Group/Control/core/workflow/callable"
 	"github.com/spf13/viper"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/connectivity"
@@ -50,7 +52,7 @@ type Plugin struct {
 	dcsClient      *RpcClient
 }
 
-func NewPlugin(endpoint string) *Plugin {
+func NewPlugin(endpoint string) integration.Plugin {
 	u, err := url.Parse(endpoint)
 	if err != nil {
 		log.WithField("endpoint", endpoint).
@@ -105,7 +107,12 @@ func (p *Plugin) Init(instanceId string) error {
 	return nil
 }
 
-func (p *Plugin) ObjectStack(varStack map[string]string) (stack map[string]interface{}) {
+func (p *Plugin) ObjectStack(data interface{}) (stack map[string]interface{}) {
+	call, ok := data.(*callable.Call)
+	if !ok {
+		return
+	}
+	varStack := call.VarStack
 	stack = make(map[string]interface{})
 	stack["StartOfRun"] = func() (out string) {	// must formally return string even when we return nothing
 		log.Debug("performing DCS SOR")

core/integration/dcs/protos/dcs.pb.go

@@ -0,0 +1,75 @@
+/*
+ * === This file is part of ALICE O² ===
+ *
+ * Copyright 2021 CERN and copyright holders of ALICE O².
+ * Author: Teo Mrnjavac <teo.mrnjavac@cern.ch>
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ * In applying this license CERN does not waive the privileges and
+ * immunities granted to it by virtue of its status as an
+ * Intergovernmental Organization or submit itself to any jurisdiction.
+ */
+
+package ddsched
+
+import (
+	"context"
+
+	"github.com/AliceO2Group/Control/common/logger"
+	ddpb "github.com/AliceO2Group/Control/core/integration/ddsched/protos"
+	"github.com/sirupsen/logrus"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/connectivity"
+)
+
+var log = logger.New(logrus.StandardLogger(),"ddschedclient")
+
+
+type RpcClient struct {
+	ddpb.DataDistributionControlClient
+	conn *grpc.ClientConn
+}
+
+func NewClient(cxt context.Context, cancel context.CancelFunc, endpoint string) *RpcClient {
+	log.WithFields(logrus.Fields{
+		"endpoint": endpoint,
+	}).Debug("dialing DD scheduler client")
+	conn, err := grpc.DialContext(cxt, endpoint, grpc.WithInsecure(), grpc.WithBlock())
+	if err != nil {
+		log.WithField("error", err.Error()).
+			WithField("endpoint", endpoint).
+			Errorf("cannot dial RPC endpoint")
+		cancel()
+		return nil
+	}
+
+	client := &RpcClient {
+		DataDistributionControlClient: ddpb.NewDataDistributionControlClient(conn),
+		conn: conn,
+	}
+
+	return client
+}
+
+func (m *RpcClient) GetConnState() connectivity.State {
+	if m.conn == nil {
+		return connectivity.Idle
+	}
+	return m.conn.GetState()
+}
+
+func (m *RpcClient) Close() error {
+	return m.conn.Close()
+}