Merge pull request #647 from ohsu-comp-bio/close-files

FIX: Open File leak

Author: kellrott

Makefile

@@ -99,8 +99,8 @@ start-mongodb:
 	@docker run -d --name funnel-mongodb-test -p 27000:27017 docker.io/mongo:3.5.13 > /dev/null
 
 test-mongodb:
-	@go test ./tests/core/ -funnel-config `pwd`/tests/mongo.config.yml
-	@go test ./tests/scheduler/ -funnel-config `pwd`/tests/mongo.config.yml
+	@go test ./tests/core/ --funnel-config `pwd`/tests/mongo.config.yml
+	@go test ./tests/scheduler/ --funnel-config `pwd`/tests/mongo.config.yml
 
 test-badger:
 	@go test ./tests/core/ -funnel-config `pwd`/tests/badger.config.yml

cmd/server/run.go

@@ -47,7 +47,6 @@ type Server struct {
 // Database represents the base funnel database interface
 type Database interface {
 	tes.ReadOnlyServer
-	events.Writer
 	Init() error
 }
 

cmd/worker/run.go

@@ -123,7 +123,7 @@ func newDatabaseTaskReader(taskID string, db tes.ReadOnlyServer, err error) (wor
 	if err != nil {
 		return nil, fmt.Errorf("creating database task reader: %v", err)
 	}
-	return worker.NewGenericTaskReader(db.GetTask, taskID), nil
+	return worker.NewGenericTaskReader(db.GetTask, taskID, db.Close), nil
 }
 
 // eventWriterBuilder is a helper for building a set of event writers,

compute/batch/backend.go

@@ -62,6 +62,11 @@ func (b *Backend) WriteEvent(ctx context.Context, ev *events.Event) error {
 	return nil
 }
 
+func (b *Backend) Close() {
+	b.database.Close()
+	b.event.Close()
+}
+
 // Submit submits a task to the AWS batch service.
 func (b *Backend) Submit(task *tes.Task) error {
 	ctx := context.Background()

compute/hpc_backend.go

@@ -55,6 +55,8 @@ func (b *HPCBackend) WriteEvent(ctx context.Context, ev *events.Event) error {
 	return nil
 }
 
+func (b *HPCBackend) Close() {}
+
 // Submit submits a task via "qsub", "condor_submit", "sbatch", etc.
 func (b *HPCBackend) Submit(task *tes.Task) error {
 	ctx := context.Background()

compute/kubernetes/backend.go

@@ -105,6 +105,10 @@ func (b *Backend) WriteEvent(ctx context.Context, ev *events.Event) error {
 	return nil
 }
 
+func (b *Backend) Close() {
+	//TODO: close database?
+}
+
 // createJob uses the configured template to create a kubernetes batch job.
 func (b *Backend) createJob(task *tes.Task) (*v1.Job, error) {
 	submitTpl, err := template.New(task.Id).Parse(b.template)

compute/local/backend.go

@@ -32,6 +32,8 @@ func (b *Backend) WriteEvent(ctx context.Context, ev *events.Event) error {
 	return nil
 }
 
+func (b *Backend) Close() {}
+
 // Submit submits a task. For the Local backend this results in the task
 // running immediately.
 func (b *Backend) Submit(task *tes.Task) error {
@@ -46,6 +48,7 @@ func (b *Backend) Submit(task *tes.Task) error {
 
 	go func() {
 		w.Run(ctx)
+		w.Close()
 	}()
 	return nil
 }

compute/noop/backend.go

@@ -20,3 +20,5 @@ type Backend struct{}
 func (b *Backend) WriteEvent(context.Context, *events.Event) error {
 	return nil
 }
+
+func (b *Backend) Close() {}

database/badger/new.go

@@ -32,6 +32,10 @@ func (db *Badger) Init() error {
 	return nil
 }
 
+func (db *Badger) Close() {
+	db.db.Close()
+}
+
 var taskKeyPrefix = []byte("tasks")
 
 func taskKey(id string) []byte {

database/boltdb/new.go

@@ -95,3 +95,7 @@ func (taskBolt *BoltDB) Init() error {
 		return nil
 	})
 }
+
+func (taskBolt *BoltDB) Close() {
+	taskBolt.db.Close()
+}