api/controller/cleanup_node_launch.go

Rama main · réplica de lectura del repositorio original.

package controller

import (
	"context"
	"errors"
	"net/http"
	"os"

	guard "github.com/goodrain/rainbond/pkg/cleanup"
	"github.com/goodrain/rainbond/pkg/cleanup/kubeidentity"
	httputil "github.com/goodrain/rainbond/util/http"
)

// NodeJobSelection contains the saved scan identity; paths and executor settings
// are intentionally absent. All supplied locators are checked against Kubernetes.
type NodeJobSelection struct {
	guard.CoordinationRequest
	SourcePod        string `json:"source_pod"`
	SourceUID        string `json:"source_uid"`
	NodeName         string `json:"node_name"`
	NodeUID          string `json:"node_uid"`
	Entry            string `json:"entry"`
	EntryFingerprint string `json:"entry_fingerprint"`
}

func systemNodeJobSettings() kubeidentity.NodeJobSettings {
	return kubeidentity.NodeJobSettings{Region: os.Getenv("REGION_NAME"), Image: os.Getenv("CLEANUP_NODE_EXECUTOR_IMAGE"), Endpoint: os.Getenv("CLEANUP_NODE_CORE_ENDPOINT"), CredentialSecret: os.Getenv("CLEANUP_NODE_CORE_SECRET"), StateClaim: os.Getenv("CLEANUP_NODE_STATE_CLAIM"), ConsoleEnterprise: os.Getenv("CLEANUP_NODE_CONSOLE_ENTERPRISE"), AllowConsoleHTTP: os.Getenv("CLEANUP_NODE_CONSOLE_ALLOW_HTTP") == "true"}
}

// SubmitNodeJob builds an immutable suspended executor for an admitted selection.
func (h *CleanupCoordinationHandler) SubmitNodeJob(w http.ResponseWriter, r *http.Request) {
	var body NodeJobSelection
	if !coordinationDecode(w, r, &body) || !coordinationScopeFromRoute(w, r, &body.CoordinationRequest) {
		return
	}
	if h.gcTarget == nil || h.nodeSettings == nil {
		coordinationError(w, r, guard.ErrCoordinationUnavailable)
		return
	}
	client, namespace, _, err := h.gcTarget()
	if err != nil || client == nil {
		coordinationError(w, r, guard.ErrCoordinationUnavailable)
		return
	}
	database := h.database()
	binding, err := guard.StorageBinding(database, body.StorageID, body.Generation)
	if err != nil {
		coordinationError(w, r, err)
		return
	}
	intent := guard.NodeJobIntent{Namespace: namespace, NodeName: body.NodeName, NodeUID: body.NodeUID, Entry: body.Entry, Fingerprint: body.EntryFingerprint}
	expected, err := guard.ManagedNodeRequest(binding, body.Owner, body.OperationID, body.Fingerprint, intent)
	if err != nil || expected != body.CoordinationRequest {
		coordinationError(w, r, guard.ErrCoordinationChanged)
		return
	}
	existing, err := guard.ReadNodeJobBinding(database, body.CoordinationRequest)
	if err == nil {
		if existing.Namespace != namespace || existing.NodeName != intent.NodeName || existing.NodeUID != intent.NodeUID || existing.Entry != intent.Entry || existing.Fingerprint != intent.Fingerprint {
			coordinationError(w, r, guard.ErrCoordinationChanged)
			return
		}
		if _, err := guard.ReconcileNodeJob(r.Context(), database, client.BatchV1().Jobs(namespace), body.CoordinationRequest); err != nil {
			coordinationError(w, r, err)
			return
		}
	} else {
		if !errors.Is(err, guard.ErrNodeJobNotPrepared) {
			coordinationError(w, r, err)
			return
		}
		template, err := kubeidentity.BuildManagedNodeJob(r.Context(), client, body.SourcePod, body.SourceUID, binding, body.CoordinationRequest, intent, h.nodeSettings())
		if err != nil {
			coordinationError(w, r, guard.ErrCoordinationChanged)
			return
		}
		if _, err := guard.SubmitSuspendedNodeJob(r.Context(), database, client.BatchV1().Jobs(namespace), body.CoordinationRequest, intent, template); err != nil {
			coordinationError(w, r, err)
			return
		}
	}
	httputil.ReturnSuccess(r, w, struct {
		Protocol int  `json:"protocol"`
		Recorded bool `json:"recorded"`
	}{1, true})
}

// StartNodeJob reconstructs the trusted source before starting the original Job.
func (h *CleanupCoordinationHandler) StartNodeJob(w http.ResponseWriter, r *http.Request) {
	var request guard.CoordinationRequest
	if !coordinationDecode(w, r, &request) || !coordinationScopeFromRoute(w, r, &request) {
		return
	}
	if h.gcTarget == nil || h.nodeSettings == nil {
		coordinationError(w, r, guard.ErrCoordinationUnavailable)
		return
	}
	client, namespace, _, err := h.gcTarget()
	if err != nil || client == nil {
		coordinationError(w, r, guard.ErrCoordinationUnavailable)
		return
	}
	database := h.database()
	binding, err := guard.ReadNodeJobBinding(database, request)
	if err != nil {
		coordinationError(w, r, err)
		return
	}
	if binding.Namespace != namespace {
		coordinationError(w, r, guard.ErrCoordinationChanged)
		return
	}
	job, err := guard.ReconcileNodeJob(r.Context(), database, client.BatchV1().Jobs(namespace), request)
	if err != nil {
		coordinationError(w, r, err)
		return
	}
	storage, err := guard.StorageBinding(database, request.StorageID, request.Generation)
	if err != nil {
		coordinationError(w, r, err)
		return
	}
	source := job.Spec.Template.Annotations
	current, err := kubeidentity.BuildManagedNodeJob(r.Context(), client, source["rainbond.io/node-source-pod"], source["rainbond.io/node-source-uid"], storage, request, binding.NodeJobIntent, h.nodeSettings())
	if err != nil || !kubeidentity.SameManagedNodeSource(job, current) {
		coordinationError(w, r, guard.ErrCoordinationChanged)
		return
	}
	if _, ready, err := h.certifyManagedNodeWriters(r.Context(), storage, &request); err != nil || !ready {
		if err == nil {
			err = guard.ErrCoordinationBusy
		}
		nodeAPIError(w, r, err)
		return
	}
	if _, err := guard.StartNodeJob(r.Context(), database, client.BatchV1().Jobs(namespace), request); err != nil {
		coordinationError(w, r, err)
		return
	}
	httputil.ReturnSuccess(r, w, struct {
		Protocol int  `json:"protocol"`
		Recorded bool `json:"recorded"`
	}{1, true})
}

// Recheck immediately before the one-use grant; result recovery must not depend
// on the continued availability of the builder that originally supplied a mount.
func (h *CleanupCoordinationHandler) validateNodeLaunchSource(ctx context.Context, request guard.CoordinationRequest) error {
	if h.gcTarget == nil || h.nodeSettings == nil {
		return guard.ErrCoordinationUnavailable
	}
	client, namespace, _, err := h.gcTarget()
	if err != nil || client == nil {
		return guard.ErrCoordinationUnavailable
	}
	binding, err := guard.ReadNodeJobBinding(h.database(), request)
	if err != nil {
		return err
	}
	if binding.Namespace != namespace {
		return guard.ErrCoordinationChanged
	}
	job, err := guard.ReconcileNodeJob(ctx, h.database(), client.BatchV1().Jobs(namespace), request)
	if err != nil {
		return err
	}
	storage, err := guard.StorageBinding(h.database(), request.StorageID, request.Generation)
	if err != nil {
		return err
	}
	source := job.Spec.Template.Annotations
	current, err := kubeidentity.BuildManagedNodeJob(ctx, client, source["rainbond.io/node-source-pod"], source["rainbond.io/node-source-uid"], storage, request, binding.NodeJobIntent, h.nodeSettings())
	if err != nil || !kubeidentity.SameManagedNodeSource(job, current) {
		return guard.ErrCoordinationChanged
	}
	if _, ready, err := h.certifyManagedNodeWriters(ctx, storage, &request); err != nil || !ready {
		if err != nil {
			return err
		}
		return guard.ErrCoordinationBusy
	}

	return nil
}

// CancelNodeBeforeGrant reconciles an original task that never obtained native
// authority. Core's storage transaction, not caller claims, proves that fact.
func (h *CleanupCoordinationHandler) CancelNodeBeforeGrant(w http.ResponseWriter, r *http.Request) {
	var body NodeJobSelection
	if !coordinationDecode(w, r, &body) || !coordinationScopeFromRoute(w, r, &body.CoordinationRequest) {
		return
	}
	if h.gcTarget == nil {
		coordinationError(w, r, guard.ErrCoordinationUnavailable)
		return
	}
	_, namespace, _, err := h.gcTarget()
	if err != nil || namespace == "" {
		coordinationError(w, r, guard.ErrCoordinationUnavailable)
		return
	}
	intent := guard.NodeJobIntent{Namespace: namespace, NodeName: body.NodeName, NodeUID: body.NodeUID, Entry: body.Entry, Fingerprint: body.EntryFingerprint}
	if err := guard.CancelNodeBeforeGrant(h.database(), body.CoordinationRequest, intent); err != nil {
		nodeAPIError(w, r, err)
		return
	}
	nodeRecorded(w, r)
}