82 lines
2.2 KiB
Go
82 lines
2.2 KiB
Go
// Package controller composes independently runnable scheduler and backend workers.
|
|
package controller
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"slices"
|
|
"strings"
|
|
|
|
"golang.org/x/sync/errgroup"
|
|
)
|
|
|
|
type ComponentName string
|
|
|
|
const (
|
|
Scheduler ComponentName = "scheduler"
|
|
KubernetesWorker ComponentName = "kubernetes-worker"
|
|
OpenSandboxWorker ComponentName = "opensandbox-worker"
|
|
)
|
|
|
|
var defaultComponents = []ComponentName{Scheduler, KubernetesWorker, OpenSandboxWorker}
|
|
|
|
// Selection parses --components. An empty value enables all components.
|
|
type Selection []ComponentName
|
|
|
|
func ParseSelection(value string) (Selection, error) {
|
|
if strings.TrimSpace(value) == "" || strings.TrimSpace(value) == "all" {
|
|
return append(Selection(nil), defaultComponents...), nil
|
|
}
|
|
var selected Selection
|
|
for _, raw := range strings.Split(value, ",") {
|
|
name := ComponentName(strings.TrimSpace(raw))
|
|
switch name {
|
|
case "pod-worker":
|
|
name = KubernetesWorker
|
|
case "vm-worker":
|
|
name = OpenSandboxWorker
|
|
}
|
|
if !slices.Contains(defaultComponents, name) {
|
|
return nil, fmt.Errorf("unknown controller component %q", name)
|
|
}
|
|
if !slices.Contains(selected, name) {
|
|
selected = append(selected, name)
|
|
}
|
|
}
|
|
if len(selected) == 0 {
|
|
return nil, errors.New("at least one controller component is required")
|
|
}
|
|
return selected, nil
|
|
}
|
|
|
|
type Component interface {
|
|
Run(context.Context) error
|
|
}
|
|
|
|
type Registry map[ComponentName]Component
|
|
|
|
// Run starts exactly the selected components in one process. The first real
|
|
// failure cancels its peers; ordinary context cancellation is graceful.
|
|
func Run(ctx context.Context, selection Selection, registry Registry) error {
|
|
group, groupContext := errgroup.WithContext(ctx)
|
|
for _, name := range selection {
|
|
component, ok := registry[name]
|
|
if !ok || component == nil {
|
|
return fmt.Errorf("component %q is not configured", name)
|
|
}
|
|
name, component := name, component
|
|
group.Go(func() error {
|
|
err := component.Run(groupContext)
|
|
if errors.Is(err, context.Canceled) && groupContext.Err() != nil {
|
|
return nil
|
|
}
|
|
if err != nil {
|
|
return fmt.Errorf("component %s: %w", name, err)
|
|
}
|
|
return nil
|
|
})
|
|
}
|
|
return group.Wait()
|
|
}
|