From 990f62cc6900861446119b534b36c94712fac96c Mon Sep 17 00:00:00 2001
From: dave vader <48764154+plyr4@users.noreply.github.com>
Date: Mon, 12 Dec 2022 11:22:51 -0600
Subject: [PATCH] Revert "fix: Allow log streaming to take longer than build
 execution (#390)"

This reverts commit b50e19fb68488f990fb61ca82c774e41fb19263e.
---
 cmd/vela-worker/exec.go          |  12 ---
 cmd/vela-worker/run.go           |   1 -
 executor/flags.go                |   8 --
 executor/linux/build.go          |  12 +--
 executor/linux/build_test.go     |   1 -
 executor/linux/linux.go          |   2 -
 executor/linux/opts.go           |  13 ---
 executor/linux/opts_test.go      |  41 --------
 executor/setup.go                |   5 -
 executor/setup_test.go           |   2 -
 internal/context/context.go      |  49 ----------
 internal/context/context_test.go | 156 -------------------------------
 internal/context/doc.go          |  10 --
 13 files changed, 2 insertions(+), 310 deletions(-)
 delete mode 100644 internal/context/context.go
 delete mode 100644 internal/context/context_test.go
 delete mode 100644 internal/context/doc.go

diff --git a/cmd/vela-worker/exec.go b/cmd/vela-worker/exec.go
index 10411047..45f2f039 100644
--- a/cmd/vela-worker/exec.go
+++ b/cmd/vela-worker/exec.go
@@ -6,7 +6,6 @@ package main
 
 import (
 	"context"
-	"sync"
 	"time"
 
 	"github.com/go-vela/worker/executor"
@@ -76,7 +75,6 @@ func (w *Worker) exec(index int) error {
 		Driver:              w.Config.Executor.Driver,
 		LogMethod:           w.Config.Executor.LogMethod,
 		MaxLogSize:          w.Config.Executor.MaxLogSize,
-		LogStreamingTimeout: w.Config.Executor.LogStreamingTimeout,
 		EnforceTrustedRepos: w.Config.Executor.EnforceTrustedRepos,
 		PrivilegedImages:    w.Config.Runtime.PrivilegedImages,
 		Client:              w.VelaClient,
@@ -110,13 +108,7 @@ func (w *Worker) exec(index int) error {
 	timeoutCtx, timeout := context.WithTimeout(buildCtx, t)
 	defer timeout()
 
-	// This WaitGroup delays calling DestroyBuild until the StreamBuild goroutine finishes.
-	var wg sync.WaitGroup
-
 	defer func() {
-		// if exec() exits before starting StreamBuild, this returns immediately.
-		wg.Wait()
-
 		logger.Info("destroying build")
 
 		// destroy the build with the executor (pass a background
@@ -145,12 +137,8 @@ func (w *Worker) exec(index int) error {
 		return nil
 	}
 
-	// add StreamBuild goroutine to WaitGroup
-	wg.Add(1)
-
 	// log/event streaming uses buildCtx so that it is not subject to the timeout.
 	go func() {
-		defer wg.Done()
 		logger.Info("streaming build logs")
 		// execute the build with the executor
 		err = _executor.StreamBuild(buildCtx)
diff --git a/cmd/vela-worker/run.go b/cmd/vela-worker/run.go
index 482911e0..2fadcddc 100644
--- a/cmd/vela-worker/run.go
+++ b/cmd/vela-worker/run.go
@@ -96,7 +96,6 @@ func run(c *cli.Context) error {
 				Driver:              c.String("executor.driver"),
 				LogMethod:           c.String("executor.log_method"),
 				MaxLogSize:          c.Uint("executor.max_log_size"),
-				LogStreamingTimeout: c.Duration("executor.log_streaming_timeout"),
 				EnforceTrustedRepos: c.Bool("executor.enforce-trusted-repos"),
 			},
 			// logger configuration
diff --git a/executor/flags.go b/executor/flags.go
index 63de3dc6..95c96cfc 100644
--- a/executor/flags.go
+++ b/executor/flags.go
@@ -5,8 +5,6 @@
 package executor
 
 import (
-	"time"
-
 	"github.com/go-vela/types/constants"
 
 	"github.com/urfave/cli/v2"
@@ -39,12 +37,6 @@ var Flags = []cli.Flag{
 		Name:     "executor.max_log_size",
 		Usage:    "maximum log size (in bytes)",
 	},
-	&cli.DurationFlag{
-		EnvVars: []string{"WORKER_LOG_STREAMING_TIMEOUT", "VELA_LOG_STREAMING_TIMEOUT", "LOG_STREAMING_TIMEOUT"},
-		Name:    "executor.log_streaming_timeout",
-		Usage:   "maximum amount of time to wait for log streaming after build completes",
-		Value:   5 * time.Minute,
-	},
 	&cli.BoolFlag{
 		EnvVars:  []string{"VELA_EXECUTOR_ENFORCE_TRUSTED_REPOS", "EXECUTOR_ENFORCE_TRUSTED_REPOS"},
 		FilePath: "/vela/executor/enforce_trusted_repos",
diff --git a/executor/linux/build.go b/executor/linux/build.go
index f57656a9..8fa38f24 100644
--- a/executor/linux/build.go
+++ b/executor/linux/build.go
@@ -16,7 +16,6 @@ import (
 	"github.com/go-vela/types/constants"
 	"github.com/go-vela/types/library"
 	"github.com/go-vela/worker/internal/build"
-	context2 "github.com/go-vela/worker/internal/context"
 	"github.com/go-vela/worker/internal/image"
 	"github.com/go-vela/worker/internal/step"
 )
@@ -567,15 +566,10 @@ func (c *client) ExecBuild(ctx context.Context) error {
 // StreamBuild receives a StreamRequest and then
 // runs StreamService or StreamStep in a goroutine.
 func (c *client) StreamBuild(ctx context.Context) error {
-	// cancel streaming after a timeout once the build has finished
-	delayedCtx, cancelStreaming := context2.
-		WithDelayedCancelPropagation(ctx, c.logStreamingTimeout, "streaming", c.Logger)
-	defer cancelStreaming()
-
 	// create an error group with the parent context
 	//
 	// https://pkg.go.dev/golang.org/x/sync/errgroup?tab=doc#WithContext
-	streams, streamCtx := errgroup.WithContext(delayedCtx)
+	streams, streamCtx := errgroup.WithContext(ctx)
 
 	defer func() {
 		c.Logger.Trace("waiting for stream functions to return")
@@ -585,8 +579,6 @@ func (c *client) StreamBuild(ctx context.Context) error {
 			c.Logger.Errorf("error in a stream request, %v", err)
 		}
 
-		cancelStreaming()
-
 		c.Logger.Info("all stream functions have returned")
 	}()
 
@@ -615,7 +607,7 @@ func (c *client) StreamBuild(ctx context.Context) error {
 
 				return nil
 			})
-		case <-delayedCtx.Done():
+		case <-ctx.Done():
 			c.Logger.Debug("streaming context canceled")
 			// build done or canceled
 			return nil
diff --git a/executor/linux/build_test.go b/executor/linux/build_test.go
index 97a99136..42d465b3 100644
--- a/executor/linux/build_test.go
+++ b/executor/linux/build_test.go
@@ -1238,7 +1238,6 @@ func TestLinux_StreamBuild(t *testing.T) {
 				WithPipeline(_pipeline),
 				WithRepo(_repo),
 				WithRuntime(_runtime),
-				WithLogStreamingTimeout(1*time.Second),
 				WithUser(_user),
 				WithVelaClient(_client),
 				withStreamRequests(streamRequests),
diff --git a/executor/linux/linux.go b/executor/linux/linux.go
index 4a78d2fb..a2346731 100644
--- a/executor/linux/linux.go
+++ b/executor/linux/linux.go
@@ -7,7 +7,6 @@ package linux
 import (
 	"reflect"
 	"sync"
-	"time"
 
 	"github.com/go-vela/sdk-go/vela"
 	"github.com/go-vela/types/library"
@@ -35,7 +34,6 @@ type (
 		init                *pipeline.Container
 		logMethod           string
 		maxLogSize          uint
-		logStreamingTimeout time.Duration
 		privilegedImages    []string
 		enforceTrustedRepos bool
 		build               *library.Build
diff --git a/executor/linux/opts.go b/executor/linux/opts.go
index 1b143152..a75d0a42 100644
--- a/executor/linux/opts.go
+++ b/executor/linux/opts.go
@@ -6,7 +6,6 @@ package linux
 
 import (
 	"fmt"
-	"time"
 
 	"github.com/go-vela/sdk-go/vela"
 	"github.com/go-vela/types/library"
@@ -65,18 +64,6 @@ func WithMaxLogSize(size uint) Opt {
 	}
 }
 
-// WithLogStreamingTimeout sets the log streaming timeout in the executor client for Linux.
-func WithLogStreamingTimeout(timeout time.Duration) Opt {
-	return func(c *client) error {
-		c.Logger.Trace("configuring log streaming timeout in linux executor client")
-
-		// set the maximum log size in the client
-		c.logStreamingTimeout = timeout
-
-		return nil
-	}
-}
-
 // WithPrivilegedImages sets the privileged images in the executor client for Linux.
 func WithPrivilegedImages(images []string) Opt {
 	return func(c *client) error {
diff --git a/executor/linux/opts_test.go b/executor/linux/opts_test.go
index f4fe9efa..f01c6b87 100644
--- a/executor/linux/opts_test.go
+++ b/executor/linux/opts_test.go
@@ -8,7 +8,6 @@ import (
 	"net/http/httptest"
 	"reflect"
 	"testing"
-	"time"
 
 	"github.com/gin-gonic/gin"
 	"github.com/sirupsen/logrus"
@@ -162,46 +161,6 @@ func TestLinux_Opt_WithMaxLogSize(t *testing.T) {
 	}
 }
 
-func TestLinux_Opt_WithLogStreamingTimeout(t *testing.T) {
-	// setup tests
-	tests := []struct {
-		name                string
-		failure             bool
-		logStreamingTimeout time.Duration
-	}{
-		{
-			name:                "defined",
-			failure:             false,
-			logStreamingTimeout: 1 * time.Second,
-		},
-	}
-
-	// run tests
-	for _, test := range tests {
-		t.Run(test.name, func(t *testing.T) {
-			_engine, err := New(
-				WithLogStreamingTimeout(test.logStreamingTimeout),
-			)
-
-			if test.failure {
-				if err == nil {
-					t.Errorf("WithLogStreamingTimeout should have returned err")
-				}
-
-				return // continue to next test
-			}
-
-			if err != nil {
-				t.Errorf("WithLogStreamingTimeout returned err: %v", err)
-			}
-
-			if !reflect.DeepEqual(_engine.logStreamingTimeout, test.logStreamingTimeout) {
-				t.Errorf("WithLogStreamingTimeout is %v, want %v", _engine.logStreamingTimeout, test.logStreamingTimeout)
-			}
-		})
-	}
-}
-
 func TestLinux_Opt_WithPrivilegedImages(t *testing.T) {
 	// setup tests
 	tests := []struct {
diff --git a/executor/setup.go b/executor/setup.go
index 333a83d6..2b07f7ae 100644
--- a/executor/setup.go
+++ b/executor/setup.go
@@ -7,7 +7,6 @@ package executor
 import (
 	"fmt"
 	"strings"
-	"time"
 
 	"github.com/go-vela/sdk-go/vela"
 
@@ -41,9 +40,6 @@ type Setup struct {
 	LogMethod string
 	// specifies the maximum log size
 	MaxLogSize uint
-	// specifies how long to wait after the build finishes
-	// for log streaming to complete
-	LogStreamingTimeout time.Duration
 	// specifies a list of privileged images to use
 	PrivilegedImages []string
 	// configuration for enforcing that only trusted repos may run privileged images
@@ -89,7 +85,6 @@ func (s *Setup) Linux() (Engine, error) {
 		linux.WithBuild(s.Build),
 		linux.WithLogMethod(s.LogMethod),
 		linux.WithMaxLogSize(s.MaxLogSize),
-		linux.WithLogStreamingTimeout(s.LogStreamingTimeout),
 		linux.WithPrivilegedImages(s.PrivilegedImages),
 		linux.WithEnforceTrustedRepos(s.EnforceTrustedRepos),
 		linux.WithHostname(s.Hostname),
diff --git a/executor/setup_test.go b/executor/setup_test.go
index 615b9ccd..069993d1 100644
--- a/executor/setup_test.go
+++ b/executor/setup_test.go
@@ -7,7 +7,6 @@ package executor
 import (
 	"net/http/httptest"
 	"testing"
-	"time"
 
 	"github.com/gin-gonic/gin"
 	"github.com/google/go-cmp/cmp"
@@ -80,7 +79,6 @@ func TestExecutor_Setup_Linux(t *testing.T) {
 		linux.WithBuild(_build),
 		linux.WithLogMethod("byte-chunks"),
 		linux.WithMaxLogSize(2097152),
-		linux.WithLogStreamingTimeout(1*time.Second),
 		linux.WithHostname("localhost"),
 		linux.WithPipeline(_pipeline),
 		linux.WithRepo(_repo),
diff --git a/internal/context/context.go b/internal/context/context.go
deleted file mode 100644
index 04cea667..00000000
--- a/internal/context/context.go
+++ /dev/null
@@ -1,49 +0,0 @@
-// Copyright (c) 2022 Target Brands, Inc. All rights reserved.
-//
-// Use of this source code is governed by the LICENSE file in this repository.
-
-package context
-
-import (
-	"context"
-	"time"
-
-	"github.com/sirupsen/logrus"
-)
-
-func WithDelayedCancelPropagation(parent context.Context, timeout time.Duration, name string, logger *logrus.Entry) (context.Context, context.CancelFunc) {
-	ctx, cancel := context.WithCancel(context.Background())
-
-	go func() {
-		var timer *time.Timer
-
-		// start the timer once the parent context is canceled
-		select {
-		case <-parent.Done():
-			logger.Tracef("parent context is done, starting %s timer for %s", name, timeout)
-			timer = time.NewTimer(timeout)
-
-			break
-		case <-ctx.Done():
-			logger.Tracef("%s finished before the parent context", name)
-
-			return
-		}
-
-		// wait for the timer to elapse or the context to naturally finish.
-		select {
-		case <-timer.C:
-			logger.Tracef("%s timed out, propagating cancel to %s context", name, name)
-			cancel()
-
-			return
-		case <-ctx.Done():
-			logger.Tracef("%s finished, stopping timeout timer", name)
-			timer.Stop()
-
-			return
-		}
-	}()
-
-	return ctx, cancel
-}
diff --git a/internal/context/context_test.go b/internal/context/context_test.go
deleted file mode 100644
index 140fa5fb..00000000
--- a/internal/context/context_test.go
+++ /dev/null
@@ -1,156 +0,0 @@
-// Copyright (c) 2022 Target Brands, Inc. All rights reserved.
-//
-// Use of this source code is governed by the LICENSE file in this repository.
-
-package context
-
-import (
-	"context"
-	"errors"
-	"fmt"
-	"testing"
-	"time"
-
-	"github.com/sirupsen/logrus"
-	logrusTest "github.com/sirupsen/logrus/hooks/test"
-)
-
-// shortDuration copied from
-// https://github.com/golang/go/blob/go1.19.1/src/context/context_test.go#L45
-const shortDuration = 1 * time.Millisecond // a reasonable duration to block in a test
-
-// quiescent returns an arbitrary duration by which the program should have
-// completed any remaining work and reached a steady (idle) state.
-//
-// copied from https://github.com/golang/go/blob/go1.19.1/src/context/context_test.go#L49-L59
-func quiescent(t *testing.T) time.Duration {
-	deadline, ok := t.Deadline()
-	if !ok {
-		return 5 * time.Second
-	}
-
-	const arbitraryCleanupMargin = 1 * time.Second
-
-	return time.Until(deadline) - arbitraryCleanupMargin
-}
-
-// testCancelPropagated is a helper that tests deadline/timeouts.
-//
-// based on testDeadline from
-// https://github.com/golang/go/blob/go1.19.1/src/context/context_test.go#L272-L285
-func testCancelPropagated(c context.Context, name string, t *testing.T) {
-	t.Helper()
-	d := quiescent(t)
-
-	timer := time.NewTimer(d)
-	defer timer.Stop()
-
-	select {
-	case <-timer.C:
-		t.Fatalf("%s: context not timed out after %v", name, d)
-	case <-c.Done():
-	}
-	//if e := c.Err(); e != context.DeadlineExceeded { // original line
-	if e := c.Err(); !errors.Is(e, context.Canceled) { // delayedCancelPropagation triggers this instead
-		t.Errorf("%s: c.Err() == %v; want %v", name, e, context.Canceled)
-	}
-}
-
-func TestWithDelayedCancelPropagation(t *testing.T) {
-	parentLogger := logrus.New()
-	parentLogger.SetLevel(logrus.TraceLevel)
-
-	loggerHook := logrusTest.NewLocal(parentLogger)
-
-	nameArg := "streaming"
-
-	tests := []struct {
-		name           string
-		cancelParent   string // before, after, never
-		timeout        time.Duration
-		cancelCtxAfter time.Duration
-		lastLogMessage string
-	}{
-		{
-			name:           "cancel parent before call-child finishes before timeout",
-			cancelParent:   "before",
-			timeout:        shortDuration * 5,
-			cancelCtxAfter: shortDuration,
-			lastLogMessage: nameArg + " finished, stopping timeout timer",
-		},
-		{
-			name:           "cancel parent before call-child exceeds timeout",
-			cancelParent:   "before",
-			timeout:        shortDuration,
-			cancelCtxAfter: shortDuration * 5,
-			lastLogMessage: nameArg + " timed out, propagating cancel to " + nameArg + " context",
-		},
-		{
-			name:           "child finished before timeout and before parent cancel",
-			cancelParent:   "never",
-			timeout:        shortDuration * 5,
-			cancelCtxAfter: shortDuration,
-			lastLogMessage: nameArg + " finished before the parent context",
-		},
-		{
-			name:           "cancel parent after call-child finishes before timeout",
-			cancelParent:   "after",
-			timeout:        shortDuration * 5,
-			cancelCtxAfter: shortDuration,
-			lastLogMessage: nameArg + " finished, stopping timeout timer",
-		},
-		{
-			name:           "cancel parent after call-child exceeds timeout",
-			cancelParent:   "after",
-			timeout:        shortDuration,
-			cancelCtxAfter: shortDuration * 5,
-			lastLogMessage: nameArg + " timed out, propagating cancel to " + nameArg + " context",
-		},
-	}
-	for _, test := range tests {
-		t.Run(test.name, func(t *testing.T) {
-			defer loggerHook.Reset()
-
-			logger := parentLogger.WithFields(logrus.Fields{"test": test.name})
-
-			parentCtx, parentCtxCancel := context.WithCancel(context.Background())
-			defer parentCtxCancel() // handles test.CancelParent == "never" case
-
-			if test.cancelParent == "before" {
-				parentCtxCancel()
-			}
-
-			ctx, cancel := WithDelayedCancelPropagation(parentCtx, test.timeout, nameArg, logger)
-			defer cancel()
-
-			// test based on test for context.WithCancel
-			if got, want := fmt.Sprint(ctx), "context.Background.WithCancel"; got != want {
-				t.Errorf("ctx.String() = %q want %q", got, want)
-			}
-
-			if d := ctx.Done(); d == nil {
-				t.Errorf("ctx.Done() == %v want non-nil", d)
-			}
-			if e := ctx.Err(); e != nil {
-				t.Errorf("ctx.Err() == %v want nil", e)
-			}
-
-			if test.cancelParent == "after" {
-				parentCtxCancel()
-			}
-
-			go func() {
-				time.Sleep(test.cancelCtxAfter)
-				cancel()
-			}()
-
-			testCancelPropagated(ctx, "WithDelayedCancelPropagation", t)
-
-			time.Sleep(shortDuration)
-			lastLogEntry := loggerHook.LastEntry()
-			if lastLogEntry.Message != test.lastLogMessage {
-				t.Errorf("unexpected last log entry: want = %s ; got = %s", test.lastLogMessage, lastLogEntry.Message)
-			}
-		})
-	}
-}
diff --git a/internal/context/doc.go b/internal/context/doc.go
deleted file mode 100644
index c7c2f795..00000000
--- a/internal/context/doc.go
+++ /dev/null
@@ -1,10 +0,0 @@
-// Copyright (c) 2022 Target Brands, Inc. All rights reserved.
-//
-// Use of this source code is governed by the LICENSE file in this repository.
-
-// Package context provides context utilities.
-//
-// Usage:
-//
-//	import "github.com/go-vela/worker/internal/context"
-package context