fix(cloudcp): isolate provider MSP recovery per ingress

Merge the reviewed recovery candidate onto current main while retaining the already landed PR implementation. Filter shared support roles by the configured provider network, verify tenant-network ownership before attachment, and cover two same-host providers plus startup reconciliation.

Change-source: pulse-maintainer
This commit is contained in:
pulse-triage[bot]
2026-09-24 03:21:59 +01:00
5 changed files with 302 additions and 16 deletions
@@ -3500,6 +3500,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
@@ -2432,16 +2432,14 @@ has not repeated that live-host proof.
### Provider MSP clients survive a support-container recreate
Recreating the control plane or Traefik (every `upgrade.sh` run recreates the
control plane) used to detach them from each client's isolated network. The
health monitor then saw every client as failing and stopped it, and nothing
restarted it, so an upgrade on a v6.4.1 install with clients took all of them
offline. The monitor now reattaches the support containers on startup and on
every pass, and restarts rather than stops an unhealthy client. Verified on
2026-09-23 on a v6.4.1 provider bundle with two clients: a control-plane
recreate rejoined both client networks within a second, a Traefik-only
recreate had its client routes back on the next 60-second pass, and neither
client was stopped.
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. Reconnection selects support containers from the configured
provider ingress network and requires exact tenant ownership labels before
mutating an existing network.
### Provider MSP operations accept a renewed licence
+15 -1
View File
@@ -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,
})
@@ -887,12 +894,19 @@ func (m *Manager) EnsureSupportContainersOnTenantNetwork(ctx context.Context, te
if networkName == "" {
return nil
}
if _, err := m.cli.NetworkInspect(ctx, networkName, client.NetworkInspectOptions{}); err != 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)
}
+33
View File
@@ -641,3 +641,36 @@ func TestEnsureSupportContainersOnTenantNetworkSkipsMissingNetwork(t *testing.T)
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)
}
}
+231 -5
View File
@@ -9,6 +9,7 @@ import (
"strings"
"sync"
"testing"
"time"
"github.com/rcourtman/pulse-go-rewrite/internal/cloudcp/docker"
"github.com/rcourtman/pulse-go-rewrite/internal/cloudcp/registry"
@@ -24,6 +25,7 @@ type fakeProviderDaemon struct {
support map[string]string
health string
calls []string
healthChecked chan struct{}
}
var dockerAPIVersionPrefix = regexp.MustCompile(`^/v[0-9.]+`)
@@ -43,7 +45,10 @@ func (d *fakeProviderDaemon) ServeHTTP(w http.ResponseWriter, r *http.Request) {
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})
_ = 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)
@@ -51,14 +56,26 @@ func (d *fakeProviderDaemon) ServeHTTP(w http.ResponseWriter, r *http.Request) {
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(r.URL.Query().Get("filters"), label) {
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}},
@@ -94,7 +111,8 @@ func TestHealthMonitorReattachesSupportContainersAndRestartsInsteadOfStopping(t
"pulse.provider-msp.role=traefik": "traefik-recreated",
"pulse.provider-msp.role=control-plane": "control-plane-recreated",
},
health: "healthy",
health: "healthy",
healthChecked: make(chan struct{}, 1),
}
srv := httptest.NewServer(daemon)
t.Cleanup(srv.Close)
@@ -123,8 +141,37 @@ func TestHealthMonitorReattachesSupportContainersAndRestartsInsteadOfStopping(t
t.Fatalf("Create tenant: %v", err)
}
monitor := NewMonitor(reg, mgr, MonitorConfig{RestartOnFail: true, FailThreshold: 3})
monitor.checkAll(context.Background())
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"} {
@@ -157,3 +204,182 @@ func TestHealthMonitorReattachesSupportContainersAndRestartsInsteadOfStopping(t
t.Fatalf("unhealthy client was stopped, which unless-stopped never undoes; daemon saw %q", got)
}
}
// Two provider installations can share a Docker daemon and the same support
// role labels. A recreated container must only join the tenant network of the
// installation whose ingress network already contains it.
type multiProviderDaemon struct {
mu sync.Mutex
networks map[string]*multiProviderNetwork
support map[string]multiProviderSupport
health map[string]string
calls []string
}
type multiProviderNetwork struct {
tenantID string
attached map[string]bool
}
type multiProviderSupport struct {
role string
ingress string
}
func (d *multiProviderDaemon) 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 && strings.HasPrefix(path, "/networks/"):
name := strings.TrimPrefix(path, "/networks/")
net, ok := d.networks[name]
if !ok {
w.WriteHeader(http.StatusNotFound)
_, _ = w.Write([]byte(`{"message":"network not found"}`))
return
}
containers := make(map[string]any, len(net.attached))
for id := range net.attached {
containers[id] = map[string]any{"Name": id}
}
labels := map[string]string{}
if net.tenantID != "" {
labels = map[string]string{"pulse.tenant.id": net.tenantID, "pulse.provider-msp.network": "tenant"}
}
_ = json.NewEncoder(w).Encode(map[string]any{
"Name": name, "Id": name, "Containers": containers, "Labels": labels,
})
case r.Method == http.MethodPost && strings.HasPrefix(path, "/networks/") && strings.HasSuffix(path, "/connect"):
name := strings.TrimSuffix(strings.TrimPrefix(path, "/networks/"), "/connect")
var body struct{ Container string }
if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
w.WriteHeader(http.StatusBadRequest)
return
}
if _, ok := d.networks[name]; !ok {
w.WriteHeader(http.StatusNotFound)
return
}
d.networks[name].attached[body.Container] = true
d.calls = append(d.calls, "connect "+body.Container+" "+name)
w.WriteHeader(http.StatusOK)
case r.Method == http.MethodGet && path == "/containers/json":
var filters map[string]map[string]bool
if err := json.Unmarshal([]byte(r.URL.Query().Get("filters")), &filters); err != nil {
w.WriteHeader(http.StatusBadRequest)
return
}
var items []map[string]any
for id, support := range d.support {
if matchesDockerFilter(filters["label"], support.role) && matchesDockerFilter(filters["network"], support.ingress) {
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"):
id := strings.TrimSuffix(strings.TrimPrefix(path, "/containers/"), "/json")
_ = json.NewEncoder(w).Encode(map[string]any{
"Id": id,
"State": map[string]any{"Status": "running", "Running": true, "Health": map[string]any{"Status": d.health[id]}},
"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)
_, _ = w.Write([]byte(`{"message":"no such object"}`))
}
}
func matchesDockerFilter(values map[string]bool, candidate string) bool {
if len(values) == 0 {
return true
}
return values[candidate]
}
func (d *multiProviderDaemon) recorded() []string {
d.mu.Lock()
defer d.mu.Unlock()
return append([]string(nil), d.calls...)
}
func TestHealthMonitorKeepsProviderInstallationsIsolatedOnRecovery(t *testing.T) {
daemon := &multiProviderDaemon{
networks: map[string]*multiProviderNetwork{
"provider-a-tenant-t-alpha": {tenantID: "t-alpha", attached: map[string]bool{"client-alpha": true}},
"provider-b-tenant-t-beta": {tenantID: "t-beta", attached: map[string]bool{"client-beta": true}},
},
support: map[string]multiProviderSupport{
"traefik-a": {role: "pulse.provider-msp.role=traefik", ingress: "provider-a"},
"control-a": {role: "pulse.provider-msp.role=control-plane", ingress: "provider-a"},
"traefik-b": {role: "pulse.provider-msp.role=traefik", ingress: "provider-b"},
"control-b": {role: "pulse.provider-msp.role=control-plane", ingress: "provider-b"},
},
health: map[string]string{"client-alpha": "healthy", "client-beta": "unhealthy"},
}
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", "")
for _, client := range []struct {
provider, tenantID, containerID string
passes int
}{
{provider: "provider-a", tenantID: "t-alpha", containerID: "client-alpha", passes: 1},
{provider: "provider-b", tenantID: "t-beta", containerID: "client-beta", passes: 3},
} {
mgr, err := docker.NewManager(docker.ManagerConfig{
Image: "pulse:test", Network: client.provider, IsolateTenantNetworks: true,
TenantNetworkPrefix: client.provider + "-tenant", BaseDomain: "msp.example.com",
})
if err != nil {
t.Fatalf("NewManager(%s): %v", client.provider, err)
}
t.Cleanup(func() { _ = mgr.Close() })
reg, err := registry.NewTenantRegistry(t.TempDir())
if err != nil {
t.Fatalf("NewTenantRegistry(%s): %v", client.provider, err)
}
t.Cleanup(func() { _ = reg.Close() })
if err := reg.Create(&registry.Tenant{
ID: client.tenantID, AccountID: client.provider, State: registry.TenantStateActive,
ContainerID: client.containerID,
}); err != nil {
t.Fatalf("Create(%s): %v", client.tenantID, err)
}
monitor := NewMonitor(reg, mgr, MonitorConfig{RestartOnFail: true, FailThreshold: 3})
for range client.passes {
monitor.checkAll(context.Background())
}
}
got := daemon.recorded()
want := map[string]int{
"connect traefik-a provider-a-tenant-t-alpha": 1,
"connect control-a provider-a-tenant-t-alpha": 1,
"connect traefik-b provider-b-tenant-t-beta": 1,
"connect control-b provider-b-tenant-t-beta": 1,
"client-beta/restart": 1,
}
for _, call := range got {
if want[call] == 0 {
t.Fatalf("unexpected Docker mutation %q (including cross-client attachment or stop); all=%v", call, got)
}
want[call]--
}
for call, count := range want {
if count != 0 {
t.Fatalf("missing Docker mutation %q; all=%v", call, got)
}
}
}