Use contexts in database queries.

cmd/metal-api/internal/datastore/event.go

@@ -1,37 +1,39 @@
 package datastore
 
 import (
+	"context"
+
 	"github.com/metal-stack/metal-api/cmd/metal-api/internal/metal"
 )
 
 // ListProvisioningEventContainers returns all machine provisioning event containers.
-func (rs *RethinkStore) ListProvisioningEventContainers() (metal.ProvisioningEventContainers, error) {
+func (rs *RethinkStore) ListProvisioningEventContainers(ctx context.Context) (metal.ProvisioningEventContainers, error) {
 	es := make(metal.ProvisioningEventContainers, 0)
-	err := rs.listEntities(rs.eventTable(), &es)
+	err := rs.listEntities(ctx, rs.eventTable(), &es)
 	return es, err
 }
 
 // FindProvisioningEventContainer finds a provisioning event container to a given machine id.
-func (rs *RethinkStore) FindProvisioningEventContainer(id string) (*metal.ProvisioningEventContainer, error) {
+func (rs *RethinkStore) FindProvisioningEventContainer(ctx context.Context, id string) (*metal.ProvisioningEventContainer, error) {
 	var e metal.ProvisioningEventContainer
-	err := rs.findEntityByID(rs.eventTable(), &e, id)
+	err := rs.findEntityByID(ctx, rs.eventTable(), &e, id)
 	if err != nil {
 		return nil, err
 	}
 	return &e, nil
 }
 
 // UpdateProvisioningEventContainer updates a provisioning event container.
-func (rs *RethinkStore) UpdateProvisioningEventContainer(old *metal.ProvisioningEventContainer, new *metal.ProvisioningEventContainer) error {
-	return rs.updateEntity(rs.eventTable(), new, old)
+func (rs *RethinkStore) UpdateProvisioningEventContainer(ctx context.Context, old *metal.ProvisioningEventContainer, new *metal.ProvisioningEventContainer) error {
+	return rs.updateEntity(ctx, rs.eventTable(), new, old)
 }
 
 // CreateProvisioningEventContainer creates a new provisioning event container.
-func (rs *RethinkStore) CreateProvisioningEventContainer(ec *metal.ProvisioningEventContainer) error {
-	return rs.createEntity(rs.eventTable(), ec)
+func (rs *RethinkStore) CreateProvisioningEventContainer(ctx context.Context, ec *metal.ProvisioningEventContainer) error {
+	return rs.createEntity(ctx, rs.eventTable(), ec)
 }
 
 // UpsertProvisioningEventContainer inserts a machine's event container.
-func (rs *RethinkStore) UpsertProvisioningEventContainer(ec *metal.ProvisioningEventContainer) error {
-	return rs.upsertEntity(rs.eventTable(), ec)
+func (rs *RethinkStore) UpsertProvisioningEventContainer(ctx context.Context, ec *metal.ProvisioningEventContainer) error {
+	return rs.upsertEntity(ctx, rs.eventTable(), ec)
 }

cmd/metal-api/internal/datastore/filesystem.go

@@ -1,35 +1,39 @@
 package datastore
 
-import "github.com/metal-stack/metal-api/cmd/metal-api/internal/metal"
+import (
+	"context"
+
+	"github.com/metal-stack/metal-api/cmd/metal-api/internal/metal"
+)
 
 // FindFilesystemLayout return a filesystemlayout for a given id.
-func (rs *RethinkStore) FindFilesystemLayout(id string) (*metal.FilesystemLayout, error) {
+func (rs *RethinkStore) FindFilesystemLayout(ctx context.Context, id string) (*metal.FilesystemLayout, error) {
 	var fl metal.FilesystemLayout
-	err := rs.findEntityByID(rs.filesystemLayoutTable(), &fl, id)
+	err := rs.findEntityByID(ctx, rs.filesystemLayoutTable(), &fl, id)
 	if err != nil {
 		return nil, err
 	}
 	return &fl, nil
 }
 
 // ListFilesystemLayouts returns all filesystemlayouts.
-func (rs *RethinkStore) ListFilesystemLayouts() (metal.FilesystemLayouts, error) {
+func (rs *RethinkStore) ListFilesystemLayouts(ctx context.Context) (metal.FilesystemLayouts, error) {
 	fls := make(metal.FilesystemLayouts, 0)
-	err := rs.listEntities(rs.filesystemLayoutTable(), &fls)
+	err := rs.listEntities(ctx, rs.filesystemLayoutTable(), &fls)
 	return fls, err
 }
 
 // CreateFilesystemLayout creates a new filesystemlayout.
-func (rs *RethinkStore) CreateFilesystemLayout(fl *metal.FilesystemLayout) error {
-	return rs.createEntity(rs.filesystemLayoutTable(), fl)
+func (rs *RethinkStore) CreateFilesystemLayout(ctx context.Context, fl *metal.FilesystemLayout) error {
+	return rs.createEntity(ctx, rs.filesystemLayoutTable(), fl)
 }
 
 // DeleteFilesystemLayout deletes a filesystemlayout.
-func (rs *RethinkStore) DeleteFilesystemLayout(fl *metal.FilesystemLayout) error {
-	return rs.deleteEntity(rs.filesystemLayoutTable(), fl)
+func (rs *RethinkStore) DeleteFilesystemLayout(ctx context.Context, fl *metal.FilesystemLayout) error {
+	return rs.deleteEntity(ctx, rs.filesystemLayoutTable(), fl)
 }
 
 // UpdateFilesystemLayout updates a filesystemlayout.
-func (rs *RethinkStore) UpdateFilesystemLayout(oldFilesystemLayout *metal.FilesystemLayout, newFilesystemLayout *metal.FilesystemLayout) error {
-	return rs.updateEntity(rs.filesystemLayoutTable(), newFilesystemLayout, oldFilesystemLayout)
+func (rs *RethinkStore) UpdateFilesystemLayout(ctx context.Context, oldFilesystemLayout *metal.FilesystemLayout, newFilesystemLayout *metal.FilesystemLayout) error {
+	return rs.updateEntity(ctx, rs.filesystemLayoutTable(), newFilesystemLayout, oldFilesystemLayout)
 }

cmd/metal-api/internal/datastore/health.go

@@ -2,6 +2,7 @@ package datastore
 
 import (
 	"context"
+	"time"
 
 	"github.com/metal-stack/metal-lib/rest"
 	r "gopkg.in/rethinkdb/rethinkdb-go.v6"
@@ -13,13 +14,16 @@ func (rs *RethinkStore) ServiceName() string {
 
 // Check implements the health interface and tests if the database is healthy.
 func (rs *RethinkStore) Check(ctx context.Context) (rest.HealthStatus, error) {
+	healthCtx, cancel := context.WithTimeout(ctx, 2*time.Second)
+	defer cancel()
+
 	t := r.Branch(
 		rs.db().TableList().SetIntersection(r.Expr(tables)).Count().Eq(len(tables)),
 		r.Expr(true),
 		r.Error("required tables are missing"),
 	)
 
-	err := t.Exec(rs.session, r.ExecOpts{Context: ctx})
+	err := t.Exec(rs.session, r.ExecOpts{Context: healthCtx})
 	if err != nil {
 		return rest.HealthStatusUnhealthy, err
 	}

cmd/metal-api/internal/datastore/image.go

@@ -1,6 +1,7 @@
 package datastore
 
 import (
+	"context"
 	"fmt"
 	"sort"
 	"strings"
@@ -12,18 +13,18 @@ import (
 )
 
 // GetImage return a image for a given id without semver matching.
-func (rs *RethinkStore) GetImage(id string) (*metal.Image, error) {
+func (rs *RethinkStore) GetImage(ctx context.Context, id string) (*metal.Image, error) {
 	var i metal.Image
-	err := rs.findEntityByID(rs.imageTable(), &i, id)
+	err := rs.findEntityByID(ctx, rs.imageTable(), &i, id)
 	if err != nil {
 		return nil, err
 	}
 	return &i, nil
 }
 
 // FindImages returns all images for the given image id.
-func (rs *RethinkStore) FindImages(id string) ([]metal.Image, error) {
-	allImages, err := rs.ListImages()
+func (rs *RethinkStore) FindImages(ctx context.Context, id string) ([]metal.Image, error) {
+	allImages, err := rs.ListImages(ctx)
 	if err != nil {
 		return nil, err
 	}
@@ -36,8 +37,8 @@ func (rs *RethinkStore) FindImages(id string) ([]metal.Image, error) {
 }
 
 // FindImage returns an image for the given image id.
-func (rs *RethinkStore) FindImage(id string) (*metal.Image, error) {
-	allImages, err := rs.ListImages()
+func (rs *RethinkStore) FindImage(ctx context.Context, id string) (*metal.Image, error) {
+	allImages, err := rs.ListImages(ctx)
 	if err != nil {
 		return nil, err
 	}
@@ -52,40 +53,40 @@ func (rs *RethinkStore) FindImage(id string) (*metal.Image, error) {
 }
 
 // ListImages returns all images.
-func (rs *RethinkStore) ListImages() (metal.Images, error) {
+func (rs *RethinkStore) ListImages(ctx context.Context) (metal.Images, error) {
 	imgs := make(metal.Images, 0)
-	err := rs.listEntities(rs.imageTable(), &imgs)
+	err := rs.listEntities(ctx, rs.imageTable(), &imgs)
 	return imgs, err
 }
 
 // CreateImage creates a new image.
-func (rs *RethinkStore) CreateImage(i *metal.Image) error {
-	return rs.createEntity(rs.imageTable(), i)
+func (rs *RethinkStore) CreateImage(ctx context.Context, i *metal.Image) error {
+	return rs.createEntity(ctx, rs.imageTable(), i)
 }
 
 // DeleteImage deletes an image.
-func (rs *RethinkStore) DeleteImage(i *metal.Image) error {
-	return rs.deleteEntity(rs.imageTable(), i)
+func (rs *RethinkStore) DeleteImage(ctx context.Context, i *metal.Image) error {
+	return rs.deleteEntity(ctx, rs.imageTable(), i)
 }
 
 // UpdateImage updates an image.
-func (rs *RethinkStore) UpdateImage(oldImage *metal.Image, newImage *metal.Image) error {
-	return rs.updateEntity(rs.imageTable(), newImage, oldImage)
+func (rs *RethinkStore) UpdateImage(ctx context.Context, oldImage *metal.Image, newImage *metal.Image) error {
+	return rs.updateEntity(ctx, rs.imageTable(), newImage, oldImage)
 }
 
 // DeleteOrphanImages deletes Images which are no longer allocated by a machine and older than allowed.
 // Always at least one image per OS is kept even if no longer valid and not allocated.
 // This ensures to have always at least a usable image left.
-func (rs *RethinkStore) DeleteOrphanImages(images metal.Images, machines metal.Machines) (metal.Images, error) {
+func (rs *RethinkStore) DeleteOrphanImages(ctx context.Context, images metal.Images, machines metal.Machines) (metal.Images, error) {
 	if images == nil {
-		is, err := rs.ListImages()
+		is, err := rs.ListImages(ctx)
 		if err != nil {
 			return nil, err
 		}
 		images = is
 	}
 	if machines == nil {
-		ms, err := rs.ListMachines()
+		ms, err := rs.ListMachines(ctx)
 		if err != nil {
 			return nil, err
 		}
@@ -105,7 +106,7 @@ func (rs *RethinkStore) DeleteOrphanImages(images metal.Images, machines metal.M
 		}
 
 		if isOrphanImage(image, machines) {
-			err := rs.DeleteImage(&image)
+			err := rs.DeleteImage(ctx, &image)
 			if err != nil {
 				return nil, fmt.Errorf("unable to delete image:%s err:%w", image.ID, err)
 			}

cmd/metal-api/internal/datastore/image_test.go

@@ -1,6 +1,7 @@
 package datastore
 
 import (
+	"context"
 	"reflect"
 	"testing"
 	"time"
@@ -52,7 +53,7 @@ func TestRethinkStore_FindImage(t *testing.T) {
 	for i := range tests {
 		tt := tests[i]
 		t.Run(tt.name, func(t *testing.T) {
-			got, err := tt.rs.FindImage(tt.args.id)
+			got, err := tt.rs.FindImage(context.Background(), tt.args.id)
 			if (err != nil) != tt.wantErr {
 				t.Errorf("RethinkStore.FindImage() error = %v, wantErr %v", err, tt.wantErr)
 				return
@@ -86,7 +87,7 @@ func TestRethinkStore_ListImages(t *testing.T) {
 	for i := range tests {
 		tt := tests[i]
 		t.Run(tt.name, func(t *testing.T) {
-			got, err := tt.rs.ListImages()
+			got, err := tt.rs.ListImages(context.Background())
 			if (err != nil) != tt.wantErr {
 				t.Errorf("RethinkStore.ListImages() error = %v, wantErr %v", err, tt.wantErr)
 				return
@@ -127,7 +128,7 @@ func TestRethinkStore_CreateImage(t *testing.T) {
 	for i := range tests {
 		tt := tests[i]
 		t.Run(tt.name, func(t *testing.T) {
-			err := tt.rs.CreateImage(tt.args.i)
+			err := tt.rs.CreateImage(context.Background(), tt.args.i)
 			if (err != nil) != tt.wantErr {
 				t.Errorf("RethinkStore.CreateImage() error = %v, wantErr %v", err, tt.wantErr)
 				return
@@ -171,7 +172,7 @@ func TestRethinkStore_DeleteImage(t *testing.T) {
 	for i := range tests {
 		tt := tests[i]
 		t.Run(tt.name, func(t *testing.T) {
-			err := tt.rs.DeleteImage(tt.args.img)
+			err := tt.rs.DeleteImage(context.Background(), tt.args.img)
 			if (err != nil) != tt.wantErr {
 				t.Errorf("RethinkStore.DeleteImage() error = %v, wantErr %v", err, tt.wantErr)
 				return
@@ -216,7 +217,7 @@ func TestRethinkStore_UpdateImage(t *testing.T) {
 	for i := range tests {
 		tt := tests[i]
 		t.Run(tt.name, func(t *testing.T) {
-			if err := tt.rs.UpdateImage(tt.args.oldImage, tt.args.newImage); (err != nil) != tt.wantErr {
+			if err := tt.rs.UpdateImage(context.Background(), tt.args.oldImage, tt.args.newImage); (err != nil) != tt.wantErr {
 				t.Errorf("RethinkStore.UpdateImage() error = %v, wantErr %v", err, tt.wantErr)
 			}
 		})
@@ -539,7 +540,7 @@ func TestRethinkStore_DeleteOrphanImages(t *testing.T) {
 	for i := range tests {
 		tt := tests[i]
 		t.Run(tt.name, func(t *testing.T) {
-			got, err := tt.rs.DeleteOrphanImages(tt.images, tt.machines)
+			got, err := tt.rs.DeleteOrphanImages(context.Background(), tt.images, tt.machines)
 			if (err != nil) != tt.wantErr {
 				t.Errorf("RethinkStore.DeleteOrphanImages() error = %v, wantErr %v", err, tt.wantErr)
 				return

cmd/metal-api/internal/datastore/integer.go

@@ -1,6 +1,7 @@
 package datastore
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"time"
@@ -138,7 +139,9 @@ func (ip *IntegerPool) initIntegerPool(log *zap.SugaredLogger) error {
 	_, err = ip.infoTable.Insert(integerinfo{
 		ID:            ip.String(),
 		IsInitialized: true,
-	}).RunWrite(ip.session)
+	}).RunWrite(ip.session, r.RunOpts{
+		Context: context.Background(), // this can take a while, do not abort
+	})
 	if err != nil {
 		return err
 	}
@@ -147,14 +150,14 @@ func (ip *IntegerPool) initIntegerPool(log *zap.SugaredLogger) error {
 }
 
 // AcquireRandomUniqueInteger returns a random unique integer from the pool.
-func (ip *IntegerPool) AcquireRandomUniqueInteger() (uint, error) {
+func (ip *IntegerPool) AcquireRandomUniqueInteger(ctx context.Context) (uint, error) {
 	t := ip.poolTable.Limit(1)
 
 	var integer uint
 	err := retry.Do(
 		func() error {
 			var err2 error
-			integer, err2 = ip.genericAcquire(&t)
+			integer, err2 = ip.genericAcquire(ctx, &t)
 			return err2
 		},
 		retry.Attempts(10),
@@ -166,13 +169,13 @@ func (ip *IntegerPool) AcquireRandomUniqueInteger() (uint, error) {
 }
 
 // AcquireUniqueInteger returns a unique integer from the pool.
-func (ip *IntegerPool) AcquireUniqueInteger(value uint) (uint, error) {
+func (ip *IntegerPool) AcquireUniqueInteger(ctx context.Context, value uint) (uint, error) {
 	err := ip.verifyRange(value)
 	if err != nil {
 		return 0, err
 	}
 	t := ip.poolTable.Get(value)
-	return ip.genericAcquire(&t)
+	return ip.genericAcquire(ctx, &t)
 }
 
 // ReleaseUniqueInteger returns a unique integer to the pool.
@@ -193,14 +196,16 @@ func (ip *IntegerPool) ReleaseUniqueInteger(id uint) error {
 	return nil
 }
 
-func (ip *IntegerPool) genericAcquire(term *r.Term) (uint, error) {
+func (ip *IntegerPool) genericAcquire(ctx context.Context, term *r.Term) (uint, error) {
 	res, err := term.Delete(r.DeleteOpts{ReturnChanges: true}).RunWrite(ip.session)
 	if err != nil {
 		return 0, err
 	}
 
 	if len(res.Changes) == 0 {
-		res, err := ip.poolTable.Count().Run(ip.session)
+		res, err := ip.poolTable.Count().Run(ip.session, r.RunOpts{
+			Context: ctx,
+		})
 		if err != nil {
 			return 0, err
 		}