database: add lock support

database/database.go

@@ -1,6 +1,9 @@
 package database
 
-import "errors"
+import (
+	"errors"
+	"time"
+)
 
 var (
 	// ErrTransaction is an error that occurs when a database transaction fails.
@@ -39,9 +42,9 @@ type Datastore interface {
 	GetKeyValue(key string) (string, error)
 
 	// Lock
-	// Lock(name string, duration time.Duration, owner string) (bool, time.Time)
+	Lock(name string, owner string, duration time.Duration, renew bool) (bool, time.Time)
-	// Unlock(name, owner string)
+	Unlock(name, owner string)
-	// LockInfo(name string) (string, time.Time, error)
+	FindLock(name string) (string, time.Time, error)
 
 	Close()
 }

database/pgsql/layer.go

@@ -352,6 +352,6 @@ func createNV(features []database.FeatureVersion) (map[string]*database.FeatureV
 }
 
 func (pgSQL *pgSQL) DeleteLayer(name string) error {
-	// TODO
+	// TODO(Quentin-M): Implement and test me.
 	return nil
 }

database/pgsql/lock.go

@@ -0,0 +1,88 @@
+package pgsql
+
+import (
+	"database/sql"
+	"time"
+
+	cerrors "github.com/coreos/clair/utils/errors"
+)
+
+// Lock tries to set a temporary lock in the database.
+//
+// Lock does not block, instead, it returns true and its expiration time
+// is the lock has been successfully acquired or false otherwise
+func (pgSQL *pgSQL) Lock(name string, owner string, duration time.Duration, renew bool) (bool, time.Time) {
+	if name == "" || owner == "" || duration == 0 {
+		log.Warning("could not create an invalid lock")
+		return false, time.Time{}
+	}
+
+	// Prune locks.
+	pgSQL.pruneLocks()
+
+	// Compute expiration.
+	until := time.Now().Add(duration)
+
+	if renew {
+		// Renew lock.
+		r, err := pgSQL.Exec(getQuery("u_lock"), name, owner, until)
+		if err != nil {
+			handleError("u_lock", err)
+			return false, until
+		}
+		if n, _ := r.RowsAffected(); n > 0 {
+			// Updated successfully.
+			return true, until
+		}
+	}
+
+	// Lock.
+	_, err := pgSQL.Exec(getQuery("i_lock"), name, owner, until)
+	if err != nil {
+		if !isErrUniqueViolation(err) {
+			handleError("i_lock", err)
+		}
+		return false, until
+	}
+
+	return true, until
+}
+
+// Unlock unlocks a lock specified by its name if I own it
+func (pgSQL *pgSQL) Unlock(name, owner string) {
+	if name == "" || owner == "" {
+		log.Warning("could not delete an invalid lock")
+		return
+	}
+
+	pgSQL.Exec(getQuery("r_lock"), name, owner)
+}
+
+// FindLock returns the owner of a lock specified by its name and its
+// expiration time.
+func (pgSQL *pgSQL) FindLock(name string) (string, time.Time, error) {
+	if name == "" {
+		log.Warning("could not find an invalid lock")
+		return "", time.Time{}, cerrors.NewBadRequestError("could not find an invalid lock")
+	}
+
+	var owner string
+	var until time.Time
+	err := pgSQL.QueryRow(getQuery("f_lock"), name).Scan(&owner, &until)
+
+	if err == sql.ErrNoRows {
+		return owner, until, cerrors.ErrNotFound
+	}
+	if err != nil {
+		return owner, until, handleError("f_lock", err)
+	}
+
+	return owner, until, nil
+}
+
+// pruneLocks removes every expired locks from the database
+func (pgSQL *pgSQL) pruneLocks() {
+	if _, err := pgSQL.Exec(getQuery("r_lock_expired")); err != nil {
+		handleError("r_lock_expired", err)
+	}
+}

database/pgsql/lock_test.go

@@ -0,0 +1,55 @@
+package pgsql
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestLock(t *testing.T) {
+	datastore, err := OpenForTest("InsertNamespace", false)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	defer datastore.Close()
+
+	var l bool
+	var et time.Time
+
+	// Create a first lock.
+	l, _ = datastore.Lock("test1", "owner1", time.Minute, false)
+	assert.True(t, l)
+
+	// Try to lock the same lock with another owner.
+	l, _ = datastore.Lock("test1", "owner2", time.Minute, true)
+	assert.False(t, l)
+
+	l, _ = datastore.Lock("test1", "owner2", time.Minute, false)
+	assert.False(t, l)
+
+	// Renew the lock.
+	l, _ = datastore.Lock("test1", "owner1", 2*time.Minute, true)
+	assert.True(t, l)
+
+	// Unlock and then relock by someone else.
+	datastore.Unlock("test1", "owner1")
+
+	l, et = datastore.Lock("test1", "owner2", time.Minute, false)
+	assert.True(t, l)
+
+	// LockInfo
+	o, et2, err := datastore.FindLock("test1")
+	assert.Nil(t, err)
+	assert.Equal(t, "owner2", o)
+	assert.Equal(t, et.Second(), et2.Second())
+
+	// Create a second lock which is actually already expired ...
+	l, _ = datastore.Lock("test2", "owner1", -time.Minute, false)
+	assert.True(t, l)
+
+	// Take over the lock
+	l, _ = datastore.Lock("test2", "owner2", time.Minute, false)
+	assert.True(t, l)
+}

database/pgsql/migrations/20151222113213_Initial.sql

@@ -15,12 +15,13 @@ CREATE TABLE IF NOT EXISTS Layer (
   id SERIAL PRIMARY KEY,
   name VARCHAR(128) NOT NULL UNIQUE,
   engineversion SMALLINT NOT NULL,
-  parent_id INT NULL REFERENCES Layer,
+  parent_id INT NULL REFERENCES Layer ON DELETE CASCADE,
   namespace_id INT NULL REFERENCES Namespace);
 
 CREATE INDEX ON Layer (parent_id);
 CREATE INDEX ON Layer (namespace_id);
 
+
 -- -----------------------------------------------------
 -- Table Feature
 -- -----------------------------------------------------
@@ -31,6 +32,7 @@ CREATE TABLE IF NOT EXISTS Feature (
 
   UNIQUE (namespace_id, name));
 
+
 -- -----------------------------------------------------
 -- Table FeatureVersion
 -- -----------------------------------------------------
@@ -113,6 +115,18 @@ CREATE TABLE IF NOT EXISTS KeyValue (
   key VARCHAR(128) NOT NULL UNIQUE,
   value TEXT);
 
+
+-- -----------------------------------------------------
+-- Table Lock
+-- -----------------------------------------------------
+CREATE TABLE IF NOT EXISTS Lock (
+  id SERIAL PRIMARY KEY,
+  name VARCHAR(64) NOT NULL UNIQUE,
+  owner VARCHAR(64) NOT NULL,
+  until TIMESTAMP WITH TIME ZONE);
+
+CREATE INDEX ON Lock (owner);
+
 -- +goose Down
 
 DROP TABLE IF EXISTS Namespace,
@@ -124,4 +138,5 @@ DROP TABLE IF EXISTS Namespace,
                      Vulnerability_FixedIn_Feature,
                      Vulnerability_Affects_FeatureVersion,
                      KeyValue
+                     Lock
             CASCADE;

database/pgsql/queries.go

@@ -131,6 +131,17 @@ func init() {
       SELECT $1, fv.id, $2
 	    FROM FeatureVersion fv
 	    WHERE fv.id = ANY($3::integer[])`
+
+	// lock.go
+	queries["i_lock"] = `INSERT INTO Lock(name, owner, until) VALUES($1, $2, $3)`
+
+	queries["f_lock"] = `SELECT owner, until FROM Lock WHERE name = $1`
+
+	queries["u_lock"] = `UPDATE Lock SET until = $3 WHERE name = $1 AND owner = $2`
+
+	queries["r_lock"] = `DELETE FROM Lock WHERE name = $1 AND owner = $2`
+
+	queries["r_lock_expired"] = `DELETE FROM LOCK WHERE until < CURRENT_TIMESTAMP`
 }
 
 func getQuery(name string) string {

updater/updater.go

@@ -101,7 +101,7 @@ func Run(config *config.UpdaterConfig, st *utils.Stopper) {
 				}
 				continue
 			} else {
-				lockOwner, lockExpiration, err := database.LockInfo(flagName)
+				lockOwner, lockExpiration, err := database.FindLock(flagName)
 				if err != nil {
 					log.Debug("update lock is already taken")
 					nextUpdate = hasLockUntil