fix: the etcd recovery client and tests

This is the follow-up fix to the PR #5129.

1. Correctly catch only expected errors in the tests.
2. Rewind the snapshot each time the upload is retried.
3. Correctly unwrap errors in the `EtcdRecovery` client.
4. Update the `grpc-proxy` library to pass through the EOF error.

Signed-off-by: Andrey Smirnov <andrey.smirnov@talos-systems.com>
(cherry picked from commit f477507)

Author: smira

go.mod

@@ -98,7 +98,7 @@ require (
 	github.com/talos-systems/go-procfs v0.1.0
 	github.com/talos-systems/go-retry v0.3.1
 	github.com/talos-systems/go-smbios v0.1.1
-	github.com/talos-systems/grpc-proxy v0.2.0
+	github.com/talos-systems/grpc-proxy v0.3.0
 	github.com/talos-systems/net v0.3.2
 	github.com/talos-systems/siderolink v0.1.0
 	github.com/talos-systems/talos/pkg/machinery v1.0.0-beta.2

go.sum

@@ -1255,8 +1255,8 @@ github.com/talos-systems/go-retry v0.3.1 h1:GjjyHB8i1CJpb1O5qYPMljq74cRQ5uiDoyMa
 github.com/talos-systems/go-retry v0.3.1/go.mod h1:HiXQqyVStZ35uSY/MTLWVvQVmC3lIW2MS5VdDaMtoKM=
 github.com/talos-systems/go-smbios v0.1.1 h1:Au6obB/Pp0i0JHhvPlzONk5aoNseosO2BUsmvWWi7y8=
 github.com/talos-systems/go-smbios v0.1.1/go.mod h1:vk76naUSZaWE8Z95wbDn51FgH0goECM4oK3KY2hYSMU=
-github.com/talos-systems/grpc-proxy v0.2.0 h1:DN75bLfaW4xfhq0r0mwFRnfGhSB+HPhK1LNzuMEs9Pw=
-github.com/talos-systems/grpc-proxy v0.2.0/go.mod h1:sm97Vc/z2cok3pu6ruNeszQej4KDxFrDgfWs4C1mtC4=
+github.com/talos-systems/grpc-proxy v0.3.0 h1:pM4P6GqzjmZzq6jAxeP2mzjuTbx9B2E5peVdi2bufM4=
+github.com/talos-systems/grpc-proxy v0.3.0/go.mod h1:1TNaCLt0NTdFdz48OCnOM+HbPEIkj3DhCi85+z6zSM8=
 github.com/talos-systems/net v0.3.2 h1:IMseRyuha8fNsv/3FbQPRE9hLVRBEFR+9sxcoETQ5vI=
 github.com/talos-systems/net v0.3.2/go.mod h1:zhcGixNJz9dgwFiUwc7gkkAqdVqXagU1SNNoIVXYKGo=
 github.com/talos-systems/siderolink v0.1.0 h1:7mkJ9EicQ8J9DHHkwiNYGoccCgFcEIFcmfcKRyI7Y+8=

internal/integration/api/etcd-recover.go

@@ -17,6 +17,7 @@ import (
 	"time"
 
 	"github.com/talos-systems/go-retry/retry"
+	"google.golang.org/grpc/codes"
 
 	"github.com/talos-systems/talos/internal/integration/base"
 	machineapi "github.com/talos-systems/talos/pkg/machinery/api/machine"
@@ -156,7 +157,7 @@ func (suite *EtcdRecoverSuite) TestSnapshotRecover() {
 
 	suite.T().Logf("recovering etcd snapshot at node %q", recoverNode)
 
-	suite.Require().NoError(suite.recoverEtcd(recoverNode, &snapshot))
+	suite.Require().NoError(suite.recoverEtcd(recoverNode, bytes.NewReader(snapshot.Bytes())))
 
 	suite.AssertClusterHealthy(suite.ctx)
 
@@ -197,21 +198,40 @@ func (suite *EtcdRecoverSuite) snapshotEtcd(snapshotNode string, dest io.Writer)
 	return err
 }
 
-func (suite *EtcdRecoverSuite) recoverEtcd(recoverNode string, src io.Reader) error {
+func (suite *EtcdRecoverSuite) recoverEtcd(recoverNode string, src io.ReadSeeker) error {
 	ctx := client.WithNodes(suite.ctx, recoverNode)
 
+	suite.T().Log("uploading the snapshot")
+
 	if err := retry.Constant(time.Minute, retry.WithUnits(time.Millisecond*200)).RetryWithContext(ctx, func(ctx context.Context) error {
-		_, err := suite.Client.EtcdRecover(ctx, src)
+		_, err := src.Seek(0, io.SeekStart)
+		if err != nil {
+			return err
+		}
+
+		_, err = suite.Client.EtcdRecover(ctx, src)
+
+		if client.StatusCode(err) == codes.FailedPrecondition {
+			return retry.ExpectedError(err)
+		}
 
-		return retry.ExpectedError(err)
+		return err
 	}); err != nil {
 		return fmt.Errorf("error uploading snapshot: %w", err)
 	}
 
+	suite.T().Log("bootstrapping from the snapshot")
+
 	return retry.Constant(time.Minute, retry.WithUnits(time.Millisecond*200)).RetryWithContext(ctx, func(ctx context.Context) error {
-		return retry.ExpectedError(suite.Client.Bootstrap(ctx, &machineapi.BootstrapRequest{
+		err := suite.Client.Bootstrap(ctx, &machineapi.BootstrapRequest{
 			RecoverEtcd: true,
-		}))
+		})
+
+		if client.StatusCode(err) == codes.FailedPrecondition || client.StatusCode(err) == codes.DeadlineExceeded {
+			return retry.ExpectedError(err)
+		}
+
+		return err
 	})
 }
 

pkg/machinery/client/client.go

@@ -907,7 +907,9 @@ func (c *Client) EtcdRecover(ctx context.Context, snapshot io.Reader, callOption
 		default:
 		}
 
-		n, err := snapshot.Read(buf)
+		var n int
+
+		n, err = snapshot.Read(buf)
 		if err != nil {
 			if errors.Is(err, io.EOF) {
 				break
@@ -919,11 +921,21 @@ func (c *Client) EtcdRecover(ctx context.Context, snapshot io.Reader, callOption
 		if err = cli.Send(&common.Data{
 			Bytes: buf[:n],
 		}); err != nil {
+			if errors.Is(err, io.EOF) {
+				break
+			}
+
 			return nil, err
 		}
 	}
 
-	return cli.CloseAndRecv()
+	resp, err := cli.CloseAndRecv()
+
+	var filtered interface{}
+	filtered, err = FilterMessages(resp, err)
+	resp, _ = filtered.(*machineapi.EtcdRecoverResponse) //nolint:errcheck
+
+	return resp, err
 }
 
 // GenerateClientConfiguration implements proto.MachineServiceClient interface.