auditLogAnalyzer: find resources being updated too often

Author: Vadim Rutkovsky

pkg/monitortests/kubeapiserver/auditloganalyzer/handle_excessive_applies.go

@@ -1,21 +1,70 @@
 package auditloganalyzer
 
 import (
+	"fmt"
+	"sort"
+	"strings"
+	"sync"
+
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	auditv1 "k8s.io/apiserver/pkg/apis/audit/v1"
 	"k8s.io/apiserver/pkg/authentication/serviceaccount"
-	"strings"
-	"sync"
 )
 
+type userApplyCount struct {
+	User    string
+	Applies int
+}
+
+type numberOfAppliesWithExamples struct {
+	numberOfApplies   int
+	users             map[string]int
+	firstAuditEventID string
+	lastAuditEventID  string
+	firstOccurredAt   metav1.Time
+	lastOccurredAt    metav1.Time
+}
+
+func (n numberOfAppliesWithExamples) toErrorString() string {
+	appliesPerUser := []userApplyCount{}
+	for k, v := range n.users {
+		appliesPerUser = append(appliesPerUser, userApplyCount{k, v})
+	}
+	sort.Slice(appliesPerUser, func(i, j int) bool {
+		return appliesPerUser[i].Applies > appliesPerUser[j].Applies
+	})
+	topUsernames := []string{}
+	maxUsernames := 5
+	if len(appliesPerUser) < maxUsernames {
+		maxUsernames = len(appliesPerUser)
+	}
+	for i := 0; i < maxUsernames; i++ {
+		topUsernames = append(topUsernames, appliesPerUser[i].User)
+	}
+
+	return fmt.Sprintf(`
+Time: started %s, finished at %s
+Top 5 usernames: %s
+Audit Log IDs: %s ... %s
+`,
+		n.firstOccurredAt,
+		n.lastOccurredAt,
+		strings.Join(topUsernames, ", "),
+		n.firstAuditEventID,
+		n.lastAuditEventID,
+	)
+}
+
 type excessiveApplies struct {
 	lock                              sync.Mutex
 	namespacesToUserToNumberOfApplies map[string]map[string]int
+	resourcesToNumberOfApplies        map[string]numberOfAppliesWithExamples
 }
 
 func CheckForExcessiveApplies() *excessiveApplies {
 	return &excessiveApplies{
 		namespacesToUserToNumberOfApplies: map[string]map[string]int{},
+		resourcesToNumberOfApplies:        map[string]numberOfAppliesWithExamples{},
 	}
 }
 
@@ -43,4 +92,36 @@ func (s *excessiveApplies) HandleAuditLogEvent(auditEvent *auditv1.Event, beginn
 	}
 	users[auditEvent.User.Username] = users[auditEvent.User.Username] + 1
 	s.namespacesToUserToNumberOfApplies[nsName] = users
+
+	obj := auditEvent.ObjectRef
+	if obj == nil {
+		return
+	}
+	resource := fmt.Sprintf("%s/%s", obj.Resource, obj.Name)
+	if obj.Namespace != "" {
+		resource = fmt.Sprintf("%s -n %s", resource, obj.Namespace)
+	}
+	if obj.APIGroup != "" {
+		resource = fmt.Sprintf("%s.%s", obj.APIGroup, resource)
+	}
+	objApplies, ok := s.resourcesToNumberOfApplies[resource]
+	if !ok {
+		objApplies = numberOfAppliesWithExamples{
+			numberOfApplies:   0,
+			firstAuditEventID: string(auditEvent.AuditID),
+			lastAuditEventID:  string(auditEvent.AuditID),
+			users:             map[string]int{},
+			firstOccurredAt:   metav1.Time(auditEvent.RequestReceivedTimestamp),
+			lastOccurredAt:    metav1.Time(auditEvent.RequestReceivedTimestamp),
+		}
+	}
+	objApplies.numberOfApplies += 1
+	objApplies.lastAuditEventID = string(auditEvent.AuditID)
+	objApplies.lastOccurredAt = metav1.Time(auditEvent.RequestReceivedTimestamp)
+	userApplies, ok := objApplies.users[auditEvent.User.Username]
+	if !ok {
+		userApplies = 0
+	}
+	objApplies.users[auditEvent.User.Username] = userApplies + 1
+	s.resourcesToNumberOfApplies[resource] = objApplies
 }

pkg/monitortests/kubeapiserver/auditloganalyzer/monitortest.go

@@ -256,6 +256,40 @@ func (w *auditLogAnalyzer) EvaluateTestsFromConstructedIntervals(ctx context.Con
 
 	}
 
+	testName := `[Jira:"kube-apiserver"] API resources are not updated excessively`
+	flakes := []string{}
+	for resource, applies := range w.excessiveApplyChecker.resourcesToNumberOfApplies {
+		if applies.numberOfApplies < 200 {
+			continue
+		}
+		errorMessage := fmt.Sprintf("resource %s had %d applies, %s", resource, applies.numberOfApplies, applies.toErrorString())
+		flakes = append(flakes, errorMessage)
+	}
+	switch {
+	case len(flakes) > 1:
+		ret = append(ret,
+			&junitapi.JUnitTestCase{
+				Name: testName,
+				FailureOutput: &junitapi.FailureOutput{
+					Message: strings.Join(flakes, "\n"),
+					Output:  "details in audit log",
+				},
+			},
+		)
+		ret = append(ret,
+			&junitapi.JUnitTestCase{
+				Name: testName,
+			},
+		)
+
+	default:
+		ret = append(ret,
+			&junitapi.JUnitTestCase{
+				Name: testName,
+			},
+		)
+	}
+
 	for verb, namespacesToUserToNumberOf422s := range w.invalidRequestsChecker.verbToNamespacesTouserToNumberOf422s {
 		for _, namespace := range allPlatformNamespaces {
 			testName := fmt.Sprintf("users in ns/%s must not produce too many invalid %q requests", namespace, verb)
@@ -346,7 +380,7 @@ func (w *auditLogAnalyzer) EvaluateTestsFromConstructedIntervals(ctx context.Con
 		}
 	}
 
-	testName := "API LBs follow /readyz of kube-apiserver and stop sending requests before server shutdowns for external clients"
+	testName = "API LBs follow /readyz of kube-apiserver and stop sending requests before server shutdowns for external clients"
 	switch {
 	case len(w.requestsDuringShutdownChecker.auditIDs) > 0:
 		ret = append(ret,