internal/resources/metricfallback_test.go

Rama main · réplica de lectura del repositorio original.

package resources

import (
	"context"
	"errors"
	"testing"
	"time"

	authv1 "k8s.io/api/authorization/v1"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/client-go/dynamic/fake"
	k8sfake "k8s.io/client-go/kubernetes/fake"
	k8stesting "k8s.io/client-go/testing"

	"github.com/sophotechlabs/spinoza/internal/access"
	"github.com/sophotechlabs/spinoza/internal/auth"
	"github.com/sophotechlabs/spinoza/internal/prom"
)

func prometheusService() *corev1.Service {
	return &corev1.Service{
		Namespace: "monitoring",
		Name:      "prometheus-operated",
		Labels:    map[string]string{"operated-prometheus": "true"},
		Spec:      corev1.ServiceSpec{Ports: []corev1.ServicePort{{Name: "http-web", Port: 9090}}},
	}
}

func podMetric(namespace, name, cpu, memory string) *unstructured.Unstructured {
	return &unstructured.Unstructured{Object: map[string]any{
		"apiVersion": "metrics.k8s.io/v1beta1",
		"kind":       "PodMetrics",
		"metadata":   map[string]any{"namespace": namespace, "name": name},
		"containers": []any{map[string]any{
			"name":  "app",
			"usage": map[string]any{"cpu": cpu, "memory": memory},
		}},
	}}
}

func measuredCluster(t *testing.T) *fake.FakeDynamicClient {
	t.Helper()
	dyn := fake.NewSimpleDynamicClientWithCustomListKinds(runtime.NewScheme(), metricsKinds())
	seedNodeMetric(t, dyn, "n1")
	gvr := schema.GroupVersionResource{Group: "metrics.k8s.io", Version: "v1beta1", Resource: "pods"}
	_, err := dyn.Resource(gvr).Namespace("prod").
		Create(context.Background(), podMetric("prod", "web", "250m", "512Mi"), metav1.CreateOptions{})
	if err != nil {
		t.Fatalf("seed pod metrics: %v", err)
	}
	return dyn
}

func TestMetricHistoryIsMeasuredHereWhenThereIsNoPrometheus(t *testing.T) {
	mgr := NewManager(t.Context(), Deps{Dynamic: measuredCluster(t), Clientset: k8sfake.NewClientset()})

	history, err := mgr.MetricHistory(context.Background(), "prod", "web", time.Hour)
	if err != nil {
		t.Fatalf("history: %v", err)
	}
	if !history.Sampled {
		t.Fatal("the answer did not say spinoza measured it")
	}
	if len(history.CPU) != 1 {
		t.Fatalf("cpu points = %d, want the one reading taken on the way", len(history.CPU))
	}
	if history.CPU[0].Value != 0.25 {
		t.Fatalf("cpu = %v cores, want the 250m the cluster reported", history.CPU[0].Value)
	}
}

func TestAskingForAChartIsWhatTakesTheFirstReading(t *testing.T) {
	mgr := NewManager(t.Context(), Deps{Dynamic: measuredCluster(t), Clientset: k8sfake.NewClientset()})

	history, err := mgr.MetricHistory(context.Background(), "prod", "web", time.Hour)
	if err != nil {
		t.Fatalf("history: %v", err)
	}
	if len(history.CPU) == 0 {
		t.Fatal("nothing was measured, so nobody else had asked for metrics first")
	}
}

func metricsDeniedTo(t *testing.T) *k8sfake.Clientset {
	t.Helper()
	cs := k8sfake.NewClientset()
	cs.PrependReactor("create", "selfsubjectaccessreviews", func(action k8stesting.Action) (bool, runtime.Object, error) {
		create, ok := action.(k8stesting.CreateAction)
		if !ok {
			return false, nil, nil
		}
		review, ok := create.GetObject().(*authv1.SelfSubjectAccessReview)
		if !ok {
			return false, nil, nil
		}
		attributes := review.Spec.ResourceAttributes
		review.Status.Allowed = attributes.Group != "metrics.k8s.io"
		if !review.Status.Allowed {
			review.Status.Reason = "no metrics for you"
		}
		return true, review, nil
	})
	return cs
}

func TestAReaderDeniedMetricsIsNotServedSomebodyElsesSamples(t *testing.T) {
	dyn := measuredCluster(t)
	mgr := NewManager(t.Context(), Deps{Dynamic: dyn, Clientset: metricsDeniedTo(t)})
	primed, err := mgr.MetricHistory(t.Context(), "prod", "web", time.Hour)
	if err != nil || len(primed.CPU) != 1 {
		t.Fatalf("priming as the server: history = %+v, error = %v", primed, err)
	}
	reader := auth.WithIdentity(t.Context(), auth.Identity{User: "reader", Role: auth.RoleViewer})

	after, err := mgr.MetricHistory(reader, "prod", "web", time.Hour)

	if !errors.Is(err, access.ErrDenied) {
		t.Fatalf("error = %v, want the reader's own denial", err)
	}
	if err.Error() != "kubernetes authorization denied: no metrics for you" {
		t.Fatalf("error = %q", err.Error())
	}
	if len(after.CPU) != 0 {
		t.Fatalf("the denied reader still received %d cached samples", len(after.CPU))
	}
}

func TestAReaderAllowedMetricsStillGetsTheSampledHistory(t *testing.T) {
	dyn := measuredCluster(t)
	cs := k8sfake.NewClientset()
	cs.PrependReactor("create", "selfsubjectaccessreviews", func(action k8stesting.Action) (bool, runtime.Object, error) {
		create, ok := action.(k8stesting.CreateAction)
		if !ok {
			return false, nil, nil
		}
		review, ok := create.GetObject().(*authv1.SelfSubjectAccessReview)
		if !ok {
			return false, nil, nil
		}
		review.Status.Allowed = true
		return true, review, nil
	})
	mgr := NewManager(t.Context(), Deps{Dynamic: dyn, Clientset: cs})
	if _, err := mgr.MetricHistory(t.Context(), "prod", "web", time.Hour); err != nil {
		t.Fatalf("priming as the server: %v", err)
	}
	reader := auth.WithIdentity(t.Context(), auth.Identity{User: "reader", Role: auth.RoleViewer})

	history, err := mgr.MetricHistory(reader, "prod", "web", time.Hour)
	if err != nil {
		t.Fatalf("history: %v", err)
	}
	if !history.Sampled || len(history.CPU) != 1 {
		t.Fatalf("history = %+v, want the sampled series an authorized reader may see", history)
	}
}

func TestTheMetricsGateAsksAboutTheNamespaceThatWasRequested(t *testing.T) {
	var asked []authv1.ResourceAttributes
	cs := k8sfake.NewClientset()
	cs.PrependReactor("create", "selfsubjectaccessreviews", func(action k8stesting.Action) (bool, runtime.Object, error) {
		create, ok := action.(k8stesting.CreateAction)
		if !ok {
			return false, nil, nil
		}
		review, ok := create.GetObject().(*authv1.SelfSubjectAccessReview)
		if !ok {
			return false, nil, nil
		}
		asked = append(asked, *review.Spec.ResourceAttributes)
		review.Status.Allowed = true
		return true, review, nil
	})
	mgr := NewManager(t.Context(), Deps{Dynamic: measuredCluster(t), Clientset: cs})
	reader := auth.WithIdentity(t.Context(), auth.Identity{User: "reader", Role: auth.RoleViewer})

	if _, err := mgr.MetricHistory(reader, "prod", "web", time.Hour); err != nil {
		t.Fatalf("history: %v", err)
	}

	want := authv1.ResourceAttributes{Namespace: "prod", Verb: "list", Group: "metrics.k8s.io", Resource: "pods"}
	found := false
	for _, one := range asked {
		if one == want {
			found = true
		}
	}
	if !found {
		t.Fatalf("questions = %+v, want %+v among them", asked, want)
	}
}

type refusingProxy struct{}

func (*refusingProxy) Get(context.Context, prom.Target, string, map[string]string) ([]byte, error) {
	return nil, errors.New("dial tcp: connection refused")
}

func TestMetricHistoryFallsBackWhenPrometheusCannotBeReached(t *testing.T) {
	cs := k8sfake.NewClientset()
	mgr := NewManager(t.Context(), Deps{
		Dynamic:    measuredCluster(t),
		Clientset:  cs,
		Prometheus: prom.NewClientWithProxy(cs, &refusingProxy{}, prom.Target{}),
	})

	history, err := mgr.MetricHistory(context.Background(), "prod", "web", time.Hour)
	if err != nil {
		t.Fatalf("history: %v", err)
	}
	if !history.Sampled {
		t.Fatal("a prometheus that could not be reached left the chart empty")
	}
}

type brokenProxy struct{}

func (*brokenProxy) Get(context.Context, prom.Target, string, map[string]string) ([]byte, error) {
	return []byte(`{"status":"error","errorType":"bad_data","error":"parse error"}`), nil
}

func TestMetricHistoryPassesOnAnErrorThatIsNotAMissingPrometheus(t *testing.T) {
	cs := k8sfake.NewClientset(prometheusService())
	mgr := NewManager(t.Context(), Deps{
		Dynamic:    measuredCluster(t),
		Clientset:  cs,
		Prometheus: prom.NewClientWithProxy(cs, &brokenProxy{}, prom.Target{}),
	})

	_, err := mgr.MetricHistory(context.Background(), "prod", "web", time.Hour)

	if err == nil {
		t.Fatal("an error from prometheus itself was swallowed")
	}
	if errors.Is(err, prom.ErrUnavailable) {
		t.Fatalf("error = %v, want the one prometheus gave rather than a missing one", err)
	}
}

func TestMetricHistoryRefusesAnotherNamespace(t *testing.T) {
	mgr := NewManager(t.Context(), Deps{
		Clientset: decidingClientset(false, "not for you"),
	})
	ctx := auth.WithIdentity(t.Context(), auth.Identity{User: "alice"})

	_, err := mgr.MetricHistory(ctx, "prod", "web", time.Hour)

	if !errors.Is(err, ErrOutOfScope) {
		t.Fatalf("error = %v, want the namespace refused", err)
	}
}

func TestMetricsInRefusesAnotherNamespace(t *testing.T) {
	mgr := NewManager(t.Context(), Deps{
		Clientset: decidingClientset(false, "not for you"),
	})
	ctx := auth.WithIdentity(t.Context(), auth.Identity{User: "alice"})

	_, err := mgr.MetricsIn(ctx, "prod")

	if !errors.Is(err, ErrOutOfScope) {
		t.Fatalf("error = %v, want the namespace refused", err)
	}
}