run golangci-lint and apply suggested fixes

daemon/control.go

@@ -89,7 +89,7 @@ func (j *controlJob) Run(ctx context.Context) {
 
 	mux := http.NewServeMux()
 	mux.Handle(ControlJobEndpointPProf,
-		requestLogger{log: log, handler: jsonRequestResponder{func(decoder jsonDecoder) (interface{}, error) {
+		requestLogger{log: log, handler: jsonRequestResponder{log, func(decoder jsonDecoder) (interface{}, error) {
 			var msg PprofServerControlMsg
 			err := decoder(&msg)
 			if err != nil {
@@ -100,19 +100,19 @@ func (j *controlJob) Run(ctx context.Context) {
 		}}})
 
 	mux.Handle(ControlJobEndpointVersion,
-		requestLogger{log: log, handler: jsonResponder{func() (interface{}, error) {
+		requestLogger{log: log, handler: jsonResponder{log, func() (interface{}, error) {
 			return version.NewZreplVersionInformation(), nil
 		}}})
 
 	mux.Handle(ControlJobEndpointStatus,
 		// don't log requests to status endpoint, too spammy
-		jsonResponder{func() (interface{}, error) {
+		jsonResponder{log, func() (interface{}, error) {
 			s := j.jobs.status()
 			return s, nil
 		}})
 
 	mux.Handle(ControlJobEndpointSignal,
-		requestLogger{log: log, handler: jsonRequestResponder{func(decoder jsonDecoder) (interface{}, error) {
+		requestLogger{log: log, handler: jsonRequestResponder{log, func(decoder jsonDecoder) (interface{}, error) {
 			type reqT struct {
 				Name string
 				Op   string
@@ -153,7 +153,10 @@ outer:
 		select {
 		case <-ctx.Done():
 			log.WithError(ctx.Err()).Info("context done")
-			server.Shutdown(context.Background())
+			err := server.Shutdown(context.Background())
+			if err != nil {
+				log.WithError(err).Error("cannot shutdown server")
+			}
 			break outer
 		case err = <-served:
 			if err != nil {
@@ -167,33 +170,50 @@ outer:
 }
 
 type jsonResponder struct {
+	log      Logger
 	producer func() (interface{}, error)
 }
 
 func (j jsonResponder) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	logIoErr := func(err error) {
+		if err != nil {
+			j.log.WithError(err).Error("control handler io error")
+		}
+	}
 	res, err := j.producer()
 	if err != nil {
+		j.log.WithError(err).Error("control handler error")
 		w.WriteHeader(http.StatusInternalServerError)
-		io.WriteString(w, err.Error())
+		_, err = io.WriteString(w, err.Error())
+		logIoErr(err)
 		return
 	}
 	var buf bytes.Buffer
 	err = json.NewEncoder(&buf).Encode(res)
 	if err != nil {
+		j.log.WithError(err).Error("control handler json marshal error")
 		w.WriteHeader(http.StatusInternalServerError)
-		io.WriteString(w, err.Error())
+		_, err = io.WriteString(w, err.Error())
 	} else {
-		io.Copy(w, &buf)
+		_, err = io.Copy(w, &buf)
 	}
+	logIoErr(err)
 }
 
 type jsonDecoder = func(interface{}) error
 
 type jsonRequestResponder struct {
+	log      Logger
 	producer func(decoder jsonDecoder) (interface{}, error)
 }
 
 func (j jsonRequestResponder) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	logIoErr := func(err error) {
+		if err != nil {
+			j.log.WithError(err).Error("control handler io error")
+		}
+	}
+
 	var decodeError error
 	decoder := func(i interface{}) error {
 		err := json.NewDecoder(r.Body).Decode(&i)
@@ -205,22 +225,28 @@ func (j jsonRequestResponder) ServeHTTP(w http.ResponseWriter, r *http.Request)
 	//If we had a decode error ignore output of producer and return error
 	if decodeError != nil {
 		w.WriteHeader(http.StatusBadRequest)
-		io.WriteString(w, decodeError.Error())
+		_, err := io.WriteString(w, decodeError.Error())
+		logIoErr(err)
 		return
 	}
 	if producerErr != nil {
+		j.log.WithError(producerErr).Error("control handler error")
 		w.WriteHeader(http.StatusInternalServerError)
-		io.WriteString(w, producerErr.Error())
+		_, err := io.WriteString(w, producerErr.Error())
+		logIoErr(err)
 		return
 	}
 
 	var buf bytes.Buffer
 	encodeErr := json.NewEncoder(&buf).Encode(res)
 	if encodeErr != nil {
+		j.log.WithError(producerErr).Error("control handler json marhsal error")
 		w.WriteHeader(http.StatusInternalServerError)
-		io.WriteString(w, encodeErr.Error())
+		_, err := io.WriteString(w, encodeErr.Error())
+		logIoErr(err)
 	} else {
-		io.Copy(w, &buf)
+		_, err := io.Copy(w, &buf)
+		logIoErr(err)
 	}
 }
 

daemon/daemon.go

@@ -117,9 +117,7 @@ func newJobs() *jobs {
 }
 
 const (
-	logJobField    string = "job"
-	logTaskField   string = "task"
-	logSubsysField string = "subsystem"
+	logJobField string = "job"
 )
 
 func (s *jobs) wait() <-chan struct{} {

daemon/job/active.go

@@ -68,8 +68,7 @@ type activeSideTasks struct {
 func (a *ActiveSide) updateTasks(u func(*activeSideTasks)) activeSideTasks {
 	a.tasksMtx.Lock()
 	defer a.tasksMtx.Unlock()
-	var copy activeSideTasks
-	copy = a.tasks
+	copy := a.tasks
 	if u == nil {
 		return copy
 	}

daemon/job/job.go

@@ -94,18 +94,21 @@ func (s *Status) UnmarshalJSON(in []byte) (err error) {
 		var st SnapJobStatus
 		err = json.Unmarshal(jobJSON, &st)
 		s.JobSpecific = &st
+
 	case TypePull:
 		fallthrough
 	case TypePush:
 		var st ActiveSideStatus
 		err = json.Unmarshal(jobJSON, &st)
 		s.JobSpecific = &st
+
 	case TypeSource:
 		fallthrough
 	case TypeSink:
 		var st PassiveStatus
 		err = json.Unmarshal(jobJSON, &st)
 		s.JobSpecific = &st
+
 	case TypeInternal:
 		// internal jobs do not report specifics
 	default:

daemon/logging/logging_formatters.go

@@ -161,10 +161,16 @@ func (f *LogfmtFormatter) Format(e *logger.Entry) ([]byte, error) {
 	enc := logfmt.NewEncoder(&buf)
 
 	if f.metadataFlags&MetadataTime != 0 {
-		enc.EncodeKeyval(FieldTime, e.Time)
+		err := enc.EncodeKeyval(FieldTime, e.Time)
+		if err != nil {
+			return nil, errors.Wrap(err, "logfmt: encode time")
+		}
 	}
 	if f.metadataFlags&MetadataLevel != 0 {
-		enc.EncodeKeyval(FieldLevel, e.Level)
+		err := enc.EncodeKeyval(FieldLevel, e.Level)
+		if err != nil {
+			return nil, errors.Wrap(err, "logfmt: encode level")
+		}
 	}
 
 	// at least try and put job and task in front
@@ -181,8 +187,10 @@ func (f *LogfmtFormatter) Format(e *logger.Entry) ([]byte, error) {
 		prefixed[pf] = true
 	}
 
-	enc.EncodeKeyval(FieldMessage, e.Message)
-
+	err := enc.EncodeKeyval(FieldMessage, e.Message)
+	if err != nil {
+		return nil, errors.Wrap(err, "logfmt: encode message")
+	}
 	for k, v := range e.Fields {
 		if !prefixed[k] {
 			if err := logfmtTryEncodeKeyval(enc, k, v); err != nil {
@@ -201,7 +209,10 @@ func logfmtTryEncodeKeyval(enc *logfmt.Encoder, field, value interface{}) error
 	case nil: // ok
 		return nil
 	case logfmt.ErrUnsupportedValueType:
-		enc.EncodeKeyval(field, fmt.Sprintf("<%T>", value))
+		err := enc.EncodeKeyval(field, fmt.Sprintf("<%T>", value))
+		if err != nil {
+			return errors.Wrap(err, "cannot encode unsuuported value type Go type")
+		}
 		return nil
 	}
 	return errors.Wrapf(err, "cannot encode field '%s'", field)

daemon/logging/logging_outlets.go

@@ -30,7 +30,10 @@ func (h WriterOutlet) WriteEntry(entry logger.Entry) error {
 		return err
 	}
 	_, err = h.writer.Write(bytes)
-	h.writer.Write([]byte("\n"))
+	if err != nil {
+		return err
+	}
+	_, err = h.writer.Write([]byte("\n"))
 	return err
 }
 
@@ -94,8 +97,10 @@ func (h *TCPOutlet) outLoop(retryInterval time.Duration) {
 				conn = nil
 			}
 		}
-		conn.SetWriteDeadline(time.Now().Add(retryInterval))
-		_, err = io.Copy(conn, msg)
+		err = conn.SetWriteDeadline(time.Now().Add(retryInterval))
+		if err == nil {
+			_, err = io.Copy(conn, msg)
+		}
 		if err != nil {
 			retry = time.Now().Add(retryInterval)
 			conn.Close()

daemon/pprof.go

@@ -7,11 +7,12 @@ import (
 	"context"
 	"net"
 	"net/http/pprof"
+
+	"github.com/zrepl/zrepl/daemon/job"
 )
 
 type pprofServer struct {
 	cc       chan PprofServerControlMsg
-	state    PprofServerControlMsg
 	listener net.Listener
 }
 
@@ -63,7 +64,14 @@ outer:
 			mux.Handle("/debug/pprof/profile", http.HandlerFunc(pprof.Profile))
 			mux.Handle("/debug/pprof/symbol", http.HandlerFunc(pprof.Symbol))
 			mux.Handle("/debug/pprof/trace", http.HandlerFunc(pprof.Trace))
-			go http.Serve(s.listener, mux)
+			go func() {
+				err := http.Serve(s.listener, mux)
+				if ctx.Err() != nil {
+					return
+				} else if err != nil {
+					job.GetLogger(ctx).WithError(err).Error("pprof server serve error")
+				}
+			}()
 			continue
 		}
 

daemon/prometheus.go

@@ -65,17 +65,15 @@ func (j *prometheusJob) Run(ctx context.Context) {
 		log.WithError(err).Error("cannot listen")
 	}
 	go func() {
-		select {
-		case <-ctx.Done():
-			l.Close()
-		}
+		<-ctx.Done()
+		l.Close()
 	}()
 
 	mux := http.NewServeMux()
 	mux.Handle("/metrics", promhttp.Handler())
 
 	err = http.Serve(l, mux)
-	if err != nil {
+	if err != nil && ctx.Err() == nil {
 		log.WithError(err).Error("error while serving")
 	}
 

daemon/pruner/pruner.go

@@ -86,19 +86,6 @@ type LocalPrunerFactory struct {
 	promPruneSecs *prometheus.HistogramVec
 }
 
-func checkContainsKeep1(rules []pruning.KeepRule) error {
-	if len(rules) == 0 {
-		return nil //No keep rules means keep all - ok
-	}
-	for _, e := range rules {
-		switch e.(type) {
-		case *pruning.KeepLastN:
-			return nil
-		}
-	}
-	return errors.New("sender keep rules must contain last_n or be empty so that the last snapshot is definitely kept")
-}
-
 func NewLocalPrunerFactory(in config.PruningLocal, promPruneSecs *prometheus.HistogramVec) (*LocalPrunerFactory, error) {
 	rules, err := pruning.RulesFromConfig(in.Keep)
 	if err != nil {

daemon/snapper/snapper.go

@@ -203,7 +203,7 @@ func syncUp(a args, u updater) state {
 	u(func(s *Snapper) {
 		s.sleepUntil = syncPoint
 	})
-	t := time.NewTimer(syncPoint.Sub(time.Now()))
+	t := time.NewTimer(time.Until(syncPoint))
 	defer t.Stop()
 	select {
 	case <-t.C:
@@ -307,7 +307,7 @@ func wait(a args, u updater) state {
 		logFunc("enter wait-state after error")
 	})
 
-	t := time.NewTimer(sleepUntil.Sub(time.Now()))
+	t := time.NewTimer(time.Until(sleepUntil))
 	defer t.Stop()
 
 	select {