/*

* === This file is part of ALICE O² ===

*

* Copyright 2018 CERN and copyright holders of ALICE O².

* Author: Teo Mrnjavac <teo.mrnjavac@cern.ch>

*

* Portions from examples in <https://github.com/mesos/mesos-go>:

* Copyright 2013-2015, Mesosphere, Inc.

*

* 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.

*/

//go:generate protoc -I ../occ --gofast_out=plugins=grpc:. protos/occ.proto

// Package executor implements the O² Control executor binary.

package executor

import (

"context"

"encoding/json"

"errors"

"fmt"

"io"

"net/url"

"os"

"os/exec"

"strings"

"sync"

"time"

"github.com/AliceO2Group/Control/common"

"github.com/AliceO2Group/Control/common/event"

"github.com/AliceO2Group/Control/common/logger"

"github.com/AliceO2Group/Control/executor/executorcmd"

"github.com/AliceO2Group/Control/executor/protos"

"github.com/golang/protobuf/proto"

"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/encoding"

"github.com/mesos/mesos-go/api/v1/lib/encoding/codecs"

"github.com/mesos/mesos-go/api/v1/lib/executor"

"github.com/mesos/mesos-go/api/v1/lib/executor/calls"

"github.com/mesos/mesos-go/api/v1/lib/executor/config"

"github.com/mesos/mesos-go/api/v1/lib/executor/events"

"github.com/mesos/mesos-go/api/v1/lib/httpcli"

"github.com/mesos/mesos-go/api/v1/lib/httpcli/httpexec"

"github.com/pborman/uuid"

"github.com/sirupsen/logrus"

"google.golang.org/grpc"

)

const (

apiPath = "/api/v1/executor"

httpTimeout = 10 * time.Second

startupPollingInterval = 500 * time.Millisecond

startupTimeout = 30 * time.Second

)

var log = logger.New(logrus.StandardLogger(), "executor")

var errMustAbort = errors.New("executor received abort signal from Mesos, will attempt to re-subscribe")

// maybeReconnect returns a backoff.Notifier chan if framework checkpointing is enabled.

func maybeReconnect(cfg config.Config) <-chan struct{} {

if cfg.Checkpoint {

return backoff.Notifier(1*time.Second, cfg.SubscriptionBackoffMax*3/4, nil)

}

return nil

}

// Run is the actual entry point of the executor.

func Run(cfg config.Config) {

var (

apiURL = url.URL{

Scheme: "http",

Host: cfg.AgentEndpoint,

Path: apiPath,

}

http = httpcli.New(

httpcli.Endpoint(apiURL.String()),

httpcli.Codec(codecs.ByMediaType[codecs.MediaTypeProtobuf]),

httpcli.Do(httpcli.With(httpcli.Timeout(httpTimeout))),

)

// Fill in the Framework and Executor IDs as call parameters

callOptions = executor.CallOptions{

calls.Framework(cfg.FrameworkID),

calls.Executor(cfg.ExecutorID),

}

state = &internalState{

// With this we inject the callOptions into all outgoing calls

cli: calls.SenderWith(

httpexec.NewSender(http.Send),

callOptions...,

),

// The executor keeps lists of unacknowledged tasks and unacknowledged updates. In case of unexpected

// disconnection, the executor should SUBSCRIBE again and send these lists to Mesos in the SUBSCRIBE

// call.

unackedTasks: make(map[mesos.TaskID]mesos.TaskInfo),

unackedUpdates: make(map[string]executor.Call_Update),

failedTasks: make(map[mesos.TaskID]mesos.TaskStatus),

rpcClients: make(map[mesos.TaskID]*executorcmd.RpcClient),

}

subscriber = calls.SenderWith(

// Here too, callOptions for all outgoing subscriber calls

httpexec.NewSender(http.Send, httpcli.Close(true)),

callOptions...,

)

// Chan which receives a struct every once in a while

shouldReconnect = maybeReconnect(cfg)

disconnected = time.Now()

handler = buildEventHandler(state)

)

// Main loop for (re)subscription. Once we're subscribed, we jump into the event loop for handling the agent.

for {

func() {

// We create the subscription call. If we haven't had an unclean disconnect, the lists of unacknowledged

// tasks and updates are empty.

state.mu.RLock()

subscribe := calls.Subscribe(unacknowledgedTasks(state), unacknowledgedUpdates(state))

state.mu.RUnlock()

log.Info("subscribing to agent for events")

// ↓ empty context ↓ we get a plain RequestFunc from the executor.Call value

resp, err := subscriber.Send(context.TODO(), calls.NonStreaming(subscribe))

if resp != nil {

defer resp.Close()

}

if err == nil {

log.Info("subscription ok, ready to receive events")

// We're officially connected, start decoding events

err = eventLoop(state, resp, handler)

// If we're out of the eventLoop, means a disconnect happened, willingly or not.

disconnected = time.Now()

log.Info("event loop finished")

}

if err != nil && err != io.EOF {

log.WithField("error", err).Error("executor disconnected with error")

} else {

log.Info("executor disconnected")

}

}()

if state.shouldQuit {

log.Info("gracefully shutting down because we were told to")

return

}

// The purpose of checkpointing is to handle recovery when mesos-agent exits.

if !cfg.Checkpoint {

log.Info("gracefully exiting because framework checkpointing is NOT enabled")

return

}

if time.Now().Sub(disconnected) > cfg.RecoveryTimeout {

log.WithField("timeout", cfg.RecoveryTimeout).

Error("failed to re-establish subscription with agent within recovery timeout, aborting")

return

}

log.Info("waiting for reconnect timeout")

<-shouldReconnect // wait for some amount of time before retrying subscription

}

}

// unacknowledgedTasks generates the value of the UnacknowledgedTasks field of a Subscribe call.

func unacknowledgedTasks(state *internalState) (result []mesos.TaskInfo) {

if n := len(state.unackedTasks); n > 0 {

result = make([]mesos.TaskInfo, 0, n)

for k := range state.unackedTasks {

result = append(result, state.unackedTasks[k])

}

}

return

}

// unacknowledgedUpdates generates the value of the UnacknowledgedUpdates field of a Subscribe call.

func unacknowledgedUpdates(state *internalState) (result []executor.Call_Update) {

if n := len(state.unackedUpdates); n > 0 {

result = make([]executor.Call_Update, 0, n)

for k := range state.unackedUpdates {

result = append(result, state.unackedUpdates[k])

}

}

return

}

// eventLoop dispatches incoming events from mesos-agent to the events.Handler (built in buildEventhandler).

func eventLoop(state *internalState, decoder encoding.Decoder, h events.Handler) (err error) {

log.Info("listening for events from agent")

ctx := context.TODO() // dummy context

for err == nil && !state.shouldQuit {

log.Debug("begin new event loop iteration")

// housekeeping

state.mu.Lock()

sendFailedTasks(state)

state.mu.Unlock()

var e executor.Event

if err = decoder.Decode(&e); err == nil {

err = h.HandleEvent(ctx, &e)

}

}

return err

}

// buildEventHandler builds an events.Handler, whose HandleEvent is triggered from the eventLoop.

func buildEventHandler(state *internalState) events.Handler {

return events.HandlerFuncs{

executor.Event_SUBSCRIBED: func(_ context.Context, e *executor.Event) error {

log.WithField("event", e.Type.String()).Debug("handling event")

state.mu.Lock()

// With this event we get FrameworkInfo, ExecutorInfo, AgentInfo:

state.framework = e.Subscribed.FrameworkInfo

state.executor = e.Subscribed.ExecutorInfo

state.agent = e.Subscribed.AgentInfo

state.mu.Unlock()

return nil

},

executor.Event_LAUNCH: func(_ context.Context, e *executor.Event) error {

// Launch one task. We're not handling LAUNCH_GROUP.

log.WithField("event", e.Type.String()).Debug("handling event")

launch(state, e.Launch.Task)

return nil

},

executor.Event_KILL: func(_ context.Context, e *executor.Event) error {

log.WithField("event", e.Type.String()).Debug("handling event")

// TODO: ask the process to kindly transition to the end state and quit, and if it doesn't do so gracefully,

// kill it and report back with a MESSAGE.

log.Warning("event KILL not implemented")

return nil

},

executor.Event_ACKNOWLEDGED: func(_ context.Context, e *executor.Event) error {

log.WithField("event", e.Type.String()).Debug("handling event")

state.mu.Lock()

delete(state.unackedTasks, e.Acknowledged.TaskID)

delete(state.unackedUpdates, string(e.Acknowledged.UUID))

state.mu.Unlock()

return nil

},

executor.Event_MESSAGE: func(_ context.Context, e *executor.Event) error {

log.WithField("event", e.Type.String()).Debug("handling event")

log.WithFields(logrus.Fields{

"length": len(e.Message.Data),

"message": string(e.Message.Data[:]),

}).

Debug("received message data")

err := handleMessage(state, e.Message.Data)

if err != nil {

log.WithField("error", err.Error()).Debug("MESSAGE handler error")

}

return err

},

executor.Event_SHUTDOWN: func(_ context.Context, e *executor.Event) error {

log.WithField("event", e.Type.String()).Debug("handling event")

state.shouldQuit = true

return nil

},

executor.Event_ERROR: func(_ context.Context, e *executor.Event) error {

log.WithField("event", e.Type.String()).Debug("handling event")

return errMustAbort

},

}.Otherwise(func(_ context.Context, e *executor.Event) error {

log.Fatal("unexpected event", e)

return nil

})

}

// sendFailedTasks runs on every iteration of eventLoop to send an UPDATE on any failed tasks to the agent.

func sendFailedTasks(state *internalState) {

for taskID, status := range state.failedTasks {

updateErr := update(state, status)

if updateErr != nil {

log.WithFields(logrus.Fields{

"taskId": taskID.Value,

"error": updateErr,

}).

Error("failed to send status update for task")

} else {

// If we have successfully notified Mesos, we clear our list of failed tasks.

delete(state.failedTasks, taskID)

}

}

}

func handleMessage(state *internalState, data []byte) (err error) {

var incoming struct {

Name string `json:"name"`

TargetList []struct{

TaskId mesos.TaskID

} `json:"targetList"`

}

err = json.Unmarshal(data, &incoming)

if err != nil {

return

}

if len(incoming.TargetList) != 1 {

err = fmt.Errorf("cannot apply ExecutorCommand with %s!=1 target taskIds", len(incoming.TargetList))

return

}

log.WithField("name", incoming.Name).

WithField("payload", string(data[:])).

Debug("processing incoming MESSAGE")

switch incoming.Name {

case "MesosCommand_Transition":

taskId := incoming.TargetList[0].TaskId

state.mu.RLock()

// Check whether we have a control connection to the running task

_, ok := state.rpcClients[taskId]

if !ok {

err = fmt.Errorf("no RPC client for taskId %s", taskId.Value)

log.WithFields(logrus.Fields{

"name": incoming.Name,

"message": string(data[:]),

"error": err.Error(),

}).

Error("cannot unmarshal incoming MESSAGE")

state.mu.RUnlock()

return

}

var cmd *executorcmd.ExecutorCommand_Transition

cmd, err = state.rpcClients[taskId].UnmarshalTransition(data)

if err != nil {

log.WithFields(logrus.Fields{

"name": cmd.Name,

"message": string(data[:]),

"error": err.Error(),

}).

Error("cannot unmarshal incoming MESSAGE")

state.mu.RUnlock()

return

}

state.mu.RUnlock()

if cmd.Name == "CONFIGURE" {

log.WithFields(logrus.Fields{"map": cmd.Arguments, "taskId": taskId}).Debug("CONFIGURE pushing FairMQ properties")

}

newState, transitionError := cmd.Commit()

go func() {

response := cmd.PrepareResponse(transitionError, newState, taskId.Value)

data, marshalError := json.Marshal(response)

if marshalError != nil {

log.WithFields(logrus.Fields{

"commandName": response.GetCommandName(),

"commandId": response.GetCommandId(),

"error": response.Err().Error(),

"marshalError": marshalError,

}).

Error("cannot marshal MesosCommandResponse for sending as MESSAGE")

return

}

state.mu.Lock()

defer state.mu.Unlock()

state.cli.Send(context.TODO(), calls.NonStreaming(calls.Message(data)))

log.WithFields(logrus.Fields{

"commandName": response.GetCommandName(),

"commandId": response.GetCommandId(),

"error": response.Err().Error(),

"state": response.CurrentState,

}).

Debug("response sent")

}()

default:

err = errors.New(fmt.Sprintf("unrecognized controlcommand %s", incoming.Name))

}

return

}

// launch tries to launch a task described by a mesos.TaskInfo.

func launch(state *internalState, task mesos.TaskInfo) {

state.mu.Lock()

state.unackedTasks[task.TaskID] = task

jsonTask, _ := json.MarshalIndent(task, "", "\t")

log.WithField("payload", fmt.Sprintf("%s", jsonTask[:])).Debug("received task to launch")

status := newStatus(state, task.TaskID)

var commandInfo common.TaskCommandInfo

tciData := task.GetData()

log.WithField("json", string(tciData[:])).Debug("received TaskCommandInfo")

if err := json.Unmarshal(tciData, &commandInfo); tciData != nil && err == nil {

log.WithFields(logrus.Fields{

"shell": *commandInfo.Shell,

"value": *commandInfo.Value,

"args": commandInfo.Arguments,

"task": task.Name,

}).

Info("launching task")

} else {

if err != nil {

log.WithError(err).WithField("task", task.Name).Error("could not launch task")

} else {

log.WithError(errors.New("command data is nil")).WithField("task", task.Name).Error("could not launch task")

}

status.State = mesos.TASK_FAILED.Enum()

status.Message = protoString("TaskInfo.Data is nil")

state.failedTasks[task.TaskID] = status

state.mu.Unlock()

return

}

state.mu.Unlock()

var taskCmd *exec.Cmd

if *commandInfo.Shell {

rawCommand := strings.Join(append([]string{*commandInfo.Value}, commandInfo.Arguments...), " ")

taskCmd = exec.Command("/bin/sh", []string{"-c", rawCommand}...)

} else {

taskCmd = exec.Command(*commandInfo.Value, commandInfo.Arguments...)

}

taskCmd.Env = append(os.Environ(), commandInfo.Env...)

var errStdout, errStderr error

stdoutIn, _ := taskCmd.StdoutPipe()

stderrIn, _ := taskCmd.StderrPipe()

log.WithField("payload", string(task.GetData()[:])).WithField("task", task.Name).Debug("starting task")

err := taskCmd.Start()

if err != nil {

log.WithFields(logrus.Fields{

"id": task.TaskID.Value,

"task": task.Name,

"error": err,

"command": *commandInfo.Value,

}).

Error("failed to run task")

status.State = mesos.TASK_FAILED.Enum()

status.Message = protoString(err.Error())

state.mu.Lock()

state.failedTasks[task.TaskID] = status

state.mu.Unlock()

return

}

log.WithField("id", task.TaskID.Value).WithField("task", task.Name).Debug("task started")

go func() {

_, errStdout = io.Copy(log.WithPrefix("task-stdout").WithField("task", task.Name).Writer(), stdoutIn)

}()

go func() {

_, errStderr = io.Copy(log.WithPrefix("task-stderr").WithField("task", task.Name).Writer(), stderrIn)

}()

state.mu.Lock()

log.WithFields(logrus.Fields{

"controlPort": commandInfo.ControlPort,

"controlMode": commandInfo.ControlMode.String(),

"task": task.Name,

"id": task.TaskID.Value,

}).

Debug("starting gRPC client")

state.rpcClients[task.TaskID] = executorcmd.NewClient(commandInfo.ControlPort, commandInfo.ControlMode)

state.mu.Unlock()

go func() {

elapsed := 0 * time.Second

for {

log.WithFields(logrus.Fields{

"id": task.TaskID.Value,

"task": task.Name,

"elapsed": elapsed.String(),

}).

Debug("polling task for IDLE state reached")

state.mu.RLock()

response, err := state.rpcClients[task.TaskID].GetState(context.TODO(), &pb.GetStateRequest{}, grpc.EmptyCallOption{})

if err != nil {

log.WithError(err).WithField("task", task.Name).Error("cannot query task status")

} else {

log.WithField("state", response.GetState()).WithField("task", task.Name).Debug("task status queried")

}

// NOTE: we acquire the transitioner-dependent STANDBY equivalent state

reachedState := state.rpcClients[task.TaskID].FromDeviceState(response.GetState())

state.mu.RUnlock()

if reachedState == "STANDBY" && err == nil {

log.WithField("id", task.TaskID.Value).

WithField("task", task.Name).

Debug("task running and ready for control input")

break

} else if elapsed >= startupTimeout {

err = errors.New("timeout while waiting for task startup")

log.WithField("task", task.Name).Error(err.Error())

status.State = mesos.TASK_FAILED.Enum()

status.Message = protoString(err.Error())

state.mu.Lock()

state.failedTasks[task.TaskID] = status

state.rpcClients[task.TaskID].Close()

delete(state.rpcClients, task.TaskID)

state.mu.Unlock()

return

} else {

log.WithField("task", task.Name).Debugf("task not ready yet, waiting %s", startupPollingInterval.String())

time.Sleep(startupPollingInterval)

elapsed += startupPollingInterval

}

}

// Set up event stream from task

state.mu.RLock()

esc, err := state.rpcClients[task.TaskID].EventStream(context.TODO(), &pb.EventStreamRequest{}, grpc.EmptyCallOption{})

state.mu.RUnlock()

if err != nil {

log.WithField("task", task.Name).WithError(err).Error("cannot set up event stream from task")

status.State = mesos.TASK_FAILED.Enum()

status.Message = protoString(err.Error())

state.mu.Lock()

state.failedTasks[task.TaskID] = status

state.rpcClients[task.TaskID].Close()

delete(state.rpcClients, task.TaskID)

state.mu.Unlock()

return

}

log.WithField("task", task.Name).Debug("notifying of task running state")

status.State = mesos.TASK_RUNNING.Enum()

// send RUNNING

state.mu.Lock()

err = update(state, status)

if err != nil {

log.WithFields(logrus.Fields{

"id": task.TaskID.Value,

"task": task.Name,

"error": err.Error(),

}).

Error("failed to send TASK_RUNNING")

status.State = mesos.TASK_FAILED.Enum()

status.Message = protoString(err.Error())

state.failedTasks[task.TaskID] = status

state.rpcClients[task.TaskID].Close()

delete(state.rpcClients, task.TaskID)

if taskCmd.Process != nil {

log.WithFields(logrus.Fields{

"process": taskCmd.Process.Pid,

"id": task.TaskID.Value,

"task": task.Name,

}).

Warning("killing leftover process")

err = taskCmd.Process.Kill()

if err != nil {

log.WithFields(logrus.Fields{

"process": taskCmd.Process.Pid,

"id": task.TaskID.Value,

"task": task.Name,

}).

Error("cannot kill process")

}

}

state.mu.Unlock()

return

}

state.mu.Unlock()

// Process events from task

go func() {

deo := event.DeviceEventOrigin{

AgentId: task.AgentID,

ExecutorId: task.GetExecutor().ExecutorID,

TaskId: task.TaskID,

}

for {

esr, err := esc.Recv()

if err == io.EOF {

log.WithError(err).Warning("event stream EOF")

break

}

if err != nil {

log.WithError(err).Warning("error receiving event from task")

continue

}

ev := esr.GetEvent()

deviceEvent := event.NewDeviceEvent(deo, ev.GetType())

jsonEvent, err := json.Marshal(deviceEvent)

if err != nil {

log.WithError(err).Warning("error marshaling event from task")

continue

}

state.mu.RLock()

state.cli.Send(context.TODO(), calls.NonStreaming(calls.Message(jsonEvent)))

log.WithFields(logrus.Fields{

"task": task.TaskID.Value,

"event": string(jsonEvent),

}).

Debug("event sent")

state.mu.RUnlock()

}

}()

err = taskCmd.Wait()

state.mu.Lock()

defer state.mu.Unlock()

if _, ok := state.rpcClients[task.TaskID]; ok {

state.rpcClients[task.TaskID].Close() // NOTE: might return non-nil error, but we don't care much

delete(state.rpcClients, task.TaskID)

}

if err != nil {

log.WithFields(logrus.Fields{

"id": task.TaskID.Value,

"task": task.Name,

"error": err.Error(),

}).

Error("process terminated with error")

status.State = mesos.TASK_FAILED.Enum()

status.Message = protoString(err.Error())

state.failedTasks[task.TaskID] = status

return

}

if errStdout != nil || errStderr != nil {

log.WithFields(logrus.Fields{

"errStderr": errStderr,

"errStdout": errStdout,

"id": task.TaskID.Value,

"task": task.Name,

}).

Warning("failed to capture stdout or stderr of task")

}

log.WithField("task", task.Name).Debug("sending TASK_FINISHED")

// send FINISHED

status = newStatus(state, task.TaskID)

status.State = mesos.TASK_FINISHED.Enum()

err = update(state, status)

if err != nil {

log.WithFields(logrus.Fields{

"id": task.TaskID.Value,

"name": task.Name,

"error": err.Error(),

}).

Error("failed to send TASK_FINISHED")

status.State = mesos.TASK_FAILED.Enum()

status.Message = protoString(err.Error())

state.failedTasks[task.TaskID] = status

}

}()

log.WithField("task", task.Name).Debug("gRPC client running, handler forked")

}

// helper func to package strings up nicely for protobuf

func protoString(s string) *string { return &s }

// update sends UPDATE to agent.

func update(state *internalState, status mesos.TaskStatus) error {

status.Timestamp = proto.Float64(float64(time.Now().Unix()))

log.WithFields(logrus.Fields{

"status": status.State.String(),

"id": status.TaskID.Value,

}).

Debug("sending UPDATE on task status")

upd := calls.Update(status)

resp, err := state.cli.Send(context.TODO(), calls.NonStreaming(upd))

if resp != nil {

resp.Close()

}

if err != nil {

log.WithField("error", err).Error("failed to send update")

debugJSON(upd)

} else {

state.unackedUpdates[string(status.UUID)] = *upd.Update

}

return err

}

// newStatus constructs a new mesos.TaskStatus to describe a task.

func newStatus(state *internalState, id mesos.TaskID) mesos.TaskStatus {

return mesos.TaskStatus{

TaskID: id,

Source: mesos.SOURCE_EXECUTOR.Enum(),

ExecutorID: &state.executor.ExecutorID,

UUID: []byte(uuid.NewRandom()),

}

}

// internalState of the executor.

type internalState struct {

mu sync.RWMutex

cli calls.Sender

cfg config.Config

framework mesos.FrameworkInfo

executor mesos.ExecutorInfo

agent mesos.AgentInfo

unackedTasks map[mesos.TaskID]mesos.TaskInfo

unackedUpdates map[string]executor.Call_Update

failedTasks map[mesos.TaskID]mesos.TaskStatus // send updates for these as we can

rpcClients map[mesos.TaskID]*executorcmd.RpcClient

shouldQuit bool

}