issue-141 - bug: Excessive Mem Usage/Freezing (#142)

Author: vladimirvivien

metrics/prom/prom_source.go

@@ -44,9 +44,9 @@ type PromConfig struct {
 func DefaultPromConfig() *PromConfig {
 	return &PromConfig{
 		Enabled:        true,
-		ScrapeInterval: 5 * time.Second,
-		RetentionTime:  1 * time.Hour,
-		MaxSamples:     10000,
+		ScrapeInterval: 4 * time.Second,
+		RetentionTime:  5 * time.Minute,
+		MaxSamples:     50,
 		Components: []prom.ComponentType{
 			prom.ComponentKubelet,
 			prom.ComponentCAdvisor,
@@ -444,26 +444,26 @@ func (p *PromMetricsSource) GetMetricsForPod(ctx context.Context, pod metav1.Obj
 
 // GetAllPodMetrics retrieves metrics for all pods
 func (p *PromMetricsSource) GetAllPodMetrics(ctx context.Context) ([]*metrics.PodMetrics, error) {
+	// Check health and get store reference under lock
 	p.mu.RLock()
-	defer p.mu.RUnlock()
-
 	if !p.isHealthyLocked() {
+		p.mu.RUnlock()
 		return nil, fmt.Errorf("prometheus source is not healthy")
 	}
-
 	if p.store == nil {
+		p.mu.RUnlock()
 		return nil, fmt.Errorf("metrics store not initialized")
 	}
+	store := p.store
+	p.mu.RUnlock()
 
-	// Get all unique pod/namespace combinations from labels
-	// This requires querying the store for label values
-	namespaces := p.store.GetLabelValues("namespace")
-	pods := p.store.GetLabelValues("pod")
+	// Get label values (store has its own lock)
+	namespaces := store.GetLabelValues("namespace")
+	pods := store.GetLabelValues("pod")
 
 	var allPodMetrics []*metrics.PodMetrics
 
-	// This is a simplified implementation - in production would need better logic
-	// to match pods with their namespaces
+	// Call GetPodMetrics without holding p.mu to avoid deadlock
 	for _, namespace := range namespaces {
 		for _, pod := range pods {
 			if podMetrics, err := p.GetPodMetrics(ctx, namespace, pod); err == nil {
@@ -539,12 +539,8 @@ func (p *PromMetricsSource) SetHealthCallback(callback func(healthy bool, info m
 }
 
 // buildSourceInfoLocked builds SourceInfo while holding the lock.
-// Must be called with p.mu held.
-func (p *PromMetricsSource) buildSourceInfoLocked(healthy bool) metrics.SourceInfo {
-	metricCount := 0
-	if p.store != nil {
-		metricCount = len(p.store.GetMetricNames())
-	}
+// metricCount must be obtained before acquiring lock to avoid nested locking.
+func (p *PromMetricsSource) buildSourceInfoLocked(healthy bool, metricCount int) metrics.SourceInfo {
 	return metrics.SourceInfo{
 		Type:         metrics.SourceTypePrometheus,
 		Version:      "v1.0.0",
@@ -557,6 +553,12 @@ func (p *PromMetricsSource) buildSourceInfoLocked(healthy bool) metrics.SourceIn
 
 // handleError is called when an error occurs during metrics collection
 func (p *PromMetricsSource) handleError(component prom.ComponentType, err error) {
+	// Get metric count BEFORE acquiring lock to avoid nested locking
+	metricCount := 0
+	if p.store != nil {
+		metricCount = len(p.store.GetMetricNames())
+	}
+
 	var shouldNotify bool
 	var info metrics.SourceInfo
 
@@ -572,7 +574,7 @@ func (p *PromMetricsSource) handleError(component prom.ComponentType, err error)
 	// Check if we need to notify (overall state changed from healthy to unhealthy)
 	if wasHealthy && !nowHealthy {
 		shouldNotify = true
-		info = p.buildSourceInfoLocked(false)
+		info = p.buildSourceInfoLocked(false, metricCount)
 	}
 	p.mu.Unlock()
 
@@ -584,6 +586,12 @@ func (p *PromMetricsSource) handleError(component prom.ComponentType, err error)
 
 // handleMetricsCollected is called when metrics are successfully collected
 func (p *PromMetricsSource) handleMetricsCollected(component prom.ComponentType, collectedMetrics *prom.ScrapedMetrics) {
+	// Get metric count BEFORE acquiring lock to avoid nested locking
+	metricCount := 0
+	if p.store != nil {
+		metricCount = len(p.store.GetMetricNames())
+	}
+
 	var shouldNotify bool
 	var info metrics.SourceInfo
 
@@ -604,7 +612,7 @@ func (p *PromMetricsSource) handleMetricsCollected(component prom.ComponentType,
 	// Check if we need to notify (overall state changed from unhealthy to healthy)
 	if !wasHealthy && nowHealthy {
 		shouldNotify = true
-		info = p.buildSourceInfoLocked(true)
+		info = p.buildSourceInfoLocked(true, metricCount)
 	}
 	p.mu.Unlock()
 

metrics/prom/prom_source_test.go

@@ -204,16 +204,16 @@ func TestNewPromMetricsSource_WithNilConfig(t *testing.T) {
 	}
 
 	// Verify default config values
-	if source.config.ScrapeInterval != 5*time.Second {
-		t.Errorf("Expected default scrape interval 5s, got %v", source.config.ScrapeInterval)
+	if source.config.ScrapeInterval != 4*time.Second {
+		t.Errorf("Expected default scrape interval 4s, got %v", source.config.ScrapeInterval)
 	}
 
-	if source.config.RetentionTime != 1*time.Hour {
-		t.Errorf("Expected default retention time 1h, got %v", source.config.RetentionTime)
+	if source.config.RetentionTime != 5*time.Minute {
+		t.Errorf("Expected default retention time 5m, got %v", source.config.RetentionTime)
 	}
 
-	if source.config.MaxSamples != 10000 {
-		t.Errorf("Expected default max samples 10000, got %d", source.config.MaxSamples)
+	if source.config.MaxSamples != 50 {
+		t.Errorf("Expected default max samples 50, got %d", source.config.MaxSamples)
 	}
 }
 
@@ -224,16 +224,16 @@ func TestDefaultPromConfig(t *testing.T) {
 		t.Error("Expected Enabled to be true")
 	}
 
-	if config.ScrapeInterval != 5*time.Second {
-		t.Errorf("Expected ScrapeInterval 5s, got %v", config.ScrapeInterval)
+	if config.ScrapeInterval != 4*time.Second {
+		t.Errorf("Expected ScrapeInterval 4s, got %v", config.ScrapeInterval)
 	}
 
-	if config.RetentionTime != 1*time.Hour {
-		t.Errorf("Expected RetentionTime 1h, got %v", config.RetentionTime)
+	if config.RetentionTime != 5*time.Minute {
+		t.Errorf("Expected RetentionTime 5m, got %v", config.RetentionTime)
 	}
 
-	if config.MaxSamples != 10000 {
-		t.Errorf("Expected MaxSamples 10000, got %d", config.MaxSamples)
+	if config.MaxSamples != 50 {
+		t.Errorf("Expected MaxSamples 50, got %d", config.MaxSamples)
 	}
 
 	// Check default components

prom/controller.go

@@ -138,7 +138,9 @@ func (cc *CollectorController) runCollector(ctx context.Context) {
 		case <-ctx.Done():
 			return
 		case <-ticker.C:
-			cc.collectFromAllComponents(ctx)
+			cycleCtx, cancel := context.WithTimeout(ctx, cc.config.Timeout)
+			cc.collectFromAllComponents(cycleCtx)
+			cancel()
 		}
 	}
 }

prom/scraper.go

@@ -437,9 +437,12 @@ func (ks *KubernetesScraper) scrapeTarget(ctx context.Context, target *ScrapeTar
 		}, err
 	}
 
-	// Convert to our internal format
+	// Convert to our internal format, filtering to only allowed metrics
 	metricFamilies := make(map[string]*MetricFamily)
 	for name, family := range families {
+		if !DefaultMetricAllowlist[name] {
+			continue
+		}
 		metricFamily := ks.convertMetricFamily(name, family)
 		metricFamilies[name] = metricFamily
 	}

prom/storage.go

@@ -6,6 +6,7 @@ import (
 	"strings"
 	"sync"
 	"time"
+	"unique"
 
 	"github.com/prometheus/prometheus/model/labels"
 )
@@ -22,6 +23,10 @@ type InMemoryStore struct {
 	labelNames  map[string]bool
 	labelValues map[string]map[string]bool // labelName -> values
 
+	// String interning - keeps handles alive to retain canonical strings
+	stringHandles map[string]unique.Handle[string]
+	handlesMu     sync.RWMutex
+
 	// Configuration
 	maxSamples    int
 	retentionTime time.Duration
@@ -39,12 +44,45 @@ func NewInMemoryStore(config *ScrapeConfig) *InMemoryStore {
 		metricNames:   make(map[string]bool),
 		labelNames:    make(map[string]bool),
 		labelValues:   make(map[string]map[string]bool),
+		stringHandles: make(map[string]unique.Handle[string]),
 		maxSamples:    config.MaxSamples,
 		retentionTime: config.RetentionTime,
 		lastCleanup:   time.Now(),
 	}
 }
 
+// InternString returns the canonical version of a string.
+// Uses Go's unique package to deduplicate repeated strings.
+func (store *InMemoryStore) InternString(s string) string {
+	store.handlesMu.RLock()
+	if h, ok := store.stringHandles[s]; ok {
+		store.handlesMu.RUnlock()
+		return h.Value()
+	}
+	store.handlesMu.RUnlock()
+
+	store.handlesMu.Lock()
+	defer store.handlesMu.Unlock()
+	if h, ok := store.stringHandles[s]; ok {
+		return h.Value()
+	}
+	h := unique.Make(s)
+	store.stringHandles[s] = h
+	return h.Value()
+}
+
+// internLabels creates a new labels.Labels with all strings interned.
+func (store *InMemoryStore) internLabels(lbls labels.Labels) labels.Labels {
+	interned := make(labels.Labels, len(lbls))
+	for i, l := range lbls {
+		interned[i] = labels.Label{
+			Name:  store.InternString(l.Name),
+			Value: store.InternString(l.Value),
+		}
+	}
+	return interned
+}
+
 // AddMetrics stores scraped metrics in the store
 func (store *InMemoryStore) AddMetrics(metrics *ScrapedMetrics) error {
 	if metrics.Error != nil {
@@ -55,6 +93,9 @@ func (store *InMemoryStore) AddMetrics(metrics *ScrapedMetrics) error {
 	defer store.mutex.Unlock()
 
 	for metricName, family := range metrics.Families {
+		// Intern metric name for consistent memory usage
+		metricName = store.InternString(metricName)
+
 		// Ensure metric exists in index
 		store.metricNames[metricName] = true
 
@@ -67,21 +108,23 @@ func (store *InMemoryStore) AddMetrics(metrics *ScrapedMetrics) error {
 		for _, ts := range family.TimeSeries {
 			seriesKey := ts.Labels.String()
 
-			// Update label indexes
+			// Update label indexes with interned strings
 			for _, label := range ts.Labels {
-				store.labelNames[label.Name] = true
-				if store.labelValues[label.Name] == nil {
-					store.labelValues[label.Name] = make(map[string]bool)
+				name := store.InternString(label.Name)
+				value := store.InternString(label.Value)
+				store.labelNames[name] = true
+				if store.labelValues[name] == nil {
+					store.labelValues[name] = make(map[string]bool)
 				}
-				store.labelValues[label.Name][label.Value] = true
+				store.labelValues[name][value] = true
 			}
 
 			// Get or create time series
 			existingSeries, exists := store.series[metricName][seriesKey]
 			if !exists {
-				// Create new series with ring buffer
+				// Create new series with ring buffer and interned labels
 				existingSeries = &TimeSeries{
-					Labels:  ts.Labels,
+					Labels:  store.internLabels(ts.Labels),
 					Samples: NewRingBuffer[MetricSample](store.maxSamples),
 				}
 				store.series[metricName][seriesKey] = existingSeries

prom/types.go

@@ -23,6 +23,19 @@ const (
 	ComponentKubeProxy         ComponentType = "kube-proxy"
 )
 
+// DefaultMetricAllowlist contains only the metrics ktop actually uses.
+// All other metrics are filtered out during scraping to reduce memory usage.
+var DefaultMetricAllowlist = map[string]bool{
+	"container_cpu_usage_seconds_total":      true,
+	"container_memory_working_set_bytes":     true,
+	"container_network_receive_bytes_total":  true,
+	"container_network_transmit_bytes_total": true,
+	"container_fs_reads_bytes_total":         true,
+	"container_fs_writes_bytes_total":        true,
+	"kubelet_running_pods":                   true,
+	"container_count":                        true,
+}
+
 // MetricSample represents a single metric data point
 type MetricSample struct {
 	Timestamp int64