node: add "admin list-nodes" command

Change-Id: Idbcfb449be300e0dc7f4a655b3f1fc455e4d8158

bridge/cmd/guardiand/adminclient.go

@@ -3,6 +3,7 @@ package guardiand
 import (
 	"context"
 	"fmt"
+	publicrpcv1 "github.com/certusone/wormhole/bridge/pkg/proto/publicrpc/v1"
 	"io/ioutil"
 	"log"
 	"time"
@@ -24,9 +25,16 @@ func init() {
 	if err != nil {
 		panic(err)
 	}
+	pf = AdminClientListNodes.Flags()
+	clientSocketPath = pf.String("socket", "", "gRPC admin server socket to connect to")
+	err = cobra.MarkFlagRequired(pf, "socket")
+	if err != nil {
+		panic(err)
+	}
 
 	AdminCmd.AddCommand(AdminClientInjectGuardianSetUpdateCmd)
 	AdminCmd.AddCommand(AdminClientGovernanceVAAVerifyCmd)
+	AdminCmd.AddCommand(AdminClientListNodes)
 }
 
 var AdminCmd = &cobra.Command{
@@ -52,6 +60,17 @@ func getAdminClient(ctx context.Context, addr string) (*grpc.ClientConn, error,
 	return conn, err, c
 }
 
+func getPublicrpcClient(ctx context.Context, addr string) (*grpc.ClientConn, error, publicrpcv1.PublicrpcClient) {
+	conn, err := grpc.DialContext(ctx, fmt.Sprintf("unix:///%s", addr), grpc.WithInsecure())
+
+	if err != nil {
+		log.Fatalf("failed to connect to %s: %v", addr, err)
+	}
+
+	c := publicrpcv1.NewPublicrpcClient(conn)
+	return conn, err, c
+}
+
 func runInjectGovernanceVAA(cmd *cobra.Command, args []string) {
 	path := args[0]
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
@@ -59,6 +78,9 @@ func runInjectGovernanceVAA(cmd *cobra.Command, args []string) {
 
 	conn, err, c := getAdminClient(ctx, *clientSocketPath)
 	defer conn.Close()
+	if err != nil {
+		log.Fatalf("failed to get admin client: %v", err)
+	}
 
 	b, err := ioutil.ReadFile(path)
 	if err != nil {

bridge/cmd/guardiand/adminnodes.go

@@ -0,0 +1,57 @@
+package guardiand
+
+import (
+	"context"
+	"fmt"
+	publicrpcv1 "github.com/certusone/wormhole/bridge/pkg/proto/publicrpc/v1"
+	"github.com/spf13/cobra"
+	"io"
+	"log"
+	"os"
+	"text/tabwriter"
+)
+
+// How to test in container:
+//    kubectl exec guardian-0 -- /guardiand admin list-nodes --socket /tmp/admin.sock
+
+var AdminClientListNodes = &cobra.Command{
+	Use:   "list-nodes",
+	Short: "Listens to heartbeats and displays an aggregated list of nodes (UNTRUSTED)",
+	Run:   runListNodes,
+}
+
+func runListNodes(cmd *cobra.Command, args []string) {
+	ctx := context.Background()
+	conn, err, c := getPublicrpcClient(ctx, *clientSocketPath)
+	defer conn.Close()
+	if err != nil {
+		log.Fatalf("failed to get publicrpc client: %v", err)
+	}
+
+	stream, err := c.GetRawHeartbeats(ctx, &publicrpcv1.GetRawHeartbeatsRequest{})
+	if err != nil {
+		log.Fatalf("failed to stream heartbeats: %v", err)
+	}
+
+	log.Print("connected, streaming updates")
+
+	seen := make(map[string]bool)
+	w := tabwriter.NewWriter(os.Stdout, 20, 8, 1, '\t', 0)
+	for {
+		hb, err := stream.Recv()
+		if err == io.EOF {
+			log.Print("server closed connection, exiting")
+			return
+		} else if err != nil {
+			log.Fatalf("error streaming updates: %v", err)
+		}
+
+		if seen[hb.GuardianAddr] {
+			continue
+		}
+
+		fmt.Fprintf(w, "%s\t%s\t%s\t\n", hb.GuardianAddr, hb.NodeName, hb.Version)
+		w.Flush()
+		seen[hb.GuardianAddr] = true
+	}
+}