openlineage

package
v0.1.1 Latest Latest
Warning

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

 Go to latest Published: Oct 13, 2026 License: Apache-2.0 Imports: 16 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	// MaxEventSize bounds one event rather than one request: a batch body may
	// be several megabytes as long as each event in it fits here.
	MaxEventSize = 1 << 20
	// The identity caps keep the unique btree indexes of openlineage_run and
	// openlineage_task below PostgreSQL's index item size. A value longer than
	// the cap used to be stored and then fail the insert forever, so the event
	// never became retryable; rejecting it up front makes the failure the
	// producer's to fix.
	MaxJobNamespaceLength = 255
	MaxJobNameLength      = 255
	MaxJobTypeLength      = 64
	MaxRunIDLength        = 255
	// Dataset names are longer than job names by nature: they can carry a
	// database, a schema and a table, and BigQuery prefixes a project.
	MaxDatasetNamespaceLength = 255
	MaxDatasetNameLength      = 512
	// MaxEventDatasets bounds the references one event may declare, and with
	// them the ingestion work and stored rows one run costs.
	MaxEventDatasets = 1000
	// MaxGUIDLength bounds the escaped GUID a run or task is indexed under.
	// PostgreSQL's btree item limit is about 2704 bytes, and the per-field caps
	// above can approach it once every character needs percent-encoding.
	MaxGUIDLength = 2000
	// The remaining strings an event writes into the run row. The run and task
	// lists read them for every row of a page, so an uncapped one would let one
	// page read an unbounded number of bytes even though raw_payload is not
	// selected, and these columns are what a producer can put in a single event.
	MaxEventTypeLength          = 64
	MaxProducerLength           = 512
	MaxIntegrationLength        = 64
	MaxProcessingTypeLength     = 64
	MaxParentJobNamespaceLength = 255
	MaxParentJobNameLength      = 255
	MaxParentRunIDLength        = 255
	MaxRootJobNamespaceLength   = 255
	MaxRootJobNameLength        = 255
	MaxRootRunIDLength          = 255
	// MaxColumnNameLength bounds the columnLineage strings. They reach the
	// indexed source_column/target_column of column_lineage, and a dataset a
	// facet names without listing it as an input or output builds a GUID from
	// its name, so both are capped like the identity fields.
	MaxColumnNameLength = 512
	// MaxAirflowRunLogURLLength bounds the link stored on the run. It is derived
	// from the event with url.String(), which percent-encodes, so the stored link
	// can be longer than the facet it came from; past this the link is dropped
	// rather than stored and read back for every row of a page.
	MaxAirflowRunLogURLLength = 2048
)

The per-event caps. A request body limit alone leaves a single event free to be as large as the body, and every list view that later reads the run would pay for it; these keep one event, and the identity it is indexed under, small.

View Source 
const ExternalDatasetGUIDPrefix = "external:"

ExternalDatasetGUIDPrefix is the prefix for external dataset GUIDs.

Variables

View Source 
var ErrEventTooLarge = errors.New("openlineage event exceeds the size limit")

ErrEventTooLarge reports an event beyond MaxEventSize. The ingestion handler answers it with 413 rather than the 400 a malformed event earns.

Functions

func BuildOpenLineageRunGUID 

func BuildOpenLineageRunGUID(namespace, name, jobType, runID string) string

BuildOpenLineageRunGUID returns the stable GUID for a persisted OpenLineage run.

func BuildOpenLineageTaskGUID 

func BuildOpenLineageTaskGUID(namespace, name, jobType string) string

BuildOpenLineageTaskGUID returns the stable GUID for an aggregated OpenLineage task/job.

func FormatExternalGUID 

func FormatExternalGUID(namespace, name string) string

FormatExternalGUID creates an external dataset GUID.

func InferDatasetType 

func InferDatasetType(namespace string) string

InferDatasetType guesses the dataset type from the namespace scheme.

func IsExternalGUID 

func IsExternalGUID(guid string) bool

IsExternalGUID returns true if the GUID belongs to an external dataset.

func ValidateEventLimits 

func ValidateEventLimits(event *RunEvent) error

ValidateEventLimits enforces the per-event caps on a parsed event. RawJSON is set by ParseRunEvent, which is the only supported way to build one.

Types 

type AirflowLinks struct {
	DagURL    string
	RunLogURL string
}

AirflowLinks contains Airflow UI URLs derived from an OpenLineage raw payload.

func AirflowLinksFromRunLogURL 

func AirflowLinksFromRunLogURL(runLogURL string) AirflowLinks

AirflowLinksFromRunLogURL applies the http(s) whitelist and the DAG-URL derivation to a run log URL. Ingestion stores the whitelisted link on the run row, so a reader rebuilds the same links from the column without holding the payload the facet came from. 

func DeriveAirflowLinks(rawPayload []byte) AirflowLinks

DeriveAirflowLinks extracts Airflow web URLs from the original OpenLineage payload.

type ColumnLineageDatasetReference 

type ColumnLineageDatasetReference struct {
	Namespace       string        `json:"namespace"`
	Name            string        `json:"name"`
	Field           string        `json:"field"`
	Transformations []OLTransform `json:"transformations,omitempty"`
}

ColumnLineageDatasetReference holds dataset-level lineage info.

type ColumnLineageFacet 

type ColumnLineageFacet struct {
	Producer  string                          `json:"_producer,omitempty"`
	SchemaURL string                          `json:"_schemaURL,omitempty"`
	Fields    map[string]ColumnLineageField   `json:"fields"`
	Dataset   []ColumnLineageDatasetReference `json:"dataset,omitempty"`
}

ColumnLineageFacet represents the OpenLineage ColumnLineageDatasetFacet.

type ColumnLineageField 

type ColumnLineageField struct {
	InputFields []InputField `json:"inputFields"`
}

ColumnLineageField represents column-level lineage for a single output column.

type Dataset 

type Dataset struct {
	Namespace string        `json:"namespace"`
	Name      string        `json:"name"`
	Facets    DatasetFacets `json:"facets,omitempty"`
}

Dataset represents an input or output dataset in an OpenLineage event.

type DatasetFacets 

type DatasetFacets struct {
	Schema        *SchemaFacet        `json:"schema,omitempty"`
	ColumnLineage *ColumnLineageFacet `json:"columnLineage,omitempty"`

	// Additional (non-standard) facets stored as raw JSON.
	Additional map[string]json.RawMessage `json:"-"`
}

DatasetFacets holds the standard facets attached to a dataset.

type DerivedRunMetadata 

type DerivedRunMetadata struct {
	TaskGUID           string
	JobType            string
	Integration        string
	ProcessingType     string
	ParentJobNamespace string
	ParentJobName      string
	ParentRunID        string
	RootJobNamespace   string
	RootJobName        string
	RootRunID          string
	HasLineage         bool
}

DerivedRunMetadata is the normalized OpenLineage metadata extracted from raw facets.

func DeriveRunMetadata 

func DeriveRunMetadata(event *RunEvent) DerivedRunMetadata

DeriveRunMetadata extracts normalized task/job information from an OpenLineage event.

type InputField 

type InputField struct {
	Namespace       string        `json:"namespace"`
	Name            string        `json:"name"`
	Field           string        `json:"field"`
	Transformations []OLTransform `json:"transformations,omitempty"`
}

InputField identifies an input column contributing to an output column.

type Job 

type Job struct {
	Namespace string                     `json:"namespace"`
	Name      string                     `json:"name"`
	Facets    map[string]json.RawMessage `json:"facets,omitempty"`
}

Job represents the job entity in an OpenLineage event.

type OLTransform 

type OLTransform struct {
	Type        string `json:"type"`
	Subtype     string `json:"subtype"`
	Description string `json:"description,omitempty"`
	Masking     bool   `json:"masking,omitempty"`
}

OLTransform describes a transformation applied in OpenLineage.

type Processor 

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

Processor handles incoming OpenLineage events and stores derived lineage.

func NewProcessor 

func NewProcessor(s *store.Store, lineageAnalyzer *lineage.Analyzer, reporter UnmatchedNamespaceReporter) *Processor

NewProcessor creates a new event processor. reporter is told about datasets that match no registered instance; it may be nil.

func (*Processor) ProcessRunEvent 

func (p *Processor) ProcessRunEvent(ctx context.Context, event *RunEvent, persistedRun *store.OpenLineageRunMessage) error

ProcessRunEvent derives lineage from a single OpenLineage RunEvent. The run is already persisted whatever its state; only a COMPLETE event yields lineage, because that is where the producer states what the run finally read and wrote.

func (*Processor) RevalidateContradictedColumnClaims 

func (p *Processor) RevalidateContradictedColumnClaims(ctx context.Context, limit int) (revalidated, stillUnknown int, err error)

RevalidateContradictedColumnClaims re-checks the ingested edges whose column claim the registry has come to contradict: the edge names a column on a relation that is a TABLE, and the table has no such column.

These are the edges whose relation was unknown when they were stored, so ingestion had to keep the claim unchecked. The sync that made the relation known also removed the edge from the unknown-endpoint sweep's reach, so this is the only pass that can still fix them. A claim is blanked, the edge stays as a table-level dependency, and what is still unknown around it is counted.

func (*Processor) RevalidateUnresolvedLineage 

func (p *Processor) RevalidateUnresolvedLineage(ctx context.Context, limit int) (revalidated, stillUnknown int, err error)

RevalidateUnresolvedLineage re-checks the edges whose endpoints the registry did not know when they were stored.

An ingested edge is kept when its relation is missing, because ingestion runs before the next schema sync and dropping it would lose the lineage of a table created minutes ago. That keeps real lineage, but it also means the edge's columns were never checked and the edge may name a relation that never appears. Once the relation does appear - or once a second look proves it is still absent - the same validation the ingestion path runs decides what the edge should say now: a column the relation does not have is blanked, and the object types are filled in. Edges whose relation is still unknown stay, and are counted so the backlog is visible instead of silent.

It returns the number of edges re-checked and the number of meta objects that still reference an unknown relation.

type ResolutionStore 

type ResolutionStore interface {
	GetNamespaceMapping(ctx context.Context, find *store.FindNamespaceMappingMessage) (*store.NamespaceMappingMessage, error)
	GetInstance(ctx context.Context, find *store.FindInstanceMessage) (*store.InstanceMessage, error)
	ListInstances(ctx context.Context, find *store.FindInstanceMessage) ([]*store.InstanceMessage, error)
	GetOrCreateExternalDataset(ctx context.Context, namespace, name, datasetType string) (*store.ExternalDatasetMessage, error)
}

ResolutionStore is the store surface the resolver reads: the namespace mapping, the instances behind it, and the external dataset it creates for a namespace that names no instance. It is the seam that lets the resolution rules and their per-request memoization be tested without a database.

type ResolvedDataset 

type ResolvedDataset struct {
	GUID     string
	MetaType storepb.MetaType
	// True if this dataset was resolved to an internal instance.
	Internal bool
}

ResolvedDataset holds the result of resolving an OpenLineage dataset to an internal or external entity.

type Resolver 

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

Resolver maps OpenLineage dataset namespaces and names to internal GUIDs or external datasets.

func NewRequestScopedResolver 

func NewRequestScopedResolver(s ResolutionStore) *Resolver

NewRequestScopedResolver creates a Resolver that memoizes lookups for the lifetime of one request. Use NewResolver for ingestion, where a resolver may serve many resolutions and cached answers could go stale.

func NewResolver 

func NewResolver(s ResolutionStore, reporter UnmatchedNamespaceReporter) *Resolver

NewResolver creates a new Resolver. reporter may be nil.

func (*Resolver) ResolveDataset 

func (r *Resolver) ResolveDataset(ctx context.Context, namespace, datasetName string) (*ResolvedDataset, error)

ResolveDataset resolves an OpenLineage dataset (namespace + name) to a GUID and MetaType. Resolution order:

  1. Manual namespace_mapping table lookup
  2. Auto-match by parsing namespace URL and matching instance DataSource host:port
  3. Create/return an external dataset

func (*Resolver) ResolveDatasetPreview 

func (r *Resolver) ResolveDatasetPreview(ctx context.Context, namespace, datasetName string) (*ResolvedDataset, error)

ResolveDatasetPreview resolves an OpenLineage dataset without creating external-dataset rows.

type Run 

type Run struct {
	RunID  string                     `json:"runId"`
	Facets map[string]json.RawMessage `json:"facets,omitempty"`
}

Run represents the run entity in an OpenLineage event.

type RunEvent 

type RunEvent struct {
	EventType string          `json:"eventType"`
	EventTime string          `json:"eventTime"`
	Run       Run             `json:"run"`
	Job       Job             `json:"job"`
	Inputs    []Dataset       `json:"inputs"`
	Outputs   []Dataset       `json:"outputs"`
	Producer  string          `json:"producer"`
	RawJSON   json.RawMessage `json:"-"`
}

RunEvent represents an OpenLineage RunEvent. See: https://openlineage.io/docs/spec/run-event

func ParseRunEvent 

func ParseRunEvent(data []byte) (*RunEvent, error)

ParseRunEvent parses a raw JSON payload into a RunEvent.

type SchemaFacet 

type SchemaFacet struct {
	Producer  string        `json:"_producer,omitempty"`
	SchemaURL string        `json:"_schemaURL,omitempty"`
	Fields    []SchemaField `json:"fields"`
}

SchemaFacet represents the OpenLineage SchemaDatasetFacet.

type SchemaField 

type SchemaField struct {
	Name        string `json:"name"`
	Type        string `json:"type"`
	Description string `json:"description,omitempty"`
}

SchemaField represents a field in the schema facet.

type UnmatchedNamespaceReporter 

type UnmatchedNamespaceReporter interface {
	ReportUnmatchedNamespace(ctx context.Context, namespace, dataset string)
}

UnmatchedNamespaceReporter is told about a dataset that matched no registered instance and was stored as an external dataset, so the lineage naming it points at something the registry does not describe. Ingestion wires the workspace notification service here; a nil reporter leaves the log as the record.

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.