Remove duplicate errors from an aggregate error input.

plugin/pkg/scheduler/core/generic_scheduler.go

@@ -178,7 +178,7 @@ func findNodesThatFit(
 		// Create filtered list with enough space to avoid growing it
 		// and allow assigning.
 		filtered = make([]*v1.Node, len(nodes))
-		errs := []error{}
+		errs := errors.MessageCountMap{}
 		var predicateResultLock sync.Mutex
 		var filteredLen int32
 
@@ -189,7 +189,7 @@ func findNodesThatFit(
 			fits, failedPredicates, err := podFitsOnNode(pod, meta, nodeNameToInfo[nodeName], predicateFuncs, ecache)
 			if err != nil {
 				predicateResultLock.Lock()
-				errs = append(errs, err)
+				errs[err.Error()]++
 				predicateResultLock.Unlock()
 				return
 			}
@@ -204,7 +204,7 @@ func findNodesThatFit(
 		workqueue.Parallelize(16, len(nodes), checkNode)
 		filtered = filtered[:filteredLen]
 		if len(errs) > 0 {
-			return []*v1.Node{}, FailedPredicateMap{}, errors.NewAggregate(errs)
+			return []*v1.Node{}, FailedPredicateMap{}, errors.CreateAggregateFromMessageCountMap(errs)
 		}
 	}
 

staging/src/k8s.io/apimachinery/pkg/util/errors/errors.go

@@ -21,6 +21,9 @@ import (
 	"fmt"
 )
 
+// MessagesgCountMap contains occurance for each error message.
+type MessageCountMap map[string]int
+
 // Aggregate represents an object that contains multiple errors, but does not
 // necessarily have singular semantic meaning.
 type Aggregate interface {
@@ -147,6 +150,22 @@ func Flatten(agg Aggregate) Aggregate {
 	return NewAggregate(result)
 }
 
+// CreateAggregateFromMessageCountMap converts MessageCountMap Aggregate
+func CreateAggregateFromMessageCountMap(m MessageCountMap) Aggregate {
+	if m == nil {
+		return nil
+	}
+	result := make([]error, 0, len(m))
+	for errStr, count := range m {
+		var countStr string
+		if count > 1 {
+			countStr = fmt.Sprintf(" (repeated %v times)", count)
+		}
+		result = append(result, fmt.Errorf("%v%v", errStr, countStr))
+	}
+	return NewAggregate(result)
+}
+
 // Reduce will return err or, if err is an Aggregate and only has one item,
 // the first item in the aggregate.
 func Reduce(err error) error {

staging/src/k8s.io/apimachinery/pkg/util/errors/errors_test.go

@@ -19,6 +19,7 @@ package errors
 import (
 	"fmt"
 	"reflect"
+	"sort"
 	"testing"
 )
 
@@ -262,6 +263,42 @@ func TestFlatten(t *testing.T) {
 	}
 }
 
+func TestCreateAggregateFromMessageCountMap(t *testing.T) {
+	testCases := []struct {
+		name     string
+		mcp      MessageCountMap
+		expected Aggregate
+	}{
+		{
+			"input has single instance of one message",
+			MessageCountMap{"abc": 1},
+			aggregate{fmt.Errorf("abc")},
+		},
+		{
+			"input has multiple messages",
+			MessageCountMap{"abc": 2, "ghi": 1},
+			aggregate{fmt.Errorf("abc (repeated 2 times)"), fmt.Errorf("ghi")},
+		},
+	}
+
+	var expected, agg []error
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+			if testCase.expected != nil {
+				expected = testCase.expected.Errors()
+				sort.Slice(expected, func(i, j int) bool { return expected[i].Error() < expected[j].Error() })
+			}
+			if testCase.mcp != nil {
+				agg = CreateAggregateFromMessageCountMap(testCase.mcp).Errors()
+				sort.Slice(agg, func(i, j int) bool { return agg[i].Error() < agg[j].Error() })
+			}
+			if !reflect.DeepEqual(expected, agg) {
+				t.Errorf("expected %v, got %v", expected, agg)
+			}
+		})
+	}
+}
+
 func TestAggregateGoroutines(t *testing.T) {
 	testCases := []struct {
 		errs     []error