diff --git a/flyteplugins/go/tasks/plugins/k8s/ray/ray.go b/flyteplugins/go/tasks/plugins/k8s/ray/ray.go index bc4d22bd02..7dfbb5850d 100644 --- a/flyteplugins/go/tasks/plugins/k8s/ray/ray.go +++ b/flyteplugins/go/tasks/plugins/k8s/ray/ray.go @@ -371,6 +371,22 @@ func injectLogsSidecar(primaryContainer *v1.Container, podSpec *v1.PodSpec) { podSpec.Containers = append(podSpec.Containers, *sidecar) } +// logNoiseDisablingEnvVars turns off Ray's terminal-oriented log decorations (ANSI-colored log +// prefixes and Ray Data progress bars), which are unreadable once collected from a non-interactive +// pod. Shared by the head and worker builders to keep the two in sync. +func logNoiseDisablingEnvVars() []v1.EnvVar { + return []v1.EnvVar{ + { + Name: "RAY_COLOR_PREFIX", + Value: "0", + }, + { + Name: "RAY_DATA_DISABLE_PROGRESS_BARS", + Value: "1", + }, + } +} + func buildHeadPodTemplate(primaryContainer *v1.Container, basePodSpec *v1.PodSpec, objectMeta *metav1.ObjectMeta, taskCtx pluginsCore.TaskExecutionContext, spec *plugins.HeadGroupSpec) (v1.PodTemplateSpec, error) { // Some configs are copy from https://github.com/ray-project/kuberay/blob/b72e6bdcd9b8c77a9dc6b5da8560910f3a0c3ffd/apiserver/pkg/util/cluster.go#L97 // They should always be the same, so we could hard code here. @@ -386,6 +402,7 @@ func buildHeadPodTemplate(primaryContainer *v1.Container, basePodSpec *v1.PodSpe }, }, } + envs = append(envs, logNoiseDisablingEnvVars()...) primaryContainer.Args = []string{} @@ -512,6 +529,7 @@ func buildWorkerPodTemplate(primaryContainer *v1.Container, basePodSpec *v1.PodS }, }, } + envs = append(envs, logNoiseDisablingEnvVars()...) primaryContainer.Env = append(primaryContainer.Env, envs...) diff --git a/flyteplugins/go/tasks/plugins/k8s/ray/ray_test.go b/flyteplugins/go/tasks/plugins/k8s/ray/ray_test.go index ce0b0c9b28..c8ab6e3815 100644 --- a/flyteplugins/go/tasks/plugins/k8s/ray/ray_test.go +++ b/flyteplugins/go/tasks/plugins/k8s/ray/ray_test.go @@ -350,6 +350,56 @@ func TestBuildResourceRayEntrypointPreservesEmptyArgs(t *testing.T) { assert.Contains(t, rayJobObj.Spec.Entrypoint, "vars '' resolver") } +func TestBuildResourceRayDisablesLogNoise(t *testing.T) { + rayJobResourceHandler := rayJobResourceHandler{} + assert.NoError(t, config.SetK8sPluginConfig(&config.K8sPluginConfig{})) + + taskTemplate := dummyRayTaskTemplate("ray-id", dummyRayCustomObj()) + rayCtx := dummyRayTaskContext(taskTemplate, resourceRequirements, nil, "", serviceAccount) + r, err := rayJobResourceHandler.BuildResource(context.TODO(), rayCtx) + assert.Nil(t, err) + require.NotNil(t, r) + + rayJob, ok := r.(*rayv1.RayJob) + require.True(t, ok) + require.NotEmpty(t, rayJob.Spec.RayClusterSpec.WorkerGroupSpecs) + + // Select by container name: the vars belong on the Ray container, not on an injected sidecar. + envByContainer := func(containers []corev1.Container, name string) map[string]string { + for _, cnt := range containers { + if cnt.Name != name { + continue + } + env := make(map[string]string, len(cnt.Env)) + for _, e := range cnt.Env { + env[e.Name] = e.Value + } + return env + } + return nil + } + + headEnv := envByContainer(rayJob.Spec.RayClusterSpec.HeadGroupSpec.Template.Spec.Containers, "ray-head") + workerEnv := envByContainer(rayJob.Spec.RayClusterSpec.WorkerGroupSpecs[0].Template.Spec.Containers, "ray-worker") + require.NotNil(t, headEnv) + require.NotNil(t, workerEnv) + + fixtures := []struct { + name string + value string + }{ + {name: "RAY_COLOR_PREFIX", value: "0"}, + {name: "RAY_DATA_DISABLE_PROGRESS_BARS", value: "1"}, + } + + for _, f := range fixtures { + t.Run(f.name, func(t *testing.T) { + assert.Equal(t, f.value, headEnv[f.name], "head container") + assert.Equal(t, f.value, workerEnv[f.name], "worker container") + }) + } +} + func TestBuildPodTemplate(t *testing.T) { taskTemplate := dummyRayTaskTemplate("id", dummyRayCustomObj()) resources := &corev1.ResourceRequirements{