X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/0eb72b526bf8bbb011551ecf019f604e17a534f1..c3c538444c15e68e96780f157935f2baa4ba0bc5:/services/crunch-run/logging.go

diff --git a/services/crunch-run/logging.go b/services/crunch-run/logging.go
index 0083f0999c..f8ddd563c6 100644
--- a/services/crunch-run/logging.go
+++ b/services/crunch-run/logging.go
@@ -196,7 +196,9 @@ var crunchLogThrottlePeriod time.Duration = time.Second * 60
 var crunchLogThrottleLines int64 = 1024
 var crunchLogPartialLineThrottlePeriod time.Duration = time.Second * 5
 var crunchLogBytesPerEvent int64 = 4096
-var crunchLogSecondsBetweenEvents time.Duration = time.Second * 1
+var crunchLogSecondsBetweenEvents = time.Second
+var crunchLogUpdatePeriod = time.Hour / 2
+var crunchLogUpdateSize = int64(1 << 25)
 
 // ArvLogWriter is an io.WriteCloser that processes each write by
 // writing it through to another io.WriteCloser (typically a
@@ -221,7 +223,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 {
@@ -230,7 +232,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
@@ -275,7 +276,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
 
@@ -284,7 +284,7 @@ func (arvlog *ArvLogWriter) Write(p []byte) (n int, err error) {
 		}
 	}
 
-	return bytesWritten, nil
+	return len(p), nil
 }
 
 // Close the underlying writer
@@ -375,38 +375,33 @@ func (arvlog *ArvLogWriter) rateLimit(line []byte, now time.Time) (bool, []byte)
 
 // load the rate limit discovery config parameters
 func loadLogThrottleParams(clnt IArvadosClient) {
-	param, err := clnt.Discovery("crunchLimitLogBytesPerJob")
-	if err == nil {
-		crunchLimitLogBytesPerJob = int64(param.(float64))
-	}
-
-	param, err = clnt.Discovery("crunchLogThrottleBytes")
-	if err == nil {
-		crunchLogThrottleBytes = int64(param.(float64))
-	}
-
-	param, err = clnt.Discovery("crunchLogThrottlePeriod")
-	if err == nil {
-		crunchLogThrottlePeriod = time.Duration(float64(time.Second) * param.(float64))
-	}
-
-	param, err = clnt.Discovery("crunchLogThrottleLines")
-	if err == nil {
-		crunchLogThrottleLines = int64(param.(float64))
+	loadDuration := func(dst *time.Duration, key string) {
+		if param, err := clnt.Discovery(key); err != nil {
+			return
+		} else if d, ok := param.(float64); !ok {
+			return
+		} else {
+			*dst = time.Duration(d) * time.Second
+		}
 	}
-
-	param, err = clnt.Discovery("crunchLogPartialLineThrottlePeriod")
-	if err == nil {
-		crunchLogPartialLineThrottlePeriod = time.Duration(float64(time.Second) * param.(float64))
+	loadInt64 := func(dst *int64, key string) {
+		if param, err := clnt.Discovery(key); err != nil {
+			return
+		} else if val, ok := param.(float64); !ok {
+			return
+		} else {
+			*dst = int64(val)
+		}
 	}
 
-	param, err = clnt.Discovery("crunchLogBytesPerEvent")
-	if err == nil {
-		crunchLogBytesPerEvent = int64(param.(float64))
-	}
+	loadInt64(&crunchLimitLogBytesPerJob, "crunchLimitLogBytesPerJob")
+	loadInt64(&crunchLogThrottleBytes, "crunchLogThrottleBytes")
+	loadDuration(&crunchLogThrottlePeriod, "crunchLogThrottlePeriod")
+	loadInt64(&crunchLogThrottleLines, "crunchLogThrottleLines")
+	loadDuration(&crunchLogPartialLineThrottlePeriod, "crunchLogPartialLineThrottlePeriod")
+	loadInt64(&crunchLogBytesPerEvent, "crunchLogBytesPerEvent")
+	loadDuration(&crunchLogSecondsBetweenEvents, "crunchLogSecondsBetweenEvents")
+	loadInt64(&crunchLogUpdateSize, "crunchLogUpdateSize")
+	loadDuration(&crunchLogUpdatePeriod, "crunchLogUpdatePeriod")
 
-	param, err = clnt.Discovery("crunchLogSecondsBetweenEvents")
-	if err == nil {
-		crunchLogSecondsBetweenEvents = time.Duration(float64(time.Second) * param.(float64))
-	}
 }