Documentation
¶
Index ¶
- Constants
- Variables
- func ConnectGocql(ctx context.Context, cluster *gocql.ClusterConfig) (*gocql.Session, error)
- type MetricsSource
- func (s *MetricsSource) DiscoverColumns(ctx context.Context, table string) ([]hermod.ColumnInfo, error)
- func (s *MetricsSource) DiscoverDatabases(ctx context.Context) ([]string, error)
- func (s *MetricsSource) DiscoverPublications(ctx context.Context) ([]hermod.PublicationInfo, error)
- func (s *MetricsSource) DiscoverReplicationSlots(ctx context.Context) ([]hermod.ReplicationSlotInfo, error)
- func (s *MetricsSource) DiscoverTables(ctx context.Context) ([]string, error)
- func (s *MetricsSource) ExecuteSQL(ctx context.Context, query string) ([]map[string]any, error)
- func (s *MetricsSource) GetLag(ctx context.Context) (uint64, error)
- func (s *MetricsSource) GetState() map[string]string
- func (s *MetricsSource) IsReady(ctx context.Context) error
- func (s *MetricsSource) LastStreamActivity() time.Time
- func (s *MetricsSource) PendingWork() (pending bool, known bool)
- func (s *MetricsSource) Read(ctx context.Context) (hermod.Message, error)
- func (s *MetricsSource) Sample(ctx context.Context, table string) (hermod.Message, error)
- func (s *MetricsSource) SetLogger(logger hermod.Logger)
- func (s *MetricsSource) SetState(state map[string]string)
- func (s *MetricsSource) Snapshot(ctx context.Context, tables ...string) error
- func (s *MetricsSource) StreamSilenceThreshold() time.Duration
- type PathRegistry
- type PollingSource
Constants ¶
const DefaultSourceBuffer = 64
DefaultSourceBuffer is the capacity used for the per-source message channel that connectors place decoded change events onto.
It is intentionally small: a shallow buffer absorbs short bursts while applying backpressure to the producer (the upstream database/stream reader) as soon as the downstream sink lags, instead of hoarding up to a thousand fully-decoded messages in RAM per active pipeline. With multi-KB payloads and several concurrent pipelines, deep buffers were a primary driver of the resident set; bounding them keeps Hermod within its lightweight footprint.
Variables ¶
var ( ErrPathNotRegistered = errors.New("no source registered for path") ErrPathBufferFull = errors.New("source buffer full for path") )
ErrPathNotRegistered and ErrPathBufferFull are what Dispatch reports. Each push source words its own message around them, because those messages are what its callers already read.
Functions ¶
func ConnectGocql ¶
ConnectGocql establishes a gocql session without blocking past the caller's context.
gocql exposes no context-aware CreateSession: it blocks for its own ConnectTimeout (and retries across the host list) regardless of what the caller asked for. Against an unreachable cluster that is tens of seconds. A readiness probe with a two-second budget therefore does not time out — it hangs, holding a goroutine, and reports nothing either way.
This runs the connect off the caller's goroutine and returns as soon as either the session is ready or ctx is done.
The spawned goroutine is not leaked. The result channel is buffered, so a late CreateSession still completes rather than blocking forever on send, and the abandoned session is closed so an established connection is not stranded.
Types ¶
type MetricsSource ¶
MetricsSource wraps a Source and records metrics.
func NewMetricsSource ¶
func (*MetricsSource) DiscoverColumns ¶
func (s *MetricsSource) DiscoverColumns(ctx context.Context, table string) ([]hermod.ColumnInfo, error)
func (*MetricsSource) DiscoverDatabases ¶
func (s *MetricsSource) DiscoverDatabases(ctx context.Context) ([]string, error)
func (*MetricsSource) DiscoverPublications ¶
func (s *MetricsSource) DiscoverPublications(ctx context.Context) ([]hermod.PublicationInfo, error)
func (*MetricsSource) DiscoverReplicationSlots ¶
func (s *MetricsSource) DiscoverReplicationSlots(ctx context.Context) ([]hermod.ReplicationSlotInfo, error)
func (*MetricsSource) DiscoverTables ¶
func (s *MetricsSource) DiscoverTables(ctx context.Context) ([]string, error)
func (*MetricsSource) ExecuteSQL ¶
func (*MetricsSource) GetLag ¶
func (s *MetricsSource) GetLag(ctx context.Context) (uint64, error)
GetLag forwards lag reporting to the wrapped source.
MetricsSource embeds the hermod.Source interface, and embedding an interface does not promote methods the interface does not declare — so wrapping a PostgresSource silently hid its GetLag. Every lag-based check downstream (workflow status, WAL-retention alerting, the stall watchdog) therefore read zero and concluded there was no outstanding work, which is exactly wrong when a replication slot is retaining WAL because the pipeline has stopped.
func (*MetricsSource) GetState ¶
func (s *MetricsSource) GetState() map[string]string
func (*MetricsSource) LastStreamActivity ¶
func (s *MetricsSource) LastStreamActivity() time.Time
LastStreamActivity forwards stream liveness from the wrapped source, for the same reason GetLag has to be forwarded: an embedded interface does not promote methods it does not declare, and a hidden liveness signal reads as "this source has no stream to watch" rather than as "this source's stream is fine".
func (*MetricsSource) PendingWork ¶
func (s *MetricsSource) PendingWork() (pending bool, known bool)
PendingWork forwards the wrapped source's outstanding-work signal. A source that cannot answer reports false, which leaves the stall watchdog on its lag fallback rather than asserting there is nothing outstanding.
func (*MetricsSource) SetLogger ¶
func (s *MetricsSource) SetLogger(logger hermod.Logger)
func (*MetricsSource) SetState ¶
func (s *MetricsSource) SetState(state map[string]string)
func (*MetricsSource) Snapshot ¶
func (s *MetricsSource) Snapshot(ctx context.Context, tables ...string) error
func (*MetricsSource) StreamSilenceThreshold ¶
func (s *MetricsSource) StreamSilenceThreshold() time.Duration
StreamSilenceThreshold forwards the wrapped source's silence deadline. Zero for a source that does not hold a push stream, which disables the check.
type PathRegistry ¶ added in v1.17.0
type PathRegistry struct {
// contains filtered or unexported fields
}
PathRegistry hands pushed messages to the source that holds a path. The gRPC, webhook, GraphQL and form sources each keep one: a request arrives for a path, and the source registered under it receives.
A path has one holder. A source built while the path is free holds it at once. A source built while another holds it waits, and takes it over on its first Read — or when the holder closes, whichever comes first.
The wait is what lets a source be built without being run. Testing a connection, sampling and the worker's health check all build a source from its configuration, ping it and close it. When the newest registration simply owned the path, that probe took it from the workflow that was receiving, and closing the probe then deleted it: the workflow reported itself running and every request was refused as unregistered until it was restarted. A probe is never read, so it never takes over.
Taking over on Read keeps what replacing-on-registration was for: when a workflow moves between engines, the one that starts reading is the one that receives, and the outgoing one's teardown cannot remove its successor.
func NewPathRegistry ¶ added in v1.17.0
func NewPathRegistry(buffer int) *PathRegistry
NewPathRegistry returns a registry whose channels buffer the given number of messages.
func (*PathRegistry) Dispatch ¶ added in v1.17.0
func (r *PathRegistry) Dispatch(path string, msg hermod.Message) error
Dispatch hands a message to the path's holder without blocking.
The lock is held across the send. Released first, an Unregister could close the channel between the lookup and the send, and a send on a closed channel panics.
func (*PathRegistry) Register ¶ added in v1.17.0
func (r *PathRegistry) Register(path string) chan hermod.Message
Register creates the channel a source reads its path from. It holds the path if nothing else does, and waits otherwise.
func (*PathRegistry) TakeOver ¶ added in v1.17.0
func (r *PathRegistry) TakeOver(path string, ch chan hermod.Message)
TakeOver makes a waiting channel the path's holder. A source calls it when it starts reading. The channel it replaces is left alone: its reader is on its way out, and its own Unregister finds it no longer holds anything.
It does nothing for a channel that already holds the path, or that was never registered for it.
func (*PathRegistry) Unregister ¶ added in v1.17.0
func (r *PathRegistry) Unregister(path string, ch chan hermod.Message)
Unregister releases a channel's claim on a path.
The holder's channel is closed and the path passes to the newest source waiting for it, or is removed if none is. A waiting channel is only taken off the list. A channel that has been replaced holds nothing, so releasing it changes nothing — which is what stops an outgoing engine's teardown from removing the one that took over from it.
type PollingSource ¶
type PollingSource struct {
// contains filtered or unexported fields
}
PollingSource is a generic implementation of a source that polls a database for changes.
func NewPollingSource ¶
func (*PollingSource) Close ¶
func (s *PollingSource) Close() error
func (*PollingSource) SetLogger ¶
func (s *PollingSource) SetLogger(logger hermod.Logger)
Source Files
Directories
¶
| Path | Synopsis |
|---|---|
|
Package metis reads a BPMN 2.0 workflow engine into a Hermod pipeline.
|
Package metis reads a BPMN 2.0 workflow engine into a Hermod pipeline. |