Refactor job controller and websocket logic.

[nishkrishnan]

A number of things are going on here:

• Removed the feature aware related items for log streaming work
• Created websocket package which handles writing to individual websockets
• Created Multiplexor which primarily should be all the controller interacts with and moved all primary logic there
• Created PartitionKeyGenerator for pulling the projectinfo which makes it generic.
• Modified Receive/removeChan -> Register/Deregister to remove the reading logic from the outputhandler since that struct has way too many responsibilities right now.

[msarvar]

Looks nice!

[Aayyush]

Could we use the writer.NewWriter() method instead of instantiating the upgrader here.

[nishkrishnan]

that method is unused. I'll remove it.

[Aayyush]

We probably need to assign the registry to the multiplexor? Don't we have a linter that checks for these kind of errors where the passed argument is not used in the method body?

[nishkrishnan]

Oops, classic. i don't think we have linters for that. Will fix.

[Aayyush]

I'm curious how the buffer would behave if there's new messages coming in to the projectOutputBuffers while we send messages into the ch channel. Since we only add the channel ch to the receiverBuffers[pull] later in the method, the writelogLine method would not forward the new messages into the channel neither will we have it copied over from the output buffer unless the buffer is being passed by reference. If it is being passed by reference should we just leave it inside the Rlock to ensure two goroutines don't read and write in the buffer at the same time?

atlantis/server/handlers/project_command_output_handler.go

Lines 154 to 166 in 60e3d59

	for ch := range p.receiverBuffers[pull] {
	select {
	case ch <- line:
	default:
	// Client ws conn could be closed in two ways:
	// 1. Client closes the conn gracefully -> the closeHandler() is executed which
	// closes the channel and cleans up resources.
	// 2. Client does not close the conn and the closeHandler() is not executed -> the
	// receiverChan will be blocking for N number of messages (equal to buffer size)
	// before we delete the channel and clean up the resources.
	delete(p.receiverBuffers[pull], ch)
	}
	}

[nishkrishnan]

it's not a reference, the underlying array gets copied. Realistically we should be using sync.Map here, not sure why we aren't. I plan on moving to that in a future PR.

The problem you're mentioning would happen regardless of whether it's in the Rlock since there is a chance we don't get the receiver buffer lock in this method and then lines are stored in the buffer but not forwarded to the channel. Something we'll have to fix with a future refactor to this. These two data objects should be accessed atomically imo.

[Aayyush]

Gotcha! Thanks for the clarification :)

[Aayyush]

Left a few comments but looks great overall 👍🏽 ! This will definitely help when we start working on log persistence :)