scriptrunner

Documentation

Index

• Constants
• Variables
• func VizierStatusToStatus(s *vizierpb.Status) (*statuspb.Status, error)
• type CloudSource
  • func NewCloudSource(nc *nats.Conn, csClient metadatapb.CronScriptStoreServiceClient, ...) *CloudSource
  • func (source *CloudSource) Start(baseCtx context.Context, updatesCh chan<- *cvmsgspb.CronScriptUpdate) (map[string]*cvmsgspb.CronScript, error)
  • func (source *CloudSource) Stop()
• type ConfigMapSource
  • func NewConfigMapSource(client kubernetes.Interface, namespace string) *ConfigMapSource
  • func (source *ConfigMapSource) Start(ctx context.Context, updatesCh chan<- *cvmsgspb.CronScriptUpdate) (map[string]*cvmsgspb.CronScript, error)
  • func (source *ConfigMapSource) Stop()
• type ScriptRunner
  • func New(csClient metadatapb.CronScriptStoreServiceClient, ...) *ScriptRunner
  • func (s *ScriptRunner) Stop()
  • func (s *ScriptRunner) SyncScripts() error
• type Source
  • func Sources(nc *nats.Conn, csClient metadatapb.CronScriptStoreServiceClient, ...) []Source

Constants

const (
	CloudSourceName     = "cloud"
	ConfigMapSourceName = "configmaps"
)

Names of [Source]s

Variables

var (
	// CronScriptChecksumRequestChannel is the NATS channel to make checksum requests to.
	CronScriptChecksumRequestChannel = messagebus.V2CTopic(cvmsgs.CronScriptChecksumRequestChannel)
	// CronScriptChecksumResponseChannel is the NATS channel that checksum responses are published to.
	CronScriptChecksumResponseChannel = messagebus.C2VTopic(cvmsgs.CronScriptChecksumResponseChannel)
	// GetCronScriptsRequestChannel is the NATS channel script requests are sent to.
	GetCronScriptsRequestChannel = messagebus.V2CTopic(cvmsgs.GetCronScriptsRequestChannel)
	// GetCronScriptsResponseChannel is the NATS channel that script responses are published to.
	GetCronScriptsResponseChannel = messagebus.C2VTopic(cvmsgs.GetCronScriptsResponseChannel)
	// CronScriptUpdatesChannel is the NATS channel that any cron script updates are published to.
	CronScriptUpdatesChannel = messagebus.C2VTopic(cvmsgs.CronScriptUpdatesChannel)
	// CronScriptUpdatesResponseChannel is the NATS channel that script updates are published to.
	CronScriptUpdatesResponseChannel = messagebus.V2CTopic(cvmsgs.CronScriptUpdatesResponseChannel)
)

var DefaultSources = []string{CloudSourceName}

DefaultSources is a list of sources enabled by default

Functions

func VizierStatusToStatus

func VizierStatusToStatus(s *vizierpb.Status) (*statuspb.Status, error)

VizierStatusToStatus converts the Vizier status to the internal storable version statuspb.Status

Types

type CloudSource

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

CloudSource is a Source that pulls cron scripts from the cloud.

func NewCloudSource

func NewCloudSource(nc *nats.Conn, csClient metadatapb.CronScriptStoreServiceClient, signingKey string) *CloudSource

NewCloudSource constructs a Source that will pull cron scripts from the cloud.

func (*CloudSource) Start

func (source *CloudSource) Start(baseCtx context.Context, updatesCh chan<- *cvmsgspb.CronScriptUpdate) (map[string]*cvmsgspb.CronScript, error)

Start subscribes to updates from the cloud on the CronScriptUpdatesChannel and sends resulting updates on updatesCh.

func (*CloudSource) Stop

func (source *CloudSource) Stop()

Stop stops further updates from being sent.

type ConfigMapSource

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

ConfigMapSource pulls cron scripts from config maps.

func NewConfigMapSource

func NewConfigMapSource(client kubernetes.Interface, namespace string) *ConfigMapSource

NewConfigMapSource constructs a Source that extracts cron scripts from config maps with the label "purpose=cron-script". Each config map must contain

• a script.pxl with the pixel script
• a configs.yaml which will be stored in the Configs field of cvmsgspb.CronScript
• a cron.yaml that contains a "frequency_s" key

func (*ConfigMapSource) Start

func (source *ConfigMapSource) Start(ctx context.Context, updatesCh chan<- *cvmsgspb.CronScriptUpdate) (map[string]*cvmsgspb.CronScript, error)

Start watches for updates to matching configmaps and sends resulting updates on updatesCh.

func (*ConfigMapSource) Stop

func (source *ConfigMapSource) Stop()

Stop stops further updates from being sent.

type ScriptRunner

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

ScriptRunner tracks registered cron scripts and runs them according to schedule.

func New

func New(
	csClient metadatapb.CronScriptStoreServiceClient,
	vzClient vizierpb.VizierServiceClient,
	signingKey string,
	scriptSources ...Source,
) *ScriptRunner

New creates a new script runner.

func (*ScriptRunner) Stop

func (s *ScriptRunner) Stop()

Stop performs any necessary cleanup before shutdown.

func (*ScriptRunner) SyncScripts

func (s *ScriptRunner) SyncScripts() error

SyncScripts syncs the known set of scripts in Vizier with scripts in Cloud.

type Source

type Source interface {
	// Start sends updates on updatesCh and returns the initial set of scripts which the updates are based on. It does not block.
	Start(baseCtx context.Context, updatesCh chan<- *cvmsgspb.CronScriptUpdate) (map[string]*cvmsgspb.CronScript, error)

	// Stop sending updates on the updatesCh provided in Start.
	// This method must not be called before Start.
	Stop()
}

A Source provides an initial set of cron scripts and sends incremental updates to that set.

func Sources

func Sources(nc *nats.Conn, csClient metadatapb.CronScriptStoreServiceClient, signingKey string, namespace string, sourceNames []string) []Source

Sources initializes multiple sources based on the set of sourceNames provided.