feat: add LogPersistenceSpec proto and KubeRay collector sidecar injection

Proto:
- Add LogPersistenceSpec message (storage backend, bucket, path prefix,
  region, credentials secret, collector image)
- Add log_persistence to RayClusterSpec
- Add log_url to RayClusterStatus

Mapper:
- Load LogPersistenceConfig from controllers.rayCluster.logPersistence
  YAML config block (FX-injected)
- Implement InjectCollectorSidecar() that injects the KubeRay History
  Server collector sidecar into Ray pod templates (head + workers) with
  the env-var S3 config pattern (no ConfigMap mount)
- Reuse an existing /tmp/ray volume when present so the collector and
  Ray container share log data
- Set ImagePullPolicy: IfNotPresent on the collector container
- Add lifecycle PostStart hook to extract raylet node ID
- Wire sidecar injection into mapRayCluster() when config is enabled

Tests:
- Cover env-var pattern (no ConfigMap mount), head vs worker env-var
  differences, port and resources, lifecycle preservation

Author: dragod812

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

@@ -6,12 +6,15 @@ import (
 	"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 +25,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},
 	}
 }