consensus

package
v0.21.0 Latest Latest
Warning

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

 Go to latest Published: Jul 2, 2026 License: Apache-2.0 Imports: 27 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	ReasonSoakStarted           = "SoakStarted"
	ReasonSoakResumed           = "SoakResumed" // daemon restarted mid-soak; elapsed time preserved
	ReasonSoakCheck             = "SoakCheck"
	ReasonCriterionMet          = "CriterionMet"
	ReasonFleetThresholdReached = "FleetThresholdReached"
	ReasonDecommissionTriggered = "DecommissionTriggered"
	ReasonDecommissionCompleted = "DecommissionCompleted"
	ReasonDecommissionFailed    = "DecommissionFailed"
	ReasonSoakWatcherPanicked   = "SoakWatcherPanicked"  // panic recovered in watcher goroutine
	ReasonSoakStateCorrupted    = "SoakStateCorrupted"   // state file malformed/invalid; soak abandoned
	ReasonSoakStateWriteFailed  = "SoakStateWriteFailed" // state file not persisted; crash recovery compromised
)

HIP-defined JSONL event reasons — written verbatim to consensus-migrate-events.jsonl. Do NOT change without a corresponding HIP amendment; external tooling (Alloy pipelines, Loki alert rules) depends on their exact spelling.

View Source 
const DefaultSoakPeriod = 48 * time.Hour

DefaultSoakPeriod is the HIP-specified minimum soak duration.

View Source 
const MigrationPlanBaseDir = "/opt/solo/weaver/migration/consensus"

MigrationPlanBaseDir is the only directory from which a consensus-node migration plan may be loaded. SoakStartRequest.Validate anchors MigrationPlanPath within this directory so a socket client (any weaver-group member) cannot point the daemon at an arbitrary file on the host. Keep this in sync with where the migration workflow stages plans on disk.

Variables

View Source 
var (
	ErrNamespace = errorx.NewNamespace("daemon.consensus")

	// ErrK8sClient is returned when the Kubernetes dynamic client cannot be built
	// (e.g. kubeconfig missing or malformed).
	ErrK8sClient = ErrNamespace.NewType("k8s_client")

	// ErrWatchFailed is returned when the Kubernetes watch API call fails or
	// returns an error event (e.g. 401/403, server disconnect).
	ErrWatchFailed = ErrNamespace.NewType("watch_failed")

	// ErrSoakWatcher is returned when the soak watcher encounters an error
	// (e.g. state file I/O, decommission failure).
	ErrSoakWatcher = ErrNamespace.NewType("soak_watcher")
)

Functions

This section is empty.

Types

type ComponentConfig 

type ComponentConfig struct {
	NodeID           string
	KubeconfigPath   string
	Orbit            string
	UpgradeEnabled   bool
	MigrationEnabled bool
	UpgradeEventsDir string
	HomeDir          string
	UpgradeDir       string
	MigrateEventsDir string
}

ComponentConfig holds all inputs needed to build the consensus-node component. Constructed by daemon.go from ConsensusNodeComponentConfig and WeaverPaths so that this package does not import the daemon package (avoiding cycles).

type ComponentResult 

type ComponentResult struct {
	// Monitors is the ordered slice of monitors to run under the supervisor.
	Monitors []daemonkit.MonitorRunner

	// MigrationMonitor is non-nil when the migration monitor is enabled.
	// daemon.go uses this to construct ConsensusNodeHandler with the correct
	// migrationStateFn closure after the component's StatusTracker is created.
	MigrationMonitor *MigrationMonitor
}

ComponentResult contains the monitors built by NewComponent and a reference to the MigrationMonitor (when enabled) so daemon.go can wire the HTTP handler with the per-component StatusTracker closure after the component is assembled.

func NewComponent 

func NewComponent(cfg ComponentConfig) (ComponentResult, error)

NewComponent constructs all enabled monitors for the consensus-node component and returns them alongside any references needed for HTTP handler wiring. It is the single entry point for consensus-node component assembly — daemon.go only needs to call this and wire the result into the Daemon struct.

type ConsensusMigrationStatusResponse 

type ConsensusMigrationStatusResponse struct {
	Monitor daemonkit.MonitorState `json:"monitor"`
	Soak    SoakStatusResponse     `json:"soak"`
}

ConsensusMigrationStatusResponse is the combined view returned by GET /consensus_node/migration/status: supervisor health + current soak state.

type ConsensusNodeHandler 

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

ConsensusNodeHandler implements daemonkit.ComponentHandler for all consensus-node HTTP routes under the /consensus_node/ prefix.

To add a new monitor for the consensus-node component:

  1. Add a field for the monitor and its state func to this struct.
  2. Register the new route(s) in RegisterRoutes.
  3. Implement the handler method(s) on *ConsensusNodeHandler.

func NewConsensusNodeHandler 

func NewConsensusNodeHandler(mm *MigrationMonitor, migrationStateFn func() daemonkit.MonitorState) *ConsensusNodeHandler

NewConsensusNodeHandler constructs a ConsensusNodeHandler. mm and migrationStateFn may be nil when the migration monitor is disabled.

func (*ConsensusNodeHandler) RegisterRoutes 

func (h *ConsensusNodeHandler) RegisterRoutes(mux *http.ServeMux)

RegisterRoutes implements daemonkit.ComponentHandler. All routes are prefixed with /consensus_node/.

Current routes: 

GET    /consensus_node/migration/status       — combined: monitor health + soak state
GET    /consensus_node/migration/soak/status  — soak-run state only
POST   /consensus_node/migration/soak/start   — enqueue a new soak run
DELETE /consensus_node/migration/soak         — stop the running soak watcher

type ConsensusParticipationNominal 

type ConsensusParticipationNominal struct{}

ConsensusParticipationNominal is green when the CN is actively participating in consensus — signing and submitting transactions within acceptable bounds. Stub — real implementation in a subsequent story.

func (ConsensusParticipationNominal) Check 

func (ConsensusParticipationNominal) Check(_ context.Context, _ SoakStartRequest) (bool, error)

func (ConsensusParticipationNominal) Name 

func (ConsensusParticipationNominal) Name() string

type Decommissioner 

type Decommissioner interface {
	Decommission(ctx context.Context, nodeID string) error
}

Decommissioner triggers node decommission once all soak criteria are met.

type MigrationMonitor 

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

MigrationMonitor manages the migration soak lifecycle: it owns the activation channel, shared status, and the goroutine that monitors soak criteria. Once all mainnet nodes are migrated to the new deployment model, this can be safely disabled or removed from the codebase.

func NewMigrationMonitor 

func NewMigrationMonitor() *MigrationMonitor

NewMigrationMonitor returns a zero-config MigrationMonitor. Provided for backward compatibility with existing tests that only need the dispatch loop and HTTP handler wiring. Production code should use NewMigrationMonitorWith.

func NewMigrationMonitorWith 

func NewMigrationMonitorWith(nodeID string, logger *eventlog.EventLogger, d Decommissioner, cfg MigrationMonitorConfig, stateDir string) *MigrationMonitor

NewMigrationMonitorWith constructs a MigrationMonitor wired with the given dependencies.

func (*MigrationMonitor) Close 

func (mm *MigrationMonitor) Close() error

Close implements io.Closer. It flushes and closes the event logger if one was injected. Safe to call when logger is nil (no-op). Must be called after Run returns — not safe to call concurrently with logMigrateEvent.

func (*MigrationMonitor) Name 

func (mm *MigrationMonitor) Name() string

Name implements daemon.MonitorRunner.

func (*MigrationMonitor) Run 

func (mm *MigrationMonitor) Run(ctx context.Context) error

Run is the dispatch loop. It blocks until ctx is cancelled, then waits for any in-flight watcher goroutines to finish before returning.

func (*MigrationMonitor) Status 

func (mm *MigrationMonitor) Status() *SoakStatusResponse

Status returns the current soak status. Never returns nil. Returns the live pointer when a watcher is active to avoid a copy on the hot GET /consensus_node/migration/soak/status read path.

func (*MigrationMonitor) TryEnqueue 

func (mm *MigrationMonitor) TryEnqueue(req SoakStartRequest) bool

TryEnqueue queues a soak activation request. Returns false if the request cannot be accepted; the caller should respond with 409 Conflict. Two conditions both map to false:

  • a watcher goroutine is already running (soakActive = true)
  • the channel is already full (a prior request is queued but not yet dispatched)

Both cases are intentionally indistinguishable to callers — only one soak activation may be in-flight at any time.

soakStatus is set to Active=true here, synchronously, so that GET /consensus_node/migration/soak/status reflects the accepted request immediately — without waiting for the watcher goroutine (spawned by Run) to store the same value. The watcher stores it again on startup (idempotent), and clears it on exit.

func (*MigrationMonitor) TryStop 

func (mm *MigrationMonitor) TryStop(deleteState bool) bool

TryStop cancels the running soak watcher and waits for it to drain. Returns false when no watcher is currently active (caller should 409). When deleteState is true the persisted cutover-state.jsonl is removed so the daemon does not auto-resume the soak on the next restart.

Transient false: soakActive is set (by TryEnqueue/resumeIfNeeded) before the watcher goroutine runs and stores soakCancel. A stop that lands in that brief accept→spawn window sees a nil soakCancel and returns false even though a soak is logically active. Stop is idempotent, so the documented client behaviour is to retry; the window closes as soon as the scheduled goroutine reaches its soakCancel.Store. We intentionally do not block here to wait for it — that would couple TryStop to goroutine scheduling for a sub-millisecond race.

func (*MigrationMonitor) WithCriteria 

func (mm *MigrationMonitor) WithCriteria(criteria ...SoakCriterion) *MigrationMonitor

WithCriteria sets the soak criteria to evaluate on each poll tick. Call before Run — not safe to call concurrently.

type MigrationMonitorConfig 

type MigrationMonitorConfig struct {
	// PollInterval is how often the soak-watcher evaluates criteria.
	// Zero value defaults to 900s (15 min) per HIP spec.
	PollInterval time.Duration

	// FleetThresholdPath is the host flag file whose presence means ≥26/39
	// fleet nodes have migrated successfully. Zero value defaults to the
	// HIP-specified path below.
	FleetThresholdPath string
}

MigrationMonitorConfig holds tunable parameters for MigrationMonitor.

type NoPodRestarts 

type NoPodRestarts struct {
	// KubeconfigPath is the path to the daemon kubeconfig used to contact the K8s API.
	KubeconfigPath string

	// Namespace is the orbit namespace where the CN pod runs.
	Namespace string

	// PodLabelSelector is the label selector used to identify the CN pod managed by
	// this daemon instance. The selector must match exactly the pods that belong to
	// this node's ConsensusCapsule StatefulSet.
	// Example: "operator.solo.hedera.com/orbit=mainnet-00,operator.solo.hedera.com/node-id=0.0.3"
	PodLabelSelector string
	// contains filtered or unexported fields
}

NoPodRestarts is green when the CN pod — identified by PodLabelSelector in Namespace — was started after the cutover timestamp and has accumulated zero container restarts since then, indicating stable operation.

Logic:

  • List pods matching PodLabelSelector in Namespace.
  • Consider only pods whose creation timestamp is after req.CutoverTimestamp.
  • If at least one such pod exists and all of its containers have RestartCount == 0: green.
  • If no post-cutover pod is found yet (still starting up): not-green (not an error).
  • If any post-cutover pod has at least one container restart: not-green (not an error).

TotalRestarts returns the sum of all container restart counts across post-cutover pods from the last Check call. Use it to populate diagnostic fields in SoakCheck.

func (*NoPodRestarts) Check 

func (c *NoPodRestarts) Check(ctx context.Context, req SoakStartRequest) (bool, error)

func (*NoPodRestarts) Name 

func (c *NoPodRestarts) Name() string

func (*NoPodRestarts) TotalRestarts 

func (c *NoPodRestarts) TotalRestarts() int

TotalRestarts returns the sum of all container restart counts across post-cutover pods observed during the last Check call. Zero when no post-cutover pods exist yet.

type NoopDecommissioner 

type NoopDecommissioner struct{}

NoopDecommissioner logs the call and returns nil. Used in this story while the real K8s decommission API call is deferred to a subsequent story.

func (*NoopDecommissioner) Decommission 

func (*NoopDecommissioner) Decommission(_ context.Context, nodeID string) error

type RestartCounter 

type RestartCounter interface {
	TotalRestarts() int
}

RestartCounter is an optional interface that a SoakCriterion may implement to expose the total container restart count observed during the last Check call. The MigrationMonitor checks for this interface when building the SoakCheck payload.

type SoakCriterion 

type SoakCriterion interface {
	// Name returns the unique criterion identifier used in CriterionMet events.
	Name() string
	// Check evaluates the criterion. Returns (true, nil) when the criterion is
	// green. On error, the caller logs a warning and treats the criterion as
	// not-green — a flaky check never triggers decommission.
	Check(ctx context.Context, req SoakStartRequest) (bool, error)
}

SoakCriterion is the interface every soak criterion must implement.

type SoakDuration 

type SoakDuration struct {
	// Period is the minimum time that must elapse since cutover before this
	// criterion is considered met. Defaults to 48h when zero.
	Period time.Duration
}

SoakDuration is green when at least Period has elapsed since the cutover timestamp in the activation request. Zero value of Period defaults to DefaultSoakPeriod (48h) per HIP spec.

func (SoakDuration) Check 

func (c SoakDuration) Check(_ context.Context, req SoakStartRequest) (bool, error)

func (SoakDuration) Name 

func (c SoakDuration) Name() string

type SoakStartRequest 

type SoakStartRequest struct {
	NodeID            string    `json:"node_id"`
	CutoverTimestamp  time.Time `json:"cutover_timestamp"`
	MigrationPlanPath string    `json:"migration_plan_path"`
}

SoakStartRequest is the payload for POST /consensus_node/migration/soak/start.

func (SoakStartRequest) Validate 

func (r SoakStartRequest) Validate() error

Validate checks that all required fields are present and that MigrationPlanPath is a traversal-free path anchored within MigrationPlanBaseDir. Called by the HTTP handler and by resumeIfNeeded (story #520) before activating a watcher from a persisted request — the path arrives over the daemon socket, so validation here is the trust boundary against arbitrary file reads / path traversal.

type SoakStartResponse 

type SoakStartResponse struct {
	Accepted bool `json:"accepted"`
}

SoakStartResponse is returned by POST /consensus_node/migration/soak/start on accept.

type SoakStatusResponse 

type SoakStatusResponse struct {
	Active  bool              `json:"active"`
	Request *SoakStartRequest `json:"request,omitempty"`
}

SoakStatusResponse is returned by GET /consensus_node/migration/soak/status.

type UpgradeMonitor 

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

UpgradeMonitor watches the Kubernetes API for NetworkUpgradeExecute CRs whose status.phase transitions to ReadyForProvisionerDaemon and triggers the execute-phase workflow. It is self-healing: transient errors and auth failures are retried with exponential backoff so the daemon recovers without a restart.

func NewUpgradeMonitor 

func NewUpgradeMonitor(cfg UpgradeMonitorConfig) (*UpgradeMonitor, error)

NewUpgradeMonitor constructs an UpgradeMonitor and builds the Kubernetes dynamic client from the kubeconfig at cfg.KubeconfigPath.

func NewUpgradeMonitorWithClient 

func NewUpgradeMonitorWithClient(cfg UpgradeMonitorConfig, client dynamic.Interface) *UpgradeMonitor

NewUpgradeMonitorWithClient constructs an UpgradeMonitor with an injected client — used in unit tests to avoid a real kubeconfig on disk.

func (*UpgradeMonitor) ConnectivityError 

func (um *UpgradeMonitor) ConnectivityError() *daemonkit.StatusError

ConnectivityError implements daemonkit.ConnectivityMonitor. Returns the current connectivity failure (list or watch error), or nil when the last cycle completed without error. The monitor continues retrying automatically; callers use this for observability only.

func (*UpgradeMonitor) Name 

func (um *UpgradeMonitor) Name() string

Name implements daemonkit.MonitorRunner.

func (*UpgradeMonitor) RequiredProbe 

func (um *UpgradeMonitor) RequiredProbe() daemonkit.Probe

RequiredProbe implements daemon.ProbableMonitor. It returns a composite probe that verifies the disk prerequisites that must be correct before the first upgrade CR arrives. Catching these at startup avoids silent failures mid-automation when human intervention is no longer practical.

Kube RBAC / CRD availability is intentionally excluded: the watch loop already retries those on its own backoff schedule, so a missing CRD at startup is not a problem — it resolves automatically once the CRD is installed.

  1. Parent dir ownership — the upgrade staging root (e.g. .../data/upgrade) is owned by hedera:hedera with at least rwxr-xr-x (0755).
  2. Current dir ownership — the active staging subdir (UpgradeDir) is owned by hedera:hedera with at least rwxrwxr-x (0775).
  3. Current dir write access — the running daemon process can actually write to UpgradeDir, proving `usermod -aG hedera weaver` was applied and mount flags, ACLs, and SELinux/AppArmor policies all allow writes.

func (*UpgradeMonitor) Run 

func (um *UpgradeMonitor) Run(ctx context.Context) error

Run blocks until ctx is cancelled. On every iteration it lists all NetworkUpgradeExecute CRs to seed completedOpIDs and dispatch any CRs already at ReadyForProvisionerDaemon (recovery after a crash or restart), then watches from the List's ResourceVersion so no events are missed between the two calls. Clean watch expiry reconnects after backoffInitial; real errors retry with exponential backoff; auth errors additionally rebuild the dynamic client from the kubeconfig on disk before retrying.

type UpgradeMonitorConfig 

type UpgradeMonitorConfig struct {
	// KubeconfigPath is the path to the daemon's scoped kubeconfig. Built once
	// from daemon.yaml at startup; systemd restarts the daemon if the credential
	// changes after a cluster rebuild.
	KubeconfigPath string

	// Namespace is the orbit namespace to watch for NetworkUpgradeExecute CRs.
	// One daemon manages one CN, so one namespace is sufficient.
	Namespace string

	// NodeID is the Hedera node identifier for the consensus node managed by
	// this daemon (e.g. "0.0.3"). Populated as nodeId in all JSONL event log
	// entries emitted by handleExecute.
	NodeID string

	// UpgradeEventsDir is the directory where per-operation consensus-upgrade-*.jsonl
	// files are written by handleExecute. The monitor prunes this directory at the
	// start of each Run() invocation (covers both startup and post-crash restarts).
	// Empty string disables pruning.
	UpgradeEventsDir string

	// HomeDir is the weaver home directory used as a safety base for path
	// validation during pruning. Pruning is skipped when UpgradeEventsDir falls
	// outside this tree.
	HomeDir string

	// UpgradeDir is the active upgrade staging directory (the `current`
	// subdirectory under the CN's upgrade path). RequiredProbe verifies
	// ownership and write-access on this directory and its parent before the
	// monitor is allowed to start. Defaults to the CN standard path when empty.
	//
	// Example: /opt/hgcapp/services-hedera/HapiApp2.0/data/upgrade/current
	UpgradeDir string
}

UpgradeMonitorConfig holds configuration for the UpgradeMonitor.

type UploaderBacklogCleared 

type UploaderBacklogCleared struct{}

UploaderBacklogCleared is green when the record stream uploader has no backlogged events from the old CN pending upload to mirror nodes. Stub — real implementation in a subsequent story.

func (UploaderBacklogCleared) Check 

func (UploaderBacklogCleared) Check(_ context.Context, _ SoakStartRequest) (bool, error)

func (UploaderBacklogCleared) Name 

func (UploaderBacklogCleared) Name() string

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.