relay

package
v0.0.0-...-ae9409a Latest Latest
Warning

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

 Go to latest Published: Feb 24, 2026 License: MIT Imports: 28 Imported by: 0

Documentation

Overview

Package relay — High Availability support for relay servers.

When the fleet exceeds ~100 nodes, a single relay server becomes a bottleneck and a single point of failure. HAProxy implements:

  • Multiple relay server instances sharing a common store (Postgres)
  • Consistent hashing to route nodes to preferred relay instances
  • Automatic failover: if a relay goes down, its nodes reconnect to another
  • Health monitoring between relay peers
  • Graceful draining for zero-downtime upgrades

Architecture: 

  ┌──────────────────────────────────┐
  │        Load Balancer (L4)        │
  └──┬──────────┬──────────┬─────────┘
     │          │          │
┌────▼──┐  ┌───▼───┐  ┌──▼─────┐
│Relay-1│  │Relay-2│  │Relay-3 │
└───┬───┘  └───┬───┘  └───┬────┘
    │          │          │
┌───▼──────────▼──────────▼───┐
│     Shared Store (PG)       │
└────────────────────────────-┘

Package relay — mTLS authentication for relay connections.

mTLS (mutual TLS) replaces shared bearer tokens with X.509 certificate-based authentication. Both the relay server and node agents present certificates signed by a shared CA, providing strong bidirectional identity verification.

Certificate hierarchy: 

[CA cert] ─── signs ──► [Server cert]   (relay server identity)
     │
     └── signs ──► [Node cert]          (per-node identity, CN=<node-id>)

Usage: 

# Generate CA, server, and node certs:
devopsclaw relay cert-gen --ca --out /etc/devopsclaw/certs/
devopsclaw relay cert-gen --server --ca-cert ca.pem --ca-key ca-key.pem
devopsclaw relay cert-gen --node --node-id web-01 --ca-cert ca.pem --ca-key ca-key.pem

Package relay provides NAT-traversal connectivity between the control plane and fleet nodes. Nodes make outbound connections to the relay server, eliminating the need for port forwarding, VPNs, or direct SSH access.

Architecture: 

[Control Plane / CLI / Chat] ─── gRPC ───► [Relay Server] ◄─── gRPC ─── [Node Agent]
                                              │                              │
                                         mTLS + auth                   outbound only
                                         session mgmt                  auto-reconnect
                                         multiplexing                  heartbeat

The relay server runs as a standalone service or embedded in the control plane. Node agents connect outbound (NAT-friendly) and maintain persistent streams. Commands are forwarded through the relay to target nodes.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func ClientTLSConfig 

func ClientTLSConfig(cfg MTLSConfig) (*tls.Config, error)

ClientTLSConfig builds a *tls.Config for a node agent connecting to the relay. The agent presents its own cert and verifies the server's cert against the CA.

func ExtractNodeIDFromCert 

func ExtractNodeIDFromCert(state *tls.ConnectionState) (string, error)

ExtractNodeIDFromCert extracts the node ID from a verified client certificate. The node ID is stored in the certificate's Common Name (CN) field.

func GenerateCA 

func GenerateCA(org string, validFor time.Duration) (certPEM, keyPEM []byte, err error)

GenerateCA creates a self-signed CA certificate and private key.

func GenerateNodeCert 

func GenerateNodeCert(caCertPEM, caKeyPEM []byte, nodeID string, validFor time.Duration) (certPEM, keyPEM []byte, err error)

GenerateNodeCert creates a client certificate for a fleet node, signed by the CA. The nodeID is embedded as the certificate's Common Name.

func GenerateServerCert 

func GenerateServerCert(caCertPEM, caKeyPEM []byte, hosts []string, validFor time.Duration) (certPEM, keyPEM []byte, err error)

GenerateServerCert creates a server certificate signed by the given CA.

func ServerTLSConfig 

func ServerTLSConfig(cfg MTLSConfig) (*tls.Config, error)

ServerTLSConfig builds a *tls.Config for the relay server with mTLS. The server presents its own cert and requires clients to present a CA-signed cert.

func WriteCertFiles 

func WriteCertFiles(certPath, keyPath string, certPEM, keyPEM []byte) error

WriteCertFiles writes cert and key PEM files to disk.

Types

type Agent 

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

Agent runs on each fleet node, maintaining an outbound connection to the relay.

func NewAgent 

func NewAgent(config AgentConfig, executor LocalExecutor, logger *slog.Logger) *Agent

NewAgent creates a node relay agent.

func (*Agent) IsConnected 

func (a *Agent) IsConnected() bool

IsConnected returns whether the agent has an active relay connection.

func (*Agent) Run 

func (a *Agent) Run(ctx context.Context) error

Run connects to the relay and processes commands. It reconnects automatically. This is the main loop for a fleet node agent.

func (*Agent) Stop 

func (a *Agent) Stop()

Stop gracefully stops the agent.

type AgentConfig 

type AgentConfig struct {
	RelayAddr         string        `json:"relay_addr"` // e.g., "relay.example.com:9443"
	NodeID            fleet.NodeID  `json:"node_id"`
	AuthToken         string        `json:"auth_token"` // legacy, prefer mTLS
	TLSConfig         *tls.Config   `json:"tls_config,omitempty"`
	MTLS              *MTLSConfig   `json:"mtls,omitempty"` // mTLS config (replaces AuthToken)
	ReconnectInterval time.Duration `json:"reconnect_interval"`
	HeartbeatInterval time.Duration `json:"heartbeat_interval"`
}

AgentConfig configures the node-side relay agent.

type ClientIdentity 

type ClientIdentity struct {
	NodeID       string    `json:"node_id"`
	Fingerprint  string    `json:"fingerprint"`
	Organization string    `json:"organization"`
	ValidUntil   time.Time `json:"valid_until"`
}

ClientIdentity holds verified identity from a client certificate.

func VerifyClientCert 

func VerifyClientCert(state *tls.ConnectionState) (*ClientIdentity, error)

VerifyClientCert checks that the client certificate is valid and extracts identity information. Used in the WebSocket handler.

type ClusterState 

type ClusterState struct {
	Self             *PeerState   `json:"self"`
	Peers            []*PeerState `json:"peers"`
	TotalNodes       int          `json:"total_nodes"`
	TotalInstances   int          `json:"total_instances"`
	HealthyInstances int          `json:"healthy_instances"`
}

ClusterState represents the complete HA cluster state.

type CommandEnvelope 

type CommandEnvelope struct {
	RequestID string
	Command   fleet.TypedCommand
	Deadline  time.Time
}

CommandEnvelope wraps a command with routing metadata.

type HAConfig 

type HAConfig struct {
	Enabled        bool          `json:"enabled"`
	InstanceID     string        `json:"instance_id"`     // Unique ID for this relay instance
	PeerAddrs      []string      `json:"peer_addrs"`      // Addresses of peer relay instances
	AdvertiseAddr  string        `json:"advertise_addr"`  // Address this instance advertises to peers
	HealthInterval time.Duration `json:"health_interval"` // How often to check peer health
	DrainTimeout   time.Duration `json:"drain_timeout"`   // How long to wait for connections to drain
}

HAConfig configures the relay HA cluster.

type HACoordinator 

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

HACoordinator manages relay HA cluster coordination.

func NewHACoordinator 

func NewHACoordinator(config HAConfig, server *WSServer, logger *slog.Logger) *HACoordinator

NewHACoordinator creates a new HA coordinator for the relay server.

func (*HACoordinator) ClusterStatus 

func (ha *HACoordinator) ClusterStatus() *ClusterState

ClusterStatus returns the current HA cluster state.

func (*HACoordinator) PreferredInstance 

func (ha *HACoordinator) PreferredInstance(nodeID string) string

PreferredInstance returns the preferred relay instance for a given node ID using consistent hashing. This helps distribute nodes across relay instances.

func (*HACoordinator) ShouldAcceptNode 

func (ha *HACoordinator) ShouldAcceptNode(nodeID string) bool

ShouldAcceptNode returns true if this relay instance should accept the given node, based on consistent hashing.

func (*HACoordinator) Start 

func (ha *HACoordinator) Start(ctx context.Context) error

Start begins HA coordination — peer health checking and status broadcasting.

func (*HACoordinator) Stop 

func (ha *HACoordinator) Stop(ctx context.Context) error

Stop gracefully stops the HA coordinator, draining connections.

type LocalExecutor 

type LocalExecutor interface {
	Execute(ctx context.Context, cmd fleet.TypedCommand) (*fleet.NodeResult, error)
}

LocalExecutor executes commands locally on the fleet node.

type MTLSConfig 

type MTLSConfig struct {
	// Server-side
	CACertFile     string `json:"ca_cert_file"`     // Path to CA certificate (PEM)
	ServerCertFile string `json:"server_cert_file"` // Path to server certificate (PEM)
	ServerKeyFile  string `json:"server_key_file"`  // Path to server private key (PEM)

	// Client-side (node agent)
	ClientCertFile string `json:"client_cert_file"` // Path to node certificate (PEM)
	ClientKeyFile  string `json:"client_key_file"`  // Path to node private key (PEM)

	// Policy
	RequireClientCert  bool `json:"require_client_cert"`  // Reject connections without valid client cert
	AllowTokenFallback bool `json:"allow_token_fallback"` // Allow bearer token if no client cert (migration mode)
}

MTLSConfig configures mutual TLS for the relay.

type PeerState 

type PeerState struct {
	InstanceID     string    `json:"instance_id"`
	Address        string    `json:"address"`
	Status         string    `json:"status"` // "healthy", "unhealthy", "draining"
	ConnectedNodes int       `json:"connected_nodes"`
	LastSeen       time.Time `json:"last_seen"`
	LastError      string    `json:"last_error,omitempty"`
}

PeerState tracks the health of a relay peer.

type ResultEnvelope 

type ResultEnvelope struct {
	RequestID string
	Result    fleet.NodeResult
}

ResultEnvelope wraps a result with routing metadata.

type Server 

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

Server is the relay server that brokers connections between the control plane and fleet nodes.

func NewServer 

func NewServer(config ServerConfig, store fleet.Store, logger *slog.Logger) *Server

NewServer creates a relay server.

func (*Server) ConnectedNodes 

func (s *Server) ConnectedNodes() []fleet.NodeID

ConnectedNodes returns the list of currently connected node IDs.

func (*Server) DeregisterTunnel 

func (s *Server) DeregisterTunnel(nodeID fleet.NodeID)

DeregisterTunnel removes a node's tunnel.

func (*Server) RegisterTunnel 

func (s *Server) RegisterTunnel(nodeID fleet.NodeID, remoteAddr string) (*Tunnel, error)

RegisterTunnel is called when a node agent connects.

func (*Server) SendCommand 

func (s *Server) SendCommand(ctx context.Context, nodeID fleet.NodeID, env *CommandEnvelope) (*ResultEnvelope, error)

SendCommand routes a command to a node through its tunnel.

type ServerConfig 

type ServerConfig struct {
	ListenAddr   string        `json:"listen_addr"` // e.g., ":9443"
	TLSConfig    *tls.Config   `json:"tls_config,omitempty"`
	AuthToken    string        `json:"auth_token"` // shared secret for node auth (legacy, prefer mTLS)
	MaxNodes     int           `json:"max_nodes"`
	PingInterval time.Duration `json:"ping_interval"`
	MTLS         *MTLSConfig   `json:"mtls,omitempty"` // mTLS config (replaces AuthToken)
}

ServerConfig configures the relay server.

type ShellExecutor 

type ShellExecutor struct {
	WorkDir      string
	DenyPatterns []string
}

ShellExecutor implements LocalExecutor by running shell commands locally.

func NewShellExecutor 

func NewShellExecutor(workDir string) *ShellExecutor

NewShellExecutor creates a local shell command executor.

func (*ShellExecutor) Execute 

func (e *ShellExecutor) Execute(ctx context.Context, cmd fleet.TypedCommand) (*fleet.NodeResult, error)

Execute runs a typed command locally on this node.

type Tunnel 

type Tunnel struct {
	NodeID      fleet.NodeID
	ConnectedAt time.Time
	LastPing    time.Time
	RemoteAddr  string
	CommandCh   chan *CommandEnvelope // commands to send to node
	ResultCh    chan *ResultEnvelope  // results from node
	Done        chan struct{}
}

Tunnel represents a persistent connection from a node to the relay.

type TunnelRelayClient 

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

TunnelRelayClient routes commands through the relay server.

func NewTunnelRelayClient 

func NewTunnelRelayClient(server *Server, logger *slog.Logger) *TunnelRelayClient

NewTunnelRelayClient creates a relay-based fleet client.

func (*TunnelRelayClient) Execute 

func (c *TunnelRelayClient) Execute(ctx context.Context, node *fleet.Node, cmd fleet.TypedCommand) (*fleet.NodeResult, error)

Execute sends a command to a node through the relay tunnel.

func (*TunnelRelayClient) Ping 

func (c *TunnelRelayClient) Ping(ctx context.Context, node *fleet.Node) error

Ping checks if a node is reachable through the relay.

type WSAgent 

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

WSAgent runs on each fleet node, connecting outbound to the relay server.

func NewWSAgent 

func NewWSAgent(config AgentConfig, executor LocalExecutor, logger *slog.Logger) *WSAgent

NewWSAgent creates a WebSocket-based node agent.

func (*WSAgent) IsConnected 

func (a *WSAgent) IsConnected() bool

IsConnected returns whether the agent has an active relay connection.

func (*WSAgent) Run 

func (a *WSAgent) Run(ctx context.Context) error

Run connects to the relay and processes commands with automatic reconnection.

func (*WSAgent) Stop 

func (a *WSAgent) Stop()

Stop gracefully stops the agent.

type WSMessage 

type WSMessage struct {
	Type      string          `json:"type"` // "register", "command", "result", "ping", "pong", "error"
	RequestID string          `json:"request_id,omitempty"`
	NodeID    string          `json:"node_id,omitempty"`
	Payload   json.RawMessage `json:"payload,omitempty"`
	Error     string          `json:"error,omitempty"`
	Timestamp time.Time       `json:"ts"`
}

WSMessage is the wire format for relay messages.

type WSRelayClient 

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

WSRelayClient routes commands through the WebSocket relay server.

func NewWSRelayClient 

func NewWSRelayClient(server *WSServer, logger *slog.Logger) *WSRelayClient

NewWSRelayClient creates a relay client backed by the WS server.

func (*WSRelayClient) Execute 

func (c *WSRelayClient) Execute(ctx context.Context, node *fleet.Node, cmd fleet.TypedCommand) (*fleet.NodeResult, error)

Execute sends a command to a node through the relay tunnel.

func (*WSRelayClient) Ping 

func (c *WSRelayClient) Ping(ctx context.Context, node *fleet.Node) error

Ping checks if a node is connected to the relay.

type WSServer 

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

WSServer is a WebSocket-based relay server that brokers connections between the control plane (CLI/SDK) and fleet node agents. Nodes connect outbound via WSS — no inbound ports required on nodes.

func NewWSServer 

func NewWSServer(config ServerConfig, store fleet.Store, logger *slog.Logger) *WSServer

NewWSServer creates a WebSocket relay server.

func (*WSServer) ConnectedNodeIDs 

func (s *WSServer) ConnectedNodeIDs() []fleet.NodeID

ConnectedNodeIDs returns the list of currently connected node IDs.

func (*WSServer) SendCommandWS 

func (s *WSServer) SendCommandWS(ctx context.Context, nodeID fleet.NodeID, env *CommandEnvelope) (*ResultEnvelope, error)

SendCommandWS sends a command to a node through its WebSocket tunnel.

func (*WSServer) Start 

func (s *WSServer) Start(ctx context.Context) error

Start starts the WebSocket relay server.

func (*WSServer) Stop 

func (s *WSServer) Stop(ctx context.Context) error

Stop gracefully shuts down the relay server.

type WSTunnel 

type WSTunnel struct {
	NodeID      fleet.NodeID
	Conn        *websocket.Conn
	ConnectedAt time.Time
	LastPing    time.Time
	RemoteAddr  string
	// contains filtered or unexported fields
}

WSTunnel is a WebSocket connection from a node agent to the relay.

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.