X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/e54bce82515daeef450f53aa93d3acae397bfa5f..95be914af0ab0a82c4fa92b3f9c29ebec88e8595:/services/crunch-run/logging.go

diff --git a/services/crunch-run/logging.go b/services/crunch-run/logging.go
index 3aedf51779..ce0a661263 100644
--- a/services/crunch-run/logging.go
+++ b/services/crunch-run/logging.go
@@ -1,3 +1,7 @@
+// Copyright (C) The Arvados Authors. All rights reserved.
+//
+// SPDX-License-Identifier: AGPL-3.0
+
 package main
 
 import (
@@ -34,9 +38,10 @@ type ThrottledLogger struct {
 	*log.Logger
 	buf *bytes.Buffer
 	sync.Mutex
-	writer  io.WriteCloser
-	flush   chan struct{}
-	stopped chan struct{}
+	writer   io.WriteCloser
+	flush    chan struct{}
+	stopped  chan struct{}
+	stopping chan struct{}
 	Timestamper
 	Immediate    *log.Logger
 	pendingFlush bool
@@ -97,9 +102,10 @@ func (tl *ThrottledLogger) flusher() {
 	defer ticker.Stop()
 	for stopping := false; !stopping; {
 		select {
-		case _, open := <-tl.flush:
-			// if !open, will flush tl.buf and exit the loop
-			stopping = !open
+		case <-tl.stopping:
+			// flush tl.buf and exit the loop
+			stopping = true
+		case <-tl.flush:
 		case <-ticker.C:
 		}
 
@@ -120,10 +126,10 @@ func (tl *ThrottledLogger) flusher() {
 // underlying Writer.
 func (tl *ThrottledLogger) Close() error {
 	select {
-	case <-tl.flush:
+	case <-tl.stopping:
 		// already stopped
 	default:
-		close(tl.flush)
+		close(tl.stopping)
 	}
 	<-tl.stopped
 	return tl.writer.Close()
@@ -175,6 +181,7 @@ func NewThrottledLogger(writer io.WriteCloser) *ThrottledLogger {
 	tl := &ThrottledLogger{}
 	tl.flush = make(chan struct{}, 1)
 	tl.stopped = make(chan struct{})
+	tl.stopping = make(chan struct{})
 	tl.writer = writer
 	tl.Logger = log.New(tl, "", 0)
 	tl.Timestamper = RFC3339Timestamp
@@ -214,7 +221,7 @@ type ArvLogWriter struct {
 	closing                      bool
 }
 
-func (arvlog *ArvLogWriter) Write(p []byte) (n int, err error) {
+func (arvlog *ArvLogWriter) Write(p []byte) (int, error) {
 	// Write to the next writer in the chain (a file in Keep)
 	var err1 error
 	if arvlog.writeCloser != nil {
@@ -223,7 +230,6 @@ func (arvlog *ArvLogWriter) Write(p []byte) (n int, err error) {
 
 	// write to API after checking rate limit
 	now := time.Now()
-	bytesWritten := 0
 
 	if now.After(arvlog.logThrottleResetTime) {
 		// It has been more than throttle_period seconds since the last
@@ -258,7 +264,7 @@ func (arvlog *ArvLogWriter) Write(p []byte) (n int, err error) {
 		}
 	}
 
-	if (int64(arvlog.bufToFlush.Len()) > crunchLogBytesPerEvent ||
+	if (int64(arvlog.bufToFlush.Len()) >= crunchLogBytesPerEvent ||
 		(now.Sub(arvlog.bufFlushedAt) >= crunchLogSecondsBetweenEvents) ||
 		arvlog.closing) && (arvlog.bufToFlush.Len() > 0) {
 		// write to API
@@ -268,7 +274,6 @@ func (arvlog *ArvLogWriter) Write(p []byte) (n int, err error) {
 			"properties":  map[string]string{"text": arvlog.bufToFlush.String()}}}
 		err2 := arvlog.ArvClient.Create("logs", lr, nil)
 
-		bytesWritten = arvlog.bufToFlush.Len()
 		arvlog.bufToFlush = bytes.Buffer{}
 		arvlog.bufFlushedAt = now
 
@@ -277,7 +282,7 @@ func (arvlog *ArvLogWriter) Write(p []byte) (n int, err error) {
 		}
 	}
 
-	return bytesWritten, nil
+	return len(p), nil
 }
 
 // Close the underlying writer
@@ -366,7 +371,7 @@ func (arvlog *ArvLogWriter) rateLimit(line []byte, now time.Time) (bool, []byte)
 	}
 }
 
-// load the rate limit discovery config paramters
+// load the rate limit discovery config parameters
 func loadLogThrottleParams(clnt IArvadosClient) {
 	param, err := clnt.Discovery("crunchLimitLogBytesPerJob")
 	if err == nil {