[core] Use confsys.Service and/or configuration.ROSource

Author: teo

core/environment/manager.go

@@ -27,26 +27,27 @@ package environment
 import (
 	"errors"
 	"fmt"
+	"strings"
 	"sync"
-	"github.com/pborman/uuid"
+
+	"github.com/AliceO2Group/Control/core/confsys"
 	"github.com/AliceO2Group/Control/core/task"
 	"github.com/AliceO2Group/Control/core/workflow"
-	"github.com/AliceO2Group/Control/configuration"
-	"strings"
+	"github.com/pborman/uuid"
 )
 
 type Manager struct {
 	mu      sync.RWMutex
 	m       map[uuid.Array]*Environment
 	taskman *task.Manager
-	cfg     configuration.Source
+	confSvc *confsys.Service
 }
 
-func NewEnvManager(tm *task.Manager, cfg configuration.Source) *Manager {
+func NewEnvManager(tm *task.Manager, confSvc *confsys.Service) *Manager {
 	return &Manager{
 		m:       make(map[uuid.Array]*Environment),
 		taskman: tm,
-		cfg: cfg,
+		confSvc: confSvc,
 	}
 }
 
@@ -137,5 +138,5 @@ func (envs *Manager) loadWorkflow(workflowPath string, parent workflow.Updatable
 	if strings.Contains(workflowPath, "://") {
 		return nil, errors.New("workflow loading from file not implemented yet")
 	}
-	return workflow.Load(envs.cfg, workflowPath, parent)
+	return workflow.Load(envs.confSvc.GetROSource(), workflowPath, parent)
 }

core/server.go

@@ -462,7 +462,7 @@ func (m *RpcServer) GetWorkflowTemplates(cxt context.Context, req *pb.GetWorkflo
 	m.state.RLock()
 	defer m.state.RUnlock()
 
-	wfTree, err := m.state.cfgman.GetRecursive("o2/control/workflows")
+	wfTree, err := m.state.confSvc.GetROSource().GetRecursive("o2/control/workflows")
 	if err != nil {
 		return nil, status.New(codes.FailedPrecondition, "cannot query available workflows").Err()
 	}

core/state.go

@@ -32,12 +32,12 @@ import (
 	"time"
 	"sync"
 
+	"github.com/AliceO2Group/Control/core/confsys"
 	"github.com/mesos/mesos-go/api/v1/lib"
 	"github.com/mesos/mesos-go/api/v1/lib/backoff"
 	"github.com/mesos/mesos-go/api/v1/lib/scheduler/calls"
 	"github.com/looplab/fsm"
 	"github.com/AliceO2Group/Control/core/environment"
-	"github.com/AliceO2Group/Control/configuration"
 	"encoding/json"
 	"github.com/AliceO2Group/Control/core/controlcommands"
 	"context"
@@ -61,9 +61,12 @@ func newInternalState(cfg Config, shutdown func()) (*internalState, error) {
 		return nil, err
 	}
 
-	cfgman, err := configuration.NewSource(cfg.configurationUri)
+	confSvc, err := confsys.NewService(cfg.configurationUri)
+	if err != nil {
+		return nil, err
+	}
 	if cfg.veryVerbose {
-		cfgDump, err := cfgman.GetRecursive("o2/control")
+		cfgDump, err := confSvc.GetROSource().GetRecursive("o2/control")
 		if err != nil {
 			log.WithError(err).Fatal("cannot retrieve configuration")
 			return nil, err
@@ -75,9 +78,6 @@ func newInternalState(cfg Config, shutdown func()) (*internalState, error) {
 		}
 		log.WithField("data", string(cfgBytes)).Debug("configuration dump")
 	}
-	if err != nil {
-		return nil, err
-	}
 
 	resourceOffersDone := make(chan task.DeploymentMap)
 	tasksToDeploy := make(chan task.Descriptors)
@@ -96,7 +96,7 @@ func newInternalState(cfg Config, shutdown func()) (*internalState, error) {
 		random:             rand.New(rand.NewSource(time.Now().Unix())),
 		shutdown:           shutdown,
 		environments:       nil,
-		cfgman:             cfgman,
+		confSvc:            confSvc,
 	}
 
 	state.servent = controlcommands.NewServent(
@@ -106,14 +106,14 @@ func newInternalState(cfg Config, shutdown func()) (*internalState, error) {
 	)
 	state.commandqueue = controlcommands.NewCommandQueue(state.servent)
 
-	taskman := task.NewManager(cfgman, resourceOffersDone,
+	taskman := task.NewManager(confSvc.GetROSource(), resourceOffersDone,
 		tasksToDeploy, reviveOffersTrg, state.commandqueue)
 	err = taskman.RefreshClasses()
 	if err != nil {
 		log.WithField("error", err).Warning("bad configuration, some task templates were not refreshed")
 	}
 	state.taskman = taskman
-	state.environments = environment.NewEnvManager(state.taskman, state.cfgman)
+	state.environments = environment.NewEnvManager(state.taskman, confSvc)
 	state.commandqueue.Start()	// FIXME: there should be 1 cq per env
 
 	return state, nil
@@ -149,8 +149,9 @@ type internalState struct {
 	sm           *fsm.FSM
 	environments *environment.Manager
 	taskman      *task.Manager
-	cfgman       configuration.Source
 	commandqueue *controlcommands.CommandQueue
 	servent      *controlcommands.Servent
+
+	confSvc      *confsys.Service
 }
 

core/task/manager.go

@@ -46,14 +46,14 @@ type Manager struct {
 	classes            map[string]*TaskClass
 	roster             Tasks
 
-	cfgman             configuration.Source
+	cfgman             configuration.ROSource
 	resourceOffersDone <-chan DeploymentMap
 	tasksToDeploy      chan<- Descriptors
 	reviveOffersTrg    chan struct{}
 	cq                 *controlcommands.CommandQueue
 }
 
-func NewManager(cfgman configuration.Source,
+func NewManager(cfgman configuration.ROSource,
                 resourceOffersDone <-chan DeploymentMap,
                 tasksToDeploy chan<- Descriptors,
                 reviveOffersTrg chan struct{},

core/workflow/load.go

@@ -32,7 +32,7 @@ import (
 )
 
 // FIXME: workflowPath should be of type configuration.Path, not string
-func Load(cfg configuration.Source, workflowPath string, parent Updatable) (workflow Role, err error) {
+func Load(cfg configuration.ROSource, workflowPath string, parent Updatable) (workflow Role, err error) {
 	completePath := fmt.Sprintf("%s/%s", ConfigBasePath, strings.Trim(workflowPath, "/"))
 	var yamlDoc []byte
 	yamlDoc, err = cfg.GetRecursiveYaml(completePath)