feat: add resilient container restart controls
This commit is contained in:
1 parent
1bbb14470c
commit
7bb33e528c
13 files changed
+747
-28
No files matched your search
@@ -22,6 +22,16 @@ import (
|
||||
|
||||
var ErrCredentialMissing = errors.New("Update-Zugang ist noch nicht eingerichtet")
|
||||
|
||||
const (
|
||||
agentUnavailableRetryInterval = 2 * time.Second
|
||||
agentAvailableCheckInterval = time.Minute
|
||||
)
|
||||
|
||||
type updateAgent interface {
|
||||
Start(context.Context, updateprotocol.Request) error
|
||||
Status(context.Context) (updateprotocol.Status, error)
|
||||
}
|
||||
|
||||
type State struct {
|
||||
CurrentVersion string
|
||||
LatestVersion string
|
||||
@@ -46,9 +56,10 @@ type Manager struct {
|
||||
logger *slog.Logger
|
||||
apiURL string
|
||||
interval time.Duration
|
||||
agent *AgentClient
|
||||
agent updateAgent
|
||||
httpClient *http.Client
|
||||
checkMu sync.Mutex
|
||||
requestMu sync.Mutex
|
||||
stateMu sync.RWMutex
|
||||
state State
|
||||
checkSignal chan struct{}
|
||||
@@ -87,8 +98,8 @@ func New(database *store.Store, cipher *secure.Cipher, apiURL, agentSocket strin
|
||||
}
|
||||
|
||||
func (m *Manager) Start(ctx context.Context) {
|
||||
go m.monitorAgent(ctx, agentUnavailableRetryInterval, agentAvailableCheckInterval)
|
||||
go func() {
|
||||
m.refreshAgentStatus(ctx)
|
||||
initial := time.NewTimer(15 * time.Second)
|
||||
defer initial.Stop()
|
||||
var ticker *time.Ticker
|
||||
@@ -121,7 +132,28 @@ func (m *Manager) backgroundCheck(ctx context.Context) {
|
||||
if err := m.CheckNow(checkCtx); err != nil && !errors.Is(err, ErrCredentialMissing) && !errors.Is(err, context.Canceled) {
|
||||
m.logger.Warn("software update check failed", "error", err)
|
||||
}
|
||||
m.refreshAgentStatus(ctx)
|
||||
}
|
||||
|
||||
func (m *Manager) monitorAgent(ctx context.Context, unavailableInterval, availableInterval time.Duration) {
|
||||
for {
|
||||
m.refreshAgentStatus(ctx)
|
||||
delay := unavailableInterval
|
||||
if m.State().AgentAvailable {
|
||||
delay = availableInterval
|
||||
}
|
||||
timer := time.NewTimer(delay)
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
if !timer.Stop() {
|
||||
select {
|
||||
case <-timer.C:
|
||||
default:
|
||||
}
|
||||
}
|
||||
return
|
||||
case <-timer.C:
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (m *Manager) TriggerCheck() {
|
||||
@@ -190,6 +222,9 @@ func (m *Manager) CheckNow(ctx context.Context) error {
|
||||
}
|
||||
|
||||
func (m *Manager) RequestUpdate(ctx context.Context, version string) error {
|
||||
m.requestMu.Lock()
|
||||
defer m.requestMu.Unlock()
|
||||
|
||||
version = strings.TrimSpace(version)
|
||||
state := m.State()
|
||||
if !validReleaseVersion(version) || version != state.LatestVersion {
|
||||
@@ -198,12 +233,16 @@ func (m *Manager) RequestUpdate(ctx context.Context, version string) error {
|
||||
if comparison, err := compareVersions(state.CurrentVersion, version); err != nil || comparison >= 0 {
|
||||
return errors.New("die ausgewählte Version ist nicht neuer als die laufende Version")
|
||||
}
|
||||
if !state.AgentAvailable {
|
||||
status, err := m.UpdateStatus(ctx)
|
||||
if err != nil {
|
||||
return errors.New("Update-Agent ist nicht erreichbar")
|
||||
}
|
||||
if !state.AgentCompatible {
|
||||
if !agentCompatible(status) {
|
||||
return errors.New("Web-App und Update-Agent müssen zuerst gemeinsam mit Docker Compose aktualisiert werden")
|
||||
}
|
||||
if status.Busy() {
|
||||
return errors.New("ein Software-Update läuft bereits")
|
||||
}
|
||||
settings, err := m.store.GetSoftwareUpdateSettings(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
|
||||
@@ -3,12 +3,14 @@ package appupdate
|
||||
import (
|
||||
"context"
|
||||
"encoding/base64"
|
||||
"errors"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
@@ -19,6 +21,48 @@ import (
|
||||
"github.com/root-kdw12/hetzner-api-dyndns-records-webinterface/internal/updateprotocol"
|
||||
)
|
||||
|
||||
type fakeUpdateAgent struct {
|
||||
mu sync.Mutex
|
||||
status updateprotocol.Status
|
||||
statusErr error
|
||||
startErr error
|
||||
statusCalls int
|
||||
startCalls int
|
||||
requests []updateprotocol.Request
|
||||
}
|
||||
|
||||
func (a *fakeUpdateAgent) Status(context.Context) (updateprotocol.Status, error) {
|
||||
a.mu.Lock()
|
||||
defer a.mu.Unlock()
|
||||
a.statusCalls++
|
||||
return a.status, a.statusErr
|
||||
}
|
||||
|
||||
func (a *fakeUpdateAgent) Start(_ context.Context, request updateprotocol.Request) error {
|
||||
a.mu.Lock()
|
||||
defer a.mu.Unlock()
|
||||
a.startCalls++
|
||||
a.requests = append(a.requests, request)
|
||||
if a.startErr == nil {
|
||||
a.status.Phase = updateprotocol.PhaseQueued
|
||||
a.status.TargetVersion = request.Version
|
||||
}
|
||||
return a.startErr
|
||||
}
|
||||
|
||||
func (a *fakeUpdateAgent) setStatus(status updateprotocol.Status, err error) {
|
||||
a.mu.Lock()
|
||||
a.status = status
|
||||
a.statusErr = err
|
||||
a.mu.Unlock()
|
||||
}
|
||||
|
||||
func (a *fakeUpdateAgent) counts() (status, start int) {
|
||||
a.mu.Lock()
|
||||
defer a.mu.Unlock()
|
||||
return a.statusCalls, a.startCalls
|
||||
}
|
||||
|
||||
func TestManagerKeepsUpdateReminderWhenGiteaBecomesUnavailable(t *testing.T) {
|
||||
t.Parallel()
|
||||
const credential = " test-password-or-token-that-must-stay-secret "
|
||||
@@ -138,6 +182,116 @@ func TestAgentCompatibilityStartsAtStableV14Protocol(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestRequestUpdateRefreshesStaleAgentStateBeforeStarting(t *testing.T) {
|
||||
t.Parallel()
|
||||
agent := &fakeUpdateAgent{status: compatibleAgentStatus(updateprotocol.PhaseIdle)}
|
||||
manager, database := updateReadyManager(t, agent)
|
||||
if manager.State().AgentAvailable {
|
||||
t.Fatal("test precondition failed: cached agent state is already available")
|
||||
}
|
||||
if err := manager.RequestUpdate(context.Background(), "v1.4.4"); err != nil {
|
||||
t.Fatalf("live agent was rejected because of stale cached state: %v", err)
|
||||
}
|
||||
statusCalls, startCalls := agent.counts()
|
||||
if statusCalls != 1 || startCalls != 1 {
|
||||
t.Fatalf("agent was not checked immediately before start: status=%d start=%d", statusCalls, startCalls)
|
||||
}
|
||||
settings, err := database.GetSoftwareUpdateSettings(context.Background())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if settings.DesiredVersion != "v1.4.4" {
|
||||
t.Fatalf("target version was not persisted: %#v", settings)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRequestUpdateRejectsUnavailableIncompatibleAndBusyAgents(t *testing.T) {
|
||||
t.Parallel()
|
||||
tests := []struct {
|
||||
name string
|
||||
status updateprotocol.Status
|
||||
statusErr error
|
||||
}{
|
||||
{name: "unavailable", statusErr: errors.New("offline")},
|
||||
{name: "incompatible", status: updateprotocol.Status{Phase: updateprotocol.PhaseIdle, AgentVersion: "v1.3.0", Protocol: 1}},
|
||||
{name: "busy", status: compatibleAgentStatus(updateprotocol.PhasePulling)},
|
||||
}
|
||||
for _, test := range tests {
|
||||
test := test
|
||||
t.Run(test.name, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
agent := &fakeUpdateAgent{status: test.status, statusErr: test.statusErr}
|
||||
manager, database := updateReadyManager(t, agent)
|
||||
if err := manager.RequestUpdate(context.Background(), "v1.4.4"); err == nil {
|
||||
t.Fatal("unsafe update request was accepted")
|
||||
}
|
||||
_, startCalls := agent.counts()
|
||||
if startCalls != 0 {
|
||||
t.Fatalf("agent start was called %d times", startCalls)
|
||||
}
|
||||
settings, err := database.GetSoftwareUpdateSettings(context.Background())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if settings.DesiredVersion != "" {
|
||||
t.Fatalf("rejected update changed desired version: %#v", settings)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestRequestUpdateSerializesParallelClicks(t *testing.T) {
|
||||
t.Parallel()
|
||||
agent := &fakeUpdateAgent{status: compatibleAgentStatus(updateprotocol.PhaseIdle)}
|
||||
manager, _ := updateReadyManager(t, agent)
|
||||
results := make(chan error, 2)
|
||||
start := make(chan struct{})
|
||||
for index := 0; index < 2; index++ {
|
||||
go func() {
|
||||
<-start
|
||||
results <- manager.RequestUpdate(context.Background(), "v1.4.4")
|
||||
}()
|
||||
}
|
||||
close(start)
|
||||
var successes int
|
||||
for index := 0; index < 2; index++ {
|
||||
if err := <-results; err == nil {
|
||||
successes++
|
||||
}
|
||||
}
|
||||
_, startCalls := agent.counts()
|
||||
if successes != 1 || startCalls != 1 {
|
||||
t.Fatalf("parallel clicks started an unexpected number of updates: successes=%d starts=%d", successes, startCalls)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAgentMonitorRecoversAfterDelayedAgentStartup(t *testing.T) {
|
||||
t.Parallel()
|
||||
agent := &fakeUpdateAgent{statusErr: errors.New("not listening yet")}
|
||||
manager := &Manager{agent: agent}
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
manager.monitorAgent(ctx, 2*time.Millisecond, 20*time.Millisecond)
|
||||
close(done)
|
||||
}()
|
||||
waitForCondition(t, time.Second, func() bool {
|
||||
statusCalls, _ := agent.counts()
|
||||
return statusCalls > 0
|
||||
})
|
||||
agent.setStatus(compatibleAgentStatus(updateprotocol.PhaseIdle), nil)
|
||||
waitForCondition(t, time.Second, func() bool {
|
||||
state := manager.State()
|
||||
return state.AgentAvailable && state.AgentCompatible
|
||||
})
|
||||
cancel()
|
||||
select {
|
||||
case <-done:
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("agent monitor did not stop after context cancellation")
|
||||
}
|
||||
}
|
||||
|
||||
func TestManagerClearsCompletedDesiredVersionOnStartup(t *testing.T) {
|
||||
t.Parallel()
|
||||
ctx := context.Background()
|
||||
@@ -176,3 +330,58 @@ func TestManagerClearsCompletedDesiredVersionOnStartup(t *testing.T) {
|
||||
t.Fatalf("completed desired version remained persisted: %#v", settings)
|
||||
}
|
||||
}
|
||||
|
||||
func compatibleAgentStatus(phase string) updateprotocol.Status {
|
||||
return updateprotocol.Status{Phase: phase, AgentVersion: buildinfo.MinimumAgentVersion, Protocol: buildinfo.MinimumAgentProtocol}
|
||||
}
|
||||
|
||||
func updateReadyManager(t *testing.T, agent updateAgent) (*Manager, *store.Store) {
|
||||
t.Helper()
|
||||
ctx := context.Background()
|
||||
database, err := store.Open(ctx, filepath.Join(t.TempDir(), "app.db"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = database.Close() })
|
||||
keyFile := filepath.Join(t.TempDir(), "master-key")
|
||||
key := make([]byte, 32)
|
||||
for index := range key {
|
||||
key[index] = byte(index + 1)
|
||||
}
|
||||
if err := os.WriteFile(keyFile, []byte(base64.RawStdEncoding.EncodeToString(key)), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
cipher, err := secure.LoadCipher(keyFile)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
manager, err := New(database, cipher, "https://git.example/api/releases/latest", filepath.Join(t.TempDir(), "agent.sock"), time.Hour, slog.New(slog.NewTextHandler(os.Stderr, nil)))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
manager.agent = agent
|
||||
if err := manager.SaveCredential(ctx, "registry-user", "registry-password"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := database.RecordSoftwareUpdateSuccess(ctx, "v1.4.4", "https://git.example/releases/v1.4.4", time.Now()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
settings, err := database.GetSoftwareUpdateSettings(ctx)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
manager.setStoredState(settings)
|
||||
return manager, database
|
||||
}
|
||||
|
||||
func waitForCondition(t *testing.T, timeout time.Duration, condition func() bool) {
|
||||
t.Helper()
|
||||
deadline := time.Now().Add(timeout)
|
||||
for time.Now().Before(deadline) {
|
||||
if condition() {
|
||||
return
|
||||
}
|
||||
time.Sleep(time.Millisecond)
|
||||
}
|
||||
t.Fatal("condition did not become true in time")
|
||||
}
|
||||
Reference in new issue
Block a user