mirror of
https://github.com/rcourtman/Pulse.git
synced 2026-10-24 04:33:17 +00:00
Merge provider MSP client recovery onto current main
Combine the current provider setup correction with the isolated-network recovery candidate. Retain both deployment contracts and preserve the client-restart and support-container ownership guards. Change-source: pulse-maintainer
This commit is contained in:
@@ -3482,6 +3482,21 @@ coverage:
|
||||
`TestProviderMSPWorkspaceLadderRisesFromTheEvaluation` in
|
||||
`pkg/licensing/features_test.go`.
|
||||
|
||||
### Provider MSP health monitor recovers clients across upgrades
|
||||
|
||||
The health monitor rejoins Traefik and the control plane to each active
|
||||
client's labelled isolated network on startup and before every health check.
|
||||
Recreating either support container drops its tenant-network attachments;
|
||||
without reconnection, routes fail and a healthy client can appear unreachable.
|
||||
The monitor now restarts an unhealthy client with Docker's restart operation,
|
||||
rather than stopping it: `unless-stopped` does not revive a container stopped
|
||||
through the API. Missing isolated networks are skipped; a network whose labels
|
||||
do not identify that tenant is refused rather than connected. Support containers
|
||||
must also have the role label and belong to the configured provider ingress
|
||||
network, avoiding a same-host provider stack with the same role label. Regression
|
||||
coverage is in `internal/cloudcp/health_monitor_test.go` and
|
||||
`internal/cloudcp/docker/manager_test.go`.
|
||||
|
||||
### Provider MSP status reads client health that any caller can observe
|
||||
|
||||
`provider-msp status`, which `upgrade.sh` gates on before every provider
|
||||
|
||||
@@ -2430,6 +2430,16 @@ which have working defaults or are generated. PR #2212 reports a v6.4.1
|
||||
first-run 404 and a 200 after starting the platform; this assigned candidate
|
||||
has not repeated that live-host proof.
|
||||
|
||||
### Provider MSP clients survive a support-container recreate
|
||||
|
||||
Recreating the control plane or Traefik detaches it from each client's
|
||||
isolated network. The health monitor now reattaches both support containers
|
||||
on startup and each pass, and restarts rather than stops an unhealthy client.
|
||||
The original PR reports a two-client v6.4.1 lab reproduction and recovery on
|
||||
the same installation on 2026-09-23; that is not installed acceptance of this
|
||||
integrated source. The network reconnection requires exact tenant ownership
|
||||
labels before mutating an existing network.
|
||||
|
||||
### Provider MSP operations accept a renewed licence
|
||||
|
||||
`provider-msp preflight`, `proof`, `recover` and `backup` treat a licence
|
||||
|
||||
@@ -409,8 +409,15 @@ func (m *Manager) connectSupportContainersToTenantNetwork(ctx context.Context, n
|
||||
}
|
||||
|
||||
func (m *Manager) connectSupportContainersByLabel(ctx context.Context, networkName, label string) error {
|
||||
providerNetwork := strings.TrimSpace(m.cfg.Network)
|
||||
if providerNetwork == "" {
|
||||
return fmt.Errorf("provider ingress network is required to select support containers")
|
||||
}
|
||||
filters := client.Filters{}
|
||||
filters = filters.Add("label", label)
|
||||
// Role labels are shared by every provider MSP installation on a host.
|
||||
// Never attach another installation's support container to this tenant.
|
||||
filters = filters.Add("network", providerNetwork)
|
||||
result, err := m.cli.ContainerList(ctx, client.ContainerListOptions{
|
||||
Filters: filters,
|
||||
})
|
||||
@@ -863,6 +870,46 @@ func (m *Manager) Stop(ctx context.Context, containerID string) error {
|
||||
return err
|
||||
}
|
||||
|
||||
// Restart stops and starts a container in one daemon call. The health
|
||||
// monitor relies on it to recover an unhealthy tenant: a container stopped
|
||||
// through the API is never brought back by the unless-stopped restart
|
||||
// policy, so stopping it left the client workspace down for good.
|
||||
func (m *Manager) Restart(ctx context.Context, containerID string) error {
|
||||
timeout := 30
|
||||
_, err := m.cli.ContainerRestart(ctx, containerID, client.ContainerRestartOptions{Timeout: &timeout})
|
||||
return err
|
||||
}
|
||||
|
||||
// EnsureSupportContainersOnTenantNetwork reattaches the provider support
|
||||
// containers (Traefik and the control plane) to a tenant's isolated network.
|
||||
// They join it when the tenant is created, but recreating either one, as an
|
||||
// upgrade or any compose change does, drops the attachment: Traefik then has
|
||||
// no route to the client and the control plane cannot reach it. A tenant
|
||||
// without an isolated network has nothing to reattach.
|
||||
func (m *Manager) EnsureSupportContainersOnTenantNetwork(ctx context.Context, tenantID string) error {
|
||||
if m == nil || m.cli == nil || !m.cfg.IsolateTenantNetworks || strings.TrimSpace(tenantID) == "" {
|
||||
return nil
|
||||
}
|
||||
networkName := m.tenantNetworkName(tenantID)
|
||||
if networkName == "" {
|
||||
return nil
|
||||
}
|
||||
inspected, err := m.cli.NetworkInspect(ctx, networkName, client.NetworkInspectOptions{})
|
||||
if err != nil {
|
||||
if errdefs.IsNotFound(err) {
|
||||
return nil
|
||||
}
|
||||
return fmt.Errorf("inspect tenant network %q: %w", networkName, err)
|
||||
}
|
||||
// A name alone does not establish ownership. Never attach provider support
|
||||
// containers to an unrelated network that happens to have the derived name.
|
||||
labels := inspected.Network.Labels
|
||||
if labels["pulse.tenant.id"] != tenantID || labels[tenantRuntimeNetworkLabel] != tenantRuntimeNetworkLabelValue {
|
||||
return fmt.Errorf("network %q is not the isolated network for tenant %q", networkName, tenantID)
|
||||
}
|
||||
return m.connectSupportContainersToTenantNetwork(ctx, networkName)
|
||||
}
|
||||
|
||||
// Start starts a stopped tenant container.
|
||||
func (m *Manager) Start(ctx context.Context, containerID string) error {
|
||||
_, err := m.cli.ContainerStart(ctx, containerID, client.ContainerStartOptions{})
|
||||
|
||||
@@ -2,6 +2,8 @@ package docker
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
@@ -603,3 +605,72 @@ func TestDockerRecordedHealthDecidesWhenTheImageDeclaresOne(t *testing.T) {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// A client created before per-client networks has no isolated network to
|
||||
// rejoin. Reconciling it must be a quiet no-op, not a warning every minute.
|
||||
func TestEnsureSupportContainersOnTenantNetworkSkipsMissingNetwork(t *testing.T) {
|
||||
var mutations []string
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Api-Version", "1.47")
|
||||
if strings.HasSuffix(r.URL.Path, "/_ping") {
|
||||
_, _ = w.Write([]byte("OK"))
|
||||
return
|
||||
}
|
||||
if r.Method != http.MethodGet {
|
||||
mutations = append(mutations, r.Method+" "+r.URL.Path)
|
||||
}
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
w.WriteHeader(http.StatusNotFound)
|
||||
_, _ = w.Write([]byte(`{"message":"network not found"}`))
|
||||
}))
|
||||
t.Cleanup(srv.Close)
|
||||
t.Setenv("DOCKER_HOST", "tcp://"+strings.TrimPrefix(srv.URL, "http://"))
|
||||
t.Setenv("DOCKER_TLS_VERIFY", "")
|
||||
t.Setenv("DOCKER_CERT_PATH", "")
|
||||
|
||||
mgr, err := NewManager(ManagerConfig{Image: "pulse:test", Network: "pulse-provider-msp", IsolateTenantNetworks: true})
|
||||
if err != nil {
|
||||
t.Fatalf("NewManager: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { _ = mgr.Close() })
|
||||
|
||||
if err := mgr.EnsureSupportContainersOnTenantNetwork(context.Background(), "t-legacy"); err != nil {
|
||||
t.Fatalf("EnsureSupportContainersOnTenantNetwork = %v, want nil for a client without an isolated network", err)
|
||||
}
|
||||
if len(mutations) != 0 {
|
||||
t.Fatalf("reconcile changed the host for a missing network: %v", mutations)
|
||||
}
|
||||
}
|
||||
|
||||
func TestEnsureSupportContainersOnTenantNetworkRejectsWrongOwner(t *testing.T) {
|
||||
var mutations []string
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Api-Version", "1.47")
|
||||
if strings.HasSuffix(r.URL.Path, "/_ping") {
|
||||
_, _ = w.Write([]byte("OK"))
|
||||
return
|
||||
}
|
||||
if r.Method != http.MethodGet {
|
||||
mutations = append(mutations, r.Method+" "+r.URL.Path)
|
||||
}
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
_, _ = w.Write([]byte(`{"Name":"pulse-provider-msp-tenant-t-acme","Id":"net-1","Labels":{"pulse.tenant.id":"t-other","pulse.provider-msp.network":"tenant"}}`))
|
||||
}))
|
||||
t.Cleanup(srv.Close)
|
||||
t.Setenv("DOCKER_HOST", "tcp://"+strings.TrimPrefix(srv.URL, "http://"))
|
||||
t.Setenv("DOCKER_TLS_VERIFY", "")
|
||||
t.Setenv("DOCKER_CERT_PATH", "")
|
||||
|
||||
mgr, err := NewManager(ManagerConfig{Image: "pulse:test", Network: "pulse-provider-msp", IsolateTenantNetworks: true})
|
||||
if err != nil {
|
||||
t.Fatalf("NewManager: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { _ = mgr.Close() })
|
||||
|
||||
if err := mgr.EnsureSupportContainersOnTenantNetwork(context.Background(), "t-acme"); err == nil {
|
||||
t.Fatal("expected wrong-owner network to be rejected")
|
||||
}
|
||||
if len(mutations) != 0 {
|
||||
t.Fatalf("reconcile changed wrong-owner network: %v", mutations)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -49,6 +49,12 @@ func (m *Monitor) Run(ctx context.Context) {
|
||||
Bool("restart_on_fail", m.cfg.RestartOnFail).
|
||||
Msg("Health monitor started")
|
||||
|
||||
// Check once straight away. The control plane starts fresh after every
|
||||
// upgrade, and the first pass is what puts the recreated support
|
||||
// containers back on each client's network; waiting a full interval
|
||||
// leaves every client route broken for that long.
|
||||
m.checkAll(ctx)
|
||||
|
||||
ticker := time.NewTicker(m.cfg.Interval)
|
||||
defer ticker.Stop()
|
||||
|
||||
@@ -64,6 +70,9 @@ func (m *Monitor) Run(ctx context.Context) {
|
||||
}
|
||||
|
||||
func (m *Monitor) checkAll(ctx context.Context) {
|
||||
if ctx.Err() != nil {
|
||||
return
|
||||
}
|
||||
activeContainerIDs := make(map[string]struct{})
|
||||
tenants, err := m.registry.ListByState(registry.TenantStateActive)
|
||||
if err != nil {
|
||||
@@ -80,6 +89,16 @@ func (m *Monitor) checkAll(ctx context.Context) {
|
||||
}
|
||||
activeContainerIDs[tenant.ContainerID] = struct{}{}
|
||||
|
||||
// Rejoin Traefik and the control plane to this client's network
|
||||
// first. A recreated support container (every upgrade recreates the
|
||||
// control plane) has lost it, which cuts the client's route and
|
||||
// would make a healthy client look unreachable.
|
||||
if err := m.docker.EnsureSupportContainersOnTenantNetwork(ctx, tenant.ID); err != nil {
|
||||
log.Warn().Err(err).
|
||||
Str("tenant_id", tenant.ID).
|
||||
Msg("Health monitor: could not reattach support containers to tenant network")
|
||||
}
|
||||
|
||||
healthy, err := m.docker.HealthCheck(ctx, tenant.ContainerID)
|
||||
if err != nil {
|
||||
log.Warn().Err(err).
|
||||
@@ -117,14 +136,16 @@ func (m *Monitor) checkAll(ctx context.Context) {
|
||||
Int("fail_threshold", m.cfg.FailThreshold).
|
||||
Msg("Container unhealthy, attempting restart")
|
||||
|
||||
if err := m.docker.Stop(ctx, tenant.ContainerID); err != nil {
|
||||
// Restart, never Stop: the unless-stopped policy does not bring
|
||||
// back a container stopped through the API, so a stop took the
|
||||
// client workspace down until someone started it by hand.
|
||||
if err := m.docker.Restart(ctx, tenant.ContainerID); err != nil {
|
||||
log.Error().
|
||||
Err(err).
|
||||
Str("tenant_id", tenant.ID).
|
||||
Str("container_id", tenant.ContainerID).
|
||||
Msg("Failed to stop unhealthy container")
|
||||
Msg("Failed to restart unhealthy container")
|
||||
}
|
||||
// Docker restart policy (unless-stopped) will restart the container
|
||||
m.failures[tenant.ContainerID] = 0
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,206 @@
|
||||
package cloudcp
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"regexp"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/rcourtman/pulse-go-rewrite/internal/cloudcp/docker"
|
||||
"github.com/rcourtman/pulse-go-rewrite/internal/cloudcp/registry"
|
||||
)
|
||||
|
||||
// fakeProviderDaemon answers the few Docker Engine API calls the health
|
||||
// monitor makes for one provider-hosted client workspace, and records every
|
||||
// mutating call so a test can assert what the monitor did to the host.
|
||||
type fakeProviderDaemon struct {
|
||||
mu sync.Mutex
|
||||
tenantNetwork string
|
||||
attached map[string]bool // container IDs on the tenant network
|
||||
support map[string]string
|
||||
health string
|
||||
calls []string
|
||||
healthChecked chan struct{}
|
||||
}
|
||||
|
||||
var dockerAPIVersionPrefix = regexp.MustCompile(`^/v[0-9.]+`)
|
||||
|
||||
func (d *fakeProviderDaemon) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
||||
d.mu.Lock()
|
||||
defer d.mu.Unlock()
|
||||
|
||||
path := dockerAPIVersionPrefix.ReplaceAllString(r.URL.Path, "")
|
||||
w.Header().Set("Api-Version", "1.47")
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
switch {
|
||||
case path == "/_ping":
|
||||
_, _ = w.Write([]byte("OK"))
|
||||
case r.Method == http.MethodGet && path == "/networks/"+d.tenantNetwork:
|
||||
containers := map[string]any{}
|
||||
for id := range d.attached {
|
||||
containers[id] = map[string]any{"Name": id}
|
||||
}
|
||||
_ = json.NewEncoder(w).Encode(map[string]any{
|
||||
"Name": d.tenantNetwork, "Id": "net-1", "Containers": containers,
|
||||
"Labels": map[string]string{"pulse.tenant.id": "t-acme", "pulse.provider-msp.network": "tenant"},
|
||||
})
|
||||
case r.Method == http.MethodPost && path == "/networks/"+d.tenantNetwork+"/connect":
|
||||
var body struct{ Container string }
|
||||
_ = json.NewDecoder(r.Body).Decode(&body)
|
||||
d.attached[body.Container] = true
|
||||
d.calls = append(d.calls, "connect "+body.Container)
|
||||
w.WriteHeader(http.StatusOK)
|
||||
case r.Method == http.MethodGet && path == "/containers/json":
|
||||
filters := r.URL.Query().Get("filters")
|
||||
if !strings.Contains(filters, `"network"`) || !strings.Contains(filters, "pulse-provider-msp") {
|
||||
w.WriteHeader(http.StatusBadRequest)
|
||||
_, _ = w.Write([]byte(`{"message":"support containers must be selected from the provider ingress network"}`))
|
||||
return
|
||||
}
|
||||
var items []map[string]any
|
||||
for label, id := range d.support {
|
||||
if strings.Contains(filters, label) {
|
||||
items = append(items, map[string]any{"Id": id, "State": "running"})
|
||||
}
|
||||
}
|
||||
_ = json.NewEncoder(w).Encode(items)
|
||||
case r.Method == http.MethodGet && strings.HasPrefix(path, "/containers/") && strings.HasSuffix(path, "/json"):
|
||||
if d.healthChecked != nil {
|
||||
select {
|
||||
case d.healthChecked <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
}
|
||||
_ = json.NewEncoder(w).Encode(map[string]any{
|
||||
"Id": strings.TrimSuffix(strings.TrimPrefix(path, "/containers/"), "/json"),
|
||||
"State": map[string]any{"Status": "running", "Running": true, "Health": map[string]any{"Status": d.health}},
|
||||
"Config": map[string]any{"Labels": map[string]string{}},
|
||||
})
|
||||
case r.Method == http.MethodPost && strings.HasPrefix(path, "/containers/"):
|
||||
d.calls = append(d.calls, strings.TrimPrefix(path, "/containers/"))
|
||||
w.WriteHeader(http.StatusNoContent)
|
||||
default:
|
||||
w.WriteHeader(http.StatusNotFound)
|
||||
_ = json.NewEncoder(w).Encode(map[string]string{"message": "no such object: " + path})
|
||||
}
|
||||
}
|
||||
|
||||
func (d *fakeProviderDaemon) recorded() []string {
|
||||
d.mu.Lock()
|
||||
defer d.mu.Unlock()
|
||||
return append([]string(nil), d.calls...)
|
||||
}
|
||||
|
||||
// An upgrade recreates the control plane (and often Traefik), and the new
|
||||
// containers are not on any client's isolated network. The monitor used to
|
||||
// read that as a failing client and, three minutes later, Stop it; Docker's
|
||||
// unless-stopped policy never restarts a container stopped through the API,
|
||||
// so every client workspace stayed down. The monitor must put the support
|
||||
// containers back on the network, and when a client really is unhealthy it
|
||||
// must restart it rather than stop it.
|
||||
func TestHealthMonitorReattachesSupportContainersAndRestartsInsteadOfStopping(t *testing.T) {
|
||||
daemon := &fakeProviderDaemon{
|
||||
tenantNetwork: "pulse-provider-msp-tenant-t-acme",
|
||||
attached: map[string]bool{"tenant-container": true},
|
||||
support: map[string]string{
|
||||
"pulse.provider-msp.role=traefik": "traefik-recreated",
|
||||
"pulse.provider-msp.role=control-plane": "control-plane-recreated",
|
||||
},
|
||||
health: "healthy",
|
||||
healthChecked: make(chan struct{}, 1),
|
||||
}
|
||||
srv := httptest.NewServer(daemon)
|
||||
t.Cleanup(srv.Close)
|
||||
t.Setenv("DOCKER_HOST", "tcp://"+strings.TrimPrefix(srv.URL, "http://"))
|
||||
t.Setenv("DOCKER_TLS_VERIFY", "")
|
||||
t.Setenv("DOCKER_CERT_PATH", "")
|
||||
|
||||
mgr, err := docker.NewManager(docker.ManagerConfig{
|
||||
Image: "pulse:test",
|
||||
Network: "pulse-provider-msp",
|
||||
IsolateTenantNetworks: true,
|
||||
TenantNetworkPrefix: "pulse-provider-msp-tenant",
|
||||
BaseDomain: "msp.example.com",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("NewManager: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { _ = mgr.Close() })
|
||||
|
||||
reg, err := registry.NewTenantRegistry(t.TempDir())
|
||||
if err != nil {
|
||||
t.Fatalf("NewTenantRegistry: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { _ = reg.Close() })
|
||||
if err := reg.Create(®istry.Tenant{ID: "t-acme", AccountID: "a-1", State: registry.TenantStateActive, ContainerID: "tenant-container"}); err != nil {
|
||||
t.Fatalf("Create tenant: %v", err)
|
||||
}
|
||||
|
||||
monitor := NewMonitor(reg, mgr, MonitorConfig{Interval: time.Hour, RestartOnFail: true, FailThreshold: 3})
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
monitor.Run(ctx)
|
||||
close(done)
|
||||
}()
|
||||
select {
|
||||
case <-daemon.healthChecked:
|
||||
case <-time.After(5 * time.Second):
|
||||
cancel()
|
||||
t.Fatal("monitor did not check the tenant on startup")
|
||||
}
|
||||
deadline := time.Now().Add(5 * time.Second)
|
||||
for {
|
||||
observed, err := reg.Get("t-acme")
|
||||
if err == nil && observed != nil && observed.HealthCheckOK {
|
||||
break
|
||||
}
|
||||
if time.Now().After(deadline) {
|
||||
cancel()
|
||||
t.Fatalf("monitor did not persist healthy result on startup: tenant=%+v err=%v", observed, err)
|
||||
}
|
||||
time.Sleep(5 * time.Millisecond)
|
||||
}
|
||||
cancel()
|
||||
select {
|
||||
case <-done:
|
||||
case <-time.After(5 * time.Second):
|
||||
t.Fatal("monitor did not stop after cancellation")
|
||||
}
|
||||
|
||||
got := strings.Join(daemon.recorded(), ", ")
|
||||
for _, want := range []string{"connect traefik-recreated", "connect control-plane-recreated"} {
|
||||
if !strings.Contains(got, want) {
|
||||
t.Fatalf("after one pass the daemon saw %q, want %q", got, want)
|
||||
}
|
||||
}
|
||||
tenant, err := reg.Get("t-acme")
|
||||
if err != nil || tenant == nil || !tenant.HealthCheckOK {
|
||||
t.Fatalf("tenant after healthy pass = %+v, err %v; want HealthCheckOK", tenant, err)
|
||||
}
|
||||
|
||||
// Already attached: later passes change nothing on the network.
|
||||
monitor.checkAll(context.Background())
|
||||
if again := daemon.recorded(); len(again) != 2 {
|
||||
t.Fatalf("second pass reconnected attached containers: %v", again)
|
||||
}
|
||||
|
||||
daemon.mu.Lock()
|
||||
daemon.health = "unhealthy"
|
||||
daemon.mu.Unlock()
|
||||
for range 3 {
|
||||
monitor.checkAll(context.Background())
|
||||
}
|
||||
got = strings.Join(daemon.recorded(), ", ")
|
||||
if !strings.Contains(got, "tenant-container/restart") {
|
||||
t.Fatalf("unhealthy client was not restarted; daemon saw %q", got)
|
||||
}
|
||||
if strings.Contains(got, "tenant-container/stop") {
|
||||
t.Fatalf("unhealthy client was stopped, which unless-stopped never undoes; daemon saw %q", got)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user