opensandbox-group--opensandbox/kubernetes/internal/task-executor/server/handler.go

// Copyright 2025 Alibaba Group Holding Ltd.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package server

import (
	"encoding/json"
	"fmt"
	"net/http"
	"time"

	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/klog/v2"

	"github.com/alibaba/OpenSandbox/sandbox-k8s/internal/task-executor/config"
	"github.com/alibaba/OpenSandbox/sandbox-k8s/internal/task-executor/manager"
	"github.com/alibaba/OpenSandbox/sandbox-k8s/internal/task-executor/types"
	api "github.com/alibaba/OpenSandbox/sandbox-k8s/pkg/task-executor"
)

// ErrorResponse represents a standard error response
type ErrorResponse struct {
	Code    string `json:"code"`
	Message string `json:"message"`
}

type Handler struct {
	manager manager.TaskManager
	config  *config.Config
}

func NewHandler(mgr manager.TaskManager, cfg *config.Config) *Handler {
	if mgr == nil {
		klog.Warning("TaskManager is nil, handler may not work properly")
	}
	if cfg == nil {
		klog.Warning("Config is nil, handler may not work properly")
	}
	return &Handler{
		manager: mgr,
		config:  cfg,
	}
}

func (h *Handler) CreateTask(w http.ResponseWriter, r *http.Request) {
	if h.manager == nil {
		writeError(w, http.StatusInternalServerError, "task manager not initialized")
		return
	}

	var apiTask api.Task
	if err := json.NewDecoder(r.Body).Decode(&apiTask); err != nil {
		writeError(w, http.StatusBadRequest, fmt.Sprintf("invalid request body: %v", err))
		return
	}

	if apiTask.Name == "" {
		writeError(w, http.StatusBadRequest, "task name is required")
		return
	}

	task := h.convertAPIToInternalTask(&apiTask)
	if task == nil {
		writeError(w, http.StatusBadRequest, "failed to convert task")
		return
	}

	created, err := h.manager.Create(r.Context(), task)
	if err != nil {
		klog.ErrorS(err, "failed to create task", "name", apiTask.Name)
		writeError(w, http.StatusInternalServerError, fmt.Sprintf("failed to create task: %v", err))
		return
	}

	response := convertInternalToAPITask(created)

	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(http.StatusCreated)
	json.NewEncoder(w).Encode(response)

	klog.InfoS("task created via API", "name", apiTask.Name)
}

func (h *Handler) SyncTasks(w http.ResponseWriter, r *http.Request) {
	if h.manager == nil {
		writeError(w, http.StatusInternalServerError, "task manager not initialized")
		return
	}

	var apiTasks []api.Task
	if err := json.NewDecoder(r.Body).Decode(&apiTasks); err != nil {
		writeError(w, http.StatusBadRequest, fmt.Sprintf("invalid request body: %v", err))
		return
	}

	desired := make([]*types.Task, 0, len(apiTasks))
	for i := range apiTasks {
		if apiTasks[i].Name == "" {
			continue
		}
		task := h.convertAPIToInternalTask(&apiTasks[i])
		if task != nil {
			desired = append(desired, task)
		}
	}

	current, err := h.manager.Sync(r.Context(), desired)
	if err != nil {
		klog.ErrorS(err, "failed to sync tasks")
		writeError(w, http.StatusInternalServerError, fmt.Sprintf("failed to sync tasks: %v", err))
		return
	}

	response := make([]api.Task, 0, len(current))
	for _, task := range current {
		if task != nil {
			response = append(response, *convertInternalToAPITask(task))
		}
	}

	w.Header().Set("Content-Type", "application/json")
	json.NewEncoder(w).Encode(response)

	klog.V(1).InfoS("tasks synced via API", "count", len(response))
}

func (h *Handler) GetTask(w http.ResponseWriter, r *http.Request) {
	if h.manager == nil {
		writeError(w, http.StatusInternalServerError, "task manager not initialized")
		return
	}

	// Extract task ID from path
	taskID := r.PathValue("id")
	if taskID == "" {
		writeError(w, http.StatusBadRequest, "task id is required")
		return
	}

	task, err := h.manager.Get(r.Context(), taskID)
	if err != nil {
		klog.ErrorS(err, "failed to get task", "id", taskID)
		writeError(w, http.StatusNotFound, fmt.Sprintf("task not found: %v", err))
		return
	}

	response := convertInternalToAPITask(task)

	w.Header().Set("Content-Type", "application/json")
	json.NewEncoder(w).Encode(response)
}

func (h *Handler) ListTasks(w http.ResponseWriter, r *http.Request) {
	if h.manager == nil {
		writeError(w, http.StatusInternalServerError, "task manager not initialized")
		return
	}

	tasks, err := h.manager.List(r.Context())
	if err != nil {
		klog.ErrorS(err, "failed to list tasks")
		writeError(w, http.StatusInternalServerError, fmt.Sprintf("failed to list tasks: %v", err))
		return
	}

	response := make([]api.Task, 0, len(tasks))
	for _, task := range tasks {
		if task != nil {
			response = append(response, *convertInternalToAPITask(task))
		}
	}

	w.Header().Set("Content-Type", "application/json")
	json.NewEncoder(w).Encode(response)
}

func (h *Handler) Health(w http.ResponseWriter, r *http.Request) {
	response := map[string]string{
		"status": "healthy",
	}
	w.Header().Set("Content-Type", "application/json")
	json.NewEncoder(w).Encode(response)
}

func (h *Handler) DeleteTask(w http.ResponseWriter, r *http.Request) {
	if h.manager == nil {
		writeError(w, http.StatusInternalServerError, "task manager not initialized")
		return
	}

	// Extract task ID from path
	taskID := r.PathValue("id")
	if taskID == "" {
		writeError(w, http.StatusBadRequest, "task id is required")
		return
	}

	err := h.manager.Delete(r.Context(), taskID)
	if err != nil {
		klog.ErrorS(err, "failed to delete task", "id", taskID)
		writeError(w, http.StatusInternalServerError, fmt.Sprintf("failed to delete task: %v", err))
		return
	}

	w.WriteHeader(http.StatusNoContent)
	klog.InfoS("task deleted via API", "id", taskID)
}

func writeError(w http.ResponseWriter, code int, message string) {
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(code)
	json.NewEncoder(w).Encode(ErrorResponse{
		Code:    http.StatusText(code),
		Message: message,
	})
}

func (h *Handler) convertAPIToInternalTask(apiTask *api.Task) *types.Task {
	if apiTask == nil {
		return nil
	}
	task := &types.Task{
		Name:            apiTask.Name,
		Process:         apiTask.Process,
		PodTemplateSpec: apiTask.PodTemplateSpec,
	}
	task.Status = types.Status{
		State: types.TaskStatePending,
	}

	return task
}

func convertInternalToAPITask(task *types.Task) *api.Task {
	if task == nil {
		return nil
	}

	apiTask := &api.Task{
		Name:            task.Name,
		Process:         task.Process,
		PodTemplateSpec: task.PodTemplateSpec,
	}

	if task.Process != nil && len(task.Status.SubStatuses) > 0 {
		sub := task.Status.SubStatuses[0]
		apiStatus := &api.ProcessStatus{}

		if task.Status.State == types.TaskStateTimeout {
			term := &api.Terminated{
				ExitCode: 137,
				Reason:   sub.Reason,
				Message:  sub.Message,
			}
			if sub.StartedAt != nil {
				term.StartedAt = metav1.NewTime(*sub.StartedAt)
			}
			term.FinishedAt = metav1.Now()
			apiStatus.Terminated = term
		} else if sub.FinishedAt != nil {
			term := &api.Terminated{
				ExitCode: int32(sub.ExitCode),
				Reason:   sub.Reason,
				Message:  sub.Message,
			}
			term.FinishedAt = metav1.NewTime(*sub.FinishedAt)
			if sub.StartedAt != nil {
				term.StartedAt = metav1.NewTime(*sub.StartedAt)
			}
			apiStatus.Terminated = term
		} else if sub.StartedAt != nil {
			apiStatus.Running = &api.Running{
				StartedAt: metav1.NewTime(*sub.StartedAt),
			}
		} else {
			apiStatus.Waiting = &api.Waiting{
				Reason:  sub.Reason,
				Message: sub.Message,
			}
		}
		apiTask.ProcessStatus = apiStatus
	}

	if task.PodTemplateSpec != nil {
		podStatus := &corev1.PodStatus{
			Phase: corev1.PodUnknown,
		}

		switch task.Status.State {
		case types.TaskStatePending:
			podStatus.Phase = corev1.PodPending
		case types.TaskStateRunning:
			podStatus.Phase = corev1.PodRunning
		case types.TaskStateSucceeded:
			podStatus.Phase = corev1.PodSucceeded
		case types.TaskStateFailed:
			podStatus.Phase = corev1.PodFailed
		}

		for _, sub := range task.Status.SubStatuses {
			cs := corev1.ContainerStatus{
				Name: sub.Name,
			}
			if sub.FinishedAt != nil {
				cs.State.Terminated = &corev1.ContainerStateTerminated{
					ExitCode:   int32(sub.ExitCode),
					Reason:     sub.Reason,
					Message:    sub.Message,
					FinishedAt: metav1.NewTime(*sub.FinishedAt),
				}
				if sub.StartedAt != nil {
					cs.State.Terminated.StartedAt = metav1.NewTime(*sub.StartedAt)
				}
			} else if sub.StartedAt != nil {
				cs.State.Running = &corev1.ContainerStateRunning{
					StartedAt: metav1.NewTime(*sub.StartedAt),
				}
				cs.Ready = true
			} else {
				cs.State.Waiting = &corev1.ContainerStateWaiting{
					Reason:  sub.Reason,
					Message: sub.Message,
				}
			}
			podStatus.ContainerStatuses = append(podStatus.ContainerStatuses, cs)
		}

		allReady := len(podStatus.ContainerStatuses) > 0
		for _, cs := range podStatus.ContainerStatuses {
			if !cs.Ready {
				allReady = false
				break
			}
		}
		readyStatus := corev1.ConditionFalse
		if allReady {
			readyStatus = corev1.ConditionTrue
		}

		var latestTransition time.Time
		for _, sub := range task.Status.SubStatuses {
			if sub.StartedAt != nil && sub.StartedAt.After(latestTransition) {
				latestTransition = *sub.StartedAt
			}
			if sub.FinishedAt != nil && sub.FinishedAt.After(latestTransition) {
				latestTransition = *sub.FinishedAt
			}
		}
		ltt := metav1.NewTime(latestTransition)
		if latestTransition.IsZero() {
			ltt = metav1.Now()
		}

		podStatus.Conditions = append(podStatus.Conditions,
			corev1.PodCondition{
				Type:               corev1.PodReady,
				Status:             readyStatus,
				LastTransitionTime: ltt,
			},
			corev1.PodCondition{
				Type:               corev1.ContainersReady,
				Status:             readyStatus,
				LastTransitionTime: ltt,
			},
		)

		apiTask.PodStatus = podStatus
	}

	return apiTask
}