clustergateway

package
v1.1.3 Latest Latest
Warning

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

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

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type AgentConnection 

type AgentConnection struct {
	ID              string // Unique connection identifier
	Conn            Connection
	PlaneType       string // e.g., "dataplane", "workflowplane", "observabilityplane"
	PlaneID         string // Logical plane identifier (shared across CRs with same physical plane)
	PlaneIdentifier string // Simplified identifier: {planeType}/{planeID}
	ConnectedAt     time.Time
	LastSeen        time.Time
	ValidCRs        []string // List of CRs (namespace/name) this connection is authorized for
	// contains filtered or unexported fields
}

AgentConnection represents an active agent connection Multiple agent replicas for the same plane can share the same PlaneIdentifier for HA One agent per physical plane handles multiple CRs with the same planeID

func (*AgentConnection) AddValidCR added in v0.14.0 

func (ac *AgentConnection) AddValidCR(crKey string)

AddValidCR adds a CR to the ValidCRs list if not already present

func (*AgentConnection) Close 

func (ac *AgentConnection) Close() error

Close closes the agent connection

func (*AgentConnection) GetValidCRs added in v0.14.0 

func (ac *AgentConnection) GetValidCRs() []string

GetValidCRs returns a copy of the ValidCRs list

func (*AgentConnection) IsValidForCR added in v0.14.0 

func (ac *AgentConnection) IsValidForCR(crKey string) bool

IsValidForCR checks if this connection is authorized for the specified CR

func (*AgentConnection) RemoveValidCR added in v0.14.0 

func (ac *AgentConnection) RemoveValidCR(crKey string)

RemoveValidCR removes a CR from the ValidCRs list

func (*AgentConnection) SendHTTPTunnelRequest added in v0.7.0 

func (ac *AgentConnection) SendHTTPTunnelRequest(req *messaging.HTTPTunnelRequest) error

SendHTTPTunnelRequest sends an HTTPTunnelRequest through this connection

func (*AgentConnection) SendRawMessage added in v1.1.0 

func (ac *AgentConnection) SendRawMessage(data []byte) error

SendRawMessage sends a pre-marshaled JSON message through this connection.

func (*AgentConnection) SetValidCRs added in v0.14.0 

func (ac *AgentConnection) SetValidCRs(validCRs []string)

SetValidCRs replaces the ValidCRs list (used during re-validation)

func (*AgentConnection) UpdateCRValidity added in v0.14.0 

func (ac *AgentConnection) UpdateCRValidity(crKey string, certPool *x509.CertPool) (granted bool, revoked bool, err error)

UpdateCRValidity validates the connection's client certificate against the given CA pool and updates the CR's authorization status accordingly. Returns (granted=true) if authorization was newly granted, (revoked=true) if authorization was revoked. Returns error if certificate verification fails and authorization is revoked.

type AllPlaneStatusResponse added in v0.14.0 

type AllPlaneStatusResponse struct {
	Planes []PlaneConnectionStatus `json:"planes"`
	Total  int                     `json:"total"`
}

AllPlaneStatusResponse represents the response for all plane statuses

type Config 

type Config struct {
	Port           int
	InternalPort   int
	ServerCertPath string
	ServerKeyPath  string
	// SkipClientCertVerify has no effect: agent client certificates are always
	// verified per plane CR on the public listener, and internal listener
	// verification is controlled by InternalMTLSEnabled.
	//
	// Deprecated: will be removed in a future release.
	SkipClientCertVerify bool
	// InternalMTLSEnabled requires and verifies client certificates on the
	// internal API listener (/api/*) against InternalClientCAPath.
	InternalMTLSEnabled bool
	// InternalClientCAPath is the path to the CA bundle used to verify
	// internal API clients. Required when InternalMTLSEnabled is true.
	InternalClientCAPath string
	ReadTimeout          time.Duration
	WriteTimeout         time.Duration
	IdleTimeout          time.Duration
	ShutdownTimeout      time.Duration
	HeartbeatInterval    time.Duration
	HeartbeatTimeout     time.Duration
}

Config holds configuration for the agent server

type Connection added in v1.1.0 

type Connection interface {
	ReadMessage() (messageType int, p []byte, err error)
	WriteMessage(messageType int, data []byte) error
	WriteControl(messageType int, data []byte, deadline time.Time) error
	SetReadDeadline(t time.Time) error
	SetPongHandler(h func(appData string) error)
	Close() error
}

Connection abstracts a WebSocket connection for testability. *websocket.Conn satisfies this interface.

type ConnectionManager 

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

ConnectionManager manages active agent connections Supports multiple concurrent connections per plane identifier for HA One agent per physical plane (planeID) handles multiple CRs

func NewConnectionManager 

func NewConnectionManager(logger *slog.Logger) *ConnectionManager

NewConnectionManager creates a new ConnectionManager

func (*ConnectionManager) Count 

func (cm *ConnectionManager) Count() int

Count returns the total number of active connections across all planes

func (*ConnectionManager) DisconnectAllForPlane added in v0.9.0 

func (cm *ConnectionManager) DisconnectAllForPlane(planeType, planeID string) int

DisconnectAllForPlane disconnects all agent connections for a specific planeType/planeID This is used when a CR is deleted or updated to force agents to reconnect with new configuration

func (*ConnectionManager) Get 

func (cm *ConnectionManager) Get(planeIdentifier string) (*AgentConnection, error)

Get retrieves an agent connection by plane identifier using round-robin selection If multiple connections exist for the plane, it rotates between them

func (*ConnectionManager) GetAll 

func (cm *ConnectionManager) GetAll() []*AgentConnection

func (*ConnectionManager) GetAllPlaneStatuses added in v0.12.0 

func (cm *ConnectionManager) GetAllPlaneStatuses() []PlaneConnectionStatus

GetAllPlaneStatuses returns connection status for all planes

func (*ConnectionManager) GetCRAuthorizationStatus added in v0.14.0 

func (cm *ConnectionManager) GetCRAuthorizationStatus(planeType, planeID, namespace, name string) *PlaneConnectionStatus

GetCRAuthorizationStatus returns authorization status for a specific CR within a plane This is different from GetPlaneStatus which only checks if agents are connected to the plane This method checks if the connected agents are actually authorized for the specific CR (namespace/name)

func (*ConnectionManager) GetForCR added in v0.14.0 

func (cm *ConnectionManager) GetForCR(planeIdentifier, crKey string) (*AgentConnection, error)

GetForCR retrieves an agent connection authorized for the specified CR using round-robin selection Only connections where the agent's certificate is valid for the requested CR are considered This enforces per-CR security boundaries in multi-tenant scenarios Returns error if no authorized connections are found

func (*ConnectionManager) GetPlaneStatus added in v0.12.0 

func (cm *ConnectionManager) GetPlaneStatus(planeType, planeID string) *PlaneConnectionStatus

GetPlaneStatus returns connection status for a specific plane

func (*ConnectionManager) Register 

func (cm *ConnectionManager) Register(
	planeType, planeID string,
	conn Connection,
	validCRs []string,
	clientCert *x509.Certificate,
) (string, error)

Register registers a new agent connection with per-CR authorization planeIdentifier format: {planeType}/{planeID} Multiple agent replicas (for HA) for the same plane share the same planeIdentifier validCRs contains the list of CRs (namespace/name) this connection is authorized for clientCert is stored for re-validation when CRs are updated Returns the connection ID which should be used for unregistration

func (*ConnectionManager) RevalidateCR added in v0.14.0 

func (cm *ConnectionManager) RevalidateCR(
	planeType, planeID, crNamespace, crName string,
	newCAData []byte,
) (updated int, removed int, err error)

RevalidateCR re-validates all connections for a specific CR without disconnecting This is called when a CR is updated (e.g., certificate rotation) to enforce new security policy without causing service disruption. Returns counts of authorizations granted and revoked.

func (*ConnectionManager) SendHTTPTunnelRequest added in v0.7.0 

func (cm *ConnectionManager) SendHTTPTunnelRequest(planeIdentifier string, req *messaging.HTTPTunnelRequest) error

func (*ConnectionManager) Unregister 

func (cm *ConnectionManager) Unregister(planeIdentifier, connID string)

func (*ConnectionManager) UpdateConnectionLastSeen 

func (cm *ConnectionManager) UpdateConnectionLastSeen(planeIdentifier, connID string)

UpdateConnectionLastSeen updates the last seen time for a specific connection

type PlaneAPI added in v0.9.0 

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

func NewPlaneAPI added in v0.9.0 

func NewPlaneAPI(connMgr *ConnectionManager, server *Server, logger *slog.Logger) *PlaneAPI

func (*PlaneAPI) RegisterRoutes added in v0.9.0 

func (api *PlaneAPI) RegisterRoutes(mux *http.ServeMux)

type PlaneConnectionStatus added in v0.12.0 

type PlaneConnectionStatus struct {
	PlaneType       string    `json:"planeType"`
	PlaneID         string    `json:"planeID"`
	Connected       bool      `json:"connected"`
	ConnectedAgents int       `json:"connectedAgents"`
	LastSeen        time.Time `json:"lastSeen,omitempty"`
}

PlaneConnectionStatus holds connection status for a specific plane

type PlaneNotification added in v0.9.0 

type PlaneNotification struct {
	PlaneType string `json:"planeType"` // "dataplane", "workflowplane", "observabilityplane"
	PlaneID   string `json:"planeID"`
	Event     string `json:"event"` // "created", "updated", "deleted"
	Namespace string `json:"namespace"`
	Name      string `json:"name"`
}

type PlaneNotificationResponse added in v0.14.0 

type PlaneNotificationResponse struct {
	Success               bool   `json:"success"`
	Action                string `json:"action"` // "disconnect", "disconnect_fallback", "revalidate"
	DisconnectedAgents    *int   `json:"disconnectedAgents,omitempty"`
	AuthorizationsGranted *int   `json:"authorizationsGranted,omitempty"`
	AuthorizationsRevoked *int   `json:"authorizationsRevoked,omitempty"`
	Error                 string `json:"error,omitempty"`
}

PlaneNotificationResponse represents the response to a plane notification

type PlaneReconnectResponse added in v0.14.0 

type PlaneReconnectResponse struct {
	Success            bool `json:"success"`
	DisconnectedAgents int  `json:"disconnectedAgents"`
}

PlaneReconnectResponse represents the response to a manual reconnect request

type RemoteServerClientConfig 

type RemoteServerClientConfig struct {
	// ServerURL is the URL of the agent server (e.g., https://cluster-agent-server:8443)
	ServerURL string

	// InsecureSkipVerify disables TLS certificate verification (development only)
	InsecureSkipVerify bool

	// ServerCAPath is the path to the CA certificate for verifying the server's certificate
	// If empty and InsecureSkipVerify is false, system CA pool will be used
	ServerCAPath string

	// ClientCertPath is the path to the client certificate for mTLS (optional)
	ClientCertPath string

	// ClientKeyPath is the path to the client private key for mTLS (optional)
	ClientKeyPath string

	// Timeout is the HTTP client timeout
	Timeout time.Duration
}

RemoteServerClientConfig holds configuration for RemoteServerClient

type RequestValidator added in v0.7.0 

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

func NewRequestValidator added in v0.7.0 

func NewRequestValidator() *RequestValidator

func (*RequestValidator) AllowTarget added in v0.7.0 

func (v *RequestValidator) AllowTarget(target string)

func (*RequestValidator) BlockPath added in v0.7.0 

func (v *RequestValidator) BlockPath(path string)

func (*RequestValidator) SetMaxRequestBodySize added in v0.7.0 

func (v *RequestValidator) SetMaxRequestBodySize(size int64)

func (*RequestValidator) ValidateRequest added in v0.7.0 

func (v *RequestValidator) ValidateRequest(r *http.Request, target, path string) error

type Server 

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

func New 

func New(config *Config, k8sClient client.Client, logger *slog.Logger) *Server

func (*Server) GetConnectionManager 

func (s *Server) GetConnectionManager() *ConnectionManager

func (*Server) SendHTTPTunnelRequest added in v0.7.0 

func (s *Server) SendHTTPTunnelRequest(planeName string, req *messaging.HTTPTunnelRequest, timeout time.Duration) (*messaging.HTTPTunnelResponse, error)

SendHTTPTunnelRequest sends an HTTP tunnel request to an agent and waits for the response

func (*Server) SendHTTPTunnelRequestForCR added in v0.14.0 

func (s *Server) SendHTTPTunnelRequestForCR(
	planeName, crKey string,
	req *messaging.HTTPTunnelRequest,
	timeout time.Duration,
) (*messaging.HTTPTunnelResponse, error)

SendHTTPTunnelRequestForCR sends an HTTP tunnel request to an agent authorized for a specific CR and waits for the response. This enforces per-CR security boundaries.

func (*Server) Start 

func (s *Server) Start() error

type ValidationError added in v0.7.0 

type ValidationError struct {
	Code    int
	Message string
}

func (*ValidationError) Error added in v0.7.0 

func (e *ValidationError) Error() 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.