A self-hosted orchestration system crafted for power users. composia.xyz
Something went wrong. Try again.
10 kB · 267 lines
Go
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268package controller
import ( "context" "encoding/json" "errors" "fmt" "os" "path/filepath" "strings" "sync" "time"
"connectrpc.com/connect"
controllerv1 "forgejo.alexma.top/alexma233/composia/gen/go/proto/composia/controller/v1" "forgejo.alexma.top/alexma233/composia/internal/core/config" "forgejo.alexma.top/alexma233/composia/internal/core/repo" "forgejo.alexma.top/alexma233/composia/internal/platform/store")
type serviceQueryServer struct { db *store.DB cfg *config.ControllerConfig availableNodeIDs map[string]struct{} taskQueue *taskQueueNotifier taskResults *taskResultNotifier dockerQueries *dockerQueryBroker repoMu *sync.Mutex}
func (server *serviceQueryServer) ListServices(ctx context.Context, req *connect.Request[controllerv1.ListServicesRequest]) (*connect.Response[controllerv1.ListServicesResponse], error) { if req.Msg == nil { req.Msg = &controllerv1.ListServicesRequest{} }
services, totalCount, err := server.db.ListDeclaredServices(ctx, req.Msg.GetRuntimeStatus(), req.Msg.GetPage(), req.Msg.GetPageSize()) if err != nil { return nil, connect.NewError(connect.CodeInternal, err) }
response := &controllerv1.ListServicesResponse{ Services: make([]*controllerv1.ServiceSummary, 0, len(services)), TotalCount: totalCount, } for _, service := range services { response.Services = append(response.Services, &controllerv1.ServiceSummary{ Name: service.Name, IsDeclared: service.IsDeclared, RuntimeStatus: service.RuntimeStatus, UpdatedAt: service.UpdatedAt, InstanceCount: service.InstanceCount, RunningCount: service.RunningCount, TargetNodeCount: service.TargetNodeCount, }) }
return connect.NewResponse(response), nil}
func (server *serviceQueryServer) ListServiceWorkspaces(ctx context.Context, _ *connect.Request[controllerv1.ListServiceWorkspacesRequest]) (*connect.Response[controllerv1.ListServiceWorkspacesResponse], error) { workspaces, err := server.listServiceWorkspaceSummaries(ctx) if err != nil { return nil, connect.NewError(connect.CodeInternal, err) } return connect.NewResponse(&controllerv1.ListServiceWorkspacesResponse{Workspaces: workspaces}), nil}
func (server *serviceQueryServer) GetService(ctx context.Context, req *connect.Request[controllerv1.GetServiceRequest]) (*connect.Response[controllerv1.GetServiceResponse], error) { if req.Msg == nil || req.Msg.GetServiceName() == "" { return nil, connect.NewError(connect.CodeInvalidArgument, errors.New("service_name is required")) } service, err := repo.FindService(server.cfg.RepoDir, server.availableNodeIDs, req.Msg.GetServiceName()) if err != nil { return nil, connect.NewError(connect.CodeNotFound, err) } snapshot, err := server.db.GetServiceSnapshot(ctx, service.Name) if err != nil { if errors.Is(err, store.ErrServiceNotFound) { return nil, connect.NewError(connect.CodeNotFound, err) } return nil, connect.NewError(connect.CodeInternal, err) } instances, err := server.db.ListServiceInstances(ctx, service.Name) if err != nil { return nil, connect.NewError(connect.CodeInternal, err) } snapshotByNodeID, err := buildNodeSnapshotMap(ctx, server.db) if err != nil { return nil, connect.NewError(connect.CodeInternal, err) } response := &controllerv1.GetServiceResponse{ Name: service.Name, RuntimeStatus: snapshot.RuntimeStatus, UpdatedAt: snapshot.UpdatedAt, Nodes: append([]string(nil), service.TargetNodes...), Enabled: service.Enabled, Directory: filepath.ToSlash(mustRelativeServiceDir(server.cfg.RepoDir, service.Directory)), Instances: make([]*controllerv1.ServiceInstanceDetail, 0, len(instances)), Actions: buildServiceActionCapabilities(server.cfg, server.availableNodeIDs, snapshotByNodeID, service), } for _, instance := range instances { if !req.Msg.GetIncludeContainers() { response.Instances = append(response.Instances, serviceInstanceDetailMessage(instance, nil)) continue } detail, err := buildServiceInstanceDetail(ctx, server.db, server.cfg, server.dockerQueries, service, instance, requestTaskSource(req.Header())) if err != nil { return nil, err } response.Instances = append(response.Instances, detail) } return connect.NewResponse(response), nil}
func (server *serviceQueryServer) GetServiceWorkspace(ctx context.Context, req *connect.Request[controllerv1.GetServiceWorkspaceRequest]) (*connect.Response[controllerv1.GetServiceWorkspaceResponse], error) { if req.Msg == nil || strings.TrimSpace(req.Msg.GetFolder()) == "" { return nil, connect.NewError(connect.CodeInvalidArgument, errors.New("folder is required")) } workspaces, err := server.listServiceWorkspaceSummaries(ctx) if err != nil { return nil, connect.NewError(connect.CodeInternal, err) } folder := filepath.ToSlash(strings.TrimSpace(req.Msg.GetFolder())) for _, workspace := range workspaces { if workspace.GetFolder() == folder { return connect.NewResponse(&controllerv1.GetServiceWorkspaceResponse{Workspace: workspace}), nil } } return nil, connect.NewError(connect.CodeNotFound, fmt.Errorf("service folder %q not found", folder))}
func (server *serviceQueryServer) listServiceWorkspaceSummaries(ctx context.Context) ([]*controllerv1.ServiceWorkspaceSummary, error) { if server.cfg == nil || strings.TrimSpace(server.cfg.RepoDir) == "" { return nil, errors.New("controller repo_dir is not configured") } entries, err := repo.ListFiles(server.cfg.RepoDir, "", false) if err != nil { return nil, err } declaredServices, _, err := server.db.ListDeclaredServices(ctx, "", 1, 10000) if err != nil { return nil, err } snapshotByNodeID, err := buildNodeSnapshotMap(ctx, server.db) if err != nil { return nil, err } declaredByName := make(map[string]store.ServiceSummary, len(declaredServices)) for _, service := range declaredServices { declaredByName[service.Name] = service } workspaces := make([]*controllerv1.ServiceWorkspaceSummary, 0, len(entries)) for _, entry := range entries { if !entry.IsDir { continue } workspace, err := server.buildServiceWorkspaceSummary(entry.Path, entry.Name, declaredByName, snapshotByNodeID) if err != nil { return nil, err } workspaces = append(workspaces, workspace) } return workspaces, nil}
func (server *serviceQueryServer) buildServiceWorkspaceSummary(folder, defaultName string, declaredByName map[string]store.ServiceSummary, snapshotByNodeID map[string]store.NodeSnapshot) (*controllerv1.ServiceWorkspaceSummary, error) { workspace := &controllerv1.ServiceWorkspaceSummary{ Folder: folder, DisplayName: defaultName, RuntimeStatus: "uninitialized", Nodes: []string{}, Actions: buildDisabledServiceActionCapabilities(reasonMissingServiceMeta), } metaPath := filepath.Join(server.cfg.RepoDir, filepath.FromSlash(folder), repo.MetaFileName) metaInfo, err := os.Stat(metaPath) if err != nil { if errors.Is(err, os.ErrNotExist) { return workspace, nil } return nil, fmt.Errorf("stat service meta %q: %w", metaPath, err) } if metaInfo.IsDir() { return nil, fmt.Errorf("service meta %q must be a file", metaPath) } workspace.HasMeta = true workspace.Actions = buildDisabledServiceActionCapabilities(reasonServiceNotDeclared) meta, err := repo.LoadServiceMeta(metaPath) if err != nil { workspace.RuntimeStatus = "needs_validation" return workspace, nil //nolint:nilerr } serviceName := strings.TrimSpace(meta.Name) if serviceName != "" { workspace.DisplayName = serviceName workspace.ServiceName = serviceName } workspace.Nodes = normalizeWorkspaceNodeIDs(meta.Nodes) workspace.Enabled = meta.Enabled == nil || *meta.Enabled service, err := repo.LoadServiceFromMetaPath(metaPath, server.availableNodeIDs) if err != nil { workspace.RuntimeStatus = "needs_validation" return workspace, nil //nolint:nilerr } workspace.Actions = buildServiceActionCapabilities(server.cfg, server.availableNodeIDs, snapshotByNodeID, service) declared, ok := declaredByName[workspace.ServiceName] if !ok { workspace.RuntimeStatus = "needs_validation" workspace.Actions = buildDisabledServiceActionCapabilities(reasonServiceNotDeclared) return workspace, nil } workspace.IsDeclared = true workspace.RuntimeStatus = declared.RuntimeStatus workspace.UpdatedAt = declared.UpdatedAt return workspace, nil}
func normalizeWorkspaceNodeIDs(nodeIDs []string) []string { normalized := make([]string, 0, len(nodeIDs)) for _, nodeID := range nodeIDs { nodeID = strings.TrimSpace(nodeID) if nodeID == "" { continue } normalized = append(normalized, nodeID) } return normalized}
func (server *serviceQueryServer) GetServiceImageUpdateChecks(ctx context.Context, req *connect.Request[controllerv1.GetServiceImageUpdateChecksRequest]) (*connect.Response[controllerv1.GetServiceImageUpdateChecksResponse], error) { if req.Msg == nil || req.Msg.GetServiceName() == "" { return nil, connect.NewError(connect.CodeInvalidArgument, errors.New("service_name is required")) } checks, err := server.db.LatestServiceImageUpdateChecks(ctx, req.Msg.GetServiceName(), req.Msg.GetNodeId()) if err != nil { return nil, connect.NewError(connect.CodeInternal, err) } response := &controllerv1.GetServiceImageUpdateChecksResponse{Checks: make([]*controllerv1.ServiceImageUpdateCheckSummary, 0, len(checks))} for _, check := range checks { var candidateTags []string if check.CandidateTagsJSON != "" { _ = json.Unmarshal([]byte(check.CandidateTagsJSON), &candidateTags) } checkedAt := "" if !check.CheckedAt.IsZero() { checkedAt = check.CheckedAt.UTC().Format(time.RFC3339) } response.Checks = append(response.Checks, &controllerv1.ServiceImageUpdateCheckSummary{ ServiceName: check.ServiceName, NodeId: check.NodeID, ImageName: check.ImageName, ImageRef: check.ImageRef, PolicyType: check.PolicyType, CurrentValue: check.CurrentValue, CurrentTag: check.CurrentTag, CurrentDigest: check.CurrentDigest, CandidateTag: check.CandidateTag, CandidateDigest: check.CandidateDigest, CandidateTags: candidateTags, UpdateAvailable: check.UpdateAvailable, CheckStatus: check.CheckStatus, ErrorSummary: check.ErrorSummary, CheckedAt: checkedAt, }) } return connect.NewResponse(response), nil}