feat: implement simplified client method to consume events

This provides interface to consume (watch) events in the same way (with
the same API) as server code does. So some processing might be moved
back and forth between client and server without much code change.

Also moved peer methods out of helpers to the main client package, they
might be useful for any client, not only talosctl.

Signed-off-by: Andrey Smirnov <smirnov.andrey@gmail.com>

cmd/talosctl/cmd/talos/containers.go

@@ -19,7 +19,6 @@ import (
 
 	"github.com/talos-systems/talos/api/common"
 	osapi "github.com/talos-systems/talos/api/os"
-	"github.com/talos-systems/talos/cmd/talosctl/pkg/talos/helpers"
 	"github.com/talos-systems/talos/pkg/cli"
 	"github.com/talos-systems/talos/pkg/client"
 	"github.com/talos-systems/talos/pkg/constants"
@@ -65,7 +64,7 @@ func containerRender(remotePeer *peer.Peer, resp *osapi.ContainersResponse) erro
 	w := tabwriter.NewWriter(os.Stdout, 0, 0, 3, ' ', 0)
 	fmt.Fprintln(w, "NODE\tNAMESPACE\tID\tIMAGE\tPID\tSTATUS")
 
-	defaultNode := helpers.AddrFromPeer(remotePeer)
+	defaultNode := client.AddrFromPeer(remotePeer)
 
 	for _, msg := range resp.Messages {
 		resp := msg

cmd/talosctl/cmd/talos/dmesg.go

@@ -14,7 +14,6 @@ import (
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 
-	"github.com/talos-systems/talos/cmd/talosctl/pkg/talos/helpers"
 	"github.com/talos-systems/talos/pkg/client"
 )
 
@@ -33,7 +32,7 @@ var dmesgCmd = &cobra.Command{
 				return fmt.Errorf("error getting dmesg: %w", err)
 			}
 
-			defaultNode := helpers.RemotePeer(stream.Context())
+			defaultNode := client.RemotePeer(stream.Context())
 
 			for {
 				resp, err := stream.Recv()

cmd/talosctl/cmd/talos/events.go

@@ -7,18 +7,12 @@ package talos
 import (
 	"context"
 	"fmt"
-	"io"
-	"log"
 	"os"
 	"text/tabwriter"
 
-	"github.com/golang/protobuf/proto"
 	"github.com/spf13/cobra"
-	"google.golang.org/grpc/codes"
-	"google.golang.org/grpc/status"
 
 	"github.com/talos-systems/talos/api/machine"
-	"github.com/talos-systems/talos/cmd/talosctl/pkg/talos/helpers"
 	"github.com/talos-systems/talos/pkg/client"
 )
 
@@ -29,47 +23,31 @@ var eventsCmd = &cobra.Command{
 	Long:  ``,
 	RunE: func(cmd *cobra.Command, args []string) error {
 		return WithClient(func(ctx context.Context, c *client.Client) error {
-			stream, err := c.Events(ctx)
-			if err != nil {
-				return fmt.Errorf("error fetching events: %s", err)
-			}
-
-			defaultNode := helpers.RemotePeer(stream.Context())
-
 			w := tabwriter.NewWriter(os.Stdout, 0, 0, 3, ' ', 0)
 			fmt.Fprintln(w, "NODE\tEVENT\tMESSAGE")
 
+			return c.EventsWatch(ctx, func(ch <-chan client.Event) {
 				for {
-				event, err := stream.Recv()
+					var (
-				if err != nil {
+						event client.Event
-					if err == io.EOF || status.Code(err) == codes.Canceled {
+						ok    bool
-						return nil
+					)
+
+					select {
+					case event, ok = <-ch:
+						if !ok {
+							return
 						}
-
+					case <-ctx.Done():
-					return fmt.Errorf("failed to watch events: %w", err)
+						return
 					}
 
-				node := defaultNode
-
-				if event.Metadata != nil {
-					node = event.Metadata.Hostname
-				}
-
-				typeURL := event.GetData().GetTypeUrl()
-
 					format := "%s\t%s\t%s\n"
 
 					var args []interface{}
 
-				switch event.GetData().GetTypeUrl() {
+					switch msg := event.Payload.(type) {
-				case "talos/runtime/" + proto.MessageName(&machine.SequenceEvent{}):
+					case *machine.SequenceEvent:
-					msg := &machine.SequenceEvent{}
-
-					if err = proto.Unmarshal(event.GetData().GetValue(), msg); err != nil {
-						log.Printf("failed to unmarshal message: %v", err)
-						continue
-					}
-
 						if msg.Error != nil {
 							args = []interface{}{msg.GetSequence() + " error:" + " " + msg.GetError().GetMessage()}
 						} else {
@@ -80,13 +58,14 @@ var eventsCmd = &cobra.Command{
 						continue
 					}
 
-				args = append([]interface{}{node, typeURL}, args...)
+					args = append([]interface{}{event.Node, event.TypeURL}, args...)
 					fmt.Fprintf(w, format, args...)
 
 					// nolint: errcheck
 					w.Flush()
 				}
 			})
+		})
 	},
 }
 

cmd/talosctl/cmd/talos/interfaces.go

@@ -15,7 +15,6 @@ import (
 	"google.golang.org/grpc/peer"
 
 	networkapi "github.com/talos-systems/talos/api/network"
-	"github.com/talos-systems/talos/cmd/talosctl/pkg/talos/helpers"
 	"github.com/talos-systems/talos/pkg/cli"
 	"github.com/talos-systems/talos/pkg/client"
 )
@@ -48,7 +47,7 @@ func intersRender(remotePeer *peer.Peer, resp *networkapi.InterfacesResponse) er
 	w := tabwriter.NewWriter(os.Stdout, 0, 0, 3, ' ', 0)
 	fmt.Fprintln(w, "NODE\tINDEX\tNAME\tMAC\tMTU\tADDRESS")
 
-	defaultNode := helpers.AddrFromPeer(remotePeer)
+	defaultNode := client.AddrFromPeer(remotePeer)
 
 	for _, msg := range resp.Messages {
 		node := defaultNode

cmd/talosctl/cmd/talos/list.go

@@ -18,7 +18,6 @@ import (
 	"google.golang.org/grpc/status"
 
 	machineapi "github.com/talos-systems/talos/api/machine"
-	"github.com/talos-systems/talos/cmd/talosctl/pkg/talos/helpers"
 	"github.com/talos-systems/talos/pkg/client"
 )
 
@@ -55,7 +54,7 @@ var lsCmd = &cobra.Command{
 				return fmt.Errorf("error fetching logs: %s", err)
 			}
 
-			defaultNode := helpers.RemotePeer(stream.Context())
+			defaultNode := client.RemotePeer(stream.Context())
 
 			if !long {
 				w := tabwriter.NewWriter(os.Stdout, 0, 0, 3, ' ', 0)

cmd/talosctl/cmd/talos/logs.go

@@ -19,7 +19,6 @@ import (
 
 	"github.com/talos-systems/talos/api/common"
 	"github.com/talos-systems/talos/api/machine"
-	"github.com/talos-systems/talos/cmd/talosctl/pkg/talos/helpers"
 	"github.com/talos-systems/talos/pkg/cli"
 	"github.com/talos-systems/talos/pkg/client"
 	"github.com/talos-systems/talos/pkg/constants"
@@ -55,7 +54,7 @@ var logsCmd = &cobra.Command{
 				return fmt.Errorf("error fetching logs: %s", err)
 			}
 
-			defaultNode := helpers.RemotePeer(stream.Context())
+			defaultNode := client.RemotePeer(stream.Context())
 
 			respCh, errCh := newLineSlicer(stream)
 

cmd/talosctl/cmd/talos/memory.go

@@ -15,7 +15,6 @@ import (
 	"google.golang.org/grpc/peer"
 
 	osapi "github.com/talos-systems/talos/api/os"
-	"github.com/talos-systems/talos/cmd/talosctl/pkg/talos/helpers"
 	"github.com/talos-systems/talos/pkg/cli"
 	"github.com/talos-systems/talos/pkg/client"
 )
@@ -55,7 +54,7 @@ func briefRender(remotePeer *peer.Peer, resp *osapi.MemoryResponse) error {
 	w := tabwriter.NewWriter(os.Stdout, 0, 0, 3, ' ', 0)
 	fmt.Fprintln(w, "NODE\tTOTAL\tUSED\tFREE\tSHARED\tBUFFERS\tCACHE\tAVAILABLE")
 
-	defaultNode := helpers.AddrFromPeer(remotePeer)
+	defaultNode := client.AddrFromPeer(remotePeer)
 
 	for _, msg := range resp.Messages {
 		node := defaultNode
@@ -81,7 +80,7 @@ func briefRender(remotePeer *peer.Peer, resp *osapi.MemoryResponse) error {
 }
 
 func verboseRender(remotePeer *peer.Peer, resp *osapi.MemoryResponse) error {
-	defaultNode := helpers.AddrFromPeer(remotePeer)
+	defaultNode := client.AddrFromPeer(remotePeer)
 
 	// Dump as /proc/meminfo
 	for _, msg := range resp.Messages {

cmd/talosctl/cmd/talos/mounts.go

@@ -16,7 +16,6 @@ import (
 	"google.golang.org/grpc/peer"
 
 	machineapi "github.com/talos-systems/talos/api/machine"
-	"github.com/talos-systems/talos/cmd/talosctl/pkg/talos/helpers"
 	"github.com/talos-systems/talos/pkg/cli"
 	"github.com/talos-systems/talos/pkg/client"
 )
@@ -50,7 +49,7 @@ func mountsRender(remotePeer *peer.Peer, resp *machineapi.MountsResponse) error
 	w := tabwriter.NewWriter(os.Stdout, 0, 0, 3, ' ', 0)
 	fmt.Fprintln(w, "NODE\tFILESYSTEM\tSIZE(GB)\tUSED(GB)\tAVAILABLE(GB)\tPERCENT USED\tMOUNTED ON")
 
-	defaultNode := helpers.AddrFromPeer(remotePeer)
+	defaultNode := client.AddrFromPeer(remotePeer)
 
 	for _, msg := range resp.Messages {
 		for _, r := range msg.Stats {

cmd/talosctl/cmd/talos/processes.go

@@ -22,7 +22,6 @@ import (
 	"google.golang.org/grpc/peer"
 
 	osapi "github.com/talos-systems/talos/api/os"
-	"github.com/talos-systems/talos/cmd/talosctl/pkg/talos/helpers"
 	"github.com/talos-systems/talos/pkg/cli"
 	"github.com/talos-systems/talos/pkg/client"
 )
@@ -184,7 +183,7 @@ func processesOutput(ctx context.Context, c *client.Client) (output string, err
 		return output, nil
 	}
 
-	defaultNode := helpers.AddrFromPeer(&remotePeer)
+	defaultNode := client.AddrFromPeer(&remotePeer)
 
 	s := []string{}
 

cmd/talosctl/cmd/talos/routes.go

@@ -15,7 +15,6 @@ import (
 	"google.golang.org/grpc/peer"
 
 	networkapi "github.com/talos-systems/talos/api/network"
-	"github.com/talos-systems/talos/cmd/talosctl/pkg/talos/helpers"
 	"github.com/talos-systems/talos/pkg/cli"
 	"github.com/talos-systems/talos/pkg/client"
 )
@@ -46,7 +45,7 @@ func routesRender(remotePeer *peer.Peer, resp *networkapi.RoutesResponse) error
 	w := tabwriter.NewWriter(os.Stdout, 0, 0, 3, ' ', 0)
 	fmt.Fprintln(w, "NODE\tINTERFACE\tDESTINATION\tGATEWAY\tMETRIC")
 
-	defaultNode := helpers.AddrFromPeer(remotePeer)
+	defaultNode := client.AddrFromPeer(remotePeer)
 
 	for _, msg := range resp.Messages {
 		node := defaultNode

cmd/talosctl/cmd/talos/service.go

@@ -17,7 +17,6 @@ import (
 	"google.golang.org/grpc/peer"
 
 	machineapi "github.com/talos-systems/talos/api/machine"
-	"github.com/talos-systems/talos/cmd/talosctl/pkg/talos/helpers"
 	"github.com/talos-systems/talos/pkg/cli"
 	"github.com/talos-systems/talos/pkg/client"
 )
@@ -79,7 +78,7 @@ func serviceList(ctx context.Context, c *client.Client) error {
 	w := tabwriter.NewWriter(os.Stdout, 0, 0, 3, ' ', 0)
 	fmt.Fprintln(w, "NODE\tSERVICE\tSTATE\tHEALTH\tLAST CHANGE\tLAST EVENT")
 
-	defaultNode := helpers.AddrFromPeer(&remotePeer)
+	defaultNode := client.AddrFromPeer(&remotePeer)
 
 	for _, msg := range resp.Messages {
 		for _, s := range msg.Services {
@@ -112,7 +111,7 @@ func serviceInfo(ctx context.Context, c *client.Client, id string) error {
 
 	w := tabwriter.NewWriter(os.Stdout, 0, 0, 3, ' ', 0)
 
-	defaultNode := helpers.AddrFromPeer(&remotePeer)
+	defaultNode := client.AddrFromPeer(&remotePeer)
 
 	for _, s := range services {
 		node := defaultNode
@@ -163,7 +162,7 @@ func serviceStart(ctx context.Context, c *client.Client, id string) error {
 		cli.Warning("%s", err)
 	}
 
-	defaultNode := helpers.AddrFromPeer(&remotePeer)
+	defaultNode := client.AddrFromPeer(&remotePeer)
 
 	w := tabwriter.NewWriter(os.Stdout, 0, 0, 3, ' ', 0)
 	fmt.Fprintln(w, "NODE\tRESPONSE")
@@ -193,7 +192,7 @@ func serviceStop(ctx context.Context, c *client.Client, id string) error {
 		cli.Warning("%s", err)
 	}
 
-	defaultNode := helpers.AddrFromPeer(&remotePeer)
+	defaultNode := client.AddrFromPeer(&remotePeer)
 
 	w := tabwriter.NewWriter(os.Stdout, 0, 0, 3, ' ', 0)
 	fmt.Fprintln(w, "NODE\tRESPONSE")
@@ -223,7 +222,7 @@ func serviceRestart(ctx context.Context, c *client.Client, id string) error {
 		cli.Warning("%s", err)
 	}
 
-	defaultNode := helpers.AddrFromPeer(&remotePeer)
+	defaultNode := client.AddrFromPeer(&remotePeer)
 
 	w := tabwriter.NewWriter(os.Stdout, 0, 0, 3, ' ', 0)
 	fmt.Fprintln(w, "NODE\tRESPONSE")

cmd/talosctl/cmd/talos/stats.go

@@ -19,7 +19,6 @@ import (
 
 	"github.com/talos-systems/talos/api/common"
 	osapi "github.com/talos-systems/talos/api/os"
-	"github.com/talos-systems/talos/cmd/talosctl/pkg/talos/helpers"
 	"github.com/talos-systems/talos/pkg/cli"
 	"github.com/talos-systems/talos/pkg/client"
 	"github.com/talos-systems/talos/pkg/constants"
@@ -65,7 +64,7 @@ func statsRender(remotePeer *peer.Peer, resp *osapi.StatsResponse) error {
 
 	fmt.Fprintln(w, "NODE\tNAMESPACE\tID\tMEMORY(MB)\tCPU")
 
-	defaultNode := helpers.AddrFromPeer(remotePeer)
+	defaultNode := client.AddrFromPeer(remotePeer)
 
 	for _, msg := range resp.Messages {
 		resp := msg

cmd/talosctl/cmd/talos/time.go

@@ -17,7 +17,6 @@ import (
 	"google.golang.org/grpc/peer"
 
 	timeapi "github.com/talos-systems/talos/api/time"
-	"github.com/talos-systems/talos/cmd/talosctl/pkg/talos/helpers"
 	"github.com/talos-systems/talos/pkg/cli"
 	"github.com/talos-systems/talos/pkg/client"
 )
@@ -57,7 +56,7 @@ var timeCmd = &cobra.Command{
 			w := tabwriter.NewWriter(os.Stdout, 0, 0, 3, ' ', 0)
 			fmt.Fprintln(w, "NODE\tNTP-SERVER\tLOCAL-TIME\tREMOTE-TIME")
 
-			defaultNode := helpers.AddrFromPeer(&remotePeer)
+			defaultNode := client.AddrFromPeer(&remotePeer)
 
 			var localtime, remotetime time.Time
 			for _, msg := range resp.Messages {

cmd/talosctl/cmd/talos/upgrade.go

@@ -15,7 +15,6 @@ import (
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/peer"
 
-	"github.com/talos-systems/talos/cmd/talosctl/pkg/talos/helpers"
 	"github.com/talos-systems/talos/pkg/cli"
 	"github.com/talos-systems/talos/pkg/client"
 )
@@ -60,7 +59,7 @@ func upgrade() error {
 		w := tabwriter.NewWriter(os.Stdout, 0, 0, 3, ' ', 0)
 		fmt.Fprintln(w, "NODE\tACK\tSTARTED")
 
-		defaultNode := helpers.AddrFromPeer(&remotePeer)
+		defaultNode := client.AddrFromPeer(&remotePeer)
 
 		for _, msg := range resp.Messages {
 			node := defaultNode

cmd/talosctl/cmd/talos/version.go

@@ -12,7 +12,6 @@ import (
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/peer"
 
-	"github.com/talos-systems/talos/cmd/talosctl/pkg/talos/helpers"
 	"github.com/talos-systems/talos/pkg/cli"
 	"github.com/talos-systems/talos/pkg/client"
 	"github.com/talos-systems/talos/pkg/version"
@@ -54,7 +53,7 @@ var versionCmd = &cobra.Command{
 				cli.Warning("%s", err)
 			}
 
-			defaultNode := helpers.AddrFromPeer(&remotePeer)
+			defaultNode := client.AddrFromPeer(&remotePeer)
 
 			for _, msg := range resp.Messages {
 				node := defaultNode

internal/app/machined/internal/server/v1alpha1/v1alpha1_server.go

@@ -689,7 +689,11 @@ func (s *Server) Events(req *machine.EventsRequest, l machine.MachineService_Eve
 				select {
 				case <-l.Context().Done():
 					return l.Context().Err()
-				case event := <-events:
+				case event, ok := <-events:
+					if !ok {
+						return nil
+					}
+
 					msg, err := event.ToMachineEvent()
 					if err != nil {
 						return err

pkg/client/client.go

@@ -14,8 +14,11 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"log"
 	"strings"
+	"sync"
 
+	"github.com/golang/protobuf/proto"
 	"github.com/golang/protobuf/ptypes/empty"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
@@ -743,3 +746,81 @@ func ReadStream(stream MachineStream) (io.ReadCloser, <-chan error, error) {
 func (c *Client) Events(ctx context.Context) (stream machineapi.MachineService_EventsClient, err error) {
 	return c.MachineClient.Events(ctx, &machineapi.EventsRequest{})
 }
+
+// Event as received from the API.
+type Event struct {
+	Node    string
+	TypeURL string
+	Payload proto.Message
+}
+
+// EventsWatch wraps Events by providing more simple interface.
+//
+//nolint: gocyclo
+func (c *Client) EventsWatch(ctx context.Context, watchFunc func(<-chan Event)) error {
+	stream, err := c.Events(ctx)
+	if err != nil {
+		return fmt.Errorf("error fetching events: %s", err)
+	}
+
+	defaultNode := RemotePeer(stream.Context())
+
+	var wg sync.WaitGroup
+
+	defer wg.Wait()
+
+	ch := make(chan Event)
+	defer close(ch)
+
+	wg.Add(1)
+
+	go func() {
+		defer wg.Done()
+
+		watchFunc(ch)
+	}()
+
+	for {
+		event, err := stream.Recv()
+		if err != nil {
+			if err == io.EOF || status.Code(err) == codes.Canceled {
+				return nil
+			}
+
+			return fmt.Errorf("failed to watch events: %w", err)
+		}
+
+		typeURL := event.GetData().GetTypeUrl()
+
+		var msg proto.Message
+
+		switch typeURL {
+		case "talos/runtime/" + proto.MessageName(&machineapi.SequenceEvent{}):
+			msg = &machineapi.SequenceEvent{}
+
+			if err = proto.Unmarshal(event.GetData().GetValue(), msg); err != nil {
+				log.Printf("failed to unmarshal message: %v", err) // TODO: this should be fixed to return errors
+				continue
+			}
+		default:
+			// We haven't implemented the handling of this event yet.
+			continue
+		}
+
+		ev := Event{
+			Node:    defaultNode,
+			TypeURL: typeURL,
+			Payload: msg,
+		}
+
+		if event.Metadata != nil {
+			ev.Node = event.Metadata.Hostname
+		}
+
+		select {
+		case ch <- ev:
+		case <-ctx.Done():
+			return nil
+		}
+	}
+}

pkg/client/peer.go

@@ -2,7 +2,7 @@
 // License, v. 2.0. If a copy of the MPL was not distributed with this
 // file, You can obtain one at http://mozilla.org/MPL/2.0/.
 
-package helpers
+package client
 
 import (
 	"context"