<commit_before>package operators

import (
	"context"
	"fmt"
	"strconv"
	"time"

	"k8s.io/client-go/kubernetes"

	corev1 "k8s.io/api/core/v1"
	"k8s.io/client-go/discovery"
	"k8s.io/client-go/restmapper"
	"k8s.io/client-go/scale"

	g "github.com/onsi/ginkgo"
	o "github.com/onsi/gomega"
	"github.com/stretchr/objx"

	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/client-go/dynamic"
	e2e "k8s.io/kubernetes/test/e2e/framework"
	e2eskipper "k8s.io/kubernetes/test/e2e/framework/skipper"

	"github.com/openshift/origin/test/extended/util/ibmcloud"
)

const (
	machineAPIGroup       = "machine.openshift.io"
	machineSetOwningLabel = "machine.openshift.io/cluster-api-machineset"
	scalingTime           = 12 * time.Minute
)

// machineSetClient returns a client for machines scoped to the proper namespace
func machineSetClient(dc dynamic.Interface) dynamic.ResourceInterface {
	machineSetClient := dc.Resource(schema.GroupVersionResource{Group: machineAPIGroup, Resource: "machinesets", Version: "v1beta1"})
	return machineSetClient.Namespace(machineAPINamespace)
}

// listWorkerMachineSets list all worker machineSets
func listWorkerMachineSets(dc dynamic.Interface) ([]objx.Map, error) {
	mc := machineSetClient(dc)
	obj, err := mc.List(context.Background(), metav1.ListOptions{})
	if err != nil {
		return nil, err
	}
	machineSets := []objx.Map{}
	for _, ms := range objects(objx.Map(obj.UnstructuredContent()).Get("items")) {
		e2e.Logf("Labels %v", ms.Get("spec.template.metadata.labels"))
		labels := (*ms.Get("spec.template.metadata.labels")).Data().(map[string]interface{})
		if val, ok := labels[machineLabelRole]; ok {
			if val == "worker" {
				machineSets = append(machineSets, ms)
				continue
			}
		}
	}
	return machineSets, nil
}

func getMachineSetReplicaNumber(item objx.Map) int {
	replicas, _ := strconv.Atoi(item.Get("spec.replicas").String())
	return replicas
}

// getNodesFromMachineSet returns an array of nodes backed by machines owned by a given machineSet
func getNodesFromMachineSet(c *kubernetes.Clientset, dc dynamic.Interface, machineSetName string) ([]*corev1.Node, error) {
	machines, err := listMachines(dc, fmt.Sprintf("%s=%s", machineSetOwningLabel, machineSetName))
	if err != nil {
		return nil, fmt.Errorf("failed to list machines: %v", err)
	}

	// fetch nodes
	allWorkerNodes, err := c.CoreV1().Nodes().List(context.Background(), metav1.ListOptions{
		LabelSelector: nodeLabelSelectorWorker,
	})
	if err != nil {
		return nil, fmt.Errorf("failed to list worker nodes: %v", err)
	}

	e2e.Logf("Machines found %v, nodes found: %v", machines, allWorkerNodes.Items)
	machineToNodes, match := mapMachineNameToNodeName(machines, allWorkerNodes.Items)
	if !match {
		return nil, fmt.Errorf("not all machines have a node reference: %v", machineToNodes)
	}
	var nodes []*corev1.Node
	for machineName := range machineToNodes {
		node, err := c.CoreV1().Nodes().Get(context.Background(), machineToNodes[machineName], metav1.GetOptions{})
		if err != nil {
			return nil, fmt.Errorf("failed to get worker nodes %q: %v", machineToNodes[machineName], err)
		}
		nodes = append(nodes, node)
	}

	return nodes, nil
}

func getScaleClient() (scale.ScalesGetter, error) {
	cfg, err := e2e.LoadConfig()
	if err != nil {
		return nil, fmt.Errorf("error getting config: %v", err)
	}

	discoveryClient, err := discovery.NewDiscoveryClientForConfig(cfg)
	if err != nil {
		return nil, fmt.Errorf("error discovering client: %v", err)
	}

	groupResources, err := restmapper.GetAPIGroupResources(discoveryClient)
	if err != nil {
		return nil, fmt.Errorf("error getting API resources: %v", err)
	}
	restMapper := restmapper.NewDiscoveryRESTMapper(groupResources)
	scaleKindResolver := scale.NewDiscoveryScaleKindResolver(discoveryClient)

	scaleClient, err := scale.NewForConfig(cfg, restMapper, dynamic.LegacyAPIPathResolverFunc, scaleKindResolver)
	if err != nil {
		return nil, fmt.Errorf("error creating scale client: %v", err)
	}
	return scaleClient, nil
}

// scaleMachineSet scales a machineSet with a given name to the given number of replicas
func scaleMachineSet(name string, replicas int) error {
	scaleClient, err := getScaleClient()
	if err != nil {
		return fmt.Errorf("error calling getScaleClient: %v", err)
	}

	scale, err := scaleClient.Scales(machineAPINamespace).Get(context.Background(), schema.GroupResource{Group: machineAPIGroup, Resource: "MachineSet"}, name, metav1.GetOptions{})
	if err != nil {
		return fmt.Errorf("error calling scaleClient.Scales get: %v", err)
	}

	scaleUpdate := scale.DeepCopy()
	scaleUpdate.Spec.Replicas = int32(replicas)
	_, err = scaleClient.Scales(machineAPINamespace).Update(context.Background(), schema.GroupResource{Group: machineAPIGroup, Resource: "MachineSet"}, scaleUpdate, metav1.UpdateOptions{})
	if err != nil {
		return fmt.Errorf("error calling scaleClient.Scales update while setting replicas to %d: %v", err, replicas)
	}
	return nil
}

var _ = g.Describe("[sig-cluster-lifecycle][Feature:Machines][Serial] Managed cluster should", func() {
	g.It("grow and decrease when scaling different machineSets simultaneously", func() {
		if e2e.TestContext.Provider == ibmcloud.ProviderName {
			e2eskipper.Skipf("IBM Cloud clusters do not contain machineset resources")
		}

		// expect new nodes to come up for machineSet
		verifyNodeScalingFunc := func(c *kubernetes.Clientset, dc dynamic.Interface, expectedScaleOut int, machineSet objx.Map) bool {
			nodes, err := getNodesFromMachineSet(c, dc, machineName(machineSet))
			if err != nil {
				e2e.Logf("Error getting nodes from machineSet: %v", err)
				return false
			}
			e2e.Logf("node count : %v, expectedCount %v", len(nodes), expectedScaleOut)
			notReady := false
			for i := range nodes {
				e2e.Logf("node: %v", nodes[i].Name)
				if !isNodeReady(*nodes[i]) {
					e2e.Logf("Node %q is not ready", nodes[i].Name)
					notReady = true
				}
			}
			return !notReady && len(nodes) == expectedScaleOut
		}

		cfg, err := e2e.LoadConfig()
		o.Expect(err).NotTo(o.HaveOccurred())
		c, err := e2e.LoadClientset()
		o.Expect(err).NotTo(o.HaveOccurred())
		dc, err := dynamic.NewForConfig(cfg)
		o.Expect(err).NotTo(o.HaveOccurred())

		g.By("fetching worker machineSets")
		machineSets, err := listWorkerMachineSets(dc)
		o.Expect(err).NotTo(o.HaveOccurred())
		if len(machineSets) == 0 {
			e2eskipper.Skipf("Expects at least one worker machineset. Found none!!!")
		}

		g.By("checking initial cluster workers size")
		nodeList, err := c.CoreV1().Nodes().List(context.Background(), metav1.ListOptions{
			LabelSelector: nodeLabelSelectorWorker,
		})
		o.Expect(err).NotTo(o.HaveOccurred())

		initialNumberOfWorkers := len(nodeList.Items)
		g.By(fmt.Sprintf("initial cluster workers size is %v", initialNumberOfWorkers))

		initialReplicasMachineSets := map[string]int{}

		for _, machineSet := range machineSets {
			initialReplicasMachineSet := getMachineSetReplicaNumber(machineSet)
			expectedScaleOut := initialReplicasMachineSet + 1
			initialReplicasMachineSets[machineName(machineSet)] = initialReplicasMachineSet
			g.By(fmt.Sprintf("scaling %q from %d to %d replicas", machineName(machineSet), initialReplicasMachineSet, expectedScaleOut))
			err = scaleMachineSet(machineName(machineSet), expectedScaleOut)
			o.Expect(err).NotTo(o.HaveOccurred())
		}

		g.By("checking scaled up worker nodes are ready")
		for _, machineSet := range machineSets {
			expectedScaleOut := initialReplicasMachineSets[machineName(machineSet)] + 1
			o.Eventually(func() bool {
				return verifyNodeScalingFunc(c, dc, expectedScaleOut, machineSet)
			}, scalingTime, 5*time.Second).Should(o.BeTrue())
		}

		for _, machineSet := range machineSets {
			scaledReplicasMachineSet := initialReplicasMachineSets[machineName(machineSet)] + 1
			g.By(fmt.Sprintf("scaling %q from %d to %d replicas", machineName(machineSet), scaledReplicasMachineSet, initialReplicasMachineSets[machineName(machineSet)]))
			err = scaleMachineSet(machineName(machineSet), initialReplicasMachineSets[machineName(machineSet)])
			o.Expect(err).NotTo(o.HaveOccurred())
		}

		g.By(fmt.Sprintf("waiting for cluster to get back to original size. Final size should be %d worker nodes", initialNumberOfWorkers))
		o.Eventually(func() bool {
			nodeList, err := c.CoreV1().Nodes().List(context.Background(), metav1.ListOptions{
				LabelSelector: nodeLabelSelectorWorker,
			})
			o.Expect(err).NotTo(o.HaveOccurred())
			g.By(fmt.Sprintf("got %v nodes, expecting %v", len(nodeList.Items), initialNumberOfWorkers))
			return len(nodeList.Items) == initialNumberOfWorkers
			// Azure actuator takes something over 3 minutes to delete a machine.
			// The worst observable case to delete a machine was 5m15s however.
			// Also, there are two instances to be deleted.
			// Rounding to 10 minutes to accommodate for future new and slower cloud providers.
			// https://bugzilla.redhat.com/show_bug.cgi?id=1812240
		}, 10*time.Minute, 5*time.Second).Should(o.BeTrue())
	})
})<commit_msg>make the machine scale test wait until nodes are ready to run workloads<commit_after>package operators

import (
	"context"
	"fmt"
	"strconv"
	"time"

	g "github.com/onsi/ginkgo"
	o "github.com/onsi/gomega"
	"github.com/openshift/origin/test/extended/util/ibmcloud"
	"github.com/stretchr/objx"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/client-go/discovery"
	"k8s.io/client-go/dynamic"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/restmapper"
	"k8s.io/client-go/scale"
	e2e "k8s.io/kubernetes/test/e2e/framework"
	e2eskipper "k8s.io/kubernetes/test/e2e/framework/skipper"
)

const (
	machineAPIGroup       = "machine.openshift.io"
	machineSetOwningLabel = "machine.openshift.io/cluster-api-machineset"
	scalingTime           = 12 * time.Minute
)

// machineSetClient returns a client for machines scoped to the proper namespace
func machineSetClient(dc dynamic.Interface) dynamic.ResourceInterface {
	machineSetClient := dc.Resource(schema.GroupVersionResource{Group: machineAPIGroup, Resource: "machinesets", Version: "v1beta1"})
	return machineSetClient.Namespace(machineAPINamespace)
}

// listWorkerMachineSets list all worker machineSets
func listWorkerMachineSets(dc dynamic.Interface) ([]objx.Map, error) {
	mc := machineSetClient(dc)
	obj, err := mc.List(context.Background(), metav1.ListOptions{})
	if err != nil {
		return nil, err
	}
	machineSets := []objx.Map{}
	for _, ms := range objects(objx.Map(obj.UnstructuredContent()).Get("items")) {
		e2e.Logf("Labels %v", ms.Get("spec.template.metadata.labels"))
		labels := (*ms.Get("spec.template.metadata.labels")).Data().(map[string]interface{})
		if val, ok := labels[machineLabelRole]; ok {
			if val == "worker" {
				machineSets = append(machineSets, ms)
				continue
			}
		}
	}
	return machineSets, nil
}

func getMachineSetReplicaNumber(item objx.Map) int {
	replicas, _ := strconv.Atoi(item.Get("spec.replicas").String())
	return replicas
}

// getNodesFromMachineSet returns an array of nodes backed by machines owned by a given machineSet
func getNodesFromMachineSet(c *kubernetes.Clientset, dc dynamic.Interface, machineSetName string) ([]*corev1.Node, error) {
	machines, err := listMachines(dc, fmt.Sprintf("%s=%s", machineSetOwningLabel, machineSetName))
	if err != nil {
		return nil, fmt.Errorf("failed to list machines: %v", err)
	}

	// fetch nodes
	allWorkerNodes, err := c.CoreV1().Nodes().List(context.Background(), metav1.ListOptions{
		LabelSelector: nodeLabelSelectorWorker,
	})
	if err != nil {
		return nil, fmt.Errorf("failed to list worker nodes: %v", err)
	}

	e2e.Logf("Machines found %v, nodes found: %v", machines, allWorkerNodes.Items)
	machineToNodes, match := mapMachineNameToNodeName(machines, allWorkerNodes.Items)
	if !match {
		return nil, fmt.Errorf("not all machines have a node reference: %v", machineToNodes)
	}
	var nodes []*corev1.Node
	for machineName := range machineToNodes {
		node, err := c.CoreV1().Nodes().Get(context.Background(), machineToNodes[machineName], metav1.GetOptions{})
		if err != nil {
			return nil, fmt.Errorf("failed to get worker nodes %q: %v", machineToNodes[machineName], err)
		}
		nodes = append(nodes, node)
	}

	return nodes, nil
}

func getScaleClient() (scale.ScalesGetter, error) {
	cfg, err := e2e.LoadConfig()
	if err != nil {
		return nil, fmt.Errorf("error getting config: %v", err)
	}

	discoveryClient, err := discovery.NewDiscoveryClientForConfig(cfg)
	if err != nil {
		return nil, fmt.Errorf("error discovering client: %v", err)
	}

	groupResources, err := restmapper.GetAPIGroupResources(discoveryClient)
	if err != nil {
		return nil, fmt.Errorf("error getting API resources: %v", err)
	}
	restMapper := restmapper.NewDiscoveryRESTMapper(groupResources)
	scaleKindResolver := scale.NewDiscoveryScaleKindResolver(discoveryClient)

	scaleClient, err := scale.NewForConfig(cfg, restMapper, dynamic.LegacyAPIPathResolverFunc, scaleKindResolver)
	if err != nil {
		return nil, fmt.Errorf("error creating scale client: %v", err)
	}
	return scaleClient, nil
}

// scaleMachineSet scales a machineSet with a given name to the given number of replicas
func scaleMachineSet(name string, replicas int) error {
	scaleClient, err := getScaleClient()
	if err != nil {
		return fmt.Errorf("error calling getScaleClient: %v", err)
	}

	scale, err := scaleClient.Scales(machineAPINamespace).Get(context.Background(), schema.GroupResource{Group: machineAPIGroup, Resource: "MachineSet"}, name, metav1.GetOptions{})
	if err != nil {
		return fmt.Errorf("error calling scaleClient.Scales get: %v", err)
	}

	scaleUpdate := scale.DeepCopy()
	scaleUpdate.Spec.Replicas = int32(replicas)
	_, err = scaleClient.Scales(machineAPINamespace).Update(context.Background(), schema.GroupResource{Group: machineAPIGroup, Resource: "MachineSet"}, scaleUpdate, metav1.UpdateOptions{})
	if err != nil {
		return fmt.Errorf("error calling scaleClient.Scales update while setting replicas to %d: %v", err, replicas)
	}
	return nil
}

var _ = g.Describe("[sig-cluster-lifecycle][Feature:Machines][Serial] Managed cluster should", func() {
	g.It("grow and decrease when scaling different machineSets simultaneously", func() {
		if e2e.TestContext.Provider == ibmcloud.ProviderName {
			e2eskipper.Skipf("IBM Cloud clusters do not contain machineset resources")
		}

		// expect new nodes to come up for machineSet
		verifyNodeScalingFunc := func(c *kubernetes.Clientset, dc dynamic.Interface, expectedScaleOut int, machineSet objx.Map) bool {
			nodes, err := getNodesFromMachineSet(c, dc, machineName(machineSet))
			if err != nil {
				e2e.Logf("Error getting nodes from machineSet: %v", err)
				return false
			}
			e2e.Logf("node count : %v, expectedCount %v", len(nodes), expectedScaleOut)
			notReady := false
			for i := range nodes {
				e2e.Logf("node: %v", nodes[i].Name)
				if !isNodeReady(*nodes[i]) {
					e2e.Logf("Node %q is not ready", nodes[i].Name)
					notReady = true
				}
			}
			return !notReady && len(nodes) == expectedScaleOut
		}

		cfg, err := e2e.LoadConfig()
		o.Expect(err).NotTo(o.HaveOccurred())
		c, err := e2e.LoadClientset()
		o.Expect(err).NotTo(o.HaveOccurred())
		dc, err := dynamic.NewForConfig(cfg)
		o.Expect(err).NotTo(o.HaveOccurred())

		g.By("fetching worker machineSets")
		machineSets, err := listWorkerMachineSets(dc)
		o.Expect(err).NotTo(o.HaveOccurred())
		if len(machineSets) == 0 {
			e2eskipper.Skipf("Expects at least one worker machineset. Found none!!!")
		}

		g.By("checking initial cluster workers size")
		nodeList, err := c.CoreV1().Nodes().List(context.Background(), metav1.ListOptions{
			LabelSelector: nodeLabelSelectorWorker,
		})
		o.Expect(err).NotTo(o.HaveOccurred())

		initialNumberOfWorkers := len(nodeList.Items)
		g.By(fmt.Sprintf("initial cluster workers size is %v", initialNumberOfWorkers))

		initialReplicasMachineSets := map[string]int{}

		for _, machineSet := range machineSets {
			initialReplicasMachineSet := getMachineSetReplicaNumber(machineSet)
			expectedScaleOut := initialReplicasMachineSet + 1
			initialReplicasMachineSets[machineName(machineSet)] = initialReplicasMachineSet
			g.By(fmt.Sprintf("scaling %q from %d to %d replicas", machineName(machineSet), initialReplicasMachineSet, expectedScaleOut))
			err = scaleMachineSet(machineName(machineSet), expectedScaleOut)
			o.Expect(err).NotTo(o.HaveOccurred())
		}

		g.By("checking scaled up worker nodes are ready")
		for _, machineSet := range machineSets {
			expectedScaleOut := initialReplicasMachineSets[machineName(machineSet)] + 1
			o.Eventually(func() bool {
				return verifyNodeScalingFunc(c, dc, expectedScaleOut, machineSet)
			}, scalingTime, 5*time.Second).Should(o.BeTrue())
		}

		for _, machineSet := range machineSets {
			scaledReplicasMachineSet := initialReplicasMachineSets[machineName(machineSet)] + 1
			g.By(fmt.Sprintf("scaling %q from %d to %d replicas", machineName(machineSet), scaledReplicasMachineSet, initialReplicasMachineSets[machineName(machineSet)]))
			err = scaleMachineSet(machineName(machineSet), initialReplicasMachineSets[machineName(machineSet)])
			o.Expect(err).NotTo(o.HaveOccurred())
		}

		g.By(fmt.Sprintf("waiting for cluster to get back to original size. Final size should be %d worker nodes", initialNumberOfWorkers))
		o.Eventually(func() bool {
			nodeList, err := c.CoreV1().Nodes().List(context.Background(), metav1.ListOptions{
				LabelSelector: nodeLabelSelectorWorker,
			})
			o.Expect(err).NotTo(o.HaveOccurred())
			g.By(fmt.Sprintf("got %v nodes, expecting %v", len(nodeList.Items), initialNumberOfWorkers))
			if len(nodeList.Items) != initialNumberOfWorkers {
				return false
			}

			g.By(fmt.Sprintf("ensure worker taints and conditions are consistent"))
			for _, node := range nodeList.Items {
				if _, ok := node.Labels["node-role.kubernetes.io/worker"]; !ok {
					continue
				}
				if len(node.Spec.Taints) > 0 {
					e2e.Logf("node/%s had unexpected taints: %#v", node.Name, node.Spec.Taints)
					return false
				}
				for _, condition := range node.Status.Conditions {
					switch condition.Type {
					case corev1.NodeReady:
						if condition.Status != corev1.ConditionTrue {
							e2e.Logf("node/%s had unexpected condition %q == %v: %#v", node.Name, condition.Reason, condition.Status)
							return false
						}
					case corev1.NodeMemoryPressure,
						corev1.NodeDiskPressure,
						corev1.NodePIDPressure,
						corev1.NodeNetworkUnavailable:
						if condition.Status != corev1.ConditionFalse {
							e2e.Logf("node/%s had unexpected condition %q == %v: %#v", node.Name, condition.Reason, condition.Status)
							return false
						}

					default:
						e2e.Logf("node/%s had unhandled condition %q == %v: %#v", node.Name, condition.Reason, condition.Status)

					}
				}
				e2e.Logf("node/%s taints and conditions seem ok")
			}

			return true
			// Azure actuator takes something over 3 minutes to delete a machine.
			// The worst observable case to delete a machine was 5m15s however.
			// Also, there are two instances to be deleted.
			// Rounding to 10 minutes to accommodate for future new and slower cloud providers.
			// https://bugzilla.redhat.com/show_bug.cgi?id=1812240
		}, 10*time.Minute, 5*time.Second).Should(o.BeTrue())
	})
})
{'fraction_non_alphanumeric': 0.1296900067277338, 'fraction_numerical': 0.007918025151374011, 'mean_word_length': 4.782166367444644, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 4, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '949022', 'n_tokens_mistral': 6888, 'n_tokens_neox': 6269, 'n_words': 1456}