mirror of
https://github.com/mulgadc/spinifex.git
synced 2026-10-20 11:21:49 +00:00
Josh Siviour
4e6a2eb46a
On startup the agent had no reconcile pass: Run registered, heartbeat, and immediately polled assignments with an in-memory dedup map. After a restart the gateway re-delivers the instance's assignments and the agent re-ran them, orphaning containers still running from before and double-starting tasks. Add a reconcile() pass that runs once before polling: it lists the runtime's containers, and for each task still running in this cluster re-registers its IAM credentials, re-attaches the exit-wait, and refreshes RUNNING on the gateway. The adopted task set seeds pollAssignments so re-delivered assignments are acked but not re-run. - runtime: add a label-agnostic List(ctx) []Container to the Runner seam, with a containerd implementation and fake support. - taskLabels: stamp credID / taskRoleArn / eniMac (when set) so a container is self-describing enough to re-register creds and tear down its netns on adoption. - pollAssignments takes the seed dedup map. Best-effort: a nil runner or a List error degrades to today's poll+run behaviour.
126 lines
2.8 KiB
Go
126 lines
2.8 KiB
Go
package runtime
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"sync"
|
|
)
|
|
|
|
// FakePuller is an in-memory ImagePuller for tests and the scheduler-only build.
|
|
// It records pull requests and replays a programmed result or error.
|
|
type FakePuller struct {
|
|
mu sync.Mutex
|
|
Pulls []PullSpec
|
|
Authzd []string
|
|
Result Image
|
|
Err error
|
|
Closed bool
|
|
OnPull func(PullSpec) (Image, error)
|
|
|
|
// Run bookkeeping.
|
|
Runs []RunSpec
|
|
RunErr error
|
|
WaitCode int
|
|
WaitErr error
|
|
Waited []string
|
|
Removed []string
|
|
OnRun func(string, RunSpec) (string, error)
|
|
|
|
// List bookkeeping: Containers is replayed by List; ListErr forces a failure;
|
|
// Listed records that List was called.
|
|
Containers []Container
|
|
ListErr error
|
|
Listed bool
|
|
}
|
|
|
|
var (
|
|
_ ImagePuller = (*FakePuller)(nil)
|
|
_ Runner = (*FakePuller)(nil)
|
|
)
|
|
|
|
// Pull records the spec, exercises the resolver, and returns the programmed
|
|
// result. If OnPull is set it takes precedence over Result/Err.
|
|
func (f *FakePuller) Pull(ctx context.Context, spec PullSpec, r Resolver) (Image, error) {
|
|
f.mu.Lock()
|
|
f.Pulls = append(f.Pulls, spec)
|
|
f.mu.Unlock()
|
|
|
|
if r != nil {
|
|
user, _, _, err := r.Authorize(ctx, spec.Ref)
|
|
if err != nil {
|
|
return Image{}, fmt.Errorf("authorize %s: %w", spec.Ref, err)
|
|
}
|
|
f.mu.Lock()
|
|
f.Authzd = append(f.Authzd, user)
|
|
f.mu.Unlock()
|
|
}
|
|
|
|
if f.OnPull != nil {
|
|
return f.OnPull(spec)
|
|
}
|
|
if f.Err != nil {
|
|
return Image{}, f.Err
|
|
}
|
|
res := f.Result
|
|
if res.Ref == "" {
|
|
res.Ref = spec.Ref
|
|
}
|
|
return res, nil
|
|
}
|
|
|
|
// Close marks the puller closed.
|
|
func (f *FakePuller) Close() error {
|
|
f.mu.Lock()
|
|
defer f.mu.Unlock()
|
|
f.Closed = true
|
|
return nil
|
|
}
|
|
|
|
// Run records the spec and returns id (or the programmed result/error).
|
|
func (f *FakePuller) Run(_ context.Context, id string, spec RunSpec) (string, error) {
|
|
f.mu.Lock()
|
|
f.Runs = append(f.Runs, spec)
|
|
f.mu.Unlock()
|
|
if f.OnRun != nil {
|
|
return f.OnRun(id, spec)
|
|
}
|
|
if f.RunErr != nil {
|
|
return "", f.RunErr
|
|
}
|
|
return id, nil
|
|
}
|
|
|
|
// Wait records the container ID and returns the programmed exit code/error.
|
|
func (f *FakePuller) Wait(_ context.Context, containerID string) (RunStatus, error) {
|
|
f.mu.Lock()
|
|
f.Waited = append(f.Waited, containerID)
|
|
f.mu.Unlock()
|
|
return RunStatus{ExitCode: f.WaitCode}, f.WaitErr
|
|
}
|
|
|
|
// Remove records the container ID.
|
|
func (f *FakePuller) Remove(_ context.Context, containerID string) error {
|
|
f.mu.Lock()
|
|
defer f.mu.Unlock()
|
|
f.Removed = append(f.Removed, containerID)
|
|
return nil
|
|
}
|
|
|
|
// Waits returns a copy of the container IDs passed to Wait, for assertions.
|
|
func (f *FakePuller) Waits() []string {
|
|
f.mu.Lock()
|
|
defer f.mu.Unlock()
|
|
return append([]string(nil), f.Waited...)
|
|
}
|
|
|
|
// List replays the programmed container set (or error).
|
|
func (f *FakePuller) List(_ context.Context) ([]Container, error) {
|
|
f.mu.Lock()
|
|
defer f.mu.Unlock()
|
|
f.Listed = true
|
|
if f.ListErr != nil {
|
|
return nil, f.ListErr
|
|
}
|
|
return f.Containers, nil
|
|
}
|