[Pipelines] Replace logger with zerolog (#670)

app/pipeline/logger/zerolog.go

@@ -0,0 +1,119 @@
+// Copyright 2023 Harness, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package logger
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/drone/runner-go/logger"
+	"github.com/rs/zerolog"
+	"github.com/rs/zerolog/log"
+)
+
+// WithWrappedZerolog adds a wrapped copy of the zerolog logger to the context.
+func WithWrappedZerolog(ctx context.Context) context.Context {
+	return logger.WithContext(
+		ctx,
+		&wrapZerolog{
+			inner: log.Ctx(ctx).With().Logger(),
+			err:   nil,
+		})
+}
+
+// WithUnwrappedZerolog adds an unwrapped copy of the zerolog logger to the context.
+func WithUnwrappedZerolog(ctx context.Context) context.Context {
+	// try to get the logger from the wrapped zerologger
+	if wrappedLogger, ok := logger.FromContext(ctx).(*wrapZerolog); ok {
+		return wrappedLogger.inner.WithContext(ctx)
+	}
+
+	// if there's no logger, fall-back to global logger instance
+	return log.Logger.WithContext(ctx)
+}
+
+// wrapZerolog wraps the zerolog logger to be used within drone packages.
+type wrapZerolog struct {
+	inner zerolog.Logger
+	err   error
+}
+
+func (w *wrapZerolog) WithError(err error) logger.Logger {
+	return &wrapZerolog{inner: w.inner, err: err}
+}
+
+func (w *wrapZerolog) WithField(key string, value interface{}) logger.Logger {
+	return &wrapZerolog{inner: w.inner.With().Str(key, fmt.Sprint(value)).Logger(), err: w.err}
+}
+
+func (w *wrapZerolog) Debug(args ...interface{}) {
+	w.inner.Debug().Err(w.err).Msg(fmt.Sprint(args...))
+}
+
+func (w *wrapZerolog) Debugf(format string, args ...interface{}) {
+	w.inner.Debug().Err(w.err).Msgf(format, args...)
+}
+
+func (w *wrapZerolog) Debugln(args ...interface{}) {
+	w.inner.Debug().Err(w.err).Msg(fmt.Sprintln(args...))
+}
+
+func (w *wrapZerolog) Error(args ...interface{}) {
+	w.inner.Error().Err(w.err).Msg(fmt.Sprint(args...))
+}
+
+func (w *wrapZerolog) Errorf(format string, args ...interface{}) {
+	w.inner.Error().Err(w.err).Msgf(format, args...)
+}
+
+func (w *wrapZerolog) Errorln(args ...interface{}) {
+	w.inner.Error().Err(w.err).Msg(fmt.Sprintln(args...))
+}
+
+func (w *wrapZerolog) Info(args ...interface{}) {
+	w.inner.Info().Err(w.err).Msg(fmt.Sprint(args...))
+}
+
+func (w *wrapZerolog) Infof(format string, args ...interface{}) {
+	w.inner.Info().Err(w.err).Msgf(format, args...)
+}
+
+func (w *wrapZerolog) Infoln(args ...interface{}) {
+	w.inner.Info().Err(w.err).Msg(fmt.Sprintln(args...))
+}
+
+func (w *wrapZerolog) Trace(args ...interface{}) {
+	w.inner.Trace().Err(w.err).Msg(fmt.Sprint(args...))
+}
+
+func (w *wrapZerolog) Tracef(format string, args ...interface{}) {
+	w.inner.Trace().Err(w.err).Msgf(format, args...)
+}
+
+func (w *wrapZerolog) Traceln(args ...interface{}) {
+	w.inner.Trace().Err(w.err).Msg(fmt.Sprintln(args...))
+}
+
+func (w *wrapZerolog) Warn(args ...interface{}) {
+	w.inner.Warn().Err(w.err).Msg(fmt.Sprint(args...))
+}
+
+func (w *wrapZerolog) Warnf(format string, args ...interface{}) {
+	w.inner.Warn().Err(w.err).Msgf(format, args...)
+}
+
+func (w *wrapZerolog) Warnln(args ...interface{}) {
+	w.inner.Warn().Err(w.err).Msg(fmt.Sprintln(args...))
+}

app/pipeline/manager/manager.go

@@ -414,7 +414,7 @@ func (m *Manager) AfterStep(ctx context.Context, step *types.Step) error {
 		log.Warn().Err(err).Msg("manager: cannot update step")
 	}
 
-	if err := m.Logz.Delete(noContext, step.ID); err != nil {
+	if err := m.Logz.Delete(noContext, step.ID); err != nil && !errors.Is(err, livelog.ErrStreamNotFound) {
 		log.Warn().Err(err).Msg("manager: cannot teardown log stream")
 	}
 	return retErr

app/pipeline/manager/teardown.go

@@ -101,7 +101,7 @@ func (t *teardown) do(ctx context.Context, stage *types.Stage) error {
 
 	for _, step := range stage.Steps {
 		err = t.Logs.Delete(noContext, step.ID)
-		if err != nil {
+		if err != nil && !errors.Is(err, livelog.ErrStreamNotFound) {
 			log.Warn().Err(err).Msgf("failed to delete log stream for step %d", step.ID)
 		}
 	}

app/pipeline/runner/poller.go

@@ -19,6 +19,7 @@ import (
 	"fmt"
 	"runtime/debug"
 
+	"github.com/harness/gitness/app/pipeline/logger"
 	"github.com/harness/gitness/types"
 
 	"github.com/drone-runners/drone-runner-docker/engine/resource"
@@ -26,6 +27,7 @@ import (
 	"github.com/drone/drone-go/drone"
 	runnerclient "github.com/drone/runner-go/client"
 	"github.com/drone/runner-go/poller"
+	"github.com/rs/zerolog/log"
 )
 
 func NewExecutionPoller(
@@ -33,11 +35,16 @@ func NewExecutionPoller(
 	config *types.Config,
 	client runnerclient.Client,
 ) *poller.Poller {
-	// taking the cautious approach of recovering in case of panics
 	runWithRecovery := func(ctx context.Context, stage *drone.Stage) (err error) {
+		ctx = logger.WithUnwrappedZerolog(ctx)
 		defer func() {
 			if r := recover(); r != nil {
-				err = fmt.Errorf("panic received while executing run: %s", debug.Stack())
+				err = fmt.Errorf("panic received: %s", debug.Stack())
+			}
+
+			// the caller of this method (poller.Poller) discards the error - log it here
+			if err != nil {
+				log.Ctx(ctx).Error().Err(err).Msgf("An error occured while calling runner.Run in Poller")
 			}
 		}()
 		return runner.Run(ctx, stage)

cli/server/server.go

@@ -22,16 +22,15 @@ import (
 	"syscall"
 	"time"
 
+	"github.com/harness/gitness/app/pipeline/logger"
 	"github.com/harness/gitness/profiler"
 	"github.com/harness/gitness/types"
 	"github.com/harness/gitness/version"
 
-	"github.com/drone/runner-go/logger"
 	"github.com/joho/godotenv"
 	"github.com/mattn/go-isatty"
 	"github.com/rs/zerolog"
 	"github.com/rs/zerolog/log"
-	"github.com/sirupsen/logrus"
 	"golang.org/x/sync/errgroup"
 	"gopkg.in/alecthomas/kingpin.v2"
 )
@@ -117,11 +116,10 @@ func (c *command) run(*kingpin.ParseContext) error {
 		})
 		// start poller for CI build executions.
 		g.Go(func() error {
-			log := logrus.New()
-			log.Out = os.Stdout
-			log.Level = logrus.DebugLevel // print all debug logs in common runner code.
-			ctx = logger.WithContext(ctx, logger.Logrus(log.WithContext(ctx)))
-			system.poller.Poll(ctx, config.CI.ParallelWorkers)
+			system.poller.Poll(
+				logger.WithWrappedZerolog(ctx),
+				config.CI.ParallelWorkers,
+			)
 			return nil
 		})
 	}

livelog/memory.go

@@ -20,9 +20,9 @@ import (
 	"sync"
 )
 
-// error returned when a stream is not registered with
+// ErrStreamNotFound is returned when a stream is not registered with
 // the streamer.
-var errStreamNotFound = errors.New("stream: not found")
+var ErrStreamNotFound = errors.New("stream: not found")
 
 type streamer struct {
 	sync.Mutex
@@ -52,7 +52,7 @@ func (s *streamer) Delete(ctx context.Context, id int64) error {
 	}
 	s.Unlock()
 	if !ok {
-		return errStreamNotFound
+		return ErrStreamNotFound
 	}
 	return stream.close()
 }
@@ -62,7 +62,7 @@ func (s *streamer) Write(ctx context.Context, id int64, line *Line) error {
 	stream, ok := s.streams[id]
 	s.Unlock()
 	if !ok {
-		return errStreamNotFound
+		return ErrStreamNotFound
 	}
 	return stream.write(line)
 }