Documentation
¶
Index ¶
- func InitCommands(conn adbc.Connection, c *config.Conf) error
- func InitTables(conn adbc.Connection, c *config.Conf) error
- func InitUDFs(c *config.Conf) error
- func OpenState(ctx context.Context, path string) (*duckdb.DB, error)
- type ErrorPolicy
- type Handler
- type Mark
- type MarkCommitter
- type Marks
- type Message
- type MetadataWriter
- type Metrics
- type OffsetStat
- type OffsetStore
- type PipelineErrorPolicies
- type Sink
- type Source
- type StateStats
- type Stats
- type TableStat
- type Turbine
- type TurbineOption
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func InitCommands ¶
func InitCommands(conn adbc.Connection, c *config.Conf) error
func InitTables ¶
func InitTables(conn adbc.Connection, c *config.Conf) error
InitTables creates the tables that live across the pipeline's lifetime, such as the aggregate tables a tumbling window manager maintains.
func InitUDFs ¶
InitUDFs reports that a config declares UDFs, which turbine does not support. The Python engine registers arbitrary Python callables through conn.create_function; Go has no equivalent, and rather than inventing a plugin mechanism, user-defined functions are left to DuckDB itself (a macro, an extension, or an ATTACHed database that defines them).
This is an error rather than a silent skip: ignoring the block would defer the failure to an unhelpful binder error when the handler SQL calls the function.
func OpenState ¶ added in v1.0.5
OpenState opens the DuckDB database backing a pipeline's durable state.
It exists to separate two cases that duckdb.OpenPath cannot tell apart. A path with no file is the first run, and creating it is correct. A path that holds something DuckDB refuses to open is a damaged state file, and starting over would replay from the beginning while reporting healthy.
A damaged file is never repaired, moved, or truncated here. It holds the only copy of the pipeline's positions, so it is evidence an operator needs.
Types ¶
type ErrorPolicy ¶
type ErrorPolicy int
const ( PolicyRaise ErrorPolicy = iota PolicyIgnore PolicyDLQ )
func ParseErrorPolicy ¶
func ParseErrorPolicy(s string) (ErrorPolicy, error)
ParseErrorPolicy resolves the configured policy name. Matching the Python engine, the name is case-insensitive and an empty value means RAISE.
type Mark ¶ added in v1.0.4
Mark is the position of the last message the pipeline has finished with in one partition: written to the handler, or dropped by an error policy.
type MarkCommitter ¶ added in v1.0.4
MarkCommitter is implemented by sources that can commit an explicit position. The pipeline prefers it to Commit, because a source that reads ahead of the pipeline -- the Kafka source polls into a buffer -- has fetched messages the pipeline has not processed, and committing "everything fetched" commits those too.
type Marks ¶ added in v1.0.5
type Marks struct {
// contains filtered or unexported fields
}
Marks records the last position the pipeline has finished with in each partition it has read: a message written to the handler, or one dropped by an error policy.
Topic-then-partition nesting is the natural shape for this and the awkward one to pass around. Owning it here keeps two rules in one place rather than repeated at every call site: a partition only ever moves forward, and iteration is ordered so anything rendered from it -- the stats endpoint, the CLI -- is stable between runs.
func (*Marks) Advance ¶ added in v1.0.5
Advance records a position, ignoring one that would move a partition backwards. Offsets arrive in order within a partition, so the guard only matters if a source redelivers -- and a redelivery must not rewind a position that has already been committed.
func (*Marks) Each ¶ added in v1.0.5
Each calls fn for every position, ordered by topic then partition. The order is deliberate: map iteration is random, and callers render these into JSON and log lines where a shuffling order makes diffs unreadable.
type Message ¶
type Message struct {
Value []byte
Topic string
Partition int32
Offset int64
// LeaderEpoch is the Kafka leader epoch the record was read under. It is
// carried through so a commit can name it, which lets the broker detect
// log truncation. Only meaningful when HasMetadata is true; a source with
// no positions leaves it zero along with the rest.
LeaderEpoch int32
// HighWatermark is the partition's high watermark at fetch time, so lag
// can be computed against the position last processed. Zero for sources
// without one.
HighWatermark int64
}
Message is one record from a source, with whatever provenance the source knows about it. Only Kafka populates the metadata fields.
func (Message) HasMetadata ¶
HasMetadata reports whether the source supplied provenance for this message. Topic is the discriminator: a Kafka record always has one, and a source that has none leaves it empty.
type MetadataWriter ¶
MetadataWriter is implemented by handlers that can use a message's source metadata. Handlers that only need the payload implement Handler alone and the consume loop hands them the value.
type Metrics ¶
type Metrics struct {
MessageCount metric.Int64Counter
ErrorCount metric.Int64Counter
SourceReadLatency metric.Float64Histogram
SinkFlushLatency metric.Float64Histogram
SinkFlushNumRows metric.Int64Gauge
SinkFlushCount metric.Int64Counter
BatchProcessingLatency metric.Float64Histogram
StateCommitLatency metric.Float64Histogram
StateCommitCount metric.Int64Counter
StateSizeBytes metric.Int64Gauge
StateTableRows metric.Int64Gauge
ConsumerLag metric.Int64Gauge
}
Metrics holds the instruments the pipeline records, mirroring the names, descriptions and units the Python engine exports.
func NewMetrics ¶
func NewMetrics(mp metric.MeterProvider) (*Metrics, error)
NewMetrics builds the instruments from a meter provider. Passing a noop provider yields instruments that record nothing, so the pipeline needs no nil checks.
type OffsetStat ¶ added in v1.0.5
type OffsetStat struct {
Topic string `json:"topic"`
Partition int32 `json:"partition"`
Offset int64 `json:"offset"`
LeaderEpoch int32 `json:"leader_epoch"`
}
OffsetStat is one partition's durable position.
type OffsetStore ¶ added in v1.0.5
type OffsetStore struct {
// contains filtered or unexported fields
}
OffsetStore persists Marks to DuckDB. It holds no lock and starts no transaction of its own: Save issues writes on the connection it is given, and it is the caller's job to commit -- typically together with the batch that advanced those offsets.
func NewOffsetStore ¶ added in v1.0.5
func NewOffsetStore(conn adbc.Connection) *OffsetStore
NewOffsetStore wraps a DuckDB connection. The connection is expected to be the same one the pipeline uses for its batch writes, so offsets land in the same transaction as the state they describe.
func (*OffsetStore) Init ¶ added in v1.0.5
func (s *OffsetStore) Init(ctx context.Context) error
Init prepares the offsets table, and refuses to run against a state file it cannot read.
The table is created only when it is absent, which is the first run. When it is present, its schema has to match offsetsSchema exactly. A table that carries the right name and the wrong shape means the file was written by something else, or damaged; either way the positions in it cannot be trusted.
The alternative -- CREATE TABLE IF NOT EXISTS, as this did before -- accepts the damaged table, finds no readable positions, and replays the topic from the beginning while reporting healthy. A silent restart from zero is worse than a refusal to start, so this returns CodeStateCorrupt and changes nothing on disk.
func (*OffsetStore) Load ¶ added in v1.0.5
func (s *OffsetStore) Load(ctx context.Context) (*Marks, error)
Load returns every position recorded in the offsets table. A store with no rows yet returns an empty, non-nil Marks -- the caller must not treat that as offset zero.
func (*OffsetStore) Save ¶ added in v1.0.5
func (s *OffsetStore) Save(ctx context.Context, marks *Marks) error
Save upserts one row per topic/partition in marks. It does not commit: the caller owns the transaction, so these writes can land together with whatever state change they are recording the position for.
type PipelineErrorPolicies ¶
type PipelineErrorPolicies struct {
Policy ErrorPolicy
DLQSink Sink
}
type Sink ¶
type Sink interface {
WriteTable(ctx context.Context, batch arrow.Table) error
Flush(ctx context.Context) error
Batch() (arrow.Table, error)
}
Sink is where a batch leaves the pipeline.
Both write paths take a context so a sink that retries can be interrupted. Without it, a retry ladder outlives the SIGTERM that asked the pipeline to stop, and the graceful drain waits out the full retry deadline before it can finish. See #110.
type StateStats ¶ added in v1.0.5
type StateStats struct {
Path string `json:"path"`
SizeBytes int64 `json:"size_bytes"`
Tables []TableStat `json:"tables"`
Offsets []OffsetStat `json:"offsets"`
}
StateStats is a snapshot of a pipeline's durable state: how large it is, what is in it, and how far through the stream it has committed.
It is produced once and rendered by two consumers -- the /stats endpoint of a running pipeline and the CLI reading a stopped one -- so the two can never disagree about what a pipeline's state contains.
func CollectStateStats ¶ added in v1.0.5
func CollectStateStats(ctx context.Context, conn adbc.Connection, path string) (*StateStats, error)
CollectStateStats reads a snapshot of the state database.
The connection should be one dedicated to reading, not the connection the pipeline writes on. Connections to a DuckDB database have independent transaction state, so a reader sees committed rows only: the numbers here describe what would survive a crash at this moment, and collecting them cannot block the writer or be blocked by it.
type Stats ¶
func (*Stats) AddMessagesConsumed ¶
func (*Stats) GetThroughput ¶
func (*Stats) MessagesConsumed ¶
func (*Stats) SetNumMessagesConsumed ¶
func (*Stats) SetThroughput ¶
type Turbine ¶
type Turbine struct {
// contains filtered or unexported fields
}
func NewTurbine ¶
func (*Turbine) ConsumeLoop ¶
func (*Turbine) SyncState ¶ added in v1.0.5
SyncState closes the open state transaction, making everything written since the last commit durable. A pipeline with no state database does nothing.
Shutdown uses it twice: once before the table managers run their final poll, so that poll sees a current clock rather than one frozen at the last batch, and once after, so the rows that poll published are actually deleted instead of being rolled back when the connection closes and republished on the next start.
type TurbineOption ¶
type TurbineOption func(turbine *Turbine)
func WithMetrics ¶
func WithMetrics(m *Metrics) TurbineOption
WithMetrics records pipeline instruments through the given provider.
func WithStateStats ¶ added in v1.0.5
func WithStateStats(fn func() (*StateStats, error)) TurbineOption
WithStateStats supplies the snapshot function backing the state gauges. It must read a connection dedicated to reading; passing the pipeline's writer would let a scrape contend with batch processing.
func WithStateStore ¶ added in v1.0.5
func WithStateStore(offsets offsetSaver, tx stateTx) TurbineOption
WithStateStore makes each batch transactional: the handler's writes to the state database and the offsets that produced them commit together, so a crash can never leave one without the other.
func WithTurbineLogger ¶
func WithTurbineLogger(l *zap.Logger) TurbineOption
Source Files