Log Streaming Bug Fix and Refactor #101

Aayyush · 2021-09-03T19:13:03Z

No description provided.

Aayyush · 2021-09-03T19:13:30Z

/ptal @msarvar @nishkrishnan #orchestration

msarvar · 2021-09-03T19:51:03Z

server/controllers/logstreaming_controller.go


 	pull := pullInfo.String()

-	err = j.ProjectCommandOutputHandler.Receive(pull, func(msg string) error {
+	// Buffer size = 10 allows for any dealys in ws connection message.
+	wsChannel := make(chan string, 10)


I'm not sure why do we need buffer size here? What delay in connection impacts the buffer?

When we get a new message in the ProjectCmdOutput channel, we iterate through all the channels in ReceiverBuffer and forward the message. However, if a unbuffered channel is not ready to receive msg, we delete the chan here:

atlantis/server/handlers/project_command_output_handler.go

Lines 120 to 127 in 22f56bb

for ch := range p.receiverBuffers[pull] {

select {

case ch <- line:

default:

// Buffered chan of size 10. Delete chan if still blocking

delete(p.receiverBuffers[pull], ch)

}

}

This is what happens when the UI gets stuck. When one of the message takes longer to send (callback function) through the ws and the receiving channel is not ready to receive the message (shown below), we just delete the channel which causes the UI to get stuck since there's no channel that's forwarding the messages to the UI.

atlantis/server/handlers/project_command_output_handler.go

Lines 69 to 73 in 22f56bb

for msg := range ch {

if err := callback(msg); err != nil {

return err

}

}

Making the channel buffered (size 10 for example) allows the writeLogLine() method to keep on pushing messages even if the channel is not ready to receive the messages yet. If it's still not ready after 10 enqued messaged, we could assume that something's wrong with the ch and delete the channel. We could increase the buffer size imo

This was the fix to the UI bug btw 😅

Hm, ok. If that's the case we can revert this value back to 1000 as it was in the original code. And we can revisit the implementation later. Receiving part of log streaming is a bit too complicated for what it is doing.

True! I'll revisit the architecture in a later PR.

nishkrishnan

Nice work!

server/handlers/project_command_output_handler.go

nishkrishnan · 2021-09-03T19:53:22Z

server/controllers/logstreaming_controller.go

+		for {
+			_, _, err := c.ReadMessage()
+			if err != nil {
+				j.Logger.Warn("Failed to read WS message: %s", err)
+				return
+			}
+		}


I'm not sure I understand the point of this

Apparently, there needs to be a reader in order to listen for events from the ws client. Since we're using the close handler here:

atlantis/server/controllers/logstreaming_controller.go

Lines 140 to 145 in 22f56bb

c.SetCloseHandler(func(code int, text string) error {

// Close the channnel after websocket connection closed.

// Will gracefully exit the ProjectCommandOutputHandler.Receive() call and cleanup.

close(wsChannel)

return nil

})

Removing the reader makes the closeHandler() non-functional.

server/controllers/logstreaming_controller.go

nishkrishnan · 2021-09-03T20:19:27Z

server/handlers/project_command_output_handler.go

 	if p.receiverBuffers[pull] == nil {
 		p.receiverBuffers[pull] = map[chan string]bool{}
 	}
 	p.receiverBuffers[pull][ch] = true
-	p.controllerBufferLock.Unlock()
-	return ch
+	p.receiverBuffersLock.Unlock()


There are so many locks being used everywhere it makes everything kind of difficult to understand/read. Not necessary to change as part of this PR, but as a thought exercise it would be good to think about basically having a single thread manage receiver buffers and projectBuffers and basically using channels to communicate between that thread and other threads thereby eliminating the need to have locks everywhere.

I wonder if we can implement something more event driven? We can register new webhooks to output_handler instances and emit any new requests to that webhook that should remove the need to using channels to sending messages to the webhook sessions.

Hmm that's true. I was trying to fix what was already working but I could definitely work on some refactoring and making the code-flow more understandable. I'll work on it after this PR.

server/handlers/project_command_output_handler.go

nishkrishnan · 2021-09-03T20:22:55Z

server/handlers/project_command_output_handler.go

+
+	// Avoid deadlock when projectOutputBuffer size is greater than the channel (currently set to 1000)
+	// Running this as a goroutine allows for the channel to be read in callback
+	go p.addChan(ch, projectInfo)


note: I think my suggestion would eliminate the need for this. Kinda weird to be creating a new thread for every ws conn just to register it.

Yup, will rethink about how we do this in a later PR.

server/handlers/project_command_output_handler.go

msarvar · 2021-09-08T18:30:31Z

server/controllers/logstreaming_controller.go

-	defer c.Close()
+	// Buffer size set to 1000 to ensure messages get queued (upto 1000) if the receiverCh is not ready to
+	// receive messages before the channel is closed and resources cleaned up.
+	receiverChan := make(chan string, 1000)


small nit: don't need to suffix the variable with its type.

msarvar · 2021-09-08T18:31:32Z

server/handlers/project_command_output_handler.go

@@ -29,7 +31,7 @@ type ProjectCommandOutputHandler interface {
 	Clear(ctx models.ProjectCommandContext)

 	// Receive will create a channel for projectPullInfo and run a callback function argument when the new channel receives a message.
-	Receive(projectInfo string, callback func(msg string) error) error
+	Receive(projectInfo string, ch chan string, callback func(msg string) error) error


You can probably remove argument names from the interface. Or update the ch to receiver

msarvar

Looks good, couple nits

Aayyush added 2 commits September 2, 2021 17:31

Fixing bugs and refactoring

2c2bd0b

Fixing tests

a54669b

Fixing lint error

22f56bb

msarvar reviewed Sep 3, 2021

View reviewed changes

nishkrishnan previously approved these changes Sep 3, 2021

View reviewed changes

Addressing comments

0e80e19

Aayyush dismissed nishkrishnan’s stale review via 0e80e19 September 7, 2021 18:25

msarvar reviewed Sep 8, 2021

View reviewed changes

msarvar previously approved these changes Sep 8, 2021

View reviewed changes

Addressing comments

5e12473

Aayyush dismissed msarvar’s stale review via 5e12473 September 9, 2021 17:17

msarvar approved these changes Sep 9, 2021

View reviewed changes

Aayyush merged commit 73abc76 into release-v0.17.1-lyft.1 Sep 9, 2021

Aayyush deleted the aayush/logstreaming branch September 9, 2021 17:28

msarvar pushed a commit that referenced this pull request Sep 27, 2021

Log Streaming Bug Fix and Refactor (#101)

1cfeb2e

Aayyush added a commit that referenced this pull request Dec 9, 2021

Log Streaming Bug Fix and Refactor (#101)

4fc5e78

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Log Streaming Bug Fix and Refactor #101

Log Streaming Bug Fix and Refactor #101

Aayyush commented Sep 3, 2021 •

edited

Loading

Aayyush commented Sep 3, 2021

msarvar Sep 3, 2021

Aayyush Sep 3, 2021

Aayyush Sep 3, 2021

msarvar Sep 7, 2021

Aayyush Sep 7, 2021

nishkrishnan left a comment

nishkrishnan Sep 3, 2021

Aayyush Sep 7, 2021

nishkrishnan Sep 3, 2021

msarvar Sep 7, 2021

Aayyush Sep 7, 2021

nishkrishnan Sep 3, 2021

Aayyush Sep 7, 2021

msarvar Sep 8, 2021

Aayyush Sep 9, 2021

msarvar Sep 8, 2021

Aayyush Sep 9, 2021

msarvar left a comment

	for ch := range p.receiverBuffers[pull] {
	select {
	case ch <- line:
	default:
	// Buffered chan of size 10. Delete chan if still blocking
	delete(p.receiverBuffers[pull], ch)
	}
	}

	for msg := range ch {
	if err := callback(msg); err != nil {
	return err
	}
	}

	c.SetCloseHandler(func(code int, text string) error {
	// Close the channnel after websocket connection closed.
	// Will gracefully exit the ProjectCommandOutputHandler.Receive() call and cleanup.
	close(wsChannel)
	return nil
	})

Log Streaming Bug Fix and Refactor #101

Log Streaming Bug Fix and Refactor #101

Conversation

Aayyush commented Sep 3, 2021 • edited Loading

Aayyush commented Sep 3, 2021

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

nishkrishnan left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

msarvar left a comment

Choose a reason for hiding this comment

Aayyush commented Sep 3, 2021 •

edited

Loading