Documentation
¶
Overview ¶
Package sdk is the harness for SDK (Go) jobs run through the mailer control plane. A user writes a pipeline builder and calls Run:
func main() { sdk.Run(Build) }
func Build(env *mailer.StreamExecutionEnv) {
env.FromSource(src).KeyBy(key).Reduce(sum).ToSink(out)
}
The harness supervises the pipeline exactly like the YAML runner: it serves the control surface (/state, /metrics, /cancel, /savepoint) for the dashboard, drains gracefully on SIGTERM, and — when checkpointing is enabled (CHECKPOINT_INTERVAL) — supports savepoints and recovery. So an SDK job is managed identically to a YAML job.
Configuration (environment):
DATA_DIR state/checkpoint root (default /data) PORT control port (default 8080) SAVEPOINT_DIR savepoint blobstore (default /savepoints) CHECKPOINT_INTERVAL enable durable checkpointing, e.g. 5s (default off) RESTORE_SAVEPOINT savepoint label to resume from (default none) JOB_NAME human-readable name for logs
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func Run ¶
func Run(build Builder)
Run is the SDK job entrypoint. It configures the environment from env vars, invokes build to wire the pipeline, and supervises it to completion, exiting the process with an appropriate status code.
func Serve ¶
func Serve(ctx context.Context, env *mailer.StreamExecutionEnv, opts ServeOptions) int
Serve supervises a configured env under a jobagent: it restores a savepoint if asked, serves the control surface, runs to completion, and promotes a savepoint on a stop-with-savepoint request. It returns a process exit code. This is the shared lifecycle behind both the YAML runner and SDK jobs, so they behave identically.
Types ¶
type Builder ¶
type Builder func(env *mailer.StreamExecutionEnv)
Builder wires a pipeline onto env: env.FromSource(...)....ToSink(...).
type ServeOptions ¶
type ServeOptions struct {
Name string // for log lines
Port string // control port; default 8080
CheckpointDir string // "" when checkpointing is disabled
SavepointDir string // savepoint blobstore; default /savepoints
RestoreSavepoint string // savepoint label to seed from, or ""
Stdout, Stderr io.Writer
}
ServeOptions configures Serve.
Source Files