[WIP] Add command cluster updatepeerclusters

go.mod

@@ -4,11 +4,15 @@ go 1.12
 
 require (
 	github.com/fatih/color v1.7.0 // indirect
+	github.com/google/go-cmp v0.3.1 // indirect
 	github.com/kris-nova/logger v0.0.0-20181127235838-fd0d87064b06
 	github.com/kris-nova/lolgopher v0.0.0-20180921204813-313b3abb0d9b // indirect
 	github.com/mattn/go-colorable v0.1.2 // indirect
 	github.com/mattn/go-runewidth v0.0.4 // indirect
 	github.com/olekukonko/tablewriter v0.0.1
+	github.com/pkg/errors v0.8.1 // indirect
 	github.com/spf13/cobra v0.0.5
 	github.com/spf13/pflag v1.0.3
+	github.com/stretchr/objx v0.2.0 // indirect
+	github.com/stretchr/testify v1.3.0
 )

go.sum

@@ -4,11 +4,16 @@ github.com/coreos/etcd v3.3.10+incompatible/go.mod h1:uF7uidLiAD3TWHmW31ZFd/JWoc
 github.com/coreos/go-etcd v2.0.0+incompatible/go.mod h1:Jez6KQU2B/sWsbdaef3ED8NzMklzPG4d5KIOhIy30Tk=
 github.com/coreos/go-semver v0.2.0/go.mod h1:nnelYz7RCh+5ahJtPPxZlU+153eP4D4r3EedlOD2RNk=
 github.com/cpuguy83/go-md2man v1.0.10/go.mod h1:SmD6nW6nTyfqj6ABTjUi3V3JVMnlJmwcJI5acqYI6dE=
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/fatih/color v1.7.0 h1:DkWD4oS2D8LGGgTQ6IvwJJXSL5Vp2ffcQg58nFV38Ys=
 github.com/fatih/color v1.7.0/go.mod h1:Zm6kSWBoL9eyXnKyktHP6abPY2pDugNf5KwzbycvMj4=
 github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
+github.com/google/go-cmp v0.3.1 h1:Xye71clBPdm5HgqGwUkwhbynsUJZhDbS20FvLhQ2izg=
+github.com/google/go-cmp v0.3.1/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
 github.com/hashicorp/hcl v1.0.0/go.mod h1:E5yfLk+7swimpb2L/Alb/PJmXilQ/rhwaUYs4T20WEQ=
+github.com/inconshreveable/mousetrap v1.0.0 h1:Z8tu5sraLXCXIcARxBp/8cbvlwVa7Z1NHg9XEKhtSvM=
 github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANytuPF1OarO4DADm73n8=
 github.com/kris-nova/logger v0.0.0-20181127235838-fd0d87064b06 h1:vN4d3jSss3ExzUn2cE0WctxztfOgiKvMKnDrydBsg00=
 github.com/kris-nova/logger v0.0.0-20181127235838-fd0d87064b06/go.mod h1:++9BgZujZd4v0ZTZCb5iPsaomXdZWyxotIAh1IiDm44=
@@ -26,6 +31,9 @@ github.com/mitchellh/mapstructure v1.1.2/go.mod h1:FVVH3fgwuzCH5S8UJGiWEs2h04kUh
 github.com/olekukonko/tablewriter v0.0.1 h1:b3iUnf1v+ppJiOfNX4yxxqfWKMQPZR5yoh8urCTFX88=
 github.com/olekukonko/tablewriter v0.0.1/go.mod h1:vsDQFd/mU46D+Z4whnwzcISnGGzXWMclvtLoiIKAKIo=
 github.com/pelletier/go-toml v1.2.0/go.mod h1:5z9KED0ma1S8pY6P1sdut58dfprrGBbd/94hg7ilaic=
+github.com/pkg/errors v0.8.1 h1:iURUrRGxPUNPdy5/HRSm+Yj6okJ6UtLINN0Q9M4+h3I=
+github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/russross/blackfriday v1.5.2/go.mod h1:JO/DiYxRf+HjHt06OyowR9PTA263kcR/rfWxYHBV53g=
 github.com/spf13/afero v1.1.2/go.mod h1:j4pytiNVoe2o6bmDsKpLACNPDBIoEAkihy7loJ1B0CQ=
@@ -36,12 +44,21 @@ github.com/spf13/jwalterweatherman v1.0.0/go.mod h1:cQK4TGJAtQXfYWX+Ddv3mKDzgVb6
 github.com/spf13/pflag v1.0.3 h1:zPAT6CGy6wXeQ7NtTnaTerfKOsV6V6F8agHXFiazDkg=
 github.com/spf13/pflag v1.0.3/go.mod h1:DYY7MBk1bdzusC3SYhjObp+wFpr4gzcvqqNjLnInEg4=
 github.com/spf13/viper v1.3.2/go.mod h1:ZiWeW+zYFKm7srdB9IoDzzZXaJaI5eL9QjNiN/DMA2s=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/objx v0.2.0 h1:Hbg2NidpLE8veEBkEZTL3CvlkUIVzuU9jDplZO54c48=
+github.com/stretchr/objx v0.2.0/go.mod h1:qt09Ya8vawLte6SNmTgCsAVtYtaKzEcn8ATUoHMkEqE=
+github.com/stretchr/testify v1.2.2 h1:bSDNvY7ZPG5RlJ8otE/7V6gMiyenm9RtJ7IUVIAoJ1w=
 github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
+github.com/stretchr/testify v1.3.0 h1:TivCn/peBQ7UY8ooIcPgZFpTNSz0Q2U6UrFlUfqbe0Q=
+github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 github.com/ugorji/go/codec v0.0.0-20181204163529-d75b2dcb6bc8/go.mod h1:VFNgLljTbGfSG7qAOspJ7OScBnGdDN/yBr0sguwnwf0=
 github.com/xordataexchange/crypt v0.0.3-0.20170626215501-b2862e3d0a77/go.mod h1:aYKd//L2LvnjZzWKhF00oedf4jCCReLcmhLdhm1A27Q=
 golang.org/x/crypto v0.0.0-20181203042331-505ab145d0a9/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
 golang.org/x/sys v0.0.0-20181205085412-a5c9d58dba9a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20190222072716-a9d3bda3a223 h1:DH4skfRX4EBpamg7iV4ZlCpblAHI6s6TDM39bFZumv8=
 golang.org/x/sys v0.0.0-20190222072716-a9d3bda3a223/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
+gotest.tools v2.2.0+incompatible h1:VsBPFP1AI068pPrMxtb/S8Zkgf9xEmTLJjfM+P5UIEo=
+gotest.tools v2.2.0+incompatible/go.mod h1:DsYFclhRJ6vuDpmuTbkuFWG+y2sxOXAzmJt81HFBacw=

pkg/cmdutils/cmdutils.go

@@ -30,7 +30,7 @@ func NewResourceCmd(use, short, long string, aliases ...string) *cobra.Command {
 
 // GetNameArg tests to ensure there is only 1 name argument
 func GetNameArg(args []string) string {
-	if len(args) > 1 {
+	if len(args) > 1 || len(args) == 0 {
 		logger.Critical("only one argument is allowed to be used as a name")
 		os.Exit(1)
 	}

pkg/ctl/cluster/cluster.go

@@ -14,6 +14,8 @@ func Command(flagGrouping *cmdutils.FlagGrouping) *cobra.Command {
 
 	cmdutils.AddVerbCmd(flagGrouping, resourceCmd, createClusterCmd)
 	cmdutils.AddVerbCmd(flagGrouping, resourceCmd, listClustersCmd)
+	cmdutils.AddVerbCmd(flagGrouping, resourceCmd, getClusterDataCmd)
+	cmdutils.AddVerbCmd(flagGrouping, resourceCmd, updatePeerClustersCmd)
 
 	return resourceCmd
 }

pkg/ctl/cluster/get.go

@@ -0,0 +1,58 @@
+package cluster
+
+import (
+	"encoding/json"
+	"errors"
+	"github.com/streamnative/pulsarctl/pkg/cmdutils"
+)
+
+
+var commandUsedFor = "This command is used for getting the cluster data of the specified cluster."
+var commandExample =
+		"{\n" +
+		"    serviceUrl : http://localhost:8080, \n" +
+		"    serviceUrlTls : https://localhost:8080, \n" +
+		"    brokerServiceUrl: pulsar://localhost:6650, \n" +
+		"    brokerServiceUrlTls: pulsar+ssl://localhost:6650, \n" +
+		"    peerClusterNames: \"\" \n" +
+		"}\n"
+var commandPermission = "This command only admin can use."
+
+func getClusterDataCmd(vc *cmdutils.VerbCmd)  {
+	vc.SetDescription(
+		"get",
+		"Get the configuration data for the specified cluster",
+		concat("\n"),
+		"get")
+
+	vc.SetRunFuncWithNameArg(func() error {
+		return doGetClusterData(vc)
+	})
+}
+
+func doGetClusterData(vc *cmdutils.VerbCmd) error {
+	clusterName := vc.NameArg
+	if clusterName == "" {
+		return errors.New("Should specified a cluster name")
+	}
+
+	admin := cmdutils.NewPulsarClient()
+	clusterData, err := admin.Clusters().Get(clusterName)
+	if err != nil {
+		cmdutils.PrintError(vc.Command.OutOrStderr(), err)
+	} else {
+		s, err  := json.MarshalIndent(clusterData,  "", "    ")
+		if err != nil {
+			return err
+		}
+		vc.Command.Println(string(s))
+	}
+
+	return err
+}
+
+func concat(join string) string {
+	return "USED FOR:" + join + "\t" + commandUsedFor  + join +
+		"PERMISSION:" + join + "\t" + commandPermission + join +
+		"EXAMPLE:" + join + commandExample
+}

pkg/ctl/cluster/get_test.go

@@ -0,0 +1,41 @@
+package cluster
+
+import (
+	"encoding/json"
+	"fmt"
+	"github.com/streamnative/pulsarctl/pkg/pulsar"
+	"github.com/stretchr/testify/assert"
+	"regexp"
+	"testing"
+)
+
+
+func TestGetClusterData(t *testing.T) {
+	args := []string{"get", "standalone"}
+	out, err := TestClusterCommands(getClusterDataCmd, args)
+	if err != nil {
+		t.Error(err)
+	}
+	c := pulsar.ClusterData{}
+	err = json.Unmarshal(out.Bytes(), &c)
+	if err != nil {
+		fmt.Println(err)
+	}
+
+	pulsarUrl, err := regexp.Compile("^pulsar://[a-z-A-Z0-9]*:6650$")
+	if err != nil {
+		t.Error(err)
+	}
+
+	res := pulsarUrl.MatchString(c.BrokerServiceURL)
+	assert.True(t, res)
+
+	httpUrl, err := regexp.Compile("^http://[a-z-A-Z0-9]*:8080$")
+	if err != nil {
+		t.Error(err)
+	}
+
+	res = httpUrl.MatchString(c.ServiceURL)
+	assert.True(t, res)
+}
+

pkg/ctl/cluster/test.go

@@ -0,0 +1,37 @@
+package cluster
+
+import (
+	"bytes"
+	"github.com/kris-nova/logger"
+	"github.com/spf13/cobra"
+	"github.com/streamnative/pulsarctl/pkg/cmdutils"
+)
+
+func TestClusterCommands(newVerb func(cmd *cmdutils.VerbCmd), args []string) (out *bytes.Buffer, err error)  {
+	var rootCmd = &cobra.Command {
+		Use:	"pulsarctl [command]",
+		Short: 	"a CLI for Apache Pulsar",
+		Run: func(cmd *cobra.Command, _ []string) {
+			if err := cmd.Help(); err != nil {
+				logger.Debug("ignoring error %q", err.Error())
+			}
+		},
+	}
+
+	buf := new(bytes.Buffer)
+	rootCmd.SetOut(buf)
+	rootCmd.SetArgs(append([]string{"clusters"}, args...))
+
+
+	resourceCmd := cmdutils.NewResourceCmd(
+		"clusters",
+		"Operations about cluster(s)",
+		"",
+		"cluster")
+	flagGrouping := cmdutils.NewGrouping()
+	cmdutils.AddVerbCmd(flagGrouping, resourceCmd, newVerb)
+	rootCmd.AddCommand(resourceCmd)
+	err = rootCmd.Execute()
+
+	return buf, err
+}

pkg/ctl/cluster/update_peer_clusters.go

@@ -0,0 +1,44 @@
+package cluster
+
+import (
+	"github.com/spf13/pflag"
+	"github.com/streamnative/pulsarctl/pkg/cmdutils"
+	"github.com/streamnative/pulsarctl/pkg/pulsar"
+)
+
+func updatePeerClustersCmd(vc *cmdutils.VerbCmd)  {
+	vc.SetDescription(
+		"update-peer-clusters",
+		"",
+		"",
+		"upc")
+
+	clusterData := &pulsar.ClusterData{}
+
+	vc.SetRunFuncWithNameArg(func() error {
+		return doUpdatePeerClusters(vc, clusterData)
+	})
+
+	vc.FlagSetGroup.InFlagSet("Update peer clusters", func(set *pflag.FlagSet) {
+		set.StringArrayVarP(
+			&clusterData.PeerClusterNames,
+			"peer-cluster",
+			"p",
+			[]string{""},
+			"Cluster to be registered as a peer-cluster of this cluster")
+	})
+
+}
+
+func doUpdatePeerClusters(vc *cmdutils.VerbCmd, clusterData *pulsar.ClusterData) error {
+	clusterData.Name = vc.NameArg
+
+	admin := cmdutils.NewPulsarClient()
+	err := admin.Clusters().UpdatePeerClusters(clusterData.Name, clusterData.PeerClusterNames)
+	if err != nil {
+		cmdutils.PrintError(vc.Command.OutOrStderr(), err)
+	} else {
+		vc.Command.Printf("Peer clusters updated")
+	}
+	return err
+}

pkg/ctl/cluster/update_peer_clusters_test.go

@@ -0,0 +1 @@
+package cluster

pkg/pulsar/admin.go

@@ -128,6 +128,28 @@ func (c *client) delete(endpoint string, obj interface{}) error {
 	return nil
 }
 
+func (c *client) post(endpoint string, in, obj interface{}) error {
+	req, err := c.newRequest(http.MethodPut, endpoint)
+	if err != nil {
+		return err
+	}
+	req.obj = in
+
+	resp, err := checkSuccessful(c.doRequest(req))
+	if err != nil {
+		return err
+	}
+	defer safeRespClose(resp)
+
+	if obj != nil {
+		if err := decodeJsonBody(resp, &obj); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
 type request struct {
 	method string
 	url *url.URL
@@ -187,6 +209,7 @@ func (c *client) useragent() string {
 
 func (c *client) doRequest(r *request) (*http.Response, error) {
 	req, err := r.toHTTP()
+	r.toHTTP()
 	if err != nil {
 		return nil, err
 	}

pkg/pulsar/cluster.go

@@ -7,6 +7,7 @@ type Clusters interface {
 	Get(string) (ClusterData, error)
 	Create(ClusterData) error
 	Delete(string) error
+	UpdatePeerClusters(string, []string) error
 }
 
 type clusters struct {
@@ -45,4 +46,7 @@ func (c *clusters) Delete(name string) error {
 	return c.client.delete(endpoint, nil)
 }
 
-
+func (c *clusters) UpdatePeerClusters(cluster string, peerClusters []string) error {
+	endpoint :=  c.client.endpoint(c.basePath, cluster, "peers")
+	return c.client.post(endpoint, peerClusters, nil)
+}

pkg/pulsar/data.go

@@ -8,4 +8,6 @@ type ClusterData struct {
 	BrokerServiceURL string `json:"brokerServiceUrl"`
 	BrokerServiceURLTls string `json:"brokerServiceUrlTls"`
 	PeerClusterNames []string `json:"peerClusterNames"`
+
+	test int
 }