Add config in main() & fix the entrance to schedule cleanup

Author: phantom5125

ray-operator/controllers/ray/metrics/ray_cluster_metrics.go

@@ -43,7 +43,7 @@ type RayClusterMetricsManager struct {
 }
 
 // NewRayClusterMetricsManager creates a new RayClusterMetricsManager instance.
-func NewRayClusterMetricsManager(ctx context.Context, client client.Client) *RayClusterMetricsManager {
+func NewRayClusterMetricsManager(ctx context.Context, client client.Client, metricsTTLSeconds int) *RayClusterMetricsManager {
 	manager := &RayClusterMetricsManager{
 		rayClusterProvisionedDurationSeconds: prometheus.NewGaugeVec(
 			prometheus.GaugeOpts{
@@ -73,7 +73,7 @@ func NewRayClusterMetricsManager(ctx context.Context, client client.Client) *Ray
 		client:       client,
 		log:          ctrl.LoggerFrom(ctx),
 		cleanupQueue: make([]RayClusterMetricCleanupItem, 0),
-		metricTTL:    5 * time.Minute, // Keep metrics for 5 minutes
+		metricTTL:    time.Duration(metricsTTLSeconds) * time.Second,
 	}
 
 	// Start the cleanup goroutine

ray-operator/controllers/ray/metrics/ray_job_metrics.go

@@ -20,7 +20,7 @@ type RayJobMetricsObserver interface {
 }
 
 // RayJobMetricCleanupItem represents an item in the RayJob metric cleanup queue
- type RayJobMetricCleanupItem struct {
+type RayJobMetricCleanupItem struct {
 	Name      string
 	Namespace string
 	DeleteAt  time.Time
@@ -41,7 +41,7 @@ type RayJobMetricsManager struct {
 }
 
 // NewRayJobMetricsManager creates a new RayJobMetricsManager instance.
-func NewRayJobMetricsManager(ctx context.Context, client client.Client) *RayJobMetricsManager {
+func NewRayJobMetricsManager(ctx context.Context, client client.Client, metricsTTLSeconds int) *RayJobMetricsManager {
 	manager := &RayJobMetricsManager{
 		rayJobExecutionDurationSeconds: prometheus.NewGaugeVec(
 			prometheus.GaugeOpts{
@@ -67,9 +67,9 @@ func NewRayJobMetricsManager(ctx context.Context, client client.Client) *RayJobM
 		client:       client,
 		log:          ctrl.LoggerFrom(ctx),
 		cleanupQueue: make([]RayJobMetricCleanupItem, 0),
-		metricTTL:    5 * time.Minute, // Keep metrics for 5 minutes
+		metricTTL:    time.Duration(metricsTTLSeconds) * time.Second,
 	}
-	
+
 	// Start the cleanup goroutine
 	go manager.startRayJobCleanupLoop(ctx)
 	return manager

ray-operator/controllers/ray/raycluster_controller.go

@@ -135,6 +135,9 @@ func (r *RayClusterReconciler) Reconcile(ctx context.Context, request ctrl.Reque
 	if errors.IsNotFound(err) {
 		// Clear all related expectations
 		r.rayClusterScaleExpectation.Delete(instance.Name, instance.Namespace)
+		if r.options.RayClusterMetricsManager != nil {
+			r.options.RayClusterMetricsManager.ScheduleRayClusterMetricForCleanup(instance.Name, instance.Namespace)
+		}
 	} else {
 		logger.Error(err, "Read request instance error!")
 	}
@@ -195,13 +198,6 @@ func (r *RayClusterReconciler) rayClusterReconcile(ctx context.Context, instance
 	// Please do NOT modify `originalRayClusterInstance` in the following code.
 	originalRayClusterInstance := instance.DeepCopy()
 
-	if instance.DeletionTimestamp != nil && !instance.DeletionTimestamp.IsZero() {
-		logger.Info("RayCluster is being deleted, scheduling metric for cleanup")
-		if r.options.RayClusterMetricsManager != nil {
-			r.options.RayClusterMetricsManager.ScheduleRayClusterMetricForCleanup(instance.Name, instance.Namespace)
-		}
-	}
-
 	// The `enableGCSFTRedisCleanup` is a feature flag introduced in KubeRay v1.0.0. It determines whether
 	// the Redis cleanup job should be activated. Users can disable the feature by setting the environment
 	// variable `ENABLE_GCS_FT_REDIS_CLEANUP` to `false`, and undertake the Redis storage namespace cleanup

ray-operator/controllers/ray/rayjob_controller.go

@@ -92,6 +92,10 @@ func (r *RayJobReconciler) Reconcile(ctx context.Context, request ctrl.Request)
 		if errors.IsNotFound(err) {
 			// Request object not found, could have been deleted after reconcile request. Stop reconciliation.
 			logger.Info("RayJob resource not found. Ignoring since object must be deleted")
+			// Schedule metrics cleanup when job reaches terminal status
+			if r.options.RayJobMetricsManager != nil {
+				r.options.RayJobMetricsManager.ScheduleRayJobMetricForCleanup(rayJobInstance.Name, rayJobInstance.Namespace)
+			}
 			return ctrl.Result{}, nil
 		}
 		// Error reading the object - requeue the request.
@@ -465,10 +469,6 @@ func (r *RayJobReconciler) Reconcile(ctx context.Context, request ctrl.Request)
 		return ctrl.Result{RequeueAfter: RayJobDefaultRequeueDuration}, err
 	}
 	emitRayJobMetrics(r.options.RayJobMetricsManager, rayJobInstance.Name, rayJobInstance.Namespace, originalRayJobInstance.Status, rayJobInstance.Status)
-	// Schedule metrics cleanup when job reaches terminal status
-	if !rayv1.IsJobDeploymentTerminal(originalRayJobInstance.Status.JobDeploymentStatus) && rayv1.IsJobDeploymentTerminal(rayJobInstance.Status.JobDeploymentStatus) {
-		r.options.RayJobMetricsManager.ScheduleRayJobMetricForCleanup(rayJobInstance.Name, rayJobInstance.Namespace)
-	}
 	return ctrl.Result{RequeueAfter: RayJobDefaultRequeueDuration}, nil
 }
 

ray-operator/main.go

@@ -69,6 +69,7 @@ func main() {
 	var enableBatchScheduler bool
 	var batchScheduler string
 	var enableMetrics bool
+	var metricsTTLSeconds int
 
 	// TODO: remove flag-based config once Configuration API graduates to v1.
 	flag.StringVar(&metricsAddr, "metrics-addr", configapi.DefaultMetricsAddr, "The address the metric endpoint binds to.")
@@ -100,6 +101,7 @@ func main() {
 		"Use Kubernetes proxy subresource when connecting to the Ray Head node.")
 	flag.StringVar(&featureGates, "feature-gates", "", "A set of key=value pairs that describe feature gates. E.g. FeatureOne=true,FeatureTwo=false,...")
 	flag.BoolVar(&enableMetrics, "enable-metrics", false, "Enable the emission of control plane metrics.")
+	flag.IntVar(&metricsTTLSeconds, "metrics-ttl-seconds", 300, "The time to live for metrics cleanup in seconds. Default is 300 seconds (5 minutes).")
 
 	opts := k8szap.Options{
 		TimeEncoder: zapcore.ISO8601TimeEncoder,
@@ -237,8 +239,8 @@ func main() {
 	var rayServiceMetricsManager *metrics.RayServiceMetricsManager
 	if config.EnableMetrics {
 		mgrClient := mgr.GetClient()
-		rayClusterMetricsManager = metrics.NewRayClusterMetricsManager(ctx, mgrClient)
-		rayJobMetricsManager = metrics.NewRayJobMetricsManager(ctx, mgrClient)
+		rayClusterMetricsManager = metrics.NewRayClusterMetricsManager(ctx, mgrClient, metricsTTLSeconds)
+		rayJobMetricsManager = metrics.NewRayJobMetricsManager(ctx, mgrClient, metricsTTLSeconds)
 		rayServiceMetricsManager = metrics.NewRayServiceMetricsManager(ctx, mgrClient)
 		ctrlmetrics.Registry.MustRegister(
 			rayClusterMetricsManager,