api/handler/overview_truthful_data_test.go
Rama main · réplica de lectura del repositorio original.
package handler
import (
"context"
"encoding/json"
"errors"
"net/http"
"net/http/httptest"
"sync"
"sync/atomic"
"testing"
"time"
"github.com/goodrain/rainbond/api/model"
rainbondfake "github.com/goodrain/rainbond/pkg/generated/clientset/versioned/fake"
"github.com/shirou/gopsutil/v4/disk"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
corev1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/api/resource"
"k8s.io/client-go/kubernetes"
"k8s.io/client-go/rest"
)
// capability_id: rainbond.overview.lightweight-nodes
func TestOverviewLightweightNodesSkipStats(t *testing.T) {
var statsCalls atomic.Int32
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
if r.URL.Path == "/api/v1/nodes" {
require.NoError(t, json.NewEncoder(w).Encode(&corev1.NodeList{Items: []corev1.Node{overviewResourceNode("worker", true)}}))
return
}
statsCalls.Add(1)
http.Error(w, "unexpected stats request", http.StatusInternalServerError)
}))
defer server.Close()
cli, err := kubernetes.NewForConfig(&rest.Config{Host: server.URL})
require.NoError(t, err)
prom := &recordingPrometheus{}
n := &nodesHandle{clientset: cli, prometheusCli: prom}
got, err := n.ListNodesWithOptions(context.Background(), false)
require.NoError(t, err)
require.Len(t, got, 1)
assert.Equal(t, "worker", got[0].Name)
assert.Equal(t, "not_requested", got[0].DiskStatsStatus)
assert.NotEmpty(t, got[0].Conditions)
assert.Zero(t, statsCalls.Load())
assert.Empty(t, prom.queries)
ctx, cancel := context.WithCancel(context.Background())
cancel()
_, err = n.ListNodesWithOptions(ctx, false)
require.ErrorIs(t, err, context.Canceled)
}
// capability_id: rainbond.overview.disk-availability
func TestOverviewDiskStatsAvailability(t *testing.T) {
tests := []struct {
name, payload, want string
fail bool
}{
{name: "both filesystems", payload: `{"node":{"fs":{"capacityBytes":100,"availableBytes":20},"runtime":{"imageFs":{"capacityBytes":100,"availableBytes":80}}}}`, want: "ok"},
{name: "only root", payload: `{"node":{"fs":{"capacityBytes":100,"availableBytes":20}}}`, want: "partial"},
{name: "no usable filesystems", payload: `{"node":{}}`, want: "unavailable"},
{name: "kubelet failed", fail: true, want: "unavailable"},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
if r.URL.Path == "/api/v1/nodes" {
require.NoError(t, json.NewEncoder(w).Encode(&corev1.NodeList{Items: []corev1.Node{overviewResourceNode("worker", true)}}))
} else if tt.fail {
http.Error(w, "denied", http.StatusForbidden)
} else {
_, _ = w.Write([]byte(tt.payload))
}
}))
defer server.Close()
cli, err := kubernetes.NewForConfig(&rest.Config{Host: server.URL})
require.NoError(t, err)
n := &nodesHandle{clientset: cli, prometheusCli: noopPrometheus{}}
got, err := n.ListNodes(context.Background())
require.NoError(t, err)
require.Len(t, got, 1)
assert.Equal(t, tt.want, got[0].DiskStatsStatus)
})
}
}
// capability_id: rainbond.overview.kubernetes-cluster-resources
func TestOverviewClusterResourcesUseKubernetesNotPrometheus(t *testing.T) {
nodes := []corev1.Node{overviewResourceNode("healthy", true), overviewResourceNode("unhealthy", false), overviewResourceNode("tainted", true)}
nodes[2].Spec.Taints = []corev1.Taint{{Key: "dedicated", Effect: corev1.TaintEffectNoSchedule}}
pods := []corev1.Pod{
overviewResourcePod("running", "healthy", corev1.PodRunning),
overviewResourcePod("pending", "unhealthy", corev1.PodPending),
overviewResourcePod("tainted", "tainted", corev1.PodRunning),
overviewResourcePod("unscheduled", "", corev1.PodPending),
overviewResourcePod("completed", "healthy", corev1.PodSucceeded),
overviewResourcePod("failed", "healthy", corev1.PodFailed),
}
pods[0].Spec.InitContainers = []corev1.Container{{Resources: corev1.ResourceRequirements{Requests: corev1.ResourceList{
corev1.ResourceCPU: resource.MustParse("1"), corev1.ResourceMemory: resource.MustParse("512Mi"),
}}}}
pods[0].Spec.Overhead = corev1.ResourceList{corev1.ResourceCPU: resource.MustParse("100m"), corev1.ResourceMemory: resource.MustParse("64Mi")}
action, calls, closeServer := overviewClusterAction(t, nodes, pods, false)
defer closeServer()
got, err := action.GetClusterInfo(context.Background())
require.NoError(t, err)
assert.Equal(t, "kubernetes", got.ResourceSource)
assert.Equal(t, "rainbond_workloads", got.ResourceScope)
assert.Equal(t, 12, got.CapCPU)
assert.Equal(t, 3*8192, got.CapMem)
assert.InDelta(t, 2.1, got.ReqCPU, .00001)
assert.Equal(t, 1088, got.ReqMem)
assert.InDelta(t, 1.6, got.HealthReqCPU, .00001)
assert.InDelta(t, .5, got.UnhealthReqCPU, .00001)
assert.Equal(t, int32(2), got.NodeReady)
assert.Equal(t, 3, got.RunPodNumber)
assert.Equal(t, 2, got.RunningPodNumber)
require.NotNil(t, got.MaxAllocatableMemoryNodeResource)
assert.Equal(t, int64(8192-576)*1024*1024, got.MaxAllocatableMemoryNodeResource.Memory)
assert.Equal(t, int64(2900), got.MaxAllocatableMemoryNodeResource.MilliCPU)
assert.Equal(t, int32(1), calls.nodes.Load())
assert.Equal(t, int32(1), calls.pods.Load())
assert.Empty(t, action.prometheusCli.(*recordingPrometheus).queries)
_, err = action.GetClusterInfo(context.Background())
require.NoError(t, err)
assert.Equal(t, int32(1), calls.pods.Load(), "cached reads must not relist Pods")
}
func TestOverviewClusterResourcesReportZeroOnlyAfterSuccessfulReads(t *testing.T) {
for _, fail := range []bool{false, true} {
t.Run(map[bool]string{false: "real zero", true: "API error is not zero"}[fail], func(t *testing.T) {
action, _, closeServer := overviewClusterAction(t, nil, nil, fail)
defer closeServer()
got, err := action.GetClusterInfo(context.Background())
if fail {
require.Error(t, err)
assert.Nil(t, got)
assert.Nil(t, action.clusterInfoCache)
return
}
require.NoError(t, err)
assert.Equal(t, "kubernetes", got.ResourceSource)
assert.Zero(t, got.ReqCPU)
assert.Zero(t, got.ReqMem)
assert.Zero(t, got.RunPodNumber)
assert.Zero(t, got.RunningPodNumber)
})
}
}
func TestOverviewClusterResourcesCoalesceConcurrentRefresh(t *testing.T) {
action, calls, closeServer := overviewClusterAction(t, []corev1.Node{overviewResourceNode("worker", true)}, nil, false)
defer closeServer()
var wg sync.WaitGroup
for i := 0; i < 8; i++ {
wg.Add(1)
go func() {
defer wg.Done()
_, err := action.GetClusterInfo(context.Background())
assert.NoError(t, err)
}()
}
wg.Wait()
assert.Equal(t, int32(1), calls.nodes.Load())
assert.Equal(t, int32(1), calls.pods.Load())
action.cacheTime = time.Now().Add(-time.Minute)
_, err := action.GetClusterInfo(context.Background())
require.NoError(t, err)
assert.Equal(t, int32(2), calls.pods.Load())
}
func TestOverviewClusterResourceRequestsIncludeInitSidecarsAndOverhead(t *testing.T) {
pod := overviewResourcePod("resource-accounting", "worker", corev1.PodRunning)
always := corev1.ContainerRestartPolicyAlways
pod.Spec.InitContainers = []corev1.Container{
{RestartPolicy: &always, Resources: corev1.ResourceRequirements{Requests: corev1.ResourceList{corev1.ResourceCPU: resource.MustParse("100m"), corev1.ResourceMemory: resource.MustParse("64Mi")}}},
{Resources: corev1.ResourceRequirements{Requests: corev1.ResourceList{corev1.ResourceCPU: resource.MustParse("2"), corev1.ResourceMemory: resource.MustParse("512Mi")}}},
}
pod.Spec.Overhead = corev1.ResourceList{corev1.ResourceCPU: resource.MustParse("50m"), corev1.ResourceMemory: resource.MustParse("32Mi")}
got := podResourceInformationFromPod(&pod)
assert.Equal(t, int64(2150), got.CPU)
assert.Equal(t, int64(608*1024*1024), got.Memory)
}
func TestOverviewClusterResourcesPreserveFractionalCapacityAndScheduledScope(t *testing.T) {
node := overviewResourceNode("worker", true)
node.Status.Allocatable[corev1.ResourceCPU] = resource.MustParse("7900m")
pods := []corev1.Pod{
overviewResourcePod("active", "worker", corev1.PodUnknown),
overviewResourcePod("not-scheduled", "", corev1.PodRunning),
overviewResourcePod("missing-node", "removed", corev1.PodRunning),
overviewResourcePod("not-rainbond", "worker", corev1.PodRunning),
}
pods[3].Labels = map[string]string{"creator": "other"}
got := buildClusterResourceSnapshot([]*corev1.Node{&node}, pods)
assert.Equal(t, int64(7900), got.CapCPUMillicores)
assert.Equal(t, 8, got.CapCPU, "legacy integer field stays compatible")
assert.Equal(t, 1, got.RunPodNumber)
assert.Zero(t, got.RunningPodNumber)
assert.InDelta(t, .5, got.ReqCPU, .00001)
}
func TestOverviewClusterResourcesIncludeEmptySchedulableNodes(t *testing.T) {
node := overviewResourceNode("empty-worker", true)
got := buildClusterResourceSnapshot([]*corev1.Node{&node}, nil)
require.NotNil(t, got.MaxAllocatableMemoryNodeResource)
assert.Equal(t, int64(8192*1024*1024), got.MaxAllocatableMemoryNodeResource.Memory)
node.Spec.Unschedulable = true
got = buildClusterResourceSnapshot([]*corev1.Node{&node}, nil)
assert.Nil(t, got.MaxAllocatableMemoryNodeResource)
assert.Equal(t, int32(1), got.NodeReady, "cordon does not mean unready")
assert.Equal(t, int64(4000), got.CapCPUMillicores)
}
func TestOverviewClusterResourceRefreshWaitRespectsCancellation(t *testing.T) {
action := &clusterAction{infoRefresh: &clusterInfoRefresh{done: make(chan struct{})}}
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond)
defer cancel()
started := time.Now()
got, err := action.GetClusterInfo(ctx)
assert.Nil(t, got)
require.ErrorIs(t, err, context.DeadlineExceeded)
assert.Less(t, time.Since(started), time.Second)
}
func TestOverviewClusterResourceOptionalVersionCannotBlockResources(t *testing.T) {
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
switch r.URL.Path {
case "/api/v1/nodes":
_, _ = w.Write([]byte(`{"items":[]}`))
case "/api/v1/pods":
_, _ = w.Write([]byte(`{"items":[]}`))
case "/version":
<-r.Context().Done()
default:
http.NotFound(w, r)
}
}))
defer server.Close()
cli, err := kubernetes.NewForConfig(&rest.Config{Host: server.URL})
require.NoError(t, err)
action := &clusterAction{clientset: cli}
started := time.Now()
got, err := action.GetClusterInfo(context.Background())
require.NoError(t, err)
assert.Equal(t, "kubernetes", got.ResourceSource)
assert.Empty(t, got.K8sVersion)
assert.Less(t, time.Since(started), time.Second)
}
// capability_id: rainbond.overview.nonblocking-local-disk
func TestOverviewClusterResourceCannotBeBlockedByLocalDisk(t *testing.T) {
action, calls, closeServer := overviewClusterAction(t, []corev1.Node{overviewResourceNode("worker", true)}, nil, false)
defer closeServer()
started := make(chan struct{})
release := make(chan struct{})
var releaseOnce sync.Once
defer releaseOnce.Do(func() { close(release) })
var diskCalls atomic.Int32
action.diskUsage = func(string) (*disk.UsageStat, error) {
if diskCalls.Add(1) == 1 {
close(started)
}
<-release
return &disk.UsageStat{Total: 1000, Used: 750}, nil
}
type response struct {
value *model.ClusterResource
err error
}
returned := make(chan response, 1)
go func() {
value, err := action.GetClusterInfo(context.Background())
returned <- response{value, err}
}()
var first *model.ClusterResource
select {
case result := <-returned:
require.NoError(t, result.err)
first = result.value
assert.Equal(t, "unavailable", result.value.LocalDiskStatsStatus)
case <-time.After(500 * time.Millisecond):
t.Fatal("local statfs blocked the successful Kubernetes resource response")
}
select {
case <-started:
case <-time.After(time.Second):
t.Fatal("background sample did not start")
}
for i := 0; i < 20; i++ {
got, err := action.GetClusterInfo(context.Background())
require.NoError(t, err)
assert.Equal(t, "kubernetes", got.ResourceSource)
assert.Equal(t, "unavailable", got.LocalDiskStatsStatus)
}
assert.Equal(t, int32(1), diskCalls.Load(), "a hung statfs must not spawn per-request goroutines")
releaseOnce.Do(func() { close(release) })
require.Eventually(t, func() bool {
got, err := action.GetClusterInfo(context.Background())
return err == nil && got.LocalDiskStatsStatus == "ok" && got.CapDisk == 1000 && got.ReqDisk == 750
}, time.Second, 10*time.Millisecond)
assert.Equal(t, int32(1), calls.pods.Load(), "disk refresh must not invalidate Kubernetes resources")
assert.Equal(t, "unavailable", first.LocalDiskStatsStatus, "background completion must not mutate a returned response")
assert.Zero(t, first.CapDisk)
}
func TestOverviewLocalDiskFailureUsesCooldownAndPreservesStaleSample(t *testing.T) {
for _, tc := range []struct {
name string
stale bool
nilNil bool
}{{name: "failure"}, {name: "nil nil result", nilNil: true}, {name: "stale success survives failure", stale: true}} {
t.Run(tc.name, func(t *testing.T) {
var calls atomic.Int32
action := &clusterAction{
clusterInfoCache: &model.ClusterResource{ResourceSource: "kubernetes"},
cacheTime: time.Now(),
diskUsage: func(string) (*disk.UsageStat, error) {
calls.Add(1)
if tc.nilNil {
return nil, nil
}
return nil, errors.New("filesystem unavailable")
},
}
if tc.stale {
action.localDisk = clusterLocalDiskSample{
capacity: 1000, used: 750, available: true,
sampledAt: time.Now().Add(-time.Minute),
}
}
_, err := action.GetClusterInfo(context.Background())
require.NoError(t, err)
require.Eventually(t, func() bool {
action.cacheMu.Lock()
defer action.cacheMu.Unlock()
return action.localDisk.failed && !action.localDisk.refreshing
}, time.Second, 10*time.Millisecond)
for i := 0; i < 20; i++ {
got, err := action.GetClusterInfo(context.Background())
require.NoError(t, err)
if tc.stale {
assert.Equal(t, "stale", got.LocalDiskStatsStatus)
assert.Equal(t, uint64(1000), got.CapDisk)
assert.Equal(t, uint64(750), got.ReqDisk)
} else {
assert.Equal(t, "unavailable", got.LocalDiskStatsStatus)
}
}
assert.Equal(t, int32(1), calls.Load(), "failures must have a 30 second retry cooldown")
action.cacheMu.Lock()
action.localDisk.lastAttempt = time.Now().Add(-time.Minute)
action.cacheMu.Unlock()
_, err = action.GetClusterInfo(context.Background())
require.NoError(t, err)
require.Eventually(t, func() bool { return calls.Load() == 2 }, time.Second, 10*time.Millisecond)
})
}
}
// capability_id: rainbond.overview.empty-installed-plugins
func TestOverviewEmptyInstalledPluginsAreAnArray(t *testing.T) {
action := &clusterAction{rainbondClient: rainbondfake.NewSimpleClientset()}
for _, official := range []bool{false, true} {
got, err := action.ListPlugins(official)
require.NoError(t, err)
encoded, err := json.Marshal(got)
require.NoError(t, err)
assert.Equal(t, "[]", string(encoded))
}
}
type overviewResourceCalls struct{ nodes, pods atomic.Int32 }
func overviewClusterAction(t *testing.T, nodes []corev1.Node, pods []corev1.Pod, fail bool) (*clusterAction, *overviewResourceCalls, func()) {
t.Helper()
calls := &overviewResourceCalls{}
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
switch r.URL.Path {
case "/api/v1/nodes":
calls.nodes.Add(1)
require.NoError(t, json.NewEncoder(w).Encode(&corev1.NodeList{Items: nodes}))
case "/api/v1/pods":
calls.pods.Add(1)
assert.Equal(t, "creator=Rainbond", r.URL.Query().Get("labelSelector"))
if fail {
http.Error(w, "forbidden", http.StatusForbidden)
return
}
require.NoError(t, json.NewEncoder(w).Encode(&corev1.PodList{Items: pods}))
case "/version":
_, _ = w.Write([]byte(`{"gitVersion":"v1.30.4"}`))
default:
http.NotFound(w, r)
}
}))
cli, err := kubernetes.NewForConfig(&rest.Config{Host: server.URL})
require.NoError(t, err)
return &clusterAction{clientset: cli, prometheusCli: &recordingPrometheus{}}, calls, server.Close
}
func overviewResourceNode(name string, ready bool) corev1.Node {
node := newResourceQueryNode(name)
node.Status.Allocatable = node.Status.Capacity.DeepCopy()
node.Status.Allocatable[corev1.ResourcePods] = resource.MustParse("110")
condition := corev1.ConditionFalse
if ready {
condition = corev1.ConditionTrue
}
node.Status.Conditions = []corev1.NodeCondition{{Type: corev1.NodeReady, Status: condition}}
return node
}
func overviewResourcePod(name, node string, phase corev1.PodPhase) corev1.Pod {
pod := newResourceQueryPod(name, node, "service-"+name)
pod.Status.Phase = phase
return pod
}