feat(logs): allow setting max log size (#244)

cmd/vela-worker/exec.go

@@ -47,16 +47,17 @@ func (w *Worker) exec(index int) error {
 	//
 	// https://godoc.org/github.com/go-vela/worker/executor#New
 	_executor, err := executor.New(&executor.Setup{
-		Driver:    w.Config.Executor.Driver,
-		LogMethod: w.Config.Executor.LogMethod,
-		Client:    w.VelaClient,
-		Hostname:  w.Config.API.Address.Hostname(),
-		Runtime:   w.Runtime,
-		Build:     item.Build,
-		Pipeline:  item.Pipeline.Sanitize(w.Config.Runtime.Driver),
-		Repo:      item.Repo,
-		User:      item.User,
-		Version:   v.Semantic(),
+		Driver:     w.Config.Executor.Driver,
+		LogMethod:  w.Config.Executor.LogMethod,
+		MaxLogSize: w.Config.Executor.MaxLogSize,
+		Client:     w.VelaClient,
+		Hostname:   w.Config.API.Address.Hostname(),
+		Runtime:    w.Runtime,
+		Build:      item.Build,
+		Pipeline:   item.Pipeline.Sanitize(w.Config.Runtime.Driver),
+		Repo:       item.Repo,
+		User:       item.User,
+		Version:    v.Semantic(),
 	})
 
 	// add the executor to the worker

cmd/vela-worker/run.go

@@ -95,8 +95,9 @@ func run(c *cli.Context) error {
 			CheckIn: c.Duration("checkIn"),
 			// executor configuration
 			Executor: &executor.Setup{
-				Driver:    c.String("executor.driver"),
-				LogMethod: c.String("executor.log_method"),
+				Driver:     c.String("executor.driver"),
+				LogMethod:  c.String("executor.log_method"),
+				MaxLogSize: c.Uint("executor.max_log_size"),
 			},
 			// logger configuration
 			Logger: &Logger{

executor/executor_test.go

@@ -45,6 +45,7 @@ func TestExecutor_New(t *testing.T) {
 		linux.WithBuild(_build),
 		linux.WithHostname("localhost"),
 		linux.WithLogMethod("byte-chunks"),
+		linux.WithMaxLogSize(2097152),
 		linux.WithPipeline(_pipeline),
 		linux.WithRepo(_repo),
 		linux.WithRuntime(_runtime),
@@ -93,15 +94,16 @@ func TestExecutor_New(t *testing.T) {
 		{
 			failure: false,
 			setup: &Setup{
-				Build:     _build,
-				Client:    _client,
-				Driver:    constants.DriverLinux,
-				LogMethod: "byte-chunks",
-				Pipeline:  _pipeline,
-				Repo:      _repo,
-				Runtime:   _runtime,
-				User:      _user,
-				Version:   "v1.0.0",
+				Build:      _build,
+				Client:     _client,
+				Driver:     constants.DriverLinux,
+				LogMethod:  "byte-chunks",
+				MaxLogSize: 2097152,
+				Pipeline:   _pipeline,
+				Repo:       _repo,
+				Runtime:    _runtime,
+				User:       _user,
+				Version:    "v1.0.0",
 			},
 			want: _linux,
 		},

executor/flags.go

@@ -49,4 +49,10 @@ var Flags = []cli.Flag{
 		Usage:    "method used to publish logs to the server - options: (byte-chunks|time-chunks)",
 		Value:    "byte-chunks",
 	},
+	&cli.UintFlag{
+		EnvVars:  []string{"VELA_EXECUTOR_MAX_LOG_SIZE", "EXECUTOR_MAX_LOG_SIZE"},
+		FilePath: "/vela/executor/max_log_size",
+		Name:     "executor.max_log_size",
+		Usage:    "maximum log size (in bytes)",
+	},
 }

executor/linux/linux.go

@@ -30,12 +30,13 @@ type (
 		secret *secretSvc
 
 		// private fields
-		init      *pipeline.Container
-		logger    *logrus.Entry
-		logMethod string
-		build     *library.Build
-		pipeline  *pipeline.Build
-		repo      *library.Repo
+		init       *pipeline.Container
+		logger     *logrus.Entry
+		logMethod  string
+		maxLogSize uint
+		build      *library.Build
+		pipeline   *pipeline.Build
+		repo       *library.Repo
 		// nolint: structcheck,unused // ignore false positives
 		secrets     sync.Map
 		services    sync.Map

executor/linux/opts.go

@@ -59,6 +59,18 @@ func WithLogMethod(method string) Opt {
 	}
 }
 
+// WithMaxLogSize set the maximum log size (in bytes) in the client.
+func WithMaxLogSize(size uint) Opt {
+	logrus.Trace("configuring maximum log size in linux client")
+
+	return func(c *client) error {
+		// set the maximum log size in the client
+		c.maxLogSize = size
+
+		return nil
+	}
+}
+
 // WithHostname sets the hostname in the client.
 func WithHostname(hostname string) Opt {
 	logrus.Trace("configuring hostname in linux client")

executor/linux/opts_test.go

@@ -109,6 +109,41 @@ func TestLinux_Opt_WithLogMethod(t *testing.T) {
 	}
 }
 
+func TestLinux_Opt_WithMaxLogSize(t *testing.T) {
+	// setup tests
+	tests := []struct {
+		failure    bool
+		maxLogSize uint
+	}{
+		{
+			failure:    false,
+			maxLogSize: 2097152,
+		},
+	}
+
+	// run tests
+	for _, test := range tests {
+		_engine, err := New(
+			WithMaxLogSize(test.maxLogSize),
+		)
+
+		if test.failure {
+			if err == nil {
+				t.Errorf("WithMaxLogSize should have returned err")
+			}
+
+			continue
+		}
+
+		if err != nil {
+			t.Errorf("WithMaxLogSize returned err: %v", err)
+		}
+
+		if !reflect.DeepEqual(_engine.maxLogSize, test.maxLogSize) {
+			t.Errorf("WithMaxLogSize is %v, want %v", _engine.maxLogSize, test.maxLogSize)
+		}
+	}
+}
 func TestLinux_Opt_WithHostname(t *testing.T) {
 	// setup tests
 	tests := []struct {

executor/linux/service.go

@@ -197,6 +197,13 @@ func (c *client) StreamService(ctx context.Context, ctn *pipeline.Container) err
 			return
 		}
 
+		// don't attempt last upload if log size exceeded
+		if c.maxLogSize > 0 && uint(len(data)) >= c.maxLogSize {
+			logger.Trace("maximum log size reached")
+
+			return
+		}
+
 		// overwrite the existing log with all bytes
 		//
 		// https://pkg.go.dev/github.com/go-vela/types/library?tab=doc#Log.SetData
@@ -277,6 +284,13 @@ func (c *client) StreamService(ctx context.Context, ctn *pipeline.Container) err
 						// flush the buffer of logs
 						logs.Reset()
 					}
+
+					// check whether we've reached the maximum log size
+					if c.maxLogSize > 0 && uint(len(_log.GetData())) >= c.maxLogSize {
+						logger.Trace("maximum log size reached")
+
+						return
+					}
 				}
 			}
 		}()
@@ -330,6 +344,13 @@ func (c *client) StreamService(ctx context.Context, ctn *pipeline.Container) err
 				// flush the buffer of logs
 				logs.Reset()
 			}
+
+			// check whether we've reached the maximum log size
+			if c.maxLogSize > 0 && uint(len(_log.GetData())) >= c.maxLogSize {
+				logger.Trace("maximum log size reached")
+
+				break
+			}
 		}
 
 		logger.Info("finished streaming logs")

executor/linux/step.go

@@ -232,6 +232,13 @@ func (c *client) StreamStep(ctx context.Context, ctn *pipeline.Container) error
 			return
 		}
 
+		// don't attempt last upload if log size exceeded
+		if c.maxLogSize > 0 && uint(len(data)) >= c.maxLogSize {
+			logger.Trace("maximum log size reached")
+
+			return
+		}
+
 		// overwrite the existing log with all bytes
 		//
 		// https://pkg.go.dev/github.com/go-vela/types/library?tab=doc#Log.SetData
@@ -312,6 +319,13 @@ func (c *client) StreamStep(ctx context.Context, ctn *pipeline.Container) error
 						// flush the buffer of logs
 						logs.Reset()
 					}
+
+					// check whether we've reached the maximum log size
+					if c.maxLogSize > 0 && uint(len(_log.GetData())) >= c.maxLogSize {
+						logger.Trace("maximum log size reached")
+
+						return
+					}
 				}
 			}
 		}()
@@ -365,6 +379,13 @@ func (c *client) StreamStep(ctx context.Context, ctn *pipeline.Container) error
 				// flush the buffer of logs
 				logs.Reset()
 			}
+
+			// check whether we've reached the maximum log size
+			if c.maxLogSize > 0 && uint(len(_log.GetData())) >= c.maxLogSize {
+				logger.Trace("maximum log size reached")
+
+				break
+			}
 		}
 
 		logger.Info("finished streaming logs")

executor/linux/step_test.go

@@ -323,6 +323,10 @@ func TestLinux_StreamStep(t *testing.T) {
 	_build := testBuild()
 	_repo := testRepo()
 	_user := testUser()
+	_logs := new(library.Log)
+
+	// fill log with bytes
+	_logs.SetData(make([]byte, 1000))
 
 	gin.SetMode(gin.TestMode)
 
@@ -346,6 +350,7 @@ func TestLinux_StreamStep(t *testing.T) {
 	}{
 		{ // init step container
 			failure: false,
+			logs:    _logs,
 			container: &pipeline.Container{
 				ID:          "step_github_octocat_1_init",
 				Directory:   "/vela/src/github.com/github/octocat",
@@ -358,6 +363,7 @@ func TestLinux_StreamStep(t *testing.T) {
 		},
 		{ // basic step container
 			failure: false,
+			logs:    _logs,
 			container: &pipeline.Container{
 				ID:          "step_github_octocat_1_echo",
 				Directory:   "/vela/src/github.com/github/octocat",
@@ -370,6 +376,7 @@ func TestLinux_StreamStep(t *testing.T) {
 		},
 		{ // step container with name not found
 			failure: true,
+			logs:    _logs,
 			container: &pipeline.Container{
 				ID:          "step_github_octocat_1_notfound",
 				Directory:   "/vela/src/github.com/github/octocat",
@@ -382,6 +389,7 @@ func TestLinux_StreamStep(t *testing.T) {
 		},
 		{ // empty step container
 			failure:   true,
+			logs:      _logs,
 			container: new(pipeline.Container),
 		},
 	}
@@ -391,6 +399,7 @@ func TestLinux_StreamStep(t *testing.T) {
 		_engine, err := New(
 			WithBuild(_build),
 			WithPipeline(new(pipeline.Build)),
+			WithMaxLogSize(10),
 			WithRepo(_repo),
 			WithRuntime(_runtime),
 			WithUser(_user),

executor/setup.go

@@ -32,6 +32,8 @@ type Setup struct {
 	Driver string
 	// specifies the executor method used to publish logs
 	LogMethod string
+	// specifies the maximum log size
+	MaxLogSize uint
 	// specifies the executor hostname
 	Hostname string
 	// specifies the executor version
@@ -72,6 +74,7 @@ func (s *Setup) Linux() (Engine, error) {
 	return linux.New(
 		linux.WithBuild(s.Build),
 		linux.WithLogMethod(s.LogMethod),
+		linux.WithMaxLogSize(s.MaxLogSize),
 		linux.WithHostname(s.Hostname),
 		linux.WithPipeline(s.Pipeline),
 		linux.WithRepo(s.Repo),