feat: coordinate Active Profile target panel

This commit is contained in:
proxyctl coordinator
2026-07-17 04:05:14 +08:00
parent 34b45c807f
commit fd113fb316
11 changed files with 652 additions and 36 deletions
+135 -2
View File
@@ -4,6 +4,7 @@ import (
"context"
"errors"
"fmt"
"strings"
"time"
"proxyctl/internal/profile"
@@ -18,6 +19,7 @@ const (
DeleteProfile
ActivateProfile
SetTargetDesired
RefreshTargets
)
type Intent struct {
@@ -44,10 +46,20 @@ type Snapshot struct {
Targets []TargetSnapshot
}
type TargetProgress struct {
TargetID string
Applying bool
State target.State
Evidence string
Err error
}
type Result struct {
Snapshot Snapshot
Notice string
Confirmation string
Progress []TargetProgress
Complete bool
Err error
}
@@ -56,6 +68,13 @@ type Module interface {
Dispatch(Intent) Result
}
// ProgressiveModule is implemented by Applications that can expose each
// coordinated Target update while an Active Profile transition is running.
// The final Result has Complete set and closes the stream.
type ProgressiveModule interface {
DispatchProgress(Intent) <-chan Result
}
type Application struct {
store profile.Store
state profile.State
@@ -103,6 +122,21 @@ func (a *Application) Snapshot() Snapshot {
}
func (a *Application) Dispatch(intent Intent) Result {
return a.dispatch(intent, nil)
}
func (a *Application) DispatchProgress(intent Intent) <-chan Result {
results := make(chan Result, 2*len(a.targets)+1)
go func() {
defer close(results)
final := a.dispatch(intent, func(update Result) { results <- update })
final.Complete = true
results <- final
}()
return results
}
func (a *Application) dispatch(intent Intent, publish func(Result)) Result {
if a.mutationLock != nil {
unlock, err := a.mutationLock.Lock(context.Background())
if err != nil {
@@ -113,14 +147,19 @@ func (a *Application) Dispatch(intent Intent) Result {
if intent.Kind == SetTargetDesired {
return a.setTargetDesired(intent)
}
if intent.Kind == RefreshTargets {
a.refreshTargets()
return Result{Snapshot: a.Snapshot(), Notice: "Refreshed Target availability and state"}
}
if intent.Kind == ActivateProfile {
return a.activateProfile(intent, publish)
}
var notice string
next, err := a.store.Mutate(func(current profile.State) (profile.State, error) {
var mutationErr error
switch intent.Kind {
case CreateProfile:
current, notice, mutationErr = create(current, intent.Profile)
case ActivateProfile:
current, notice, mutationErr = activate(current, intent.Name)
case UpdateProfile:
current, notice, mutationErr = update(current, intent.Name, intent.Profile)
case DeleteProfile:
@@ -150,6 +189,100 @@ func (a *Application) Dispatch(intent Intent) Result {
return Result{Snapshot: a.Snapshot(), Notice: notice, Err: err}
}
func (a *Application) activateProfile(intent Intent, publish func(Result)) Result {
if index(a.state, intent.Name) < 0 {
return Result{Snapshot: a.Snapshot(), Err: fmt.Errorf("Profile %q does not exist", intent.Name)}
}
if a.state.ActiveProfile == intent.Name {
return Result{Snapshot: a.Snapshot(), Notice: fmt.Sprintf("Profile %q is already Active", intent.Name)}
}
var enabled []target.Target
var warnings []string
for _, adapter := range a.targets {
if a.inspections[adapter.ID()].State != target.Enabled {
continue
}
enabled = append(enabled, adapter)
if warning := adapter.Confirmation(true); warning != "" {
warnings = append(warnings, adapter.Description().Name+": "+warning)
}
}
if len(warnings) > 0 && !intent.Confirm {
return Result{Snapshot: a.Snapshot(), Confirmation: "Changing the Active Profile requires confirmation:\n" + strings.Join(warnings, "\n")}
}
next, err := a.store.Mutate(func(current profile.State) (profile.State, error) {
next, _, mutationErr := activate(current, intent.Name)
return next, mutationErr
})
if err != nil {
var committed *profile.CommitError
if errors.As(err, &committed) {
a.state = next
if reloaded, loadErr := a.store.Load(); loadErr != nil {
err = errors.Join(err, fmt.Errorf("reload committed Profile configuration: %w", loadErr))
} else {
a.state = reloaded
}
}
return Result{Snapshot: a.Snapshot(), Err: err}
}
a.state = next
active := a.activeProfile()
// The Active Profile has changed before any Target write begins. Re-inspect
// immediately so rows not yet updated are truthfully Drifted rather than
// appearing Enabled against stale managed values.
for _, adapter := range a.targets {
a.inspections[adapter.ID()] = adapter.Inspect(context.Background(), active)
}
var progress []TargetProgress
var failures []error
for _, adapter := range enabled {
if publish != nil {
publish(Result{Snapshot: a.Snapshot(), Progress: []TargetProgress{{TargetID: adapter.ID(), Applying: true, State: a.inspections[adapter.ID()].State}}, Notice: "Updating " + adapter.Description().Name + " Target"})
}
operation := adapter.SetDesired(context.Background(), true, active)
a.inspections[adapter.ID()] = operation.Inspection
var safeErr error
if operation.Err != nil {
safeErr = errors.New(profile.RedactText(operation.Err.Error()))
}
item := TargetProgress{TargetID: adapter.ID(), State: operation.Inspection.State, Evidence: operation.Inspection.Evidence, Err: safeErr}
progress = append(progress, item)
if publish != nil {
publish(Result{Snapshot: a.Snapshot(), Progress: []TargetProgress{item}, Notice: adapter.Description().Name + " Target update completed", Err: safeErr})
}
if a.logger != nil {
outcome, detail := "success", operation.Inspection.Evidence
if safeErr != nil {
outcome, detail = "failure", safeErr.Error()
}
_ = a.logger.Write(LogEntry{Time: time.Now(), Operation: "switch-Active-Profile", Target: adapter.ID(), Result: outcome, Detail: RedactLogDetail(detail)})
}
if safeErr != nil {
failures = append(failures, fmt.Errorf("%s Target: %w", adapter.Description().Name, safeErr))
}
}
// Re-inspect Targets which were not part of the transition so any managed
// configuration is interpreted against the new Active Profile as well.
updated := make(map[string]bool, len(enabled))
for _, adapter := range enabled {
updated[adapter.ID()] = true
}
for _, adapter := range a.targets {
if !updated[adapter.ID()] {
a.inspections[adapter.ID()] = adapter.Inspect(context.Background(), active)
}
}
result := Result{Snapshot: a.Snapshot(), Progress: progress, Notice: fmt.Sprintf("Activated Profile %q", intent.Name)}
if len(failures) > 0 {
result.Err = errors.Join(failures...)
result.Notice = fmt.Sprintf("Activated Profile %q with %d Target update failure(s); successful updates were kept and no global rollback was attempted", intent.Name, len(failures))
}
return result
}
func (a *Application) activeProfile() *profile.Profile {
for i := range a.state.Profiles {
if a.state.Profiles[i].Name == a.state.ActiveProfile {
+227
View File
@@ -0,0 +1,227 @@
package app_test
import (
"context"
"errors"
"os"
"path/filepath"
"strings"
"testing"
"proxyctl/internal/app"
"proxyctl/internal/dockerclient"
"proxyctl/internal/profile"
"proxyctl/internal/shell"
"proxyctl/internal/target"
)
type coordinatedTarget struct {
id string
state target.State
applied []string
appliedName string
failName string
}
func (t *coordinatedTarget) ID() string { return t.id }
func (t *coordinatedTarget) Description() target.Description {
return target.Description{Name: t.id, Summary: "summary", Details: "effects, non-effects, ownership, permissions, and side effects"}
}
func (t *coordinatedTarget) Availability(context.Context) target.Availability {
return target.Availability{Available: true, Evidence: "detected"}
}
func (t *coordinatedTarget) Inspect(_ context.Context, active *profile.Profile) target.Inspection {
if t.state == target.Enabled && (active == nil || (t.appliedName != "" && active.Name != t.appliedName)) {
return target.Inspection{State: target.Drifted, Evidence: "managed configuration does not match the Active Profile"}
}
return target.Inspection{State: t.state, Evidence: "actual configuration"}
}
func (t *coordinatedTarget) Confirmation(bool) string { return "" }
func (t *coordinatedTarget) SetDesired(_ context.Context, enabled bool, active *profile.Profile) target.OperationResult {
if !enabled {
t.state = target.Disabled
return target.OperationResult{Inspection: target.Inspection{State: t.state, Evidence: "disabled"}}
}
t.applied = append(t.applied, active.Name)
if active.Name == t.failName {
t.state = target.Drifted
return target.OperationResult{Inspection: target.Inspection{State: t.state, Evidence: "re-inspected partial write"}, Err: errors.New("failed using http://alice:secret@proxy.example")}
}
t.state = target.Enabled
t.appliedName = active.Name
return target.OperationResult{Inspection: target.Inspection{State: t.state, Evidence: "matches " + active.Name}}
}
func TestActiveProfileSwitchCoordinatesEnabledTargetsWithoutRollback(t *testing.T) {
first := &coordinatedTarget{id: "first", state: target.Disabled}
second := &coordinatedTarget{id: "second", state: target.Disabled, failName: "new"}
application, err := app.NewWithTargets(profile.NewFileStore(filepath.Join(t.TempDir(), "profiles.json")), []target.Target{first, second})
if err != nil {
t.Fatal(err)
}
for _, name := range []string{"old", "new"} {
result := application.Dispatch(app.Intent{Kind: app.CreateProfile, Profile: profile.Profile{Name: name, SingleURL: "http://alice:secret@proxy.example:8080"}})
if result.Err != nil {
t.Fatal(result.Err)
}
}
if result := application.Dispatch(app.Intent{Kind: app.ActivateProfile, Name: "old"}); result.Err != nil {
t.Fatal(result.Err)
}
for _, id := range []string{"first", "second"} {
if result := application.Dispatch(app.Intent{Kind: app.SetTargetDesired, TargetID: id, Enabled: true}); result.Err != nil {
t.Fatal(result.Err)
}
}
result := application.Dispatch(app.Intent{Kind: app.ActivateProfile, Name: "new"})
if result.Snapshot.ActiveProfile != "new" {
t.Fatalf("Active Profile = %q", result.Snapshot.ActiveProfile)
}
if result.Err == nil || strings.Contains(result.Err.Error(), "secret") || !strings.Contains(result.Err.Error(), "second") {
t.Fatalf("redacted partial failure = %v", result.Err)
}
if got := result.Snapshot.Targets; got[0].State != target.Enabled || got[1].State != target.Drifted {
t.Fatalf("truthful partial states = %#v", got)
}
if len(result.Progress) != 2 || result.Progress[0].TargetID != "first" || result.Progress[0].State != target.Enabled || result.Progress[1].TargetID != "second" || result.Progress[1].State != target.Drifted {
t.Fatalf("per-Target progress = %#v", result.Progress)
}
if len(first.applied) != 2 || first.applied[1] != "new" {
t.Fatalf("successful Target was not updated: %#v", first.applied)
}
}
func TestApplicationCoordinatesMultipleRealAndFakeHermeticAdaptersThroughPartialFailure(t *testing.T) {
directory := t.TempDir()
shellAdapter := shell.New(shell.Configuration{
Shell: "/bin/bash", EnvironmentFile: filepath.Join(directory, "shell.env"), RecoveryFile: filepath.Join(directory, "shell.json"), TrustedRoots: []string{directory},
}, nil, shell.MapEnvironment{})
dockerAdapter := dockerclient.New(dockerclient.Configuration{
ConfigFile: filepath.Join(directory, "docker", "config.json"), RecoveryFile: filepath.Join(directory, "docker", "recovery.json"), TrustedRoots: []string{directory},
}, nil)
failure := &coordinatedTarget{id: "failing-fake", state: target.Disabled, failName: "new"}
application, err := app.NewWithTargets(profile.NewFileStore(filepath.Join(directory, "profiles.json")), []target.Target{shellAdapter, dockerAdapter, failure})
if err != nil {
t.Fatal(err)
}
for name, endpoint := range map[string]string{"old": "http://old-proxy.example:8080", "new": "http://new-proxy.example:9090"} {
if result := application.Dispatch(app.Intent{Kind: app.CreateProfile, Profile: profile.Profile{Name: name, SingleURL: endpoint}}); result.Err != nil {
t.Fatal(result.Err)
}
}
if result := application.Dispatch(app.Intent{Kind: app.ActivateProfile, Name: "old"}); result.Err != nil {
t.Fatal(result.Err)
}
for _, id := range []string{shellAdapter.ID(), dockerAdapter.ID(), failure.ID()} {
if result := application.Dispatch(app.Intent{Kind: app.SetTargetDesired, TargetID: id, Enabled: true}); result.Err != nil {
t.Fatalf("enable %s: %v", id, result.Err)
}
}
result := application.Dispatch(app.Intent{Kind: app.ActivateProfile, Name: "new"})
if result.Err == nil || result.Snapshot.ActiveProfile != "new" {
t.Fatalf("partial switch = %#v", result)
}
states := map[string]target.State{}
for _, item := range result.Snapshot.Targets {
states[item.ID] = item.State
}
if states["shell"] != target.Enabled || states["docker-containers"] != target.Enabled || states["failing-fake"] != target.Drifted {
t.Fatalf("real/fake Target states = %#v", states)
}
generated, err := os.ReadFile(filepath.Join(directory, "shell.env"))
if err != nil || !strings.Contains(string(generated), "new-proxy.example:9090") {
t.Fatalf("real Shell effect = %q, %v", generated, err)
}
dockerConfig, err := os.ReadFile(filepath.Join(directory, "docker", "config.json"))
if err != nil || !strings.Contains(string(dockerConfig), "new-proxy.example:9090") {
t.Fatalf("real Docker client effect = %q, %v", dockerConfig, err)
}
}
func TestActiveProfileSwitchStreamsApplyingAndCompletionForEachTarget(t *testing.T) {
adapter := &coordinatedTarget{id: "shell", state: target.Disabled}
application, err := app.NewWithTargets(profile.NewFileStore(filepath.Join(t.TempDir(), "profiles.json")), []target.Target{adapter})
if err != nil {
t.Fatal(err)
}
for _, name := range []string{"old", "new"} {
if result := application.Dispatch(app.Intent{Kind: app.CreateProfile, Profile: profile.Profile{Name: name, SingleURL: "http://proxy.example:8080"}}); result.Err != nil {
t.Fatal(result.Err)
}
}
if result := application.Dispatch(app.Intent{Kind: app.ActivateProfile, Name: "old"}); result.Err != nil {
t.Fatal(result.Err)
}
if result := application.Dispatch(app.Intent{Kind: app.SetTargetDesired, TargetID: "shell", Enabled: true}); result.Err != nil {
t.Fatal(result.Err)
}
updates := application.DispatchProgress(app.Intent{Kind: app.ActivateProfile, Name: "new"})
applying := <-updates
completed := <-updates
final := <-updates
if len(applying.Progress) != 1 || !applying.Progress[0].Applying || applying.Progress[0].TargetID != "shell" {
t.Fatalf("applying update = %#v", applying)
}
if len(completed.Progress) != 1 || completed.Progress[0].Applying || completed.Progress[0].State != target.Enabled {
t.Fatalf("completion update = %#v", completed)
}
if !final.Complete || final.Snapshot.ActiveProfile != "new" {
t.Fatalf("final update = %#v", final)
}
}
func TestDeletingFinalProfileLeavesManagedTargetsDriftedAndDisableStillRestoresExternal(t *testing.T) {
directory := t.TempDir()
configuration := dockerclient.Configuration{
ConfigFile: filepath.Join(directory, "docker", "config.json"), RecoveryFile: filepath.Join(directory, "state", "docker.json"), TrustedRoots: []string{directory},
}
if err := os.MkdirAll(filepath.Dir(configuration.ConfigFile), 0o700); err != nil {
t.Fatal(err)
}
const externalConfig = `{"proxies":{"default":{"httpProxy":"http://external.example:3128"}},"auths":{"registry.example":{}}}`
if err := os.WriteFile(configuration.ConfigFile, []byte(externalConfig), 0o600); err != nil {
t.Fatal(err)
}
adapter := dockerclient.New(configuration, nil)
application, err := app.NewWithTargets(profile.NewFileStore(filepath.Join(directory, "profiles.json")), []target.Target{adapter})
if err != nil {
t.Fatal(err)
}
created := application.Dispatch(app.Intent{Kind: app.CreateProfile, Profile: profile.Profile{Name: "only", SingleURL: "http://proxy.example:8080"}})
if created.Err != nil {
t.Fatal(created.Err)
}
if activated := application.Dispatch(app.Intent{Kind: app.ActivateProfile, Name: "only"}); activated.Err != nil {
t.Fatal(activated.Err)
}
if enabled := application.Dispatch(app.Intent{Kind: app.SetTargetDesired, TargetID: adapter.ID(), Enabled: true}); enabled.Err != nil {
t.Fatal(enabled.Err)
}
unconfirmed := application.Dispatch(app.Intent{Kind: app.DeleteProfile, Name: "only"})
if unconfirmed.Err == nil || unconfirmed.Snapshot.ActiveProfile != "only" {
t.Fatalf("unconfirmed final delete = %#v", unconfirmed)
}
deleted := application.Dispatch(app.Intent{Kind: app.DeleteProfile, Name: "only", Confirm: true})
if deleted.Err != nil || deleted.Snapshot.ActiveProfile != "" || deleted.Snapshot.Targets[0].State != target.Drifted {
t.Fatalf("confirmed final delete = %#v", deleted)
}
enable := application.Dispatch(app.Intent{Kind: app.SetTargetDesired, TargetID: adapter.ID(), Enabled: true})
if enable.Err == nil || !strings.Contains(enable.Err.Error(), "Profile management") {
t.Fatalf("re-enable without Active Profile = %v", enable.Err)
}
disabled := application.Dispatch(app.Intent{Kind: app.SetTargetDesired, TargetID: adapter.ID(), Enabled: false})
if disabled.Err != nil || disabled.Snapshot.Targets[0].State != target.External {
t.Fatalf("disable after final delete = %#v", disabled)
}
restored, err := os.ReadFile(configuration.ConfigFile)
if err != nil {
t.Fatal(err)
}
if !strings.Contains(string(restored), "external.example:3128") || !strings.Contains(string(restored), "registry.example") {
t.Fatalf("External Docker client configuration was not restored: %s", restored)
}
}