include/mesos/executor.hpp - mesos - Git at Google

 // Licensed to the Apache Software Foundation (ASF) under one
 // or more contributor license agreements.  See the NOTICE file
 // distributed with this work for additional information
 // regarding copyright ownership.  The ASF licenses this file
 // to you under the Apache License, Version 2.0 (the
 // "License"); you may not use this file except in compliance
 // with the License.  You may obtain a copy of the License at
 //
 //     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 // See the License for the specific language governing permissions and
 // limitations under the License.

 #ifndef __MESOS_EXECUTOR_HPP__
 #define __MESOS_EXECUTOR_HPP__

 #include <map>
 #include <mutex>
 #include <string>

 #include <mesos/mesos.hpp>

 // Mesos executor interface and executor driver. An executor is
 // responsible for launching tasks in a framework specific way (i.e.,
 // creating new threads, new processes, etc). One or more executors
 // from the same framework may run concurrently on the same machine.
 // Note that we use the term "executor" fairly loosely to refer to the
 // code that implements the Executor interface (see below) as well as
 // the program that is responsible for instantiating a new
 // MesosExecutorDriver (also below). In fact, while a Mesos slave is
 // responsible for (forking and) executing the "executor", there is no
 // reason why whatever the slave executed might itself actually
 // execute another program which actually instantiates and runs the
 // MesosSchedulerDriver. The only contract with the slave is that the
 // program that it invokes does not exit until the "executor" has
 // completed. Thus, what the slave executes may be nothing more than a
 // script which actually executes (or forks and waits) the "real"
 // executor.
 //
 // IF YOU FIND YOURSELF MODIFYING COMMENTS HERE PLEASE CONSIDER MAKING
 // THE SAME MODIFICATIONS FOR OTHER LANGUAGE BINDINGS (e.g., Java:
 // src/java/src/org/apache/mesos, Python: src/python/src, etc.).

 // Forward declaration.
 namespace process {
 class Latch;
 } // namespace process {

 namespace mesos {

 // A few forward declarations.
 class ExecutorDriver;

 namespace internal {
 class ExecutorProcess;
 }

 // Callback interface to be implemented by frameworks' executors. Note
 // that only one callback will be invoked at a time, so it is not
 // recommended that you block within a callback because it may cause a
 // deadlock.
 //
 // Each callback includes a pointer to the executor driver that was
 // used to run this executor. The pointer will not change for the
 // duration of an executor (i.e., from the point you do
 // ExecutorDriver::start() to the point that ExecutorDriver::join()
 // returns). This is intended for convenience so that an executor
 // doesn't need to store a pointer to the driver itself.
 //
 // TODO(bmahler): Consider adding a usage() callback here, that
 // provides information to the executor about its ResourceUsage.
 class Executor
 {
 public:
   // Empty virtual destructor (necessary to instantiate subclasses).
   virtual ~Executor() {}

   // Invoked once the executor driver has been able to successfully
   // connect with Mesos. In particular, a scheduler can pass some
   // data to its executors through the FrameworkInfo.ExecutorInfo's
   // data field.
   virtual void registered(
       ExecutorDriver* driver,
       const ExecutorInfo& executorInfo,
       const FrameworkInfo& frameworkInfo,
       const SlaveInfo& slaveInfo) = 0;

   // Invoked when the executor reregisters with a restarted slave.
   virtual void reregistered(
       ExecutorDriver* driver,
       const SlaveInfo& slaveInfo) = 0;

   // Invoked when the executor becomes "disconnected" from the slave
   // (e.g., the slave is being restarted due to an upgrade).
   virtual void disconnected(ExecutorDriver* driver) = 0;

   // Invoked when a task has been launched on this executor (initiated
   // via Scheduler::launchTasks). Note that this task can be realized
   // with a thread, a process, or some simple computation, however, no
   // other callbacks will be invoked on this executor until this
   // callback has returned.
   virtual void launchTask(
       ExecutorDriver* driver,
       const TaskInfo& task) = 0;

   // Invoked when a task running within this executor has been killed
   // (via SchedulerDriver::killTask). Note that no status update will
   // be sent on behalf of the executor, the executor is responsible
   // for creating a new TaskStatus (i.e., with TASK_KILLED) and
   // invoking ExecutorDriver::sendStatusUpdate.
   virtual void killTask(
       ExecutorDriver* driver,
       const TaskID& taskId) = 0;

   // Invoked when a framework message has arrived for this executor.
   // These messages are best effort; do not expect a framework message
   // to be retransmitted in any reliable fashion.
   virtual void frameworkMessage(
       ExecutorDriver* driver,
       const std::string& data) = 0;

   // Invoked when the executor should terminate all of its currently
   // running tasks. Note that after a Mesos has determined that an
   // executor has terminated any tasks that the executor did not send
   // terminal status updates for (e.g., TASK_KILLED, TASK_FINISHED,
   // TASK_FAILED, etc) a TASK_LOST status update will be created.
   virtual void shutdown(ExecutorDriver* driver) = 0;

   // Invoked when a fatal error has occurred with the executor and/or
   // executor driver. The driver will be aborted BEFORE invoking this
   // callback.
   virtual void error(
       ExecutorDriver* driver,
       const std::string& message) = 0;
 };


 // Abstract interface for connecting an executor to Mesos. This
 // interface is used both to manage the executor's lifecycle (start
 // it, stop it, or wait for it to finish) and to interact with Mesos
 // (e.g., send status updates, send framework messages, etc.). See
 // MesosExecutorDriver below for a concrete example of an
 // ExecutorDriver.
 class ExecutorDriver
 {
 public:
   // Empty virtual destructor (necessary to instantiate subclasses).
   virtual ~ExecutorDriver() {}

   // Starts the executor driver. This needs to be called before any
   // other driver calls are made.
   virtual Status start() = 0;

   // Stops the executor driver.
   virtual Status stop() = 0;

   // Aborts the driver so that no more callbacks can be made to the
   // executor. The semantics of abort and stop have deliberately been
   // separated so that code can detect an aborted driver (i.e., via
   // the return status of ExecutorDriver::join, see below), and
   // instantiate and start another driver if desired (from within the
   // same process ... although this functionality is currently not
   // supported for executors).
   virtual Status abort() = 0;

   // Waits for the driver to be stopped or aborted, possibly
   // _blocking_ the current thread indefinitely. The return status of
   // this function can be used to determine if the driver was aborted
   // (see mesos.proto for a description of Status).
   virtual Status join() = 0;

   // Starts and immediately joins (i.e., blocks on) the driver.
   virtual Status run() = 0;

   // Sends a status update to the framework scheduler, retrying as
   // necessary until an acknowledgement has been received or the
   // executor is terminated (in which case, a TASK_LOST status update
   // will be sent). See Scheduler::statusUpdate for more information
   // about status update acknowledgements.
   virtual Status sendStatusUpdate(const TaskStatus& status) = 0;

   // Sends a message to the framework scheduler. These messages are
   // best effort; do not expect a framework message to be
   // retransmitted in any reliable fashion.
   virtual Status sendFrameworkMessage(const std::string& data) = 0;
 };


 // Concrete implementation of an ExecutorDriver that connects an
 // Executor with a Mesos slave. The MesosExecutorDriver is
 // thread-safe.
 //
 // The driver is responsible for invoking the Executor callbacks as it
 // communicates with the Mesos slave.
 //
 // Note that blocking on the MesosExecutorDriver (e.g., via
 // MesosExecutorDriver::join) doesn't affect the executor callbacks in
 // anyway because they are handled by a different thread.
 //
 // Note that the driver uses GLOG to do its own logging. GLOG flags
 // can be set via environment variables, prefixing the flag name with
 // "GLOG_", e.g., "GLOG_v=1". For Mesos specific logging flags see
 // src/logging/flags.hpp. Mesos flags can also be set via environment
 // variables, prefixing the flag name with "MESOS_", e.g.,
 // "MESOS_QUIET=1".
 //
 // See src/examples/test_executor.cpp for an example of using the
 // MesosExecutorDriver.
 class MesosExecutorDriver : public ExecutorDriver
 {
 public:
   // Creates a new driver that uses the specified Executor. Note, the
   // executor pointer must outlive the driver.
   //
   // Note that the other constructor overload that accepts `environment`
   // argument is preferable to this one in a multithreaded environment,
   // because the implementation of this one accesses global environment
   // which is unsafe due to a potential concurrent modification of the
   // environment by another thread.
   explicit MesosExecutorDriver(Executor* executor);

   // Creates a new driver that uses the specified `Executor` and environment
   // variables. Note, the executor pointer must outlive the driver.
   explicit MesosExecutorDriver(
       Executor* executor,
       const std::map<std::string, std::string>& environment);

   // This destructor will block indefinitely if
   // MesosExecutorDriver::start was invoked successfully (possibly via
   // MesosExecutorDriver::run) and MesosExecutorDriver::stop has not
   // been invoked.
   ~MesosExecutorDriver() override;

   // See ExecutorDriver for descriptions of these.
   Status start() override;
   Status stop() override;
   Status abort() override;
   Status join() override;
   Status run() override;
   Status sendStatusUpdate(const TaskStatus& status) override;
   Status sendFrameworkMessage(const std::string& data) override;

 private:
   friend class internal::ExecutorProcess;

   Executor* executor;

   // Libprocess process for communicating with slave.
   internal::ExecutorProcess* process;

   // Mutex for enforcing serial execution of all non-callbacks.
   std::recursive_mutex mutex;

   // Latch for waiting until driver terminates.
   process::Latch* latch;

   // Current status of the driver.
   Status status;

   std::map<std::string, std::string> environment;
 };

 } // namespace mesos {

 #endif // __MESOS_EXECUTOR_HPP__
	// Licensed to the Apache Software Foundation (ASF) under one
	// or more contributor license agreements. See the NOTICE file
	// distributed with this work for additional information
	// regarding copyright ownership. The ASF licenses this file
	// to you under the Apache License, Version 2.0 (the
	// "License"); you may not use this file except in compliance
	// with the License. You may obtain a copy of the License at
	//
	// http://www.apache.org/licenses/LICENSE-2.0
	//
	// Unless required by applicable law or agreed to in writing, software
	// distributed under the License is distributed on an "AS IS" BASIS,
	// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	// See the License for the specific language governing permissions and
	// limitations under the License.

	#ifndef __MESOS_EXECUTOR_HPP__
	#define __MESOS_EXECUTOR_HPP__

	#include <map>
	#include <mutex>
	#include <string>

	#include <mesos/mesos.hpp>

	// Mesos executor interface and executor driver. An executor is
	// responsible for launching tasks in a framework specific way (i.e.,
	// creating new threads, new processes, etc). One or more executors
	// from the same framework may run concurrently on the same machine.
	// Note that we use the term "executor" fairly loosely to refer to the
	// code that implements the Executor interface (see below) as well as
	// the program that is responsible for instantiating a new
	// MesosExecutorDriver (also below). In fact, while a Mesos slave is
	// responsible for (forking and) executing the "executor", there is no
	// reason why whatever the slave executed might itself actually
	// execute another program which actually instantiates and runs the
	// MesosSchedulerDriver. The only contract with the slave is that the
	// program that it invokes does not exit until the "executor" has
	// completed. Thus, what the slave executes may be nothing more than a
	// script which actually executes (or forks and waits) the "real"
	// executor.
	//
	// IF YOU FIND YOURSELF MODIFYING COMMENTS HERE PLEASE CONSIDER MAKING
	// THE SAME MODIFICATIONS FOR OTHER LANGUAGE BINDINGS (e.g., Java:
	// src/java/src/org/apache/mesos, Python: src/python/src, etc.).

	// Forward declaration.
	namespace process {
	class Latch;
	} // namespace process {

	namespace mesos {

	// A few forward declarations.
	class ExecutorDriver;

	namespace internal {
	class ExecutorProcess;
	}

	// Callback interface to be implemented by frameworks' executors. Note
	// that only one callback will be invoked at a time, so it is not
	// recommended that you block within a callback because it may cause a
	// deadlock.
	//
	// Each callback includes a pointer to the executor driver that was
	// used to run this executor. The pointer will not change for the
	// duration of an executor (i.e., from the point you do
	// ExecutorDriver::start() to the point that ExecutorDriver::join()
	// returns). This is intended for convenience so that an executor
	// doesn't need to store a pointer to the driver itself.
	//
	// TODO(bmahler): Consider adding a usage() callback here, that
	// provides information to the executor about its ResourceUsage.
	class Executor
	{
	public:
	// Empty virtual destructor (necessary to instantiate subclasses).
	virtual ~Executor() {}

	// Invoked once the executor driver has been able to successfully
	// connect with Mesos. In particular, a scheduler can pass some
	// data to its executors through the FrameworkInfo.ExecutorInfo's
	// data field.
	virtual void registered(
	ExecutorDriver* driver,
	const ExecutorInfo& executorInfo,
	const FrameworkInfo& frameworkInfo,
	const SlaveInfo& slaveInfo) = 0;

	// Invoked when the executor reregisters with a restarted slave.
	virtual void reregistered(
	ExecutorDriver* driver,
	const SlaveInfo& slaveInfo) = 0;

	// Invoked when the executor becomes "disconnected" from the slave
	// (e.g., the slave is being restarted due to an upgrade).
	virtual void disconnected(ExecutorDriver* driver) = 0;

	// Invoked when a task has been launched on this executor (initiated
	// via Scheduler::launchTasks). Note that this task can be realized
	// with a thread, a process, or some simple computation, however, no
	// other callbacks will be invoked on this executor until this
	// callback has returned.
	virtual void launchTask(
	ExecutorDriver* driver,
	const TaskInfo& task) = 0;

	// Invoked when a task running within this executor has been killed
	// (via SchedulerDriver::killTask). Note that no status update will
	// be sent on behalf of the executor, the executor is responsible
	// for creating a new TaskStatus (i.e., with TASK_KILLED) and
	// invoking ExecutorDriver::sendStatusUpdate.
	virtual void killTask(
	ExecutorDriver* driver,
	const TaskID& taskId) = 0;

	// Invoked when a framework message has arrived for this executor.
	// These messages are best effort; do not expect a framework message
	// to be retransmitted in any reliable fashion.
	virtual void frameworkMessage(
	ExecutorDriver* driver,
	const std::string& data) = 0;

	// Invoked when the executor should terminate all of its currently
	// running tasks. Note that after a Mesos has determined that an
	// executor has terminated any tasks that the executor did not send
	// terminal status updates for (e.g., TASK_KILLED, TASK_FINISHED,
	// TASK_FAILED, etc) a TASK_LOST status update will be created.
	virtual void shutdown(ExecutorDriver* driver) = 0;

	// Invoked when a fatal error has occurred with the executor and/or
	// executor driver. The driver will be aborted BEFORE invoking this
	// callback.
	virtual void error(
	ExecutorDriver* driver,
	const std::string& message) = 0;
	};


	// Abstract interface for connecting an executor to Mesos. This
	// interface is used both to manage the executor's lifecycle (start
	// it, stop it, or wait for it to finish) and to interact with Mesos
	// (e.g., send status updates, send framework messages, etc.). See
	// MesosExecutorDriver below for a concrete example of an
	// ExecutorDriver.
	class ExecutorDriver
	{
	public:
	// Empty virtual destructor (necessary to instantiate subclasses).
	virtual ~ExecutorDriver() {}

	// Starts the executor driver. This needs to be called before any
	// other driver calls are made.
	virtual Status start() = 0;

	// Stops the executor driver.
	virtual Status stop() = 0;

	// Aborts the driver so that no more callbacks can be made to the
	// executor. The semantics of abort and stop have deliberately been
	// separated so that code can detect an aborted driver (i.e., via
	// the return status of ExecutorDriver::join, see below), and
	// instantiate and start another driver if desired (from within the
	// same process ... although this functionality is currently not
	// supported for executors).
	virtual Status abort() = 0;

	// Waits for the driver to be stopped or aborted, possibly
	// _blocking_ the current thread indefinitely. The return status of
	// this function can be used to determine if the driver was aborted
	// (see mesos.proto for a description of Status).
	virtual Status join() = 0;

	// Starts and immediately joins (i.e., blocks on) the driver.
	virtual Status run() = 0;

	// Sends a status update to the framework scheduler, retrying as
	// necessary until an acknowledgement has been received or the
	// executor is terminated (in which case, a TASK_LOST status update
	// will be sent). See Scheduler::statusUpdate for more information
	// about status update acknowledgements.
	virtual Status sendStatusUpdate(const TaskStatus& status) = 0;

	// Sends a message to the framework scheduler. These messages are
	// best effort; do not expect a framework message to be
	// retransmitted in any reliable fashion.
	virtual Status sendFrameworkMessage(const std::string& data) = 0;
	};


	// Concrete implementation of an ExecutorDriver that connects an
	// Executor with a Mesos slave. The MesosExecutorDriver is
	// thread-safe.
	//
	// The driver is responsible for invoking the Executor callbacks as it
	// communicates with the Mesos slave.
	//
	// Note that blocking on the MesosExecutorDriver (e.g., via
	// MesosExecutorDriver::join) doesn't affect the executor callbacks in
	// anyway because they are handled by a different thread.
	//
	// Note that the driver uses GLOG to do its own logging. GLOG flags
	// can be set via environment variables, prefixing the flag name with
	// "GLOG_", e.g., "GLOG_v=1". For Mesos specific logging flags see
	// src/logging/flags.hpp. Mesos flags can also be set via environment
	// variables, prefixing the flag name with "MESOS_", e.g.,
	// "MESOS_QUIET=1".
	//
	// See src/examples/test_executor.cpp for an example of using the
	// MesosExecutorDriver.
	class MesosExecutorDriver : public ExecutorDriver
	{
	public:
	// Creates a new driver that uses the specified Executor. Note, the
	// executor pointer must outlive the driver.
	//
	// Note that the other constructor overload that accepts `environment`
	// argument is preferable to this one in a multithreaded environment,
	// because the implementation of this one accesses global environment
	// which is unsafe due to a potential concurrent modification of the
	// environment by another thread.
	explicit MesosExecutorDriver(Executor* executor);

	// Creates a new driver that uses the specified `Executor` and environment
	// variables. Note, the executor pointer must outlive the driver.
	explicit MesosExecutorDriver(
	Executor* executor,
	const std::map<std::string, std::string>& environment);

	// This destructor will block indefinitely if
	// MesosExecutorDriver::start was invoked successfully (possibly via
	// MesosExecutorDriver::run) and MesosExecutorDriver::stop has not
	// been invoked.
	~MesosExecutorDriver() override;

	// See ExecutorDriver for descriptions of these.
	Status start() override;
	Status stop() override;
	Status abort() override;
	Status join() override;
	Status run() override;
	Status sendStatusUpdate(const TaskStatus& status) override;
	Status sendFrameworkMessage(const std::string& data) override;

	private:
	friend class internal::ExecutorProcess;

	Executor* executor;

	// Libprocess process for communicating with slave.
	internal::ExecutorProcess* process;

	// Mutex for enforcing serial execution of all non-callbacks.
	std::recursive_mutex mutex;

	// Latch for waiting until driver terminates.
	process::Latch* latch;

	// Current status of the driver.
	Status status;

	std::map<std::string, std::string> environment;
	};

	} // namespace mesos {

	#endif // __MESOS_EXECUTOR_HPP__