jobservices

Documentation

Overview

Package jobservices handles services necessary WRT handling jobs from SDKs. Nominally this is the entry point for most users, and a job's external interactions outside of pipeline execution.

This includes handling receiving, staging, and provisioning artifacts, and orchestrating external workers, such as for loopback mode.

Execution of jobs is abstracted away to an execute function specified at server construction time.

Index

• Variables
• type Job
  • func (j *Job) AddMetricShortIDs(ids *fnpb.MonitoringInfosMetadataResponse)
  • func (j *Job) ArtifactEndpoint() string
  • func (j *Job) Canceled()
  • func (j *Job) Canceling()
  • func (j *Job) ContributeFinalMetrics(payloads *fnpb.ProcessBundleResponse) []string
  • func (j *Job) ContributeTentativeMetrics(payloads *fnpb.ProcessBundleProgressResponse) (map[string]int64, []string)
  • func (j *Job) Done()
  • func (j *Job) Failed(err error)
  • func (j *Job) JobKey() string
  • func (j *Job) LogValue() slog.Value
  • func (j *Job) PipelineOptions() *structpb.Struct
  • func (j *Job) Running()
  • func (j *Job) SendMsg(msg string)
  • func (j *Job) Start()
  • func (j *Job) String() string
• type Server
  • func NewServer(port int, execute func(*Job)) *Server
  • func (s *Server) Cancel(_ context.Context, req *jobpb.CancelJobRequest) (*jobpb.CancelJobResponse, error)
  • func (s *Server) DescribePipelineOptions(context.Context, *jobpb.DescribePipelineOptionsRequest) (*jobpb.DescribePipelineOptionsResponse, error)
  • func (s *Server) Endpoint() string
  • func (s *Server) GetArtifact(req *jobpb.GetArtifactRequest, ...) error
  • func (s *Server) GetJobMetrics(ctx context.Context, req *jobpb.GetJobMetricsRequest) (*jobpb.GetJobMetricsResponse, error)
  • func (s *Server) GetJobs(context.Context, *jobpb.GetJobsRequest) (*jobpb.GetJobsResponse, error)
  • func (s *Server) GetMessageStream(req *jobpb.JobMessagesRequest, stream jobpb.JobService_GetMessageStreamServer) error
  • func (s *Server) GetPipeline(_ context.Context, req *jobpb.GetJobPipelineRequest) (*jobpb.GetJobPipelineResponse, error)
  • func (s *Server) GetState(_ context.Context, req *jobpb.GetJobStateRequest) (*jobpb.JobStateEvent, error)
  • func (s *Server) GetStateStream(req *jobpb.GetJobStateRequest, stream jobpb.JobService_GetStateStreamServer) error
  • func (s *Server) IdleShutdown(timeout time.Duration, cancelFn context.CancelCauseFunc)
  • func (s *Server) Prepare(ctx context.Context, req *jobpb.PrepareJobRequest) (_ *jobpb.PrepareJobResponse, err error)
  • func (s *Server) ResolveArtifacts(_ context.Context, req *jobpb.ResolveArtifactsRequest) (*jobpb.ResolveArtifactsResponse, error)
  • func (s *Server) ReverseArtifactRetrievalService(stream jobpb.ArtifactStagingService_ReverseArtifactRetrievalServiceServer) error
  • func (s *Server) Run(ctx context.Context, req *jobpb.RunJobRequest) (*jobpb.RunJobResponse, error)
  • func (s *Server) Serve()
  • func (s *Server) Stop()

Variables

var (
	// ErrCancel represents a pipeline cancellation by the user.
	ErrCancel = errors.New("pipeline canceled")
)

Types

type Job

type Job struct {
	Pipeline *pipepb.Pipeline

	// Context used to terminate this job.
	RootCtx  context.Context
	CancelFn context.CancelCauseFunc
	// Logger for this job.
	Logger *slog.Logger
	// contains filtered or unexported fields
}

Job is an interface to the job services for executing pipelines. It allows the executor to communicate status, messages, and metrics back to callers of the Job Management API.

func (*Job) AddMetricShortIDs

func (j *Job) AddMetricShortIDs(ids *fnpb.MonitoringInfosMetadataResponse)

AddMetricShortIDs populates metric short IDs with their metadata.

func (*Job) ArtifactEndpoint

func (j *Job) ArtifactEndpoint() string

func (*Job) Canceled

func (j *Job) Canceled()

Canceled indicates that the job is canceled.

func (*Job) Canceling

func (j *Job) Canceling()

Canceling indicates that the job is canceling.

func (*Job) ContributeFinalMetrics

func (j *Job) ContributeFinalMetrics(payloads *fnpb.ProcessBundleResponse) []string

ContributeFinalMetrics returns any unknown monitoring short ids.

func (*Job) ContributeTentativeMetrics

func (j *Job) ContributeTentativeMetrics(payloads *fnpb.ProcessBundleProgressResponse) (map[string]int64, []string)

ContributeTentativeMetrics returns the datachannel read index, and any unknown monitoring short ids.

func (*Job) Done

func (j *Job) Done()

Done indicates that the job completed successfully.

func (*Job) Failed

func (j *Job) Failed(err error)

Failed indicates that the job completed unsuccessfully.

func (*Job) JobKey

func (j *Job) JobKey() string

func (*Job) LogValue

func (j *Job) LogValue() slog.Value

func (*Job) PipelineOptions

func (j *Job) PipelineOptions() *structpb.Struct

func (*Job) Running

func (j *Job) Running()

Running indicates that the job is executing.

func (*Job) SendMsg

func (j *Job) SendMsg(msg string)

func (*Job) Start

func (j *Job) Start()

Start indicates that the job is preparing to execute.

func (*Job) String

func (j *Job) String() string

type Server

type Server struct {
	jobpb.UnimplementedJobServiceServer
	jobpb.UnimplementedArtifactStagingServiceServer
	jobpb.UnimplementedArtifactRetrievalServiceServer
	fnpb.UnimplementedProvisionServiceServer
	// contains filtered or unexported fields
}

func NewServer

func NewServer(port int, execute func(*Job)) *Server

NewServer acquires the indicated port.

func (*Server) Cancel

func (s *Server) Cancel(_ context.Context, req *jobpb.CancelJobRequest) (*jobpb.CancelJobResponse, error)

Cancel a Job requested by the CancelJobRequest for jobs not in an already terminal state. Otherwise, returns nil if Job does not exist or the Job's existing state as part of the CancelJobResponse.

func (*Server) DescribePipelineOptions

func (s *Server) DescribePipelineOptions(context.Context, *jobpb.DescribePipelineOptionsRequest) (*jobpb.DescribePipelineOptionsResponse, error)

DescribePipelineOptions is a no-op since it's unclear how it is to function. Apparently only implemented in the Python SDK.

func (*Server) Endpoint

func (s *Server) Endpoint() string

func (*Server) GetArtifact

func (s *Server) GetArtifact(req *jobpb.GetArtifactRequest, stream jobpb.ArtifactRetrievalService_GetArtifactServer) error

func (*Server) GetJobMetrics

func (s *Server) GetJobMetrics(ctx context.Context, req *jobpb.GetJobMetricsRequest) (*jobpb.GetJobMetricsResponse, error)

GetJobMetrics Fetch metrics for a given job.

func (*Server) GetJobs

func (s *Server) GetJobs(context.Context, *jobpb.GetJobsRequest) (*jobpb.GetJobsResponse, error)

GetJobs returns the set of active jobs and associated metadata.

func (*Server) GetMessageStream

func (s *Server) GetMessageStream(req *jobpb.JobMessagesRequest, stream jobpb.JobService_GetMessageStreamServer) error

GetMessageStream subscribes to a stream of state changes and messages from the job. If throughput is high, this may cause losses of messages.

func (*Server) GetPipeline

func (s *Server) GetPipeline(_ context.Context, req *jobpb.GetJobPipelineRequest) (*jobpb.GetJobPipelineResponse, error)

GetPipeline returns pipeline proto of the requested job id.

func (*Server) GetState

func (s *Server) GetState(_ context.Context, req *jobpb.GetJobStateRequest) (*jobpb.JobStateEvent, error)

GetState returns the current state of the job with the requested id.

func (*Server) GetStateStream

func (s *Server) GetStateStream(req *jobpb.GetJobStateRequest, stream jobpb.JobService_GetStateStreamServer) error

GetStateStream returns the job state as it changes.

func (*Server) IdleShutdown

func (s *Server) IdleShutdown(timeout time.Duration, cancelFn context.CancelCauseFunc)

IdleShutdown allows the server to call the cancelFn if there have been no active jobs for at least the given timeout.

func (*Server) Prepare

func (s *Server) Prepare(ctx context.Context, req *jobpb.PrepareJobRequest) (_ *jobpb.PrepareJobResponse, err error)

func (*Server) ResolveArtifacts

func (s *Server) ResolveArtifacts(_ context.Context, req *jobpb.ResolveArtifactsRequest) (*jobpb.ResolveArtifactsResponse, error)

func (*Server) ReverseArtifactRetrievalService

func (s *Server) ReverseArtifactRetrievalService(stream jobpb.ArtifactStagingService_ReverseArtifactRetrievalServiceServer) error

func (*Server) Run

func (s *Server) Run(ctx context.Context, req *jobpb.RunJobRequest) (*jobpb.RunJobResponse, error)

func (*Server) Serve

func (s *Server) Serve()

Serve serves on the started listener. Blocks.

func (*Server) Stop

func (s *Server) Stop()

Stop the GRPC server.