worker

README

Travis Worker

Installing Travis Worker

from binary

1. download the 64-bit linux binary
2. put it somewhere in $PATH, e.g. /usr/local/bin/travis-worker

from package

Use the ./bin/travis-worker-installer script, or take a look at the packagecloud instructions.

from source

1. clone this down
2. install Go and gvt.
3. make

Configuring Travis Worker

Travis Worker is configured with environment variables or command line flags via the codegangsta/cli library. A list of the non-dynamic flags and environment variables may be found by invoking the built-in help system:

travis-worker --help

Configuring the requested provider

Each provider requires its own configuration, which must be provided via environment variables namespaced by TRAVIS_WORKER_{PROVIDER}_, e.g. for the docker provider:

export TRAVIS_WORKER_DOCKER_ENDPOINT="tcp://localhost:4243"
export TRAVIS_WORKER_DOCKER_PRIVILEGED="false"
export TRAVIS_WORKER_DOCKER_CERT_PATH="/etc/secret-docker-cert-stuff"

Verifying and exporting configuration

To inspect the parsed configuration in a format that can be used as a base environment variable configuration, use the --echo-config flag, which will exit immediately after writing to stdout:

travis-worker --echo-config

Running Travis Worker

1. make
2. ./bin/travis-worker

C-c will stop the worker. Note that any VMs for builds that were still running will have to be cleaned up manually.

Stopping Travis Worker

Travis Worker has two shutdown modes: Graceful and immediate. The graceful shutdown will tell the worker to not start any additional jobs, but finish the jobs it is currently running before it shuts down. The immediate shutdown will make the worker stop the jobs it's working on and requeue them, and clean up any open resources (shut down VMs, cleanly close connections, etc.)

To start a graceful shutdown, send an INT signal to the worker (for example using kill -INT). To start an immediate shutdown, send a TERM signal to the worker (for example using kill -TERM).

Go dependency management

Travis Worker is built via the standard go commands (with GO15VENDOREXPERIMENT=1), and dependencies managed by gvt.

To work with the dependencies you need to do the following first

• Have this repository checked out
• Install gvt with github.com/FiloSottile/gvt

Updating existing vendored dependencies

To update and existing vendored dependency, do the following:

1. cd ~/code/travis-worker
2. gvt fetch name/of/dependency e.g. gvt fetch github.com/pkg/sftp

Adding a new dependency

To add a new dependency, do the following:

1. cd ~/code/travis-worker
2. gvt fetch name/of/package e.g. gvt fetch github.com/pkg/sftp

License and Copyright Information

See LICENSE file.

© 2014-2015 Travis CI GmbH

Documentation

Index

• Variables
• type AMQPCanceller
  • func NewAMQPCanceller(ctx gocontext.Context, conn *amqp.Connection) *AMQPCanceller
  • func (d *AMQPCanceller) Run()
  • func (d *AMQPCanceller) Subscribe(id uint64, ch chan<- struct{}) error
  • func (d *AMQPCanceller) Unsubscribe(id uint64)
• type AMQPJobQueue
  • func NewAMQPJobQueue(conn *amqp.Connection, queue string) (*AMQPJobQueue, error)
  • func (q *AMQPJobQueue) Cleanup() error
  • func (q *AMQPJobQueue) Jobs(ctx gocontext.Context) (outChan <-chan Job, err error)
• type BuildPayload
• type BuildScriptGenerator
  • func NewBuildScriptGenerator(cfg *config.Config) BuildScriptGenerator
• type BuildScriptGeneratorError
• type CLI
  • func NewCLI(c *cli.Context) *CLI
  • func (i *CLI) Run()
  • func (i *CLI) Setup() (bool, error)
• type Canceller
• type FileCanceller
  • func NewFileCanceller(ctx gocontext.Context, baseDir string) *FileCanceller
  • func (c *FileCanceller) Run()
  • func (c *FileCanceller) Subscribe(id uint64, ch chan<- struct{}) error
  • func (c *FileCanceller) Unsubscribe(id uint64)
• type FileJobQueue
  • func NewFileJobQueue(baseDir, queue string, pollingInterval time.Duration) (*FileJobQueue, error)
  • func (f *FileJobQueue) Cleanup() error
  • func (f *FileJobQueue) Jobs(ctx gocontext.Context) (<-chan Job, error)
• type FinishState
• type Job
• type JobJobPayload
• type JobPayload
• type JobQueue
• type LogWriter
• type Processor
  • func NewProcessor(ctx gocontext.Context, hostname string, buildJobsChan <-chan Job, ...) (*Processor, error)
  • func (p *Processor) GracefulShutdown()
  • func (p *Processor) Run()
  • func (p *Processor) Terminate()
• type ProcessorPool
  • func NewProcessorPool(hostname string, ctx gocontext.Context, hardTimeout time.Duration, ...) *ProcessorPool
  • func (p *ProcessorPool) Decr()
  • func (p *ProcessorPool) Each(f func(int, *Processor))
  • func (p *ProcessorPool) GracefulShutdown()
  • func (p *ProcessorPool) Incr()
  • func (p *ProcessorPool) Run(poolSize int, queue JobQueue) error
  • func (p *ProcessorPool) Size() int
• type RepositoryPayload
• type SentryHook
  • func NewSentryHook(DSN string, levels []logrus.Level) (*SentryHook, error)
  • func (hook *SentryHook) Fire(entry *logrus.Entry) error
  • func (hook *SentryHook) Levels() []logrus.Level
• type TimeoutsPayload

Variables

var (
	// LogWriterTick is how often the buffer should be flushed out and sent to
	// travis-logs.
	LogWriterTick = 500 * time.Millisecond

	// LogChunkSize is a bit of a magic number, calculated like this: The
	// maximum Pusher payload is 10 kB (or 10 KiB, who knows, but let's go with
	// 10 kB since that is smaller). Looking at the travis-logs source, the
	// current message overhead (i.e. the part of the payload that isn't
	// the content of the log part) is 42 bytes + the length of the JSON-
	// encoded ID and the length of the JSON-encoded sequence number. A 64-
	// bit number is up to 20 digits long, so that means (assuming we don't
	// go over 64-bit numbers) the overhead is up to 82 bytes. That means
	// we can send up to 9918 bytes of content. However, the JSON-encoded
	// version of a string can be significantly longer than the raw bytes.
	// Worst case that I could find is "<", which with the Go JSON encoder
	// becomes "\u003c" (i.e. six bytes long). So, given a string of just
	// left angle brackets, the string would become six times as long,
	// meaning that the longest string we can take is 1653. We could still
	// get errors if we go over 64-bit numbers, but I find the likeliness
	// of that happening to both the sequence number, the ID, and us maxing
	// out the worst-case logs to be quite unlikely, so I'm willing to live
	// with that. --Henrik
	LogChunkSize = 1653

	// ErrWrotePastMaxLogLength is returned by LogWriter.Write if the write
	// caused the number of written bytes to go over the maximum log length.
	ErrWrotePastMaxLogLength = errors.New("wrote past max length")
)

var (
	// VersionString is the git describe version set at build time
	VersionString = "?"
	// RevisionString is the git revision set at build time
	RevisionString = "?"
	// GeneratedString is the build date set at build time
	GeneratedString = "?"
	// CopyrightString is the copyright set at build time
	CopyrightString = "?"
)

Types

type AMQPCanceller

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

AMQPCanceller is responsible for listening to a command queue on AMQP and dispatching the commands to the right place. Currently the only valid command is the 'cancel job' command.

func NewAMQPCanceller

func NewAMQPCanceller(ctx gocontext.Context, conn *amqp.Connection) *AMQPCanceller

NewAMQPCanceller creates a new AMQPCanceller. No network traffic occurs until you call Run()

func (*AMQPCanceller) Run

func (d *AMQPCanceller) Run()

Run will make the AMQPCanceller listen to the worker command queue and start dispatching any incoming commands.

func (*AMQPCanceller) Subscribe

func (d *AMQPCanceller) Subscribe(id uint64, ch chan<- struct{}) error

Subscribe is an implementation of Canceller.Subscribe.

func (*AMQPCanceller) Unsubscribe

func (d *AMQPCanceller) Unsubscribe(id uint64)

Unsubscribe is an implementation of Canceller.Unsubscribe.

type AMQPJobQueue

type AMQPJobQueue struct {
	DefaultLanguage, DefaultDist, DefaultGroup, DefaultOS string
	// contains filtered or unexported fields
}

AMQPJobQueue is a JobQueue that uses AMQP

func NewAMQPJobQueue

func NewAMQPJobQueue(conn *amqp.Connection, queue string) (*AMQPJobQueue, error)

NewAMQPJobQueue creates a AMQPJobQueue backed by the given AMQP connections and connects to the AMQP queue with the given name. The queue will be declared in AMQP when this function is called, so an error could be raised if the queue already exists, but with different attributes than we expect.

func (*AMQPJobQueue) Cleanup

func (q *AMQPJobQueue) Cleanup() error

Cleanup closes the underlying AMQP connection

func (*AMQPJobQueue) Jobs

func (q *AMQPJobQueue) Jobs(ctx gocontext.Context) (outChan <-chan Job, err error)

Jobs creates a new consumer on the queue, and returns three channels. The first channel gets sent every BuildJob that we receive from AMQP. The stopChan is a channel that can be closed in order to stop the consumer.

type BuildPayload

type BuildPayload struct {
	ID     uint64 `json:"id"`
	Number string `json:"number"`
}

BuildPayload contains information about the build.

type BuildScriptGenerator

type BuildScriptGenerator interface {
	Generate(gocontext.Context, *simplejson.Json) ([]byte, error)
}

A BuildScriptGenerator generates a build script for a given job payload.

func NewBuildScriptGenerator

func NewBuildScriptGenerator(cfg *config.Config) BuildScriptGenerator

NewBuildScriptGenerator creates a generator backed by an HTTP API.

type BuildScriptGeneratorError

type BuildScriptGeneratorError struct {

	// true when this error can be recovered by retrying later
	Recover bool
	// contains filtered or unexported fields
}

A BuildScriptGeneratorError is sometimes used by the Generate method on a BuildScriptGenerator to return more metadata about an error.

type CLI

type CLI struct {
	Config               *config.Config
	BuildScriptGenerator BuildScriptGenerator
	BackendProvider      backend.Provider
	ProcessorPool        *ProcessorPool
	Canceller            Canceller
	JobQueue             JobQueue
	// contains filtered or unexported fields
}

CLI is the top level of execution for the whole shebang

func NewCLI

func NewCLI(c *cli.Context) *CLI

NewCLI creates a new *CLI from a *cli.Context

func (*CLI) Run

func (i *CLI) Run()

Run starts all long-running processes and blocks until the processor pool returns from its Run func

func (*CLI) Setup

func (i *CLI) Setup() (bool, error)

Setup runs one-time preparatory actions and returns a boolean success value that is used to determine if it is safe to invoke the Run func

type Canceller

type Canceller interface {
	// Subscribe will set up a subscription for cancellation messages for the
	// given job ID. When a cancellation message comes in, the channel will be
	// closed. Only one subscription per job ID is valid, if there's already a
	// subscription set up, an error will be returned.
	Subscribe(id uint64, ch chan<- struct{}) error

	// Unsubscribe removes the existing subscription for the given job ID.
	Unsubscribe(id uint64)
}

A Canceller allows you to subscribe to and unsubscribe from cancellation messages for a given job ID.

type FileCanceller

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

FileCanceller fulfills the Canceller interface for file-based queues

func NewFileCanceller

func NewFileCanceller(ctx gocontext.Context, baseDir string) *FileCanceller

NewFileCanceller creates a *FileCanceller

func (*FileCanceller) Run

func (c *FileCanceller) Run()

Run is a no-op

func (*FileCanceller) Subscribe

func (c *FileCanceller) Subscribe(id uint64, ch chan<- struct{}) error

Subscribe is a no-op

func (*FileCanceller) Unsubscribe

func (c *FileCanceller) Unsubscribe(id uint64)

Unsubscribe is a no-op

type FileJobQueue

type FileJobQueue struct {
	DefaultLanguage, DefaultDist, DefaultGroup, DefaultOS string
	// contains filtered or unexported fields
}

FileJobQueue is a JobQueue that uses directories for input, state, and output

func NewFileJobQueue

func NewFileJobQueue(baseDir, queue string, pollingInterval time.Duration) (*FileJobQueue, error)

NewFileJobQueue creates a *FileJobQueue from a base directory and queue name

func (*FileJobQueue) Cleanup

func (f *FileJobQueue) Cleanup() error

Cleanup is a no-op

func (*FileJobQueue) Jobs

func (f *FileJobQueue) Jobs(ctx gocontext.Context) (<-chan Job, error)

Jobs returns a channel of jobs from the created directory

type FinishState

type FinishState string

FinishState is the state that a job finished with (such as pass/fail/etc.). You should not provide a string directly, but use one of the FinishStateX constants defined in this package.

const (
	FinishStatePassed    FinishState = "passed"
	FinishStateFailed    FinishState = "failed"
	FinishStateErrored   FinishState = "errored"
	FinishStateCancelled FinishState = "cancelled"
)

Valid finish states for the FinishState type

type Job

type Job interface {
	Payload() *JobPayload
	RawPayload() *simplejson.Json
	StartAttributes() *backend.StartAttributes

	Received() error
	Started() error
	Error(gocontext.Context, string) error
	Requeue() error
	Finish(FinishState) error

	LogWriter(gocontext.Context) (LogWriter, error)
}

A Job ties togeher all the elements required for a build job

type JobJobPayload

type JobJobPayload struct {
	ID     uint64 `json:"id"`
	Number string `json:"number"`
}

JobJobPayload contains information about the job.

type JobPayload

type JobPayload struct {
	Type       string                 `json:"type"`
	Job        JobJobPayload          `json:"job"`
	Build      BuildPayload           `json:"source"`
	Repository RepositoryPayload      `json:"repository"`
	UUID       string                 `json:"uuid"`
	Config     map[string]interface{} `json:"config"`
	Timeouts   TimeoutsPayload        `json:"timeouts,omitempty"`
}

JobPayload is the payload we receive over RabbitMQ.

type JobQueue

type JobQueue interface {
	Jobs(gocontext.Context) (<-chan Job, error)
	Cleanup() error
}

JobQueue is the minimal interface needed by a ProcessorPool

type LogWriter

type LogWriter interface {
	io.WriteCloser
	WriteAndClose([]byte) (int, error)
	SetTimeout(time.Duration)
	Timeout() <-chan time.Time
	SetMaxLogLength(int)
}

LogWriter is primarily an io.Writer that will send all bytes to travis-logs for processing, and also has some utility methods for timeouts and log length limiting. Each LogWriter is tied to a given job, and can be gotten by calling the LogWriter() method on a Job.

type Processor

type Processor struct {
	ID uuid.UUID

	ProcessedCount int

	SkipShutdownOnLogTimeout bool
	// contains filtered or unexported fields
}

A Processor will process build jobs on a channel, one by one, until it is told to shut down or the channel of build jobs closes.

func NewProcessor

func NewProcessor(ctx gocontext.Context, hostname string, buildJobsChan <-chan Job,
	provider backend.Provider, generator BuildScriptGenerator, canceller Canceller,
	hardTimeout time.Duration, logTimeout time.Duration) (*Processor, error)

NewProcessor creates a new processor that will run the build jobs on the given channel using the given provider and getting build scripts from the generator.

func (*Processor) GracefulShutdown

func (p *Processor) GracefulShutdown()

GracefulShutdown tells the processor to finish the job it is currently processing, but not pick up any new jobs. This method will return immediately, the processor is done when Run() returns.

func (*Processor) Run

func (p *Processor) Run()

Run starts the processor. This method will not return until the processor is terminated, either by calling the GracefulShutdown or Terminate methods, or if the build jobs channel is closed.

func (*Processor) Terminate

func (p *Processor) Terminate()

Terminate tells the processor to stop working on the current job as soon as possible.

type ProcessorPool

type ProcessorPool struct {
	Context     gocontext.Context
	Provider    backend.Provider
	Generator   BuildScriptGenerator
	Canceller   Canceller
	Hostname    string
	HardTimeout time.Duration
	LogTimeout  time.Duration

	SkipShutdownOnLogTimeout bool
	// contains filtered or unexported fields
}

A ProcessorPool spins up multiple Processors handling build jobs from the same queue.

func NewProcessorPool

func NewProcessorPool(hostname string, ctx gocontext.Context, hardTimeout time.Duration,
	logTimeout time.Duration, provider backend.Provider,
	generator BuildScriptGenerator, canceller Canceller) *ProcessorPool

NewProcessorPool creates a new processor pool using the given arguments.

func (*ProcessorPool) Decr

func (p *ProcessorPool) Decr()

Decr pops a processor out of the pool and issues a graceful shutdown

func (*ProcessorPool) Each

func (p *ProcessorPool) Each(f func(int, *Processor))

Each loops through all the processors in the pool and calls the given function for each of them, passing in the index and the processor. The order of the processors is the same for the same set of processors.

func (*ProcessorPool) GracefulShutdown

func (p *ProcessorPool) GracefulShutdown()

GracefulShutdown causes each processor in the pool to start its graceful shutdown.

func (*ProcessorPool) Incr

func (p *ProcessorPool) Incr()

Incr adds a single running processor to the pool

func (*ProcessorPool) Run

func (p *ProcessorPool) Run(poolSize int, queue JobQueue) error

Run starts up a number of processors and connects them to the given queue. This method stalls until all processors have finished.

func (*ProcessorPool) Size

func (p *ProcessorPool) Size() int

Size returns the number of processors in the pool

type RepositoryPayload

type RepositoryPayload struct {
	ID   uint64 `json:"id"`
	Slug string `json:"slug"`
}

RepositoryPayload contains information about the repository.

type SentryHook

type SentryHook struct {
	Timeout time.Duration
	// contains filtered or unexported fields
}

SentryHook delivers logs to a sentry server

func NewSentryHook

func NewSentryHook(DSN string, levels []logrus.Level) (*SentryHook, error)

NewSentryHook creates a hook to be added to an instance of logger and initializes the raven client. This method sets the timeout to 100 milliseconds.

func (*SentryHook) Fire

func (hook *SentryHook) Fire(entry *logrus.Entry) error

Fire is called when an event should be sent to sentry

func (*SentryHook) Levels

func (hook *SentryHook) Levels() []logrus.Level

Levels returns the available logging levels.

type TimeoutsPayload

type TimeoutsPayload struct {
	HardLimit  uint64 `json:"hard_limit"`
	LogSilence uint64 `json:"log_silence"`
}

TimeoutsPayload contains information about any custom timeouts. The timeouts are given in seconds, and a value of 0 means no custom timeout is set.