ship: v0.1 Foundation milestone complete (#1)
This commit was merged in pull request #1.
This commit is contained in:
@@ -0,0 +1,52 @@
|
||||
package engine
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log/slog"
|
||||
|
||||
"git.cloudinit.dev/coreci/orca/internal/store"
|
||||
)
|
||||
|
||||
// Audit wraps a slog.Logger and persists structured audit entries to SQLite.
|
||||
type Audit struct {
|
||||
repo *store.AuditRepo
|
||||
log *slog.Logger
|
||||
}
|
||||
|
||||
func NewAudit(repo *store.AuditRepo, log *slog.Logger) *Audit {
|
||||
if log == nil {
|
||||
log = slog.Default()
|
||||
}
|
||||
return &Audit{repo: repo, log: log}
|
||||
}
|
||||
|
||||
func (a *Audit) Record(ctx context.Context, actor, action, resource, result string, err error, meta map[string]any) {
|
||||
entry := &store.AuditEntry{
|
||||
Actor: actor,
|
||||
Action: action,
|
||||
Resource: resource,
|
||||
Result: result,
|
||||
Metadata: meta,
|
||||
}
|
||||
if err != nil {
|
||||
entry.Error = err.Error()
|
||||
}
|
||||
if persistErr := a.repo.Append(ctx, entry); persistErr != nil {
|
||||
a.log.Error("audit persist failed",
|
||||
slog.String("action", action),
|
||||
slog.String("resource", resource),
|
||||
slog.String("error", persistErr.Error()))
|
||||
}
|
||||
attrs := []any{
|
||||
slog.String("actor", actor),
|
||||
slog.String("action", action),
|
||||
slog.String("resource", resource),
|
||||
slog.String("result", result),
|
||||
}
|
||||
if err != nil {
|
||||
attrs = append(attrs, slog.String("error", err.Error()))
|
||||
a.log.Warn("audit", attrs...)
|
||||
} else {
|
||||
a.log.Info("audit", attrs...)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,150 @@
|
||||
package engine
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"os/exec"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/google/uuid"
|
||||
|
||||
"git.cloudinit.dev/coreci/orca/internal/model"
|
||||
"git.cloudinit.dev/coreci/orca/internal/store"
|
||||
)
|
||||
|
||||
type Executor struct {
|
||||
jobs *store.JobRepo
|
||||
tasks *store.TaskRepo
|
||||
log *slog.Logger
|
||||
mu sync.Mutex
|
||||
}
|
||||
|
||||
func NewExecutor(jobs *store.JobRepo, tasks *store.TaskRepo, log *slog.Logger) *Executor {
|
||||
if log == nil {
|
||||
log = slog.Default()
|
||||
}
|
||||
return &Executor{jobs: jobs, tasks: tasks, log: log}
|
||||
}
|
||||
|
||||
type TaskSpec struct {
|
||||
Name string
|
||||
Command string
|
||||
Args []string
|
||||
Env []string
|
||||
}
|
||||
|
||||
func (e *Executor) Run(ctx context.Context, job *model.Job, specs []TaskSpec) error {
|
||||
e.mu.Lock()
|
||||
defer e.mu.Unlock()
|
||||
|
||||
// Insert the job first so tasks can reference it via foreign key.
|
||||
if err := e.jobs.Insert(ctx, job); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := e.jobs.UpdateStatus(ctx, job.ID, model.JobStatusRunning, 0); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
var (
|
||||
wg sync.WaitGroup
|
||||
failedCount int
|
||||
exitCode int
|
||||
mu sync.Mutex
|
||||
)
|
||||
|
||||
for _, ts := range specs {
|
||||
wg.Add(1)
|
||||
go func(ts TaskSpec) {
|
||||
defer wg.Done()
|
||||
if err := e.runOne(ctx, job, ts); err != nil {
|
||||
mu.Lock()
|
||||
failedCount++
|
||||
e.log.Error("task failed",
|
||||
slog.String("job_id", job.ID),
|
||||
slog.String("task", ts.Name),
|
||||
slog.String("error", err.Error()))
|
||||
mu.Unlock()
|
||||
}
|
||||
}(ts)
|
||||
}
|
||||
wg.Wait()
|
||||
|
||||
if failedCount > 0 {
|
||||
exitCode = 1
|
||||
if err := e.jobs.UpdateStatus(ctx, job.ID, model.JobStatusFailed, exitCode); err != nil {
|
||||
return err
|
||||
}
|
||||
return fmt.Errorf("%d/%d tasks failed", failedCount, len(specs))
|
||||
}
|
||||
|
||||
if err := e.jobs.UpdateStatus(ctx, job.ID, model.JobStatusComplete, 0); err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (e *Executor) runOne(ctx context.Context, job *model.Job, ts TaskSpec) error {
|
||||
task := &model.Task{
|
||||
ID: uuid.NewString(),
|
||||
JobID: job.ID,
|
||||
Command: ts.Command,
|
||||
Args: ts.Args,
|
||||
Env: ts.Env,
|
||||
Status: model.TaskStatusPending,
|
||||
}
|
||||
if err := e.tasks.Insert(ctx, task); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
cmd := exec.CommandContext(ctx, ts.Command, ts.Args...)
|
||||
cmd.Env = append(cmd.Environ(), ts.Env...)
|
||||
// WaitDelay (Go 1.25+) bounds the time spent waiting on a child process
|
||||
// that fails to exit after the context is canceled.
|
||||
cmd.WaitDelay = 5 * time.Second
|
||||
|
||||
var stdout, stderr bytes.Buffer
|
||||
cmd.Stdout = &stdout
|
||||
cmd.Stderr = &stderr
|
||||
|
||||
if err := cmd.Start(); err != nil {
|
||||
_ = e.tasks.UpdateKilled(ctx, task.ID)
|
||||
return fmt.Errorf("start: %w", err)
|
||||
}
|
||||
|
||||
if err := e.tasks.UpdateRunning(ctx, task.ID, cmd.Process.Pid); err != nil {
|
||||
e.log.Warn("update running failed", slog.String("error", err.Error()))
|
||||
}
|
||||
|
||||
e.log.Info("task started",
|
||||
slog.String("job_id", job.ID),
|
||||
slog.String("task", ts.Name),
|
||||
slog.Int("pid", cmd.Process.Pid))
|
||||
|
||||
done := make(chan error, 1)
|
||||
go func() { done <- cmd.Wait() }()
|
||||
|
||||
select {
|
||||
case err := <-done:
|
||||
exitCode := 0
|
||||
if err != nil {
|
||||
if ee, ok := err.(*exec.ExitError); ok {
|
||||
exitCode = ee.ExitCode()
|
||||
} else {
|
||||
exitCode = 1
|
||||
}
|
||||
}
|
||||
_ = e.tasks.UpdateDone(ctx, task.ID, exitCode, stdout.String(), stderr.String())
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
case <-ctx.Done():
|
||||
// WaitDelay (set above) gives the process a grace period to exit
|
||||
// cleanly before being killed.
|
||||
_ = e.tasks.UpdateKilled(ctx, task.ID)
|
||||
return ctx.Err()
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,70 @@
|
||||
package engine
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
|
||||
"git.cloudinit.dev/coreci/orca/internal/model"
|
||||
"git.cloudinit.dev/coreci/orca/internal/store"
|
||||
)
|
||||
|
||||
type NodeRegistry struct {
|
||||
repo *store.NodeRepo
|
||||
audit *Audit
|
||||
log *slog.Logger
|
||||
}
|
||||
|
||||
func NewNodeRegistry(repo *store.NodeRepo, audit *Audit, log *slog.Logger) *NodeRegistry {
|
||||
if log == nil {
|
||||
log = slog.Default()
|
||||
}
|
||||
return &NodeRegistry{repo: repo, audit: audit, log: log}
|
||||
}
|
||||
|
||||
func (r *NodeRegistry) Join(ctx context.Context, n *model.Node) error {
|
||||
if err := r.repo.Insert(ctx, n); err != nil {
|
||||
r.audit.Record(ctx, "cli", "node.join", n.ID, "failure", err, map[string]any{
|
||||
"name": n.Name,
|
||||
"address": n.Address,
|
||||
})
|
||||
return fmt.Errorf("join node: %w", err)
|
||||
}
|
||||
r.audit.Record(ctx, "cli", "node.join", n.ID, "success", nil, map[string]any{
|
||||
"name": n.Name,
|
||||
"address": n.Address,
|
||||
})
|
||||
r.log.Info("node joined",
|
||||
slog.String("node_id", n.ID),
|
||||
slog.String("name", n.Name),
|
||||
slog.String("address", n.Address))
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *NodeRegistry) Leave(ctx context.Context, id string) error {
|
||||
if err := r.repo.UpdateState(ctx, id, model.NodeStateLeft); err != nil {
|
||||
r.audit.Record(ctx, "cli", "node.leave", id, "failure", err, nil)
|
||||
return fmt.Errorf("leave node: %w", err)
|
||||
}
|
||||
r.audit.Record(ctx, "cli", "node.leave", id, "success", nil, nil)
|
||||
r.log.Info("node left", slog.String("node_id", id))
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *NodeRegistry) Forget(ctx context.Context, id string) error {
|
||||
if err := r.repo.Delete(ctx, id); err != nil {
|
||||
r.audit.Record(ctx, "cli", "node.forget", id, "failure", err, nil)
|
||||
return fmt.Errorf("forget node: %w", err)
|
||||
}
|
||||
r.audit.Record(ctx, "cli", "node.forget", id, "success", nil, nil)
|
||||
r.log.Info("node removed from registry", slog.String("node_id", id))
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *NodeRegistry) List(ctx context.Context) ([]*model.Node, error) {
|
||||
return r.repo.List(ctx)
|
||||
}
|
||||
|
||||
func (r *NodeRegistry) Get(ctx context.Context, id string) (*model.Node, error) {
|
||||
return r.repo.Get(ctx, id)
|
||||
}
|
||||
Reference in New Issue
Block a user