Merge pull request #50904 from crassirostris/sd-logging-e2e-system-logs

test/e2e/instrumentation/logging/stackdrvier/basic.go

@@ -142,11 +142,21 @@ var _ = instrumentation.SIGDescribe("Cluster level logging implemented by Stackd
 
 	ginkgo.It("should ingest system logs from all nodes", func() {
 		withLogProviderForScope(f, systemScope, func(p *sdLogProvider) {
-			ginkgo.By("Waiting for some system logs to ingest")
-			nodeIds := utils.GetNodeIds(f.ClientSet)
-			c := utils.NewLogChecker(p, utils.UntilFirstEntry, utils.JustTimeout, nodeIds...)
-			err := utils.WaitForLogs(c, ingestionInterval, ingestionTimeout)
-			framework.ExpectNoError(err)
+			ginkgo.By("Waiting for some kubelet logs to be ingested from each node", func() {
+				nodeIds := utils.GetNodeIds(f.ClientSet)
+				log := fmt.Sprintf("projects/%s/logs/kubelet", framework.TestContext.CloudConfig.ProjectID)
+				c := utils.NewLogChecker(p, utils.UntilFirstEntryFromLog(log), utils.JustTimeout, nodeIds...)
+				err := utils.WaitForLogs(c, ingestionInterval, ingestionTimeout)
+				framework.ExpectNoError(err)
+			})
+
+			ginkgo.By("Waiting for some docker logs to be ingested from each node", func() {
+				nodeIds := utils.GetNodeIds(f.ClientSet)
+				log := fmt.Sprintf("projects/%s/logs/docker", framework.TestContext.CloudConfig.ProjectID)
+				c := utils.NewLogChecker(p, utils.UntilFirstEntryFromLog(log), utils.JustTimeout, nodeIds...)
+				err := utils.WaitForLogs(c, ingestionInterval, ingestionTimeout)
+				framework.ExpectNoError(err)
+			})
 		})
 	})
 })

test/e2e/instrumentation/logging/stackdrvier/utils.go

@@ -315,23 +315,27 @@ func (p *sdLogProvider) tryGetName(sdLogEntry sd.LogEntry) (string, bool) {
 	return "", false
 }
 
-func convertLogEntry(sdLogEntry sd.LogEntry) (utils.LogEntry, error) {
+func convertLogEntry(sdLogEntry sd.LogEntry) (entry utils.LogEntry, err error) {
+	entry = utils.LogEntry{LogName: sdLogEntry.LogName}
 	if sdLogEntry.TextPayload != "" {
-		return utils.LogEntry{TextPayload: sdLogEntry.TextPayload}, nil
+		entry.TextPayload = sdLogEntry.TextPayload
+		return
 	}
 
 	bytes, err := sdLogEntry.JsonPayload.MarshalJSON()
 	if err != nil {
-		return utils.LogEntry{}, fmt.Errorf("Failed to get jsonPayload from LogEntry %v", sdLogEntry)
+		err = fmt.Errorf("Failed to get jsonPayload from LogEntry %v", sdLogEntry)
+		return
 	}
 
 	var jsonObject map[string]interface{}
 	err = json.Unmarshal(bytes, &jsonObject)
 	if err != nil {
-		return utils.LogEntry{},
-			fmt.Errorf("Failed to deserialize jsonPayload as json object %s", string(bytes[:]))
+		err = fmt.Errorf("Failed to deserialize jsonPayload as json object %s", string(bytes[:]))
+		return
 	}
-	return utils.LogEntry{JSONPayload: jsonObject}, nil
+	entry.JSONPayload = jsonObject
+	return
 }
 
 func pullAndAck(service *pubsub.Service, subs *pubsub.Subscription) ([]*pubsub.ReceivedMessage, error) {

test/e2e/instrumentation/logging/utils/types.go

@@ -30,6 +30,7 @@ var (
 
 // LogEntry represents a log entry, received from the logging backend.
 type LogEntry struct {
+	LogName     string
 	TextPayload string
 	JSONPayload map[string]interface{}
 }

test/e2e/instrumentation/logging/utils/wait.go

@@ -42,6 +42,19 @@ var UntilFirstEntry IngestionPred = func(_ string, entries []LogEntry) (bool, er
 	return len(entries) > 0, nil
 }
 
+// UntilFirstEntryFromLog is a IngestionPred that checks that at least one
+// entry from the log with a given name was ingested.
+func UntilFirstEntryFromLog(log string) IngestionPred {
+	return func(_ string, entries []LogEntry) (bool, error) {
+		for _, e := range entries {
+			if e.LogName == log {
+				return true, nil
+			}
+		}
+		return false, nil
+	}
+}
+
 // TimeoutFun is a function that is called when the waiting times out.
 type TimeoutFun func([]string, []bool) error