metrics-server/pkg/provider/sink/sinkprov_test.go

// Copyright 2018 The Kubernetes Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package sink_test

import (
	"testing"
	"time"

	. "github.com/onsi/ginkgo"
	. "github.com/onsi/gomega"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/resource"
	apitypes "k8s.io/apimachinery/pkg/types"
	metrics "k8s.io/metrics/pkg/apis/metrics"

	"github.com/kubernetes-incubator/metrics-server/pkg/provider"
	. "github.com/kubernetes-incubator/metrics-server/pkg/provider/sink"
	"github.com/kubernetes-incubator/metrics-server/pkg/sink"
	"github.com/kubernetes-incubator/metrics-server/pkg/sources"
)

var defaultWindow = 30 * time.Second

func TestSourceManager(t *testing.T) {
	RegisterFailHandler(Fail)
	RunSpecs(t, "Provider/Sink Suite")
}

func newMilliPoint(ts time.Time, cpu, memory int64) sources.MetricsPoint {
	return sources.MetricsPoint{
		Timestamp:   ts,
		CpuUsage:    *resource.NewMilliQuantity(cpu, resource.DecimalSI),
		MemoryUsage: *resource.NewMilliQuantity(memory, resource.BinarySI),
	}
}

var _ = Describe("In-memory Sink Provider", func() {
	var (
		batch    *sources.MetricsBatch
		prov     provider.MetricsProvider
		provSink sink.MetricSink
		now      time.Time
	)

	BeforeEach(func() {
		now = time.Now()
		batch = &sources.MetricsBatch{
			Nodes: []sources.NodeMetricsPoint{
				{Name: "node1", MetricsPoint: newMilliPoint(now.Add(100*time.Millisecond), 110, 120)},
				{Name: "node2", MetricsPoint: newMilliPoint(now.Add(200*time.Millisecond), 210, 220)},
				{Name: "node3", MetricsPoint: newMilliPoint(now.Add(300*time.Millisecond), 310, 320)},
			},
			Pods: []sources.PodMetricsPoint{
				{Name: "pod1", Namespace: "ns1", Containers: []sources.ContainerMetricsPoint{
					{Name: "container1", MetricsPoint: newMilliPoint(now.Add(400*time.Millisecond), 410, 420)},
					{Name: "container2", MetricsPoint: newMilliPoint(now.Add(500*time.Millisecond), 510, 520)},
				}},
				{Name: "pod2", Namespace: "ns1", Containers: []sources.ContainerMetricsPoint{
					{Name: "container1", MetricsPoint: newMilliPoint(now.Add(600*time.Millisecond), 610, 620)},
				}},
				{Name: "pod1", Namespace: "ns2", Containers: []sources.ContainerMetricsPoint{
					{Name: "container1", MetricsPoint: newMilliPoint(now.Add(700*time.Millisecond), 710, 720)},
					{Name: "container2", MetricsPoint: newMilliPoint(now.Add(800*time.Millisecond), 810, 820)},
				}},
			},
		}

		provSink, prov = NewSinkProvider()
	})

	It("should receive batches of metrics", func() {
		By("sending the batch to the sink")
		Expect(provSink.Receive(batch)).To(Succeed())

		By("making sure that the provider contains all nodes received")
		for _, node := range batch.Nodes {
			_, _, err := prov.GetNodeMetrics(node.Name)
			Expect(err).NotTo(HaveOccurred())
		}

		By("making sure that the provider contains all pods received")
		for _, pod := range batch.Pods {
			_, _, err := prov.GetContainerMetrics(apitypes.NamespacedName{
				Name:      pod.Name,
				Namespace: pod.Namespace,
			})
			Expect(err).NotTo(HaveOccurred())
		}
	})

	It("should error out if duplicate nodes were received, without a partial store", func() {
		By("adding a duplicate node to the batch")
		batch.Nodes = append(batch.Nodes, batch.Nodes[0])

		By("sending the batch to the sink and checking for an error")
		Expect(provSink.Receive(batch)).NotTo(Succeed())

		By("making sure none of the data is in the sink")
		for _, node := range batch.Nodes {
			_, res, err := prov.GetNodeMetrics(node.Name)
			Expect(err).NotTo(HaveOccurred())
			Expect(res).To(ConsistOf(corev1.ResourceList(nil)))
		}
		for _, pod := range batch.Pods {
			_, res, err := prov.GetContainerMetrics(apitypes.NamespacedName{
				Name:      pod.Name,
				Namespace: pod.Namespace,
			})
			Expect(err).NotTo(HaveOccurred())
			Expect(res).To(Equal([][]metrics.ContainerMetrics{nil}))
		}
	})

	It("should error out if duplicate pods were received, without a partial store", func() {
		By("adding a duplicate pod to the batch")
		batch.Pods = append(batch.Pods, batch.Pods[0])

		By("sending the batch to the sink and checking for an error")
		Expect(provSink.Receive(batch)).NotTo(Succeed())

		By("making sure none of the data is in the sink")
		for _, node := range batch.Nodes {
			_, res, err := prov.GetNodeMetrics(node.Name)
			Expect(err).NotTo(HaveOccurred())
			Expect(res).To(ConsistOf(corev1.ResourceList(nil)))
		}
		for _, pod := range batch.Pods {
			_, res, err := prov.GetContainerMetrics(apitypes.NamespacedName{
				Name:      pod.Name,
				Namespace: pod.Namespace,
			})
			Expect(err).NotTo(HaveOccurred())
			Expect(res).To(Equal([][]metrics.ContainerMetrics{nil}))
		}
	})

	It("should retrieve metrics for all containers in a pod, with overall latest scrape time", func() {
		By("sending the batch to the sink")
		Expect(provSink.Receive(batch)).To(Succeed())

		By("fetching the pod")
		ts, containerMetrics, err := prov.GetContainerMetrics(apitypes.NamespacedName{
			Name:      "pod1",
			Namespace: "ns1",
		})
		Expect(err).NotTo(HaveOccurred())

		By("verifying that the timestamp is the smallest time amongst all containers")
		Expect(ts).To(ConsistOf(provider.TimeInfo{Timestamp: now.Add(400 * time.Millisecond), Window: defaultWindow}))

		By("verifying that all containers have data")
		Expect(containerMetrics).To(Equal(
			[][]metrics.ContainerMetrics{
				{
					{
						Name: "container1",
						Usage: corev1.ResourceList{
							corev1.ResourceCPU:    *resource.NewMilliQuantity(410, resource.DecimalSI),
							corev1.ResourceMemory: *resource.NewMilliQuantity(420, resource.BinarySI),
						},
					},
					{
						Name: "container2",
						Usage: corev1.ResourceList{
							corev1.ResourceCPU:    *resource.NewMilliQuantity(510, resource.DecimalSI),
							corev1.ResourceMemory: *resource.NewMilliQuantity(520, resource.BinarySI),
						},
					},
				},
			},
		))
	})

	It("should return nil metrics for missing pods", func() {
		By("sending the batch to the sink")
		Expect(provSink.Receive(batch)).To(Succeed())

		By("fetching the a present pod and a missing pod")
		ts, containerMetrics, err := prov.GetContainerMetrics(apitypes.NamespacedName{
			Name:      "pod1",
			Namespace: "ns1",
		}, apitypes.NamespacedName{
			Name:      "pod2",
			Namespace: "ns42",
		})
		Expect(err).NotTo(HaveOccurred())

		By("verifying that the timestamp is the smallest time amongst all containers")
		Expect(ts).To(Equal([]provider.TimeInfo{{Timestamp: now.Add(400 * time.Millisecond), Window: defaultWindow}, {}}))

		By("verifying that all present containers have data")
		Expect(containerMetrics).To(Equal(
			[][]metrics.ContainerMetrics{
				{
					{
						Name: "container1",
						Usage: corev1.ResourceList{
							corev1.ResourceCPU:    *resource.NewMilliQuantity(410, resource.DecimalSI),
							corev1.ResourceMemory: *resource.NewMilliQuantity(420, resource.BinarySI),
						},
					},
					{
						Name: "container2",
						Usage: corev1.ResourceList{
							corev1.ResourceCPU:    *resource.NewMilliQuantity(510, resource.DecimalSI),
							corev1.ResourceMemory: *resource.NewMilliQuantity(520, resource.BinarySI),
						},
					},
				},
				nil,
			},
		))

	})

	It("should retrieve metrics for a node, with overall latest scrape time", func() {
		By("sending the batch to the sink")
		Expect(provSink.Receive(batch)).To(Succeed())

		By("fetching the nodes")
		ts, nodeMetrics, err := prov.GetNodeMetrics("node1", "node2")
		Expect(err).NotTo(HaveOccurred())

		By("verifying that the timestamp is the smallest time amongst all containers")
		Expect(ts).To(Equal([]provider.TimeInfo{{Timestamp: now.Add(100 * time.Millisecond), Window: defaultWindow}, {Timestamp: now.Add(200 * time.Millisecond), Window: defaultWindow}}))

		By("verifying that all nodes have data")
		Expect(nodeMetrics).To(Equal(
			[]corev1.ResourceList{
				{
					corev1.ResourceCPU:    *resource.NewMilliQuantity(110, resource.DecimalSI),
					corev1.ResourceMemory: *resource.NewMilliQuantity(120, resource.BinarySI),
				},
				{
					corev1.ResourceCPU:    *resource.NewMilliQuantity(210, resource.DecimalSI),
					corev1.ResourceMemory: *resource.NewMilliQuantity(220, resource.BinarySI),
				},
			},
		))
	})

	It("should return nil metrics for missing nodes", func() {
		By("sending the batch to the sink")
		Expect(provSink.Receive(batch)).To(Succeed())

		By("fetching the nodes, plus a missing node")
		ts, nodeMetrics, err := prov.GetNodeMetrics("node1", "node2", "node42")
		Expect(err).NotTo(HaveOccurred())

		By("verifying that the timestamp is the smallest time amongst all containers")
		Expect(ts).To(Equal([]provider.TimeInfo{{Timestamp: now.Add(100 * time.Millisecond), Window: defaultWindow}, {Timestamp: now.Add(200 * time.Millisecond), Window: defaultWindow}, {}}))

		By("verifying that all present nodes have data")
		Expect(nodeMetrics).To(Equal(
			[]corev1.ResourceList{
				{
					corev1.ResourceCPU:    *resource.NewMilliQuantity(110, resource.DecimalSI),
					corev1.ResourceMemory: *resource.NewMilliQuantity(120, resource.BinarySI),
				},
				{
					corev1.ResourceCPU:    *resource.NewMilliQuantity(210, resource.DecimalSI),
					corev1.ResourceMemory: *resource.NewMilliQuantity(220, resource.BinarySI),
				},
				nil,
			},
		))

	})
})