api/controller/cleanup_node_executor.go

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

package controller

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

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

func (h *CleanupCoordinationHandler) inspectNodeExecution(ctx context.Context, request guard.CoordinationRequest, locator guard.NodeExecutorLocator, terminal bool) (guard.NodeExecutorIdentity, error) {
	denied := guard.NodeExecutorIdentity{}
	if err := ctx.Err(); err != nil {
		return denied, err
	}
	if h.gcTarget == nil {
		return denied, guard.ErrCoordinationUnavailable
	}
	database := h.database()
	execution, err := guard.ReadNodeJobBinding(database, request)
	if err != nil {
		return denied, err
	}
	client, namespace, _, err := h.gcTarget()
	if err != nil || client == nil || execution.Namespace != namespace {
		return denied, guard.ErrCoordinationChanged
	}
	job, err := guard.ReconcileNodeJob(ctx, database, client.BatchV1().Jobs(namespace), request)
	if err != nil {
		return denied, err
	}
	storage, err := guard.StorageBinding(database, request.StorageID, request.Generation)
	if err != nil {
		return denied, err
	}
	var observed guard.NodeExecutorIdentity
	if terminal {
		observed, err = kubeidentity.InspectTerminatedNodeExecutor(ctx, client, job, locator.Pod, locator.PodUID, storage, execution)
	} else {
		observed, err = kubeidentity.InspectNodeExecutor(ctx, client, job, locator.Pod, locator.PodUID, storage, execution)
	}
	if contextErr := ctx.Err(); contextErr != nil {
		return denied, contextErr
	}
	return observed, err
}
func nodeAPIError(w http.ResponseWriter, r *http.Request, err error) {
	if errors.Is(err, guard.ErrNodeJobNotPrepared) {
		httputil.ReturnError(r, w, 404, "NODE_JOB_NOT_PREPARED")
		return
	}
	if errors.Is(err, kubeidentity.ErrExecutorRunning) {
		err = guard.ErrCoordinationBusy
	}
	if errors.Is(err, kubeidentity.ErrBinding) {
		err = guard.ErrCoordinationChanged
	}
	coordinationError(w, r, err)
}
func nodeRecorded(w http.ResponseWriter, r *http.Request) {
	httputil.ReturnSuccess(r, w, struct {
		Protocol int  `json:"protocol"`
		Recorded bool `json:"recorded"`
	}{1, true})
}

// EnterNodeJob admits only a Core-bound node executor using live Kubernetes facts.
// This route is always protected by CleanupIdentity; caller runtime assertions fail decoding.
func (h *CleanupCoordinationHandler) EnterNodeJob(w http.ResponseWriter, r *http.Request) {
	var body struct {
		guard.CoordinationRequest
		guard.NodeExecutorLocator
	}
	if !coordinationDecode(w, r, &body) || !coordinationScopeFromRoute(w, r, &body.CoordinationRequest) {
		return
	}
	if !body.NodeExecutorLocator.Valid() {
		httputil.ReturnError(r, w, 400, "INVALID_NODE_EXECUTOR")
		return
	}
	if err := h.validateNodeLaunchSource(r.Context(), body.CoordinationRequest); err != nil {
		nodeAPIError(w, r, err)
		return
	}
	observed, err := h.inspectNodeExecution(r.Context(), body.CoordinationRequest, body.NodeExecutorLocator, false)
	if err != nil {
		nodeAPIError(w, r, err)
		return
	}
	if err := guard.EnterNodeExecution(h.database(), body.CoordinationRequest, observed); err != nil {
		nodeAPIError(w, r, err)
		return
	}
	nodeRecorded(w, r)
}

// RecordNodeJobResult stores effects from the original live or terminated executor.
// It never releases deletion protection or accepts caller termination assertions.
func (h *CleanupCoordinationHandler) RecordNodeJobResult(w http.ResponseWriter, r *http.Request) {
	var body struct {
		guard.CoordinationRequest
		guard.NodeExecutorLocator
		Result guard.NodeExecutionResult `json:"result"`
	}
	if !coordinationDecode(w, r, &body) || !coordinationScopeFromRoute(w, r, &body.CoordinationRequest) {
		return
	}
	if !body.NodeExecutorLocator.Valid() {
		httputil.ReturnError(r, w, 400, "INVALID_NODE_EXECUTOR")
		return
	}
	recorded, err := guard.NodeResultRecorded(h.database(), body.CoordinationRequest, body.NodeExecutorLocator, body.Result)
	if err != nil {
		nodeAPIError(w, r, err)
		return
	}
	if recorded {
		nodeRecorded(w, r)
		return
	}
	observed, err := h.inspectNodeExecution(r.Context(), body.CoordinationRequest, body.NodeExecutorLocator, false)
	if errors.Is(err, kubeidentity.ErrBinding) {
		observed, err = h.inspectNodeExecution(r.Context(), body.CoordinationRequest, body.NodeExecutorLocator, true)
	}
	if err != nil {
		nodeAPIError(w, r, err)
		return
	}
	if err := guard.RecordNodeResult(h.database(), body.CoordinationRequest, observed, body.Result); err != nil {
		nodeAPIError(w, r, err)
		return
	}
	nodeRecorded(w, r)
}

// FinishNodeJob checks original container termination before releasing its scope.
func (h *CleanupCoordinationHandler) FinishNodeJob(w http.ResponseWriter, r *http.Request) {
	var body struct {
		guard.CoordinationRequest
		guard.NodeExecutorLocator
	}
	if !coordinationDecode(w, r, &body) || !coordinationScopeFromRoute(w, r, &body.CoordinationRequest) {
		return
	}
	if !body.NodeExecutorLocator.Valid() {
		httputil.ReturnError(r, w, 400, "INVALID_NODE_EXECUTOR")
		return
	}
	progress, err := guard.ReadNodeJobProgress(h.database(), body.CoordinationRequest)
	if err != nil {
		nodeAPIError(w, r, err)
		return
	}
	if progress.State == "finished" {
		if progress.Execution.PodName != body.Pod || progress.Execution.PodUID != body.PodUID {
			nodeAPIError(w, r, guard.ErrCoordinationChanged)
			return
		}
		nodeRecorded(w, r)
		return
	}
	observed, err := h.inspectNodeExecution(r.Context(), body.CoordinationRequest, body.NodeExecutorLocator, true)
	if err != nil {
		nodeAPIError(w, r, err)
		return
	}
	if err := guard.FinishNodeExecution(h.database(), body.CoordinationRequest, observed); err != nil {
		nodeAPIError(w, r, err)
		return
	}
	nodeRecorded(w, r)
}

// NodeJobProgress reads a durable task receipt without mutating its native Job.
func (h *CleanupCoordinationHandler) NodeJobProgress(w http.ResponseWriter, r *http.Request) {
	var request guard.CoordinationRequest
	if !coordinationDecode(w, r, &request) || !coordinationScopeFromRoute(w, r, &request) {
		return
	}
	result, err := guard.ReadNodeJobProgress(h.database(), request)
	if err != nil {
		nodeAPIError(w, r, err)
		return
	}
	httputil.ReturnSuccess(r, w, struct {
		Protocol int                   `json:"protocol"`
		NodeJob  guard.NodeJobProgress `json:"node_job"`
	}{1, result})
}