feat: add admin runjob endpoint for streaming Kubernetes Job output

Adds POST /admin/.../runjob which accepts a Job manifest (JSON), creates
the Job on the cluster, streams the pod's logs back to the caller as they
arrive, then deletes the Job on completion or cancellation. Namespace
defaults to openshift-azure-operator if absent; a random 5-character suffix
is appended to the job name to prevent collisions.

Also adds KubeFollowPodLogs to KubeActions for log streaming with Follow=true,
composable by higher-level actions.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: ventifus

pkg/frontend/admin_openshiftcluster_runjob.go

@@ -0,0 +1,250 @@
+package frontend
+
+// Copyright (c) Microsoft Corporation.
+// Licensed under the Apache License 2.0.
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"path/filepath"
+	"strings"
+	"time"
+
+	"github.com/go-chi/chi/v5"
+	"github.com/sirupsen/logrus"
+
+	batchv1 "k8s.io/api/batch/v1"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	kruntime "k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	utilrand "k8s.io/apimachinery/pkg/util/rand"
+	"k8s.io/apimachinery/pkg/watch"
+
+	"github.com/Azure/ARO-RP/pkg/api"
+	"github.com/Azure/ARO-RP/pkg/database/cosmosdb"
+	"github.com/Azure/ARO-RP/pkg/frontend/adminactions"
+	"github.com/Azure/ARO-RP/pkg/frontend/middleware"
+)
+
+const (
+	runJobDefaultNamespace = "openshift-azure-operator"
+	runJobCleanupTimeout   = 30 * time.Second
+)
+
+func (f *frontend) postAdminOpenShiftClusterRunJob(w http.ResponseWriter, r *http.Request) {
+	ctx := r.Context()
+	log := ctx.Value(middleware.ContextKeyLog).(*logrus.Entry)
+	r.URL.Path = filepath.Dir(r.URL.Path)
+
+	reader, writer := io.Pipe()
+	err := f._postAdminOpenShiftClusterRunJob(ctx, r, log, writer)
+	var header http.Header
+	if err == nil {
+		header = http.Header{"Content-Type": []string{"text/plain"}}
+	}
+	f.streamResponder.AdminReplyStream(log, w, header, reader, err)
+}
+
+func (f *frontend) _postAdminOpenShiftClusterRunJob(ctx context.Context, r *http.Request, log *logrus.Entry, writer io.WriteCloser) error {
+	resType, resName, resGroupName := chi.URLParam(r, "resourceType"), chi.URLParam(r, "resourceName"), chi.URLParam(r, "resourceGroupName")
+
+	body := r.Context().Value(middleware.ContextKeyBody).([]byte)
+	job, err := parseAndValidateJob(body)
+	if err != nil {
+		return err
+	}
+
+	resourceID := strings.TrimPrefix(r.URL.Path, "/admin")
+
+	dbOpenShiftClusters, err := f.dbGroup.OpenShiftClusters()
+	if err != nil {
+		return api.NewCloudError(http.StatusInternalServerError, api.CloudErrorCodeInternalServerError, "", err.Error())
+	}
+
+	doc, err := dbOpenShiftClusters.Get(ctx, resourceID)
+	switch {
+	case cosmosdb.IsErrorStatusCode(err, http.StatusNotFound):
+		return api.NewCloudError(http.StatusNotFound, api.CloudErrorCodeResourceNotFound, "",
+			fmt.Sprintf("The Resource '%s/%s' under resource group '%s' was not found.", resType, resName, resGroupName))
+	case err != nil:
+		return err
+	}
+
+	k, err := f.kubeActionsFactory(log, f.env, doc.OpenShiftCluster)
+	if err != nil {
+		return err
+	}
+
+	go runJobStream(ctx, k, job, writer)
+	return nil
+}
+
+func parseAndValidateJob(body []byte) (*batchv1.Job, error) {
+	if len(body) == 0 {
+		return nil, api.NewCloudError(http.StatusBadRequest, api.CloudErrorCodeInvalidRequestContent, "",
+			"The request body must not be empty.")
+	}
+
+	var raw map[string]interface{}
+	if err := json.Unmarshal(body, &raw); err != nil {
+		return nil, api.NewCloudError(http.StatusBadRequest, api.CloudErrorCodeInvalidRequestContent, "",
+			fmt.Sprintf("Failed to parse request body: %v", err))
+	}
+
+	kind, _ := raw["kind"].(string)
+	if kind != "Job" {
+		return nil, api.NewCloudError(http.StatusBadRequest, api.CloudErrorCodeInvalidParameter, "",
+			fmt.Sprintf("Expected kind 'Job', got '%s'.", kind))
+	}
+
+	var job batchv1.Job
+	if err := kruntime.DefaultUnstructuredConverter.FromUnstructured(raw, &job); err != nil {
+		return nil, api.NewCloudError(http.StatusBadRequest, api.CloudErrorCodeInvalidRequestContent, "",
+			fmt.Sprintf("Failed to convert manifest to Job: %v", err))
+	}
+
+	if job.Name == "" {
+		return nil, api.NewCloudError(http.StatusBadRequest, api.CloudErrorCodeInvalidParameter, "",
+			"The provided Job manifest must have a non-empty metadata.name.")
+	}
+
+	if job.Namespace == "" {
+		job.Namespace = runJobDefaultNamespace
+	}
+
+	job.Name = job.Name + "-" + utilrand.String(5)
+
+	return &job, nil
+}
+
+// runJobStream creates a Kubernetes Job on the cluster, streams the pod's logs back to w
+// as they arrive, then deletes the Job regardless of outcome. It is called in a goroutine
+// by the HTTP handler and may also be called directly by higher-level composed actions.
+func runJobStream(ctx context.Context, k adminactions.KubeActions, job *batchv1.Job, w io.WriteCloser) {
+	defer w.Close()
+
+	namespace := job.Namespace
+	jobName := job.Name
+
+	fmt.Fprintf(w, "Creating job %s in %s...\n", jobName, namespace)
+
+	unstrMap, err := kruntime.DefaultUnstructuredConverter.ToUnstructured(job)
+	if err != nil {
+		fmt.Fprintf(w, "Failed to prepare job manifest: %v\n", err)
+		return
+	}
+	un := &unstructured.Unstructured{Object: unstrMap}
+	un.SetGroupVersionKind(schema.GroupVersionKind{Group: "batch", Version: "v1", Kind: "Job"})
+
+	if err := k.KubeCreateOrUpdate(ctx, un); err != nil {
+		fmt.Fprintf(w, "Failed to create job: %v\n", err)
+		return
+	}
+
+	fmt.Fprintf(w, "Waiting for pod...\n")
+	podName, err := waitForJobPod(ctx, k, namespace, jobName)
+	if err != nil {
+		fmt.Fprintf(w, "Error waiting for pod: %v\n", err)
+		cleanupJob(k, namespace, jobName)
+		return
+	}
+
+	fmt.Fprintf(w, "Pod %s assigned, streaming logs...\n", podName)
+
+	if err := k.KubeFollowPodLogs(ctx, namespace, podName, "", newLimitedWriter(w, "pod logs")); err != nil && ctx.Err() == nil {
+		fmt.Fprintf(w, "Log streaming error: %v\n", err)
+	}
+
+	if ctx.Err() != nil {
+		fmt.Fprintf(w, "Request cancelled.\n")
+		cleanupJob(k, namespace, jobName)
+		return
+	}
+
+	if ok, err := jobSucceeded(ctx, k, namespace, jobName); err != nil {
+		fmt.Fprintf(w, "Could not determine job result: %v\n", err)
+	} else if ok {
+		fmt.Fprintf(w, "Job succeeded.\n")
+	} else {
+		fmt.Fprintf(w, "Job failed.\n")
+	}
+
+	cleanupJob(k, namespace, jobName)
+	fmt.Fprintf(w, "Cleanup complete.\n")
+}
+
+// waitForJobPod watches pods with the job-name label until one reaches Running,
+// Succeeded, or Failed phase, then returns the pod name.
+func waitForJobPod(ctx context.Context, k adminactions.KubeActions, namespace, jobName string) (string, error) {
+	podTemplate := &unstructured.Unstructured{}
+	podTemplate.SetGroupVersionKind(schema.GroupVersionKind{Version: "v1", Kind: "Pod"})
+	podTemplate.SetNamespace(namespace)
+	podTemplate.SetLabels(map[string]string{"batch.kubernetes.io/job-name": jobName})
+
+	watcher, err := k.KubeWatch(ctx, podTemplate, "batch.kubernetes.io/job-name")
+	if err != nil {
+		return "", fmt.Errorf("watching pods: %w", err)
+	}
+	defer watcher.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return "", ctx.Err()
+		case event, ok := <-watcher.ResultChan():
+			if !ok {
+				return "", fmt.Errorf("pod watch channel closed unexpectedly")
+			}
+			if event.Type != watch.Added && event.Type != watch.Modified {
+				continue
+			}
+			pod, ok := event.Object.(*unstructured.Unstructured)
+			if !ok {
+				continue
+			}
+			name := pod.GetName()
+			phase, _, _ := unstructured.NestedString(pod.Object, "status", "phase")
+			switch corev1.PodPhase(phase) {
+			case corev1.PodRunning, corev1.PodSucceeded, corev1.PodFailed:
+				return name, nil
+			}
+		}
+	}
+}
+
+// jobSucceeded returns true if the Job has a Complete=True condition.
+func jobSucceeded(ctx context.Context, k adminactions.KubeActions, namespace, jobName string) (bool, error) {
+	data, err := k.KubeGet(ctx, "Job.batch", namespace, jobName)
+	if err != nil {
+		return false, err
+	}
+	var obj map[string]interface{}
+	if err := json.Unmarshal(data, &obj); err != nil {
+		return false, err
+	}
+	conditions, _, _ := unstructured.NestedSlice(obj, "status", "conditions")
+	for _, c := range conditions {
+		m, ok := c.(map[string]interface{})
+		if !ok {
+			continue
+		}
+		if m["type"] == "Complete" && m["status"] == "True" {
+			return true, nil
+		}
+	}
+	return false, nil
+}
+
+// cleanupJob deletes the Job with a fresh context so that cancellation of the
+// caller's request context does not prevent cleanup.
+func cleanupJob(k adminactions.KubeActions, namespace, jobName string) {
+	ctx, cancel := context.WithTimeout(context.Background(), runJobCleanupTimeout)
+	defer cancel()
+	foreground := metav1.DeletePropagationForeground
+	_ = k.KubeDelete(ctx, "Job.batch", namespace, jobName, false, &foreground)
+}