package controller import ( "context" "encoding/json" "errors" "io" "net/http" "net/url" "regexp" "strings" "time" "github.com/go-chi/chi" "github.com/goodrain/rainbond/config/configs" "github.com/goodrain/rainbond/db" guard "github.com/goodrain/rainbond/pkg/cleanup" "github.com/goodrain/rainbond/pkg/cleanup/kubeidentity" "github.com/goodrain/rainbond/pkg/component/k8s" "github.com/goodrain/rainbond/pkg/component/storage" httputil "github.com/goodrain/rainbond/util/http" "github.com/jinzhu/gorm" "k8s.io/apimachinery/pkg/util/validation" "k8s.io/client-go/kubernetes" "k8s.io/client-go/metadata" ) // CleanupCoordinationHandler is an internal Region API. Routes must always use // CleanupIdentity, requiring a configured token or verified platform mutual TLS. // This API records coordination only; it never performs deletion or enables an // unverified store. Owner identities come from trusted Region participants. type CleanupCoordinationHandler struct { uploadStorageBinding func() (string, error) deleteUploadChunks func(string) error uploadPackageReferences func(*gorm.DB, []string) (map[string]bool, error) measureUploadEvent func(context.Context, string) (storage.UploadChunkUsage, error) measureUploadChunks func(context.Context, string) (storage.UploadChunkUsage, error) inspectRegistryCoverage func(context.Context, guard.StorageRegistration, []guard.ParticipantRegistration) (guard.RegistryCoverage, error) inspectReferenceWriters func(context.Context) ([]guard.ReferenceWriter, error) inspectConsoleWriter func(context.Context, string, string) (guard.ReferenceWriter, error) clusterReferences func(context.Context) (guard.RegionReferenceInventory, error) nodeSettings func() kubeidentity.NodeJobSettings inventorySettings func() kubeidentity.NodeInventorySettings inspectManagedCache func(context.Context, string, string) (kubeidentity.ManagedCachePreparation, error) inspectManagedPackages func(context.Context, string, string) (kubeidentity.ManagedPackagePreparation, error) inspectPackageWriters func(context.Context) ([]guard.ReferenceWriter, error) gcTarget func() (kubernetes.Interface, string, string, error) database func() *gorm.DB permitKey func() []byte inspectRegistry func(context.Context, string, string) (kubeidentity.RegistryPreparation, error) inspectParticipant func(context.Context, string, string, string, guard.StorageRegistration) (guard.ParticipantRegistration, error) } // NewCleanupCoordinationHandler uses the Region database manager. func NewCleanupCoordinationHandler() *CleanupCoordinationHandler { h := &CleanupCoordinationHandler{uploadStorageBinding: systemUploadStorageBinding, deleteUploadChunks: systemDeleteUploadChunks, uploadPackageReferences: guard.ReadUploadPackageReferences, measureUploadChunks: systemMeasureUploadChunks, measureUploadEvent: systemMeasureUploadEvent, inspectReferenceWriters: inspectSystemReferenceWriters, inspectConsoleWriter: inspectSystemConsoleWriter, clusterReferences: systemClusterReferenceInventory, inventorySettings: systemNodeInventorySettings, nodeSettings: systemNodeJobSettings, database: func() *gorm.DB { return db.GetManager().DB() }, permitKey: systemRegistryPermitKey, inspectRegistry: inspectSystemRegistry, inspectManagedCache: inspectSystemManagedCache, inspectManagedPackages: inspectSystemManagedPackages, inspectPackageWriters: inspectSystemManagedPackageWriters, inspectParticipant: inspectSystemRegistryParticipant, gcTarget: systemRegistryInspectionTarget} h.inspectRegistryCoverage = h.observeRegistryCoverage return h } // DiscoverStores locates enrolled storage for authenticated platform producers. // It neither registers new storage nor enables deletion readiness. func (h *CleanupCoordinationHandler) DiscoverStores(w http.ResponseWriter, r *http.Request) { var body struct{} if !coordinationDecode(w, r, &body) { return } stores, err := guard.DiscoverStores(h.database()) if err != nil { coordinationError(w, r, err) return } httputil.ReturnSuccess(r, w, struct { Protocol int `json:"protocol"` Stores []guard.StoreIdentity `json:"stores"` }{1, stores}) } func coordinationError(w http.ResponseWriter, r *http.Request, err error) { status, code := http.StatusServiceUnavailable, "COORDINATION_UNAVAILABLE" switch { case errors.Is(err, guard.ErrCoordinationDenied): status, code = 403, "COORDINATION_DENIED" case errors.Is(err, guard.ErrCoordinationBusy): status, code = 409, "COORDINATION_BUSY" case errors.Is(err, guard.ErrCoordinationChanged): status, code = 409, "COORDINATION_CHANGED" case errors.Is(err, guard.ErrCoordinationUncertain): status, code = 409, "COORDINATION_UNCERTAIN" case errors.Is(err, gorm.ErrRecordNotFound): status, code = 404, "COORDINATION_NOT_FOUND" } httputil.ReturnError(r, w, status, code) } func coordinationDecode(w http.ResponseWriter, r *http.Request, value interface{}) bool { decoder := json.NewDecoder(http.MaxBytesReader(w, r.Body, 16<<10)) decoder.DisallowUnknownFields() if decoder.Decode(value) != nil || decoder.Decode(&struct{}{}) != io.EOF { httputil.ReturnError(r, w, 400, "INVALID_COORDINATION_REQUEST") return false } return true } func coordinationScopeFromRoute(w http.ResponseWriter, r *http.Request, request *guard.CoordinationRequest) bool { request.StorageID = chi.URLParam(r, "storage_id") if id := chi.URLParam(r, "operation_id"); id != "" && request.OperationID != id { httputil.ReturnError(r, w, 400, "INVALID_COORDINATION_SCOPE") return false } return true } // Acquire records an admission bound to the server-routed store. func (h *CleanupCoordinationHandler) Acquire(w http.ResponseWriter, r *http.Request) { var request guard.CoordinationRequest if !coordinationDecode(w, r, &request) || !coordinationScopeFromRoute(w, r, &request) { return } if err := r.Context().Err(); err != nil { coordinationError(w, r, err) return } created, err := guard.AcquireOperation(h.database(), request) if err != nil { coordinationError(w, r, err) return } httputil.ReturnSuccess(r, w, struct { Protocol int `json:"protocol"` NewlyAdmitted bool `json:"newly_admitted"` }{1, created}) } // Finish records an original operation outcome without unlocking uncertain work. func (h *CleanupCoordinationHandler) Finish(w http.ResponseWriter, r *http.Request) { var body struct { guard.CoordinationRequest Confirmed *bool `json:"confirmed"` } if !coordinationDecode(w, r, &body) || !coordinationScopeFromRoute(w, r, &body.CoordinationRequest) { return } if body.Confirmed == nil { httputil.ReturnError(r, w, 400, "CONFIRMATION_REQUIRED") return } if err := r.Context().Err(); err != nil { coordinationError(w, r, err) return } if err := guard.FinishOperation(h.database(), body.CoordinationRequest, *body.Confirmed); err != nil { coordinationError(w, r, err) return } httputil.ReturnSuccess(r, w, struct { Protocol int `json:"protocol"` Recorded bool `json:"recorded"` }{1, true}) } // Inspect returns only the exact bound operation state. func (h *CleanupCoordinationHandler) Inspect(w http.ResponseWriter, r *http.Request) { var request guard.CoordinationRequest if !coordinationDecode(w, r, &request) || !coordinationScopeFromRoute(w, r, &request) { return } if err := r.Context().Err(); err != nil { coordinationError(w, r, err) return } state, err := guard.InspectOperation(h.database(), request) if err != nil { coordinationError(w, r, err) return } httputil.ReturnSuccess(r, w, struct { Protocol int `json:"protocol"` State string `json:"state"` }{1, state}) } // RequestMaintenance closes admission but does not start GC. func (h *CleanupCoordinationHandler) RequestMaintenance(w http.ResponseWriter, r *http.Request) { var request guard.CoordinationRequest if !coordinationDecode(w, r, &request) || !coordinationScopeFromRoute(w, r, &request) { return } if err := r.Context().Err(); err != nil { coordinationError(w, r, err) return } created, err := guard.RequestMaintenance(h.database(), request) if err != nil { coordinationError(w, r, err) return } httputil.ReturnSuccess(r, w, struct { Protocol int `json:"protocol"` NewlyAdmitted bool `json:"newly_admitted"` }{1, created}) } func (h *CleanupCoordinationHandler) maintenanceTransition(w http.ResponseWriter, r *http.Request, transition func(*gorm.DB, guard.CoordinationRequest) error) { var request guard.CoordinationRequest if !coordinationDecode(w, r, &request) || !coordinationScopeFromRoute(w, r, &request) { return } if err := r.Context().Err(); err != nil { coordinationError(w, r, err) return } if err := transition(h.database(), request); err != nil { coordinationError(w, r, err) return } httputil.ReturnSuccess(r, w, struct { Protocol int `json:"protocol"` Recorded bool `json:"recorded"` }{1, true}) } // EnterMaintenance grants one execution after all earlier operations drain. func (h *CleanupCoordinationHandler) EnterMaintenance(w http.ResponseWriter, r *http.Request) { h.maintenanceTransition(w, r, guard.EnterMaintenance) } // BeginRestore records intent; it does not declare the native storage writable. func (h *CleanupCoordinationHandler) BeginRestore(w http.ResponseWriter, r *http.Request) { h.maintenanceTransition(w, r, guard.BeginMaintenanceRestore) } // CancelDrain only cancels a maintenance request before GC has been granted. func (h *CleanupCoordinationHandler) CancelDrain(w http.ResponseWriter, r *http.Request) { h.maintenanceTransition(w, r, guard.CancelMaintenanceDrain) } // CompleteMaintenanceWork records the trusted executor's observed process outcome. func (h *CleanupCoordinationHandler) CompleteMaintenanceWork(w http.ResponseWriter, r *http.Request) { var body struct { guard.CoordinationRequest Outcome string `json:"outcome"` } if !coordinationDecode(w, r, &body) || !coordinationScopeFromRoute(w, r, &body.CoordinationRequest) { return } if err := r.Context().Err(); err != nil { coordinationError(w, r, err) return } if err := guard.CompleteMaintenanceWork(h.database(), body.CoordinationRequest, body.Outcome); err != nil { coordinationError(w, r, err) return } httputil.ReturnSuccess(r, w, struct { Protocol int `json:"protocol"` Recorded bool `json:"recorded"` }{1, true}) } // RecordMaintenanceMeasurement accepts immutable observations from an internal // executor; observations alone never complete GC or restore storage writes. func (h *CleanupCoordinationHandler) RecordMaintenanceMeasurement(w http.ResponseWriter, r *http.Request) { var body struct { guard.CoordinationRequest Phase string `json:"phase"` Measurement guard.StorageMeasurement `json:"measurement"` } if !coordinationDecode(w, r, &body) || !coordinationScopeFromRoute(w, r, &body.CoordinationRequest) { return } if err := r.Context().Err(); err != nil { coordinationError(w, r, err) return } if err := guard.RecordMaintenanceMeasurement(h.database(), body.CoordinationRequest, body.Phase, body.Measurement); err != nil { coordinationError(w, r, err) return } httputil.ReturnSuccess(r, w, struct { Protocol int `json:"protocol"` Recorded bool `json:"recorded"` }{1, true}) } // FinishRestore records successful native restoration or persistent uncertainty. func (h *CleanupCoordinationHandler) FinishRestore(w http.ResponseWriter, r *http.Request) { var body struct { guard.CoordinationRequest Confirmed *bool `json:"confirmed"` } if !coordinationDecode(w, r, &body) || !coordinationScopeFromRoute(w, r, &body.CoordinationRequest) { return } if body.Confirmed == nil { httputil.ReturnError(r, w, 400, "CONFIRMATION_REQUIRED") return } if err := r.Context().Err(); err != nil { coordinationError(w, r, err) return } if err := guard.FinishMaintenanceRestore(h.database(), body.CoordinationRequest, *body.Confirmed); err != nil { coordinationError(w, r, err) return } httputil.ReturnSuccess(r, w, struct { Protocol int `json:"protocol"` Recorded bool `json:"recorded"` }{1, true}) } // RegistryReferenceInventory returns advisory Region and retained Helm image identities. func (h *CleanupCoordinationHandler) RegistryReferenceInventory(w http.ResponseWriter, r *http.Request) { var body struct { Generation string `json:"generation"` } if !coordinationDecode(w, r, &body) { return } if err := r.Context().Err(); err != nil { coordinationError(w, r, err) return } storage := chi.URLParam(r, "storage_id") result, err := guard.ReadRegionReferenceInventory(h.database(), storage, body.Generation) if err != nil { coordinationError(w, r, err) return } if h.clusterReferences == nil { coordinationError(w, r, guard.ErrCoordinationUnavailable) return } helm, err := h.clusterReferences(r.Context()) if err != nil { coordinationError(w, r, guard.ErrCoordinationUnavailable) return } result = guard.MergeReferenceInventories(result, helm) // Manual inventory collection refreshes readiness from current deployment // evidence. Health/status reads remain side-effect free. Missing evidence // preserves the reference inventory but cannot certify deletion candidates. ready := false if binding, bindingErr := guard.StorageBinding(h.database(), storage, body.Generation); bindingErr == nil { _, certified, certificationErr := h.certifyRegistry(r.Context(), binding, nil) ready = certificationErr == nil && certified } httputil.ReturnSuccess(r, w, struct { Protocol int `json:"protocol"` StorageID string `json:"storage_id"` Generation string `json:"generation"` RegistryReady bool `json:"registry_ready"` guard.RegionReferenceInventory }{1, storage, body.Generation, ready, result}) } // RegistryReferences audits retained Region records while the selected scope is held. func (h *CleanupCoordinationHandler) RegistryReferences(w http.ResponseWriter, r *http.Request) { var body struct { guard.CoordinationRequest Tags []string `json:"tags"` } if !coordinationDecode(w, r, &body) || !coordinationScopeFromRoute(w, r, &body.CoordinationRequest) { return } if err := r.Context().Err(); err != nil { coordinationError(w, r, err) return } if h.clusterReferences == nil { coordinationError(w, r, guard.ErrCoordinationUnavailable) return } state, err := guard.InspectOperation(h.database(), body.CoordinationRequest) if err != nil || state != "active" { coordinationError(w, r, guard.ErrCoordinationChanged) return } helm, err := h.clusterReferences(r.Context()) if err != nil { coordinationError(w, r, guard.ErrCoordinationUnavailable) return } // Re-observe all current platform writers before accepting absence of references. if err := h.requireReferenceWriterCoverage(r.Context()); err != nil { coordinationError(w, r, err) return } // The database audit rechecks the original admission after the remote read. result, err := guard.AuditRegionManifestReferences(h.database(), body.CoordinationRequest, body.Tags, helm) if err != nil { coordinationError(w, r, err) return } httputil.ReturnSuccess(r, w, struct { Protocol int `json:"protocol"` guard.RegionReferenceAudit }{1, result}) } // RegistryPermit issues a short-lived credential for an active immutable target. func (h *CleanupCoordinationHandler) RegistryPermit(w http.ResponseWriter, r *http.Request) { var request guard.CoordinationRequest if !coordinationDecode(w, r, &request) || !coordinationScopeFromRoute(w, r, &request) { return } if h.permitKey == nil { coordinationError(w, r, guard.ErrCoordinationDenied) return } state, err := guard.InspectOperation(h.database(), request) if err != nil { coordinationError(w, r, err) return } if state != "active" { coordinationError(w, r, guard.ErrCoordinationUncertain) return } binding, err := guard.StorageBinding(h.database(), request.StorageID, request.Generation) if err != nil { coordinationError(w, r, err) return } if _, ready, err := h.certifyRegistry(r.Context(), binding, &request); err != nil || !ready { coordinationError(w, r, guard.ErrCoordinationUnavailable) return } permit, err := guard.IssueRegistryDeletionPermit(h.permitKey(), request, time.Now()) if err != nil { coordinationError(w, r, err) return } httputil.ReturnSuccess(r, w, struct { Protocol int `json:"protocol"` Permit string `json:"permit"` }{1, permit}) } // BeginAttempt consumes the target's one-time execution grant. func (h *CleanupCoordinationHandler) BeginAttempt(w http.ResponseWriter, r *http.Request) { h.maintenanceTransition(w, r, guard.BeginDeletionAttempt) } // CompleteAttempt records the observed transport result without releasing scope. func (h *CleanupCoordinationHandler) CompleteAttempt(w http.ResponseWriter, r *http.Request) { var body struct { guard.CoordinationRequest Outcome string `json:"outcome"` } if !coordinationDecode(w, r, &body) || !coordinationScopeFromRoute(w, r, &body.CoordinationRequest) { return } if err := guard.CompleteDeletionAttempt(h.database(), body.CoordinationRequest, body.Outcome); err != nil { coordinationError(w, r, err) return } httputil.ReturnSuccess(r, w, struct { Protocol int `json:"protocol"` Recorded bool `json:"recorded"` }{1, true}) } // BindUpload persists the Registry's exact upload identity before forwarding it. func (h *CleanupCoordinationHandler) BindUpload(w http.ResponseWriter, r *http.Request) { var body struct { guard.CoordinationRequest Repository string `json:"repository"` UploadID string `json:"upload_id"` } if !coordinationDecode(w, r, &body) || !coordinationScopeFromRoute(w, r, &body.CoordinationRequest) { return } if err := guard.BindRegistryUpload(h.database(), body.CoordinationRequest, body.Repository, body.UploadID); err != nil { coordinationError(w, r, err) return } httputil.ReturnSuccess(r, w, struct { Protocol int `json:"protocol"` Recorded bool `json:"recorded"` }{1, true}) } // LookupUpload resolves an existing upload under the server-selected store. func (h *CleanupCoordinationHandler) LookupUpload(w http.ResponseWriter, r *http.Request) { var body struct { Generation string `json:"generation"` Repository string `json:"repository"` UploadID string `json:"upload_id"` } if !coordinationDecode(w, r, &body) { return } binding, err := guard.LookupRegistryUpload(h.database(), chi.URLParam(r, "storage_id"), body.Generation, body.Repository, body.UploadID) if err != nil { coordinationError(w, r, err) return } httputil.ReturnSuccess(r, w, struct { Protocol int `json:"protocol"` Binding guard.CoordinationRequest `json:"binding"` Repository string `json:"repository"` UploadID string `json:"upload_id"` }{1, binding, body.Repository, body.UploadID}) } // AcquireUpload admits only a continuation of a recorded upload session. func (h *CleanupCoordinationHandler) AcquireUpload(w http.ResponseWriter, r *http.Request) { var body struct { Parent guard.CoordinationRequest `json:"parent"` Request guard.CoordinationRequest `json:"request"` Closing *bool `json:"closing"` } if !coordinationDecode(w, r, &body) || !coordinationScopeFromRoute(w, r, &body.Parent) { return } body.Request.StorageID = body.Parent.StorageID if body.Closing == nil { httputil.ReturnError(r, w, 400, "UPLOAD_PHASE_REQUIRED") return } created, err := guard.AcquireUploadRequest(h.database(), body.Parent, body.Request, *body.Closing) if err != nil { coordinationError(w, r, err) return } httputil.ReturnSuccess(r, w, struct { Protocol int `json:"protocol"` NewlyAdmitted bool `json:"newly_admitted"` }{1, created}) } // FinishUpload keeps the upload parent until the closing request is confirmed. func (h *CleanupCoordinationHandler) FinishUpload(w http.ResponseWriter, r *http.Request) { var body struct { Parent guard.CoordinationRequest `json:"parent"` Request guard.CoordinationRequest `json:"request"` Outcome string `json:"outcome"` } if !coordinationDecode(w, r, &body) || !coordinationScopeFromRoute(w, r, &body.Parent) { return } body.Request.StorageID = body.Parent.StorageID if body.Outcome == "" { httputil.ReturnError(r, w, 400, "CONFIRMATION_REQUIRED") return } if err := guard.RecordUploadRequest(h.database(), body.Parent, body.Request, body.Outcome); err != nil { coordinationError(w, r, err) return } httputil.ReturnSuccess(r, w, struct { Protocol int `json:"protocol"` Recorded bool `json:"recorded"` }{1, true}) } // StorageStatus exposes identity and enrollment state, never a writable override. func (h *CleanupCoordinationHandler) StorageStatus(w http.ResponseWriter, r *http.Request) { var body struct { Generation string `json:"generation"` } if !coordinationDecode(w, r, &body) { return } observation, err := guard.InspectStorage(h.database(), chi.URLParam(r, "storage_id"), body.Generation) if err != nil { coordinationError(w, r, err) return } httputil.ReturnSuccess(r, w, struct { Protocol int `json:"protocol"` Storage guard.StorageObservation `json:"storage"` }{1, observation}) } func systemRegistryInspectionTarget() (kubernetes.Interface, string, string, error) { component := k8s.Default() configuration := configs.Default() if component == nil || component.Clientset == nil || configuration.PublicConfig == nil || configuration.ServerConfig == nil { return nil, "", "", kubeidentity.ErrBinding } namespace := configuration.PublicConfig.RbdNamespace endpoint, err := url.Parse(configuration.ServerConfig.RbdHub) if err != nil || endpoint.User != nil || namespace == "" { return nil, "", "", kubeidentity.ErrBinding } host := endpoint.Hostname() service := strings.Split(host, ".")[0] if host != service && host != service+"."+namespace && host != service+"."+namespace+".svc" && host != service+"."+namespace+".svc.cluster.local" { return nil, "", "", kubeidentity.ErrBinding } return component.Clientset, namespace, service, nil } func inspectSystemRegistry(ctx context.Context, pod, uid string) (kubeidentity.RegistryPreparation, error) { client, namespace, service, err := systemRegistryInspectionTarget() if err != nil { return kubeidentity.RegistryPreparation{}, err } return kubeidentity.InspectNativeRegistry(ctx, client, namespace, service, pod, uid) } func inspectSystemRegistryParticipant(ctx context.Context, pod, uid, owner string, binding guard.StorageRegistration) (guard.ParticipantRegistration, error) { client, namespace, service, err := systemRegistryInspectionTarget() if err != nil { return guard.ParticipantRegistration{}, err } return kubeidentity.InspectRegistryParticipant(ctx, client, namespace, service, pod, uid, owner, binding) } var registryPodUID = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9_.:-]{0,63}$`) // PrepareRegistry derives identity from the configured system registry, never // from caller-provided volume IDs, generations, namespaces or filesystem paths. func (h *CleanupCoordinationHandler) PrepareRegistry(w http.ResponseWriter, r *http.Request) { var body struct { Pod string `json:"pod"` PodUID string `json:"pod_uid"` } if !coordinationDecode(w, r, &body) { return } if len(validation.IsDNS1123Subdomain(body.Pod)) != 0 || !registryPodUID.MatchString(body.PodUID) { httputil.ReturnError(r, w, 400, "INVALID_REGISTRY_POD") return } if h.inspectRegistry == nil { coordinationError(w, r, guard.ErrCoordinationUnavailable) return } observed, err := h.inspectRegistry(r.Context(), body.Pod, body.PodUID) if err != nil { coordinationError(w, r, guard.ErrCoordinationUnavailable) return } binding, err := guard.ProvisionRegistryStorage(h.database(), observed.Mount.VolumeUID, observed.Root) if err != nil { coordinationError(w, r, err) return } status, err := guard.InspectStorage(h.database(), binding.StorageID, binding.Generation) if err != nil { coordinationError(w, r, err) return } httputil.ReturnSuccess(r, w, struct { Protocol int `json:"protocol"` Registration guard.StorageRegistration `json:"registration"` Storage guard.StorageObservation `json:"storage"` Container string `json:"registry_container"` }{1, binding, status, observed.Container}) } // RegisterRegistryParticipant accepts only runtime facts verified by the core. func (h *CleanupCoordinationHandler) RegisterRegistryParticipant(w http.ResponseWriter, r *http.Request) { var body struct { Generation string `json:"generation"` Owner string `json:"owner"` Pod string `json:"pod"` PodUID string `json:"pod_uid"` } if !coordinationDecode(w, r, &body) { return } if len(validation.IsDNS1123Subdomain(body.Pod)) != 0 || !registryPodUID.MatchString(body.PodUID) || body.Owner == "" || len(body.Owner) > 128 { httputil.ReturnError(r, w, 400, "INVALID_PARTICIPANT_REQUEST") return } if h.inspectParticipant == nil { coordinationError(w, r, guard.ErrCoordinationUnavailable) return } binding, err := guard.StorageBinding(h.database(), chi.URLParam(r, "storage_id"), body.Generation) if err != nil { coordinationError(w, r, err) return } observed, err := h.inspectParticipant(r.Context(), body.Pod, body.PodUID, body.Owner, binding) if err != nil { coordinationError(w, r, guard.ErrCoordinationUnavailable) return } fingerprint, _ := binding.Fingerprint() if observed.StorageID != binding.StorageID || observed.Generation != binding.Generation || observed.Owner != body.Owner || observed.PodUID != body.PodUID || observed.Role != "registry-ingress" || observed.BindingFingerprint != fingerprint { coordinationError(w, r, guard.ErrCoordinationChanged) return } if err := guard.RegisterParticipant(h.database(), observed); err != nil { coordinationError(w, r, err) return } httputil.ReturnSuccess(r, w, struct { Protocol int `json:"protocol"` Recorded bool `json:"recorded"` }{1, true}) } func inspectSystemManagedCache(ctx context.Context, pod, uid string) (kubeidentity.ManagedCachePreparation, error) { client, namespace, _, err := systemRegistryInspectionTarget() if err != nil { return kubeidentity.ManagedCachePreparation{}, err } return kubeidentity.InspectManagedBuildCacheSource(ctx, client, namespace, pod, uid) } func inspectSystemManagedPackages(ctx context.Context, pod, uid string) (kubeidentity.ManagedPackagePreparation, error) { client, namespace, _, err := systemRegistryInspectionTarget() if err != nil { return kubeidentity.ManagedPackagePreparation{}, err } return kubeidentity.InspectManagedPackageSource(ctx, client, namespace, pod, uid) } func inspectSystemManagedPackageWriters(ctx context.Context) ([]guard.ReferenceWriter, error) { client, namespace, _, err := systemRegistryInspectionTarget() if err != nil { return nil, err } return kubeidentity.InspectManagedPackageWriterCoverage(ctx, client, namespace) } // PrepareManagedPackages enrolls the two fixed package_build subtrees from one // observed rbd-chaos volume and certifies readiness from current writers. It // never accepts paths or caller-declared readiness. func (h *CleanupCoordinationHandler) PrepareManagedPackages(w http.ResponseWriter, r *http.Request) { var body struct { Pod string `json:"pod"` PodUID string `json:"pod_uid"` } if !coordinationDecode(w, r, &body) { return } if len(validation.IsDNS1123Subdomain(body.Pod)) != 0 || !registryPodUID.MatchString(body.PodUID) { httputil.ReturnError(r, w, 400, "INVALID_MANAGED_PACKAGE_POD") return } if h.inspectManagedPackages == nil { coordinationError(w, r, guard.ErrCoordinationUnavailable) return } observed, err := h.inspectManagedPackages(r.Context(), body.Pod, body.PodUID) if err != nil || observed.NodeUID == "" || observed.Root != "/grdata/package_build" { coordinationError(w, r, guard.ErrCoordinationUnavailable) return } registrations := make([]guard.StorageRegistration, 0, 2) storages := make([]guard.StorageObservation, 0, 2) quiescent := true for _, kind := range []string{"upload_events", "upload_components"} { binding, err := guard.ProvisionManagedPackageStorage(h.database(), observed.Mount.VolumeUID, kind) if err != nil { coordinationError(w, r, err) return } status, ready, err := h.certifyPackageWriters(r.Context(), binding, nil) if err != nil { coordinationError(w, r, err) return } quiescent = quiescent && ready registrations = append(registrations, binding) storages = append(storages, status) } httputil.ReturnSuccess(r, w, struct { Protocol int `json:"protocol"` Quiescent bool `json:"quiescent"` Registrations []guard.StorageRegistration `json:"registrations"` Storages []guard.StorageObservation `json:"storages"` NodeUID string `json:"node_uid"` NodeName string `json:"node_name"` }{1, quiescent, registrations, storages, observed.NodeUID, observed.Mount.NodeName}) } // PrepareManagedCache enrolls only observed system build-cache storage. It // never accepts filesystem paths or grants deletion readiness. func (h *CleanupCoordinationHandler) PrepareManagedCache(w http.ResponseWriter, r *http.Request) { var body struct { Pod string `json:"pod"` PodUID string `json:"pod_uid"` } if !coordinationDecode(w, r, &body) { return } if len(validation.IsDNS1123Subdomain(body.Pod)) != 0 || !registryPodUID.MatchString(body.PodUID) { httputil.ReturnError(r, w, 400, "INVALID_MANAGED_CACHE_POD") return } if h.inspectManagedCache == nil { coordinationError(w, r, guard.ErrCoordinationUnavailable) return } observed, err := h.inspectManagedCache(r.Context(), body.Pod, body.PodUID) if err != nil || observed.NodeUID == "" { coordinationError(w, r, guard.ErrCoordinationUnavailable) return } binding, err := guard.ProvisionManagedCacheStorage(h.database(), observed.Mount.VolumeUID, observed.Root) if err != nil { coordinationError(w, r, err) return } status, quiescent, err := h.certifyCacheWriters(r.Context(), binding, nil) if err != nil { coordinationError(w, r, err) return } httputil.ReturnSuccess(r, w, struct { Protocol int `json:"protocol"` Quiescent bool `json:"quiescent"` Registration guard.StorageRegistration `json:"registration"` Storage guard.StorageObservation `json:"storage"` NodeUID string `json:"node_uid"` NodeName string `json:"node_name"` }{1, quiescent, binding, status, observed.NodeUID, observed.Mount.NodeName}) } func systemClusterReferenceInventory(ctx context.Context) (guard.RegionReferenceInventory, error) { component := k8s.Default() if component == nil || component.RestConfig == nil || component.Clientset == nil { return guard.RegionReferenceInventory{}, guard.ErrCoordinationUnavailable } client, err := metadata.NewForConfig(component.RestConfig) if err != nil { return guard.RegionReferenceInventory{}, guard.ErrCoordinationUnavailable } helm, err := kubeidentity.ReadHelmReferenceInventory(ctx, component.Clientset, client) if err != nil { return guard.RegionReferenceInventory{}, err } configuration := configs.Default() if configuration.PublicConfig == nil { return guard.RegionReferenceInventory{}, guard.ErrCoordinationUnavailable } helpers, err := kubeidentity.ReadPlatformHelperReferences(ctx, component.Clientset, component.DynamicClient, configuration.PublicConfig.RbdNamespace) if err != nil { return guard.RegionReferenceInventory{}, err } desired, err := kubeidentity.ReadHelmAppReferenceInventory(ctx, component.RainbondClient, helm.HelmReleases) if err != nil { return guard.RegionReferenceInventory{}, err } return guard.MergeReferenceInventories(helm, helpers, desired), nil }