enhancement: Make context handling for log streaming more explicit

[cognifloyd]

Part of go-vela/community#562

Overview

This PR comes from my question: "What is responsible for ensuring that all log streaming go routines have returned?"

It addresses two related issues:

• monitoring stream goroutines to make sure they exit, and do so consistently for Step/Detach Step/Service/secret.
• allow the log capture go routine to go (at least slightly) longer than the exec even if exec goes past the deadline and gets canceled.

Current state

Nothing is monitoring the log streaming go routines to make sure they exit. ExecStep/ExecService each run StreamStep/StreamService in an errgroup goroutine, but nothing monitors to make sure the goroutine cooperatively exits when the context is canceled (secret.exec+secret.stream are very similar but without using the errgroup). Detached Steps and Services cannot be monitored from within ExecStep/ExecService, so we need to monitor the errgroup somewhere else.

Passing the ctx from ExecBuild through ExecService/ExecStage/ExecStep/secret.exec is problematic because we need to apply a timeout/deadline to that execution, but we want logging capture to continue (at least a bit longer) even when the execution exceeds the deadline to ensure all the logs get captured.

This PR

The exec timeout/deadline context is derived from another context - we make that context have its own cancel function (not a deadline) and then pass that parent context down to the log streaming methods. To facilitate changing the context like that, we now call StreamService/StreamStep/secret.stream from within a new StreamBuild function that runs concurrently with ExecBuild. Exec* signals when to begin streaming logs for a given container using a streamRequests channel.

In other words, this PR implements the idea outlined in go-vela/community#562 (comment), but instead of adding the channel to various function signatures, this adds a streamRequests channel as an internal field on the linux/local executor engine.

How Streaming Starts

Here is the interface for StreamBuild():

worker/executor/engine.go

Lines 56 to 58 in 2f7fc2a

	// StreamBuild defines a function that receives a StreamRequest
	// and then runs StreamService or StreamStep in a goroutine.
	StreamBuild(context.Context) error

Here is where ExecStep() sends a message that StreamBuild() listens for (it is similar for service and secret):

worker/executor/linux/step.go

Lines 158 to 163 in 2f7fc2a

	// trigger StreamStep goroutine with logging context
	c.streamRequests <- message.StreamRequest{
	Key: "step",
	Stream: c.StreamStep,
	Container: ctn,
	}

Here is where StreamBuild() receives the message and calls StreamStep() (or StreamService() or secret.stream()):

worker/executor/linux/build.go

Lines 515 to 524 in 2f7fc2a

	case req := <-c.streamRequests:
	streams.Go(func() error {
	// update engine logger with step metadata
	//
	// https://pkg.go.dev/github.com/sirupsen/logrus?tab=doc#Entry.WithField
	logger := c.Logger.WithField(req.Key, req.Container.Name)
	logger.Debugf("streaming %s container %s", req.Key, req.Container.ID)
	err := req.Stream(streamCtx, req.Container)

Context overview

Here is the build context and the deadline context derived from it. Note that the build context still gets canceled to ensure we don't have zombie log streaming goroutines after the build completes.

worker/cmd/vela-worker/exec.go

Lines 96 to 103 in 2f7fc2a

	// create a background context
	buildCtx, done := context.WithCancel(context.Background())
	defer done()
	// add to the background context with a timeout
	// built in for ensuring a build doesn't run forever
	ctx, timeout := context.WithTimeout(buildCtx, t)
	defer timeout()

The deadline context is used in ExecBuild() (and most other executor.*Build() functions):

worker/cmd/vela-worker/exec.go

Lines 150 to 156 in 2f7fc2a

	logger.Info("executing build")
	// execute the build with the executor
	err = _executor.ExecBuild(ctx)
	if err != nil {
	logger.Errorf("unable to execute build: %v", err)
	return nil
	}

The build context gets passed to StreamBuild() so that it is not subject to the deadline. It is only subject to the end of the build (when exec() returns):

worker/cmd/vela-worker/exec.go

Lines 132 to 140 in 2f7fc2a

	// log streaming uses buildCtx so that it is not subject to the timeout.
	go func() {
	logger.Info("streaming build logs")
	// execute the build with the executor
	err = _executor.StreamBuild(buildCtx)
	if err != nil {
	logger.Errorf("unable to stream build logs: %v", err)
	}
	}()

Tests

Pretty much all new code is tested (or at least it registers in codecov).

Testing StreamBuild() was interesting because it is designed to run concurrently with AssembleBuild() (which calls secret.exec()) and ExecBuild() (which calls PlanService()/PlanStep() and ExecService()/ExecStep()). So, I used a goroutine to simulate/mock the primary bits of ExecBuild():

worker/executor/linux/build_test.go

Lines 727 to 745 in 2f7fc2a

	// simulate ExecBuild() which runs concurrently with StreamBuild()
	go func() {
	// ExecBuild calls PlanService()/PlanStep() before ExecService()/ExecStep()
	// (ExecStage() calls PlanStep() before ExecStep()).
	_engine.err = test.planFunc(_engine)(buildCtx, test.ctn)
	// ExecService()/ExecStep()/secret.exec() send this message
	streamRequests <- message.StreamRequest{
	Key: test.messageKey,
	Stream: test.streamFunc(_engine),
	Container: test.ctn,
	}
	// simulate exec build duration
	time.Sleep(100 * time.Microsecond)
	// signal the end of the build so StreamBuild can terminate
	done()
	}()

NOTE: The test I added for Linux.WithLogger() is slightly orthogonal to this PR, but it is still "logging" so it works to leave it here.

Utilities

In order to accomplish all of the above, I needed to add a type for the messages that go from ExecStep() and friends to StreamBuild(). I did that in the internal/message package.

worker/internal/message/stream.go

Lines 16 to 25 in 2f7fc2a

	// StreamRequest is the message used to begin streaming for a container
	// (requests goes from ExecService / ExecStep to StreamBuild in executor).
	type StreamRequest struct {
	// Key is either "service" or "step".
	Key string
	// Stream is either Engine.StreamService or Engine.StreamStep.
	Stream StreamFunc
	// Container is the container for the service or step to stream logs for.
	Container *pipeline.Container
	}

Initially I tried modifying the signature of various Exec* and Stream* methods, but that meant using an internal type on an exported interface, so I moved the stream to an internal property. Here is the streamRequests channel on the executor:

worker/executor/linux/linux.go

Line 46 in 2f7fc2a

streamRequests chan message.StreamRequest

Sadly, this broke some of the existing tests because reflect.DeepEqual() couldn't handle the channel. I ended up using cmp.Diff() with a custom Equal(a, b *client) to fix those tests. Here's the Equal function:

worker/executor/linux/linux.go

Lines 58 to 85 in 2f7fc2a

	// Equal returns true if the other client is the equivalent.
	func Equal(a, b *client) bool {
	// handle any nil comparisons
	if a == nil || b == nil {
	return a == nil && b == nil
	}
	return reflect.DeepEqual(a.Logger, b.Logger) &&
	reflect.DeepEqual(a.Vela, b.Vela) &&
	reflect.DeepEqual(a.Runtime, b.Runtime) &&
	reflect.DeepEqual(a.Secrets, b.Secrets) &&
	a.Hostname == b.Hostname &&
	a.Version == b.Version &&
	reflect.DeepEqual(a.init, b.init) &&
	a.logMethod == b.logMethod &&
	a.maxLogSize == b.maxLogSize &&
	reflect.DeepEqual(a.build, b.build) &&
	reflect.DeepEqual(a.pipeline, b.pipeline) &&
	reflect.DeepEqual(a.repo, b.repo) &&
	reflect.DeepEqual(&a.secrets, &b.secrets) &&
	reflect.DeepEqual(&a.services, &b.services) &&
	reflect.DeepEqual(&a.serviceLogs, &b.serviceLogs) &&
	reflect.DeepEqual(&a.steps, &b.steps) &&
	reflect.DeepEqual(&a.stepLogs, &b.stepLogs) &&
	// do not compare streamRequests channel
	reflect.DeepEqual(a.user, b.user) &&
	reflect.DeepEqual(a.err, b.err)
	}

And here's a test that uses it:

worker/executor/setup_test.go

Lines 114 to 118 in 2f7fc2a

	// Comparing with reflect.DeepEqual(x, y interface) panics due to the
	// unexported streamRequests channel.
	if diff := cmp.Diff(want, got, cmp.Comparer(linux.Equal)); diff != "" {
	t.Errorf("linux Engine mismatch (-want +got):\n%v", diff)
	}

Visual Overview

I have a map of methods in the Worker to help myself navigate through the callstack.

This is the relevant section that shows how we're moving the Stream*() calls from Exec*() (the ❌ (red x) marks where these are removed) to StreamBuild() which is now called in a goroutine (the top layer in the diagram).

[codecov]

Codecov Report

Merging #317 (ebd5738) into master (1518490) will increase coverage by 0.81%.
The diff coverage is 94.07%.

@@            Coverage Diff             @@
##           master     #317      +/-   ##
==========================================
+ Coverage   79.68%   80.49%   +0.81%     
==========================================
  Files          67       67              
  Lines        4962     5081     +119     
==========================================
+ Hits         3954     4090     +136     
+ Misses        862      847      -15     
+ Partials      146      144       -2     

Impacted Files	Coverage Δ
executor/local/build.go	77.27% <78.57%> (+0.14%)	⬆️
executor/linux/build.go	74.57% <90.90%> (+1.41%)	⬆️
executor/linux/linux.go	100.00% <100.00%> (ø)
executor/linux/opts.go	100.00% <100.00%> (+8.33%)	⬆️
executor/linux/secret.go	71.84% <100.00%> (+1.38%)	⬆️
executor/linux/service.go	63.06% <100.00%> (+0.52%)	⬆️
executor/linux/step.go	65.68% <100.00%> (+0.57%)	⬆️
executor/local/local.go	100.00% <100.00%> (ø)
executor/local/opts.go	100.00% <100.00%> (ø)
executor/local/service.go	93.87% <100.00%> (+3.06%)	⬆️
... and 7 more

[kneal]

LGTM 🐬

[cognifloyd]

I renamed logCtx to streamCtx because I will probably use StreamBuild to stream events as well as logs in the Kubernetes runtime (in a future PR).

I did the rename now since I was also merging in master, and that'll reduce the diff for the next PR without changing the diff/functionality of this one.

[kneal]

LGTM 🐬

[wass3r]

lgtm as well - thanks for all the work! if you want to make that one log level change, i think we're good.

i do see some inconsistencies with logging mentioned somewhere else too, sometimes using .WithFields and duplicating info in the actual message and sometimes not, but that's outside of the scope of this PR.

[cognifloyd]

@wass3r Fixed the log entry. :) This should be ready when you are.

[kneal]

LGTM 🐬