serivce monitor and fixes

Author: Rishab87

clusterloader2/pkg/measurement/common/kube_state_metrics_measurement.go

@@ -29,7 +29,6 @@ import (
 	"github.com/prometheus/common/model"
 	clientset "k8s.io/client-go/kubernetes"
 	"k8s.io/klog/v2"
-
 )
 
 const (
@@ -128,11 +127,11 @@ func (m *ksmLatencyMeasurement) Execute(config *measurement.Config) ([]measureme
 		// the scrape interval so we should cancel.
 		m.startQuerying(m.ctx, client, probeIntervalDefault)
 		// Retrieve initial latency when first call is done.
-		m.initialLatency, err = m.retrieveLatencyMetrics(m.ctx, client)
+		m.initialLatency, err = m.retrieveKSMLatencyMetrics(m.ctx, client)
 		return nil, err
 	case "gather":
 		defer m.cancel()
-		return m.createLatencySummary(m.ctx, client)
+		return m.createKSMLatencySummary(m.ctx, client)
 	default:
 		return nil, fmt.Errorf("unknown action %v", action)
 	}
@@ -152,45 +151,33 @@ func (m *crsmLatencyMeasurement) Execute(config *measurement.Config) ([]measurem
 			return nil, nil
 		}
 		m.startQuerying(m.ctx, client, probeIntervalDefault)
-		m.initialLatency, err = m.retrieveLatencyMetrics(m.ctx, client)
+		m.initialLatency, err = m.retrieveCRSMLatencyMetrics(m.ctx, client)
 		return nil, err
 	case "gather":
 		defer m.cancel()
-		return m.createLatencySummary(m.ctx, client)
+		return m.createCRSMLatencySummary(m.ctx, client)
 	default:
 		return nil, fmt.Errorf("unknown action %v", action)
 	}
 }
 
-func getMetricsFromService(ctx context.Context, client clientset.Interface, namespace, serviceName string, port int) (string, error) {
-	ctx, cancel := context.WithTimeout(ctx, 2*time.Minute)
-	defer cancel()
-	out, err := client.CoreV1().RESTClient().Get().
-		Resource("services").
-		SubResource("proxy").
-		Namespace(namespace).
-		Name(fmt.Sprintf("%v:%v", serviceName, port)).
-		Suffix("metrics").
-		Do(ctx).Raw()
-	return string(out), err
-}
-
 func (m *ksmLatencyMeasurement) stop() error {
-    if !m.isRunning {
-        return fmt.Errorf("%s: measurement was not running", m)
-    }
-    m.isRunning = false
-    m.cancel()
-    m.wg.Wait()
-    return nil 
+	if !m.isRunning {
+		return fmt.Errorf("%s: measurement was not running", m)
+	}
+	m.cancel()
+	m.wg.Wait()
+	return nil
 }
 
-func (m *ksmLatencyMeasurement) createLatencySummary(ctx context.Context, client clientset.Interface) ([]measurement.Summary, error) {
-	latestLatency, err := m.retrieveLatencyMetrics(ctx, client)
+func (m *ksmLatencyMeasurement) createKSMLatencySummary(ctx context.Context, client clientset.Interface) ([]measurement.Summary, error) {
+	latestLatency, err := m.retrieveKSMLatencyMetrics(ctx, client)
 	if err != nil {
 		return nil, err
 	}
-	m.stop()
+	if err = m.stop(); err != nil {
+		return nil, err
+	}
 	// We want to subtract the latest histogram from the first one we collect.
 	finalLatency := HistogramSub(latestLatency, m.initialLatency)
 	// Pretty Print the report.
@@ -202,6 +189,12 @@ func (m *ksmLatencyMeasurement) createLatencySummary(ctx context.Context, client
 	if err != nil {
 		return nil, err
 	}
+	
+	// Log the latency results for visibility
+	klog.Infof("%s: Latency Results - P50: %v, P90: %v, P99: %v", 
+		m, result.Perc50, result.Perc90, result.Perc99)
+	klog.Infof("%s: Final latency summary: %s", m, content)
+	
 	// Create Summary.
 	return []measurement.Summary{measurement.CreateSummary(ksmLatencyName, "json", content)}, nil
 }
@@ -211,43 +204,86 @@ func (m *ksmLatencyMeasurement) createLatencySummary(ctx context.Context, client
 func (m *ksmLatencyMeasurement) startQuerying(ctx context.Context, client clientset.Interface, interval time.Duration) {
 	m.isRunning = true
 	m.wg.Add(1)
-	go func() {
-		defer m.wg.Done()
-		for {
-			select {
-			case <-ctx.Done():
-				return
-			case <-time.After(interval):
-				_, err := getMetricsFromService(ctx, client, m.namespace, m.serviceName, m.metricsPort)
-				if err != nil {
-					klog.V(2).Infof("error during fetching metrics from service: %v", err)
-				}
+	go m.queryLoop(ctx, client, interval)
+}
+
+func (m *ksmLatencyMeasurement) queryLoop(ctx context.Context, client clientset.Interface, interval time.Duration) {
+	defer m.wg.Done()
+	queryCount := 0
+	for {
+		select {
+		case <-ctx.Done():
+			klog.V(2).Infof("%s: stopping query loop after %d queries", m, queryCount)
+			return
+		case <-time.After(interval):
+			queryCount++
+			klog.V(4).Infof("%s: executing query #%d", m, queryCount)
+			output, err := m.getMetricsFromService(ctx, client, m.metricsPort)
+			if err != nil {
+				klog.V(2).Infof("%s: error during fetching metrics from service (query #%d): %v", m, queryCount, err)
+			}
+			if output == "" {
+				klog.V(2).Infof("%s: /metrics endpoint returned no data in namespace: %s from service: %s port: %d", 
+					m, m.namespace, m.serviceName, m.metricsPort)
+			} else {
+				klog.V(4).Infof("%s: successfully fetched %d bytes from metrics endpoint (query #%d)", 
+					m, len(output), queryCount)
 			}
 		}
-	}()
+	}
 }
 
-func (m *ksmLatencyMeasurement) retrieveLatencyMetrics(ctx context.Context, c clientset.Interface) (*measurementutil.Histogram, error) {
-	hist := measurementutil.NewHistogram(nil)
-	output, err := getMetricsFromService(ctx, c, m.namespace, m.serviceName, m.selfPort)
+func (m *ksmLatencyMeasurement) retrieveKSMLatencyMetrics(ctx context.Context, c clientset.Interface) (*measurementutil.Histogram, error) {
+	klog.V(4).Infof("%s: retrieving KSM latency metrics", m)
+	ksmHist := measurementutil.NewHistogram(nil)
+	output, err := m.getMetricsFromService(ctx, c, m.selfPort)
 	if err != nil {
-		return hist, err
+		klog.Errorf("%s: failed to get metrics from service: %v", m, err)
+		return ksmHist, err
 	}
 	samples, err := measurementutil.ExtractMetricSamples(output)
 	if err != nil {
-		return hist, err
+		klog.Errorf("%s: failed to extract metric samples: %v", m, err)
+		return ksmHist, err
 	}
+	
+	sampleCount := 0
 	for _, sample := range samples {
-		if sample.Metric[model.MetricNameLabel] == ksmRequestDurationMetricName {
-			measurementutil.ConvertSampleToHistogram(sample, hist)
+		switch sample.Metric[model.MetricNameLabel] {
+		case ksmRequestDurationMetricName:
+			measurementutil.ConvertSampleToHistogram(sample, ksmHist)
+			sampleCount++
 		}
 	}
-	return hist, nil
+	klog.V(4).Infof("%s: processed %d histogram samples", m, sampleCount)
+	return ksmHist, nil
+}
+
+func (m *ksmLatencyMeasurement) getMetricsFromService(ctx context.Context, client clientset.Interface, port int) (string, error) {
+	ctx, cancel := context.WithTimeout(ctx, 2*time.Minute)
+	defer cancel()
+	
+	klog.V(4).Infof("%s: fetching metrics from %s/%s:%d", m, m.namespace, m.serviceName, port)
+	
+	out, err := client.CoreV1().RESTClient().Get().
+		Resource("services").
+		SubResource("proxy").
+		Namespace(m.namespace).
+		Name(fmt.Sprintf("%v:%v", m.serviceName, port)).
+		Suffix("metrics").
+		Do(ctx).Raw()
+	
+	if err != nil {
+		klog.V(2).Infof("%s: error fetching metrics from %s/%s:%d: %v", m, m.namespace, m.serviceName, port, err)
+	}
+	
+	return string(out), err
 }
+
 // Dispose cleans up after the measurement.
 func (m *ksmLatencyMeasurement) Dispose() {
 	if err := m.stop(); err != nil {
-		klog.V(2).Infof("error during dispose call: %v", err)
+		klog.V(2).Infof("%s: error during dispose call: %v", m, err)
 	}
 }
 
@@ -257,21 +293,22 @@ func (m *ksmLatencyMeasurement) String() string {
 }
 
 func (m *crsmLatencyMeasurement) stop() error {
-    if !m.isRunning {
-        return fmt.Errorf("%s: measurement was not running", m)
-    }
-    m.isRunning = false
-    m.cancel()
-    m.wg.Wait()
-    return nil
+	if !m.isRunning {
+		return fmt.Errorf("%s: measurement was not running", m)
+	}
+	m.cancel()
+	m.wg.Wait()
+	return nil
 }
 
-func (m *crsmLatencyMeasurement) createLatencySummary(ctx context.Context, client clientset.Interface) ([]measurement.Summary, error) {
-	latestLatency, err := m.retrieveLatencyMetrics(ctx, client)
+func (m *crsmLatencyMeasurement) createCRSMLatencySummary(ctx context.Context, client clientset.Interface) ([]measurement.Summary, error) {
+	latestLatency, err := m.retrieveCRSMLatencyMetrics(ctx, client)
 	if err != nil {
 		return nil, err
 	}
-	m.stop()
+	if err = m.stop(); err != nil {
+		return nil, err
+	}
 	finalLatency := HistogramSub(latestLatency, m.initialLatency)
 	result := &measurementutil.LatencyMetric{}
 	if err = SetQuantileFromHistogram(result, finalLatency); err != nil {
@@ -281,49 +318,97 @@ func (m *crsmLatencyMeasurement) createLatencySummary(ctx context.Context, clien
 	if err != nil {
 		return nil, err
 	}
+	
+	// Log the latency results for visibility
+	klog.Infof("%s: Latency Results - P50: %v, P90: %v, P99: %v", 
+		m, result.Perc50, result.Perc90, result.Perc99)
+	klog.Infof("%s: Final latency summary: %s", m, content)
+	
 	return []measurement.Summary{measurement.CreateSummary(crsmLatencyName, "json", content)}, nil
 }
 
 func (m *crsmLatencyMeasurement) startQuerying(ctx context.Context, client clientset.Interface, interval time.Duration) {
 	m.isRunning = true
 	m.wg.Add(1)
-	go func() {
-		defer m.wg.Done()
-		for {
-			select {
-			case <-ctx.Done():
-				return
-			case <-time.After(interval):
-				_, err := getMetricsFromService(ctx, client, m.namespace, m.serviceName, m.metricsPort)
-				if err != nil {
-					klog.V(2).Infof("error during fetching metrics from service: %v", err)
-				}
+	go m.queryLoop(ctx, client, interval)
+}
+
+func (m *crsmLatencyMeasurement) queryLoop(ctx context.Context, client clientset.Interface, interval time.Duration) {
+	defer m.wg.Done()
+	queryCount := 0
+	for {
+		select {
+		case <-ctx.Done():
+			klog.V(2).Infof("%s: stopping query loop after %d queries", m, queryCount)
+			return
+		case <-time.After(interval):
+			queryCount++
+			klog.V(4).Infof("%s: executing query #%d", m, queryCount)
+			output, err := m.getMetricsFromService(ctx, client, m.metricsPort)
+			if err != nil {
+				klog.V(2).Infof("%s: error during fetching metrics from service (query #%d): %v", m, queryCount, err)
+			}
+			if output == "" {
+				klog.V(2).Infof("%s: /metrics endpoint returned no data in namespace: %s from service: %s port: %d", 
+					m, m.namespace, m.serviceName, m.metricsPort)
+			} else {
+				klog.V(4).Infof("%s: successfully fetched %d bytes from metrics endpoint (query #%d)", 
+					m, len(output), queryCount)
 			}
 		}
-	}()
+	}
 }
 
-func (m *crsmLatencyMeasurement) retrieveLatencyMetrics(ctx context.Context, c clientset.Interface) (*measurementutil.Histogram, error) {
-	hist := measurementutil.NewHistogram(nil)
-	output, err := getMetricsFromService(ctx, c, m.namespace, m.serviceName, m.selfPort)
+func (m *crsmLatencyMeasurement) retrieveCRSMLatencyMetrics(ctx context.Context, c clientset.Interface) (*measurementutil.Histogram, error) {
+	klog.V(4).Infof("%s: retrieving CRSM latency metrics", m)
+	crsmHist := measurementutil.NewHistogram(nil)
+	output, err := m.getMetricsFromService(ctx, c, m.selfPort)
 	if err != nil {
-		return hist, err
+		klog.Errorf("%s: failed to get metrics from service: %v", m, err)
+		return crsmHist, err
 	}
 	samples, err := measurementutil.ExtractMetricSamples(output)
 	if err != nil {
-		return hist, err
+		klog.Errorf("%s: failed to extract metric samples: %v", m, err)
+		return crsmHist, err
 	}
+	
+	sampleCount := 0
 	for _, sample := range samples {
-		if sample.Metric[model.MetricNameLabel] == ksmRequestDurationMetricName {
-			measurementutil.ConvertSampleToHistogram(sample, hist)
+		switch sample.Metric[model.MetricNameLabel] {
+		case ksmRequestDurationMetricName:
+			measurementutil.ConvertSampleToHistogram(sample, crsmHist)
+			sampleCount++
 		}
 	}
-	return hist, nil
+	klog.V(4).Infof("%s: processed %d histogram samples", m, sampleCount)
+	return crsmHist, nil
+}
+
+func (m *crsmLatencyMeasurement) getMetricsFromService(ctx context.Context, client clientset.Interface, port int) (string, error) {
+	ctx, cancel := context.WithTimeout(ctx, 2*time.Minute)
+	defer cancel()
+	
+	klog.V(4).Infof("%s: fetching metrics from %s/%s:%d", m, m.namespace, m.serviceName, port)
+	
+	out, err := client.CoreV1().RESTClient().Get().
+		Resource("services").
+		SubResource("proxy").
+		Namespace(m.namespace).
+		Name(fmt.Sprintf("%v:%v", m.serviceName, port)).
+		Suffix("metrics").
+		Do(ctx).Raw()
+	
+	if err != nil {
+		klog.V(2).Infof("%s: error fetching metrics from %s/%s:%d: %v", m, m.namespace, m.serviceName, port, err)
+	}
+	
+	return string(out), err
 }
 
 func (m *crsmLatencyMeasurement) Dispose() {
 	if err := m.stop(); err != nil {
-    	klog.V(2).Infof("error during dispose call: %v", err)
+		klog.V(2).Infof("%s: error during dispose call: %v", m, err)
 	}
 }
 

clusterloader2/pkg/prometheus/manifests/exporters/custom-resource-state-metrics/serviceMonitor.yaml

@@ -0,0 +1,22 @@
+apiVersion: monitoring.coreos.com/v1
+kind: ServiceMonitor
+metadata:
+  labels:
+    app.kubernetes.io/name: kube-state-metrics
+    app.kubernetes.io/version: 2.13.0
+  name: custom-resource-state-metrics-sm
+  namespace: custom-resource-state-metrics-perf-test
+spec:
+  selector:
+    matchLabels:
+      app.kubernetes.io/name: custom-resource-state-metrics
+  namespaceSelector:
+    matchNames:
+      - custom-resource-state-metrics-perf-test
+  endpoints:
+    - port: telemetry
+      path: /metrics
+      interval: 30s
+    - port: http-metrics
+      path: /metrics
+      interval: 30s