73 lines
2.0 KiB
Go
73 lines
2.0 KiB
Go
package controlplane
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
)
|
|
|
|
var ErrSnapshotStreamClosed = errors.New("worker snapshot stream closed")
|
|
|
|
type SessionRunner struct {
|
|
reporter *RuntimeReporter
|
|
watcher *SnapshotWatcher
|
|
outcomes *OutcomeReporter
|
|
}
|
|
|
|
func NewSessionRunner(reporter *RuntimeReporter, watcher *SnapshotWatcher, outcomes ...*OutcomeReporter) (*SessionRunner, error) {
|
|
if reporter == nil || watcher == nil || len(outcomes) > 1 || (len(outcomes) == 1 && outcomes[0] == nil) {
|
|
return nil, ErrInvalidOptions
|
|
}
|
|
runner := &SessionRunner{reporter: reporter, watcher: watcher}
|
|
if len(outcomes) == 1 {
|
|
runner.outcomes = outcomes[0]
|
|
}
|
|
return runner, nil
|
|
}
|
|
|
|
// Run registers exactly once, then runs snapshot intake and Runtime reporting
|
|
// against that session. A stream ending without context cancellation is a
|
|
// failure so the owning process can apply its reconnect policy.
|
|
func (runner *SessionRunner) Run(ctx context.Context) error {
|
|
if runner == nil || ctx == nil {
|
|
return ErrInvalidOptions
|
|
}
|
|
registration, err := runner.reporter.Register(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
groupCtx, cancel := context.WithCancel(ctx)
|
|
defer cancel()
|
|
runners := []func() error{
|
|
func() error { return runner.watcher.Watch(groupCtx, registration.SessionID) },
|
|
func() error { return runner.reporter.RunRegistered(groupCtx) },
|
|
}
|
|
if runner.outcomes != nil {
|
|
runners = append(runners, func() error { return runner.outcomes.RunRegistered(groupCtx, registration) })
|
|
}
|
|
results := make(chan error, len(runners))
|
|
for _, run := range runners {
|
|
go func(run func() error) { results <- run() }(run)
|
|
}
|
|
|
|
first := <-results
|
|
cancel()
|
|
others := make([]error, 0, len(runners)-1)
|
|
for range len(runners) - 1 {
|
|
others = append(others, <-results)
|
|
}
|
|
if ctx.Err() != nil {
|
|
return ctx.Err()
|
|
}
|
|
if first == nil {
|
|
return ErrSnapshotStreamClosed
|
|
}
|
|
if errors.Is(first, context.Canceled) {
|
|
for _, other := range others {
|
|
if other != nil && !errors.Is(other, context.Canceled) {
|
|
return other
|
|
}
|
|
}
|
|
}
|
|
return first
|
|
}
|