[core] Initial plumbing for task cleanup support

Author: teo

coconut/protos/o2control.pb.go

@@ -227,7 +227,10 @@ message TaskInfo {
 message CleanupTasksRequest {
     repeated string taskIds = 1;
 }
-message CleanupTasksReply {}
+message CleanupTasksReply {
+    repeated ShortTaskInfo killedTasks = 1;
+    repeated ShortTaskInfo runningTasks = 2;
+}
 
 ////////////////////////////////////////
 // Roles

core/protos/o2control.pb.go

@@ -30,6 +30,7 @@ import (
 	"runtime"
 	"time"
 
+	"github.com/AliceO2Group/Control/core/task"
 	"github.com/AliceO2Group/Control/core/task/channel"
 	"golang.org/x/net/context"
 	"google.golang.org/grpc"
@@ -348,12 +349,29 @@ func (m *RpcServer) GetTask(cxt context.Context, req *pb.GetTaskRequest) (*pb.Ge
 	return rep, nil
 }
 
-func (m *RpcServer) CleanupTasks(context.Context, *pb.CleanupTasksRequest) (*pb.CleanupTasksReply, error) {
-	log.WithPrefix("rpcserver").
-		WithField("method", "CleanupTasks").
-		Debug("implement me")
+func (m *RpcServer) CleanupTasks(cxt context.Context, req *pb.CleanupTasksRequest) (*pb.CleanupTasksReply, error) {
+	m.logMethod()
+	m.state.Lock()
+	defer m.state.Unlock()
+	idsToKill := req.GetTaskIds()
+	var(
+		killedTasks, runningTasks task.Tasks
+		err error
+	)
+	if len(idsToKill) == 0 { // by default we try to kill all, best effort
+		killedTasks, runningTasks, err = m.state.taskman.Cleanup()
+	} else {
+		killedTasks, runningTasks, err = m.state.taskman.KillTasks(idsToKill)
+	}
+
+	if err != nil {
+		log.WithError(err).Error("task cleanup error")
+	}
+	killed := tasksToShortTaskInfos(killedTasks)
+	running := tasksToShortTaskInfos(runningTasks)
 
-	return &pb.CleanupTasksReply{}, status.New(codes.Unimplemented, "not implemented").Err()
+	// FIXME: implement doKillTasks in task.Manager, then remove codes.Unimplemented
+	return &pb.CleanupTasksReply{KilledTasks: killed, RunningTasks: running}, status.New(codes.Unimplemented, "not implemented").Err()
 }
 
 

core/protos/o2control.proto

@@ -32,65 +32,65 @@ import (
 
 type TaskError interface {
 	error
-	GetTaskName() string
+	GetTaskId() string
 }
 
 type TasksError interface {
 	error
-	GetTaskNames() []string
+	GetTaskIds() []string
 }
 
 type taskErrorBase struct {
-	taskName string
+	taskId string
 }
-func (r taskErrorBase) GetTaskName() string {
-	return r.taskName
+func (r taskErrorBase) GetTaskId() string {
+	return r.taskId
 }
 
 type tasksErrorBase struct {
-	taskNames []string
+	taskIds []string
 }
-func (r tasksErrorBase) GetTaskNames() []string {
-	return r.taskNames
+func (r tasksErrorBase) GetTaskIds() []string {
+	return r.taskIds
 }
 
 type GenericTaskError struct {
 	taskErrorBase
 	message string
 }
 func (r GenericTaskError) Error() string {
-	return fmt.Sprintf("task %s error: %s", r.taskName, r.message)
+	return fmt.Sprintf("task %s error: %s", r.taskId, r.message)
 }
 
 type GenericTasksError struct {
 	tasksErrorBase
 	message string
 }
 func (r GenericTasksError) Error() string {
-	return fmt.Sprintf("tasks [%s] error: %s", strings.Join(r.taskNames, ", "), r.message)
+	return fmt.Sprintf("tasks [%s] error: %s", strings.Join(r.taskIds, ", "), r.message)
 }
 
 type TasksDeploymentError tasksErrorBase
 func (r TasksDeploymentError) Error() string {
-	return fmt.Sprintf("deployment failed for tasks [%s]", r.taskNames)
+	return fmt.Sprintf("deployment failed for tasks [%s]", r.taskIds)
 }
 
 type TaskAlreadyReleasedError taskErrorBase
 func (r TaskAlreadyReleasedError) Error() string {
-	return fmt.Sprintf("task %s already released", r.taskName)
+	return fmt.Sprintf("task %s already released", r.taskId)
 }
 
 type TaskNotFoundError taskErrorBase
 func (r TaskNotFoundError) Error() string {
-	return fmt.Sprintf("task %s not found", r.taskName)
+	return fmt.Sprintf("task %s not found", r.taskId)
 }
 
 type TaskLockedError struct {
 	taskErrorBase
 	envId uuid.Array
 }
 func (r TaskLockedError) Error() string {
-	return fmt.Sprintf("task %s is locked by environment %s", r.taskName, r.envId)
+	return fmt.Sprintf("task %s is locked by environment %s", r.taskId, r.envId)
 }
 func (r TaskLockedError) EnvironmentId() uuid.Array {
 	return r.envId

core/server.go

@@ -265,7 +265,7 @@ func (m *Manager) AcquireTasks(envId uuid.Array, taskDescriptors Descriptors) (e
 			taskPtr.parent = nil
 			deployedTaskNames = append(deployedTaskNames, taskPtr.name)
 		}
-		err = TasksDeploymentError{taskNames: deployedTaskNames}
+		err = TasksDeploymentError{taskIds: deployedTaskNames}
 	}
 
 	// Finally, we write to the roster. Point of no return!
@@ -283,14 +283,6 @@ func (m *Manager) AcquireTasks(envId uuid.Array, taskDescriptors Descriptors) (e
 	return
 }
 
-func (m *Manager) TeardownTasks(roleNames []string) error {
-	m.mu.Lock()
-	defer m.mu.Unlock()
-	//FIXME: implement
-
-	return nil
-}
-
 func (m *Manager) ReleaseTasks(envId uuid.Array, tasks Tasks) error {
 	m.mu.Lock()
 	defer m.mu.Unlock()
@@ -315,7 +307,7 @@ func (m *Manager) releaseTask(envId uuid.Array, task *Task) error {
 		return TaskNotFoundError{}
 	}
 	if task.IsLocked() && task.GetEnvironmentId() != envId {
-		return TaskLockedError{taskErrorBase: taskErrorBase{taskName: task.name}, envId: envId}
+		return TaskLockedError{taskErrorBase: taskErrorBase{taskId: task.name}, envId: envId}
 	}
 
 	task.parent = nil
@@ -477,4 +469,49 @@ func (m *Manager) UpdateTaskStatus(status *mesos.TaskStatus) {
 	}
 }
 
+func (m *Manager) Cleanup() (killed Tasks, running Tasks, err error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	toKill := m.roster.Filtered(func(t *Task) bool {
+		return !t.IsLocked()
+	})
+
+	killed, running, err = m.doKillTasks(toKill)
+	return
+}
+
+func (m *Manager) KillTasks(taskIds []string) (killed Tasks, running Tasks, err error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	toKill := m.roster.Filtered(func(t *Task) bool {
+		if t.IsLocked() {
+			return false
+		}
+		for _, id := range taskIds {
+			if t.taskId == id {
+				return true
+			}
+		}
+		return false
+	})
+
+	if len(toKill) < len(taskIds) {
+		err = TaskNotFoundError{}
+		return
+	}
+
+	killed, running, err = m.doKillTasks(toKill)
+	return
+}
+
+func (m *Manager) doKillTasks(tasks Tasks) (killed Tasks, running Tasks, err error) {
+	// FIXME: implement
+	// transition to done
+	// if it fails, kill them all
+	log.Error("no tasks killed: not implemented yet")
+	return make(Tasks, 0), m.roster, nil
+}
+