Documentation
¶
Overview ¶
Example ¶
The README shows this under "As a Library". It lived only in the README and named three packages that do not exist -- pkg/sink/stdout and pkg/buffer are really pkg/comm/sink/stdout and pkg/comm/buffer -- and passed no formatter to a constructor that requires one. Nothing caught it because the module path in go.mod matched no repository, so the example could not have been compiled by anyone, including whoever wrote it.
It is a compiled example now, in an external test package so the imports read the way a caller's would. There is no "Output:" comment on purpose: `go test` compiles an example without one but does not run it, which is what this needs -- Start blocks on a real source, and the thing worth checking is that the code a new user copies still builds.
package main
import (
"context"
"io"
"time"
"github.com/gsoultan/hermod"
"github.com/gsoultan/hermod/pkg/comm/buffer"
"github.com/gsoultan/hermod/pkg/comm/formatter/json"
"github.com/gsoultan/hermod/pkg/comm/sink/stdout"
"github.com/gsoultan/hermod/pkg/engine"
"github.com/gsoultan/hermod/pkg/engine/config"
)
// The README shows this under "As a Library". It lived only in the README and
// named three packages that do not exist -- pkg/sink/stdout and pkg/buffer are
// really pkg/comm/sink/stdout and pkg/comm/buffer -- and passed no formatter to
// a constructor that requires one. Nothing caught it because the module path in
// go.mod matched no repository, so the example could not have been compiled by
// anyone, including whoever wrote it.
//
// It is a compiled example now, in an external test package so the imports read
// the way a caller's would. There is no "Output:" comment on purpose: `go test`
// compiles an example without one but does not run it, which is what this
// needs -- Start blocks on a real source, and the thing worth checking is that
// the code a new user copies still builds.
func main() {
source := exampleSource{}
sinks := []hermod.Sink{stdout.NewStdoutSink(json.NewJSONFormatter())}
buf := buffer.NewRingBuffer(1024)
eng := engine.NewEngine(source, sinks, buf)
eng.SetConfig(config.Config{
MaxRetries: 5,
RetryInterval: 200 * time.Millisecond,
})
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
_ = eng.Start(ctx)
}
// exampleSource stands in for a real source. Read returning io.EOF is what a
// finite source does once it is drained.
type exampleSource struct{}
func (exampleSource) Read(context.Context) (hermod.Message, error) { return nil, io.EOF }
func (exampleSource) Ack(context.Context, hermod.Message) error { return nil }
func (exampleSource) Ping(context.Context) error { return nil }
func (exampleSource) Close() error { return nil }
Output:
Index ¶
- Constants
- func NewDefaultLogger() hermod.Logger
- func PendingOverReleaseCount() int64
- type BackpressureStrategy
- type CheckpointManager
- type Config
- type Engine
- func (e *Engine) AcquireNode(ctx context.Context, nodeID string) error
- func (e *Engine) Checkpoint(ctx context.Context) error
- func (e *Engine) DeadLetterNodeFailure(ctx context.Context, nodeID string, msg hermod.Message, cause error) bool
- func (e *Engine) DetectAnomaly(duration time.Duration) bool
- func (e *Engine) DrainDLQ(ctx context.Context) error
- func (e *Engine) GetConcurrency() int
- func (e *Engine) GetSinkConfigs() []config.SinkConfig
- func (e *Engine) GetSinks() []hermod.Sink
- func (e *Engine) GetSource() hermod.Source
- func (e *Engine) GetStatus() telemetry.StatusUpdate
- func (e *Engine) HardStop()
- func (e *Engine) IsSafeMode() bool
- func (e *Engine) LastMsgTime() time.Time
- func (e *Engine) RecordTraceStep(ctx context.Context, msg hermod.Message, nodeID string, start time.Time, ...)
- func (e *Engine) ReleaseNode(nodeID string)
- func (e *Engine) SetCheckpointHandler(fn func(context.Context, map[string]string) error)
- func (e *Engine) SetConfig(cfg config.Config)
- func (e *Engine) SetDeadLetterSink(snk hermod.Sink)
- func (e *Engine) SetIDs(workflowID string, sourceID string, sinkIDs []string)
- func (e *Engine) SetLogger(l hermod.Logger)
- func (e *Engine) SetOnStall(fn func(reason string))
- func (e *Engine) SetOnStatusChange(fn func(telemetry.StatusUpdate))
- func (e *Engine) SetOutboxStorage(outbox hermod.OutboxStorage)
- func (e *Engine) SetRouter(r RouterFunc)
- func (e *Engine) SetSafeMode(enabled bool)
- func (e *Engine) SetSinkConfigs(configs []config.SinkConfig)
- func (e *Engine) SetSinkIDs(ids []string)
- func (e *Engine) SetSinkTypes(types []string)
- func (e *Engine) SetSourceConfig(cfg config.SourceConfig)
- func (e *Engine) SetSourceID(id string)
- func (e *Engine) SetTraceRecorder(tr hermod.TraceRecorder)
- func (e *Engine) SetValidator(v schema.Validator)
- func (e *Engine) SetWorkflowID(id string)
- func (e *Engine) SimulateFailure(duration time.Duration)
- func (e *Engine) Start(ctx context.Context) error
- func (e *Engine) UpdateConcurrency(n int)
- func (e *Engine) UpdateEdgeMetric(sourceNodeID string, targetNodeID string, count uint64)
- func (e *Engine) UpdateNodeErrorMetric(nodeID string, count uint64)
- func (e *Engine) UpdateNodeMetric(nodeID string, count uint64)
- func (e *Engine) UpdateNodeSample(nodeID string, data map[string]any)
- func (e *Engine) UpdateSinkConfig(sinkID string, update func(*config.SinkConfig))
- type RoutedMessage
- type RouterFunc
- type Runner
- type SinkConfig
- type SourceConfig
- type StatusUpdate
Examples ¶
Constants ¶
const DefaultLagWarnBytes uint64 = 256 << 20
DefaultLagWarnBytes is how much un-acknowledged WAL a source may retain before it is reported. A replication slot that stops advancing pins WAL on the *source* database, so an unnoticed stall does not degrade Hermod — it fills the customer's primary and takes it down. 256 MiB is small enough to be a warning rather than an incident and large enough not to fire on a normal backlog.
const DefaultStallThreshold = 60 * time.Second
DefaultStallThreshold is how long a pipeline may hold outstanding work without completing any of it before it is treated as wedged rather than busy. Slow sinks and long retry backoffs are normal, so this is deliberately generous; the failure it exists to catch lasts indefinitely, not seconds.
const DefaultStreamSilenceInterval = 10 * time.Second
DefaultStreamSilenceInterval is how often a source's push stream is sampled for silence when the engine config does not say.
Variables ¶
This section is empty.
Functions ¶
func NewDefaultLogger ¶
NewDefaultLogger creates a DefaultLogger with stderr output and timestamps.
func PendingOverReleaseCount ¶
func PendingOverReleaseCount() int64
PendingOverReleaseCount reports how many times a pendingMessage was released more often than it was referenced. Any non-zero value indicates unbalanced release bookkeeping somewhere in the writer paths.
Types ¶
type BackpressureStrategy ¶
type BackpressureStrategy = config.BackpressureStrategy
type CheckpointManager ¶
type CheckpointManager struct {
// contains filtered or unexported fields
}
func NewCheckpointManager ¶
func (*CheckpointManager) Checkpoint ¶
func (m *CheckpointManager) Checkpoint(ctx context.Context) error
func (*CheckpointManager) IsInCheckpoint ¶
func (m *CheckpointManager) IsInCheckpoint() bool
type Config ¶
Re-export common types from sub-packages for backward compatibility
func DefaultConfig ¶
func DefaultConfig() Config
DefaultConfig returns the default configuration for the Engine.
type Engine ¶
type Engine struct {
// contains filtered or unexported fields
}
Engine orchestrates the data flow from Source to Sinks. It acts as a Facade, delegating complex tasks to internal components.
func (*Engine) AcquireNode ¶
AcquireNode attempts to acquire a concurrency slot for a specific node. This implements node-level backpressure.
func (*Engine) DeadLetterNodeFailure ¶
func (e *Engine) DeadLetterNodeFailure(ctx context.Context, nodeID string, msg hermod.Message, cause error) bool
DeadLetterNodeFailure sends a message that a workflow node could not process to the dead-letter sink, and reports whether it went anywhere.
Node failures did not reach the dead-letter sink at all. The traversal logged "Node %s failed" and released the message, so a workflow with a dead-letter sink configured still lost every message a transformation, condition or any other node rejected — the DLQ covered validation and sink failures only, and the gap was invisible because the log line looked like handling.
Returning false means nothing was configured to catch it, which is the caller's cue to say so rather than imply the message was kept.
func (*Engine) DrainDLQ ¶
DrainDLQ attempts to wrap the current source with a PrioritySource to drain DLQ messages.
func (*Engine) GetConcurrency ¶
func (*Engine) GetSinkConfigs ¶
func (e *Engine) GetSinkConfigs() []config.SinkConfig
func (*Engine) GetStatus ¶
func (e *Engine) GetStatus() telemetry.StatusUpdate
GetStatus returns the current status of the engine.
func (*Engine) HardStop ¶
func (e *Engine) HardStop()
HardStop attempts to forcefully and quickly stop the engine.
func (*Engine) IsSafeMode ¶
func (*Engine) LastMsgTime ¶
LastMsgTime returns the time of the last message received from the source.
func (*Engine) RecordTraceStep ¶
func (*Engine) ReleaseNode ¶
ReleaseNode releases a concurrency slot for a specific node.
func (*Engine) SetCheckpointHandler ¶
func (*Engine) SetDeadLetterSink ¶
func (*Engine) SetOnStall ¶
SetOnStall registers a supervisor for this engine. The watchdog calls it once per stall episode, on its own goroutine, when the pipeline is holding work it has stopped completing.
This exists because a data pipeline has many ways to wedge and only one reliable cure. Three separate wedge causes were found and fixed by hand (a double-pooled pendingMessage, a circuit-breaker self-deadlock, a source that retired itself silently) and the symptom still recurred. Restarting the workflow recovered it every time, losing nothing, because the replication slot holds un-acknowledged WAL until the new engine re-reads it. So rather than assume the last bug was the last bug, supervise: detect the stall and rebuild the engine, the way a process supervisor or a liveness probe would.
func (*Engine) SetOnStatusChange ¶
func (e *Engine) SetOnStatusChange(fn func(telemetry.StatusUpdate))
func (*Engine) SetOutboxStorage ¶
func (e *Engine) SetOutboxStorage(outbox hermod.OutboxStorage)
func (*Engine) SetRouter ¶
func (e *Engine) SetRouter(r RouterFunc)
func (*Engine) SetSafeMode ¶
func (*Engine) SetSinkConfigs ¶
func (e *Engine) SetSinkConfigs(configs []config.SinkConfig)
func (*Engine) SetSinkIDs ¶
func (*Engine) SetSinkTypes ¶
func (*Engine) SetSourceConfig ¶
func (e *Engine) SetSourceConfig(cfg config.SourceConfig)
func (*Engine) SetSourceID ¶
func (*Engine) SetTraceRecorder ¶
func (e *Engine) SetTraceRecorder(tr hermod.TraceRecorder)
func (*Engine) SetValidator ¶
func (*Engine) SetWorkflowID ¶
func (*Engine) SimulateFailure ¶
func (*Engine) UpdateConcurrency ¶
func (*Engine) UpdateEdgeMetric ¶
func (*Engine) UpdateNodeErrorMetric ¶
func (*Engine) UpdateNodeMetric ¶
func (*Engine) UpdateNodeSample ¶
UpdateNodeSample stores the latest payload sample for a node. Callers must pass an independent map (e.g. produced by Registry.getConsistentData) that is not mutated afterwards; the value is stored as-is to avoid an extra full-payload JSON round-trip on every message.
func (*Engine) UpdateSinkConfig ¶
func (e *Engine) UpdateSinkConfig(sinkID string, update func(*config.SinkConfig))
type RoutedMessage ¶
RoutedMessage represents a message and its target sink index.
type RouterFunc ¶
RouterFunc is a function that routes a message to one or more sinks.
type SinkConfig ¶
type SinkConfig = config.SinkConfig
type SourceConfig ¶
type SourceConfig = config.SourceConfig
type StatusUpdate ¶
type StatusUpdate = telemetry.StatusUpdate
Source Files
Directories