heartbeat

Documentation

Index

• type Client
• type Heartbeat
• type HeartbeatClient
  • func NewClient(conn *nats.Conn, nodeID string, topic string) (*HeartbeatClient, error)
  • func (h *HeartbeatClient) Close(ctx context.Context) error
  • func (h *HeartbeatClient) Publish(ctx context.Context, message Heartbeat) error
  • func (h *HeartbeatClient) SendHeartbeat(ctx context.Context, sequence uint64) error
• type HeartbeatServer
  • func NewServer(params HeartbeatServerParams) (*HeartbeatServer, error)
  • func (h *HeartbeatServer) CheckQueue(ctx context.Context)
  • func (h *HeartbeatServer) FilterNodeInfos(nodeInfos []*models.NodeInfo, state models.NodeConnectionState) []*models.NodeInfo
  • func (h *HeartbeatServer) Handle(ctx context.Context, message Heartbeat) error
  • func (h *HeartbeatServer) RemoveNode(nodeID string)
  • func (h *HeartbeatServer) Start(ctx context.Context) error
  • func (h *HeartbeatServer) UpdateNodeInfo(state *models.NodeState)
• type HeartbeatServerParams
• type TimestampedHeartbeat

Types

type Client

type Client interface {
	SendHeartbeat(ctx context.Context, sequence uint64) error
	Close(ctx context.Context) error
}

type Heartbeat

type Heartbeat struct {
	NodeID   string
	Sequence uint64
}

Heartbeat represents a heartbeat message from a specific node. It contains the node ID and the sequence number of the heartbeat which is monotonically increasing (reboots aside). We do not use timestamps on the client, we rely solely on the server-side time to avoid clock drift issues.

type HeartbeatClient

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

func NewClient

func NewClient(conn *nats.Conn, nodeID string, topic string) (*HeartbeatClient, error)

func (*HeartbeatClient) Close

func (h *HeartbeatClient) Close(ctx context.Context) error

func (*HeartbeatClient) Publish

func (h *HeartbeatClient) Publish(ctx context.Context, message Heartbeat) error

func (*HeartbeatClient) SendHeartbeat

func (h *HeartbeatClient) SendHeartbeat(ctx context.Context, sequence uint64) error

type HeartbeatServer

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

func NewServer

func NewServer(params HeartbeatServerParams) (*HeartbeatServer, error)

func (*HeartbeatServer) CheckQueue

func (h *HeartbeatServer) CheckQueue(ctx context.Context)

CheckQueue will check the queue for old heartbeats that might make a node's liveness either unhealthy or unknown, and will update the node's status accordingly.

func (*HeartbeatServer) FilterNodeInfos

func (h *HeartbeatServer) FilterNodeInfos(nodeInfos []*models.NodeInfo, state models.NodeConnectionState) []*models.NodeInfo

FilterNodeInfos will return only those NodeInfos that have the requested liveness

func (*HeartbeatServer) Handle

func (h *HeartbeatServer) Handle(ctx context.Context, message Heartbeat) error

func (*HeartbeatServer) RemoveNode

func (h *HeartbeatServer) RemoveNode(nodeID string)

RemoveNode will handle removing the liveness for a specific node. This is useful when a node is removed from the cluster.

func (*HeartbeatServer) Start

func (h *HeartbeatServer) Start(ctx context.Context) error

func (*HeartbeatServer) UpdateNodeInfo

func (h *HeartbeatServer) UpdateNodeInfo(state *models.NodeState)

UpdateNode will add the liveness for specific nodes to their NodeInfo

type HeartbeatServerParams

type HeartbeatServerParams struct {
	Client                *nats.Conn
	Topic                 string
	Clock                 clock.Clock
	CheckFrequency        time.Duration
	NodeDisconnectedAfter time.Duration
}

type TimestampedHeartbeat

type TimestampedHeartbeat struct {
	Heartbeat
	Timestamp int64
}