Pipelines

Goldsky Turbo Pipelines ingest data, transform it, and write it to a configured sink. The Go SDK manages the lifecycle and observability of a pipeline; the flexible source, transform, and sink authoring schema is carried in map[string]any because those product-specific objects evolve independently of the SDK.

Recommended workflow: discover the current authoring contract in Goldsky’s pipeline documentation, validate the definition, create the pipeline, observe status and logs, and only then automate lifecycle changes.

Core types

Type	Purpose	Important fields
PipelineDefinition	Flexible authoring payload	Sources, Transforms, Sinks, optional Description, ResourceSize, Job.
CreatePipelineRequest	Creates a persistent pipeline	Name, Definition, optional resource and dedicated-IP settings.
ValidatePipelineRequest	Validates a definition without creating	Same authoring inputs as creation.
Pipeline	Returned managed resource	Name, Status, Definition, timestamps, version, project ID.
PipelineStatusResponse	Focused operational status	Name, Status, and any server-reported errors.

Known pipeline states include RUNNING, PAUSED, RESTARTING, DEPLOYING, STOPPED, FAILED, SUCCEEDED, and UNKNOWN. The underlying type remains a string, so a future Goldsky state will decode safely even before the SDK adds a named constant.

List and inspect pipelines

List retrieves one cursor page. Pipeline names are locally checked against ^[a-z0-9-]{1,50}$ when the SDK receives a name.

page, err := client.Pipelines.List(ctx, goldsky.ListPipelinesOptions{
    Type:     "turbo",
    PageSize: 50,
})
if err != nil {
    return err
}

for _, pipeline := range page.Data {
    fmt.Printf("%s: %s\n", pipeline.Name, pipeline.Status)
}

pipeline, err := client.Pipelines.Get(ctx, "dex-trades")
if err != nil {
    return err
}
fmt.Println(pipeline.Definition.Description)

Use NewPipelinePager to consume all pages. See Errors, Retries, and Pagination for why a short page is not a completion signal.

Validate before creating

Validation is the safest way to catch invalid authoring input before a resource is created. The exact nested schema for sources, transforms, and sinks belongs to Goldsky’s current pipeline authoring documentation.1

definition := goldsky.PipelineDefinition{
    Sources: map[string]any{
        "trades": map[string]any{
            "type": "dataset",
            "dataset_name": "example-dataset",
        },
    },
    Transforms: map[string]any{},
    Sinks: map[string]any{
        "discard": map[string]any{"type": "blackhole"},
    },
    Description: "Validation-only sample pipeline",
}

validation, err := client.Pipelines.Validate(ctx, goldsky.ValidatePipelineRequest{
    Name:       "dex-trades",
    Definition: definition,
})
if err != nil {
    return err
}
if !validation.Valid {
    for _, finding := range validation.Errors {
        fmt.Printf("%s: %s\n", finding.Field, finding.Message)
    }
    return fmt.Errorf("pipeline definition is invalid")
}
for _, warning := range validation.Warnings {
    log.Printf("pipeline warning at %s: %s", warning.Field, warning.Message)
}

A valid response does not create a resource. Treat warnings as design review input, especially before enabling a production sink.

Create a pipeline

Creation is a REST mutation. It is not automatically retried unless you explicitly enable mutation retries, because Goldsky does not document idempotency keys. If the network fails after send, inspect the pipeline by name before retrying.

pipeline, err := client.Pipelines.Create(ctx, goldsky.CreatePipelineRequest{
    Name:         "dex-trades",
    Description:  "Streams example dataset events to a managed sink",
    ResourceSize: "small",
    Definition:   definition,
})
if err != nil {
    return err
}
fmt.Printf("created %s with status %s\n", pipeline.Name, pipeline.Status)

Name can also appear inside PipelineDefinition, but the top-level CreatePipelineRequest.Name takes precedence. Name the resource explicitly at the request level to reduce ambiguity.

Preview a definition

Preview provides a short-lived preview. Its optional TTLSeconds is validated locally between 1 and 600 seconds.

preview, err := client.Pipelines.Preview(ctx, goldsky.PreviewPipelineRequest{
    Definition: definition,
    TTLSeconds: json.Number("120"),
})
if err != nil {
    return err
}
fmt.Printf("preview %s expires at %s\n", preview.PipelineName, preview.ExpiresAt)

A preview is not a substitute for a production change-management process. Confirm what it does not model, such as production credentials or historical volume, in Goldsky’s live authoring documentation.

Lifecycle operations

Method	Effect	Operational guidance
Pause(ctx, name)	Stops a pipeline from running.	Check downstream freshness expectations before pausing.
Resume(ctx, name)	Resumes a paused pipeline.	Check status and logs after resumption.
Restart(ctx, name, req)	Restarts a pipeline.	Use ClearState only when you understand replay implications.
Delete(ctx, name)	Deletes a pipeline.	Verify ownership and impact; this is destructive.

if err := client.Pipelines.Pause(ctx, "dex-trades"); err != nil {
    return err
}

if err := client.Pipelines.Restart(ctx, "dex-trades", &goldsky.RestartPipelineRequest{
    ClearState: false,
}); err != nil {
    return err
}

RestartPipelineRequest.ClearState defaults to false. Clearing state can alter how a pipeline recovers or reprocesses data, so make the choice explicit in automated tooling.

Observe status, logs, errors, and state

Use a focused status call for an operator view, logs for diagnosis, and state only when your application understands the returned raw JSON schema.

status, err := client.Pipelines.Status(ctx, "dex-trades")
if err != nil {
    return err
}
for _, item := range status.Errors {
    log.Printf("pipeline error: %s", item.Message)
}

logs, err := client.Pipelines.Logs(ctx, "dex-trades", goldsky.PipelineLogsOptions{
    LogLevels: "ERROR,WARN",
    Direction: "desc",
    Search:    "sink",
})
if err != nil {
    return err
}
for _, record := range logs.Data.Results {
    fmt.Printf("%s %s %s\n", record.Timestamp, record.Level, record.Text)
}

ErrorCount(ctx, name, sinceHours) accepts an optional period of 1 through 168 hours. State(ctx, name) returns json.RawMessage inside PipelineStateResponse.Data, because Goldsky leaves the state schema open; decode it into a private type only after you have confirmed the shape you expect.

Production checklist

Before production	Why it matters
Validate the same definition you plan to create.	Detects authoring failures without resource creation.
Use an explicit pipeline name matching the SDK’s pattern.	Enables deterministic reconciliation after a timeout.
Set a context deadline for create and lifecycle calls.	Avoids jobs that wait indefinitely.
Monitor Status, Logs, and ErrorCount.	Separates lifecycle state from runtime failures.
Treat ClearState and Delete as deliberate changes.	Both can affect data continuity.
Preserve the complete Goldsky problem response in internal diagnostics with care.	Field-level validation detail can accelerate remediation.

References