From b9b9aa9104c19815929b64a77a6a522f8f46baf4 Mon Sep 17 00:00:00 2001
From: Andrey Smirnov <smirnov.andrey@gmail.com>
Date: Wed, 20 May 2020 16:11:45 +0300
Subject: [PATCH] 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          |  3 +-
 cmd/talosctl/cmd/talos/dmesg.go               |  3 +-
 cmd/talosctl/cmd/talos/events.go              | 83 +++++++------------
 cmd/talosctl/cmd/talos/interfaces.go          |  3 +-
 cmd/talosctl/cmd/talos/list.go                |  3 +-
 cmd/talosctl/cmd/talos/logs.go                |  3 +-
 cmd/talosctl/cmd/talos/memory.go              |  5 +-
 cmd/talosctl/cmd/talos/mounts.go              |  3 +-
 cmd/talosctl/cmd/talos/processes.go           |  3 +-
 cmd/talosctl/cmd/talos/routes.go              |  3 +-
 cmd/talosctl/cmd/talos/service.go             | 11 ++-
 cmd/talosctl/cmd/talos/stats.go               |  3 +-
 cmd/talosctl/cmd/talos/time.go                |  3 +-
 cmd/talosctl/cmd/talos/upgrade.go             |  3 +-
 cmd/talosctl/cmd/talos/version.go             |  3 +-
 .../server/v1alpha1/v1alpha1_server.go        |  6 +-
 pkg/client/client.go                          | 81 ++++++++++++++++++
 .../pkg/talos/helpers => pkg/client}/peer.go  |  2 +-
 18 files changed, 137 insertions(+), 87 deletions(-)
 rename {cmd/talosctl/pkg/talos/helpers => pkg/client}/peer.go (97%)

diff --git a/cmd/talosctl/cmd/talos/containers.go b/cmd/talosctl/cmd/talos/containers.go
index 2d6b4ff44..686b30892 100644
--- a/cmd/talosctl/cmd/talos/containers.go
+++ b/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
diff --git a/cmd/talosctl/cmd/talos/dmesg.go b/cmd/talosctl/cmd/talos/dmesg.go
index 9c4caa102..7101a204f 100644
--- a/cmd/talosctl/cmd/talos/dmesg.go
+++ b/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()
diff --git a/cmd/talosctl/cmd/talos/events.go b/cmd/talosctl/cmd/talos/events.go
index 536a601f9..0df7886e7 100644
--- a/cmd/talosctl/cmd/talos/events.go
+++ b/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,63 +23,48 @@ 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")
 
-			for {
-				event, err := stream.Recv()
-				if err != nil {
-					if err == io.EOF || status.Code(err) == codes.Canceled {
-						return nil
+			return c.EventsWatch(ctx, func(ch <-chan client.Event) {
+				for {
+					var (
+						event client.Event
+						ok    bool
+					)
+
+					select {
+					case event, ok = <-ch:
+						if !ok {
+							return
+						}
+					case <-ctx.Done():
+						return
 					}
 
-					return fmt.Errorf("failed to watch events: %w", err)
-				}
+					format := "%s\t%s\t%s\n"
 
-				node := defaultNode
+					var args []interface{}
 
-				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() {
-				case "talos/runtime/" + proto.MessageName(&machine.SequenceEvent{}):
-					msg := &machine.SequenceEvent{}
-
-					if err = proto.Unmarshal(event.GetData().GetValue(), msg); err != nil {
-						log.Printf("failed to unmarshal message: %v", err)
+					switch msg := event.Payload.(type) {
+					case *machine.SequenceEvent:
+						if msg.Error != nil {
+							args = []interface{}{msg.GetSequence() + " error:" + " " + msg.GetError().GetMessage()}
+						} else {
+							args = []interface{}{msg.GetSequence() + " " + msg.GetAction().String()}
+						}
+					default:
+						// We haven't implemented the handling of this event yet.
 						continue
 					}
 
-					if msg.Error != nil {
-						args = []interface{}{msg.GetSequence() + " error:" + " " + msg.GetError().GetMessage()}
-					} else {
-						args = []interface{}{msg.GetSequence() + " " + msg.GetAction().String()}
-					}
-				default:
-					// We haven't implemented the handling of this event yet.
-					continue
+					args = append([]interface{}{event.Node, event.TypeURL}, args...)
+					fmt.Fprintf(w, format, args...)
+
+					// nolint: errcheck
+					w.Flush()
 				}
-
-				args = append([]interface{}{node, typeURL}, args...)
-				fmt.Fprintf(w, format, args...)
-
-				// nolint: errcheck
-				w.Flush()
-			}
+			})
 		})
 	},
 }
diff --git a/cmd/talosctl/cmd/talos/interfaces.go b/cmd/talosctl/cmd/talos/interfaces.go
index 6fbf23a07..0c2a94ae6 100644
--- a/cmd/talosctl/cmd/talos/interfaces.go
+++ b/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
diff --git a/cmd/talosctl/cmd/talos/list.go b/cmd/talosctl/cmd/talos/list.go
index de85baccd..798bc0e33 100644
--- a/cmd/talosctl/cmd/talos/list.go
+++ b/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)
diff --git a/cmd/talosctl/cmd/talos/logs.go b/cmd/talosctl/cmd/talos/logs.go
index 2f7d35eb6..4b6e6d9bf 100644
--- a/cmd/talosctl/cmd/talos/logs.go
+++ b/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)
 
diff --git a/cmd/talosctl/cmd/talos/memory.go b/cmd/talosctl/cmd/talos/memory.go
index e3739ef8c..3a45d5d4f 100644
--- a/cmd/talosctl/cmd/talos/memory.go
+++ b/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 {
diff --git a/cmd/talosctl/cmd/talos/mounts.go b/cmd/talosctl/cmd/talos/mounts.go
index 87daeb54a..9449446f4 100644
--- a/cmd/talosctl/cmd/talos/mounts.go
+++ b/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 {
diff --git a/cmd/talosctl/cmd/talos/processes.go b/cmd/talosctl/cmd/talos/processes.go
index 6fd016f09..fd4db9e48 100644
--- a/cmd/talosctl/cmd/talos/processes.go
+++ b/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{}
 
diff --git a/cmd/talosctl/cmd/talos/routes.go b/cmd/talosctl/cmd/talos/routes.go
index 36fd7d0b9..000030f2c 100644
--- a/cmd/talosctl/cmd/talos/routes.go
+++ b/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
diff --git a/cmd/talosctl/cmd/talos/service.go b/cmd/talosctl/cmd/talos/service.go
index 9a7d97358..f110bab5b 100644
--- a/cmd/talosctl/cmd/talos/service.go
+++ b/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")
diff --git a/cmd/talosctl/cmd/talos/stats.go b/cmd/talosctl/cmd/talos/stats.go
index ea97fd01c..5fe12f545 100644
--- a/cmd/talosctl/cmd/talos/stats.go
+++ b/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
diff --git a/cmd/talosctl/cmd/talos/time.go b/cmd/talosctl/cmd/talos/time.go
index 74198b801..fbf7061f1 100644
--- a/cmd/talosctl/cmd/talos/time.go
+++ b/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 {
diff --git a/cmd/talosctl/cmd/talos/upgrade.go b/cmd/talosctl/cmd/talos/upgrade.go
index b241af963..f4f72610b 100644
--- a/cmd/talosctl/cmd/talos/upgrade.go
+++ b/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
diff --git a/cmd/talosctl/cmd/talos/version.go b/cmd/talosctl/cmd/talos/version.go
index 976797c3c..61772b1f9 100644
--- a/cmd/talosctl/cmd/talos/version.go
+++ b/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
diff --git a/internal/app/machined/internal/server/v1alpha1/v1alpha1_server.go b/internal/app/machined/internal/server/v1alpha1/v1alpha1_server.go
index ec3e8cff3..8e25c6a0e 100644
--- a/internal/app/machined/internal/server/v1alpha1/v1alpha1_server.go
+++ b/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
diff --git a/pkg/client/client.go b/pkg/client/client.go
index d4f10c3bb..3138b8923 100644
--- a/pkg/client/client.go
+++ b/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
+		}
+	}
+}
diff --git a/cmd/talosctl/pkg/talos/helpers/peer.go b/pkg/client/peer.go
similarity index 97%
rename from cmd/talosctl/pkg/talos/helpers/peer.go
rename to pkg/client/peer.go
index eeedc98f8..9b58c92f0 100644
--- a/cmd/talosctl/pkg/talos/helpers/peer.go
+++ b/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"