Merge pull request kubernetes#2127 from mborsz/params

Add unit tests to util.ToStruct

clusterloader2/pkg/execservice/exec_service.go

@@ -34,6 +34,7 @@ import (
 	"k8s.io/perf-tests/clusterloader2/pkg/framework"
 	"k8s.io/perf-tests/clusterloader2/pkg/framework/client"
 	measurementutil "k8s.io/perf-tests/clusterloader2/pkg/measurement/util"
+	"k8s.io/perf-tests/clusterloader2/pkg/util"
 )
 
 const (
@@ -95,7 +96,7 @@ func SetUpExecService(f *framework.Framework, c config.ExecServiceConfig) error
 
 	ctx, cancel := context.WithTimeout(context.TODO(), execPodCheckTimeout)
 	defer cancel()
-	selector := &measurementutil.ObjectSelector{
+	selector := &util.ObjectSelector{
 		Namespace:     execDeploymentNamespace,
 		LabelSelector: execPodSelector,
 		FieldSelector: "",

clusterloader2/pkg/measurement/common/loadbalancer_nodesync_latency.go

@@ -20,6 +20,8 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"time"
+
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/fields"
@@ -34,7 +36,6 @@ import (
 	"k8s.io/perf-tests/clusterloader2/pkg/measurement"
 	measurementutil "k8s.io/perf-tests/clusterloader2/pkg/measurement/util"
 	"k8s.io/perf-tests/clusterloader2/pkg/util"
-	"time"
 )
 
 const (
@@ -65,15 +66,15 @@ func init() {
 
 func createLoadBalancerNodeSyncMeasurement() measurement.Measurement {
 	return &LoadBalancerNodeSyncMeasurement{
-		selector:                  measurementutil.NewObjectSelector(),
+		selector:                  util.NewObjectSelector(),
 		svcNodeSyncLatencyTracker: measurementutil.NewObjectTransitionTimes(loadBalancerNodeSyncLatencyName),
 	}
 }
 
 type LoadBalancerNodeSyncMeasurement struct {
 	client clientset.Interface
 	// selector used to select relevant load balancer type service used for measurement
-	selector *measurementutil.ObjectSelector
+	selector *util.ObjectSelector
 	// waitTimeout specify for the timeout for node sync on all LBs to complete
 	waitTimeout time.Duration
 	// svcNodeSyncLatencyTracker tracks the nodesync latency

clusterloader2/pkg/measurement/common/network/network_performance_measurement.go

@@ -228,7 +228,7 @@ func (npm *networkPerformanceMeasurement) createAndWaitForWorkerPods() error {
 	// Wait for all worker pods to be ready
 	ctx, cancel := context.WithTimeout(context.TODO(), podReadyTimeout)
 	defer cancel()
-	selector := &measurementutil.ObjectSelector{Namespace: netperfNamespace}
+	selector := &util.ObjectSelector{Namespace: netperfNamespace}
 	options := &measurementutil.WaitForPodOptions{
 		DesiredPodCount:     func() int { return npm.numberOfClients + npm.numberOfServers },
 		CallerName:          networkPerformanceMetricsName,

clusterloader2/pkg/measurement/common/ooms_tracker.go

@@ -34,7 +34,6 @@ import (
 	"k8s.io/client-go/tools/pager"
 	"k8s.io/klog"
 	"k8s.io/perf-tests/clusterloader2/pkg/measurement"
-	measurementutil "k8s.io/perf-tests/clusterloader2/pkg/measurement/util"
 	"k8s.io/perf-tests/clusterloader2/pkg/measurement/util/informer"
 	"k8s.io/perf-tests/clusterloader2/pkg/util"
 )
@@ -63,7 +62,7 @@ func createClusterOOMsTrackerMeasurement() measurement.Measurement {
 }
 
 type clusterOOMsTrackerMeasurement struct {
-	selector                *measurementutil.ObjectSelector
+	selector                *util.ObjectSelector
 	msgRegex                *regexp.Regexp
 	isRunning               bool
 	startTime               time.Time
@@ -199,7 +198,7 @@ func (m *clusterOOMsTrackerMeasurement) initFields(config *measurement.Config) e
 	m.isRunning = true
 	m.startTime = time.Now()
 	m.stopCh = make(chan struct{})
-	m.selector = &measurementutil.ObjectSelector{
+	m.selector = &util.ObjectSelector{
 		FieldSelector: fields.Set{"reason": oomEventReason}.AsSelector().String(),
 		Namespace:     metav1.NamespaceAll,
 	}

clusterloader2/pkg/measurement/common/scheduling_throughput.go

@@ -67,7 +67,7 @@ func (s *schedulingThroughputMeasurement) Execute(config *measurement.Config) ([
 			klog.V(3).Infof("%s: measurement already running", s)
 			return nil, nil
 		}
-		selector := measurementutil.NewObjectSelector()
+		selector := util.NewObjectSelector()
 		if err := selector.Parse(config.Params); err != nil {
 			return nil, err
 		}
@@ -112,7 +112,7 @@ func (*schedulingThroughputMeasurement) String() string {
 	return schedulingThroughputMeasurementName
 }
 
-func (s *schedulingThroughputMeasurement) start(clientSet clientset.Interface, selector *measurementutil.ObjectSelector, measurmentInterval time.Duration) error {
+func (s *schedulingThroughputMeasurement) start(clientSet clientset.Interface, selector *util.ObjectSelector, measurmentInterval time.Duration) error {
 	ps, err := measurementutil.NewPodStore(clientSet, selector)
 	if err != nil {
 		return fmt.Errorf("pod store creation error: %v", err)

clusterloader2/pkg/measurement/common/service_creation_latency.go

@@ -62,15 +62,15 @@ func init() {
 
 func createServiceCreationLatencyMeasurement() measurement.Measurement {
 	return &serviceCreationLatencyMeasurement{
-		selector:      measurementutil.NewObjectSelector(),
+		selector:      util.NewObjectSelector(),
 		queue:         workerqueue.NewWorkerQueue(serviceCreationLatencyWorkers),
 		creationTimes: measurementutil.NewObjectTransitionTimes(serviceCreationLatencyName),
 		pingCheckers:  checker.NewMap(),
 	}
 }
 
 type serviceCreationLatencyMeasurement struct {
-	selector      *measurementutil.ObjectSelector
+	selector      *util.ObjectSelector
 	waitTimeout   time.Duration
 	stopCh        chan struct{}
 	isRunning     bool

clusterloader2/pkg/measurement/common/slos/pod_startup_latency.go

@@ -57,7 +57,7 @@ func init() {
 
 func createPodStartupLatencyMeasurement() measurement.Measurement {
 	return &podStartupLatencyMeasurement{
-		selector:          measurementutil.NewObjectSelector(),
+		selector:          util.NewObjectSelector(),
 		podStartupEntries: measurementutil.NewObjectTransitionTimes(podStartupLatencyMeasurementName),
 		podMetadata:       measurementutil.NewPodsMetadata(podStartupLatencyMeasurementName),
 		eventQueue:        workqueue.New(),
@@ -70,7 +70,7 @@ type eventData struct {
 }
 
 type podStartupLatencyMeasurement struct {
-	selector  *measurementutil.ObjectSelector
+	selector  *util.ObjectSelector
 	isRunning bool
 	stopCh    chan struct{}
 	// This queue can potentially grow indefinitely, beacause we put all changes here.

clusterloader2/pkg/measurement/common/wait_for_controlled_pods.go

@@ -103,7 +103,7 @@ func (s *sharedPodIndexerFactory) start(c clientset.Interface) (*measurementutil
 
 func createWaitForControlledPodsRunningMeasurement() measurement.Measurement {
 	return &waitForControlledPodsRunningMeasurement{
-		selector:   measurementutil.NewObjectSelector(),
+		selector:   util.NewObjectSelector(),
 		queue:      workerqueue.NewWorkerQueue(waitForControlledPodsWorkers),
 		objectKeys: sets.NewString(),
 		checkerMap: checker.NewMap(),
@@ -113,7 +113,7 @@ func createWaitForControlledPodsRunningMeasurement() measurement.Measurement {
 type waitForControlledPodsRunningMeasurement struct {
 	apiVersion       string
 	kind             string
-	selector         *measurementutil.ObjectSelector
+	selector         *util.ObjectSelector
 	operationTimeout time.Duration
 	// countErrorMargin orders measurement to wait for number of pods to be in
 	// <desired count - countErrorMargin, desired count> range

clusterloader2/pkg/measurement/common/wait_for_jobs.go

@@ -35,7 +35,6 @@ import (
 
 	"k8s.io/perf-tests/clusterloader2/pkg/framework"
 	"k8s.io/perf-tests/clusterloader2/pkg/measurement"
-	measurementutil "k8s.io/perf-tests/clusterloader2/pkg/measurement/util"
 	"k8s.io/perf-tests/clusterloader2/pkg/measurement/util/informer"
 	"k8s.io/perf-tests/clusterloader2/pkg/measurement/util/workerqueue"
 	"k8s.io/perf-tests/clusterloader2/pkg/util"
@@ -56,14 +55,14 @@ func init() {
 
 func createWaitForFinishedJobsMeasurement() measurement.Measurement {
 	return &waitForFinishedJobsMeasurement{
-		selector:     measurementutil.NewObjectSelector(),
+		selector:     util.NewObjectSelector(),
 		queue:        workerqueue.NewWorkerQueue(waitForFinishedJobsWorkers),
 		finishedJobs: make(map[string]batchv1.JobConditionType),
 	}
 }
 
 type waitForFinishedJobsMeasurement struct {
-	selector *measurementutil.ObjectSelector
+	selector *util.ObjectSelector
 
 	queue            workerqueue.Interface
 	isRunning        bool

clusterloader2/pkg/measurement/common/wait_for_nodes.go

@@ -52,7 +52,7 @@ func (w *waitForNodesMeasurement) Execute(config *measurement.Config) ([]measure
 		return nil, err
 	}
 
-	selector := measurementutil.NewObjectSelector()
+	selector := util.NewObjectSelector()
 	if err := selector.Parse(config.Params); err != nil {
 		return nil, err
 	}

clusterloader2/pkg/measurement/common/wait_for_pods.go

@@ -52,7 +52,7 @@ func (w *waitForRunningPodsMeasurement) Execute(config *measurement.Config) ([]m
 	if err != nil {
 		return nil, err
 	}
-	selector := measurementutil.NewObjectSelector()
+	selector := util.NewObjectSelector()
 	if err := selector.Parse(config.Params); err != nil {
 		return nil, err
 	}

clusterloader2/pkg/measurement/common/wait_for_pvcs.go

@@ -51,7 +51,7 @@ func (w *waitForBoundPVCsMeasurement) Execute(config *measurement.Config) ([]mea
 	if err != nil {
 		return nil, err
 	}
-	selector := measurementutil.NewObjectSelector()
+	selector := util.NewObjectSelector()
 	if err := selector.Parse(config.Params); err != nil {
 		return nil, err
 	}

clusterloader2/pkg/measurement/common/wait_for_pvs.go

@@ -54,7 +54,7 @@ func (w *waitForAvailablePVsMeasurement) Execute(config *measurement.Config) ([]
 	if err != nil {
 		return nil, err
 	}
-	selector := measurementutil.NewObjectSelector()
+	selector := util.NewObjectSelector()
 	if err := selector.Parse(config.Params); err != nil {
 		return nil, err
 	}

clusterloader2/pkg/measurement/util/informer/informer.go

@@ -25,8 +25,7 @@ import (
 	"k8s.io/client-go/dynamic"
 	"k8s.io/client-go/dynamic/dynamicinformer"
 	"k8s.io/client-go/tools/cache"
-
-	measurementutil "k8s.io/perf-tests/clusterloader2/pkg/measurement/util"
+	"k8s.io/perf-tests/clusterloader2/pkg/util"
 )
 
 // NewInformer creates a new informer.
@@ -44,7 +43,7 @@ func NewInformer(
 func NewDynamicInformer(
 	c dynamic.Interface,
 	gvr schema.GroupVersionResource,
-	selector *measurementutil.ObjectSelector,
+	selector *util.ObjectSelector,
 	handleObj func(interface{}, interface{}),
 ) cache.SharedInformer {
 	optionsModifier := func(options *metav1.ListOptions) {

clusterloader2/pkg/measurement/util/object_store.go

@@ -35,6 +35,7 @@ import (
 	"k8s.io/apimachinery/pkg/watch"
 	clientset "k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/tools/cache"
+	"k8s.io/perf-tests/clusterloader2/pkg/util"
 )
 
 // ObjectStore is a convenient wrapper around cache.Store.
@@ -45,7 +46,7 @@ type ObjectStore struct {
 }
 
 // newObjectStore creates ObjectStore based on given object selector.
-func newObjectStore(obj runtime.Object, lw *cache.ListWatch, selector *ObjectSelector) (*ObjectStore, error) {
+func newObjectStore(obj runtime.Object, lw *cache.ListWatch, selector *util.ObjectSelector) (*ObjectStore, error) {
 	store := cache.NewStore(cache.MetaNamespaceKeyFunc)
 	stopCh := make(chan struct{})
 	name := fmt.Sprintf("%sStore: %s", reflect.TypeOf(obj).String(), selector.String())
@@ -75,11 +76,11 @@ func (s *ObjectStore) Stop() {
 // PodStore is a convenient wrapper around cache.Store.
 type PodStore struct {
 	*ObjectStore
-	selector *ObjectSelector
+	selector *util.ObjectSelector
 }
 
 // NewPodStore creates PodStore based on given object selector.
-func NewPodStore(c clientset.Interface, selector *ObjectSelector) (*PodStore, error) {
+func NewPodStore(c clientset.Interface, selector *util.ObjectSelector) (*PodStore, error) {
 	lw := &cache.ListWatch{
 		ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
 			options.LabelSelector = selector.LabelSelector
@@ -157,7 +158,7 @@ type PVCStore struct {
 }
 
 // NewPVCStore creates PVCStore based on a given object selector.
-func NewPVCStore(c clientset.Interface, selector *ObjectSelector) (*PVCStore, error) {
+func NewPVCStore(c clientset.Interface, selector *util.ObjectSelector) (*PVCStore, error) {
 	lw := &cache.ListWatch{
 		ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
 			options.LabelSelector = selector.LabelSelector
@@ -194,7 +195,7 @@ type PVStore struct {
 }
 
 // NewPVStore creates PVStore based on a given object selector.
-func NewPVStore(c clientset.Interface, selector *ObjectSelector, provisioner string) (*PVStore, error) {
+func NewPVStore(c clientset.Interface, selector *util.ObjectSelector, provisioner string) (*PVStore, error) {
 	lw := &cache.ListWatch{
 		ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
 			options.LabelSelector = selector.LabelSelector
@@ -234,7 +235,7 @@ type NodeStore struct {
 }
 
 // NewNodeStore creates NodeStore based on a given object selector.
-func NewNodeStore(c clientset.Interface, selector *ObjectSelector) (*NodeStore, error) {
+func NewNodeStore(c clientset.Interface, selector *util.ObjectSelector) (*NodeStore, error) {
 	lw := &cache.ListWatch{
 		ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
 			options.LabelSelector = selector.LabelSelector

clusterloader2/pkg/measurement/util/wait_for_nodes.go

@@ -28,7 +28,7 @@ import (
 
 // WaitForNodeOptions is an options object used by WaitForNodes methods.
 type WaitForNodeOptions struct {
-	Selector             *ObjectSelector
+	Selector             *util.ObjectSelector
 	MinDesiredNodeCount  int
 	MaxDesiredNodeCount  int
 	CallerName           string

clusterloader2/pkg/measurement/util/wait_for_pvcs.go

@@ -23,11 +23,12 @@ import (
 
 	clientset "k8s.io/client-go/kubernetes"
 	"k8s.io/klog"
+	"k8s.io/perf-tests/clusterloader2/pkg/util"
 )
 
 // WaitForPVCOptions is an options used by WaitForPVCs methods.
 type WaitForPVCOptions struct {
-	Selector            *ObjectSelector
+	Selector            *util.ObjectSelector
 	DesiredPVCCount     int
 	CallerName          string
 	WaitForPVCsInterval time.Duration

clusterloader2/pkg/measurement/util/wait_for_pvs.go

@@ -23,11 +23,12 @@ import (
 
 	clientset "k8s.io/client-go/kubernetes"
 	"k8s.io/klog"
+	"k8s.io/perf-tests/clusterloader2/pkg/util"
 )
 
 // WaitForPVOptions is an options used by WaitForPVs methods.
 type WaitForPVOptions struct {
-	Selector           *ObjectSelector
+	Selector           *util.ObjectSelector
 	DesiredPVCount     int
 	Provisioner        string
 	CallerName         string

clusterloader2/pkg/measurement/util/selector.go → clusterloader2/pkg/util/selector.go

@@ -21,8 +21,6 @@ import (
 	"strings"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-
-	"k8s.io/perf-tests/clusterloader2/pkg/util"
 )
 
 // ObjectSelector is an aggregation of namespace, labelSelector and fieldSelector.
@@ -42,15 +40,15 @@ func NewObjectSelector() *ObjectSelector {
 // Parse parses namespace, labelSelector and fieldSelector from params map.
 func (o *ObjectSelector) Parse(params map[string]interface{}) error {
 	var err error
-	o.Namespace, err = util.GetStringOrDefault(params, "namespace", metav1.NamespaceAll)
+	o.Namespace, err = GetStringOrDefault(params, "namespace", metav1.NamespaceAll)
 	if err != nil {
 		return err
 	}
-	o.LabelSelector, err = util.GetStringOrDefault(params, "labelSelector", "")
+	o.LabelSelector, err = GetStringOrDefault(params, "labelSelector", "")
 	if err != nil {
 		return err
 	}
-	o.FieldSelector, err = util.GetStringOrDefault(params, "fieldSelector", "")
+	o.FieldSelector, err = GetStringOrDefault(params, "fieldSelector", "")
 	if err != nil {
 		return err
 	}

clusterloader2/pkg/util/util.go

@@ -41,7 +41,7 @@ func IsErrKeyNotFound(err error) bool {
 	return isErrKeyNotFound
 }
 
-// ToStruct converts map[string]interface{} to standard object (e.g. struct).
+// ToStruct converts map[string]interface{} to standard object (e.g. struct). It preserves fields that are not set in dict.
 func ToStruct(dict map[string]interface{}, out interface{}) error {
 	output := &bytes.Buffer{}
 	if err := json.NewEncoder(output).Encode(dict); err != nil {