Function TestServer_ForwarderContinuesAfterError

agent/boundarylogproxy/proxy_test.go:321–404

(t *testing.T)

}

func TestServer_ForwarderContinuesAfterError(t *testing.T) {
	t.Parallel()

	socketPath := filepath.Join(testutil.TempDirUnixSocket(t), "boundary.sock")
	srv := boundarylogproxy.NewServer(testutil.Logger(t), socketPath, prometheus.NewRegistry())

	err := srv.Start()
	require.NoError(t, err)
	t.Cleanup(func() { require.NoError(t, srv.Close()) })

	reportNotify := make(chan struct{}, 1)
	reporter := &fakeReporter{
		// Simulate an error on the first call.
		err:     context.DeadlineExceeded,
		errOnce: true,
		reportCb: func() {
			reportNotify <- struct{}{}
		},
	}

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	forwarderDone := make(chan error, 1)
	go func() {
		forwarderDone <- srv.RunForwarder(ctx, reporter)
	}()

	conn, err := net.Dial("unix", socketPath)
	require.NoError(t, err)
	defer conn.Close()

	// Send the first message to be processed and wait for failure.
	req1 := &agentproto.ReportBoundaryLogsRequest{
		Logs: []*agentproto.BoundaryLog{
			{
				Allowed: true,
				Time:    timestamppb.Now(),
				Resource: &agentproto.BoundaryLog_HttpRequest_{
					HttpRequest: &agentproto.BoundaryLog_HttpRequest{
						Method: "GET",
						Url:    "https://example.com/first",
					},
				},
			},
		},
	}
	sendLogs(t, conn, req1)

	select {
	case <-reportNotify:
	case <-time.After(testutil.WaitShort):
		t.Fatal("timed out waiting for first message to be processed")
	}

	// Send the second message, which should succeed.
	req2 := &agentproto.ReportBoundaryLogsRequest{
		Logs: []*agentproto.BoundaryLog{
			{