UPSTREAM: <carry>: add http logging for kubelet metrics endpoint

Author: tkashem

pkg/kubelet/metrics/collectors/resource_metrics_test.go

@@ -19,11 +19,15 @@ package collectors
 import (
 	"context"
 	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
 	"strings"
 	"testing"
 	"time"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	compbasemetrics "k8s.io/component-base/metrics"
 	"k8s.io/component-base/metrics/testutil"
 	statsapi "k8s.io/kubelet/pkg/apis/stats/v1alpha1"
 	summaryprovidertest "k8s.io/kubernetes/pkg/kubelet/server/stats/testing"
@@ -419,3 +423,112 @@ func TestCollectResourceMetrics(t *testing.T) {
 func uint64Ptr(u uint64) *uint64 {
 	return &u
 }
+
+type fakeSummaryProvider struct {
+	stats *statsapi.Summary
+}
+
+func (p *fakeSummaryProvider) Get(ctx context.Context, updateStats bool) (*statsapi.Summary, error) {
+	return nil, fmt.Errorf("should not be invoked")
+}
+
+func (p *fakeSummaryProvider) GetCPUAndMemoryStats(ctx context.Context) (*statsapi.Summary, error) {
+	return p.stats, nil
+}
+
+func createSummary() *statsapi.Summary {
+	staticTimestamp := time.Unix(0, 1624396278302091597)
+	testTime := metav1.NewTime(staticTimestamp)
+
+	pods := make([]statsapi.PodStats, 0)
+	for i := 0; i < 500; i++ {
+		podstat := statsapi.PodStats{
+			PodRef: statsapi.PodReference{
+				Name:      fmt.Sprintf("pod_%d", i),
+				Namespace: fmt.Sprintf("namespace_%d", i),
+			},
+			CPU: &statsapi.CPUStats{
+				Time:                 testTime,
+				UsageCoreNanoSeconds: uint64Ptr(10000000000),
+			},
+			Memory: &statsapi.MemoryStats{
+				Time:            testTime,
+				WorkingSetBytes: uint64Ptr(1000),
+			},
+			Swap: &statsapi.SwapStats{
+				Time:           testTime,
+				SwapUsageBytes: uint64Ptr(5000),
+			},
+		}
+		for j := 0; j < 10; j++ {
+			podstat.Containers = append(podstat.Containers, statsapi.ContainerStats{
+				Name:      fmt.Sprintf("container_%d", j),
+				StartTime: metav1.NewTime(staticTimestamp.Add(-30 * time.Second)),
+				CPU: &statsapi.CPUStats{
+					Time:                 testTime,
+					UsageCoreNanoSeconds: uint64Ptr(10000000000),
+				},
+				Memory: &statsapi.MemoryStats{
+					Time:            testTime,
+					WorkingSetBytes: uint64Ptr(1000),
+				},
+				Swap: &statsapi.SwapStats{
+					Time:           testTime,
+					SwapUsageBytes: uint64Ptr(1000),
+				},
+			})
+		}
+		pods = append(pods, podstat)
+	}
+
+	return &statsapi.Summary{
+		Node: statsapi.NodeStats{
+			CPU: &statsapi.CPUStats{
+				Time:                 testTime,
+				UsageCoreNanoSeconds: uint64Ptr(10000000000),
+			},
+			Memory: &statsapi.MemoryStats{
+				Time:            testTime,
+				WorkingSetBytes: uint64Ptr(1000),
+			},
+			Swap: &statsapi.SwapStats{
+				Time:           testTime,
+				SwapUsageBytes: uint64Ptr(500),
+			},
+		},
+		Pods: pods,
+	}
+}
+
+func TestResourceMetricsCollector(t *testing.T) {
+	provider := &fakeSummaryProvider{stats: createSummary()}
+	collector := NewResourceMetricsCollector(provider)
+
+	registry := compbasemetrics.NewKubeRegistry()
+	registry.CustomMustRegister(collector)
+
+	handler := compbasemetrics.HandlerFor(registry, compbasemetrics.HandlerOpts{ErrorHandling: compbasemetrics.ContinueOnError})
+	// handler = http.HandlerFunc(WithHTTPLogging(handler))
+
+	server := httptest.NewUnstartedServer(handler)
+	defer server.Close()
+	server.EnableHTTP2 = false
+	server.StartTLS()
+
+	client := server.Client()
+	req, err := http.NewRequestWithContext(context.Background(), http.MethodGet, server.URL, nil)
+	if err != nil {
+		t.Errorf("failed to create a new http request - %v", err)
+		return
+	}
+
+	resp, err := client.Do(req)
+	if err != nil {
+		t.Fatalf("unexpected error from client.Do - %v", err)
+	}
+	bytes, err := io.ReadAll(resp.Body)
+	if err != nil {
+		t.Errorf("unexpected error while reading the response body - %v", err)
+	}
+	t.Logf("%v", len(bytes)/1024)
+}

pkg/kubelet/server/server.go

@@ -474,15 +474,18 @@ func (s *Server) InstallAuthNotRequiredHandlers() {
 		r.RawMustRegister(metrics.NewPrometheusCollector(prometheusHostAdapter{s.host}, containerPrometheusLabelsFunc(s.host), includedMetrics, clock.RealClock{}, cadvisorOpts))
 	}
 	s.restfulCont.Handle(cadvisorMetricsPath,
-		compbasemetrics.HandlerFor(r, compbasemetrics.HandlerOpts{ErrorHandling: compbasemetrics.ContinueOnError}),
-	)
+		WithHTTPLogging(
+			compbasemetrics.HandlerFor(r, compbasemetrics.HandlerOpts{ErrorHandling: compbasemetrics.ContinueOnError}),
+		))
 
 	s.addMetricsBucketMatcher("metrics/resource")
 	resourceRegistry := compbasemetrics.NewKubeRegistry()
-	resourceRegistry.CustomMustRegister(collectors.NewResourceMetricsCollector(s.resourceAnalyzer))
+	analyzer := &SummaryProviderTracker{ResourceAnalyzer: s.resourceAnalyzer}
+	resourceRegistry.CustomMustRegister(collectors.NewResourceMetricsCollector(analyzer))
 	s.restfulCont.Handle(resourceMetricsPath,
-		compbasemetrics.HandlerFor(resourceRegistry, compbasemetrics.HandlerOpts{ErrorHandling: compbasemetrics.ContinueOnError}),
-	)
+		WithHTTPLogging(
+			compbasemetrics.HandlerFor(resourceRegistry, compbasemetrics.HandlerOpts{ErrorHandling: compbasemetrics.ContinueOnError}),
+		))
 
 	// prober metrics are exposed under a different endpoint
 

pkg/kubelet/server/server_patch.go

@@ -0,0 +1,135 @@
+package server
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"time"
+
+	"k8s.io/apiserver/pkg/audit"
+	"k8s.io/apiserver/pkg/endpoints/responsewriter"
+	"k8s.io/klog/v2"
+	statsapi "k8s.io/kubelet/pkg/apis/stats/v1alpha1"
+	"k8s.io/kubernetes/pkg/kubelet/server/stats"
+
+	"github.com/google/uuid"
+)
+
+func WithHTTPLogging(handler http.Handler) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
+		logger := newHTTPLogger(w, req)
+		defer logger.log()
+
+		w = responsewriter.WrapForHTTP1Or2(logger)
+		handler.ServeHTTP(w, req)
+	})
+}
+
+func newHTTPLogger(w http.ResponseWriter, req *http.Request) *httpLogger {
+	auditID := audit.GetAuditIDTruncated(req.Context())
+	if len(auditID) == 0 {
+		auditID = uuid.New().String()
+	}
+	return &httpLogger{
+		w:          w,
+		startedAt:  time.Now(),
+		method:     req.Method,
+		requestURI: req.RequestURI,
+		auditID:    auditID,
+		userAgent:  req.UserAgent(),
+		srcIP:      req.RemoteAddr,
+	}
+}
+
+type httpLogger struct {
+	w http.ResponseWriter
+
+	method     string
+	requestURI string
+	auditID    string
+	userAgent  string
+	srcIP      string
+
+	startedAt      time.Time
+	writeLatency   time.Duration
+	flushLatency   time.Duration
+	writeBytes     int
+	statusRecorded bool
+	statusCode     int
+}
+
+var _ http.ResponseWriter = &httpLogger{}
+var _ responsewriter.UserProvidedDecorator = &httpLogger{}
+
+func (l *httpLogger) Unwrap() http.ResponseWriter {
+	return l.w
+}
+
+// Header implements http.ResponseWriter.
+func (l *httpLogger) Header() http.Header {
+	return l.w.Header()
+}
+
+// Write implements http.ResponseWriter.
+func (l *httpLogger) Write(b []byte) (int, error) {
+	if !l.statusRecorded {
+		l.record(http.StatusOK) // Default if WriteHeader hasn't been called
+	}
+	now := time.Now()
+	var written int
+	defer func() {
+		l.writeLatency += time.Since(now)
+		l.writeBytes += written
+	}()
+	written, err := l.w.Write(b)
+	return written, err
+}
+
+func (l *httpLogger) Flush() {
+	now := time.Now()
+	defer func() {
+		l.flushLatency += time.Since(now)
+	}()
+	l.w.(http.Flusher).Flush()
+}
+
+// WriteHeader implements http.ResponseWriter.
+func (l *httpLogger) WriteHeader(status int) {
+	l.record(status)
+	l.w.WriteHeader(status)
+}
+
+func (l *httpLogger) record(status int) {
+	l.statusCode = status
+	l.statusRecorded = true
+}
+
+func (l *httpLogger) log() {
+	latency := time.Since(l.startedAt)
+	kvs := []interface{}{
+		"startedAt", l.startedAt,
+		"method", l.method,
+		"URI", l.requestURI,
+		"latency", latency,
+		"userAgent", l.userAgent,
+		"audit-ID", l.auditID,
+		"srcIP", l.srcIP,
+		"status", l.statusCode,
+		"writeLatency", l.writeLatency,
+		"writtenBytes", fmt.Sprintf("%dK", l.writeBytes/1024),
+		"flushLatency", l.flushLatency,
+	}
+	klog.V(1).InfoSDepth(1, "HTTP", kvs...)
+}
+
+type SummaryProviderTracker struct {
+	stats.ResourceAnalyzer
+}
+
+func (t *SummaryProviderTracker) GetCPUAndMemoryStats(ctx context.Context) (*statsapi.Summary, error) {
+	now := time.Now()
+	defer func() {
+		klog.InfoS("GetCPUAndMemoryStats", "latency", time.Since(now))
+	}()
+	return t.ResourceAnalyzer.GetCPUAndMemoryStats(ctx)
+}