primary

package
v1.1.1 Latest Latest
Warning

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

 Go to latest Published: Jun 12, 2026 License: Apache-2.0 Imports: 28 Imported by: 0

Documentation

Overview

Package primary implements the Primary gRPC service and domain layer, including transaction processing, the Follow replication stream, chunk buffering for async object storage writes, Replica heartbeat and receipt tracking, and heartbeat-based degradation.

Index

Constants

This section is empty.

Variables

View Source 
var ErrInvalidTxnRequest = errors.New("invalid txn request")

ErrInvalidTxnRequest is returned when a Txn request is malformed or outside the supported request shapes.

View Source 
var ErrUnsupported = errors.New("Unsupported request - netsy only implementes the Kubernetes etcd API subet")

Functions

func BuildTxnResponse 

func BuildTxnResponse(record *proto.Record, rangeResp *pb.RangeResponse) (*pb.TxnResponse, error)

BuildTxnResponse converts a proto.Record or pb.RangeResponse to a pb.TxnResponse

func ParseTxnRequest 

func ParseTxnRequest(r *pb.TxnRequest) (*proto.Record, error)

ParseTxnRequest validates a pb.TxnRequest and creates a proto.Record

Types

type Metrics 

type Metrics struct {
	// Write path
	WritePath         *prometheus.GaugeVec
	WriteTransactions *prometheus.CounterVec
	WriteDuration     *prometheus.HistogramVec
	QuorumRollbacks   *prometheus.CounterVec
	RequiredReceipts  prometheus.Gauge
	HealthyReplicas   prometheus.Gauge
	ReceiptedReplicas prometheus.Gauge

	// Chunk buffer
	ChunkBufferRecords  prometheus.Gauge
	ChunkBufferBytes    prometheus.Gauge
	ChunkBufferAge      *collectTimeGauge
	ChunkBufferFlushes  *prometheus.CounterVec
	ChunkBufferFlushDur *prometheus.HistogramVec

	// Replication
	ReplicationStreams prometheus.Gauge

	// Drain
	DrainDuration *prometheus.HistogramVec

	// Preflight
	PreflightStageDur *prometheus.HistogramVec

	// Object storage revision
	ObjectStorageRevision prometheus.Gauge

	// Compaction (Primary-scoped)
	CompactionsTotal          *prometheus.CounterVec
	CompactionCoordDuration   *prometheus.HistogramVec
	CompactionConfirmFailures *prometheus.CounterVec
}

Metrics holds Primary-scoped Prometheus metrics. These are registered through a RoleGroup and disappear from scrape output when the node is not the Primary.

func NewMetrics 

func NewMetrics() *Metrics

NewMetrics creates all Primary-scoped Prometheus metrics.

func (*Metrics) Collectors 

func (m *Metrics) Collectors() []prometheus.Collector

Collectors returns all Primary-scoped collectors for registration with a RoleGroup.

func (*Metrics) SetWritePath 

func (m *Metrics) SetWritePath(path string)

SetWritePath sets exactly one write path label to 1 and the other to 0.

type PeerKVServer 

type PeerKVServer struct {
	pb.UnimplementedKVServer
	// contains filtered or unexported fields
}

PeerKVServer implements the etcd KV gRPC service on the Peer API server. Only Txn is implemented; all other KV methods return Unimplemented. This allows Replicas to proxy write requests to the Primary over the existing peer connection without custom proto definitions.

func NewPeerKVServer 

func NewPeerKVServer(handler TxnHandler) *PeerKVServer

NewPeerKVServer returns a PeerKVServer that delegates Txn writes to the given handler function.

func (*PeerKVServer) Txn 

func (s *PeerKVServer) Txn(ctx context.Context, r *pb.TxnRequest) (*pb.TxnResponse, error)

Txn processes a write request received from a Replica via the Peer API. It delegates to the handler, which runs the same local write path used for directly-received client requests.

type Replica 

type Replica struct {
	NodeID         string
	LastHeartbeat  atomic.Int64 // unix nano — last standalone or Receipt-embedded heartbeat
	LastReceipt    atomic.Int64 // unix nano — last successful Receipt for a transaction
	ReceiptCount   atomic.Int64 // total Receipts; must reach 1 before counted as healthy for quorum
	HealthState    atomic.Int32 // int32 representation of nodestate.HealthState
	PrimaryState   atomic.Int32 // int32 representation of nodestate.PrimaryState
	LatestRevision atomic.Int64
}

Replica tracks a Replica's health and heartbeat state as seen by the Primary. Fields use atomic operations so they can be updated independently without holding a lock. HealthState and PrimaryState are stored as int32 because Go atomic operations require numeric types.

func (*Replica) Health 

func (r *Replica) Health() nodestate.HealthState

Health returns the Replica's health state.

func (*Replica) Primary 

func (r *Replica) Primary() nodestate.PrimaryState

Primary returns the Replica's primary state.

func (*Replica) SetHealth 

func (r *Replica) SetHealth(h nodestate.HealthState)

SetHealth atomically updates the Replica's health state.

func (*Replica) SetPrimary 

func (r *Replica) SetPrimary(p nodestate.PrimaryState)

SetPrimary atomically updates the Replica's primary state.

type Replicas 

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

Replicas tracks connected Replicas for quorum and heartbeat monitoring. The Primary uses this to determine which Replicas are healthy for quorum transactions and to detect missed heartbeats.

func NewReplicas 

func NewReplicas() *Replicas

NewReplicas creates an empty Replicas tracker.

func (*Replicas) Add 

func (m *Replicas) Add(nodeID string) *Replica

Add registers a Replica in the map. If the Replica already exists, the entry is replaced — resetting ReceiptCount to zero so the Replica must receipt at least once before being counted toward quorum.

func (*Replicas) All 

func (m *Replicas) All() []*Replica

All returns all Replica entries.

func (*Replicas) Get 

func (m *Replicas) Get(nodeID string) (*Replica, bool)

Get returns the entry for a Replica. The second return value indicates whether the Replica was found.

func (*Replicas) HealthyCount 

func (m *Replicas) HealthyCount() int

HealthyCount returns the number of Replicas currently reporting the Healthy health state, regardless of whether they have receipted a write yet.

func (*Replicas) HealthyForQuorumCount 

func (m *Replicas) HealthyForQuorumCount() int

HealthyForQuorumCount returns the number of Replicas eligible to count toward quorum decisions: they must be Healthy and have receipted at least once.

func (*Replicas) ReceiptedCount 

func (m *Replicas) ReceiptedCount() int

ReceiptedCount returns the number of Replicas that have successfully sent at least one Receipt to the Primary.

func (*Replicas) Remove 

func (m *Replicas) Remove(nodeID string)

Remove removes a Replica from the map.

func (*Replicas) Reset 

func (m *Replicas) Reset()

Reset clears all Replica entries. Called when this node loses Primary leadership.

func (*Replicas) UpdateHeartbeat 

func (m *Replicas) UpdateHeartbeat(nodeID string, health nodestate.HealthState, primary nodestate.PrimaryState, latestRevision int64) bool

UpdateHeartbeat updates a Replica's heartbeat timestamp and state atomically. This is the single code path shared by standalone Heartbeats and Receipt-embedded Heartbeats. It returns false if the Replica is not in the map.

func (*Replicas) UpdateReceipt 

func (m *Replicas) UpdateReceipt(nodeID string, health nodestate.HealthState, primary nodestate.PrimaryState, latestRevision int64) bool

UpdateReceipt updates a Replica's heartbeat timestamp, state, and receipt tracking atomically. Called when a Receipt is received from a Replica via the Follow replication stream. It returns false if the Replica is not in the map.

type Server 

type Server struct {
	proto.UnimplementedPrimaryServer
	// contains filtered or unexported fields
}

Server implements the Primary domain layer and the proto.PrimaryServer gRPC interface. It handles transaction processing, the replication stream, and heartbeat collection from Replicas.

func NewServer 

func NewServer(
	logger *slog.Logger,
	conf *config.Config,
	db localdb.Database,
	snapshotWorker *snapshot.Worker,
	storageClient storage.ObjectStorage,
	state *nodestate.State,
	peerClients *peerclient.Manager,
	watchManager *watch.Manager,
	m *Metrics,
	heartbeatInterval time.Duration,
	degradationCount int,
	compactionMetrics *metrics.CompactionMetrics,
	retryMetrics *metrics.RetryMetrics,
	storageMetrics *metrics.ObjectStorageMetrics,
) (*Server, error)

NewServer constructs the Primary server and seeds its next revision counter from the database.

func (*Server) BroadcastCommit 

func (s *Server) BroadcastCommit(committedRevision int64)

BroadcastCommit sends a committed_revision update to all connected replicas via the follow streams.

func (*Server) BroadcastCompact 

func (s *Server) BroadcastCompact(compactionRevision int64)

BroadcastCompact sends a compaction_revision update to all connected replicas via the follow streams.

func (*Server) BroadcastRecord 

func (s *Server) BroadcastRecord(record *proto.Record)

BroadcastRecord sends a record to all connected replicas via the follow streams.

func (*Server) Follow 

func (s *Server) Follow(stream proto.Primary_FollowServer) error

Follow implements the Primary.Follow bidirectional streaming RPC. Each Replica opens a Follow stream to receive PrimaryMessages and send ReplicaMessages (receipts with embedded heartbeats).

func (*Server) GracefulDrain 

func (s *Server) GracefulDrain(ctx context.Context) error

GracefulDrain transitions the Primary to Draining, flushes the chunk buffer to object storage, and stops Primary services. It is the shutdown sequence that must be called before the normal node deregistration path when the shutting-down node is the Primary. Returns an error if the flush fails; callers should still proceed with shutdown even on error.

func (*Server) LeaderTxn 

func (ps *Server) LeaderTxn(ctx context.Context, r *pb.TxnRequest) (record *proto.Record, parsed *pb.TxnResponse, err error)

LeaderTxn is our backend for the etcd transaction API, responsible for committing changes.

It receives a pb.TxnRequest: https://pkg.go.dev/go.etcd.io/etcd/api/v3/etcdserverpb#TxnRequest

It returns a pb.TxnResponse: https://pkg.go.dev/go.etcd.io/etcd/api/v3/etcdserverpb#TxnResponse

The Kubernetes etcd client only uses a subset of the etcd transaction API.

In the Kubernetes codebase, a storage interface implementation translates calls to a Kubernetes etcd client (in the etcd repository): * Create (->OptimisticPut): https://github.com/kubernetes/kubernetes/blob/master/staging/src/k8s.io/apiserver/pkg/storage/etcd3/store.go#L259 * GuaranteedUpdate (->OptimisticPut): https://github.com/kubernetes/kubernetes/blob/master/staging/src/k8s.io/apiserver/pkg/storage/etcd3/store.go#L448C17-L448C33 * Delete ->conditionalDelete(->OptimisticDelete): https://github.com/kubernetes/kubernetes/blob/master/staging/src/k8s.io/apiserver/pkg/storage/etcd3/store.go#L327 The etcd repository contains the Kubernetes etcd client: * OptimisticPut: https://github.com/kubernetes/kubernetes/blob/master/vendor/go.etcd.io/etcd/client/v3/kubernetes/client.go#L83 * OptimisticDelete: https://github.com/etcd-io/etcd/blob/main/client/v3/kubernetes/client.go#L109

To summarise all Kubernetes etcd transaction request combinations:

  1. compare, which checks if the mod_revision of the field: -> for create requests: =0. meaning, there's no record or the key was deleted. -> for update requests: =prev revision. meaning, it must match kubernetes known version. -> for delete requests: =prev revision. meaning, it must match kubernetes known version.
  2. 1x success, executed if compare succeeds. -> create -> update -> delete
  3. 0 or 1 failure, executed if compare fails: -> create: can have no failure conditions, or range for existing key, returning single/first result. -> update: range for existing key, returning single/first result. -> delete: range for existing key, returning single/first result.

Essentially the compare and failure condition for update and delete are the same, just success differs. Note that create and update can have a lease ID specified, which gets recorded in the success operation.

func (*Server) Replicas 

func (s *Server) Replicas() *Replicas

Replicas returns the server's Replica tracker.

func (*Server) ResignLeadership 

func (s *Server) ResignLeadership() error

ResignLeadership transitions the Primary from Draining back to Replica, giving up leadership so the Elector can elect a new Primary. It is used by the self-degradation path after a successful drain and flush.

func (*Server) RunCompactionScheduler 

func (s *Server) RunCompactionScheduler(ctx context.Context)

RunCompactionScheduler runs a periodic loop that queries all registered nodes for their minimum watch revision, computes the cluster-wide global minimum, and drives the compaction notice/confirmation protocol.

func (*Server) RunDegradationLoop 

func (s *Server) RunDegradationLoop(ctx context.Context)

RunDegradationLoop periodically checks all Replicas and marks any as Degraded if they have missed the configured number of consecutive heartbeats. It runs until ctx is cancelled.

func (*Server) SendHeartbeat 

func (s *Server) SendHeartbeat(_ context.Context, req *proto.NodeState) (_ *emptypb.Empty, err error)

SendHeartbeat receives a standalone heartbeat from a Replica. The same processing code path is used for Receipt-embedded heartbeats via ReplicaMap.UpdateHeartbeat.

func (*Server) StartServices 

func (s *Server) StartServices(parent context.Context)

StartServices starts Primary background services and, when the node is in the Starting state, begins the preflight loop that must complete before the Primary becomes Active. It is a no-op if services are already running.

func (*Server) StopServices 

func (s *Server) StopServices()

StopServices stops Primary background services and resets the follow hub and replica tracker. It is a no-op if services are not running.

type TxnHandler 

type TxnHandler func(ctx context.Context, r *pb.TxnRequest) (*pb.TxnResponse, error)

TxnHandler processes a TxnRequest through the full local write path, including SQLite commit, replication, and watch distribution.

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.