api/controller/cleanup_registryproxy_test.go
Rama main · réplica de lectura del repositorio original.
package controller
import (
"bytes"
"context"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"io"
"net"
"net/http"
"net/http/httptest"
"net/url"
"os"
"os/exec"
"path/filepath"
"runtime"
"strings"
"sync"
"sync/atomic"
"testing"
"time"
"github.com/go-chi/chi"
"github.com/goodrain/rainbond/api/middleware"
"github.com/goodrain/rainbond/db/model"
guard "github.com/goodrain/rainbond/pkg/cleanup"
"github.com/goodrain/rainbond/pkg/cleanup/registryproxy"
"github.com/jinzhu/gorm"
)
// This owns every process, database, image and file. The fixture's ready state
// proves isolated protocol behavior, not enrollment of production participants.
// capability_id: rainbond.cleanup.coordinated-registry-delete-gc
func TestCoordinatedRegistryRealDeletionAndGC(t *testing.T) {
runCoordinatedRegistryGC(t, false)
}
// capability_id: rainbond.cleanup.registry-gc-executor
func TestCoordinatedRegistryExecutorRealDeletionAndGC(t *testing.T) {
if runtime.GOOS != "linux" {
t.Skip("Linux descriptor-pinned GC executor requires /proc/self/fd")
}
runCoordinatedRegistryGC(t, true)
}
func runCoordinatedRegistryGC(t *testing.T, useExecutor bool) {
binary := os.Getenv("CLEANUP_TEST_REGISTRY_BINARY")
if binary == "" {
t.Skip("set CLEANUP_TEST_REGISTRY_BINARY to an official local Distribution binary")
}
listener, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatal(err)
}
address := listener.Addr().String()
listener.Close()
root := t.TempDir()
storage := filepath.Join(root, "storage")
if err := os.MkdirAll(storage, 0755); err != nil {
t.Fatal(err)
}
storageHash := sha256.Sum256([]byte("registry-filesystem\x00owned-test-volume"))
storageID := hex.EncodeToString(storageHash[:])
measurementBinding := guard.StorageRegistration{StorageID: storageID, Generation: "one", VolumeUID: "owned-test-volume", RootPath: storage}
if err := registryproxy.InitializeStorageIdentity(storage, measurementBinding); err != nil {
t.Fatal(err)
}
bindingFingerprint, _ := measurementBinding.Fingerprint()
bindingJSON, _ := json.Marshal(measurementBinding)
configPath := filepath.Join(root, "registry.yml")
testID := filepath.Base(root)
config := fmt.Sprintf("version: 0.1\nlog:\n level: error\nstorage:\n filesystem:\n rootdirectory: %q\n delete:\n enabled: true\nhttp:\n addr: %q\n headers:\n X-Cleanup-Test-ID: [%q]\n", storage, address, testID)
if err := os.WriteFile(configPath, []byte(config), 0600); err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second)
defer cancel()
environment := []string{"PATH=" + os.Getenv("PATH"), "HOME=" + root, "TMPDIR=" + root, "OTEL_TRACES_EXPORTER=none", "OTEL_METRICS_EXPORTER=none"}
httpClient := &http.Client{Timeout: 5 * time.Second, CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }}
origin := "http://" + address
startRegistry := func() func() {
command := exec.CommandContext(ctx, binary, "serve", configPath)
command.Env = environment
command.Stdout = io.Discard
command.Stderr = io.Discard
if err := command.Start(); err != nil {
t.Fatal("isolated Registry failed to start")
}
exited := make(chan error, 1)
go func() { exited <- command.Wait() }()
var once sync.Once
stop := func() {
once.Do(func() {
command.Process.Signal(os.Interrupt)
select {
case <-exited:
case <-time.After(3 * time.Second):
command.Process.Kill()
<-exited
}
})
}
for deadline := time.Now().Add(15 * time.Second); time.Now().Before(deadline); {
response, err := httpClient.Get(origin + "/v2/")
if err == nil {
response.Body.Close()
if response.StatusCode == 200 && response.Header.Get("X-Cleanup-Test-ID") == testID {
return stop
}
}
time.Sleep(25 * time.Millisecond)
}
stop()
t.Fatal("owned Registry readiness failed; no other endpoint used")
return nil
}
stop := startRegistry()
defer func() { stop() }()
database, err := gorm.Open("sqlite3", filepath.Join(root, "coordination.db"))
if err != nil {
t.Fatal(err)
}
defer database.Close()
database.LogMode(false)
database.DB().SetMaxOpenConns(1)
if err := database.AutoMigrate(&model.CleanupStorage{}, &model.CleanupOperation{}).Error; err != nil {
t.Fatal(err)
}
if err := database.Create(&model.CleanupStorage{StorageID: storageID, Generation: "one", Mode: "ready", RegistrationFingerprint: bindingFingerprint, RegistrationJSON: string(bindingJSON)}).Error; err != nil {
t.Fatal(err)
}
fixtureKey := strings.Repeat("isolated-fixture-", 3)
t.Setenv("TOKEN", fixtureKey)
h := &CleanupCoordinationHandler{database: func() *gorm.DB { return database }, permitKey: func() []byte { return []byte(fixtureKey) }}
installRegistryCoverageFixture(t, h, measurementBinding)
router := chi.NewRouter()
router.Use(middleware.FullToken)
base := "/v2/cleanup/stores/{storage_id}/operations"
router.Post(base, h.Acquire)
for suffix, handler := range map[string]http.HandlerFunc{"finish": h.Finish, "inspect": h.Inspect, "registry-permit": h.RegistryPermit, "attempt": h.BeginAttempt, "attempt/complete": h.CompleteAttempt, "upload": h.BindUpload, "upload/requests": h.AcquireUpload, "upload/requests/finish": h.FinishUpload, "maintenance/measurement": h.RecordMaintenanceMeasurement, "maintenance/request": h.RequestMaintenance, "maintenance/enter": h.EnterMaintenance, "maintenance/complete": h.CompleteMaintenanceWork, "maintenance/restore": h.BeginRestore, "maintenance/restored": h.FinishRestore} {
router.Post(base+"/{operation_id}/"+suffix, handler)
}
router.Post("/v2/cleanup/stores/{storage_id}/uploads/lookup", h.LookupUpload)
apiServer := httptest.NewServer(router)
defer apiServer.Close()
client, err := guard.NewCoordinationClient(apiServer.URL, fixtureKey, true)
if err != nil {
t.Fatal(err)
}
newProxy := func(owner string) *registryproxy.Proxy {
coordinator, err := registryproxy.NewCoordinator(registryproxy.CoordinatorConfig{StorageID: storageID, Generation: "one", Owner: owner, Backend: client, PermitKey: func() []byte { return []byte(fixtureKey) }})
if err != nil {
t.Fatal(err)
}
proxy, err := registryproxy.NewProxy(origin, coordinator, nil)
if err != nil {
t.Fatal(err)
}
return proxy
}
var active atomic.Value
active.Store(newProxy("sidecar-first"))
sidecar := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { active.Load().(*registryproxy.Proxy).ServeHTTP(w, r) }))
defer sidecar.Close()
request := func(method, target, media string, data []byte, permit string, want int) http.Header {
t.Helper()
req, err := http.NewRequestWithContext(ctx, method, target, bytes.NewReader(data))
if err != nil {
t.Fatal("invalid isolated request")
}
if media != "" {
req.Header.Set("Content-Type", media)
}
req.Header.Set("Accept", "application/vnd.oci.image.manifest.v1+json")
if permit != "" {
req.Header.Set("X-Rainbond-Cleanup-Permit", permit)
}
response, err := httpClient.Do(req)
if err != nil {
t.Fatal("isolated Registry request failed")
}
defer response.Body.Close()
io.Copy(io.Discard, io.LimitReader(response.Body, 1<<20))
if response.StatusCode != want {
t.Fatalf("isolated %s response=%d want=%d", method, response.StatusCode, want)
}
return response.Header
}
digest := func(raw []byte) string { sum := sha256.Sum256(raw); return "sha256:" + hex.EncodeToString(sum[:]) }
upload := func(repository string, data []byte) string {
headers := request("POST", sidecar.URL+"/v2/"+repository+"/blobs/uploads/", "", nil, "", 202)
// Replace the sidecar instance between requests; no in-memory upload map survives.
active.Store(newProxy("sidecar-replaced"))
location, err := url.Parse(headers.Get("Location"))
if err != nil {
t.Fatal("invalid upload location")
}
originURL, _ := url.Parse(sidecar.URL)
location = originURL.ResolveReference(location)
if location.Host != originURL.Host || location.User != nil {
t.Fatal("upload location bypassed sidecar")
}
value := digest(data)
query := location.Query()
query.Set("digest", value)
location.RawQuery = query.Encode()
request("PUT", location.String(), "application/octet-stream", data, "", 201)
return value
}
shared := []byte("owned shared layer")
manifests := map[string]string{}
configs := map[string]string{}
for _, repository := range []string{"test/selected", "test/retained", "test/untagged"} {
configBody := []byte(fmt.Sprintf(`{"architecture":"amd64","os":"linux","config":{"Labels":{"fixture":%q}},"rootfs":{"type":"layers","diff_ids":[]}}`, repository))
configDigest := upload(repository, configBody)
layerDigest := upload(repository, shared)
configs[repository] = configDigest
manifest, _ := json.Marshal(map[string]interface{}{"schemaVersion": 2, "mediaType": "application/vnd.oci.image.manifest.v1+json", "config": map[string]interface{}{"mediaType": "application/vnd.oci.image.config.v1+json", "digest": configDigest, "size": len(configBody)}, "layers": []interface{}{map[string]interface{}{"mediaType": "application/vnd.oci.image.layer.v1.tar", "digest": layerDigest, "size": len(shared)}}})
reference := "v1"
if repository == "test/untagged" {
reference = digest(manifest)
}
request("PUT", sidecar.URL+"/v2/"+repository+"/manifests/"+reference, "application/vnd.oci.image.manifest.v1+json", manifest, "", 201)
manifests[repository] = digest(manifest)
}
for deadline := time.Now().Add(5 * time.Second); ; {
var outstanding int
if err := database.Model(&model.CleanupOperation{}).Where("state <> ?", "finished").Count(&outstanding).Error; err != nil {
t.Fatal(err)
}
if outstanding == 0 {
break
}
if time.Now().After(deadline) {
t.Fatal("upload leases failed to close")
}
time.Sleep(10 * time.Millisecond)
}
selected := guard.CoordinationRequest{StorageID: storageID, Generation: "one", Owner: "cleanup-test", OperationID: "selected-delete", Kind: "delete", Scope: "test/selected", Fingerprint: "explicit-selection", Target: manifests["test/selected"]}
if admitted, err := client.Acquire(ctx, selected); err != nil || !admitted {
t.Fatal("selected deletion not admitted", err)
}
permit, err := client.RegistryPermit(ctx, selected)
if err != nil {
t.Fatal("permit not issued")
}
request("DELETE", sidecar.URL+"/v2/test/retained/manifests/"+manifests["test/retained"], "", nil, permit, 403)
request("DELETE", sidecar.URL+"/v2/test/selected/manifests/"+selected.Target, "", nil, permit, 202)
request("DELETE", sidecar.URL+"/v2/test/selected/manifests/"+selected.Target, "", nil, permit, 409)
request("GET", sidecar.URL+"/v2/test/selected/manifests/"+selected.Target, "", nil, "", 404)
request("GET", sidecar.URL+"/v2/test/retained/manifests/"+manifests["test/retained"], "", nil, "", 200)
if err := client.Finish(ctx, selected, true); err != nil {
t.Fatal(err)
}
gc := guard.CoordinationRequest{StorageID: storageID, Generation: "one", Owner: "cleanup-test", OperationID: "manual-gc", Kind: "gc", Scope: "*", Fingerprint: "separate-confirmation"}
if admitted, err := client.RequestMaintenance(ctx, gc); err != nil || !admitted {
t.Fatal(err)
}
request("POST", sidecar.URL+"/v2/test/new/blobs/uploads/", "", nil, "", 409)
stop()
t.Setenv("REGISTRY_STORAGE_FILESYSTEM_ROOTDIRECTORY", t.TempDir())
recorder, err := registryproxy.NewGCRecorder(client, measurementBinding, gc)
if err != nil {
t.Fatal(err)
}
if useExecutor {
if err := registryproxy.ExecuteGC(ctx, storage, measurementBinding, gc, binary, recorder); err != nil {
t.Fatal("owned executor GC failed", err)
}
if err := registryproxy.RecoverGCReceipt(ctx, storage, measurementBinding, gc, recorder); err != nil {
t.Fatal("original executor receipt could not be reconciled", err)
}
} else {
// This exercises the protocol and native GC, not the Linux executor.
before, err := registryproxy.MeasureStorage(storage, measurementBinding)
if err != nil {
t.Fatal(err)
}
if err := recorder.BeginGC(ctx, before); err != nil {
t.Fatal(err)
}
command := exec.CommandContext(ctx, binary, "garbage-collect", configPath)
command.Env, command.Stdout, command.Stderr = environment, io.Discard, io.Discard
if err := command.Run(); err != nil {
t.Fatal("owned manual GC failed", err)
}
if err := recorder.CompleteGC(ctx, "succeeded"); err != nil {
t.Fatal(err)
}
after, err := registryproxy.MeasureStorage(storage, measurementBinding)
if err != nil {
t.Fatal(err)
}
if err := recorder.ObserveGC(ctx, after); err != nil {
t.Fatal(err)
}
}
beforeGC, afterGC, err := guard.MaintenanceMeasurements(database, gc)
if err != nil || beforeGC == nil || afterGC == nil || beforeGC.FilesystemID != afterGC.FilesystemID {
t.Fatal("bound GC observations missing", err)
}
if err := client.RecordMaintenanceMeasurement(ctx, gc, "after", *afterGC); err != nil {
t.Fatal("identical measurement retry failed", err)
}
changedMeasurement := *afterGC
changedMeasurement.FilesystemID = "different-filesystem"
if err := client.RecordMaintenanceMeasurement(ctx, gc, "after", changedMeasurement); !errors.Is(err, guard.ErrCoordinationChanged) {
t.Fatal("HTTP observation replaced immutable evidence", err)
}
wrongIdentity := gc
wrongIdentity.Owner = "another-executor"
if err := client.RecordMaintenanceMeasurement(ctx, wrongIdentity, "after", *afterGC); !errors.Is(err, guard.ErrCoordinationChanged) {
t.Fatal("observation accepted for another operation owner", err)
}
// Other processes also use the host filesystem. Its free-space delta is an
// observation, not proof that all changed bytes were reclaimed by this GC.
t.Logf("verified backing filesystem: available bytes before=%d after=%d; inode counts available=%t", beforeGC.AvailableBytes, afterGC.AvailableBytes, beforeGC.FreeInodes != nil && afterGC.FreeInodes != nil)
blobPath := func(d string) string {
hex := strings.TrimPrefix(d, "sha256:")
return filepath.Join(storage, "docker/registry/v2/blobs/sha256", hex[:2], hex, "data")
}
if _, err := os.Stat(blobPath(configs["test/selected"])); !os.IsNotExist(err) {
t.Fatal("selected-only config blob was not reclaimed")
}
if _, err := os.Stat(blobPath(digest(shared))); err != nil {
t.Fatal("GC removed retained shared layer")
}
if _, err := os.Stat(blobPath(configs["test/untagged"])); err != nil {
t.Fatal("unselected untagged manifest content was deleted", err)
}
if err := client.BeginRestore(ctx, gc); err != nil {
t.Fatal(err)
}
stop = startRegistry()
request("GET", sidecar.URL+"/v2/test/untagged/manifests/"+manifests["test/untagged"], "", nil, "", 200)
request("GET", sidecar.URL+"/v2/test/retained/manifests/"+manifests["test/retained"], "", nil, "", 200)
request("POST", sidecar.URL+"/v2/test/new/blobs/uploads/", "", nil, "", 409)
if err := client.FinishRestore(ctx, gc, true); err != nil {
t.Fatal(err)
}
upload("test/after-gc", []byte("writes restored"))
t.Log("real Distribution: uploads survive coordinator replacement; only selected manifest deleted; replay blocked; manual GC preserved shared content; writes restored")
}