storj/satellite/repair/repair_test.go

// Copyright (C) 2019 Storj Labs, Inc.
// See LICENSE for copying information.

package repair_test

import (
	"context"
	"math"
	"testing"

	"github.com/stretchr/testify/require"
	"go.uber.org/zap"

	"storj.io/storj/internal/memory"
	"storj.io/storj/internal/testcontext"
	"storj.io/storj/internal/testplanet"
	"storj.io/storj/internal/testrand"
	"storj.io/storj/pkg/pb"
	"storj.io/storj/pkg/storj"
	"storj.io/storj/satellite"
	"storj.io/storj/satellite/overlay"
	"storj.io/storj/uplink"
)

// TestDataRepair does the following:
// - Uploads test data
// - Kills some nodes and disqualifies 1
// - Triggers data repair, which repairs the data from the remaining nodes to
//	 the numbers of nodes determined by the upload repair max threshold
// - Shuts down several nodes, but keeping up a number equal to the minim
//	 threshold
// - Downloads the data from those left nodes and check that it's the same than
//   the uploaded one
func TestDataRepair(t *testing.T) {
	var repairMaxExcessRateOptimalThreshold float64

	testplanet.Run(t, testplanet.Config{
		SatelliteCount:   1,
		StorageNodeCount: 14,
		UplinkCount:      1,
		Reconfigure: testplanet.Reconfigure{
			Satellite: func(log *zap.Logger, index int, config *satellite.Config) {
				config.Overlay.Node.OnlineWindow = 0
				repairMaxExcessRateOptimalThreshold = config.Repairer.MaxExcessRateOptimalThreshold
			},
		},
	}, func(t *testing.T, ctx *testcontext.Context, planet *testplanet.Planet) {
		// first, upload some remote data
		uplinkPeer := planet.Uplinks[0]
		satellitePeer := planet.Satellites[0]
		// stop discovery service so that we do not get a race condition when we delete nodes from overlay
		satellitePeer.Discovery.Service.Discovery.Stop()
		satellitePeer.Discovery.Service.Refresh.Stop()
		// stop audit to prevent possible interactions i.e. repair timeout problems
		satellitePeer.Audit.Service.Loop.Stop()

		satellitePeer.Repair.Checker.Loop.Pause()
		satellitePeer.Repair.Repairer.Loop.Pause()

		var (
			testData         = testrand.Bytes(8 * memory.KiB)
			minThreshold     = 3
			successThreshold = 7
		)
		err := uplinkPeer.UploadWithConfig(ctx, satellitePeer, &uplink.RSConfig{
			MinThreshold:     minThreshold,
			RepairThreshold:  5,
			SuccessThreshold: successThreshold,
			MaxThreshold:     10,
		}, "testbucket", "test/path", testData)
		require.NoError(t, err)

		pointer, path := getRemoteSegment(t, ctx, satellitePeer)

		// calculate how many storagenodes to kill
		redundancy := pointer.GetRemote().GetRedundancy()
		minReq := redundancy.GetMinReq()
		remotePieces := pointer.GetRemote().GetRemotePieces()
		numPieces := len(remotePieces)
		// disqualify one storage node
		toDisqualify := 1
		toKill := numPieces - toDisqualify - int(minReq+1)
		require.True(t, toKill >= 1)
		maxNumRepairedPieces := int(
			math.Ceil(
				float64(successThreshold) * (1 + repairMaxExcessRateOptimalThreshold),
			),
		)
		numStorageNodes := len(planet.StorageNodes)
		// Ensure that there are enough storage nodes to upload repaired segments
		require.Falsef(t,
			(numStorageNodes-toKill-toDisqualify) < maxNumRepairedPieces,
			"there is not enough available nodes for repairing: need= %d, have= %d",
			maxNumRepairedPieces, (numStorageNodes - toKill - toDisqualify),
		)

		// kill nodes and track lost pieces
		nodesToKill := make(map[storj.NodeID]bool)
		nodesToDisqualify := make(map[storj.NodeID]bool)
		nodesToKeepAlive := make(map[storj.NodeID]bool)

		var numDisqualified int
		for i, piece := range remotePieces {
			if i >= toKill {
				if numDisqualified < toDisqualify {
					nodesToDisqualify[piece.NodeId] = true
					numDisqualified++
				}
				nodesToKeepAlive[piece.NodeId] = true
				continue
			}
			nodesToKill[piece.NodeId] = true
		}

		for _, node := range planet.StorageNodes {
			if nodesToDisqualify[node.ID()] {
				disqualifyNode(t, ctx, satellitePeer, node.ID())
				continue
			}
			if nodesToKill[node.ID()] {
				err = planet.StopPeer(node)
				require.NoError(t, err)
				_, err = satellitePeer.Overlay.Service.UpdateUptime(ctx, node.ID(), false)
				require.NoError(t, err)
			}
		}

		satellitePeer.Repair.Checker.Loop.Restart()
		satellitePeer.Repair.Checker.Loop.TriggerWait()
		satellitePeer.Repair.Checker.Loop.Pause()
		satellitePeer.Repair.Repairer.Loop.Restart()
		satellitePeer.Repair.Repairer.Loop.TriggerWait()
		satellitePeer.Repair.Repairer.Loop.Pause()
		satellitePeer.Repair.Repairer.Limiter.Wait()

		// repaired segment should not contain any piece in the killed and DQ nodes
		metainfoService := satellitePeer.Metainfo.Service
		pointer, err = metainfoService.Get(ctx, path)
		require.NoError(t, err)

		nodesToKillForMinThreshold := len(remotePieces) - minThreshold
		remotePieces = pointer.GetRemote().GetRemotePieces()
		for _, piece := range remotePieces {
			require.NotContains(t, nodesToKill, piece.NodeId, "there shouldn't be pieces in killed nodes")
			require.NotContains(t, nodesToDisqualify, piece.NodeId, "there shouldn't be pieces in DQ nodes")

			// Kill the original nodes which were kept alive to ensure that we can
			// download from the new nodes that the repaired pieces have been uploaded
			if _, ok := nodesToKeepAlive[piece.NodeId]; ok && nodesToKillForMinThreshold > 0 {
				stopNodeByID(t, ctx, planet, piece.NodeId)
				nodesToKillForMinThreshold--
			}
		}

		// we should be able to download data without any of the original nodes
		newData, err := uplinkPeer.Download(ctx, satellitePeer, "testbucket", "test/path")
		require.NoError(t, err)
		require.Equal(t, newData, testData)
	})
}

// TestRemoveIrreparableSegmentFromQueue
// - Upload tests data to 7 nodes
// - Kill nodes so that repair threshold > online nodes > minimum threshold
// - Call checker to add segment to the repair queue
// - Kill nodes so that online nodes < minimum threshold
// - Run the repairer
// - Verify segment is no longer in the repair queue and segment should be the same
func TestRemoveIrreparableSegmentFromQueue(t *testing.T) {
	testplanet.Run(t, testplanet.Config{
		SatelliteCount:   1,
		StorageNodeCount: 10,
		UplinkCount:      1,
	}, func(t *testing.T, ctx *testcontext.Context, planet *testplanet.Planet) {
		// first, upload some remote data
		uplinkPeer := planet.Uplinks[0]
		satellitePeer := planet.Satellites[0]
		// stop discovery service so that we do not get a race condition when we delete nodes from overlay
		satellitePeer.Discovery.Service.Discovery.Stop()
		satellitePeer.Discovery.Service.Refresh.Stop()
		// stop audit to prevent possible interactions i.e. repair timeout problems
		satellitePeer.Audit.Service.Loop.Stop()

		satellitePeer.Repair.Checker.Loop.Pause()
		satellitePeer.Repair.Repairer.Loop.Pause()

		testData := testrand.Bytes(8 * memory.KiB)

		err := uplinkPeer.UploadWithConfig(ctx, satellitePeer, &uplink.RSConfig{
			MinThreshold:     3,
			RepairThreshold:  5,
			SuccessThreshold: 7,
			MaxThreshold:     7,
		}, "testbucket", "test/path", testData)
		require.NoError(t, err)

		pointer, _ := getRemoteSegment(t, ctx, satellitePeer)

		// kill nodes and track lost pieces
		nodesToDQ := make(map[storj.NodeID]bool)

		// Kill 3 nodes so that pointer has 4 left (less than repair threshold)
		toKill := 3

		remotePieces := pointer.GetRemote().GetRemotePieces()

		for i, piece := range remotePieces {
			if i >= toKill {
				continue
			}
			nodesToDQ[piece.NodeId] = true
		}

		for nodeID := range nodesToDQ {
			disqualifyNode(t, ctx, satellitePeer, nodeID)
		}

		// trigger checker to add segment to repair queue
		satellitePeer.Repair.Checker.Loop.Restart()
		satellitePeer.Repair.Checker.Loop.TriggerWait()
		satellitePeer.Repair.Checker.Loop.Pause()

		// TODO: Verify segment is in queue by making a query to the database

		// Kill nodes so that online nodes < minimum threshold
		// This will make the segment irreparable
		for _, piece := range remotePieces {
			disqualifyNode(t, ctx, satellitePeer, piece.NodeId)
		}

		count, err := satellitePeer.DB.RepairQueue().Count(ctx)
		require.NoError(t, err)
		require.Equal(t, count, 1)

		// Run the repairer
		satellitePeer.Repair.Repairer.Loop.Restart()
		satellitePeer.Repair.Repairer.Loop.TriggerWait()
		satellitePeer.Repair.Repairer.Loop.Pause()
		satellitePeer.Repair.Repairer.Limiter.Wait()

		count, err = satellitePeer.DB.RepairQueue().Count(ctx)
		require.NoError(t, err)
		require.Equal(t, count, 0)
	})
}

// TestRepairMultipleDisqualified does the following:
// - Uploads test data to 7 nodes
// - Disqualifies 3 nodes
// - Triggers data repair, which repairs the data from the remaining 4 nodes to additional 3 new nodes
// - Shuts down the 4 nodes from which the data was repaired
// - Now we have just the 3 new nodes to which the data was repaired
// - Downloads the data from these 3 nodes (succeeds because 3 nodes are enough for download)
func TestRepairMultipleDisqualified(t *testing.T) {
	testplanet.Run(t, testplanet.Config{
		SatelliteCount:   1,
		StorageNodeCount: 12,
		UplinkCount:      1,
	}, func(t *testing.T, ctx *testcontext.Context, planet *testplanet.Planet) {
		// first, upload some remote data
		uplinkPeer := planet.Uplinks[0]
		satellitePeer := planet.Satellites[0]
		// stop discovery service so that we do not get a race condition when we delete nodes from overlay
		satellitePeer.Discovery.Service.Discovery.Stop()
		satellitePeer.Discovery.Service.Refresh.Stop()

		satellitePeer.Repair.Checker.Loop.Pause()
		satellitePeer.Repair.Repairer.Loop.Pause()

		testData := testrand.Bytes(8 * memory.KiB)

		err := uplinkPeer.UploadWithConfig(ctx, satellitePeer, &uplink.RSConfig{
			MinThreshold:     3,
			RepairThreshold:  5,
			SuccessThreshold: 7,
			MaxThreshold:     7,
		}, "testbucket", "test/path", testData)
		require.NoError(t, err)

		// get a remote segment from metainfo
		metainfo := satellitePeer.Metainfo.Service
		listResponse, _, err := metainfo.List(ctx, "", "", "", true, 0, 0)
		require.NoError(t, err)

		var path string
		var pointer *pb.Pointer
		for _, v := range listResponse {
			path = v.GetPath()
			pointer, err = metainfo.Get(ctx, path)
			require.NoError(t, err)
			if pointer.GetType() == pb.Pointer_REMOTE {
				break
			}
		}

		// calculate how many storagenodes to disqualify
		numStorageNodes := len(planet.StorageNodes)
		redundancy := pointer.GetRemote().GetRedundancy()
		remotePieces := pointer.GetRemote().GetRemotePieces()
		minReq := redundancy.GetMinReq()
		numPieces := len(remotePieces)
		toDisqualify := numPieces - (int(minReq + 1))
		// we should have enough storage nodes to repair on
		require.True(t, (numStorageNodes-toDisqualify) >= numPieces)

		// disqualify nodes and track lost pieces
		var lostPieces []int32
		nodesToDisqualify := make(map[storj.NodeID]bool)
		nodesToKeepAlive := make(map[storj.NodeID]bool)

		for i, piece := range remotePieces {
			if i >= toDisqualify {
				nodesToKeepAlive[piece.NodeId] = true
				continue
			}
			nodesToDisqualify[piece.NodeId] = true
			lostPieces = append(lostPieces, piece.GetPieceNum())
		}

		for _, node := range planet.StorageNodes {
			if nodesToDisqualify[node.ID()] {
				disqualifyNode(t, ctx, satellitePeer, node.ID())
			}
		}

		err = satellitePeer.Repair.Checker.RefreshReliabilityCache(ctx)
		require.NoError(t, err)

		satellitePeer.Repair.Checker.Loop.TriggerWait()
		satellitePeer.Repair.Repairer.Loop.TriggerWait()
		satellitePeer.Repair.Repairer.Limiter.Wait()

		// kill nodes kept alive to ensure repair worked
		for _, node := range planet.StorageNodes {
			if nodesToKeepAlive[node.ID()] {
				err = planet.StopPeer(node)
				require.NoError(t, err)

				_, err = satellitePeer.Overlay.Service.UpdateUptime(ctx, node.ID(), false)
				require.NoError(t, err)
			}
		}

		// we should be able to download data without any of the original nodes
		newData, err := uplinkPeer.Download(ctx, satellitePeer, "testbucket", "test/path")
		require.NoError(t, err)
		require.Equal(t, newData, testData)

		// updated pointer should not contain any of the disqualified nodes
		pointer, err = metainfo.Get(ctx, path)
		require.NoError(t, err)

		remotePieces = pointer.GetRemote().GetRemotePieces()
		for _, piece := range remotePieces {
			require.False(t, nodesToDisqualify[piece.NodeId])
		}
	})
}

// TestDataRepairUploadLimits does the following:
// - Uploads test data to nodes
// - Get one segment of that data to check in which nodes its pieces are stored
// - Kills as many nodes as needed which store such segment pieces
// - Triggers data repair
// - Verify that the number of pieces which repaired has uploaded don't overpass
//	 the established limit (success threshold + % of excess)
func TestDataRepairUploadLimit(t *testing.T) {
	var repairMaxExcessRateOptimalThreshold float64

	testplanet.Run(t, testplanet.Config{
		SatelliteCount:   1,
		StorageNodeCount: 13,
		UplinkCount:      1,
		Reconfigure: testplanet.Reconfigure{
			Satellite: func(log *zap.Logger, index int, config *satellite.Config) {
				repairMaxExcessRateOptimalThreshold = config.Repairer.MaxExcessRateOptimalThreshold
			},
		},
	}, func(t *testing.T, ctx *testcontext.Context, planet *testplanet.Planet) {
		satellite := planet.Satellites[0]
		// stop discovery service so that we do not get a race condition when we delete nodes from overlay
		satellite.Discovery.Service.Discovery.Stop()
		satellite.Discovery.Service.Refresh.Stop()
		// stop audit to prevent possible interactions i.e. repair timeout problems
		satellite.Audit.Service.Loop.Stop()
		satellite.Repair.Checker.Loop.Pause()
		satellite.Repair.Repairer.Loop.Pause()

		const (
			repairThreshold  = 5
			successThreshold = 7
			maxThreshold     = 9
		)
		var (
			maxRepairUploadThreshold = int(
				math.Ceil(
					float64(successThreshold) * (1 + repairMaxExcessRateOptimalThreshold),
				),
			)
			ul       = planet.Uplinks[0]
			testData = testrand.Bytes(8 * memory.KiB)
		)

		err := ul.UploadWithConfig(ctx, satellite, &uplink.RSConfig{
			MinThreshold:     3,
			RepairThreshold:  repairThreshold,
			SuccessThreshold: successThreshold,
			MaxThreshold:     maxThreshold,
		}, "testbucket", "test/path", testData)
		require.NoError(t, err)

		pointer, path := getRemoteSegment(t, ctx, satellite)
		originalPieces := pointer.GetRemote().GetRemotePieces()
		require.True(t, len(originalPieces) <= maxThreshold)

		{ // Check that there is enough nodes in the network which don't contain
			// pieces of the segment for being able to repair the lost pieces
			availableNumNodes := len(planet.StorageNodes) - len(originalPieces)
			neededNodesForRepair := maxRepairUploadThreshold - repairThreshold
			require.Truef(t,
				availableNumNodes >= neededNodesForRepair,
				"Not enough remaining nodes in the network for repairing the pieces: have= %d, need= %d",
				availableNumNodes, neededNodesForRepair,
			)
		}

		originalStorageNodes := make(map[storj.NodeID]struct{})
		for _, p := range originalPieces {
			originalStorageNodes[p.NodeId] = struct{}{}
		}

		killedNodes := make(map[storj.NodeID]struct{})
		{ // Register nodes of the network which don't have pieces for the segment
			// to be injured and ill nodes which have pieces of the segment in order
			// to injure it
			numNodesToKill := len(originalPieces) - repairThreshold
			for _, node := range planet.StorageNodes {
				if _, ok := originalStorageNodes[node.ID()]; !ok {
					continue
				}

				if len(killedNodes) < numNodesToKill {
					err = planet.StopPeer(node)
					require.NoError(t, err)
					_, err = satellite.Overlay.Service.UpdateUptime(ctx, node.ID(), false)
					require.NoError(t, err)

					killedNodes[node.ID()] = struct{}{}
				}
			}
		}

		satellite.Repair.Checker.Loop.Restart()
		satellite.Repair.Checker.Loop.TriggerWait()
		satellite.Repair.Checker.Loop.Pause()
		satellite.Repair.Repairer.Loop.Restart()
		satellite.Repair.Repairer.Loop.TriggerWait()
		satellite.Repair.Repairer.Loop.Pause()
		satellite.Repair.Repairer.Limiter.Wait()

		// Get the pointer after repair to check the nodes where the pieces are
		// stored
		pointer, err = satellite.Metainfo.Service.Get(ctx, path)
		require.NoError(t, err)

		// Check that repair has uploaded missed pieces to an expected number of
		// nodes
		afterRepairPieces := pointer.GetRemote().GetRemotePieces()
		require.Falsef(t,
			len(afterRepairPieces) > maxRepairUploadThreshold,
			"Repaired pieces cannot be over max repair upload threshold. maxRepairUploadThreshold= %d, have= %d",
			maxRepairUploadThreshold, len(afterRepairPieces),
		)
		require.Falsef(t,
			len(afterRepairPieces) < successThreshold,
			"Repaired pieces shouldn't be under success threshold. successThreshold= %d, have= %d",
			successThreshold, len(afterRepairPieces),
		)

		// Check that after repair, the segment doesn't have more pieces on the
		// killed nodes
		for _, p := range afterRepairPieces {
			require.NotContains(t, killedNodes, p.NodeId, "there shouldn't be pieces in killed nodes")
		}
	})
}

func isDisqualified(t *testing.T, ctx *testcontext.Context, satellite *satellite.Peer, nodeID storj.NodeID) bool {
	node, err := satellite.Overlay.Service.Get(ctx, nodeID)
	require.NoError(t, err)

	return node.Disqualified != nil
}

func disqualifyNode(t *testing.T, ctx *testcontext.Context, satellite *satellite.Peer, nodeID storj.NodeID) {
	_, err := satellite.DB.OverlayCache().UpdateStats(ctx, &overlay.UpdateRequest{
		NodeID:       nodeID,
		IsUp:         true,
		AuditSuccess: false,
		AuditLambda:  0,
		AuditWeight:  1,
		AuditDQ:      0.5,
		UptimeLambda: 1,
		UptimeWeight: 1,
		UptimeDQ:     0.5,
	})
	require.NoError(t, err)
	require.True(t, isDisqualified(t, ctx, satellite, nodeID))
}

// getRemoteSegment returns a remote pointer its path from satellite.
// nolint:golint
func getRemoteSegment(
	t *testing.T, ctx context.Context, satellite *satellite.Peer,
) (_ *pb.Pointer, path string) {
	t.Helper()

	// get a remote segment from metainfo
	metainfo := satellite.Metainfo.Service
	listResponse, _, err := metainfo.List(ctx, "", "", "", true, 0, 0)
	require.NoError(t, err)

	for _, v := range listResponse {
		path := v.GetPath()
		pointer, err := metainfo.Get(ctx, path)
		require.NoError(t, err)
		if pointer.GetType() == pb.Pointer_REMOTE {
			return pointer, path
		}
	}

	t.Fatal("satellite doesn't have any remote segment")
	return nil, ""
}

// nolint:golint
func stopNodeByID(t *testing.T, ctx context.Context, planet *testplanet.Planet, nodeID storj.NodeID) {
	t.Helper()

	for _, node := range planet.StorageNodes {
		if node.ID() == nodeID {
			err := planet.StopPeer(node)
			require.NoError(t, err)

			for _, sat := range planet.Satellites {
				_, err = sat.Overlay.Service.UpdateUptime(ctx, node.ID(), false)
				require.NoError(t, err)
			}

			break
		}
	}
}