Files
coder/agent/agentsocket/service.go
T
Sas Swart 2840fdcb54 feat(agent): add agent socket API (#20717)
relates to: https://github.com/coder/internal/issues/1094

This is number 2 of 5 pull requests in an effort to add agent script
ordering. It adds a drpc API that is exposed via a local socket. This
API serves access to a lightweight DAG based dependency manager that was
inspired by systemd.

In follow-up PRs:

* This unit manager will be plumbed into the workspace agent struct.
* CLI commands will use this agentsocket api to express dependencies
between coder scripts

I used an LLM to produce some of these changes, but I have conducted
thorough self review and consider this contribution to be ready for an
external reviewer.
2025-11-21 13:09:27 +02:00

143 lines
4.2 KiB
Go

package agentsocket
import (
"context"
"errors"
"golang.org/x/xerrors"
"cdr.dev/slog"
"github.com/coder/coder/v2/agent/agentsocket/proto"
"github.com/coder/coder/v2/agent/unit"
)
var _ proto.DRPCAgentSocketServer = (*DRPCAgentSocketService)(nil)
var ErrUnitManagerNotAvailable = xerrors.New("unit manager not available")
type DRPCAgentSocketService struct {
unitManager *unit.Manager
logger slog.Logger
}
func (*DRPCAgentSocketService) Ping(_ context.Context, _ *proto.PingRequest) (*proto.PingResponse, error) {
return &proto.PingResponse{}, nil
}
func (s *DRPCAgentSocketService) SyncStart(_ context.Context, req *proto.SyncStartRequest) (*proto.SyncStartResponse, error) {
if s.unitManager == nil {
return nil, xerrors.Errorf("SyncStart: %w", ErrUnitManagerNotAvailable)
}
unitID := unit.ID(req.Unit)
if err := s.unitManager.Register(unitID); err != nil {
if !errors.Is(err, unit.ErrUnitAlreadyRegistered) {
return nil, xerrors.Errorf("SyncStart: %w", err)
}
}
isReady, err := s.unitManager.IsReady(unitID)
if err != nil {
return nil, xerrors.Errorf("cannot check readiness: %w", err)
}
if !isReady {
return nil, xerrors.Errorf("cannot start unit %q: unit not ready", req.Unit)
}
err = s.unitManager.UpdateStatus(unitID, unit.StatusStarted)
if err != nil {
return nil, xerrors.Errorf("cannot start unit %q: %w", req.Unit, err)
}
return &proto.SyncStartResponse{}, nil
}
func (s *DRPCAgentSocketService) SyncWant(_ context.Context, req *proto.SyncWantRequest) (*proto.SyncWantResponse, error) {
if s.unitManager == nil {
return nil, xerrors.Errorf("cannot add dependency: %w", ErrUnitManagerNotAvailable)
}
unitID := unit.ID(req.Unit)
dependsOnID := unit.ID(req.DependsOn)
if err := s.unitManager.Register(unitID); err != nil && !errors.Is(err, unit.ErrUnitAlreadyRegistered) {
return nil, xerrors.Errorf("cannot add dependency: %w", err)
}
if err := s.unitManager.AddDependency(unitID, dependsOnID, unit.StatusComplete); err != nil {
return nil, xerrors.Errorf("cannot add dependency: %w", err)
}
return &proto.SyncWantResponse{}, nil
}
func (s *DRPCAgentSocketService) SyncComplete(_ context.Context, req *proto.SyncCompleteRequest) (*proto.SyncCompleteResponse, error) {
if s.unitManager == nil {
return nil, xerrors.Errorf("cannot complete unit: %w", ErrUnitManagerNotAvailable)
}
unitID := unit.ID(req.Unit)
if err := s.unitManager.UpdateStatus(unitID, unit.StatusComplete); err != nil {
return nil, xerrors.Errorf("cannot complete unit %q: %w", req.Unit, err)
}
return &proto.SyncCompleteResponse{}, nil
}
func (s *DRPCAgentSocketService) SyncReady(_ context.Context, req *proto.SyncReadyRequest) (*proto.SyncReadyResponse, error) {
if s.unitManager == nil {
return nil, xerrors.Errorf("cannot check readiness: %w", ErrUnitManagerNotAvailable)
}
unitID := unit.ID(req.Unit)
isReady, err := s.unitManager.IsReady(unitID)
if err != nil {
return nil, xerrors.Errorf("cannot check readiness: %w", err)
}
return &proto.SyncReadyResponse{
Ready: isReady,
}, nil
}
func (s *DRPCAgentSocketService) SyncStatus(_ context.Context, req *proto.SyncStatusRequest) (*proto.SyncStatusResponse, error) {
if s.unitManager == nil {
return nil, xerrors.Errorf("cannot get status for unit %q: %w", req.Unit, ErrUnitManagerNotAvailable)
}
unitID := unit.ID(req.Unit)
isReady, err := s.unitManager.IsReady(unitID)
if err != nil {
return nil, xerrors.Errorf("cannot check readiness: %w", err)
}
dependencies, err := s.unitManager.GetAllDependencies(unitID)
if err != nil {
return nil, xerrors.Errorf("failed to get dependencies: %w", err)
}
var depInfos []*proto.DependencyInfo
for _, dep := range dependencies {
depInfos = append(depInfos, &proto.DependencyInfo{
Unit: string(dep.Unit),
DependsOn: string(dep.DependsOn),
RequiredStatus: string(dep.RequiredStatus),
CurrentStatus: string(dep.CurrentStatus),
IsSatisfied: dep.IsSatisfied,
})
}
u, err := s.unitManager.Unit(unitID)
if err != nil {
return nil, xerrors.Errorf("cannot get status for unit %q: %w", req.Unit, err)
}
return &proto.SyncStatusResponse{
Status: string(u.Status()),
IsReady: isReady,
Dependencies: depInfos,
}, nil
}