chore: remove <-errCh where possible in grpc methods

Simplify code by passing error directly into the pipe closer.

Signed-off-by: Dmitriy Matrenichev <dmitry.matrenichev@siderolabs.com>

cmd/talosctl/cmd/talos/copy.go

@@ -10,7 +10,6 @@ import (
 	"io"
 	"os"
 	"path/filepath"
-	"sync"
 
 	"github.com/spf13/cobra"
 
@@ -48,23 +47,11 @@ captures ownership and permission bits.`,
 				return err
 			}
 
-			r, errCh, err := c.Copy(ctx, args[0])
+			r, err := c.Copy(ctx, args[0])
 			if err != nil {
 				return fmt.Errorf("error copying: %w", err)
 			}
 
-			var wg sync.WaitGroup
-
-			wg.Add(1)
-			go func() {
-				defer wg.Done()
-				for err := range errCh {
-					fmt.Fprintln(os.Stderr, err.Error())
-				}
-			}()
-
-			defer wg.Wait()
-
 			localPath := args[1]
 
 			if localPath == "-" {

cmd/talosctl/cmd/talos/etcd.go

@@ -11,7 +11,6 @@ import (
 	"io"
 	"os"
 	"strings"
-	"sync"
 	"text/tabwriter"
 
 	"github.com/dustin/go-humanize"
@@ -350,25 +349,13 @@ var etcdSnapshotCmd = &cobra.Command{
 
 			defer dest.Close() //nolint:errcheck
 
-			r, errCh, err := c.EtcdSnapshot(ctx, &machine.EtcdSnapshotRequest{})
+			r, err := c.EtcdSnapshot(ctx, &machine.EtcdSnapshotRequest{})
 			if err != nil {
 				return fmt.Errorf("error reading file: %w", err)
 			}
 
 			defer r.Close() //nolint:errcheck
 
-			var wg sync.WaitGroup
-
-			wg.Add(1)
-			go func() {
-				defer wg.Done()
-				for err := range errCh {
-					fmt.Fprintln(os.Stderr, err.Error())
-				}
-			}()
-
-			defer wg.Wait()
-
 			size, err := io.Copy(dest, r)
 			if err != nil {
 				return fmt.Errorf("error reading: %w", err)

cmd/talosctl/cmd/talos/kubeconfig.go

@@ -11,7 +11,6 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
-	"sync"
 
 	"github.com/mattn/go-isatty"
 	"github.com/siderolabs/go-kubeconfig"
@@ -92,22 +91,11 @@ Otherwise kubeconfig will be written to PWD or [local-path] if specified.`,
 				}
 			}
 
-			r, errCh, err := c.KubeconfigRaw(ctx)
+			r, err := c.KubeconfigRaw(ctx)
 			if err != nil {
 				return fmt.Errorf("error copying: %w", err)
 			}
 
-			var wg sync.WaitGroup
-
-			wg.Add(1)
-			go func() {
-				defer wg.Done()
-				for err := range errCh {
-					fmt.Fprintln(os.Stderr, err.Error())
-				}
-			}()
-
-			defer wg.Wait()
 			defer r.Close() //nolint:errcheck
 
 			data, err := helpers.ExtractFileFromTarGz("kubeconfig", r)

cmd/talosctl/cmd/talos/pcap.go

@@ -11,7 +11,6 @@ import (
 	"io"
 	"os"
 	"strings"
-	"sync"
 	"time"
 
 	"github.com/gopacket/gopacket"
@@ -90,27 +89,11 @@ e.g. by excluding packets with the port 50000.
 				return err
 			}
 
-			r, errCh, err := c.PacketCapture(ctx, &req)
+			r, err := c.PacketCapture(ctx, &req)
 			if err != nil {
 				return fmt.Errorf("error copying: %w", err)
 			}
 
-			var wg sync.WaitGroup
-
-			wg.Add(1)
-			go func() {
-				defer wg.Done()
-				for err := range errCh {
-					if client.StatusCode(err) == codes.DeadlineExceeded {
-						continue
-					}
-
-					fmt.Fprintln(os.Stderr, err.Error())
-				}
-			}()
-
-			defer wg.Wait()
-
 			if pcapCmdFlags.output == "" {
 				return dumpPackets(r)
 			}

cmd/talosctl/cmd/talos/read.go

@@ -11,7 +11,6 @@ import (
 	"os"
 
 	"github.com/spf13/cobra"
-	"golang.org/x/sync/errgroup"
 
 	"github.com/siderolabs/talos/cmd/talosctl/pkg/talos/helpers"
 	"github.com/siderolabs/talos/pkg/machinery/client"
@@ -37,37 +36,19 @@ var readCmd = &cobra.Command{
 				return err
 			}
 
-			r, errCh, err := c.Read(ctx, args[0])
+			r, err := c.Read(ctx, args[0])
 			if err != nil {
 				return fmt.Errorf("error reading file: %w", err)
 			}
 
 			defer r.Close() //nolint:errcheck
 
-			var eg errgroup.Group
-
-			eg.Go(func() error {
-				var errors error
-
-				for err := range errCh {
-					if err != nil {
-						errors = helpers.AppendErrors(errors, err)
-					}
-				}
-
-				return errors
-			})
-
 			_, err = io.Copy(os.Stdout, r)
 			if err != nil {
 				return fmt.Errorf("error reading: %w", err)
 			}
 
-			if err = r.Close(); err != nil {
-				return err
-			}
-
-			return eg.Wait()
+			return r.Close()
 		})
 	},
 }

cmd/talosctl/pkg/talos/action/tracker.go

@@ -320,7 +320,7 @@ func (a *Tracker) processNodeUpdate(update nodeUpdate) reporter.Update {
 // getBootID reads the boot ID from the node.
 // It returns the node as the first return value and the boot ID as the second.
 func getBootID(ctx context.Context, c *client.Client) (string, error) {
-	reader, errCh, err := c.Read(ctx, "/proc/sys/kernel/random/boot_id")
+	reader, err := c.Read(ctx, "/proc/sys/kernel/random/boot_id")
 	if err != nil {
 		return "", err
 	}
@@ -334,11 +334,5 @@ func getBootID(ctx context.Context, c *client.Client) (string, error) {
 
 	bootID := strings.TrimSpace(string(body))
 
-	for err = range errCh {
-		if err != nil {
-			return "", err
-		}
-	}
-
 	return bootID, reader.Close()
 }

internal/integration/api/cgroups.go

@@ -128,7 +128,7 @@ func (suite *CGroupsSuite) TestCGroupsVersion() {
 
 //nolint:gocyclo
 func (suite *CGroupsSuite) readCmdline(ctx context.Context) (string, error) {
-	reader, errCh, err := suite.Client.Read(ctx, "/proc/cmdline")
+	reader, err := suite.Client.Read(ctx, "/proc/cmdline")
 	if err != nil {
 		return "", err
 	}
@@ -147,12 +147,6 @@ func (suite *CGroupsSuite) readCmdline(ctx context.Context) (string, error) {
 		return "", err
 	}
 
-	for err = range errCh {
-		if err != nil {
-			return "", err
-		}
-	}
-
 	return bootID, reader.Close()
 }
 

internal/integration/api/dmesg.go

@@ -51,14 +51,12 @@ func (suite *DmesgSuite) TestNodeHasDmesg() {
 	)
 	suite.Require().NoError(err)
 
-	logReader, errCh, err := client.ReadStream(dmesgStream)
+	logReader, err := client.ReadStream(dmesgStream)
 	suite.Require().NoError(err)
 
 	n, err := io.Copy(io.Discard, logReader)
 	suite.Require().NoError(err)
 
-	suite.Require().NoError(<-errCh)
-
 	// dmesg shouldn't be empty
 	suite.Require().Greater(n, int64(1024))
 }

internal/integration/api/etcd-recover.go

@@ -11,7 +11,6 @@ import (
 	"context"
 	"fmt"
 	"io"
-	"sync"
 	"testing"
 	"time"
 
@@ -184,27 +183,13 @@ func (suite *EtcdRecoverSuite) TestSnapshotRecover() {
 func (suite *EtcdRecoverSuite) snapshotEtcd(snapshotNode string, dest io.Writer) error {
 	ctx := client.WithNodes(suite.ctx, snapshotNode)
 
-	r, errCh, err := suite.Client.EtcdSnapshot(ctx, &machineapi.EtcdSnapshotRequest{})
+	r, err := suite.Client.EtcdSnapshot(ctx, &machineapi.EtcdSnapshotRequest{})
 	if err != nil {
 		return fmt.Errorf("error reading snapshot: %w", err)
 	}
 
 	defer r.Close() //nolint:errcheck
 
-	var wg sync.WaitGroup
-
-	wg.Add(1)
-
-	go func() {
-		defer wg.Done()
-
-		for err := range errCh {
-			suite.T().Logf("read error: %s", err)
-		}
-	}()
-
-	defer wg.Wait()
-
 	_, err = io.Copy(dest, r)
 
 	return err

internal/integration/api/logs.go

@@ -69,15 +69,13 @@ func (suite *LogsSuite) TestServicesHaveLogs() {
 		)
 		suite.Require().NoError(err)
 
-		logReader, errCh, err := client.ReadStream(logsStream)
+		logReader, err := client.ReadStream(logsStream)
 		suite.Require().NoError(err)
 
 		n, err := io.Copy(io.Discard, logReader)
 		suite.Require().NoError(err)
 
 		logsSize += n
-
-		suite.Require().NoError(<-errCh)
 	}
 
 	// overall logs shouldn't be empty
@@ -104,7 +102,7 @@ func (suite *LogsSuite) TestTail() {
 		)
 		suite.Require().NoError(err)
 
-		logReader, errCh, err := client.ReadStream(logsStream)
+		logReader, err := client.ReadStream(logsStream)
 		suite.Require().NoError(err)
 
 		scanner := bufio.NewScanner(logReader)
@@ -116,8 +114,6 @@ func (suite *LogsSuite) TestTail() {
 
 		suite.Require().NoError(scanner.Err())
 
-		suite.Require().NoError(<-errCh)
-
 		suite.Assert().EqualValues(tailLines, lines)
 	}
 }