Add log persistence proto field and collector sidecar injection

- Add string log_url = 10 to RayClusterStatus proto for reporting
  the S3 path where persisted logs are stored
- Add LogPersistenceConfig to Mapper struct, loaded via FX from
  mapper.logPersistence YAML config
- Implement InjectCollectorSidecar() that injects KubeRay History
  Server collector sidecar into Ray pod templates (head + workers)
- Wire sidecar injection into mapRayCluster() when config is enabled
- Update NewMapper() to accept LogPersistenceConfig parameter

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

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/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 = "mapper.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},
 	}
 }
 

go/components/jobs/client/k8sengine/ray.go

@@ -6,11 +6,24 @@ import (
 	v2pb "github.com/michelangelo-ai/michelangelo/proto-go/api/v2"
 	rayv1 "github.com/ray-project/kuberay/ray-operator/apis/ray/v1"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	k8sptr "k8s.io/utils/ptr"
 )
 
+// LogPersistenceConfig holds platform-level configuration for log persistence.
+// This is loaded from YAML config at the mapper level, not from per-job proto.
+type LogPersistenceConfig struct {
+	Enabled           bool   `yaml:"enabled"`
+	StorageEndpoint   string `yaml:"storageEndpoint"`
+	Bucket            string `yaml:"bucket"`
+	PathPrefix        string `yaml:"pathPrefix"`
+	Region            string `yaml:"region"`
+	CredentialsSecret string `yaml:"credentialsSecret"`
+	CollectorImage    string `yaml:"collectorImage"`
+}
+
 func (m Mapper) mapRay(rayJob *v2pb.RayJob, jobClusterObject runtime.Object, cluster *v2pb.Cluster) (runtime.Object, error) {
 	if jobClusterObject == nil {
 		return nil, fmt.Errorf("ray job requires associated RayCluster object")
@@ -52,6 +65,14 @@ func (m Mapper) mapRay(rayJob *v2pb.RayJob, jobClusterObject runtime.Object, clu
 func (m Mapper) mapRayCluster(rayCluster *v2pb.RayCluster) (runtime.Object, error) {
 	workerGroupSpecs := getWorkerGroupSpecs(rayCluster.GetName(), rayCluster.GetSpec().Workers)
 	headGroupSpec := getHeadGroupSpec(rayCluster.GetSpec().Head)
+
+	if m.LogPersistence.Enabled {
+		InjectCollectorSidecar(&headGroupSpec.Template, m.LogPersistence, rayCluster.GetName(), "Head")
+		for i := range workerGroupSpecs {
+			InjectCollectorSidecar(&workerGroupSpecs[i].Template, m.LogPersistence, rayCluster.GetName(), "Worker")
+		}
+	}
+
 	rayV1Cluster := &rayv1.RayCluster{
 		TypeMeta: metav1.TypeMeta{
 			Kind:       RayClusterKind,
@@ -94,6 +115,124 @@ func getWorkerGroupSpecs(clusterName string, workers []*v2pb.RayWorkerSpec) []ra
 	return workerGroupSpecsJSON
 }
 
+const (
+	rayLogsVolumeName = "ray-logs"
+	rayLogsPath       = "/tmp/ray"
+	collectorPort     = 8084
+)
+
+// InjectCollectorSidecar injects a KubeRay History Server collector sidecar container
+// into the pod template. It adds a shared emptyDir volume for Ray logs, mounts it to
+// all existing containers, configures Ray event export env vars, adds a lifecycle hook
+// to write the node ID, and appends the collector sidecar container.
+func InjectCollectorSidecar(podTemplate *corev1.PodTemplateSpec, config LogPersistenceConfig, clusterName string, role string) {
+	// 1. Add ray-logs emptyDir volume
+	podTemplate.Spec.Volumes = append(podTemplate.Spec.Volumes, corev1.Volume{
+		Name: rayLogsVolumeName,
+		VolumeSource: corev1.VolumeSource{
+			EmptyDir: &corev1.EmptyDirVolumeSource{},
+		},
+	})
+
+	rayLogsVolumeMount := corev1.VolumeMount{
+		Name:      rayLogsVolumeName,
+		MountPath: rayLogsPath,
+	}
+
+	eventExportEnvVars := []corev1.EnvVar{
+		{
+			Name:  "RAY_enable_core_worker_ray_event_to_aggregator",
+			Value: "1",
+		},
+		{
+			Name:  "RAY_DASHBOARD_AGGREGATOR_AGENT_EVENTS_EXPORT_ADDR",
+			Value: fmt.Sprintf("http://localhost:%d/events", collectorPort),
+		},
+	}
+
+	// 2-3, 5. Update all existing containers: add volume mount, env vars, lifecycle hook
+	for i := range podTemplate.Spec.Containers {
+		c := &podTemplate.Spec.Containers[i]
+		c.VolumeMounts = append(c.VolumeMounts, rayLogsVolumeMount)
+		c.Env = append(c.Env, eventExportEnvVars...)
+
+		// Add postStart lifecycle hook to write node ID
+		if c.Lifecycle == nil {
+			c.Lifecycle = &corev1.Lifecycle{}
+		}
+		c.Lifecycle.PostStart = &corev1.LifecycleHandler{
+			Exec: &corev1.ExecAction{
+				Command: []string{"/bin/sh", "-c", "echo $RAY_NODE_ID > /tmp/ray/init.log"},
+			},
+		}
+	}
+
+	// 4. Add collector sidecar container
+	collectorContainer := corev1.Container{
+		Name:  "collector",
+		Image: config.CollectorImage,
+		Args: []string{
+			fmt.Sprintf("--role=%s", role),
+			"--runtime-class-name=s3",
+			fmt.Sprintf("--ray-cluster-name=%s", clusterName),
+			fmt.Sprintf("--ray-root-dir=%s", rayLogsPath),
+			fmt.Sprintf("--events-port=%d", collectorPort),
+		},
+		Env: []corev1.EnvVar{
+			{
+				Name: "AWS_ACCESS_KEY_ID",
+				ValueFrom: &corev1.EnvVarSource{
+					SecretKeyRef: &corev1.SecretKeySelector{
+						LocalObjectReference: corev1.LocalObjectReference{
+							Name: config.CredentialsSecret,
+						},
+						Key: "AWS_ACCESS_KEY_ID",
+					},
+				},
+			},
+			{
+				Name: "AWS_SECRET_ACCESS_KEY",
+				ValueFrom: &corev1.EnvVarSource{
+					SecretKeyRef: &corev1.SecretKeySelector{
+						LocalObjectReference: corev1.LocalObjectReference{
+							Name: config.CredentialsSecret,
+						},
+						Key: "AWS_SECRET_ACCESS_KEY",
+					},
+				},
+			},
+			{
+				Name:  "S3_ENDPOINT_URL",
+				Value: config.StorageEndpoint,
+			},
+			{
+				Name:  "S3_BUCKET",
+				Value: config.Bucket,
+			},
+			{
+				Name:  "S3_REGION",
+				Value: config.Region,
+			},
+		},
+		Ports: []corev1.ContainerPort{
+			{
+				Name:          "events",
+				ContainerPort: int32(collectorPort),
+				Protocol:      corev1.ProtocolTCP,
+			},
+		},
+		Resources: corev1.ResourceRequirements{
+			Requests: corev1.ResourceList{
+				corev1.ResourceCPU:    resource.MustParse("100m"),
+				corev1.ResourceMemory: resource.MustParse("128Mi"),
+			},
+		},
+		VolumeMounts: []corev1.VolumeMount{rayLogsVolumeMount},
+	}
+
+	podTemplate.Spec.Containers = append(podTemplate.Spec.Containers, collectorContainer)
+}
+
 // getRayClusterStateFromStatus maps KubeRay v1 cluster state to our internal v2pb.RayClusterState
 func getRayClusterStateFromKubeRayState(kubeRayState rayv1.ClusterState) v2pb.RayClusterState {
 	switch kubeRayState {