api/controller/cleanup_gc_executor_test.go
Rama main · réplica de lectura del repositorio original.
package controller
import (
"context"
"errors"
"net/http/httptest"
"path/filepath"
"strings"
"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/kubeidentity"
"github.com/jinzhu/gorm"
batchv1 "k8s.io/api/batch/v1"
corev1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/labels"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/util/intstr"
"k8s.io/client-go/kubernetes"
"k8s.io/client-go/kubernetes/fake"
typedbatch "k8s.io/client-go/kubernetes/typed/batch/v1"
ktesting "k8s.io/client-go/testing"
)
// capability_id: rainbond.cleanup.gc-job-execution-api
// capability_id: rainbond.cleanup.gc-job-restore
func TestGCJobAdmissionAPIRequiresVerifiedExecutorAndGrantsOnce(t *testing.T) {
database, err := gorm.Open("sqlite3", filepath.Join(t.TempDir(), "gc.db"))
if err != nil {
t.Fatal(err)
}
defer database.Close()
database.DB().SetMaxOpenConns(1)
if err := database.AutoMigrate(&model.CleanupStorage{}, &model.CleanupOperation{}).Error; err != nil {
t.Fatal(err)
}
kube, _, volume := gcAdmissionFixture(t)
binding := guard.StorageRegistration{StorageID: "store", Generation: "gen", VolumeUID: volume, RootPath: "/registry"}
if err := guard.RegisterStorage(database, binding); err != nil {
t.Fatal(err)
}
if err := database.Model(&model.CleanupStorage{}).Where("storage_id = ?", "store").Update("mode", "ready").Error; err != nil {
t.Fatal(err)
}
r := guard.CoordinationRequest{StorageID: "store", Generation: "gen", OperationID: "gc-op", Kind: "gc", Scope: "*", Owner: "executor", Fingerprint: "request"}
if _, err := guard.RequestMaintenance(database, r); err != nil {
t.Fatal(err)
}
// An unprepared/foreign pod must fail before any execution grant.
h := &CleanupCoordinationHandler{database: func() *gorm.DB { return database }, gcTarget: func() (kubernetes.Interface, string, string, error) {
return gcAdmissionKubeClient{Interface: kube}, "system", "rbd-hub", nil
}}
t.Setenv("TOKEN", "gc-isolated-fixture")
router := chi.NewRouter()
router.Use(middleware.FullToken)
router.Post("/v2/cleanup/stores/{storage_id}/operations/{operation_id}/maintenance/enter-job", h.EnterGCJob)
router.Post("/v2/cleanup/stores/{storage_id}/operations/{operation_id}/maintenance/job", h.SubmitGCJob)
router.Post("/v2/cleanup/stores/{storage_id}/operations/{operation_id}/maintenance/job/start", h.StartGCJob)
router.Post("/v2/cleanup/stores/{storage_id}/operations/{operation_id}/maintenance/job/restore", h.RestoreGCJob)
router.Post("/v2/cleanup/stores/{storage_id}/operations/{operation_id}/maintenance/job/status", h.GCJobProgress)
router.Post("/v2/cleanup/stores/{storage_id}/operations/{operation_id}/maintenance/measurement", h.RecordMaintenanceMeasurement)
router.Post("/v2/cleanup/stores/{storage_id}/operations/{operation_id}/maintenance/complete", h.CompleteMaintenanceWork)
server := httptest.NewServer(router)
defer server.Close()
client, err := guard.NewCoordinationClient(server.URL, "gc-isolated-fixture", true)
if err != nil {
t.Fatal(err)
}
locator := guard.GCExecutorLocator{Pod: "gc-pod", PodUID: "gc-pod-uid"}
if err := client.EnterGCJob(context.Background(), r, locator); err == nil {
t.Fatal("unprepared executor admitted")
}
state, err := guard.InspectOperation(database, r)
if err != nil || state != "draining" {
t.Fatal(state, err)
}
// The route must be protected even when ordinary region token auth is optional.
response := httptest.NewRecorder()
router.ServeHTTP(response, httptest.NewRequest("POST", "/v2/cleanup/stores/store/operations/gc-op/maintenance/enter-job", strings.NewReader("{}")))
if response.Code != 401 {
t.Fatal("unauthenticated admission", response.Code)
}
// Runtime overrides are not part of the API contract, even for an authenticated caller.
injected := httptest.NewRequest("POST", "/v2/cleanup/stores/store/operations/gc-op/maintenance/job", strings.NewReader(`{"image":"caller-image"}`))
injected.Header.Set("Authorization", "Token gc-isolated-fixture")
invalid := httptest.NewRecorder()
router.ServeHTTP(invalid, injected)
if invalid.Code != 400 {
t.Fatal("caller executable override accepted", invalid.Code)
}
// Store and job identity validation still runs even with valid internal auth.
for i := 0; i < 2; i++ {
if err := client.SubmitGCJob(context.Background(), r); err != nil {
t.Fatal("submit failed", err)
}
}
recorded, err := guard.ReadGCJobBinding(database, r)
if err != nil {
t.Fatal(err)
}
if err := client.StartGCJob(context.Background(), r); err != nil {
t.Fatal("start failed", err)
}
job, err := kube.BatchV1().Jobs("system").Get(context.Background(), recorded.Name, metav1.GetOptions{})
if err != nil {
t.Fatal(err)
}
yes := true
pod := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: locator.Pod, Namespace: "system", UID: "gc-pod-uid", ResourceVersion: "1", OwnerReferences: []metav1.OwnerReference{{APIVersion: "batch/v1", Kind: "Job", Name: job.Name, UID: job.UID, Controller: &yes}}}, Spec: *job.Spec.Template.Spec.DeepCopy(), Status: corev1.PodStatus{Phase: corev1.PodRunning, ContainerStatuses: []corev1.ContainerStatus{{Name: "gc", ImageID: job.Spec.Template.Spec.Containers[0].Image, ContainerID: "containerd://owned", State: corev1.ContainerState{Running: &corev1.ContainerStateRunning{}}}}}}
if _, err := kube.CoreV1().Pods("system").Create(context.Background(), pod, metav1.CreateOptions{}); err != nil {
t.Fatal(err)
}
writer := guard.CoordinationRequest{StorageID: "store", Generation: "gen", OperationID: "old-writer", Owner: "builder", Kind: "producer", Scope: "*", Fingerprint: "old-request"}
// Simulate an operation accepted before the maintenance request.
if err := database.Create(&model.CleanupOperation{StorageID: writer.StorageID, Generation: writer.Generation, OperationID: writer.OperationID, Owner: writer.Owner, Kind: writer.Kind, Scope: writer.Scope, Fingerprint: writer.Fingerprint, State: "active"}).Error; err != nil {
t.Fatal(err)
}
if err := client.EnterGCJob(context.Background(), r, locator); err == nil {
t.Fatal("active writer was ignored")
}
if err := guard.FinishOperation(database, writer, true); err != nil {
t.Fatal(err)
}
if err := client.EnterGCJob(context.Background(), r, guard.GCExecutorLocator{Pod: locator.Pod, PodUID: "replacement"}); err == nil {
t.Fatal("foreign executor admitted")
}
if err := client.EnterGCJob(context.Background(), r, locator); err != nil {
t.Fatal(err)
}
if err := client.EnterGCJob(context.Background(), r, locator); err == nil {
t.Fatal("execution grant replayed")
}
if err := guard.EnterMaintenance(database, r); err == nil {
t.Fatal("generic admission bypassed Job binding")
}
state, err = guard.InspectOperation(database, r)
if err != nil || state != "exclusive" {
t.Fatal(state, err)
}
fingerprint, _ := binding.Fingerprint()
before := guard.StorageMeasurement{Protocol: 1, StorageID: r.StorageID, Generation: r.Generation, BindingFingerprint: fingerprint, FilesystemID: "owned-fs", ObservedAt: time.Now().UTC(), TotalBytes: 1000, FreeBytes: 100, AvailableBytes: 50}
if err := client.RecordMaintenanceMeasurement(context.Background(), r, "before", before); err != nil {
t.Fatal(err)
}
if err := client.CompleteMaintenanceWork(context.Background(), r, "succeeded"); err != nil {
t.Fatal(err)
}
after := before
after.ObservedAt = before.ObservedAt.Add(time.Second)
after.FreeBytes = 200
after.AvailableBytes = 150
if err := client.RecordMaintenanceMeasurement(context.Background(), r, "after", after); err != nil {
t.Fatal(err)
}
if err := client.RestoreGCJob(context.Background(), r); !errors.Is(err, guard.ErrCoordinationBusy) {
t.Fatal("still-running executor not reported as pending", err)
}
pod.Status.Phase = corev1.PodSucceeded
pod.Status.ContainerStatuses[0].State = corev1.ContainerState{Terminated: &corev1.ContainerStateTerminated{ExitCode: 0, FinishedAt: metav1.Now()}}
if _, err := kube.CoreV1().Pods("system").Update(context.Background(), pod, metav1.UpdateOptions{}); err != nil {
t.Fatal(err)
}
hub, err := kube.CoreV1().Pods("system").Get(context.Background(), "hub", metav1.GetOptions{})
if err != nil {
t.Fatal(err)
}
original := hub.DeepCopy()
hub.Spec.Containers[0].Env = append(hub.Spec.Containers[0].Env, corev1.EnvVar{Name: "REGISTRY_STORAGE_MAINTENANCE_READONLY_ENABLED", Value: "true"})
if _, err := kube.CoreV1().Pods("system").Update(context.Background(), hub, metav1.UpdateOptions{}); err != nil {
t.Fatal(err)
}
if err := client.RestoreGCJob(context.Background(), r); err == nil {
t.Fatal("changed registry configuration accepted")
}
if _, err := kube.CoreV1().Pods("system").Update(context.Background(), original, metav1.UpdateOptions{}); err != nil {
t.Fatal(err)
}
if err := client.RestoreGCJob(context.Background(), r); err != nil {
t.Fatal("verified restoration failed", err)
}
if err := kube.CoreV1().Pods("system").Delete(context.Background(), pod.Name, metav1.DeleteOptions{}); err != nil {
t.Fatal(err)
}
if err := client.RestoreGCJob(context.Background(), r); err != nil {
t.Fatal("lost restoration acknowledgment not recoverable", err)
}
state, err = guard.InspectOperation(database, r)
if err != nil || state != "finished" {
t.Fatal(state, err)
}
progress, err := client.GCJobProgress(context.Background(), r)
if err != nil || progress.State != "finished" || progress.Outcome != "succeeded" || progress.Before == nil || progress.After == nil || progress.After.AvailableBytes != 150 {
t.Fatal("missing original outcome and measurements", progress, err)
}
}
func gcAdmissionFixture(t *testing.T) (*fake.Clientset, *batchv1.Job, string) {
t.Helper()
no := false
native := corev1.Container{Name: "gc", Image: "example.test/owned@sha256:" + strings.Repeat("a", 64), Command: []string{"/registry-gc"}, VolumeMounts: []corev1.VolumeMount{{Name: "data", MountPath: "/registry", SubPath: "owned"}}}
spec := corev1.PodSpec{NodeName: "node", RestartPolicy: corev1.RestartPolicyNever, AutomountServiceAccountToken: &no, Containers: []corev1.Container{native}, Volumes: []corev1.Volume{{Name: "data", VolumeSource: corev1.VolumeSource{PersistentVolumeClaim: &corev1.PersistentVolumeClaimVolumeSource{ClaimName: "data"}}}}}
job := &batchv1.Job{ObjectMeta: metav1.ObjectMeta{Namespace: "system"}, Spec: batchv1.JobSpec{Template: corev1.PodTemplateSpec{Spec: spec}}}
hub := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "hub", Namespace: "system", UID: "hub-uid", Labels: map[string]string{"app": "hub"}}, Spec: *spec.DeepCopy()}
hub.ResourceVersion = "1"
hub.Annotations = map[string]string{"rainbond.io/registry-gc-executor": "v1"}
hub.Spec.Containers[0].Name = "registry"
hub.Spec.Containers[0].Env = []corev1.EnvVar{{Name: "REGISTRY_STORAGE_FILESYSTEM_ROOTDIRECTORY", Value: "/registry"}, {Name: "REGISTRY_HTTP_ADDR", Value: "127.0.0.1:5000"}, {Name: "REGISTRY_STORAGE_MAINTENANCE_UPLOADPURGING_ENABLED", Value: "false"}}
svc := &corev1.Service{ObjectMeta: metav1.ObjectMeta{Name: "rbd-hub", Namespace: "system", UID: "service-uid", ResourceVersion: "1"}, Spec: corev1.ServiceSpec{Selector: map[string]string{"app": "hub"}, Ports: []corev1.ServicePort{{Port: 5000, TargetPort: intstr.FromInt(5001)}}}}
pvc := &corev1.PersistentVolumeClaim{ObjectMeta: metav1.ObjectMeta{Name: "data", Namespace: "system", UID: "pvc-uid"}, Spec: corev1.PersistentVolumeClaimSpec{VolumeName: "pv"}, Status: corev1.PersistentVolumeClaimStatus{Phase: corev1.ClaimBound}}
pv := &corev1.PersistentVolume{ObjectMeta: metav1.ObjectMeta{Name: "pv", UID: "pv-uid"}, Spec: corev1.PersistentVolumeSpec{ClaimRef: &corev1.ObjectReference{Namespace: "system", Name: "data", UID: "pvc-uid"}}, Status: corev1.PersistentVolumeStatus{Phase: corev1.VolumeBound}}
image := "example.test/chaos@sha256:" + strings.Repeat("c", 64)
cleaner := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "chaos", Namespace: "system", UID: "chaos-uid", ResourceVersion: "1", Labels: map[string]string{"name": "rbd-chaos"}}, Spec: corev1.PodSpec{Containers: []corev1.Container{{Name: "chaos", Image: image, Command: []string{"/run/rainbond-chaos"}, Args: []string{"--clean-up=false"}}}}, Status: corev1.PodStatus{Phase: corev1.PodRunning, ContainerStatuses: []corev1.ContainerStatus{{Name: "chaos", ImageID: image, ContainerID: "containerd://chaos", State: corev1.ContainerState{Running: &corev1.ContainerStateRunning{}}}}}}
client := fake.NewSimpleClientset(hub, svc, pvc, pv, cleaner)
observed, err := kubeidentity.InspectNativeRegistry(context.Background(), client, "system", "rbd-hub", "hub", "hub-uid")
if err != nil {
t.Fatal(err)
}
sidecar := corev1.Container{Name: "coordinator", Image: native.Image, Command: []string{"/registry-coordinator"}, VolumeMounts: []corev1.VolumeMount{{Name: "data", MountPath: "/registry", SubPath: "owned", ReadOnly: true}, {Name: "control", MountPath: "/control", ReadOnly: true}}, Args: []string{"--listen=:5001", "--upstream=http://127.0.0.1:5000", "--storage-id=store", "--storage-generation=gen", "--volume-uid=" + observed.Mount.VolumeUID, "--registry-path=/registry", "--storage-root=/registry", "--coordination-api=http://rbd-api.system:8443", "--allow-internal-http=true", "--credential-file=/control/token"}}
hub.Spec.Containers = append(hub.Spec.Containers, sidecar)
hub.Spec.Volumes = append(hub.Spec.Volumes, corev1.Volume{Name: "control", VolumeSource: corev1.VolumeSource{Secret: &corev1.SecretVolumeSource{SecretName: "coordination"}}})
hub.Status.ContainerStatuses = []corev1.ContainerStatus{{Name: "registry", ImageID: "example.test/native@sha256:" + strings.Repeat("b", 64), State: corev1.ContainerState{Running: &corev1.ContainerStateRunning{}}}, {Name: "coordinator", ImageID: native.Image, ContainerID: "containerd://coordinator", State: corev1.ContainerState{Running: &corev1.ContainerStateRunning{}}}}
if _, err := client.CoreV1().Pods("system").Update(context.Background(), hub, metav1.UpdateOptions{}); err != nil {
t.Fatal(err)
}
client.PrependReactor("list", "pods", func(action ktesting.Action) (bool, runtime.Object, error) {
result, err := client.Tracker().List(corev1.SchemeGroupVersion.WithResource("pods"), corev1.SchemeGroupVersion.WithKind("Pod"), action.GetNamespace())
if err == nil {
list := result.(*corev1.PodList)
selected := action.(ktesting.ListAction).GetListRestrictions().Labels
filtered := []corev1.Pod{}
for _, item := range list.Items {
if selected.Matches(labels.Set(item.Labels)) {
filtered = append(filtered, item)
}
}
list.Items = filtered
list.ResourceVersion = "snapshot"
}
return true, result, err
})
return client, job, observed.Mount.VolumeUID
}
type gcAdmissionJobClient struct{ typedbatch.JobInterface }
func (c gcAdmissionJobClient) Create(ctx context.Context, job *batchv1.Job, options metav1.CreateOptions) (*batchv1.Job, error) {
observed := job.DeepCopy()
if len(options.DryRun) > 0 {
return observed, nil
}
observed.UID = "job-uid"
observed.ResourceVersion = "1"
return c.JobInterface.Create(ctx, observed, options)
}
type gcAdmissionKubeClient struct{ kubernetes.Interface }
func (c gcAdmissionKubeClient) BatchV1() typedbatch.BatchV1Interface {
return gcAdmissionBatchClient{BatchV1Interface: c.Interface.BatchV1()}
}
type gcAdmissionBatchClient struct{ typedbatch.BatchV1Interface }
func (c gcAdmissionBatchClient) Jobs(namespace string) typedbatch.JobInterface {
return gcAdmissionJobClient{JobInterface: c.BatchV1Interface.Jobs(namespace)}
}