Help us shape the future of Go! Take the 2026 Go Developer Survey

gcpbatch

package
v1.9.1 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Nov 1, 2026 License: MIT Imports: 43 Imported by: 0

README

Google Cloud Batch

Parity grade: B · SDK cloud.google.com/go/batch@v1.24.1 · last audited 2026-10-30 (b0763c598)

Coverage

Metric Value
PARITY entries audited 14 (8 ok, 5 partial, 1 gap)
Feature families 9 (4 ok, 5 partial)
Known gaps 9
Deferred items 0
Resource leaks clean
Known gaps
  • Barrier runnables are accepted and stored but are not synchronised across tasks: they are skipped, so tasks do not wait for each other.
  • Compute Engine behaviour is not emulated: no VMs, so allocation_policy has no effect, instances stay empty, tasks are never preempted (exit codes 50001 to 50004 and 50006 cannot occur) and task_count_per_node, permissive_ssh and run_as_non_root are stored only. The 50005 exit code for exceeded run time is from the troubleshooting page.
  • Predefined environment variables other than BATCH_TASK_INDEX, BATCH_TASK_COUNT and BATCH_HOSTS_FILE (for example the retry attempt number or job identity) are not set: the primary pages fetched did not list them. The hosts file content (one name per task) is not documented.
  • Container runnable options, username and password (registry credentials) and enable_image_streaming are stored but ignored; script runnables run in a container image rather than on the VM host, so host paths a script path refers to must exist in that image.
  • Task status event texts other than the documented 'Task state is updated from X to Y' prefix, the TASK_RETRY event type and the retry description are emulator wording; the task event type names and the initial job events are not in a primary source.
  • Job.notifications are stored but no Pub/Sub message is sent.
  • Validation and state-conflict error codes and texts (AlreadyExists, FAILED_PRECONDITION for a repeated delete or cancel, the task_count upper bound of 100000 enforced here) have no primary text; the REST filter grammar of ListJobs is a subset.
  • No IAM policy methods, CancelOperation, dashboard page or proxy.
  • In docker mode a job that was RUNNING when the emulator restarts is marked FAILED on restore (its containers are gone) instead of being resumed.

More

Documentation

Index

Constants

View Source 
const (
	// DefaultPort is the dedicated gRPC and REST listener's default TCP port.
	DefaultPort = 8147
	// EngineNone records jobs and progresses them on the clock without running any container.
	EngineNone = "none"
	// EngineDocker runs every task's runnables as containers.
	EngineDocker = "docker"
	// DefaultScriptImage runs script runnables, which Batch executes on the VM host.
	DefaultScriptImage = "debian:bookworm-slim"
)
View Source 
const (

	// HostsFilePath is where BATCH_HOSTS_FILE points when require_hosts_file is set.
	HostsFilePath = "/etc/cloudbatch-taskgroup-hosts"
)

Variables

View Source 
var ErrNilAppContext = errors.New("gcpbatch: nil app context")

ErrNilAppContext is returned when Init is called with a nil AppContext.

Functions

This section is empty.

Types

type ConfigProvider 

type ConfigProvider interface {
	GetGCPBatchSettings() Settings
}

ConfigProvider extracts Batch configuration from the AppContext Config.

type Handler 

type Handler struct {
	Backend *InMemoryBackend

	Port int
	// contains filtered or unexported fields
}

Handler serves the Batch API (gRPC and REST) on a dedicated listener.

func NewHandler 

func NewHandler(backend *InMemoryBackend) *Handler

NewHandler creates a Handler; callers override Port from Settings.

func (*Handler) API 

func (h *Handler) API() batchpb.BatchServiceServer

API returns the Batch service server for in-process callers.

func (*Handler) Addr 

func (h *Handler) Addr() string

Addr returns the bound listener address, or "" before StartWorker.

func (*Handler) ExtractOperation 

func (h *Handler) ExtractOperation(*echo.Context) string

ExtractOperation is unused: requests never reach the shared router.

func (*Handler) ExtractResource 

func (h *Handler) ExtractResource(*echo.Context) string

ExtractResource is unused: requests never reach the shared router.

func (*Handler) GetSupportedOperations 

func (h *Handler) GetSupportedOperations() []string

GetSupportedOperations lists the supported gRPC methods.

func (*Handler) Handler 

func (h *Handler) Handler() echo.HandlerFunc

Handler returns a 404 handler; the real API is on the dedicated port.

func (*Handler) MatchPriority 

func (h *Handler) MatchPriority() int

MatchPriority is irrelevant because RouteMatcher never matches.

func (*Handler) Name 

func (h *Handler) Name() string

Name returns the service name.

func (*Handler) OperationsAPI 

func (h *Handler) OperationsAPI() longrunningpb.OperationsServer

OperationsAPI returns the long-running operations server for in-process callers.

func (*Handler) Reset 

func (h *Handler) Reset()

Reset clears all state.

func (*Handler) Restore 

func (h *Handler) Restore(ctx context.Context, data []byte) error

Restore implements persistence.Persistable by delegating to the backend.

func (*Handler) RouteMatcher 

func (h *Handler) RouteMatcher() service.Matcher

RouteMatcher never matches: Batch runs on its own listener.

func (*Handler) Shutdown 

func (h *Handler) Shutdown(ctx context.Context)

Shutdown stops the listener, forcing a close after a short grace period.

func (*Handler) Snapshot 

func (h *Handler) Snapshot(ctx context.Context) []byte

Snapshot implements persistence.Persistable by delegating to the backend.

func (*Handler) StartWorker 

func (h *Handler) StartWorker(ctx context.Context) error

StartWorker binds the dedicated listener synchronously so a port conflict is returned to the caller.

type InMemoryBackend 

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

InMemoryBackend holds all Batch state under one lock.

func NewInMemoryBackend 

func NewInMemoryBackend() *InMemoryBackend

NewInMemoryBackend creates an empty state-only backend on the wall clock.

func (*InMemoryBackend) CancelJob 

func (b *InMemoryBackend) CancelJob(in *batchpb.CancelJobRequest) (*longrunningpb.Operation, error)

CancelJob starts cancellation: the job reads CANCELLATION_IN_PROGRESS, then CANCELLED.

func (*InMemoryBackend) Close 

func (b *InMemoryBackend) Close()

Close cancels every running job and waits for the runners to stop.

func (*InMemoryBackend) CreateJob 

func (b *InMemoryBackend) CreateJob(in *batchpb.CreateJobRequest) (*batchpb.Job, error)

CreateJob validates and records a job. It returns the job without a long-running operation, as the API does.

func (*InMemoryBackend) DeleteJob 

func (b *InMemoryBackend) DeleteJob(in *batchpb.DeleteJobRequest) (*longrunningpb.Operation, error)

DeleteJob starts deletion: the job reads DELETION_IN_PROGRESS until the operation ends, then is gone.

func (*InMemoryBackend) DeleteOperation 

func (b *InMemoryBackend) DeleteOperation(name string) (*emptypb.Empty, error)

DeleteOperation removes a finished operation.

func (*InMemoryBackend) GetJob 

func (b *InMemoryBackend) GetJob(name string) (*batchpb.Job, error)

GetJob returns one job.

func (*InMemoryBackend) GetOperation 

func (b *InMemoryBackend) GetOperation(name string) (*longrunningpb.Operation, error)

GetOperation returns one operation.

func (*InMemoryBackend) GetTask 

func (b *InMemoryBackend) GetTask(name string) (*batchpb.Task, error)

GetTask returns one task.

func (*InMemoryBackend) ListJobs 

func (b *InMemoryBackend) ListJobs(in *batchpb.ListJobsRequest) (*batchpb.ListJobsResponse, error)

ListJobs lists the jobs of a location, or of every location for "-".

func (*InMemoryBackend) ListOperations 

func (b *InMemoryBackend) ListOperations(in *longrunningpb.ListOperationsRequest) (
	*longrunningpb.ListOperationsResponse, error,
)

ListOperations lists the operations of one location, ordered by name.

func (*InMemoryBackend) ListTasks 

func (b *InMemoryBackend) ListTasks(in *batchpb.ListTasksRequest) (*batchpb.ListTasksResponse, error)

ListTasks lists the tasks of a task group, optionally filtered by State=....

func (*InMemoryBackend) ProjectIDs 

func (b *InMemoryBackend) ProjectIDs() []string

ProjectIDs lists the projects that own at least one job, sorted.

func (*InMemoryBackend) Reset 

func (b *InMemoryBackend) Reset()

Reset stops every runner and clears all state.

func (*InMemoryBackend) Restore 

func (b *InMemoryBackend) Restore(ctx context.Context, data []byte) error

Restore loads a snapshot, replacing all state.

func (*InMemoryBackend) SetClock 

func (b *InMemoryBackend) SetClock(now func() time.Time)

SetClock replaces the clock lifecycle states and timestamps derive from.

func (*InMemoryBackend) SetContext 

func (b *InMemoryBackend) SetContext(ctx context.Context)

SetContext sets the parent context of job runners; cancelling it stops them.

func (*InMemoryBackend) SetRuntime 

func (b *InMemoryBackend) SetRuntime(rt Runtime, scriptImage string)

SetRuntime turns the container engine on: new jobs run their runnables as containers.

func (*InMemoryBackend) SetServices 

func (b *InMemoryBackend) SetServices(s Services)

SetServices connects Secret Manager and Cloud Logging.

func (*InMemoryBackend) SetTimer 

func (b *InMemoryBackend) SetTimer(t func(d time.Duration) (<-chan time.Time, func() bool))

SetTimer replaces the timer behind max_run_duration and runnable timeouts.

func (*InMemoryBackend) SetTiming 

func (b *InMemoryBackend) SetTiming(t Timing)

SetTiming replaces the state-only durations.

func (*InMemoryBackend) Snapshot 

func (b *InMemoryBackend) Snapshot(ctx context.Context) []byte

Snapshot serialises jobs, tasks, operations and request ids. Running containers are not kept.

func (*InMemoryBackend) WaitIdle 

func (b *InMemoryBackend) WaitIdle()

WaitIdle blocks until every job runner has finished.

func (*InMemoryBackend) WaitOperation 

func (b *InMemoryBackend) WaitOperation(ctx context.Context, in *longrunningpb.WaitOperationRequest) (
	*longrunningpb.Operation, error,
)

WaitOperation returns once the operation is done or the timeout passes, whichever comes first.

type Provider 

type Provider struct{}

Provider implements service.Provider for Batch. It serves gRPC, REST and the gateway data plane on its own listener (see Handler.StartWorker) and never matches on the shared AWS router.

func (*Provider) Init 

func (p *Provider) Init(ctx *service.AppContext) (service.Registerable, error)

Init creates the backend and handler; the port is bound in StartWorker.

func (*Provider) Name 

func (p *Provider) Name() string

Name returns the service provider name.

type Runtime 

type Runtime interface {
	CreateAndStart(ctx context.Context, spec container.Spec) (string, error)
	StopAndRemove(ctx context.Context, containerID string) error
}

Runtime is the slice of the container runtime a task needs. The runtime must also implement container.ExitWaiter; it should implement container.LogStreamer (Cloud Logging output).

type SecretResolver 

type SecretResolver interface {
	AccessSecret(versionName string) ([]byte, error)
}

SecretResolver reads a Secret Manager secret version ("projects/p/secrets/s/versions/v").

type Services 

type Services struct {
	Secrets SecretResolver
	Logs    gcpserve.LogSink
}

Services are the emulator backends a job reads from and writes to; absent ones stay nil.

type Settings 

type Settings struct {
	Engine string `` //nolint:lll // config struct tags are intentionally verbose
	/* 290-byte string literal not displayed */
	ScriptImage string `` //nolint:lll // config struct tags are intentionally verbose
	/* 234-byte string literal not displayed */
	Port int `` //nolint:lll // config struct tags are intentionally verbose
	/* 223-byte string literal not displayed */
}

Settings holds service-level configuration, embedded in the root CLI command as CLI.GCPBatch.

func DefaultSettings 

func DefaultSettings() Settings

DefaultSettings returns the settings used when no ConfigProvider is available.

type Timing 

type Timing struct {
	Schedule time.Duration
	Start    time.Duration
	TaskRun  time.Duration
	Cancel   time.Duration
	Delete   time.Duration
}

Timing sets how long a state-only job takes on the backend clock. Every task runs for TaskRun and at most parallelism tasks run at once.

func DefaultTiming 

func DefaultTiming() Timing

DefaultTiming returns the durations used unless a test overrides them.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.