executor

README

Mesos Executor

Customizable Apache Mesos task executor. It allows controlled graceful task shutdown and performing various additional actions during the task lifecycle, by providing hook mechanisms (see hook package).

How task execution works?

Executor uses Mesos HTTP API to communicate with agent. When executor is started, before doing anything else, it tries to subscribe to Mesos agent. After successful subscription it waits for events from agent to handle. During the whole process executor keeps connection with a Mesos agent. When connection is lost it tries to reconnect for a configured time and when it fails to do so it tries to finish the started task and stops the whole executor process.

Task is started when Event_LAUNCH is received with required TaskInfo. Before starting the received command executor fires BeforeTaskStartEvent event hook, and if any of registered hooks fail to do their jobs, it stops the execution process and fails. This hook can be used to modify the environment of the task by returning formatted variable strings ("VAR=value"). Right after starting the command, executor fires AfterTaskStartEvent event hook - and again if any of the hooks fail, executor fails also. It is worth noting that executor may exit without even starting a task.

Executor may exit in the following cases:

• started tasks fail to start or run - executor quits with TASK_FAILED sent to Mesos agent
• started tasks exit with 0 return code - executor quits with TASK_FINISHED sent to Mesos agent
• executor receives Event_SHUTDOWN or Event_KILL - executor quits with TASK_KILLED sent to Mesos agent

Executor always fires BeforeTerminateEvent event hook when exiting - regardless of whether it started a task or not.

Graceful Shutdown

Graceful Shutdown is a feature to minimize task killing impact on other systems. It is performed in the following steps:

1. Call all hooks with BeforeTerminateEvent.
2. Sent SIGTERM to process tree.
3. Wait KillPolicyGracePeriod (can be overridden with Task Kill Policy Grace Period).
4. Sent SIGKILL to process tree.

Executor can be configured to exclude certain processes from SIGTERM signal. Provide process names to exclude in ALLEGRO_EXECUTOR_SIGTERM_EXCLUDE_PROCESSES environment variable as a comma-separated string. This feature requires pgrep -g to be available on the machine.

Log scraping

By default executor forwards service stdout/stderr to its own standard streams. It can however redirect them to data processing pipeline - Logstash. This requires you to set up the connection to the Logstash service in the executor's environmental variables:

ALLEGRO_EXECUTOR_SERVICELOG_LOGSTASH_PROTOCOL="tcp" # tcp or udp
ALLEGRO_EXECUTOR_SERVICELOG_LOGSTASH_ADDRESS="localhost:1234" # host and port

Currently, the executor is able to parse and send only logs in the logfmt format. To enable log scraping you need to set log-scraping label in Mesos TaskInfo to logfmt. For more information see documentation of servicelog package.

Hooks

Executor supports integration with external system via hooks. The hook is an interface with functions that will be called when specific actions occur. To use hooks just implement hook.Hook and plug it into hook.Manager. Hooks calls are blocking.

Consul integration

Integration with Consul is based on a hook. It mimics the behavior of allegro/marathon-consul. Task is registered in Consul once it becomes healthy and deregistered before kill. Required task metadata such as name, labels and ports are obtained from task definition. Service name is taken from consul label. Labels are transformed to Consul tags only when value is equal tag. Client does not use any ACL Token by default, this can be changed by setting CONSUL_TOKEN environment variable.

VaaS integration

VaaS integration is based on a hook. Task is registered once it becomes healthy and deregistered before kill. Task’s first port will be registered under director provided in a label named director. If task has defined weight in a label it will be used. Weight could be overridden with VAAS_INITIAL_WEIGHT environment variable. If task is a canary instance (has non empty canary label) backend is marked as a canary.

Requirements

To run executor tests locally you need following tools installed:

• Go 1.14+
• Make
• Linux or Docker

If you want to test executor locally, you will need additionally:

• Vagrant
• Ansible 2.2+

Debug mode

Executor offers a debug mode that provide extended logging and capabilities during runtime. Enabling this can significantly increase the amount of resources the executor needs to operate, so do not turn this on, when it is not needed. To enable debug mode add -debug flag to executor command or set ALLEGRO_EXECUTOR_DEBUG environment variable to true.

Development

Using Vagrant environment

To create your Vagrant environment execute following command in project root folder:

$ vagrant up

It will create a virtual machine with Apache Mesos and Marathon installed and running on it. Mesos UI will be available on http://localhost:5050 and Marathon UI on http://localhost:8080.

If you want to test executor on Vagrant Mesos you will have to create release build of it. To do this, execute the following command:

$ make release

Binary will be immediately available on virtual machine on following address:

http://localhost/executor

You can use above address to configure your Marathon application to be executed by your freshly build executor. An example application is available in marathon-test-app.json

Executor configuration can be altered via environment variables in the following way:

ALLEGRO_EXECUTOR_STATE_UPDATE_BUFFER_SIZE="2048"
ALLEGRO_EXECUTOR_STATE_UPDATE_WAIT_TIMEOUT="3s"

sets the StateUpdateBufferSize Config property to 2048, and StateUpdateWaitTimeout to 3 seconds. For all available settings and their defaults see executor.go.

Additionally, a Consul instance is available for testing purposes, its logs can be viewed by running:

$ vagrant ssh
vagrant@localhost:~$ sudo supervisorctl tail -f consul

Known Issues

1. Executor may not send a SIGKILL to process tree after grace period, so service process may be still running when executor finishes. To clean up executor and launched tasks properly use pid isolator.

Contributing

See CONTRIBUTING for more details and code of conduct.

License

Mesos Executor is distributed under the Apache 2.0 License.

Documentation

Index

• Constants
• func DoHealthChecks(check mesos.HealthCheck, healthStates chan<- Event)
• func ForwardCmdOutput() func(*exec.Cmd) error
• func GetCertFromEnvVariables(env []string) (*x509.Certificate, error)
• func HealthCheckAddress(port uint32) string
• func ScrapCmdOutput(s scraper.Scraper, a appender.Appender, extenders ...servicelog.Extender) func(*exec.Cmd) error
• func StartExecutor(conf Config, hooks []hook.Hook) error
• type Command
  • func NewCommand(commandInfo mesos.CommandInfo, env []string, options ...func(*exec.Cmd) error) (Command, error)
• type Config
• type Event
• type EventType
• type Executor
  • func NewExecutor(cfg Config, hooks ...hook.Hook) *Executor
  • func (e *Executor) Start() error
• type TaskExitCode
• type TaskExitState

Constants

const EnvironmentPrefix = "allegro_executor"

EnvironmentPrefix is a prefix for environmental configuration

Functions

func DoHealthChecks

func DoHealthChecks(check mesos.HealthCheck, healthStates chan<- Event)

DoHealthChecks schedules health check defined in check. HealthState updates are delivered on provided healthStates channel.

func ForwardCmdOutput

func ForwardCmdOutput() func(*exec.Cmd) error

ForwardCmdOutput configures command to forward its output to the system stderr and stdout.

func GetCertFromEnvVariables

func GetCertFromEnvVariables(env []string) (*x509.Certificate, error)

GetCertFromEnvVariables returns certificate stored in environment variables. If no certificate is found then empty string is returned

func HealthCheckAddress

func HealthCheckAddress(port uint32) string

HealthCheckAddress returns host and port that should be used for health checking service.

func ScrapCmdOutput

func ScrapCmdOutput(s scraper.Scraper, a appender.Appender, extenders ...servicelog.Extender) func(*exec.Cmd) error

ScrapCmdOutput configures command so itd output will be scraped and forwarded by provided log appender.

func StartExecutor

func StartExecutor(conf Config, hooks []hook.Hook) error

StartExecutor creates a new executor instance nad starts it

Types

type Command

type Command interface {
	Start() error
	Wait() <-chan TaskExitState
	Stop(gracePeriod time.Duration, sigtermExcludeProcesses []string)
}

Command is an interface to abstract command running on a system.

func NewCommand

func NewCommand(commandInfo mesos.CommandInfo, env []string, options ...func(*exec.Cmd) error) (Command, error)

NewCommand returns a new command based on passed CommandInfo.

type Config

type Config struct {
	// Sets logging level to `debug` when true, `info` otherwise
	Debug bool `default:"false" split_words:"true"`
	// Mesos API path
	APIPath string `default:"/api/v1/executor" split_words:"true"`
	// Delay between sending TERM and KILL signals
	KillPolicyGracePeriod time.Duration `default:"5s" split_words:"true"`
	// Timeout for communication with Mesos
	HTTPTimeout time.Duration `default:"10s" split_words:"true"`
	// Number of state messages to keep in buffer
	StateUpdateBufferSize int `default:"1024" split_words:"true"`
	// Timeout for attempts to send messages in buffer
	StateUpdateWaitTimeout time.Duration `default:"5s" split_words:"true"`

	// Mesos framework configuration
	MesosConfig config.Config `ignore:"true"`

	// SentryDSN is an address used for sending logs to Sentry
	SentryDSN string `split_words:"true"`

	// ServicelogBufferSize sets a line buffer size used by log scraping module
	ServicelogBufferSize uint `default:"2000" split_words:"true"`

	// ServicelogIgnoreKeys is a list of ignored keys for log scraping module
	ServicelogIgnoreKeys []string `split_words:"true"`

	// Range in which certificate will be considered as expired. Used to
	// prevent shutdown of all tasks at once.
	RandomExpirationRange time.Duration `default:"3h" split_words:"true"`

	// SigtermExcludeProcesses specifies process names to omit when sending SIGTERM to process tree during shutdown
	SigtermExcludeProcesses []string `split_words:"true"`
}

Config settable from the environment

type Event

type Event struct {
	Type EventType
	// Message store the human readable information about
	// current event. For example reason of the event or
	// additional debug message.
	Message string
	// contains filtered or unexported fields
}

Event is an internal executor event that triggers specific actions driven by current state and Type.

type EventType

type EventType int

EventType defines type of the Event.

const (
	// Healthy means task health check passed and task healthy.
	Healthy EventType = iota
	// Unhealthy means task health check failed. Fail reason should be
	// passed in Event Message field.
	Unhealthy
	// FailedDueToUnhealthy means task health check failed and task should be killed
	// because it's unhealthy for longer period of time. Fail reason should be
	// passed in Event Message field.
	FailedDueToUnhealthy
	// FailedDueToExpiredCertificate means task certificate expired (or will expire soon)
	// and task should be killed because it can't work with invalid certificate.
	FailedDueToExpiredCertificate

	// CommandExited means command has exited. Message should contains information
	// about exit code.
	CommandExited

	// Kill means command should be killed and executor exit.
	Kill

	// Shutdown means executor should kill all tasks and exit.
	Shutdown

	// Subscribed means executor attach to mesos Agent.
	Subscribed
	// Launch means executor should start a task.
	Launch
)

type Executor

type Executor struct {
	// contains filtered or unexported fields
}

Executor is responsible for launching and monitoring single Mesos task.

func NewExecutor

func NewExecutor(cfg Config, hooks ...hook.Hook) *Executor

NewExecutor creates new instance of executor configured with by `cfg` with hooks

func (*Executor) Start

func (e *Executor) Start() error

Start registers executor in Mesos agent and waits for events from it.

type TaskExitCode

type TaskExitCode int8

TaskExitCode is an enum.

const (
	// SuccessCode means task exited successfully.
	SuccessCode TaskExitCode = iota
	// FailedCode means task exited with error.
	FailedCode
	// KilledCode means task was killed and it's code was ignored.
	KilledCode
)

type TaskExitState

type TaskExitState struct {
	Code TaskExitCode
	Err  error
}

TaskExitState is a type describing reason of program execution interuption.