ppsutil

Documentation

Overview

Package ppsutil contains utilities for various PPS-related tasks, which are shared by both the PPS API and the worker binary. These utilities include: - Getting the RC name and querying k8s reguarding pipelines - Reading and writing pipeline resource requests and limits - Reading and writing EtcdPipelineInfos and PipelineInfos[1]

[1] Note that PipelineInfo in particular is complicated because it contains fields that are not always set or are stored in multiple places ('job_state', for example, is not stored in PFS along with the rest of each PipelineInfo, because this field is volatile and we cannot commit to PFS every time it changes. 'job_counts' is the same, and 'reason' is in etcd because it is only updated alongside 'job_state'). As of 12/7/2017, these are the only fields not stored in PFS.

Index

• func ContainsS3Inputs(in *pps.Input) bool
• func FailPipeline(ctx context.Context, etcdClient *etcd.Client, ...) error
• func GetExpectedNumHashtrees(spec *pps.HashtreeSpec) (int64, error)
• func GetExpectedNumWorkers(kubeClient *kube.Clientset, spec *pps.ParallelismSpec) (int, error)
• func GetLimitsResourceListFromPipeline(pipelineInfo *pps.PipelineInfo) (*v1.ResourceList, error)
• func GetPipelineInfo(pachClient *client.APIClient, ptr *pps.EtcdPipelineInfo) (*pps.PipelineInfo, error)
• func GetRequestsResourceListFromPipeline(pipelineInfo *pps.PipelineInfo) (*v1.ResourceList, error)
• func IsTerminal(state pps.JobState) bool
• func JobInput(pipelineInfo *pps.PipelineInfo, outputCommitInfo *pfs.CommitInfo) *pps.Input
• func PipelineRcName(name string, version uint64) string
• func PipelineRepo(pipeline *pps.Pipeline) *pfs.Repo
• func PipelineReqFromInfo(pipelineInfo *pps.PipelineInfo) *pps.CreatePipelineRequest
• func SidecarS3GatewayService(jobID string) string
• func UpdateJobState(pipelines col.ReadWriteCollection, jobs col.ReadWriteCollection, ...) error
• type PipelineManifestReader
  • func NewPipelineManifestReader(path string) (result *PipelineManifestReader, retErr error)
  • func (r *PipelineManifestReader) NextCreatePipelineRequest() (*ppsclient.CreatePipelineRequest, error)

Functions

func ContainsS3Inputs

func ContainsS3Inputs(in *pps.Input) bool

ContainsS3Inputs returns 'true' if 'in' is or contains any PFS inputs with 'S3' set to true. Any pipelines with s3 inputs lj

func FailPipeline

func FailPipeline(ctx context.Context, etcdClient *etcd.Client, pipelinesCollection col.Collection, pipelineName string, reason string) error

FailPipeline updates the pipeline's state to failed and sets the failure reason

func GetExpectedNumHashtrees

func GetExpectedNumHashtrees(spec *pps.HashtreeSpec) (int64, error)

GetExpectedNumHashtrees computes the expected number of hashtrees that Pachyderm will create given the HashtreeSpec 'spec'.

func GetExpectedNumWorkers

func GetExpectedNumWorkers(kubeClient *kube.Clientset, spec *pps.ParallelismSpec) (int, error)

GetExpectedNumWorkers computes the expected number of workers that pachyderm will start given the ParallelismSpec 'spec'.

This is only exported for testing

func GetLimitsResourceListFromPipeline

func GetLimitsResourceListFromPipeline(pipelineInfo *pps.PipelineInfo) (*v1.ResourceList, error)

GetLimitsResourceListFromPipeline returns a list of resources that the pipeline, maximally is limited to.

func GetPipelineInfo

func GetPipelineInfo(pachClient *client.APIClient, ptr *pps.EtcdPipelineInfo) (*pps.PipelineInfo, error)

GetPipelineInfo retrieves and returns a valid PipelineInfo from PFS. It does the PFS read/unmarshalling of bytes as well as filling in missing fields

func GetRequestsResourceListFromPipeline

func GetRequestsResourceListFromPipeline(pipelineInfo *pps.PipelineInfo) (*v1.ResourceList, error)

GetRequestsResourceListFromPipeline returns a list of resources that the pipeline, minimally requires.

func IsTerminal

func IsTerminal(state pps.JobState) bool

IsTerminal returns 'true' if 'state' indicates that the job is done (i.e. the state will not change later: SUCCESS, FAILURE, KILLED) and 'false' otherwise.

func JobInput

func JobInput(pipelineInfo *pps.PipelineInfo, outputCommitInfo *pfs.CommitInfo) *pps.Input

JobInput fills in the commits for a JobInfo

func PipelineRcName

func PipelineRcName(name string, version uint64) string

PipelineRcName generates the name of the k8s replication controller that manages a pipeline's workers

func PipelineRepo

func PipelineRepo(pipeline *pps.Pipeline) *pfs.Repo

PipelineRepo creates a pfs repo for a given pipeline.

func PipelineReqFromInfo

func PipelineReqFromInfo(pipelineInfo *pps.PipelineInfo) *pps.CreatePipelineRequest

PipelineReqFromInfo converts a PipelineInfo into a CreatePipelineRequest.

func SidecarS3GatewayService

func SidecarS3GatewayService(jobID string) string

SidecarS3GatewayService returns the name of the kubernetes service created for the job 'jobID' to hand sidecar s3 gateway requests. This helper is in ppsutil because both PPS (which creates the service, in the s3 gateway sidecar server) and the worker (which passes the endpoint to the user code) need to know it.

func UpdateJobState

func UpdateJobState(pipelines col.ReadWriteCollection, jobs col.ReadWriteCollection, jobPtr *pps.EtcdJobInfo, state pps.JobState, reason string) error

UpdateJobState performs the operations involved with a job state transition.

Types

type PipelineManifestReader

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

PipelineManifestReader helps with unmarshalling pipeline configs from JSON. It's used by 'create pipeline' and 'update pipeline'

func NewPipelineManifestReader

func NewPipelineManifestReader(path string) (result *PipelineManifestReader, retErr error)

NewPipelineManifestReader creates a new manifest reader from a path.

func (*PipelineManifestReader) NextCreatePipelineRequest

func (r *PipelineManifestReader) NextCreatePipelineRequest() (*ppsclient.CreatePipelineRequest, error)

NextCreatePipelineRequest gets the next request from the manifest reader.