[core] Implementation of vastly expanded control API

Author: teo

core/server.go

@@ -30,6 +30,7 @@ import (
 	"runtime"
 	"time"
 
+	"github.com/AliceO2Group/Control/core/task/channel"
 	"golang.org/x/net/context"
 	"google.golang.org/grpc"
 
@@ -90,11 +91,13 @@ func (m *RpcServer) GetFrameworkInfo(context.Context, *pb.GetFrameworkInfoReques
 	m.logMethod()
 	m.state.RLock()
 	defer m.state.RUnlock()
+
 	r := &pb.GetFrameworkInfoReply{
 		FrameworkId:        store.GetIgnoreErrors(m.fidStore)(),
 		EnvironmentsCount:  int32(len(m.state.environments.Ids())),
 		TasksCount:         int32(m.state.taskman.TaskCount()),
 		State:              m.state.sm.Current(),
+		HostsCount:         int32(m.state.taskman.AgentCache.Count()),
 	}
 	return r, nil
 }
@@ -124,16 +127,14 @@ func (m *RpcServer) GetEnvironments(context.Context, *pb.GetEnvironmentsRequest)
 			continue
 		}
 		tasks := env.Workflow().GetTasks()
-		taskNames := make([]string, len(tasks))
-		for i, task := range tasks {
-			taskNames[i] = task.GetName()
-		}
 		e := &pb.EnvironmentInfo{
 			Id:             env.Id().String(),
 			CreatedWhen:    env.CreatedWhen().Format(time.RFC3339),
 			State:          env.CurrentState(),
-			Tasks:          taskNames,
+			Tasks:          tasksToShortTaskInfos(tasks),
+			RootRole:       env.Workflow().GetName(),
 		}
+
 		r.Environments = append(r.Environments, e)
 	}
 	return r, nil
@@ -164,7 +165,7 @@ func (m *RpcServer) NewEnvironment(cxt context.Context, request *pb.NewEnvironme
 	}
 
 	// Create new Environment instance with some roles, we get back a UUID
-	id, err := m.state.environments.CreateEnvironment(request.GetWorkflow())
+	id, err := m.state.environments.CreateEnvironment(request.GetWorkflowTemplate())
 	if err != nil {
 		return nil, status.Newf(codes.Internal, "cannot create new environment: %s", err.Error()).Err()
 	}
@@ -174,9 +175,15 @@ func (m *RpcServer) NewEnvironment(cxt context.Context, request *pb.NewEnvironme
 		return nil, status.Newf(codes.Internal, "cannot get newly created environment: %s", err.Error()).Err()
 	}
 
+	tasks := newEnv.Workflow().GetTasks()
 	r := &pb.NewEnvironmentReply{
-		Id: id.String(),
-		State: newEnv.Sm.Current(),
+		Environment: &pb.EnvironmentInfo{
+			Id: newEnv.Id().String(),
+			CreatedWhen: newEnv.CreatedWhen().Format(time.RFC3339),
+			State: newEnv.CurrentState(),
+			Tasks: tasksToShortTaskInfos(tasks),
+			RootRole: newEnv.Workflow().GetName(),
+		},
 	}
 
 	return r, nil
@@ -197,17 +204,15 @@ func (m *RpcServer) GetEnvironment(cxt context.Context, req *pb.GetEnvironmentRe
 	}
 
 	tasks := env.Workflow().GetTasks()
-	taskNames := make([]string, len(tasks))
-	for i, task := range tasks {
-		taskNames[i] = task.GetName()
-	}
 	r := &pb.GetEnvironmentReply{
 		Environment: &pb.EnvironmentInfo{
 			Id: env.Id().String(),
 			CreatedWhen: env.CreatedWhen().Format(time.RFC3339),
 			State: env.CurrentState(),
-			Tasks: taskNames,
+			Tasks: tasksToShortTaskInfos(tasks),
+			RootRole: env.Workflow().GetName(),
 		},
+		Workflow: workflowToRoleTree(env.Workflow()),
 	}
 	return r, nil
 }
@@ -291,17 +296,87 @@ func (m *RpcServer) GetTasks(context.Context, *pb.GetTasksRequest) (*pb.GetTasks
 	m.state.RLock()
 	defer m.state.RUnlock()
 
+	tasks := m.state.taskman.GetTasks()
 	r := &pb.GetTasksReply{
-		Tasks: make([]*pb.TaskInfo, 0, 0),
+		Tasks: tasksToShortTaskInfos(tasks),
 	}
 
-	for _, task := range m.state.taskman.GetTasks() {
-		ri := &pb.TaskInfo{
-			Locked:   task.IsLocked(),
-			Hostname: task.GetHostname(),
-			Name:     task.GetName(),
+	return r, nil
+}
+
+func (m *RpcServer) GetTask(cxt context.Context, req *pb.GetTaskRequest) (*pb.GetTaskReply, error) {
+	m.logMethod()
+	m.state.RLock()
+	defer m.state.RUnlock()
+
+	task := m.state.taskman.GetTask(req.TaskId)
+	if task == nil {
+		return &pb.GetTaskReply{}, status.New(codes.NotFound, "task not found").Err()
+	}
+	taskClass := task.GetTaskClass()
+	commandInfo := task.BuildTaskCommand()
+	var outbound []channel.Outbound
+	taskPath := ""
+	// TODO: probably not the nicest way to do this... the outbound assignments should be cached
+	// in the Task
+	if task.IsLocked() {
+		type parentRole interface {
+			CollectOutboundChannels() []channel.Outbound
+			GetPath() string
+		}
+		parent, ok := task.GetParentRole().(parentRole)
+		if ok {
+			outbound = parent.CollectOutboundChannels()
+			taskPath = parent.GetPath()
 		}
-		r.Tasks = append(r.Tasks, ri)
 	}
-	return r, nil
+
+	rep := &pb.GetTaskReply{
+		Task: &pb.TaskInfo{
+			ShortInfo: taskToShortTaskInfo(task),
+			ClassInfo: &pb.TaskClassInfo{
+				Name: task.GetClassName(),
+				ControlMode: taskClass.Control.Mode.String(),
+			},
+			InboundChannels: inboundChannelsToPbChannels(taskClass.Bind),
+			OutboundChannels: outboundChannelsToPbChannels(outbound),
+			CommandInfo: commandInfoToPbCommandInfo(commandInfo),
+			TaskPath: taskPath,
+			EnvId: task.GetEnvironmentId().String(),
+		},
+	}
+	return rep, nil
 }
+
+func (m *RpcServer) CleanupTasks(context.Context, *pb.CleanupTasksRequest) (*pb.CleanupTasksReply, error) {
+	log.WithPrefix("rpcserver").
+		WithField("method", "CleanupTasks").
+		Debug("implement me")
+
+	return &pb.CleanupTasksReply{}, status.New(codes.Unimplemented, "not implemented").Err()
+}
+
+
+func (m *RpcServer) GetRoles(cxt context.Context, req *pb.GetRolesRequest) (*pb.GetRolesReply, error) {
+	m.logMethod()
+	m.state.RLock()
+	defer m.state.RUnlock()
+
+	if req == nil || len(req.EnvId) == 0 {
+		return nil, status.New(codes.InvalidArgument, "received nil request").Err()
+	}
+
+	env, err := m.state.environments.Environment(uuid.Parse(req.EnvId))
+	if err != nil {
+		return nil, status.Newf(codes.NotFound, "environment not found: %s", err.Error()).Err()
+	}
+
+	resultRoles := env.QueryRoles(req.PathSpec)
+
+	roleInfos := make([]*pb.RoleInfo, len(resultRoles))
+	for i, rr := range resultRoles {
+		roleInfos[i] = workflowToRoleTree(rr)
+	}
+	return &pb.GetRolesReply{Roles: roleInfos}, nil
+}
+

core/serverutil.go

@@ -0,0 +1,156 @@
+/*
+ * === This file is part of ALICE O² ===
+ *
+ * Copyright 2018 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 core
+
+import (
+	"github.com/AliceO2Group/Control/common"
+	"github.com/AliceO2Group/Control/core/protos"
+	"github.com/AliceO2Group/Control/core/task/channel"
+
+	"github.com/AliceO2Group/Control/core/task"
+	"github.com/AliceO2Group/Control/core/workflow"
+)
+
+func commandInfoToPbCommandInfo(c *common.TaskCommandInfo) (pci *pb.CommandInfo) {
+	if c == nil {
+		return
+	}
+	pci = &pb.CommandInfo{
+		Env: c.GetEnv(),
+		Shell: c.GetShell(),
+		Value: c.GetValue(),
+		Arguments: c.GetArguments(),
+		User: c.GetUser(),
+	}
+	return
+}
+
+func inboundChannelsToPbChannels(chs []channel.Inbound) (pchs []*pb.ChannelInfo) {
+	if chs == nil {
+		return
+	}
+	pchs = make([]*pb.ChannelInfo, len(chs))
+	for i, c := range chs {
+		pchs[i] = inboundChannelToPbChannel(c)
+	}
+	return
+}
+
+func inboundChannelToPbChannel(ch channel.Inbound) (pch *pb.ChannelInfo) {
+	pch = &pb.ChannelInfo{
+		Name: ch.Name,
+		Type: ch.Type.String(),
+	}
+	return
+}
+
+func outboundChannelsToPbChannels(chs []channel.Outbound) (pchs []*pb.ChannelInfo) {
+	if chs == nil {
+		return
+	}
+	pchs = make([]*pb.ChannelInfo, len(chs))
+	for i, c := range chs {
+		pchs[i] = outboundChannelToPbChannel(c)
+	}
+	return
+}
+
+func outboundChannelToPbChannel(ch channel.Outbound) (pch *pb.ChannelInfo) {
+	pch = &pb.ChannelInfo{
+		Name: ch.Name,
+		Type: ch.Type.String(),
+		Target: ch.Target,
+	}
+	return
+}
+
+func taskToShortTaskInfo(t *task.Task) (sti *pb.ShortTaskInfo) {
+	if t == nil {
+		return
+	}
+	sti = &pb.ShortTaskInfo{
+		Name:   t.GetName(),
+		Locked: t.IsLocked(),
+		TaskId: t.GetTaskId(),
+		Status: "UNKNOWN",
+		State:  "UNKNOWN",
+		ClassName: t.GetClassName(),
+		DeploymentInfo: &pb.TaskDeploymentInfo{
+			Hostname: t.GetHostname(),
+			AgentId: t.GetAgentId(),
+			OfferId: t.GetOfferId(),
+			ExecutorId: t.GetExecutorId(),
+		},
+	}
+	parentRole, ok := t.GetParentRole().(workflow.Role)
+	if ok && parentRole != nil {
+		sti.Status = parentRole.GetStatus().String()
+		sti.State = parentRole.GetState().String()
+	}
+	return
+}
+
+func tasksToShortTaskInfos(tasks []*task.Task) (stis []*pb.ShortTaskInfo) {
+	if tasks == nil {
+		return
+	}
+	stis = make([]*pb.ShortTaskInfo, len(tasks))
+	for i, t := range tasks {
+		shortTaskInfo := taskToShortTaskInfo(t)
+		stis[i] = shortTaskInfo
+	}
+	return
+}
+
+func tasksToTaskIds(tasks []*task.Task) (ss []string) {
+	if tasks == nil {
+		return
+	}
+	ss = make([]string, len(tasks))
+	for i, t := range tasks {
+		ss[i] = t.GetTaskId()
+	}
+	return
+}
+
+func workflowToRoleTree(root workflow.Role) (ri *pb.RoleInfo) {
+	if root == nil {
+		return
+	}
+	childRoles := root.GetRoles()
+	childRoleInfos := make([]*pb.RoleInfo, len(childRoles))
+	for i, cr := range childRoles {
+		childRoleInfos[i] = workflowToRoleTree(cr)
+	}
+	ri = &pb.RoleInfo{
+		Name: root.GetName(),
+		Status: root.GetStatus().String(),
+		State: root.GetState().String(),
+		FullPath: root.GetPath(),
+		TaskIds: tasksToTaskIds(root.GetTasks()),
+		Roles: childRoleInfos,
+	}
+	return
+}