Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

fix race condition in Test_StartStop #1700

Merged
merged 2 commits into from
Oct 13, 2023
Merged

fix race condition in Test_StartStop #1700

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
067acff
fix race condition in Test_StartStop
gflarity Sep 29, 2023
2cd8362
Merge branch 'main' into log_producer_race
mdelapenya Oct 13, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
69 changes: 55 additions & 14 deletions logconsumer_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,19 +22,35 @@ const lastMessage = "DONE"

type TestLogConsumer struct {
Msgs []string
Ack chan bool
Done chan bool

// Accepted provides a blocking way of ensuring the logs messages have been consumed.
// This allows for proper synchronization during Test_StartStop in particular.
// Please see func devNullAcceptorChan if you're not interested in this synchronization.
Accepted chan string
mdelapenya marked this conversation as resolved.
Show resolved Hide resolved
}

func (g *TestLogConsumer) Accept(l Log) {
s := string(l.Content)
if s == fmt.Sprintf("echo %s\n", lastMessage) {
g.Ack <- true
g.Done <- true
return
}

g.Accepted <- s
g.Msgs = append(g.Msgs, s)
}

// devNullAcceptorChan returns string channel that essentially sends all strings to dev null
func devNullAcceptorChan() chan string {
c := make(chan string)
go func(c <-chan string) {
for range c {
// do nothing, just pull off channel
}
}(c)
return c
}

func Test_LogConsumerGetsCalled(t *testing.T) {
ctx := context.Background()
req := ContainerRequest{
Expand All @@ -58,8 +74,9 @@ func Test_LogConsumerGetsCalled(t *testing.T) {
require.NoError(t, err)

g := TestLogConsumer{
Msgs: []string{},
Ack: make(chan bool),
Msgs: []string{},
Done: make(chan bool),
Accepted: devNullAcceptorChan(),
}

c.FollowOutput(&g)
Expand All @@ -77,7 +94,7 @@ func Test_LogConsumerGetsCalled(t *testing.T) {
require.NoError(t, err)

select {
case <-g.Ack:
case <-g.Done:
case <-time.After(5 * time.Second):
t.Fatal("never received final log message")
}
Expand Down Expand Up @@ -174,8 +191,16 @@ func Test_MultipleLogConsumers(t *testing.T) {
ep, err := c.Endpoint(ctx, "http")
require.NoError(t, err)

first := TestLogConsumer{Msgs: []string{}, Ack: make(chan bool)}
second := TestLogConsumer{Msgs: []string{}, Ack: make(chan bool)}
first := TestLogConsumer{
Msgs: []string{},
Done: make(chan bool),
Accepted: devNullAcceptorChan(),
}
second := TestLogConsumer{
Msgs: []string{},
Done: make(chan bool),
Accepted: devNullAcceptorChan(),
}

c.FollowOutput(&first)
c.FollowOutput(&second)
Expand All @@ -189,8 +214,8 @@ func Test_MultipleLogConsumers(t *testing.T) {
_, err = http.Get(ep + "/stdout?echo=" + lastMessage)
require.NoError(t, err)

<-first.Ack
<-second.Ack
<-first.Done
<-second.Done
assert.Nil(t, c.StopLogProducer())

assert.Equal(t, []string{"ready\n", "echo mlem\n"}, first.Msgs)
Expand Down Expand Up @@ -220,27 +245,38 @@ func Test_StartStop(t *testing.T) {
ep, err := c.Endpoint(ctx, "http")
require.NoError(t, err)

g := TestLogConsumer{Msgs: []string{}, Ack: make(chan bool)}

g := TestLogConsumer{
Msgs: []string{},
Done: make(chan bool),
Accepted: make(chan string),
}
c.FollowOutput(&g)

require.NoError(t, c.StopLogProducer(), "nothing should happen even if the producer is not started")

require.NoError(t, c.StartLogProducer(ctx))
require.Equal(t, <-g.Accepted, "ready\n")

require.Error(t, c.StartLogProducer(ctx), "log producer is already started")

_, err = http.Get(ep + "/stdout?echo=mlem")
require.NoError(t, err)
require.Equal(t, <-g.Accepted, "echo mlem\n")

require.NoError(t, c.StopLogProducer())

require.NoError(t, c.StartLogProducer(ctx))
require.Equal(t, <-g.Accepted, "ready\n")
require.Equal(t, <-g.Accepted, "echo mlem\n")

_, err = http.Get(ep + "/stdout?echo=mlem2")
require.NoError(t, err)
require.Equal(t, <-g.Accepted, "echo mlem2\n")

_, err = http.Get(ep + "/stdout?echo=" + lastMessage)
require.NoError(t, err)

<-g.Ack
<-g.Done
// Do not close producer here, let's delegate it to c.Terminate

assert.Equal(t, []string{
Expand Down Expand Up @@ -333,7 +369,12 @@ func TestContainerLogWithErrClosed(t *testing.T) {
t.Fatal(err)
}

var consumer TestLogConsumer
consumer := TestLogConsumer{
Msgs: []string{},
Done: make(chan bool),
Accepted: devNullAcceptorChan(),
}

if err = nginx.StartLogProducer(ctx); err != nil {
t.Fatal(err)
}
Expand Down