Documentation
¶
Index ¶
- Constants
- 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 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 ¶
This section is empty.
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 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