Documentation
¶
Index ¶
- Constants
- func ApplyServiceLabels(c *cmodel.Container, serviceLabels map[string]string)
- func ComputeClusterHealth(services []*SwarmService, nodes []*SwarmNode) string
- func IsSwarmManaged(labels map[string]string) bool
- func StackName(labels map[string]string) string
- type CrashLoopDetector
- func (cld *CrashLoopDetector) CheckRecoveries()
- func (cld *CrashLoopDetector) IsCrashLooping(serviceID string) bool
- func (cld *CrashLoopDetector) RecordFailure(serviceID, serviceName, lastError string)
- func (cld *CrashLoopDetector) SetAlertCallback(cb NodeAlertCallback)
- func (cld *CrashLoopDetector) SetEventCallback(cb EventCallback)
- type DetectionResult
- type Detector
- type EventCallback
- type EventProcessor
- type InfoProvider
- type IngestService
- func (s *IngestService) HandleAgentEvent(ctx context.Context, agentID string, ev *agentpb.SwarmTopology) error
- func (s *IngestService) Reconcile(ctx context.Context, agentID string, snap TopologySnapshot) error
- func (s *IngestService) ReconcileServicesTasks(ctx context.Context, agentID string, snap TopologySnapshot) error
- func (s *IngestService) SetBroadcaster(fn func(eventType string, data any))
- type NetworkAttachment
- type NetworkResolver
- type NodeAlertCallback
- type NodeReconciler
- type NodeService
- type NodeStore
- type PortConfig
- type ReplicaHealthChecker
- type ServiceClient
- type ServiceDiscovery
- func (sd *ServiceDiscovery) DiscoverAll(ctx context.Context) ([]*cmodel.Container, []*SwarmService, error)
- func (sd *ServiceDiscovery) GetService(serviceID string) *SwarmService
- func (sd *ServiceDiscovery) GetTasksForService(serviceID string) []*SwarmTask
- func (sd *ServiceDiscovery) ListServices() []*SwarmService
- func (sd *ServiceDiscovery) RefreshService(ctx context.Context, serviceID string) (*SwarmService, error)
- func (sd *ServiceDiscovery) RemoveService(serviceID string)
- func (sd *ServiceDiscovery) SetNetworkResolver(resolver NetworkResolver)
- type ServiceTaskStore
- type SwarmCluster
- type SwarmNode
- type SwarmService
- type SwarmTask
- type TaskPlacement
- type TaskTracker
- type TopologySnapshot
- type UpdateProgress
- type UpdateStatus
- type UpdateTracker
- func (ut *UpdateTracker) CheckService(ctx context.Context, serviceID string)
- func (ut *UpdateTracker) GetUpdateStatus(ctx context.Context, serviceID string) (*UpdateProgress, error)
- func (ut *UpdateTracker) SetAlertCallback(cb NodeAlertCallback)
- func (ut *UpdateTracker) SetEventCallback(cb EventCallback)
Constants ¶
const ClusterHealthDegraded = "degraded"
ClusterHealthDegraded indicates some nodes are drained/paused or services are under-replicated.
const ClusterHealthHealthy = "healthy"
ClusterHealthHealthy indicates all nodes are ready and all services fully replicated.
const ClusterHealthUnhealthy = "unhealthy"
ClusterHealthUnhealthy indicates nodes are down/disconnected or services have zero running replicas.
Variables ¶
This section is empty.
Functions ¶
func ApplyServiceLabels ¶
ApplyServiceLabels maps Swarm service-level labels to Container model fields. This applies maintenant.* labels from the service definition to the container.
func ComputeClusterHealth ¶
func ComputeClusterHealth(services []*SwarmService, nodes []*SwarmNode) string
ComputeClusterHealth computes the overall cluster health from services and nodes. Priority: unhealthy > degraded > healthy.
func IsSwarmManaged ¶
IsSwarmManaged returns true if the container has a Swarm service ID label.
Types ¶
type CrashLoopDetector ¶
type CrashLoopDetector struct {
// contains filtered or unexported fields
}
CrashLoopDetector detects crash-loop patterns per service.
func NewCrashLoopDetector ¶
func NewCrashLoopDetector(logger *slog.Logger) *CrashLoopDetector
NewCrashLoopDetector creates a new crash-loop detector.
func (*CrashLoopDetector) CheckRecoveries ¶
func (cld *CrashLoopDetector) CheckRecoveries()
CheckRecoveries checks if any services have recovered from crash-loop.
func (*CrashLoopDetector) IsCrashLooping ¶
func (cld *CrashLoopDetector) IsCrashLooping(serviceID string) bool
IsCrashLooping returns whether a service is currently in crash-loop state.
func (*CrashLoopDetector) RecordFailure ¶
func (cld *CrashLoopDetector) RecordFailure(serviceID, serviceName, lastError string)
RecordFailure records a task failure for a service and checks for crash-loop.
func (*CrashLoopDetector) SetAlertCallback ¶
func (cld *CrashLoopDetector) SetAlertCallback(cb NodeAlertCallback)
SetAlertCallback sets the alert callback.
func (*CrashLoopDetector) SetEventCallback ¶
func (cld *CrashLoopDetector) SetEventCallback(cb EventCallback)
SetEventCallback sets the SSE event callback.
type DetectionResult ¶
type DetectionResult struct {
Active bool `json:"active"`
IsManager bool `json:"is_manager"`
ClusterID string `json:"cluster_id,omitempty"`
}
DetectionResult holds the outcome of Swarm mode detection.
type Detector ¶
type Detector struct {
// contains filtered or unexported fields
}
Detector detects whether the Docker engine is part of a Swarm cluster and whether this node is a manager.
func NewDetector ¶
func NewDetector(provider InfoProvider, logger *slog.Logger) *Detector
NewDetector creates a new Swarm mode detector.
func (*Detector) Detect ¶
func (d *Detector) Detect(ctx context.Context) (DetectionResult, error)
Detect checks the Docker engine for Swarm mode status.
func (*Detector) Result ¶
func (d *Detector) Result() DetectionResult
Result returns the cached detection result.
type EventCallback ¶
type EventCallback func(eventType string, data interface{})
EventCallback is called when a Swarm event produces a domain event.
type EventProcessor ¶
type EventProcessor struct {
// contains filtered or unexported fields
}
EventProcessor handles Swarm-specific runtime events (service/node type).
func NewEventProcessor ¶
func NewEventProcessor(discovery *ServiceDiscovery, logger *slog.Logger) *EventProcessor
NewEventProcessor creates a new Swarm event processor.
func (*EventProcessor) ProcessEvent ¶
func (ep *EventProcessor) ProcessEvent(ctx context.Context, evt runtime.RuntimeEvent)
ProcessEvent handles a runtime event of type "service" or "node".
func (*EventProcessor) SetAlertCallback ¶
func (ep *EventProcessor) SetAlertCallback(cb NodeAlertCallback)
SetAlertCallback sets the alert callback for replica health alerts.
func (*EventProcessor) SetCallback ¶
func (ep *EventProcessor) SetCallback(cb EventCallback)
SetCallback sets the event callback for broadcasting SSE events.
func (*EventProcessor) SetNodeService ¶
func (ep *EventProcessor) SetNodeService(ns *NodeService)
SetNodeService sets the node service for routing node events.
type InfoProvider ¶
InfoProvider abstracts docker info retrieval for the Swarm detector.
type IngestService ¶ added in v1.3.0
type IngestService struct {
// contains filtered or unexported fields
}
IngestService reconciles a swarm topology snapshot reported by an agent (or by the server's own local runtime under the LocalAgent id) into the store. It implements agentserver.SwarmTopologyHandler.
func NewIngestService ¶ added in v1.3.0
func NewIngestService(store ServiceTaskStore, nodes NodeReconciler, logger *slog.Logger) *IngestService
NewIngestService wires the swarm topology ingest service.
func (*IngestService) HandleAgentEvent ¶ added in v1.3.0
func (s *IngestService) HandleAgentEvent(ctx context.Context, agentID string, ev *agentpb.SwarmTopology) error
HandleAgentEvent reconciles a full swarm topology snapshot reported by a remote agent over gRPC.
func (*IngestService) Reconcile ¶ added in v1.3.0
func (s *IngestService) Reconcile(ctx context.Context, agentID string, snap TopologySnapshot) error
Reconcile hard-reconciles a domain snapshot for agentID into the store. Used directly by the server's local-runtime reconcile loop (under LocalAgent) and indirectly by HandleAgentEvent for remote agents.
func (*IngestService) ReconcileServicesTasks ¶ added in v1.3.0
func (s *IngestService) ReconcileServicesTasks(ctx context.Context, agentID string, snap TopologySnapshot) error
ReconcileServicesTasks reconciles only the services and tasks of a snapshot, leaving nodes untouched. Used by the server's local reconcile loop, where the Pro NodeService owns the local swarm_nodes rows; reconciling nodes here too would let the two writers fight over the same LocalAgent rows.
func (*IngestService) SetBroadcaster ¶ added in v1.3.0
func (s *IngestService) SetBroadcaster(fn func(eventType string, data any))
SetBroadcaster wires an SSE broadcaster. When set, a swarm.topology_changed event carrying the agent id is emitted after a remote agent's snapshot is reconciled, so connected clients scoped to that agent refetch.
type NetworkAttachment ¶
type NetworkAttachment struct {
NetworkID string `json:"network_id"`
NetworkName string `json:"network_name"`
Scope string `json:"scope"` // "swarm", "local"
}
NetworkAttachment represents a network attached to a Swarm service.
type NetworkResolver ¶
NetworkResolver resolves network IDs to names and scopes.
type NodeAlertCallback ¶
NodeAlertCallback is called when a node health alert is detected.
type NodeReconciler ¶ added in v1.3.0
type NodeReconciler interface {
ReplaceNodesForAgent(ctx context.Context, agentID string, nodes []*SwarmNode) error
}
NodeReconciler persists a full per-agent node set (SQLite-backed).
type NodeService ¶
type NodeService struct {
// contains filtered or unexported fields
}
NodeService monitors Swarm node health and detects status transitions.
func NewNodeService ¶
func NewNodeService(client ServiceClient, store NodeStore, logger *slog.Logger) *NodeService
NewNodeService creates a new node health monitoring service.
func (*NodeService) Reconcile ¶
func (ns *NodeService) Reconcile(ctx context.Context) error
Reconcile fetches the current node list from Docker, reconciles with the store, and detects status transitions that should trigger alerts.
func (*NodeService) SetAlertCallback ¶
func (ns *NodeService) SetAlertCallback(cb NodeAlertCallback)
SetAlertCallback sets the alert callback for node health alerts.
func (*NodeService) SetEventCallback ¶
func (ns *NodeService) SetEventCallback(cb EventCallback)
SetEventCallback sets the SSE event callback for node status changes.
type NodeStore ¶
type NodeStore interface {
UpsertNode(ctx context.Context, node *SwarmNode) error
ListNodes(ctx context.Context, agentID string) ([]*SwarmNode, error)
GetNodeByNodeID(ctx context.Context, nodeID string) (*SwarmNode, error)
UpdateNodeStatus(ctx context.Context, nodeID, status, availability string) error
UpdateNodeTaskCount(ctx context.Context, nodeID string, count int) error
}
NodeStore abstracts persistence for swarm nodes.
type PortConfig ¶
type PortConfig struct {
Protocol string `json:"protocol"` // "tcp", "udp", "sctp"
TargetPort uint32 `json:"target_port"` // container port
PublishedPort uint32 `json:"published_port"` // host/ingress port
PublishMode string `json:"publish_mode"` // "ingress" or "host"
}
PortConfig represents a published port on a Swarm service.
type ReplicaHealthChecker ¶
type ReplicaHealthChecker struct {
// contains filtered or unexported fields
}
ReplicaHealthChecker detects sustained under-replication and emits alerts after a configurable delay (default 5 minutes).
func NewReplicaHealthChecker ¶
func NewReplicaHealthChecker(logger *slog.Logger) *ReplicaHealthChecker
NewReplicaHealthChecker creates a new replica health checker.
func (*ReplicaHealthChecker) Check ¶
func (rhc *ReplicaHealthChecker) Check(services []*SwarmService)
Check evaluates replica health for all services and emits alerts for services that have been under-replicated beyond the alert delay.
func (*ReplicaHealthChecker) SetAlertCallback ¶
func (rhc *ReplicaHealthChecker) SetAlertCallback(cb NodeAlertCallback)
SetAlertCallback sets the alert callback.
func (*ReplicaHealthChecker) SetAlertDelay ¶
func (rhc *ReplicaHealthChecker) SetAlertDelay(d time.Duration)
SetAlertDelay overrides the default alert delay for testing.
func (*ReplicaHealthChecker) SetEventCallback ¶
func (rhc *ReplicaHealthChecker) SetEventCallback(cb EventCallback)
SetEventCallback sets the SSE event callback.
type ServiceClient ¶
type ServiceClient interface {
ServiceList(ctx context.Context) ([]swarm.Service, error)
ServiceInspect(ctx context.Context, serviceID string) (swarm.Service, error)
TaskList(ctx context.Context) ([]swarm.Task, error)
NodeList(ctx context.Context) ([]swarm.Node, error)
}
ServiceClient abstracts Docker SDK calls needed for Swarm service discovery.
type ServiceDiscovery ¶
type ServiceDiscovery struct {
// contains filtered or unexported fields
}
ServiceDiscovery handles Swarm service discovery and mapping to Container models.
func NewServiceDiscovery ¶
func NewServiceDiscovery(client ServiceClient, logger *slog.Logger) *ServiceDiscovery
NewServiceDiscovery creates a new Swarm service discovery instance.
func (*ServiceDiscovery) DiscoverAll ¶
func (sd *ServiceDiscovery) DiscoverAll(ctx context.Context) ([]*cmodel.Container, []*SwarmService, error)
DiscoverAll discovers all Swarm services and maps them to Container models. Returns containers representing Swarm tasks with Swarm fields populated.
func (*ServiceDiscovery) GetService ¶
func (sd *ServiceDiscovery) GetService(serviceID string) *SwarmService
GetService returns a cached Swarm service by ID.
func (*ServiceDiscovery) GetTasksForService ¶
func (sd *ServiceDiscovery) GetTasksForService(serviceID string) []*SwarmTask
RemoveService removes a service from the cache. GetTasksForService returns tasks for a given service by querying the Docker API.
func (*ServiceDiscovery) ListServices ¶
func (sd *ServiceDiscovery) ListServices() []*SwarmService
ListServices returns all cached Swarm services.
func (*ServiceDiscovery) RefreshService ¶
func (sd *ServiceDiscovery) RefreshService(ctx context.Context, serviceID string) (*SwarmService, error)
RefreshService re-inspects a single service and updates the cache.
func (*ServiceDiscovery) RemoveService ¶
func (sd *ServiceDiscovery) RemoveService(serviceID string)
func (*ServiceDiscovery) SetNetworkResolver ¶
func (sd *ServiceDiscovery) SetNetworkResolver(resolver NetworkResolver)
SetNetworkResolver sets the function used to resolve network names from IDs.
type ServiceTaskStore ¶ added in v1.3.0
type ServiceTaskStore interface {
ReplaceServicesForAgent(ctx context.Context, agentID string, services []SwarmService) error
ReplaceTasksForAgent(ctx context.Context, agentID string, tasks []SwarmTask) error
}
ServiceTaskStore persists per-agent swarm services and tasks (SQLite-backed).
type SwarmCluster ¶
type SwarmCluster struct {
ID string `json:"cluster_id"`
CreatedAt time.Time `json:"created_at"`
ManagerCount int `json:"manager_count"`
WorkerCount int `json:"worker_count"`
IsManager bool `json:"is_manager"`
}
SwarmCluster represents the Swarm cluster state (in-memory only, reconstructed from docker info).
type SwarmNode ¶
type SwarmNode struct {
ID string `json:"id"`
AgentID string `json:"agent_id"`
NodeID string `json:"node_id"`
Hostname string `json:"hostname"`
Role string `json:"role"` // "manager" or "worker"
Status string `json:"status"` // "ready", "down", "disconnected", "unknown"
Availability string `json:"availability"` // "active", "pause", "drain"
EngineVersion string `json:"engine_version,omitempty"`
Address string `json:"address,omitempty"`
TaskCount int `json:"task_count"`
FirstSeenAt time.Time `json:"first_seen_at"`
LastSeenAt time.Time `json:"last_seen_at"`
LastStatusChangeAt time.Time `json:"last_status_change_at"`
}
SwarmNode represents a machine in the Swarm cluster (persisted for Pro).
type SwarmService ¶
type SwarmService struct {
ServiceID string `json:"service_id"`
AgentID string `json:"agent_id,omitempty"`
Name string `json:"name"`
Image string `json:"image"`
Mode string `json:"mode"` // "replicated" or "global"
DesiredReplicas int `json:"desired_replicas"`
RunningReplicas int `json:"running_replicas"`
Labels map[string]string `json:"labels,omitempty"`
StackName string `json:"stack_name,omitempty"`
Networks []NetworkAttachment `json:"networks,omitempty"`
Ports []PortConfig `json:"ports,omitempty"`
UpdateStatus *UpdateStatus `json:"update_status,omitempty"`
CreatedAt time.Time `json:"created_at"`
}
SwarmService represents a Swarm service with aggregated state (volatile/in-memory).
type SwarmTask ¶
type SwarmTask struct {
TaskID string `json:"task_id"`
AgentID string `json:"agent_id,omitempty"`
ServiceID string `json:"service_id"`
NodeID string `json:"node_id"`
Slot int `json:"slot"`
State string `json:"state"`
DesiredState string `json:"desired_state"`
ContainerID string `json:"container_id"`
Error string `json:"error,omitempty"`
ExitCode *int `json:"exit_code,omitempty"`
Timestamp time.Time `json:"timestamp"`
NodeHostname string `json:"node_hostname,omitempty"`
}
SwarmTask represents a single instance (replica) of a service (volatile/in-memory).
type TaskPlacement ¶
type TaskPlacement struct {
ServiceID string `json:"service_id"`
ServiceName string `json:"service_name"`
ByNode map[string][]SwarmTask `json:"by_node"` // keyed by node ID
}
TaskPlacement represents the distribution of tasks across nodes for a service.
type TaskTracker ¶
type TaskTracker struct {
// contains filtered or unexported fields
}
TaskTracker aggregates task placement information from the Docker API.
func NewTaskTracker ¶
func NewTaskTracker(client ServiceClient, logger *slog.Logger) *TaskTracker
NewTaskTracker creates a new task placement tracker.
func (*TaskTracker) GetPlacementForService ¶
func (tt *TaskTracker) GetPlacementForService(ctx context.Context, serviceID, serviceName string) (*TaskPlacement, error)
GetPlacementForService returns the task distribution across nodes for a given service.
type TopologySnapshot ¶ added in v1.3.0
type TopologySnapshot struct {
Services []SwarmService
Tasks []SwarmTask
Nodes []SwarmNode
}
TopologySnapshot is a full point-in-time view of an agent's swarm: every service, task and node it currently sees. The server reconciles it against the rows already held for that agent, hard-deleting anything absent from the snapshot. Produced by both the server's local runtime (under the LocalAgent id) and remote agents (under their own id).
func SnapshotFromClient ¶ added in v1.3.0
func SnapshotFromClient(ctx context.Context, disc *ServiceDiscovery, client ServiceClient) (TopologySnapshot, error)
SnapshotFromClient builds a full topology snapshot from the live swarm API. Shared by the agent collector (which marshals it to proto and pushes it) and the server's own local-runtime reconcile (which feeds it straight to the ingest service under the LocalAgent id). disc supplies services and tasks; client supplies nodes.
type UpdateProgress ¶
type UpdateProgress struct {
ServiceID string `json:"service_id"`
ServiceName string `json:"service_name"`
State string `json:"state"`
OldImage string `json:"old_image"`
NewImage string `json:"new_image"`
TasksUpdated int `json:"tasks_updated"`
TasksTotal int `json:"tasks_total"`
Message string `json:"message"`
StartedAt *time.Time `json:"started_at,omitempty"`
CompletedAt *time.Time `json:"completed_at,omitempty"`
}
UpdateProgress represents the progress of a rolling update.
type UpdateStatus ¶
type UpdateStatus struct {
State string `json:"state"` // "updating", "paused", "completed", "rollback_started", "rollback_completed", "rollback_paused"
StartedAt *time.Time `json:"started_at,omitempty"`
CompletedAt *time.Time `json:"completed_at,omitempty"`
Message string `json:"message,omitempty"`
}
UpdateStatus represents the rolling update status of a Swarm service (Pro).
type UpdateTracker ¶
type UpdateTracker struct {
// contains filtered or unexported fields
}
UpdateTracker monitors rolling update progress for Swarm services.
func NewUpdateTracker ¶
func NewUpdateTracker(client ServiceClient, logger *slog.Logger) *UpdateTracker
NewUpdateTracker creates a new rolling update tracker.
func (*UpdateTracker) CheckService ¶
func (ut *UpdateTracker) CheckService(ctx context.Context, serviceID string)
CheckService checks the update status of a service after it was updated.
func (*UpdateTracker) GetUpdateStatus ¶
func (ut *UpdateTracker) GetUpdateStatus(ctx context.Context, serviceID string) (*UpdateProgress, error)
GetUpdateStatus returns the current update status for a service.
func (*UpdateTracker) SetAlertCallback ¶
func (ut *UpdateTracker) SetAlertCallback(cb NodeAlertCallback)
SetAlertCallback sets the alert callback.
func (*UpdateTracker) SetEventCallback ¶
func (ut *UpdateTracker) SetEventCallback(cb EventCallback)
SetEventCallback sets the SSE event callback.