feat: add log-persistence end-to-end (proto + mapper + controller)

Proto:
- Add LogPersistenceSpec message and log_url to RayClusterStatus
  (proto/api/v2/ray_cluster.proto)

Mapper (go/components/jobs/client/k8sengine):
- Load LogPersistenceConfig from controllers.rayCluster.logPersistence
  via FX
- injectCollectorSidecar(): inject the KubeRay History Server collector
  into Ray pod templates with env-var S3 config (no ConfigMap), reuse an
  existing /tmp/ray volume when present, set ImagePullPolicy: IfNotPresent,
  add the postStart raylet-node-id hook
- Wire sidecar injection into mapRayCluster() when logPersistence.enabled
- buildLogURL() and MapLocalClusterStatusToGlobal(): compute the
  human-browsable log URL during local→global status translation
  (mapper owns the formula since it knows config + local cluster name +
   compute-cluster Ray namespace = RayLocalNamespace)
  Format: {LogUrlBaseUrl}/{Bucket}/{PathPrefix}/{cluster}_{rayLocalNamespace}/

Controller (go/components/ray/cluster):
- applyRayClusterStatus(): copy clusterStatus.Ray.LogUrl onto
  rayCluster.Status.LogUrl so callers see the value the mapper computed
- Controller intentionally does NOT own the log_url formula or any
  LogPersistenceConfig — single source of truth lives in the mapper

Defaults (go/cmd/controllermgr/config/base.yaml):
- controllers.rayCluster.logPersistence:
    enabled: true
    storageEndpoint: "k3d-michelangelo-sandbox-agent-0:30007"
    bucket: ray-history
    pathPrefix: log
    region: us-east-1
    credentialsSecret: minio-credentials
    collectorImage: kuberay-collector:v0.1.0
    logUrlBaseUrl: "http://localhost:9090/browser"
- Sandbox MinIO defaults; production overlays flip these.

Tests cover: mapper sidecar injection (env vars, lifecycle preservation,
config knobs), controller log_url pass-through.

Author: dragod812

go/cmd/controllermgr/config/base.yaml

@@ -12,7 +12,7 @@ workflowClient:
   # taskList is a used as a fallback if project.Metatdata.Annotations["michelangelo/worker_queue"] is empty
   taskList: default # TODO(#934): MaCTL support for adding project.Metatdata.Annotations["michelangelo/worker_queue"] to project CRD
   executionUrlFormat: http://localhost:8088/namespaces/{{.Domain}}/workflows/{{.ExecutionID}}
-    
+
 k8s:
   qps: 300
   burst: 600
@@ -23,6 +23,15 @@ controllers:
   rayCluster:
     k8sQps: 300
     k8sBurst: 600
+    logPersistence:
+      enabled: true
+      storageEndpoint: "k3d-michelangelo-sandbox-agent-0:30007"
+      bucket: ray-history
+      pathPrefix: log
+      region: us-east-1
+      credentialsSecret: minio-credentials
+      collectorImage: kuberay-collector:v0.1.0
+      logUrlBaseUrl: "http://localhost:9090/browser"
 
 minio:
   awsRegion: us-east-1

go/components/jobs/client/client_test.go

@@ -188,7 +188,7 @@ func TestDeleteJob(t *testing.T) {
 			k8sc := Client{
 				factory: f,
 				helper:  NewHelper(),
-				mapper:  k8sengine.NewMapper().Mapper,
+				mapper:  k8sengine.NewMapper(k8sengine.LogPersistenceConfig{}).Mapper,
 			}
 
 			// test
@@ -255,7 +255,7 @@ func TestDeletePromConfigMap(t *testing.T) {
 			k8sc := Client{
 				factory: f,
 				helper:  NewHelper(),
-				mapper:  k8sengine.NewMapper().Mapper,
+				mapper:  k8sengine.NewMapper(k8sengine.LogPersistenceConfig{}).Mapper,
 			}
 
 			// test
@@ -322,7 +322,7 @@ func TestDeleteSecret(t *testing.T) {
 			k8sc := Client{
 				factory: f,
 				helper:  NewHelper(),
-				mapper:  k8sengine.NewMapper().Mapper,
+				mapper:  k8sengine.NewMapper(k8sengine.LogPersistenceConfig{}).Mapper,
 			}
 
 			// test
@@ -419,7 +419,7 @@ func TestCreatePromConfigMap(t *testing.T) {
 				return NewClient(Params{
 					Factory: f,
 					Logger:  zaptest.NewLogger(t),
-					Mapper:  k8sengine.NewMapper().Mapper,
+					Mapper:  k8sengine.NewMapper(k8sengine.LogPersistenceConfig{}).Mapper,
 					Helper:  NewHelper(),
 				})
 			},
@@ -440,7 +440,7 @@ func TestCreatePromConfigMap(t *testing.T) {
 					Factory: f,
 					Logger:  zaptest.NewLogger(t),
 					Helper:  NewHelper(),
-					Mapper:  k8sengine.NewMapper().Mapper,
+					Mapper:  k8sengine.NewMapper(k8sengine.LogPersistenceConfig{}).Mapper,
 				})
 			},
 			wantError: true,
@@ -476,7 +476,7 @@ func TestCreatePromConfigMap(t *testing.T) {
 					Factory: f,
 					Logger:  zaptest.NewLogger(t),
 					Helper:  NewHelper(),
-					Mapper:  k8sengine.NewMapper().Mapper,
+					Mapper:  k8sengine.NewMapper(k8sengine.LogPersistenceConfig{}).Mapper,
 				})
 			},
 			// file path will be created as temp file in test body
@@ -570,7 +570,7 @@ func TestCreateSecret(t *testing.T) {
 				factory:         f,
 				helper:          NewHelper(),
 				secretsProvider: provider,
-				mapper:          k8sengine.NewMapper().Mapper,
+				mapper:          k8sengine.NewMapper(k8sengine.LogPersistenceConfig{}).Mapper,
 			}
 
 			// test
@@ -639,7 +639,7 @@ func TestGetJobStatus(t *testing.T) {
 				gctrl := gomock.NewController(t)
 				defer gctrl.Finish()
 				f := computemocks.NewMockFactory(gctrl)
-				cl := &Client{factory: f, helper: NewHelper(), mapper: k8sengine.NewMapper().Mapper, logger: zaptest.NewLogger(t)}
+				cl := &Client{factory: f, helper: NewHelper(), mapper: k8sengine.NewMapper(k8sengine.LogPersistenceConfig{}).Mapper, logger: zaptest.NewLogger(t)}
 				if tt.setupFactory != nil {
 					tt.setupFactory(f)
 				}
@@ -661,7 +661,7 @@ func TestGetJobStatus(t *testing.T) {
 				return
 			}
 			// When not using GetJobStatus path, validate mapper-based conversion on local RayJob object
-			js, err := k8sengine.NewMapper().Mapper.MapLocalJobStatusToGlobal(tt.jobObject)
+			js, err := k8sengine.NewMapper(k8sengine.LogPersistenceConfig{}).Mapper.MapLocalJobStatusToGlobal(tt.jobObject)
 			require.NoError(t, err)
 			if js != nil && js.Ray != nil {
 				assert.Equal(t, tt.expectStatus, js.Ray.JobStatus)
@@ -753,7 +753,7 @@ func TestCreateJob(t *testing.T) {
 			k8sc := Client{
 				factory: f,
 				helper:  NewHelper(),
-				mapper:  k8sengine.NewMapper().Mapper,
+				mapper:  k8sengine.NewMapper(k8sengine.LogPersistenceConfig{}).Mapper,
 			}
 
 			// test
@@ -772,7 +772,7 @@ func TestCreateJob(t *testing.T) {
 func TestGetJobClusterStatus_ClientSetError(t *testing.T) {
 	g := gomock.NewController(t)
 	f := computemocks.NewMockFactory(g)
-	cl := &Client{factory: f, helper: NewHelper(), mapper: k8sengine.NewMapper().Mapper}
+	cl := &Client{factory: f, helper: NewHelper(), mapper: k8sengine.NewMapper(k8sengine.LogPersistenceConfig{}).Mapper}
 	jobCluster := &v2pb.RayCluster{ObjectMeta: metav1.ObjectMeta{Name: "rc1"}}
 	cluster := &v2pb.Cluster{ObjectMeta: metav1.ObjectMeta{Name: "c1"}}
 

go/components/jobs/client/k8sengine/BUILD.bazel

@@ -15,23 +15,29 @@ go_library(
         "//proto/api/v2:go_default_library",
         "@com_github_ray_project_kuberay_ray_operator//apis/ray/v1:go_default_library",
         "@io_k8s_api//core/v1:go_default_library",
+        "@io_k8s_apimachinery//pkg/api/resource:go_default_library",
         "@io_k8s_apimachinery//pkg/apis/meta/v1:go_default_library",
         "@io_k8s_apimachinery//pkg/runtime:go_default_library",
         "@io_k8s_utils//ptr:go_default_library",
+        "@org_uber_go_config//:go_default_library",
         "@org_uber_go_fx//:go_default_library",
     ],
 )
 
 go_test(
     name = "go_default_test",
-    srcs = ["mapper_test.go"],
+    srcs = [
+        "mapper_test.go",
+        "ray_collector_test.go",
+    ],
     embed = [":go_default_library"],
     deps = [
         "//proto/api/v2:go_default_library",
         "@com_github_ray_project_kuberay_ray_operator//apis/ray/v1:go_default_library",
         "@com_github_stretchr_testify//assert:go_default_library",
         "@com_github_stretchr_testify//require:go_default_library",
         "@io_k8s_api//core/v1:go_default_library",
+        "@io_k8s_apimachinery//pkg/api/resource:go_default_library",
         "@io_k8s_apimachinery//pkg/apis/meta/v1:go_default_library",
         "@io_k8s_apimachinery//pkg/runtime:go_default_library",
     ],

go/components/jobs/client/k8sengine/mapper.go

@@ -2,16 +2,20 @@ package k8sengine
 
 import (
 	"fmt"
+	"strings"
 
 	"github.com/michelangelo-ai/michelangelo/go/components/jobs/common/types"
 	v2pb "github.com/michelangelo-ai/michelangelo/proto-go/api/v2"
 	rayv1 "github.com/ray-project/kuberay/ray-operator/apis/ray/v1"
+	"go.uber.org/config"
 	"go.uber.org/fx"
 	"k8s.io/apimachinery/pkg/runtime"
 )
 
 // Mapper helps to map global to local crds and vice versa
-type Mapper struct{}
+type Mapper struct {
+	LogPersistence LogPersistenceConfig
+}
 
 // MapperResult has Mapper result
 type MapperResult struct {
@@ -22,10 +26,23 @@ type MapperResult struct {
 
 const _mapperName = "k8sengineMapper"
 
+const logPersistenceConfigKey = "controllers.rayCluster.logPersistence"
+
+// NewLogPersistenceConfig loads LogPersistenceConfig from YAML config provider.
+func NewLogPersistenceConfig(provider config.Provider) (LogPersistenceConfig, error) {
+	conf := LogPersistenceConfig{}
+	err := provider.Get(logPersistenceConfigKey).Populate(&conf)
+	if err != nil {
+		// Config is optional — return zero-value (disabled) if not present
+		return LogPersistenceConfig{}, nil
+	}
+	return conf, nil
+}
+
 // NewMapper constructs the Mapper
-func NewMapper() MapperResult {
+func NewMapper(logPersistence LogPersistenceConfig) MapperResult {
 	return MapperResult{
-		Mapper: Mapper{},
+		Mapper: Mapper{LogPersistence: logPersistence},
 	}
 }
 
@@ -86,6 +103,12 @@ func (m Mapper) GetLocalName(obj runtime.Object) (namespace, name string) {
 
 // MapLocalClusterStatusToGlobal converts a local (Kubernetes) cluster status object
 // to the global Michelangelo ClusterStatus representation.
+//
+// Also computes log_url here (vs in the controller) because the mapper owns
+// every piece of knowledge required: the LogPersistenceConfig, the local
+// cluster name (off the local object), and the compute-cluster Ray namespace
+// (RayLocalNamespace). The controller just copies clusterStatus.Ray.LogUrl
+// onto its RayClusterStatus during applyRayClusterStatus.
 func (m Mapper) MapLocalClusterStatusToGlobal(localClusterObject runtime.Object) (*types.JobClusterStatus, error) {
 	if localClusterObject == nil {
 		return nil, fmt.Errorf("localClusterObject cannot be nil")
@@ -94,6 +117,7 @@ func (m Mapper) MapLocalClusterStatusToGlobal(localClusterObject runtime.Object)
 	switch obj := localClusterObject.(type) {
 	case *rayv1.RayCluster:
 		v2Status := convertRayV1ClusterStatusToV2(obj)
+		v2Status.LogUrl = m.buildLogURL(obj.GetName())
 		reason := obj.Status.Reason
 		return &types.JobClusterStatus{
 			Ray:    v2Status,
@@ -104,6 +128,29 @@ func (m Mapper) MapLocalClusterStatusToGlobal(localClusterObject runtime.Object)
 	}
 }
 
+// buildLogURL returns the human-browsable log URL for a Ray cluster, or "" when
+// log persistence is disabled / no base URL is configured. The URL points at
+// where the collector sidecar wrote the cluster's logs in object storage.
+//
+//	{LogUrlBaseUrl}/{Bucket}/{PathPrefix}/{cluster}_{rayLocalNamespace}/
+//
+// rayLocalNamespace is RayLocalNamespace ("default") — the namespace where Ray
+// pods actually run on the compute cluster, not the control-plane namespace
+// of the v2 RayCluster CR. The bucket layout is created by the collector and
+// keyed off the compute-cluster namespace, so the browse URL must match.
+func (m Mapper) buildLogURL(clusterName string) string {
+	if !m.LogPersistence.Enabled || m.LogPersistence.LogUrlBaseUrl == "" {
+		return ""
+	}
+	return fmt.Sprintf("%s/%s/%s/%s_%s/",
+		strings.TrimRight(m.LogPersistence.LogUrlBaseUrl, "/"),
+		m.LogPersistence.Bucket,
+		strings.Trim(m.LogPersistence.PathPrefix, "/"),
+		clusterName,
+		RayLocalNamespace,
+	)
+}
+
 // MapLocalJobStatusToGlobal converts a local (Kubernetes) job status object
 // to the global Michelangelo JobStatus representation.
 func (m Mapper) MapLocalJobStatusToGlobal(localJobObject runtime.Object) (*types.JobStatus, error) {