Xw/net 6307 grpc client apply (#20107)

* add acl token

* apply commmand works

* add info to error message

* add tests to apply method

* add more description to flag

* code refactor

* update the error message

* fix the failing test

command/registry.go

@@ -115,6 +115,7 @@ import (
 	"github.com/hashicorp/consul/command/reload"
 	"github.com/hashicorp/consul/command/resource"
 	resourceapply "github.com/hashicorp/consul/command/resource/apply"
+	resourceapplygrpc "github.com/hashicorp/consul/command/resource/apply-grpc"
 	resourcedelete "github.com/hashicorp/consul/command/resource/delete"
 	resourcelist "github.com/hashicorp/consul/command/resource/list"
 	resourceread "github.com/hashicorp/consul/command/resource/read"
@@ -258,6 +259,8 @@ func RegisteredCommands(ui cli.Ui) map[string]mcli.CommandFactory {
 		entry{"resource read", func(ui cli.Ui) (cli.Command, error) { return resourceread.New(ui), nil }},
 		entry{"resource delete", func(ui cli.Ui) (cli.Command, error) { return resourcedelete.New(ui), nil }},
 		entry{"resource apply", func(ui cli.Ui) (cli.Command, error) { return resourceapply.New(ui), nil }},
+		// will be refactored to resource apply
+		entry{"resource apply-grpc", func(ui cli.Ui) (cli.Command, error) { return resourceapplygrpc.New(ui), nil }},
 		entry{"resource list", func(ui cli.Ui) (cli.Command, error) { return resourcelist.New(ui), nil }},
 		entry{"rtt", func(ui cli.Ui) (cli.Command, error) { return rtt.New(ui), nil }},
 		entry{"services", func(cli.Ui) (cli.Command, error) { return services.New(), nil }},

command/resource/apply-grpc/apply.go

@@ -0,0 +1,155 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: BUSL-1.1
+
+package apply
+
+import (
+	"encoding/json"
+	"errors"
+	"flag"
+	"fmt"
+	"io"
+
+	"github.com/mitchellh/cli"
+
+	"github.com/hashicorp/consul/command/resource"
+	"github.com/hashicorp/consul/command/resource/client"
+)
+
+func New(ui cli.Ui) *cmd {
+	c := &cmd{UI: ui}
+	c.init()
+	return c
+}
+
+type cmd struct {
+	UI        cli.Ui
+	flags     *flag.FlagSet
+	grpcFlags *client.GRPCFlags
+	help      string
+
+	filePath string
+
+	testStdin io.Reader
+}
+
+func (c *cmd) init() {
+	c.flags = flag.NewFlagSet("", flag.ContinueOnError)
+	c.flags.StringVar(&c.filePath, "f", "",
+		"File path with resource definition")
+
+	c.grpcFlags = &client.GRPCFlags{}
+	client.MergeFlags(c.flags, c.grpcFlags.ClientFlags())
+	c.help = client.Usage(help, c.flags)
+}
+
+func (c *cmd) Run(args []string) int {
+	if err := c.flags.Parse(args); err != nil {
+		if !errors.Is(err, flag.ErrHelp) {
+			c.UI.Error(fmt.Sprintf("Failed to parse args: %v", err))
+			return 1
+		}
+		c.UI.Error(fmt.Sprintf("Failed to run apply command: %v", err))
+		return 1
+	}
+
+	// parse resource
+	input := c.filePath
+	if input == "" {
+		c.UI.Error("Required '-f' flag was not provided to specify where to load the resource content from")
+		return 1
+	}
+	parsedResource, err := resource.ParseResourceInput(input, c.testStdin)
+	if err != nil {
+		c.UI.Error(fmt.Sprintf("Failed to decode resource from input file: %v", err))
+		return 1
+	}
+	if parsedResource == nil {
+		c.UI.Error("Unable to parse the file argument")
+		return 1
+	}
+
+	// initialize client
+	config, err := client.LoadGRPCConfig(nil)
+	if err != nil {
+		c.UI.Error(fmt.Sprintf("Error loading config: %s", err))
+		return 1
+	}
+	c.grpcFlags.MergeFlagsIntoGRPCConfig(config)
+	resourceClient, err := client.NewGRPCClient(config)
+	if err != nil {
+		c.UI.Error(fmt.Sprintf("Error connect to Consul agent: %s", err))
+		return 1
+	}
+
+	// write resource
+	gvk := &resource.GVK{
+		Group:   parsedResource.Id.Type.GetGroup(),
+		Version: parsedResource.Id.Type.GetGroupVersion(),
+		Kind:    parsedResource.Id.Type.GetKind(),
+	}
+	res := resource.ResourceGRPC{C: resourceClient}
+	entry, err := res.Apply(parsedResource)
+	if err != nil {
+		c.UI.Error(fmt.Sprintf("Error writing resource %s/%s: %v", gvk, parsedResource.Id.GetName(), err))
+		return 1
+	}
+
+	// display response
+	b, err := json.MarshalIndent(entry, "", "    ")
+	if err != nil {
+		c.UI.Error("Failed to encode output data")
+		return 1
+	}
+	c.UI.Info(fmt.Sprintf("%s.%s.%s '%s' created.", gvk.Group, gvk.Version, gvk.Kind, parsedResource.Id.GetName()))
+	c.UI.Info(string(b))
+
+	return 0
+}
+
+func (c *cmd) Synopsis() string {
+	return synopsis
+}
+
+func (c *cmd) Help() string {
+	return client.Usage(c.help, nil)
+}
+
+const synopsis = "Writes/updates resource information"
+
+const help = `
+Usage: consul resource apply [options] <resource>
+
+	Write and/or update a resource by providing the definition. The configuration
+	argument is either a file path or '-' to indicate that the resource
+    should be read from stdin. The data should be either in HCL or
+	JSON form.
+
+	Example (with flag):
+
+	$ consul resource apply -f=demo.hcl
+
+	Example (from stdin):
+
+	$ consul resource apply -f - < demo.hcl
+
+	Sample demo.hcl:
+
+	ID {
+		Type = gvk("group.version.kind")
+		Name = "resource-name"
+		Tenancy {
+		Namespace = "default"
+		Partition = "default"
+		PeerName = "local"
+		}
+	}
+
+	Data {
+		Name = "demo"
+	}
+
+	Metadata = {
+		"foo" = "bar"
+	}
+`

command/resource/apply-grpc/apply_test.go

@@ -0,0 +1,228 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: BUSL-1.1
+
+package apply
+
+import (
+	"errors"
+	"fmt"
+	"io"
+	"testing"
+
+	"github.com/mitchellh/cli"
+	"github.com/stretchr/testify/require"
+
+	"github.com/hashicorp/consul/agent"
+	"github.com/hashicorp/consul/sdk/freeport"
+	"github.com/hashicorp/consul/testrpc"
+)
+
+func TestResourceApplyCommand(t *testing.T) {
+	if testing.Short() {
+		t.Skip("too slow for testing.Short")
+	}
+
+	t.Parallel()
+	availablePort := freeport.GetOne(t)
+	a := agent.NewTestAgent(t, fmt.Sprintf("ports { grpc = %d }", availablePort))
+	testrpc.WaitForTestAgent(t, a.RPC, "dc1")
+
+	t.Cleanup(func() {
+		a.Shutdown()
+	})
+
+	cases := []struct {
+		name   string
+		output string
+		args   []string
+	}{
+		{
+			name:   "sample output",
+			args:   []string{"-f=../testdata/demo.hcl"},
+			output: "demo.v2.Artist 'korn' created.",
+		},
+		{
+			name:   "nested data format",
+			args:   []string{"-f=../testdata/nested_data.hcl"},
+			output: "mesh.v2beta1.Destinations 'api' created.",
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			ui := cli.NewMockUi()
+			c := New(ui)
+
+			args := []string{
+				fmt.Sprintf("-grpc-addr=127.0.0.1:%d", availablePort),
+				"-token=root",
+			}
+
+			args = append(args, tc.args...)
+
+			code := c.Run(args)
+			require.Equal(t, 0, code)
+			require.Empty(t, ui.ErrorWriter.String())
+			require.Contains(t, ui.OutputWriter.String(), tc.output)
+		})
+	}
+}
+
+func TestResourceApplyCommand_StdIn(t *testing.T) {
+	if testing.Short() {
+		t.Skip("too slow for testing.Short")
+	}
+
+	t.Parallel()
+
+	availablePort := freeport.GetOne(t)
+	a := agent.NewTestAgent(t, fmt.Sprintf("ports { grpc = %d }", availablePort))
+	testrpc.WaitForTestAgent(t, a.RPC, "dc1")
+
+	t.Cleanup(func() {
+		a.Shutdown()
+	})
+
+	t.Run("hcl", func(t *testing.T) {
+		stdinR, stdinW := io.Pipe()
+
+		ui := cli.NewMockUi()
+		c := New(ui)
+		c.testStdin = stdinR
+
+		stdInput := `ID {
+			Type = gvk("demo.v2.Artist")
+			Name = "korn"
+			Tenancy {
+			  Namespace = "default"
+			  Partition = "default"
+			  PeerName = "local"
+			}
+		  }
+		  
+		  Data {
+			Name = "Korn"
+			Genre = "GENRE_METAL"
+		  }
+		  
+		  Metadata = {
+			"foo" = "bar"
+		  }`
+
+		go func() {
+			stdinW.Write([]byte(stdInput))
+			stdinW.Close()
+		}()
+
+		args := []string{
+			fmt.Sprintf("-grpc-addr=127.0.0.1:%d", availablePort),
+			"-token=root",
+			"-f",
+			"-",
+		}
+
+		code := c.Run(args)
+		require.Equal(t, 0, code)
+		require.Empty(t, ui.ErrorWriter.String())
+		// Todo: make up the read result check after finishing the read command
+		//expected := readResource(t, a, []string{"demo.v2.Artist", "korn"})
+		require.Contains(t, ui.OutputWriter.String(), "demo.v2.Artist 'korn' created.")
+		//require.Contains(t, ui.OutputWriter.String(), expected)
+	})
+
+	t.Run("json", func(t *testing.T) {
+		stdinR, stdinW := io.Pipe()
+
+		ui := cli.NewMockUi()
+		c := New(ui)
+		c.testStdin = stdinR
+
+		stdInput := `{
+			"data": {
+				"genre": "GENRE_METAL",
+				"name": "Korn"
+			},
+			"id": {
+				"name": "korn",
+				"tenancy": {
+					"namespace": "default",
+					"partition": "default",
+					"peerName": "local"
+				},
+				"type": {
+					"group": "demo",
+					"groupVersion": "v2",
+					"kind": "Artist"
+				}
+			},
+			"metadata": {
+				"foo": "bar"
+			}
+		}`
+
+		go func() {
+			stdinW.Write([]byte(stdInput))
+			stdinW.Close()
+		}()
+
+		args := []string{
+			"-f",
+			"-",
+			fmt.Sprintf("-grpc-addr=127.0.0.1:%d", availablePort),
+			"-token=root",
+		}
+
+		code := c.Run(args)
+		require.Equal(t, 0, code)
+		require.Empty(t, ui.ErrorWriter.String())
+		// Todo: make up the read result check after finishing the read command
+		//expected := readResource(t, a, []string{"demo.v2.Artist", "korn"})
+		require.Contains(t, ui.OutputWriter.String(), "demo.v2.Artist 'korn' created.")
+		//require.Contains(t, ui.OutputWriter.String(), expected)
+	})
+}
+
+func TestResourceApplyInvalidArgs(t *testing.T) {
+	t.Parallel()
+
+	type tc struct {
+		args         []string
+		expectedCode int
+		expectedErr  error
+	}
+
+	cases := map[string]tc{
+		"no file path": {
+			args:         []string{"-f"},
+			expectedCode: 1,
+			expectedErr:  errors.New("Failed to parse args: flag needs an argument: -f"),
+		},
+		"missing required flag": {
+			args:         []string{},
+			expectedCode: 1,
+			expectedErr:  errors.New("Required '-f' flag was not provided to specify where to load the resource content from"),
+		},
+		"file parsing failure": {
+			args:         []string{"-f=../testdata/invalid.hcl"},
+			expectedCode: 1,
+			expectedErr:  errors.New("Failed to decode resource from input file"),
+		},
+		"file not found": {
+			args:         []string{"-f=../testdata/test.hcl"},
+			expectedCode: 1,
+			expectedErr:  errors.New("Failed to load data: Failed to read file: open ../testdata/test.hcl: no such file or directory"),
+		},
+	}
+
+	for desc, tc := range cases {
+		t.Run(desc, func(t *testing.T) {
+			ui := cli.NewMockUi()
+			c := New(ui)
+
+			code := c.Run(tc.args)
+
+			require.Equal(t, tc.expectedCode, code)
+			require.Contains(t, ui.ErrorWriter.String(), tc.expectedErr.Error())
+		})
+	}
+}

command/resource/client/grpc-client.go

@@ -22,7 +22,7 @@ type GRPCClient struct {
 func NewGRPCClient(config *GRPCConfig) (*GRPCClient, error) {
 	conn, err := dial(config)
 	if err != nil {
-		return nil, fmt.Errorf("**** error dialing grpc: %+v", err)
+		return nil, fmt.Errorf("error dialing grpc: %+v", err)
 	}
 	return &GRPCClient{
 		Client: pbresource.NewResourceServiceClient(conn),

command/resource/client/grpc-config.go

@@ -157,3 +157,18 @@ func loadEnvToDefaultConfig(config *GRPCConfig) (*GRPCConfig, error) {
 
 	return config, nil
 }
+
+func (c GRPCConfig) GetToken() (string, error) {
+	if c.TokenFile != "" {
+		data, err := os.ReadFile(c.TokenFile)
+		if err != nil {
+			return "", err
+		}
+
+		return strings.TrimSpace(string(data)), nil
+	}
+	if c.Token != "" {
+		return c.Token, nil
+	}
+	return "", nil
+}

command/resource/client/grpc-flags.go

@@ -26,8 +26,10 @@ func (f *GRPCFlags) MergeFlagsIntoGRPCConfig(c *GRPCConfig) {
 	if strings.HasPrefix(strings.ToLower(f.address.String()), "https://") {
 		c.GRPCTLS = true
 	}
-	f.address.Set(removeSchemaFromGRPCAddress(f.address.String()))
-	f.address.Merge(&c.Address)
+	if f.address.v != nil {
+		f.address.Set(removeSchemaFromGRPCAddress(f.address.String()))
+		f.address.Merge(&c.Address)
+	}
 	// won't overwrite the value if it's false
 	if f.grpcTLS.v != nil && *f.grpcTLS.v {
 		f.grpcTLS.Merge(&c.GRPCTLS)
@@ -70,7 +72,19 @@ func (f *GRPCFlags) ClientFlags() *flag.FlagSet {
 			"default to the token of the Consul agent at the GRPC address.")
 	fs.Var(&f.tokenFile, "token-file",
 		"File containing the ACL token to use in the request instead of one specified "+
-			"via the -token argument or CONSUL_GRPC_TOKEN environment variable. "+
-			"This can also be specified via the CONSUL_GRPC_TOKEN_FILE environment variable.")
+			"via the -token-file argument or CONSUL_GRPC_TOKEN_FILE environment variable. "+
+			"Notice the tokenFile takes precedence over token flag and environment variables.")
 	return fs
 }
+
+func MergeFlags(dst, src *flag.FlagSet) {
+	if dst == nil {
+		panic("dst cannot be nil")
+	}
+	if src == nil {
+		return
+	}
+	src.VisitAll(func(f *flag.Flag) {
+		dst.Var(f.Value, f.Name, f.Usage)
+	})
+}

command/resource/client/usage.go

@@ -0,0 +1,85 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: BUSL-1.1
+
+package client
+
+import (
+	"bytes"
+	"flag"
+	"fmt"
+	"io"
+	"strings"
+
+	"github.com/kr/text"
+)
+
+func Usage(txt string, flags *flag.FlagSet) string {
+	u := &Usager{
+		Usage: txt,
+		Flags: flags,
+	}
+	return u.String()
+}
+
+type Usager struct {
+	Usage string
+	Flags *flag.FlagSet
+}
+
+func (u *Usager) String() string {
+	out := new(bytes.Buffer)
+	out.WriteString(strings.TrimSpace(u.Usage))
+	out.WriteString("\n")
+	out.WriteString("\n")
+
+	if u.Flags != nil {
+		var cmdFlags *flag.FlagSet
+		u.Flags.VisitAll(func(f *flag.Flag) {
+			if cmdFlags == nil {
+				cmdFlags = flag.NewFlagSet("", flag.ContinueOnError)
+			}
+			cmdFlags.Var(f.Value, f.Name, f.Usage)
+		})
+
+		if cmdFlags != nil {
+			printTitle(out, "Command Options")
+			cmdFlags.VisitAll(func(f *flag.Flag) {
+				printFlag(out, f)
+			})
+		}
+	}
+
+	return strings.TrimRight(out.String(), "\n")
+}
+
+// printTitle prints a consistently-formatted title to the given writer.
+func printTitle(w io.Writer, s string) {
+	fmt.Fprintf(w, "%s\n\n", s)
+}
+
+// printFlag prints a single flag to the given writer.
+func printFlag(w io.Writer, f *flag.Flag) {
+	example, _ := flag.UnquoteUsage(f)
+	if example != "" {
+		fmt.Fprintf(w, "  -%s=<%s>\n", f.Name, example)
+	} else {
+		fmt.Fprintf(w, "  -%s\n", f.Name)
+	}
+
+	indented := wrapAtLength(f.Usage, 5)
+	fmt.Fprintf(w, "%s\n\n", indented)
+}
+
+// maxLineLength is the maximum width of any line.
+const maxLineLength int = 72
+
+// wrapAtLength wraps the given text at the maxLineLength, taking into account
+// any provided left padding.
+func wrapAtLength(s string, pad int) string {
+	wrapped := text.Wrap(s, maxLineLength-pad)
+	lines := strings.Split(wrapped, "\n")
+	for i, line := range lines {
+		lines[i] = strings.Repeat(" ", pad) + line
+	}
+	return strings.Join(lines, "\n")
+}

command/resource/resource-grpc.go

@@ -0,0 +1,41 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: BUSL-1.1
+
+package resource
+
+import (
+	"context"
+	"fmt"
+
+	"google.golang.org/grpc/metadata"
+
+	"github.com/hashicorp/consul/command/resource/client"
+	"github.com/hashicorp/consul/proto-public/pbresource"
+)
+
+const (
+	HeaderConsulToken = "x-consul-token"
+)
+
+type ResourceGRPC struct {
+	C *client.GRPCClient
+}
+
+func (resource *ResourceGRPC) Apply(parsedResource *pbresource.Resource) (*pbresource.Resource, error) {
+	token, err := resource.C.Config.GetToken()
+	if err != nil {
+		return nil, err
+	}
+	ctx := context.Background()
+	if token != "" {
+		ctx = metadata.AppendToOutgoingContext(context.Background(), HeaderConsulToken, token)
+	}
+
+	defer resource.C.Conn.Close()
+	writeRsp, err := resource.C.Client.Write(ctx, &pbresource.WriteRequest{Resource: parsedResource})
+	if err != nil {
+		return nil, fmt.Errorf("error writing resource: %+v", err)
+	}
+
+	return writeRsp.Resource, err
+}