[Feature] Add cleanup for terminated RayJob/RayCluster metrics

Author: phantom5125

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

@@ -89,6 +89,12 @@ func (r *RayClusterMetricsManager) ObserveRayClusterProvisionedDuration(name, na
 	r.rayClusterProvisionedDurationSeconds.WithLabelValues(name, namespace).Set(duration)
 }
 
+// DeleteRayClusterMetrics removes metrics that belongs to the specified RayCluster.
+func (r *RayClusterMetricsManager) DeleteRayClusterMetrics(name, namespace string) {
+	numCleanedUpMetrics := r.rayClusterProvisionedDurationSeconds.DeletePartialMatch(prometheus.Labels{"name": name, "namespace": namespace})
+	r.log.Info("Cleaned up expired RayCluster metric", "name", name, "namespace", namespace, "numCleanedUpMetrics", numCleanedUpMetrics)
+}
+
 func (r *RayClusterMetricsManager) collectRayClusterInfo(cluster *rayv1.RayCluster, ch chan<- prometheus.Metric) {
 	ownerKind := "None"
 	if v, ok := cluster.Labels[utils.RayOriginatedFromCRDLabelKey]; ok {

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

@@ -84,6 +84,12 @@ func (r *RayJobMetricsManager) ObserveRayJobExecutionDuration(name, namespace st
 	r.rayJobExecutionDurationSeconds.WithLabelValues(name, namespace, string(jobDeploymentStatus), strconv.Itoa(retryCount)).Set(duration)
 }
 
+// deleteRayJobMetrics removes metrics that belongs to the specified RayJob.
+func (r *RayJobMetricsManager) DeleteRayJobMetrics(name, namespace string) {
+	numCleanedUpMetrics := r.rayJobExecutionDurationSeconds.DeletePartialMatch(prometheus.Labels{"name": name, "namespace": namespace})
+	r.log.Info("Cleaned up expired rayJob metric", "name", name, "namespace", namespace, "numCleanedUpMetrics", numCleanedUpMetrics)
+}
+
 func (r *RayJobMetricsManager) collectRayJobInfo(rayJob *rayv1.RayJob, ch chan<- prometheus.Metric) {
 	ch <- prometheus.MustNewConstMetric(
 		r.rayJobInfo,

ray-operator/controllers/ray/raycluster_controller.go

@@ -135,6 +135,7 @@ 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)
+		r.options.RayClusterMetricsManager.DeleteRayClusterMetrics(request.Name, request.Namespace)
 	} else {
 		logger.Error(err, "Read request instance error!")
 	}

ray-operator/controllers/ray/rayjob_controller.go

@@ -91,7 +91,8 @@ func (r *RayJobReconciler) Reconcile(ctx context.Context, request ctrl.Request)
 	if err := r.Get(ctx, request.NamespacedName, rayJobInstance); err != nil {
 		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")
+			logger.Info("RayJob resource not found.")
+			r.options.RayJobMetricsManager.DeleteRayJobMetrics(request.Name, request.Namespace)
 			return ctrl.Result{}, nil
 		}
 		// Error reading the object - requeue the request.