Merge pull request #60794 from crassirostris/fix-audit-e2e

Automatic merge from submit-queue (batch tested with PRs 60630, 60794). If you want to cherry-pick this change to another branch, please follow the instructions <a href="https://github.com/kubernetes/community/blob/master/contributors/devel/cherry-picks.md">here</a>.

Add retrying to audit logging e2e tests

Fixes https://github.com/kubernetes/kubernetes/issues/60719

Adds retrying to the audit logging e2e tests so it can work when audit logging is in batch mode and actual writing is delayed.

```release-note
NONE
```

/cc @tallclair @liggitt @sttts

test/e2e/auth/audit.go

@@ -21,6 +21,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"strings"
+	"time"
 
 	apiv1 "k8s.io/api/core/v1"
 	extensions "k8s.io/api/extensions/v1beta1"
@@ -29,13 +30,13 @@ import (
 	"k8s.io/apiextensions-apiserver/test/integration/testserver"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/apiserver/pkg/apis/audit/v1beta1"
 	"k8s.io/kubernetes/test/e2e/framework"
 	imageutils "k8s.io/kubernetes/test/utils/image"
 
 	"github.com/evanphx/json-patch"
 	. "github.com/onsi/ginkgo"
-	. "github.com/onsi/gomega"
 )
 
 var (
@@ -630,7 +631,18 @@ var _ = SIGDescribe("Advanced Audit", func() {
 			expectedEvents = append(expectedEvents, t.events...)
 		}
 
-		expectAuditLines(f, expectedEvents)
+		// The default flush timeout is 30 seconds, therefore it should be enough to retry once
+		// to find all expected events. However, we're waiting for 5 minutes to avoid flakes.
+		pollingInterval := 30 * time.Second
+		pollingTimeout := 5 * time.Minute
+		err = wait.Poll(pollingInterval, pollingTimeout, func() (bool, error) {
+			ok, err := checkAuditLines(f, expectedEvents)
+			if err != nil {
+				framework.Logf("Failed to observe audit events: %v", err)
+			}
+			return ok, nil
+		})
+		framework.ExpectNoError(err, "after %v failed to observe audit events", pollingTimeout)
 	})
 })
 
@@ -648,7 +660,7 @@ type auditEvent struct {
 }
 
 // Search the audit log for the expected audit lines.
-func expectAuditLines(f *framework.Framework, expected []auditEvent) {
+func checkAuditLines(f *framework.Framework, expected []auditEvent) (bool, error) {
 	expectations := map[auditEvent]bool{}
 	for _, event := range expected {
 		expectations[event] = false
@@ -656,25 +668,36 @@ func expectAuditLines(f *framework.Framework, expected []auditEvent) {
 
 	// Fetch the log stream.
 	stream, err := f.ClientSet.CoreV1().RESTClient().Get().AbsPath("/logs/kube-apiserver-audit.log").Stream()
-	framework.ExpectNoError(err, "could not read audit log")
+	if err != nil {
+		return false, err
+	}
 	defer stream.Close()
 
 	scanner := bufio.NewScanner(stream)
 	for scanner.Scan() {
 		line := scanner.Text()
 		event, err := parseAuditLine(line)
-		framework.ExpectNoError(err)
+		if err != nil {
+			return false, err
+		}
 
 		// If the event was expected, mark it as found.
 		if _, found := expectations[event]; found {
 			expectations[event] = true
 		}
 	}
-	framework.ExpectNoError(scanner.Err(), "error reading audit log")
-
-	for event, found := range expectations {
-		Expect(found).To(BeTrue(), "Event %#v not found!", event)
+	if err := scanner.Err(); err != nil {
+		return false, err
 	}
+
+	noneMissing := true
+	for event, found := range expectations {
+		if !found {
+			framework.Logf("Event %#v not found!", event)
+		}
+		noneMissing = noneMissing && found
+	}
+	return noneMissing, nil
 }
 
 func parseAuditLine(line string) (auditEvent, error) {