Refactor rayCluster metrics

Signed-off-by: win5923 <ken89@kimo.com>

Author: win5923

ray-operator/controllers/ray/common/metrics_test.go

@@ -4,14 +4,16 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 )
 
-type RayClusterMetricCollector struct {
+// RayClusterCollector implements the prometheus.Collector and ray.RayClusterMetricsCollector interface to collect ray cluster metrics.
+type RayClusterCollector struct {
 	// Metrics
 	rayClusterProvisionedDuration  *prometheus.GaugeVec
 	rayClusterHeadPodReadyDuration *prometheus.GaugeVec
 }
 
-func NewRayClusterMetricCollector() *RayClusterMetricCollector {
-	collector := &RayClusterMetricCollector{
+// NewRayClusterCollector creates a new RayClusterCollector instance.
+func NewRayClusterCollector() *RayClusterCollector {
+	collector := &RayClusterCollector{
 		rayClusterProvisionedDuration: prometheus.NewGaugeVec(
 			prometheus.GaugeOpts{
 				Name: "kuberay_cluster_provisioned_duration_seconds",
@@ -31,23 +33,35 @@ func NewRayClusterMetricCollector() *RayClusterMetricCollector {
 }
 
 // Describe implements prometheus.Collector interface Describe method.
-func (c *RayClusterMetricCollector) Describe(ch chan<- *prometheus.Desc) {
+func (c *RayClusterCollector) Describe(ch chan<- *prometheus.Desc) {
 	c.rayClusterProvisionedDuration.Describe(ch)
 	c.rayClusterHeadPodReadyDuration.Describe(ch)
 }
 
 // Collect implements prometheus.Collector interface Collect method.
-func (c *RayClusterMetricCollector) Collect(ch chan<- prometheus.Metric) {
+func (c *RayClusterCollector) Collect(ch chan<- prometheus.Metric) {
 	c.rayClusterProvisionedDuration.Collect(ch)
 	c.rayClusterHeadPodReadyDuration.Collect(ch)
 }
 
-// ObserveRayClusterProvisionedDuration observes the duration of RayCluster from creation to provisioned
-func (c *RayClusterMetricCollector) ObserveRayClusterProvisionedDuration(name, namespace string, duration float64) {
+// ObserveRayClusterProvisionedDuration observes the duration of RayCluster's status transition from false (or unset) to true
+func (c *RayClusterCollector) ObserveRayClusterProvisionedDuration(name, namespace string, duration float64) {
 	c.rayClusterProvisionedDuration.WithLabelValues(name, namespace).Set(duration)
 }
 
 // ObserveRayClusterHeadPodReadyDuration observes the duration of RayCluster from creation to head pod ready
-func (c *RayClusterMetricCollector) ObserveRayClusterHeadPodReadyDuration(name, namespace string, duration float64) {
+func (c *RayClusterCollector) ObserveRayClusterHeadPodReadyDuration(name, namespace string, duration float64) {
 	c.rayClusterHeadPodReadyDuration.WithLabelValues(name, namespace).Set(duration)
 }
+
+type RayClusterNoopCollector struct{}
+
+func NewRayClusterNoopCollector() *RayClusterNoopCollector {
+	return &RayClusterNoopCollector{}
+}
+
+func (c *RayClusterNoopCollector) ObserveRayClusterProvisionedDuration(_ string, _ string, _ float64) {
+}
+
+func (c *RayClusterNoopCollector) ObserveRayClusterHeadPodReadyDuration(_ string, _ string, _ float64) {
+}

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

@@ -11,25 +11,6 @@ import (
 	"strings"
 	"time"
 
-	"k8s.io/apimachinery/pkg/api/meta"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/utils/ptr"
-
-	configapi "github.com/ray-project/kuberay/ray-operator/apis/config/v1alpha1"
-	"github.com/ray-project/kuberay/ray-operator/controllers/ray/batchscheduler"
-	"github.com/ray-project/kuberay/ray-operator/controllers/ray/common"
-	"github.com/ray-project/kuberay/ray-operator/controllers/ray/expectations"
-	"github.com/ray-project/kuberay/ray-operator/controllers/ray/metrics"
-	"github.com/ray-project/kuberay/ray-operator/controllers/ray/utils"
-	"github.com/ray-project/kuberay/ray-operator/pkg/features"
-
-	batchv1 "k8s.io/api/batch/v1"
-	rbacv1 "k8s.io/api/rbac/v1"
-
-	"k8s.io/client-go/tools/record"
-
-	rayv1 "github.com/ray-project/kuberay/ray-operator/apis/ray/v1"
-
 	"github.com/go-logr/logr"
 	routev1 "github.com/openshift/api/route/v1"
 	batchv1 "k8s.io/api/batch/v1"
@@ -57,7 +38,6 @@ import (
 	"github.com/ray-project/kuberay/ray-operator/controllers/ray/batchscheduler"
 	"github.com/ray-project/kuberay/ray-operator/controllers/ray/common"
 	"github.com/ray-project/kuberay/ray-operator/controllers/ray/expectations"
-	"github.com/ray-project/kuberay/ray-operator/controllers/ray/metrics"
 	"github.com/ray-project/kuberay/ray-operator/controllers/ray/utils"
 	"github.com/ray-project/kuberay/ray-operator/pkg/features"
 )
@@ -110,8 +90,13 @@ type RayClusterReconciler struct {
 	options                    RayClusterReconcilerOptions
 }
 
+type RayClusterMetricsCollector interface {
+	ObserveRayClusterProvisionedDuration(name, namespace string, duration float64)
+	ObserveRayClusterHeadPodReadyDuration(name, namespace string, duration float64)
+}
+
 type RayClusterReconcilerOptions struct {
-	RayClusterMetricCollector *metrics.RayClusterMetricCollector
+	RayClusterMetricCollector RayClusterMetricsCollector
 	HeadSidecarContainers     []corev1.Container
 	WorkerSidecarContainers   []corev1.Container
 	IsOpenShift               bool
@@ -348,6 +333,8 @@ func (r *RayClusterReconciler) rayClusterReconcile(ctx context.Context, instance
 		inconsistent, updateErr = r.updateRayClusterStatus(ctx, originalRayClusterInstance, newInstance)
 	}
 
+	emitRayClusterMetrics(r.options.RayClusterMetricCollector, newInstance.Name, newInstance.Namespace, originalRayClusterInstance.Status, newInstance.Status, newInstance.CreationTimestamp.Time)
+
 	// Return error based on order.
 	var err error
 	if reconcileErr != nil {
@@ -1288,16 +1275,6 @@ func (r *RayClusterReconciler) calculateStatus(ctx context.Context, instance *ra
 			})
 		} else {
 			headPodReadyCondition := utils.FindHeadPodReadyCondition(headPod)
-
-			// Record ray_cluster_head_pod_ready_duration_seconds metric
-			// Calculate the time from RayClusters created to head pod ready
-			// if headPodReadyCondition.Status == metav1.ConditionTrue {
-			// 	if !meta.IsStatusConditionTrue(newInstance.Status.Conditions, string(rayv1.HeadPodReady)) {
-			// 		readyDuration := time.Since(instance.CreationTimestamp.Time)
-			// 		common.ObserveRayClusterHeadPodReadyDuration(instance.Namespace, readyDuration)
-			// 	}
-			// }
-
 			meta.SetStatusCondition(&newInstance.Status.Conditions, headPodReadyCondition)
 		}
 
@@ -1637,21 +1614,28 @@ func (r *RayClusterReconciler) updateRayClusterStatus(ctx context.Context, origi
 	if err != nil {
 		logger.Info("Error updating status", "name", originalRayClusterInstance.Name, "error", err, "RayCluster", newInstance)
 	}
-	collectRayClusterMetrics(r.options.RayClusterMetricCollector, newInstance.Name, newInstance.Namespace, newInstance.Status, originalRayClusterInstance.Status, originalRayClusterInstance.CreationTimestamp.Time)
 
 	return inconsistent, err
 }
 
-// collectRayClusterMetrics records metrics related to the RayCluster.
-func collectRayClusterMetrics(collector *metrics.RayClusterMetricCollector, name, namespace string, newClusterStatus, oldClusterStatus rayv1.RayClusterStatus, creationTimestamp time.Time) {
-	if collector == nil {
-		return
+func emitRayClusterMetrics(rayClusterMetricsCollector RayClusterMetricsCollector, rayClusterName, rayClusterNamespace string, originRayClusterStatus, rayClusterStatus rayv1.RayClusterStatus, creationTimestamp time.Time) {
+	emitRayClusterProvisionedDuration(rayClusterMetricsCollector, rayClusterName, rayClusterNamespace, originRayClusterStatus, rayClusterStatus, creationTimestamp)
+	emitRayClusterHeadPodReadyDuration(rayClusterMetricsCollector, rayClusterName, rayClusterNamespace, originRayClusterStatus, rayClusterStatus, creationTimestamp)
+}
+
+func emitRayClusterProvisionedDuration(collector RayClusterMetricsCollector, rayClusterName, rayClusterNamespace string, originRayClusterStatus, rayClusterStatus rayv1.RayClusterStatus, creationTimestamp time.Time) {
+	// Emit kuberay_cluster_provisioned_duration_seconds when a RayCluster's RayClusterProvisioned status transitions from false (or unset) to true
+	if !meta.IsStatusConditionTrue(originRayClusterStatus.Conditions, string(rayv1.RayClusterProvisioned)) &&
+		meta.IsStatusConditionTrue(rayClusterStatus.Conditions, string(rayv1.RayClusterProvisioned)) {
+		collector.ObserveRayClusterProvisionedDuration(rayClusterName, rayClusterNamespace, time.Since(creationTimestamp).Seconds())
 	}
+}
 
-	// Record `kuberay_cluster_provisioned_duration_seconds` metric if just provisioned
-	if meta.IsStatusConditionTrue(newClusterStatus.Conditions, string(rayv1.RayClusterProvisioned)) &&
-		!meta.IsStatusConditionTrue(oldClusterStatus.Conditions, string(rayv1.RayClusterProvisioned)) {
-		collector.ObserveRayClusterProvisionedDuration(name, namespace, time.Since(creationTimestamp).Seconds())
+func emitRayClusterHeadPodReadyDuration(collector RayClusterMetricsCollector, rayClusterName, rayClusterNamespace string, originRayClusterStatus, rayClusterStatus rayv1.RayClusterStatus, creationTimestamp time.Time) {
+	// Emit kuberay_cluster_head_pod_ready_duration_seconds when a RayCluster's HeadPodReady status transitions from false (or unset) to true
+	if !meta.IsStatusConditionTrue(originRayClusterStatus.Conditions, string(rayv1.HeadPodReady)) &&
+		meta.IsStatusConditionTrue(rayClusterStatus.Conditions, string(rayv1.HeadPodReady)) {
+		collector.ObserveRayClusterHeadPodReadyDuration(rayClusterName, rayClusterNamespace, time.Since(creationTimestamp).Seconds())
 	}
 }