Merge pull request #417 from pixlise/development

Release 4.53.0

Author: pnemere

api/config/config.go

@@ -101,6 +101,13 @@ type APIConfig struct {
 	MaxFileCacheAgeSec    uint
 	MaxFileCacheSizeBytes uint
 
+	// Max time we allow memoised item to exist in DB and not be retrieved.
+	// If it hasn't been accessed in this many seconds, consider it stale & delete it!
+	MaxUnretrievedMemoisationAgeSec uint
+
+	// How often we run memoisation GC
+	MemoisationGCIntervalSec uint
+
 	ImportJobMaxTimeSec  uint32
 	PIQUANTJobMaxTimeSec uint32
 
@@ -213,6 +220,14 @@ func Init() (APIConfig, error) {
 		cfg.PIQUANTJobMaxTimeSec = uint32(15 * 60)
 	}
 
+	if cfg.MaxUnretrievedMemoisationAgeSec <= 0 {
+		cfg.MaxUnretrievedMemoisationAgeSec = 86400 * 30
+	}
+
+	if cfg.MemoisationGCIntervalSec <= 0 {
+		cfg.MemoisationGCIntervalSec = 3600
+	}
+
 	cfg.KubeConfig = *kubeconfig
 
 	return cfg, nil

api/dataimport/internal/converters/jplbreadboard/spectra_test.go

@@ -44,7 +44,7 @@ func Example_getMSASeqNo() {
 
 func Example_getSpectraFiles() {
 	files := []string{"../something/file.txt", "../something/Normal_B_1_2_3.msa", "../something/BulkSum_B_1_2_3.msa", "../something/Another_B_1_2_3.msa", "../something/Normal_B_1_2.jpg", "Normal_P_1_2_3.msa"}
-	f, l := getSpectraFiles(files, true, &logger.StdOutLogger{})
+	f, l := getSpectraFiles(files, true, &logger.StdErrLogger{})
 
 	for _, v := range f {
 		fmt.Printf(v + "\n")

api/dataimport/internal/converters/pixlem/import.go

@@ -292,6 +292,18 @@ func importEMData(creatorId string, rtt string, beamLocPath string, hkPath strin
 		}
 	}
 
+	// Check if there are beam locations for images that we don't have
+	for _, pmc := range ijPMCs {
+		if _, ok := contextImgsPerPMC[pmc]; !ok {
+			logger.Infof("WARNING: Missing image for beam location PMC: %v. Beam ij's for this won't be stored.", pmc)
+
+			// Delete all stored PMCs for this
+			for _, beamInfo := range beamLookup {
+				delete(beamInfo.IJ, pmc)
+			}
+		}
+	}
+
 	hkData, err := importerutils.ReadHousekeepingFile(hkPath, 0, logger)
 	if err != nil {
 		return nil, err

api/endpoints/memoisation.go

@@ -62,25 +62,45 @@ func GetMemoise(params apiRouter.ApiHandlerGenericParams) error {
 		return err
 	}
 
-	// Update last accessed time here
-	timestamp := uint32(params.Svcs.TimeStamper.GetTimeNowSec())
-	if timestamp != item.LastReadTimeUnixSec {
-		update := bson.D{{Key: "$set", Value: bson.D{{Key: "lastreadtimeunixsec", Value: timestamp}}}}
-		updResult, err := coll.UpdateOne(ctx, filter, update, options.Update())
+	now := uint32(params.Svcs.TimeStamper.GetTimeNowSec())
+
+	// Check if this is passed the max age we allow for an item to live in our cache
+	if item.LastReadTimeUnixSec < now-uint32(params.Svcs.Config.MaxUnretrievedMemoisationAgeSec) {
+		// It's too old, delete & don't return
+		params.Svcs.Log.Infof("Retrieved memoised item: %v that hasn't been accessed in %v sec. Deleting.", key, now-item.LastReadTimeUnixSec)
+
+		delResult, err := coll.DeleteOne(ctx, filter, options.Delete())
 		if err != nil {
 			// Don't error out on this, but do notify
-			params.Svcs.Log.Errorf("Failed to update last read time stamp for memoised item: %v. Error: %v", key, err)
-		}
+			params.Svcs.Log.Errorf("Failed to delete outdated memoised item: %v. Error: %v", key, err)
+		} else {
+			if delResult.DeletedCount != 1 {
+				params.Svcs.Log.Errorf("Memoised item delete had unexpected counts %+v key: %v", delResult, key)
+			}
 
-		if updResult.ModifiedCount != 1 {
-			params.Svcs.Log.Errorf("Memoised item timestamp update had unexpected counts %+v key: %v", updResult, key)
+			return errorwithstatus.MakeNotFoundError(key)
+		}
+	} else {
+		// Update last accessed time here
+		if now != item.LastReadTimeUnixSec {
+			update := bson.D{{Key: "$set", Value: bson.D{{Key: "lastreadtimeunixsec", Value: now}}}}
+			updResult, err := coll.UpdateOne(ctx, filter, update, options.Update())
+			if err != nil {
+				// Don't error out on this, but do notify
+				params.Svcs.Log.Errorf("Failed to update last read time stamp for memoised item: %v. Error: %v", key, err)
+			}
+
+			if updResult.ModifiedCount != 1 {
+				params.Svcs.Log.Errorf("Memoised item timestamp update had unexpected counts %+v key: %v", updResult, key)
+			}
+
+			// Also set it in the item we're replying with
+			item.LastReadTimeUnixSec = now
 		}
-
-		// Also set it in the item we're replying with
-		item.LastReadTimeUnixSec = timestamp
 	}
 
 	utils.SendProtoBinary(params.Writer, item)
+
 	return nil
 }
 

api/endpoints/middlewareAuth_test.go

@@ -99,7 +99,7 @@ func Example_isMatch() {
 
 func Example_getPermissionsForURI() {
 	var a AuthMiddleWareData
-	a.Logger = &logger.StdOutLogger{}
+	a.Logger = &logger.StdErrLogger{}
 	a.RoutePermissionsRequired = map[string]string{
 		"GET/the/{id}/something": "root3",
 		"GET/the/path":           "root1",

api/memoisation/garbageCollection.go

@@ -0,0 +1,37 @@
+package memoisation
+
+import (
+	"context"
+	"time"
+
+	"github.com/pixlise/core/v4/api/dbCollections"
+	"github.com/pixlise/core/v4/core/logger"
+	"github.com/pixlise/core/v4/core/timestamper"
+	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/mongo"
+	"go.mongodb.org/mongo-driver/mongo/options"
+)
+
+func RunMemoisationGarbageCollector(intervalSec uint32, oldestAllowedSec uint32, mongoDB *mongo.Database, ts timestamper.ITimeStamper, log logger.ILogger) {
+	for range time.Tick(time.Second * time.Duration(intervalSec)) {
+		collectGarbage(mongoDB, oldestAllowedSec, ts, log)
+	}
+}
+
+func collectGarbage(mongoDB *mongo.Database, oldestAllowedSec uint32, ts timestamper.ITimeStamper, log logger.ILogger) {
+	log.Infof("Memoisation GC starting...")
+
+	oldestAllowedUnixSec := ts.GetTimeNowSec() - int64(oldestAllowedSec)
+
+	ctx := context.TODO()
+	opts := options.Delete()
+	filter := bson.M{"lastreadtimeunixsec": bson.M{"$lt": oldestAllowedUnixSec}}
+	coll := mongoDB.Collection(dbCollections.MemoisedItemsName)
+
+	delResult, err := coll.DeleteMany(ctx, filter, opts)
+	if err != nil {
+		log.Errorf("Memoisation GC delete error: %v", err)
+	}
+
+	log.Infof("Memoisation GC deleted %v items", delResult.DeletedCount)
+}

api/memoisation/garbageCollection_test.go

@@ -0,0 +1,61 @@
+package memoisation
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/pixlise/core/v4/api/dbCollections"
+	"github.com/pixlise/core/v4/core/logger"
+	"github.com/pixlise/core/v4/core/timestamper"
+	"github.com/pixlise/core/v4/core/wstestlib"
+	protos "github.com/pixlise/core/v4/generated-protos"
+	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/mongo/options"
+)
+
+func Example_CollectGarbage() {
+	db := wstestlib.GetDB()
+	ctx := context.TODO()
+	coll := db.Collection(dbCollections.MemoisedItemsName)
+
+	// Insert an item that's too old, and one that's newly accessed
+	ts := &timestamper.MockTimeNowStamper{
+		QueuedTimeStamps: []int64{1234567890, 1234567890, 1234567890, 1234567890},
+	}
+
+	now := uint32(ts.GetTimeNowSec())
+	maxAge := uint32(100)
+
+	item := &protos.MemoisedItem{
+		Key:                 "key123",
+		MemoTimeUnixSec:     now - maxAge - 50,
+		Data:                []byte{1, 3, 5, 7},
+		ScanId:              "scan333",
+		DataSize:            uint32(4),
+		LastReadTimeUnixSec: now - maxAge - 10,
+	}
+
+	opt := options.Update().SetUpsert(true)
+	_, err := coll.UpdateByID(ctx, item.Key, bson.D{{Key: "$set", Value: item}}, opt)
+	fmt.Printf("Insert 1: %v\n", err)
+
+	item = &protos.MemoisedItem{
+		Key:                 "key456",
+		MemoTimeUnixSec:     now - maxAge - 55,
+		Data:                []byte{2, 4, 6, 8, 10},
+		ScanId:              "scan222",
+		DataSize:            uint32(5),
+		LastReadTimeUnixSec: now - 5,
+	}
+	_, err = coll.UpdateByID(ctx, item.Key, bson.D{{Key: "$set", Value: item}}, opt)
+	fmt.Printf("Insert 2: %v\n", err)
+
+	log := &logger.StdOutLogger{}
+	collectGarbage(db, maxAge, ts, log)
+
+	// Output:
+	// Insert 1: <nil>
+	// Insert 2: <nil>
+	// INFO: Memoisation GC starting...
+	// INFO: Memoisation GC deleted 1 items
+}

api/quantification/converter_test.go

@@ -140,7 +140,7 @@ func readDatasetFile(path string) (*protos.Experiment, error) {
 }
 
 func Example_matchPMCsWithDataset() {
-	l := &logger.StdOutLogger{}
+	l := &logger.StdErrLogger{}
 	data := csvData{[]string{"X", "Y", "Z", "filename", "Ca_%"}, [][]string{[]string{"1", "0.40", "0", "Roastt_Laguna_Salinas_28kV_230uA_03_03_2020_111.msa", "4.5"}}}
 
 	exp, err := readDatasetFile("./testdata/LagunaSalinasdataset.bin")

core/logger/log-stderr.go

@@ -0,0 +1,55 @@
+// Licensed to NASA JPL under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. NASA JPL licenses this file to you 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 (
+	"fmt"
+	"log"
+)
+
+// StdErrLogger - For mocking out in tests
+type StdErrLogger struct {
+	logLevel LogLevel
+}
+
+func (l *StdErrLogger) Printf(level LogLevel, format string, a ...interface{}) {
+	txt := logLevelPrefix[level] + ": " + fmt.Sprintf(format, a...)
+	log.Println(txt)
+}
+func (l *StdErrLogger) Debugf(format string, a ...interface{}) {
+	if l.logLevel <= LogDebug {
+		l.Printf(LogDebug, format, a...)
+	}
+}
+func (l *StdErrLogger) Infof(format string, a ...interface{}) {
+	if l.logLevel <= LogInfo {
+		l.Printf(LogInfo, format, a...)
+	}
+}
+func (l *StdErrLogger) Errorf(format string, a ...interface{}) {
+	l.Printf(LogError, format, a...)
+}
+
+func (l *StdErrLogger) SetLogLevel(level LogLevel) {
+	l.logLevel = level
+}
+func (l *StdErrLogger) GetLogLevel() LogLevel {
+	return l.logLevel
+}
+func (l *StdErrLogger) Close() {
+}

core/logger/log-stdout.go

@@ -19,7 +19,6 @@ package logger
 
 import (
 	"fmt"
-	"log"
 )
 
 // StdOutLogger - For mocking out in tests
@@ -29,7 +28,7 @@ type StdOutLogger struct {
 
 func (l *StdOutLogger) Printf(level LogLevel, format string, a ...interface{}) {
 	txt := logLevelPrefix[level] + ": " + fmt.Sprintf(format, a...)
-	log.Println(txt)
+	fmt.Println(txt)
 }
 func (l *StdOutLogger) Debugf(format string, a ...interface{}) {
 	if l.logLevel <= LogDebug {