e2e test case for partially dropped flows

.github/workflows/e2e.yaml

@@ -40,6 +40,9 @@ jobs:
         uses: azure/setup-helm@v3
         with:
           version: ${{ env.HELM_VERSION }}
+      - uses: actions/setup-node@v4
+        with:
+          node-version: 20
       - name: Clone cilium/hubble-ui
         uses: actions/checkout@v4
       - name: Fetch backend log-files and overwrite kind config
@@ -101,7 +104,6 @@ jobs:
         uses: cypress-io/github-action@v6
         with:
           record: true
-          working-directory: e2e
           config: baseUrl=https://${{ env.HUBBLE_UI_HOST }}
           browser: chrome
         env:
@@ -111,7 +113,6 @@ jobs:
         uses: cypress-io/github-action@v6
         with:
           record: true
-          working-directory: e2e
           config: baseUrl=https://${{ env.HUBBLE_UI_HOST}}
           browser: firefox
         env:

.gitignore

@@ -59,3 +59,6 @@ yarn-error.log*
 
 # VSCode
 .vscode
+
+# Cypress
+src/testing/e2e/cypress/screenshots

backend/internal/e2e/namespace_list_check_case.go

@@ -9,7 +9,7 @@ import (
 
 func (tc *TestsController) enableNamespaceListCheckCase(_ts *TestSettings) {
 	nsSource := sources.Namespaces([]*ns_common.NSEvent{
-		factories.CreateNSEvent(events.Added, "ns-relay"),
+		factories.CreateNSEvent(events.Added, "relay"),
 	})
 
 	tc.clients.SetSource(nsSource)

backend/internal/e2e/partially_dropped.go

@@ -0,0 +1,52 @@
+package e2e
+
+import (
+	"time"
+
+	"github.com/cilium/hubble-ui/backend/domain/events"
+	"github.com/cilium/hubble-ui/backend/internal/events_log_file"
+	"github.com/cilium/hubble-ui/backend/internal/mock/factories"
+	"github.com/cilium/hubble-ui/backend/internal/mock/sources"
+	ns_common "github.com/cilium/hubble-ui/backend/internal/ns_watcher/common"
+	"github.com/cilium/hubble-ui/backend/pkg/rate_limiter"
+)
+
+func (tc *TestsController) enablePartiallyDroppedCase() error {
+	flowsLogFile, err := tc.openLogFile("/flows/forwarded-and-dropped-2.json")
+	if err != nil {
+		return err
+	}
+
+	nsSource := sources.Namespaces([]*ns_common.NSEvent{
+		factories.CreateNSEvent(events.Added, "default"),
+		factories.CreateNSEvent(events.Added, "default"),
+	})
+
+	flowsSource := sources.LogFile(
+		flowsLogFile,
+		tc.log.WithField("source", "FnD-flows"),
+		sources.LogFileSourceOpts{
+			DebugHandler: func(i int, e *events_log_file.EventEntry) {
+				if e == nil {
+					return
+				}
+
+				tc.log.
+					WithField("index", i).
+					WithFields(e.LogEntries()).
+					Debug("unparsed entry")
+			},
+		},
+	)
+
+	combined := sources.Combine(nsSource, flowsSource)
+	tc.clients.SetSource(combined)
+
+	flowsRateLimit := rate_limiter.RateLimit{
+		Limit:  100,
+		Period: 5 * time.Second,
+	}
+	tc.clients.SetFlowsRateLimit(flowsRateLimit)
+
+	return nil
+}

backend/internal/e2e/test_settings.go

@@ -12,6 +12,7 @@ const (
 	NoPreset           TestPreset = ""
 	TenantJobs         TestPreset = "tenant-jobs"
 	NamespaceListCheck TestPreset = "ns-list-check"
+	PartiallyDropped   TestPreset = "partially-dropped"
 )
 
 type TestSettings struct {

backend/internal/e2e/tests_controller.go

@@ -56,40 +56,54 @@ func (tc *TestsController) LogFields() logrus.Fields {
 }
 
 func (tc *TestsController) applyTestSettings(ts *TestSettings) {
+	var err error
+
 	switch ts.Preset {
 	case NamespaceListCheck:
 		tc.enableNamespaceListCheckCase(ts)
 	case TenantJobs:
-		if err := tc.enableTenantJobsCase(ts); err != nil {
-			tc.log.WithError(err).Error("failed to enable tenant jobs case")
-		}
+		err = tc.enableTenantJobsCase(ts)
+	case PartiallyDropped:
+		err = tc.enablePartiallyDroppedCase()
+	}
+
+	if err != nil {
+		tc.log.
+			WithError(err).
+			WithField("preset", ts.Preset).
+			Error("failed to enable test case by preset")
 	}
 }
 
 func (tc *TestsController) openLogFiles(pathPart string) (
 	*log_file.LogFile, *log_file.LogFile, error,
 ) {
-	flowsPath := filepath.Join(tc.logFilesPath, pathPart, "hubble-events.json.log")
-	flowsLogFile, err := log_file.OpenLogFile(flowsPath)
+	flowsLogFile, err := tc.openLogFile(pathPart, "hubble-events.json.log")
 	if err != nil {
-		tc.log.
-			WithError(err).
-			WithField("path", flowsPath).
-			Error("failed to open flows log file")
+		return nil, nil, err
+	}
 
+	eventsLogFile, err := tc.openLogFile(pathPart, "fgs-events.json.log")
+	if err != nil {
 		return nil, nil, err
 	}
 
-	eventsPath := filepath.Join(tc.logFilesPath, pathPart, "fgs-events.json.log")
-	eventsLogFile, err := log_file.OpenLogFile(eventsPath)
+	return flowsLogFile, eventsLogFile, nil
+}
+
+func (tc *TestsController) openLogFile(pathParts ...string) (*log_file.LogFile, error) {
+	pathParts = append([]string{tc.logFilesPath}, pathParts...)
+	fpath := filepath.Join(pathParts...)
+
+	logFile, err := log_file.OpenLogFile(fpath)
 	if err != nil {
 		tc.log.
 			WithError(err).
-			WithField("path", eventsPath).
-			Error("failed to open fgs events log file")
+			WithField("path", fpath).
+			Error("failed to open log file")
 
-		return nil, nil, err
+		return nil, err
 	}
 
-	return flowsLogFile, eventsLogFile, nil
+	return logFile, nil
 }

backend/internal/events_log_file/event_entry.go

@@ -1,17 +1,24 @@
 package events_log_file
 
 import (
-	"github.com/cilium/cilium/api/v1/observer"
+	"encoding/json"
+
+	observerpb "github.com/cilium/cilium/api/v1/observer"
+	"github.com/sirupsen/logrus"
 	"google.golang.org/protobuf/encoding/protojson"
 )
 
 type EventEntry struct {
-	Flow     *observer.GetFlowsResponse
+	Flow           *observerpb.GetFlowsResponse
+	FlowParseError error
+
+	PEParseError error
+
 	Unparsed *string
 }
 
 func tryParseFlow(str string) (*EventEntry, error) {
-	flowEvent := &observer.GetFlowsResponse{}
+	flowEvent := &observerpb.GetFlowsResponse{}
 	err := protojson.Unmarshal([]byte(str), flowEvent)
 	if err != nil {
 		return nil, err
@@ -21,3 +28,25 @@ func tryParseFlow(str string) (*EventEntry, error) {
 		Flow: flowEvent,
 	}, nil
 }
+
+func (ee *EventEntry) LogEntries() logrus.Fields {
+	f := logrus.Fields{
+		"flow-parse-error": ee.FlowParseError,
+		"pe-parse-error":   ee.PEParseError,
+	}
+
+	if ee.Unparsed != nil {
+		f["unparsed"] = *ee.Unparsed
+	}
+
+	if ee.Flow != nil {
+		str, err := json.Marshal(ee.Flow)
+		if err != nil {
+			str = []byte{}
+		}
+
+		f["flow"] = string(str)
+	}
+
+	return f
+}

backend/internal/events_log_file/events_log_file.go

@@ -20,13 +20,14 @@ func (it *EventsLogFileIterator) Next() *EventEntry {
 		return nil
 	}
 
-	flow, _ := tryParseFlow(next)
+	flow, err1 := tryParseFlow(next)
 	if flow != nil {
 		return flow
 	}
 
 	return &EventEntry{
-		Unparsed: &next,
+		FlowParseError: err1,
+		Unparsed:       &next,
 	}
 }
 

backend/internal/mock/sources/log_file.go

@@ -23,6 +23,7 @@ type LogFileSource struct {
 
 type LogFileSourceOpts struct {
 	FlowsRateLimit rate_limiter.RateLimit
+	DebugHandler   func(int, *events_log_file.EventEntry)
 }
 
 func LogFile(
@@ -53,7 +54,10 @@ func (lfs *LogFileSource) Run(ctx context.Context) {
 		panic("failed to create an iterator over log file: " + err.Error())
 	}
 
-	defer lfs.log.Info("source is exhausted")
+	nEntries := 0
+	nFlows := 0
+	nProcEvents := 0
+	nUnparsed := 0
 
 	it := events_log_file.NewEventsIterator(_it)
 	flowsRateLimiter := rate_limiter.New(lfs.opts.FlowsRateLimit)
@@ -65,8 +69,11 @@ func (lfs *LogFileSource) Run(ctx context.Context) {
 
 	for it.HasNext() {
 		entry := it.Next()
+		nEntries += 1
 
 		if entry.Flow != nil {
+			nFlows += 1
+
 			if err := flowsRateLimiter.Wait(ctx); err != nil {
 				lfs.log.WithError(err).Error("flowsRateLimiter.Wait() error")
 				return
@@ -77,7 +84,18 @@ func (lfs *LogFileSource) Run(ctx context.Context) {
 				return
 			}
 		}
+
+		if lfs.opts.DebugHandler != nil {
+			lfs.opts.DebugHandler(nEntries-1, entry)
+		}
 	}
+
+	lfs.log.
+		WithField("nentries", nEntries).
+		WithField("nflows", nFlows).
+		WithField("nprocevents", nProcEvents).
+		WithField("nunparsed", nUnparsed).
+		Info("log file source is exhausted")
 }
 
 func (lfs *LogFileSource) sendFlow(ctx context.Context, flow *observer.GetFlowsResponse) error {

cypress.config.ts

@@ -0,0 +1,28 @@
+import { defineConfig } from 'cypress';
+
+import webpackPreprocessor from '@cypress/webpack-preprocessor';
+
+import mainWebpackConfig from './webpack.config';
+
+export const webpackConfiguration = async (f: Cypress.FileObject) => {
+  const opts = { webpackOptions: { ...mainWebpackConfig, plugins: [] } };
+
+  return await webpackPreprocessor(opts as any)(f);
+};
+
+export default defineConfig({
+  projectId: '6to9oy',
+  e2e: {
+    specPattern: `src/testing/e2e/cypress/scenarios/**/*.cy.{js,jsx,ts,tsx}`,
+    baseUrl: process.env.BASE_URL ?? 'http://127.0.0.1:8080',
+    supportFile: 'src/testing/e2e/cypress/support/index.ts',
+    chromeWebSecurity: false,
+    env: {
+      username: '[email protected]',
+      password: 'password',
+    },
+    setupNodeEvents: (on, _config) => {
+      on('file:preprocessor', async f => await webpackConfiguration(f));
+    },
+  },
+});