cmd/identity: allow using redis for RevocationDB (#3259)

cmd/identity/revocations.go

@@ -27,7 +27,7 @@ var (
 	}
 
 	revCfg struct {
-		RevocationDBURL string `default:"bolt://$CONFDIR/revocations.db" help:"url for revocation database (e.g. bolt://some.db OR redis://127.0.0.1:6378?db=2&password=abc123)"`
+		RevocationDBURL string `default:"bolt://$CONFDIR/revocations.db" help:"url for revocation database (e.g. bolt://some.db OR redis://127.0.0.1:6379?db=2&password=abc123)"`
 	}
 )
 
@@ -42,7 +42,6 @@ func cmdRevocations(cmd *cobra.Command, args []string) error {
 	if len(args) > 0 {
 		revCfg.RevocationDBURL = "bolt://" + filepath.Join(configDir, args[0], "revocations.db")
 	}
-
 	revDB, err := revocation.NewDB(revCfg.RevocationDBURL)
 	if err != nil {
 		return err

cmd/storj-sim/main.go

@@ -102,7 +102,6 @@ func main() {
 	rootCmd.AddCommand(
 		networkCmd,
 	)
-
 	rootCmd.SilenceUsage = true
 	err := rootCmd.Execute()
 	if err != nil {

cmd/storj-sim/network.go

@@ -7,6 +7,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"io/ioutil"
 	"net"
 	"os"
 	"os/exec"
@@ -61,6 +62,7 @@ const (
 	// satellite specific constants
 	debugPeerHTTP     = 7
 	debugRepairerHTTP = 8
+	redisPort         = 4
 )
 
 // port creates a port with a consistent format for storj-sim services.
@@ -103,6 +105,7 @@ func networkExec(flags *Flags, args []string, command string) error {
 
 func networkEnv(flags *Flags, args []string) error {
 	flags.OnlyEnv = true
+
 	processes, err := newNetwork(flags)
 	if err != nil {
 		return err
@@ -222,7 +225,6 @@ func newNetwork(flags *Flags) (*Processes, error) {
 	versioncontrol.ExecBefore["run"] = func(process *Process) error {
 		return readConfigString(&versioncontrol.Address, versioncontrol.Directory, "address")
 	}
-
 	// gateway must wait for the versioncontrol to start up
 
 	// Create satellites
@@ -230,6 +232,38 @@ func newNetwork(flags *Flags) (*Processes, error) {
 		return nil, fmt.Errorf("exceeded the max instance count of %d with Satellite count of %d", maxInstanceCount, flags.SatelliteCount)
 	}
 
+	// set up redis servers
+	var redisServers []*Process
+	for i := 0; i < flags.SatelliteCount; i++ {
+		rp := port(satellitePeer, i, redisPort)
+		process := processes.New(Info{
+			Name:       fmt.Sprintf("redis/%d", i),
+			Executable: "redis-server",
+			Directory:  filepath.Join(processes.Directory, "satellite", fmt.Sprint(i), "redis"),
+			Address:    net.JoinHostPort(host, rp),
+		})
+		redisServers = append(redisServers, process)
+
+		process.ExecBefore["setup"] = func(process *Process) error {
+			confpath := filepath.Join(process.Directory, "redis.conf")
+			arguments := []string{
+				"daemonize no",
+				"bind " + host,
+				"port " + rp,
+				"timeout 0",
+				"databases 2",
+				"dbfilename sim.rdb",
+				"dir ./",
+			}
+			conf := strings.Join(arguments, "\n") + "\n"
+			err := ioutil.WriteFile(confpath, []byte(conf), 0755)
+			return err
+		}
+		process.Arguments = Arguments{
+			"run": []string{filepath.Join(process.Directory, "redis.conf")},
+		}
+	}
+
 	var satellites []*Process
 	for i := 0; i < flags.SatelliteCount; i++ {
 		process := processes.New(Info{
@@ -255,6 +289,9 @@ func newNetwork(flags *Flags) (*Processes, error) {
 				"--server.address", process.Address,
 				"--server.private-address", net.JoinHostPort(host, port(satellitePeer, i, privateGRPC)),
 
+				"--live-accounting.storage-backend", "redis://" + redisServers[i].Address + "?db=0",
+				"--server.revocation-dburl", "redis://" + redisServers[i].Address + "?db=1",
+
 				"--server.extensions.revocation=false",
 				"--server.use-peer-ca-whitelist=false",
 
@@ -273,7 +310,7 @@ func newNetwork(flags *Flags) (*Processes, error) {
 				"--metainfo.database-url", pgutil.ConnstrWithSchema(flags.Postgres, fmt.Sprintf("satellite/%d/meta", i)),
 			)
 		}
-
+		process.WaitForStart(redisServers[i])
 		process.ExecBefore["run"] = func(process *Process) error {
 			return readConfigString(&process.Address, process.Directory, "server.address")
 		}
@@ -293,7 +330,6 @@ func newNetwork(flags *Flags) (*Processes, error) {
 				"--debug.addr", net.JoinHostPort(host, port(satellitePeer, i, debugPeerHTTP)),
 			},
 		})
-
 		process.WaitForStart(satellite)
 	}
 
@@ -311,7 +347,6 @@ func newNetwork(flags *Flags) (*Processes, error) {
 				"--debug.addr", net.JoinHostPort(host, port(satellitePeer, i, debugRepairerHTTP)),
 			},
 		})
-
 		process.WaitForStart(satellite)
 	}
 
@@ -529,8 +564,8 @@ func identitySetup(network *Processes) (*Processes, error) {
 	processes := NewProcesses(network.Directory)
 
 	for _, process := range network.List {
-		if process.Info.Executable == "gateway" {
+		if process.Info.Executable == "gateway" || process.Info.Executable == "redis-server" {
-			// gateways don't need an identity
+			// gateways and redis-servers don't need an identity
 			continue
 		}
 

pkg/revocation/common.go

@@ -27,7 +27,6 @@ func NewDB(dbURL string) (*DB, error) {
 	if err != nil {
 		return nil, extensions.ErrRevocationDB.Wrap(err)
 	}
-
 	var db *DB
 	switch driver {
 	case "bolt":
@@ -43,7 +42,6 @@ func NewDB(dbURL string) (*DB, error) {
 	default:
 		return nil, extensions.ErrRevocationDB.New("database scheme not supported: %s", driver)
 	}
-
 	return db, nil
 }
 

pkg/revocation/db_test.go

@@ -38,7 +38,7 @@ func TestRevocationDB_Get(t *testing.T) {
 		{
 			t.Log("missing key")
 			rev, err = revDB.Get(ctx, chain)
-			assert.NoError(t, err)
+			require.NoError(t, err)
 			assert.Nil(t, rev)
 
 			nodeID, err := identity.NodeIDFromCert(chain[peertls.CAIndex])
@@ -51,10 +51,10 @@ func TestRevocationDB_Get(t *testing.T) {
 		{
 			t.Log("existing key")
 			rev, err = revDB.Get(ctx, chain)
-			assert.NoError(t, err)
+			require.NoError(t, err)
 
 			revBytes, err := rev.Marshal()
-			assert.NoError(t, err)
+			require.NoError(t, err)
 			assert.True(t, bytes.Equal(ext.Value, revBytes))
 		}
 	})
@@ -75,7 +75,7 @@ func TestRevocationDB_Put_success(t *testing.T) {
 		// identity to be valid.
 		time.Sleep(time.Second)
 		newerRevocation, err := extensions.NewRevocationExt(keys[peertls.CAIndex], chain[peertls.LeafIndex])
-		assert.NoError(t, err)
+		require.NoError(t, err)
 
 		testcases := []struct {
 			name string
@@ -119,7 +119,7 @@ func TestRevocationDB_Put_error(t *testing.T) {
 		require.NoError(t, err)
 
 		olderRevocation, err := extensions.NewRevocationExt(keys[peertls.CAIndex], chain[peertls.LeafIndex])
-		assert.NoError(t, err)
+		require.NoError(t, err)
 
 		time.Sleep(time.Second)
 		newerRevocation, err := extensions.NewRevocationExt(keys[peertls.CAIndex], chain[peertls.LeafIndex])
@@ -151,3 +151,48 @@ func TestRevocationDB_Put_error(t *testing.T) {
 		}
 	})
 }
+
+func TestRevocationDB_List(t *testing.T) {
+	ctx := testcontext.New(t)
+	defer ctx.Cleanup()
+
+	testrevocation.RunDBs(t, func(t *testing.T, revDB extensions.RevocationDB, db storage.KeyValueStore) {
+		keys, chain, err := testpeertls.NewCertChain(2, storj.LatestIDVersion().Number)
+		require.NoError(t, err)
+		keys2, chain2, err := testpeertls.NewCertChain(2, storj.LatestIDVersion().Number)
+		require.NoError(t, err)
+
+		// test list no revocations, should not error
+		revs, err := revDB.List(ctx)
+		require.NoError(t, err)
+		assert.Nil(t, revs)
+
+		// list 1,2 revocations
+		firstRevocation, err := extensions.NewRevocationExt(keys[peertls.CAIndex], chain[peertls.LeafIndex])
+		require.NoError(t, err)
+
+		err = revDB.Put(ctx, chain, firstRevocation)
+		require.NoError(t, err)
+		revs, err = revDB.List(ctx)
+		require.NoError(t, err)
+		require.Equal(t, 1, len(revs))
+		revBytes, err := revs[0].Marshal()
+		require.NoError(t, err)
+		assert.True(t, bytes.Equal(firstRevocation.Value, revBytes))
+
+		secondRevocation, err := extensions.NewRevocationExt(keys2[peertls.CAIndex], chain2[peertls.LeafIndex])
+		require.NoError(t, err)
+		err = revDB.Put(ctx, chain2, secondRevocation)
+		require.NoError(t, err)
+		revs, err = revDB.List(ctx)
+		require.NoError(t, err)
+		assert.Equal(t, 2, len(revs))
+
+		expected := [][]byte{firstRevocation.Value, secondRevocation.Value}
+		for _, rev := range revs {
+			revBytes, err := rev.Marshal()
+			require.NoError(t, err)
+			assert.Contains(t, expected, revBytes)
+		}
+	})
+}

scripts/test-sim-backwards.sh

@@ -52,6 +52,16 @@ sed -i -e 's#/release/#/branch/#g' `storj-sim network env SATELLITE_0_DIR`/confi
 
 # replace any 140XX port with 100XX port to fix, satellite.API part removal from satellite.Peer
 sed -i -e "s#$STORJ_NETWORK_HOST4:100#$STORJ_NETWORK_HOST4:140#g" `storj-sim network env SATELLITE_0_DIR`/config.yaml
+REDIS_CONFIG=$(storj-sim network env REDIS_0_DIR)/redis.conf
+if [ ! -f "$REDIS_CONFIG" ] ; then
+    echo "daemonize no" >> $REDIS_CONFIG
+    echo "bind $STORJ_NETWORK_HOST4" >> $REDIS_CONFIG
+    echo "port 10004" >> $REDIS_CONFIG
+    echo "timeout 0" >> $REDIS_CONFIG
+    echo "databases 2" >> $REDIS_CONFIG
+    echo "dbfilename sim.rdb" >> $REDIS_CONFIG
+    echo "dir ./" >> $REDIS_CONFIG
+fi
 
 ## Ensure that partially upgraded network works
 

storage/redis/client.go

@@ -128,6 +128,9 @@ func (client *Client) Close() error {
 // is requested, an error will be returned
 func (client *Client) GetAll(ctx context.Context, keys storage.Keys) (_ storage.Values, err error) {
 	defer mon.Task()(&ctx)(&err)
+	if len(keys) == 0 {
+		return nil, nil
+	}
 	if len(keys) > storage.LookupLimit {
 		return nil, storage.ErrLimitExceeded
 	}
@@ -141,6 +144,7 @@ func (client *Client) GetAll(ctx context.Context, keys storage.Keys) (_ storage.
 	if err != nil {
 		return nil, err
 	}
+
 	values := []storage.Value{}
 	for _, result := range results {
 		if result == nil {