Pipeline db fixes (#283)

* small update to docs * fix pipeline updates and logging
SeldonIO · Jun 10, 2022 · ec665ea · ec665ea
1 parent 2d1a27d
commit ec665ea
Show file tree

Hide file tree

Showing 3 changed files with 10 additions and 6 deletions.
diff --git a/scheduler/pkg/store/pipeline/db.go b/scheduler/pkg/store/pipeline/db.go
@@ -3,15 +3,18 @@ package pipeline
 import (
 	"github.com/dgraph-io/badger/v3"
 	"github.com/seldonio/seldon-core/scheduler/apis/mlops/scheduler"
+	"github.com/sirupsen/logrus"
 	"google.golang.org/protobuf/proto"
 )
 
 type PipelineDB struct {
 	db *badger.DB
 }
 
-func NewPipelineDb(path string) (*PipelineDB, error) {
-	db, err := badger.Open(badger.DefaultOptions(path))
+func NewPipelineDb(path string, logger logrus.FieldLogger) (*PipelineDB, error) {
+	options := badger.DefaultOptions(path)
+	options.Logger = logger.WithField("source", "pipelineDb")
+	db, err := badger.Open(options)
 	if err != nil {
 		return nil, err
 	}

diff --git a/scheduler/pkg/store/pipeline/db_test.go b/scheduler/pkg/store/pipeline/db_test.go
@@ -109,7 +109,8 @@ func TestSaveAndRestore(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 			path := fmt.Sprintf("%s/db", t.TempDir())
-			db, err := NewPipelineDb(path)
+			logger := log.New()
+			db, err := NewPipelineDb(path, logger)
 			g.Expect(err).To(BeNil())
 			for _, p := range test.pipelines {
 				err := db.save(p)

diff --git a/scheduler/pkg/store/pipeline/store.go b/scheduler/pkg/store/pipeline/store.go
@@ -36,7 +36,7 @@ type PipelineStore struct {
 
 func NewPipelineStore(logger logrus.FieldLogger, eventHub *coordinator.EventHub) *PipelineStore {
 	ps := &PipelineStore{
-		logger:    logger,
+		logger:    logger.WithField("source", "pipelineStore"),
 		eventHub:  eventHub,
 		pipelines: make(map[string]*Pipeline),
 		db:        nil,
@@ -49,7 +49,7 @@ func (ps *PipelineStore) InitialiseDB(path string) error {
 	if err != nil {
 		return err
 	}
-	db, err := NewPipelineDb(path)
+	db, err := NewPipelineDb(path, ps.logger)
 	if err != nil {
 		return err
 	}
@@ -325,7 +325,7 @@ func (ps *PipelineStore) setPipelineStateImpl(name string, versionNumber uint32,
 				if status == PipelineReady {
 					evts = append(evts, ps.terminateOldUnterminatedPipelinesIfNeeded(pipeline)...)
 				}
-				if ps.db != nil {
+				if !pipeline.Deleted && ps.db != nil {
 					err := ps.db.save(pipeline)
 					if err != nil {
 						return evts, err