tasks

Documentation

Index

• Constants
• func DeferIgnoreError(f func() error)
• func GetCondition(t *v1alpha1.Task, name string) v1alpha1.StatusCondition
• func IgnoreError(err error)
• func IsDone(t *v1alpha1.Task) bool
• type FileStore
  • func NewFileStore(fileName string, log logr.Logger) (*FileStore, error)
  • func (s *FileStore) Create(ctx context.Context, t *v1alpha1.Task) (*v1alpha1.Task, error)
  • func (s *FileStore) Delete(ctx context.Context, name string) error
  • func (s *FileStore) Get(ctx context.Context, name string) (*v1alpha1.Task, error)
  • func (s *FileStore) List(ctx context.Context, workerId string, includeDone bool) ([]*v1alpha1.Task, error)
  • func (s *FileStore) Status(ctx context.Context, req *v1alpha1.StatusRequest) (*v1alpha1.StatusResponse, error)
  • func (s *FileStore) Update(ctx context.Context, t *v1alpha1.Task, worker string) (*v1alpha1.Task, error)
• type Server
  • func NewServer(tasks TaskInterface, log logr.Logger) (*Server, error)
  • func (s *Server) Create(ctx context.Context, req *v1alpha1.CreateRequest) (*v1alpha1.CreateResponse, error)
  • func (s *Server) Delete(ctx context.Context, req *v1alpha1.DeleteRequest) (*v1alpha1.DeleteResponse, error)
  • func (s *Server) Get(ctx context.Context, req *v1alpha1.GetRequest) (*v1alpha1.GetResponse, error)
  • func (s *Server) List(ctx context.Context, req *v1alpha1.ListRequest) (*v1alpha1.ListResponse, error)
  • func (s *Server) Status(ctx context.Context, req *v1alpha1.StatusRequest) (*v1alpha1.StatusResponse, error)
  • func (s *Server) Update(ctx context.Context, req *v1alpha1.UpdateRequest) (*v1alpha1.UpdateResponse, error)
• type StoredData
  • func (s *StoredData) AssignGroup(group string, workerId string)
• type TaskInterface

Constants

const (
	SUCCEEDED_CONDITION = "succeeded"
)

Functions

func DeferIgnoreError

func DeferIgnoreError(f func() error)

DeferIgnoreError is a helper function to ignore errors returned by functions called with defer.

func GetCondition

func GetCondition(t *v1alpha1.Task, name string) v1alpha1.StatusCondition

GetCondition returns the value of the specified condition. Returns UNKNOWN if there is no value for the condition.

func IgnoreError

func IgnoreError(err error)

IgnoreError is a helper function to deal with errors.

func IsDone

func IsDone(t *v1alpha1.Task) bool

IsDone Returns True iff the task is Done is true and false otherise

Types

type FileStore

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

FileStore is a simple implementation of the TaskInterface which uses a file for persistence. FileStore is intended to be a non-performant, non-scalable reference implementation.

func NewFileStore

func NewFileStore(fileName string, log logr.Logger) (*FileStore, error)

NewFileStore will create a new filestore. Tasks will be persisted to fileName. If fileName already exists it should have been created by FileStore. The directory for fileName must exist.

func (*FileStore) Create

func (s *FileStore) Create(ctx context.Context, t *v1alpha1.Task) (*v1alpha1.Task, error)

func (*FileStore) Delete

func (s *FileStore) Delete(ctx context.Context, name string) error

Delete the specified task. Delete is a null op if the task doesn't exist

func (*FileStore) Get

func (s *FileStore) Get(ctx context.Context, name string) (*v1alpha1.Task, error)

Get the task with the specified name.

func (*FileStore) List

func (s *FileStore) List(ctx context.Context, workerId string, includeDone bool) ([]*v1alpha1.Task, error)

List returns tasks for this worker. If the worker isn't already assigned a task group one is assigned.

func (*FileStore) Status

func (s *FileStore) Status(ctx context.Context, req *v1alpha1.StatusRequest) (*v1alpha1.StatusResponse, error)

Status returns status information about the taskstore. N.B this method blocks other updates and is potentially expensive.

func (*FileStore) Update

func (s *FileStore) Update(ctx context.Context, t *v1alpha1.Task, worker string) (*v1alpha1.Task, error)

Update the specified task. If the task doesn't exist it will be created

type Server

type Server struct {
	v1alpha1.UnimplementedTasksServiceServer
	// contains filtered or unexported fields
}

Server implments the gRPC interface for tasks.

func NewServer

func NewServer(tasks TaskInterface, log logr.Logger) (*Server, error)

NewServer constructs a new server.

func (*Server) Create

func (s *Server) Create(ctx context.Context, req *v1alpha1.CreateRequest) (*v1alpha1.CreateResponse, error)

func (*Server) Delete

func (s *Server) Delete(ctx context.Context, req *v1alpha1.DeleteRequest) (*v1alpha1.DeleteResponse, error)

func (*Server) Get

func (s *Server) Get(ctx context.Context, req *v1alpha1.GetRequest) (*v1alpha1.GetResponse, error)

func (*Server) List

func (s *Server) List(ctx context.Context, req *v1alpha1.ListRequest) (*v1alpha1.ListResponse, error)

func (*Server) Status

func (s *Server) Status(ctx context.Context, req *v1alpha1.StatusRequest) (*v1alpha1.StatusResponse, error)

func (*Server) Update

func (s *Server) Update(ctx context.Context, req *v1alpha1.UpdateRequest) (*v1alpha1.UpdateResponse, error)

type StoredData

type StoredData struct {

	// the tasks
	Tasks map[string]*v1alpha1.Task

	// Group to worker is a mapping from groups to workers; empty means its unassigned.
	// It is the inverse of WorkerIdToGroupNonce
	GroupToWorker map[string]string

	// Mapping from a worker id to the group nonce
	WorkerIdToGroupNonce map[string]string

	// Should we make this a list of only incomplete tasks?
	GroupToTaskNames map[string][]string
	// contains filtered or unexported fields
}

StoredData represents all the data to be stored and serialized in the file. TODO(jeremy): Should we make this a proto?

func (*StoredData) AssignGroup

func (s *StoredData) AssignGroup(group string, workerId string)

AssignGroup Assigns the group to the worker. Overrides existing assignments

type TaskInterface

type TaskInterface interface {
	Create(ctx context.Context, task *v1alpha1.Task) (*v1alpha1.Task, error)
	Get(ctx context.Context, name string) (*v1alpha1.Task, error)
	// Update departs from K8s convention because we need to provide the ID of the worker issuing the update
	// request. In principle that should be sent via a sidechannel (e.g. as part of Authn)  so it wouldn't
	// be in the body of the request.
	Update(ctx context.Context, task *v1alpha1.Task, worker string) (*v1alpha1.Task, error)
	Delete(ctx context.Context, name string) error
	List(ctx context.Context, workerId string, includeDone bool) ([]*v1alpha1.Task, error)
	Status(ctx context.Context, req *v1alpha1.StatusRequest) (*v1alpha1.StatusResponse, error)
}

TaskInterface defines the CRUD interface for the taskstore. Modeled on K8s; e.g. https://pkg.go.dev/k8s.io/client-go/kubernetes/typed/core/v1#PodInterface