queue

Documentation

Overview

Package queue provides the ability for Vela to integrate with different supported Queue backends.

Usage:

import "github.com/go-vela/server/queue"

Index

• Variables
• func WithGinContext(c *gin.Context, s Service)
• type Service
  • func FromCLICommand(ctx context.Context, c *cli.Command) (Service, error)
  • func FromContext(c context.Context) Service
  • func FromGinContext(c *gin.Context) Service
  • func New(ctx context.Context, s *Setup) (Service, error)
• type Setup
  • func (s *Setup) Kafka() (Service, error)
  • func (s *Setup) Redis(ctx context.Context) (Service, error)
  • func (s *Setup) Validate() error

Variables

var Flags = []cli.Flag{

	&cli.StringFlag{
		Name:  "queue.driver",
		Usage: "driver to be used for the queue",
		Sources: cli.NewValueSourceChain(
			cli.EnvVar("VELA_QUEUE_DRIVER"),
			cli.EnvVar("QUEUE_DRIVER"),
			cli.File("/vela/queue/driver"),
		),
		Required: true,
	},
	&cli.StringFlag{
		Name:  "queue.addr",
		Usage: "fully qualified url (<scheme>://<host>) for the queue",
		Sources: cli.NewValueSourceChain(
			cli.EnvVar("VELA_QUEUE_ADDR"),
			cli.EnvVar("QUEUE_ADDR"),
			cli.File("/vela/queue/addr"),
		),
		Action: func(_ context.Context, _ *cli.Command, v string) error {

			if !strings.Contains(v, "://") {
				return fmt.Errorf("queue address must be fully qualified (<scheme>://<host>)")
			}

			if strings.HasSuffix(v, "/") {
				return fmt.Errorf("queue address must not have trailing slash")
			}

			return nil
		},
	},
	&cli.BoolFlag{
		Name:  "queue.cluster",
		Usage: "enables connecting to a queue cluster",
		Sources: cli.NewValueSourceChain(
			cli.EnvVar("VELA_QUEUE_CLUSTER"),
			cli.EnvVar("QUEUE_CLUSTER"),
			cli.File("/vela/queue/cluster"),
		),
	},
	&cli.StringSliceFlag{
		Name:  "queue.routes",
		Usage: "list of routes (channels/topics) to publish builds",
		Sources: cli.NewValueSourceChain(
			cli.EnvVar("VELA_QUEUE_ROUTES"),
			cli.EnvVar("QUEUE_ROUTES"),
			cli.File("/vela/queue/routes"),
		),
		Value: []string{constants.DefaultRoute},
	},
	&cli.DurationFlag{
		Name:  "queue.pop.timeout",
		Usage: "timeout for requests that pop items off the queue",
		Sources: cli.NewValueSourceChain(
			cli.EnvVar("VELA_QUEUE_POP_TIMEOUT"),
			cli.EnvVar("QUEUE_POP_TIMEOUT"),
			cli.File("/vela/queue/pop_timeout"),
		),
		Value: 60 * time.Second,
	},
	&cli.StringFlag{
		Name:  "queue.private-key",
		Usage: "set value of base64 encoded queue signing private key",
		Sources: cli.NewValueSourceChain(
			cli.EnvVar("VELA_QUEUE_PRIVATE_KEY"),
			cli.EnvVar("QUEUE_PRIVATE_KEY"),
			cli.File("/vela/signing.key"),
		),
	},
	&cli.StringFlag{
		Name:  "queue.public-key",
		Usage: "set value of base64 encoded queue signing public key",
		Sources: cli.NewValueSourceChain(
			cli.EnvVar("VELA_QUEUE_PUBLIC_KEY"),
			cli.EnvVar("QUEUE_PUBLIC_KEY"),
			cli.File("/vela/signing.pub"),
		),
	},
}

Flags represents all supported command line interface (CLI) flags for the queue.

https://pkg.go.dev/github.com/urfave/cli?tab=doc#Flag

Functions

func WithGinContext

func WithGinContext(c *gin.Context, s Service)

WithGinContext inserts the queue Service into the gin.Context.

Types

type Service

type Service interface {

	// Driver defines a function that outputs
	// the configured queue driver.
	Driver() string

	// Length defines a function that outputs
	// the length of all queue routes
	Length(context.Context) (int64, error)

	// RouteLength defines a function that outputs
	// the length of a defined queue route
	RouteLength(context.Context, string) (int64, error)

	// Pop defines a function that grabs an
	// item off the queue.
	Pop(context.Context, []string) (*models.Item, error)

	// Push defines a function that publishes an
	// item to the specified route in the queue.
	Push(context.Context, string, []byte) error

	// Ping defines a function that checks the
	// connection to the queue.
	Ping(context.Context) error

	// Route defines a function that decides which
	// route a build gets placed within the queue.
	Route(*pipeline.Worker) (string, error)

	// GetSettings defines a function that returns
	// queue settings.
	GetSettings() settings.Queue

	// SetSettings defines a function that takes api settings
	// and updates the compiler Engine.
	SetSettings(*settings.Platform)
}

Service represents the interface for Vela integrating with the different supported Queue backends.

func FromCLICommand

func FromCLICommand(ctx context.Context, c *cli.Command) (Service, error)

FromCLICommand helper function to setup the queue from the CLI arguments.

func FromContext

func FromContext(c context.Context) Service

FromContext retrieves the queue Service from the context.Context.

func FromGinContext

func FromGinContext(c *gin.Context) Service

FromGinContext retrieves the queue Service from the gin.Context.

func New

func New(ctx context.Context, s *Setup) (Service, error)

New creates and returns a Vela service capable of integrating with the configured queue environment. Currently, the following queues are supported:

* redis .

type Setup

type Setup struct {

	// specifies the driver to use for the queue client
	Driver string
	// specifies the address to use for the queue client
	Address string
	// enables the queue client to integrate with a queue cluster
	Cluster bool
	// specifies a list of routes (channels/topics) for managing builds for the queue client
	Routes []string
	// specifies the timeout for pop requests for the queue client
	Timeout time.Duration
	// private key in base64 used for signing items pushed to the queue
	PrivateKey string
	// public key in base64 used for opening items popped from the queue
	PublicKey string
}

Setup represents the configuration necessary for creating a Vela service capable of integrating with a configured queue environment.

func (*Setup) Kafka

func (s *Setup) Kafka() (Service, error)

Kafka creates and returns a Vela service capable of integrating with a Kafka queue.

func (*Setup) Redis

func (s *Setup) Redis(ctx context.Context) (Service, error)

Redis creates and returns a Vela service capable of integrating with a Redis queue.

func (*Setup) Validate

func (s *Setup) Validate() error

Validate verifies the necessary fields for the provided configuration are populated correctly.