feat: Add node statistics (#450)

* add pending pods table

* rename functions

* rename remaining component functions

* bump client version

* add node statistics function

* add set pod function

* add test

* add test and fix time check

* add node statistics to cluster ping

* update comment

* add delete pod function

* rename function

* format

* watch pods

* sync pods

* get node name

* add node health

* add pods to the cache even if the node name is empty to track pods that are not scheduled yet

* revert last commit

* fix time comparison

* fix tests

* rename tables

Author: maciaszczykm

go.mod

@@ -36,7 +36,7 @@ require (
 	github.com/opencost/opencost/core v0.0.0-20241216191657-30e5d9a27f41
 	github.com/orcaman/concurrent-map/v2 v2.0.1
 	github.com/pkg/errors v0.9.1
-	github.com/pluralsh/console/go/client v1.44.0
+	github.com/pluralsh/console/go/client v1.45.0
 	github.com/pluralsh/controller-reconcile-helper v0.1.0
 	github.com/pluralsh/gophoenix v0.1.3-0.20231201014135-dff1b4309e34
 	github.com/pluralsh/polly v0.2.0

go.sum

@@ -1802,8 +1802,8 @@ github.com/pkg/sftp v1.10.1/go.mod h1:lYOWFsE0bwd1+KfKJaKeuokY15vzFx25BLbzYYoAxZ
 github.com/pkg/sftp v1.13.1/go.mod h1:3HaPG6Dq1ILlpPZRO0HVMrsydcdLt6HRDccSgb87qRg=
 github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10 h1:GFCKgmp0tecUJ0sJuv4pzYCqS9+RGSn52M3FUwPs+uo=
 github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10/go.mod h1:t/avpk3KcrXxUnYOhZhMXJlSEyie6gQbtLq5NM3loB8=
-github.com/pluralsh/console/go/client v1.44.0 h1:+CXCHhHIMKvzNl6do5nVwh1ffVzfEDgSPLhfXKvpOfI=
-github.com/pluralsh/console/go/client v1.44.0/go.mod h1:8XlMMN3LLAN9JZo69f8X/XN7Qt1+aaKpgTvvQGfSiEU=
+github.com/pluralsh/console/go/client v1.45.0 h1:eg6MlfD2NswPIM88Jhj4z//T86DoB8Ovl5QyNE8Gii0=
+github.com/pluralsh/console/go/client v1.45.0/go.mod h1:8XlMMN3LLAN9JZo69f8X/XN7Qt1+aaKpgTvvQGfSiEU=
 github.com/pluralsh/controller-reconcile-helper v0.1.0 h1:BV3dYZFH5rn8ZvZjtpkACSv/GmLEtRftNQj/Y4ddHEo=
 github.com/pluralsh/controller-reconcile-helper v0.1.0/go.mod h1:RxAbvSB4/jkvx616krCdNQXPbpGJXW3J1L3rASxeFOA=
 github.com/pluralsh/gophoenix v0.1.3-0.20231201014135-dff1b4309e34 h1:ab2PN+6if/Aq3/sJM0AVdy1SYuMAnq4g20VaKhTm/Bw=

pkg/cache/db/cache.go

@@ -62,7 +62,7 @@ func GetComponentCache() *ComponentCache {
 	return cache
 }
 
-// Children retrieves all child components and their descendants (up to 4 levels deep) for a given component UID.
+// ComponentChildren retrieves all child components and their descendants (up to 4 levels deep) for a given component UID.
 // It returns a slice of ComponentChildAttributes containing information about each child component.
 //
 // Parameters:
@@ -71,7 +71,7 @@ func GetComponentCache() *ComponentCache {
 // Returns:
 //   - []ComponentChildAttributes: A slice containing the child components and their attributes
 //   - error: An error if the database operation fails or if the connection cannot be established
-func (in *ComponentCache) Children(uid string) (result []client.ComponentChildAttributes, err error) {
+func (in *ComponentCache) ComponentChildren(uid string) (result []client.ComponentChildAttributes, err error) {
 	conn, err := in.pool.Take(context.Background())
 	if err != nil {
 		return result, err
@@ -98,11 +98,11 @@ func (in *ComponentCache) Children(uid string) (result []client.ComponentChildAt
 	return result, err
 }
 
-// Set stores or updates a component's attributes in the cache.
+// SetComponent stores or updates a component's attributes in the cache.
 // It takes a ComponentChildAttributes parameter containing the component's data.
 // If a component with the same UID exists, it will be updated; otherwise, a new entry is created.
 // Returns an error if the database operation fails or if the connection cannot be established.
-func (in *ComponentCache) Set(component client.ComponentChildAttributes) error {
+func (in *ComponentCache) SetComponent(component client.ComponentChildAttributes) error {
 	conn, err := in.pool.Take(context.Background())
 	if err != nil {
 		return err
@@ -123,6 +123,22 @@ func (in *ComponentCache) Set(component client.ComponentChildAttributes) error {
 	})
 }
 
+// DeleteComponent removes a component from the cache by its unique identifier.
+// It takes a uid string parameter identifying the component to delete.
+// Returns an error if the operation fails or if the connection cannot be established.
+func (in *ComponentCache) DeleteComponent(uid string) error {
+	conn, err := in.pool.Take(context.Background())
+	if err != nil {
+		return err
+	}
+	defer in.pool.Put(conn)
+
+	query := `DELETE FROM component WHERE uid = ?`
+	return sqlitex.ExecuteTransient(conn, query, &sqlitex.ExecOptions{
+		Args: []any{uid},
+	})
+}
+
 // HealthScore returns a percentage of healthy components to total components in the cluster.
 // The percentage is calculated as the number of healthy components divided by the total number of components.
 // Returns an int value between 0 and 100, where 100 indicates all components are healthy.
@@ -135,7 +151,7 @@ func (in *ComponentCache) HealthScore() (int64, error) {
 	defer in.pool.Put(conn)
 
 	var ratio int64
-	err = sqlitex.ExecuteTransient(conn, healthyComponentsRatio, &sqlitex.ExecOptions{
+	err = sqlitex.ExecuteTransient(conn, clusterHealthScore, &sqlitex.ExecOptions{
 		ResultFunc: func(stmt *sqlite.Stmt) error {
 			ratio = stmt.ColumnInt64(0)
 			return nil
@@ -144,22 +160,81 @@ func (in *ComponentCache) HealthScore() (int64, error) {
 	return ratio, err
 }
 
-// Delete removes a component from the cache by its unique identifier.
-// It takes a uid string parameter identifying the component to delete.
+// SetPod stores or updates a pod's attributes in the cache.
+// It takes pod name, namespace, uid, node name and creation timestamp as parameters.
+// If a pod with the same UID exists, it will be updated; otherwise, a new entry is created.
+// Returns an error if the database operation fails or if the connection cannot be established.
+func (in *ComponentCache) SetPod(name, namespace, uid, node string, createdAt int64) error {
+	conn, err := in.pool.Take(context.Background())
+	if err != nil {
+		return err
+	}
+	defer in.pool.Put(conn)
+
+	return sqlitex.ExecuteTransient(conn, setPod, &sqlitex.ExecOptions{
+		Args: []interface{}{
+			name,
+			namespace,
+			uid,
+			node,
+			createdAt,
+		},
+	})
+}
+
+// DeletePod removes a pod from the cache by its unique identifier.
+// It takes a uid string parameter identifying the pod to delete.
 // Returns an error if the operation fails or if the connection cannot be established.
-func (in *ComponentCache) Delete(uid string) error {
+func (in *ComponentCache) DeletePod(uid string) error {
 	conn, err := in.pool.Take(context.Background())
 	if err != nil {
 		return err
 	}
 	defer in.pool.Put(conn)
 
-	query := `DELETE FROM Component WHERE uid = ?`
+	query := `DELETE FROM pod WHERE uid = ?`
 	return sqlitex.ExecuteTransient(conn, query, &sqlitex.ExecOptions{
 		Args: []any{uid},
 	})
 }
 
+// NodeStatistics returns a list of node statistics including the node name and count of pending pods
+// that were created more than 5 minutes ago. Each NodeStatisticAttributes contains the node name and
+// the number of pending pods for that node. The health field is currently not implemented.
+// Returns an error if the database operation fails or if the connection cannot be established.
+func (in *ComponentCache) NodeStatistics() ([]*client.NodeStatisticAttributes, error) {
+	conn, err := in.pool.Take(context.Background())
+	if err != nil {
+		return nil, err
+	}
+	defer in.pool.Put(conn)
+
+	result := make([]*client.NodeStatisticAttributes, 0)
+	err = sqlitex.ExecuteTransient(conn, nodeStatistics, &sqlitex.ExecOptions{
+		ResultFunc: func(stmt *sqlite.Stmt) error {
+			pendingPods := stmt.ColumnInt64(1)
+			result = append(result, &client.NodeStatisticAttributes{
+				Name:        lo.ToPtr(stmt.ColumnText(0)),
+				PendingPods: &pendingPods,
+				Health:      nodeHealth(pendingPods),
+			})
+			return nil
+		},
+	})
+	return result, err
+}
+
+func nodeHealth(pendingPods int64) *client.NodeStatisticHealth {
+	switch {
+	case pendingPods == 0:
+		return lo.ToPtr(client.NodeStatisticHealthHealthy)
+	case pendingPods <= 3:
+		return lo.ToPtr(client.NodeStatisticHealthWarning)
+	default:
+		return lo.ToPtr(client.NodeStatisticHealthFailed)
+	}
+}
+
 // Close closes the connection pool and cleans up temporary file if necessary
 func (in *ComponentCache) Close() error {
 	mutex.Lock()
@@ -216,17 +291,17 @@ func (in *ComponentCache) init() error {
 	}
 
 	in.pool = pool
-	return in.initTable()
+	return in.initTables()
 }
 
-func (in *ComponentCache) initTable() error {
+func (in *ComponentCache) initTables() error {
 	conn, err := in.pool.Take(context.Background())
 	if err != nil {
 		return err
 	}
 	defer in.pool.Put(conn)
 
-	return sqlitex.ExecuteScript(conn, createTable, nil)
+	return sqlitex.ExecuteScript(conn, createTables, nil)
 }
 
 // Option represents a function that configures the ComponentCache

pkg/cache/db/cache_bench_test.go

@@ -16,16 +16,16 @@ import (
 // deleting components, and a combined workload of setting and retrieving.
 //
 // Results show that the in-memory cache is generally faster than the file-based cache,
-// especially for write operations like Set. Here's a summary of the performance comparison:
+// especially for write operations like SetComponent. Here's a summary of the performance comparison:
 //
-// 1. Set: In-memory is ~60x faster than file-based
-// 2. Children: In-memory is ~1.1x faster than file-based
-// 3. Delete: In-memory is ~1.6x faster than file-based
-// 4. SetAndChildren: In-memory is ~3.4x faster than file-based
+// 1. SetComponent: In-memory is ~60x faster than file-based
+// 2. ComponentChildren: In-memory is ~1.1x faster than file-based
+// 3. DeleteComponent: In-memory is ~1.6x faster than file-based
+// 4. SetComponentAndComponentChildren AndChildren: In-memory is ~3.4x faster than file-based
 //
-// The performance difference is most significant for write operations (Set)
+// The performance difference is most significant for write operations (SetComponent)
 // because file-based operations involve disk I/O, which is much slower than memory access.
-// Read operations (Children) show a smaller performance gap because SQLite's query
+// Read operations (ComponentChildren) show a smaller performance gap because SQLite's query
 // optimization works well for both storage modes.
 
 const (
@@ -50,7 +50,7 @@ func setupTestData(b *testing.B, cache *db.ComponentCache) {
 		State:     &state,
 	}
 
-	err := cache.Set(rootComponent)
+	err := cache.SetComponent(rootComponent)
 	require.NoError(b, err)
 
 	// Create 10 first-level children
@@ -67,7 +67,7 @@ func setupTestData(b *testing.B, cache *db.ComponentCache) {
 			State:     &state,
 		}
 
-		err := cache.Set(component)
+		err := cache.SetComponent(component)
 		require.NoError(b, err)
 
 		// Create 5 second-level children for each first-level child
@@ -84,7 +84,7 @@ func setupTestData(b *testing.B, cache *db.ComponentCache) {
 				State:     &state,
 			}
 
-			err := cache.Set(childComponent)
+			err := cache.SetComponent(childComponent)
 			require.NoError(b, err)
 
 			// Create 3 third-level children for each second-level child
@@ -101,7 +101,7 @@ func setupTestData(b *testing.B, cache *db.ComponentCache) {
 					State:     &state,
 				}
 
-				err := cache.Set(grandchildComponent)
+				err := cache.SetComponent(grandchildComponent)
 				require.NoError(b, err)
 			}
 		}
@@ -122,8 +122,8 @@ func BenchmarkMemoryCache(b *testing.B) {
 	cache := db.GetComponentCache()
 	defer cache.Close()
 
-	// Run the Set benchmark
-	b.Run("Set", func(b *testing.B) {
+	// Run the SetComponent benchmark
+	b.Run("SetComponent", func(b *testing.B) {
 		state := client.ComponentState("Healthy")
 		group := testGroup
 		namespace := testNamespace
@@ -143,39 +143,39 @@ func BenchmarkMemoryCache(b *testing.B) {
 			}
 			i++
 
-			err := cache.Set(component)
+			err := cache.SetComponent(component)
 			require.NoError(b, err)
 		}
 	})
 
 	// Setup test data for the remaining benchmarks
 	setupTestData(b, cache)
 
-	// Run the Children benchmark
-	b.Run("Children", func(b *testing.B) {
+	// Run the ComponentChildren benchmark
+	b.Run("ComponentChildren", func(b *testing.B) {
 		b.ResetTimer()
 		for b.Loop() {
-			children, err := cache.Children("root-uid")
+			children, err := cache.ComponentChildren("root-uid")
 			require.NoError(b, err)
 			require.NotEmpty(b, children)
 		}
 	})
 
-	// Run the Delete benchmark
-	b.Run("Delete", func(b *testing.B) {
+	// Run the DeleteComponent benchmark
+	b.Run("DeleteComponent", func(b *testing.B) {
 		b.ResetTimer()
 		for i := 0; i < b.N; i++ {
-			// Delete a level 1 component (which should cascade to its children)
-			err := cache.Delete("level1-uid-" + string(rune('a'+i%10)))
+			// DeleteComponent a level 1 component (which should cascade to its children)
+			err := cache.DeleteComponent("level1-uid-" + string(rune('a'+i%10)))
 			require.NoError(b, err)
 		}
 	})
 
 	// Setup test data again for the combined benchmark
 	setupTestData(b, cache)
 
-	// Run the SetAndChildren benchmark
-	b.Run("SetAndChildren", func(b *testing.B) {
+	// Run the SetComponentAndComponentChildren benchmark
+	b.Run("SetComponentAndComponentChildren", func(b *testing.B) {
 		state := client.ComponentState("Healthy")
 		group := testGroup
 		namespace := testNamespace
@@ -196,11 +196,11 @@ func BenchmarkMemoryCache(b *testing.B) {
 				State:     &state,
 			}
 
-			err := cache.Set(component)
+			err := cache.SetComponent(component)
 			require.NoError(b, err)
 
 			// Retrieve children
-			children, err := cache.Children("root-uid")
+			children, err := cache.ComponentChildren("root-uid")
 			require.NoError(b, err)
 			require.NotEmpty(b, children)
 		}
@@ -227,8 +227,8 @@ func BenchmarkFileCache(b *testing.B) {
 		_ = os.Remove(benchDBFile)
 	}()
 
-	// Run the Set benchmark
-	b.Run("Set", func(b *testing.B) {
+	// Run the SetComponent benchmark
+	b.Run("SetComponent", func(b *testing.B) {
 		state := client.ComponentState("Healthy")
 		group := testGroup
 		namespace := testNamespace
@@ -246,30 +246,30 @@ func BenchmarkFileCache(b *testing.B) {
 				State:     &state,
 			}
 
-			err := cache.Set(component)
+			err := cache.SetComponent(component)
 			require.NoError(b, err)
 		}
 	})
 
 	// Setup test data for the remaining benchmarks
 	setupTestData(b, cache)
 
-	// Run the Children benchmark
-	b.Run("Children", func(b *testing.B) {
+	// Run the ComponentChildren benchmark
+	b.Run("ComponentChildren", func(b *testing.B) {
 		b.ResetTimer()
 		for i := 0; i < b.N; i++ {
-			children, err := cache.Children("root-uid")
+			children, err := cache.ComponentChildren("root-uid")
 			require.NoError(b, err)
 			require.NotEmpty(b, children)
 		}
 	})
 
-	// Run the Delete benchmark
-	b.Run("Delete", func(b *testing.B) {
+	// Run the DeleteComponent benchmark
+	b.Run("DeleteComponent", func(b *testing.B) {
 		b.ResetTimer()
 		for i := 0; i < b.N; i++ {
-			// Delete a level 1 component (which should cascade to its children)
-			err := cache.Delete("level1-uid-" + string(rune('a'+i%10)))
+			// DeleteComponent a level 1 component (which should cascade to its children)
+			err := cache.DeleteComponent("level1-uid-" + string(rune('a'+i%10)))
 			require.NoError(b, err)
 		}
 	})
@@ -299,11 +299,11 @@ func BenchmarkFileCache(b *testing.B) {
 				State:     &state,
 			}
 
-			err := cache.Set(component)
+			err := cache.SetComponent(component)
 			require.NoError(b, err)
 
 			// Retrieve children
-			children, err := cache.Children("root-uid")
+			children, err := cache.ComponentChildren("root-uid")
 			require.NoError(b, err)
 			require.NotEmpty(b, children)
 		}