Add broker commands for pulsarctl

pkg/ctl/brokers/broker.go

@@ -0,0 +1,43 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package brokers
+
+import (
+	"github.com/spf13/cobra"
+	"github.com/streamnative/pulsarctl/pkg/cmdutils"
+)
+
+func Command(flagGrouping *cmdutils.FlagGrouping) *cobra.Command {
+	resourceCmd := cmdutils.NewResourceCmd(
+		"brokers",
+		"Operations about broker(s)",
+		"",
+		"broker")
+
+	cmdutils.AddVerbCmd(flagGrouping, resourceCmd, getBrokerListCmd)
+	cmdutils.AddVerbCmd(flagGrouping, resourceCmd, getDynamicConfigListNameCmd)
+	cmdutils.AddVerbCmd(flagGrouping, resourceCmd, getOwnedNamespacesCmd)
+	cmdutils.AddVerbCmd(flagGrouping, resourceCmd, updateDynamicConfig)
+	cmdutils.AddVerbCmd(flagGrouping, resourceCmd, deleteDynamicConfigCmd)
+	cmdutils.AddVerbCmd(flagGrouping, resourceCmd, getAllDynamicConfigsCmd)
+	cmdutils.AddVerbCmd(flagGrouping, resourceCmd, getInternalConfigCmd)
+	cmdutils.AddVerbCmd(flagGrouping, resourceCmd, getRuntimeConfigCmd)
+	cmdutils.AddVerbCmd(flagGrouping, resourceCmd, healthCheckCmd)
+
+	return resourceCmd
+}

pkg/ctl/brokers/config.go

@@ -0,0 +1,73 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package brokers
+
+import (
+	"github.com/streamnative/pulsarctl/pkg/cmdutils"
+	"github.com/streamnative/pulsarctl/pkg/pulsar"
+)
+
+func getInternalConfigCmd(vc *cmdutils.VerbCmd) {
+	desc := pulsar.LongDescription{}
+	desc.CommandUsedFor = "Get internal configuration information"
+	desc.CommandPermission = "This command requires super-user permissions."
+
+	var examples []pulsar.Example
+	list := pulsar.Example{
+		Desc:    "Get internal configuration information",
+		Command: "pulsarctl brokers get-internal-config",
+	}
+	examples = append(examples, list)
+	desc.CommandExamples = examples
+
+	var out []pulsar.Output
+	successOut := pulsar.Output{
+		Desc: "normal output",
+		Out: "{\n" +
+			"  \"zookeeperServers\": \"127.0.0.1:2181\",\n" +
+			"  \"configurationStoreServers\": \"127.0.0.1:2181\",\n" +
+			"  \"ledgersRootPath\": \"/ledgers\",\n" +
+			"  \"stateStorageServiceUrl\": \"bk://127.0.0.1:4181\"\n" +
+			"}",
+	}
+
+	out = append(out, successOut)
+	desc.CommandOutput = out
+
+	vc.SetDescription(
+		"get-internal-config",
+		"Get internal configuration information",
+		desc.ToString(),
+		desc.ExampleToString(),
+		"get-internal-config")
+
+	vc.SetRunFunc(func() error {
+		return doGetInternalConfig(vc)
+	})
+}
+
+func doGetInternalConfig(vc *cmdutils.VerbCmd) error {
+	admin := cmdutils.NewPulsarClient()
+	brokersData, err := admin.Brokers().GetInternalConfigurationData()
+	if err != nil {
+		cmdutils.PrintError(vc.Command.OutOrStderr(), err)
+	} else {
+		cmdutils.PrintJSON(vc.Command.OutOrStdout(), brokersData)
+	}
+	return err
+}

pkg/ctl/brokers/config_test.go

@@ -0,0 +1,60 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package brokers
+
+import (
+	"encoding/json"
+	"testing"
+
+	"github.com/streamnative/pulsarctl/pkg/pulsar"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestGetInternalConfig(t *testing.T) {
+	args := []string{"get-internal-config"}
+	internalOut, execErr, _, _ := TestBrokersCommands(getInternalConfigCmd, args)
+	assert.Nil(t, execErr)
+
+	var internalData pulsar.InternalConfigurationData
+	err := json.Unmarshal(internalOut.Bytes(), &internalData)
+	assert.Nil(t, err)
+
+	assert.Equal(t, "127.0.0.1:2181", internalData.ZookeeperServers)
+	assert.Equal(t, "127.0.0.1:2181", internalData.ConfigurationStoreServers)
+	assert.Equal(t, "/ledgers", internalData.LedgersRootPath)
+	assert.Equal(t, "bk://127.0.0.1:4181", internalData.StateStorageServiceURL)
+}
+
+func TestGetRuntimeConfig(t *testing.T) {
+	args := []string{"get-runtime-config"}
+	runtimeOut, execErr, _, _ := TestBrokersCommands(getRuntimeConfigCmd, args)
+	assert.Nil(t, execErr)
+
+	var runtimeConf map[string]string
+	err := json.Unmarshal(runtimeOut.Bytes(), &runtimeConf)
+	assert.Nil(t, err)
+
+	assert.Equal(t, "false", runtimeConf["authenticateOriginalAuthData"])
+	assert.Equal(t, "true", runtimeConf["backlogQuotaCheckEnabled"])
+	assert.Equal(t, "0.0.0.0", runtimeConf["bindAddress"])
+	assert.Equal(t, "CRC32", runtimeConf["managedLedgerDigestType"])
+	assert.Equal(t, "127.0.0.1:2181", runtimeConf["zookeeperServers"])
+	assert.Equal(t, "30000", runtimeConf["zooKeeperSessionTimeoutMillis"])
+	assert.Equal(t, "300000", runtimeConf["webSocketSessionIdleTimeoutMillis"])
+	assert.Equal(t, "30", runtimeConf["keepAliveIntervalSeconds"])
+}

pkg/ctl/brokers/delete_dynamic_config.go

@@ -0,0 +1,87 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package brokers
+
+import (
+	"github.com/spf13/cobra"
+	"github.com/spf13/pflag"
+	"github.com/streamnative/pulsarctl/pkg/cmdutils"
+	"github.com/streamnative/pulsarctl/pkg/pulsar"
+)
+
+func deleteDynamicConfigCmd(vc *cmdutils.VerbCmd) {
+	desc := pulsar.LongDescription{}
+	desc.CommandUsedFor = "Delete dynamic-serviceConfiguration of broker"
+	desc.CommandPermission = "This command requires super-user permissions."
+
+	var examples []pulsar.Example
+	list := pulsar.Example{
+		Desc:    "Delete dynamic-serviceConfiguration of broker",
+		Command: "pulsarctl brokers delete-dynamic-config --config (config name)",
+	}
+	examples = append(examples, list)
+	desc.CommandExamples = examples
+
+	var out []pulsar.Output
+	successOut := pulsar.Output{
+		Desc: "normal output",
+		Out:  "Deleted dynamic config: (config name) successful.",
+	}
+
+	failOut := pulsar.Output{
+		Desc: "Can't update non-dynamic configuration, please check `--config` arg.",
+		Out:  "[✖]  code: 412 reason:  Can't update non-dynamic configuration",
+	}
+
+	out = append(out, successOut, failOut)
+	desc.CommandOutput = out
+
+	vc.SetDescription(
+		"delete-dynamic-config",
+		"Delete dynamic-serviceConfiguration of broker",
+		desc.ToString(),
+		desc.ExampleToString(),
+		"delete-dynamic-config")
+
+	brokerData := &pulsar.BrokerData{}
+
+	vc.SetRunFunc(func() error {
+		return doDeleteDynamicConf(vc, brokerData)
+	})
+
+	// register the params
+	vc.FlagSetGroup.InFlagSet("BrokerData", func(flagSet *pflag.FlagSet) {
+		flagSet.StringVar(
+			&brokerData.ConfigName,
+			"config",
+			"",
+			"service-configuration name")
+		cobra.MarkFlagRequired(flagSet, "config")
+	})
+}
+
+func doDeleteDynamicConf(vc *cmdutils.VerbCmd, brokerData *pulsar.BrokerData) error {
+	admin := cmdutils.NewPulsarClient()
+	err := admin.Brokers().DeleteDynamicConfiguration(brokerData.ConfigName)
+	if err != nil {
+		cmdutils.PrintError(vc.Command.OutOrStderr(), err)
+	} else {
+		vc.Command.Printf("Deleted dynamic config: %s successful.", brokerData.ConfigName)
+	}
+	return err
+}

pkg/ctl/brokers/delete_dynamic_config_test.go

@@ -0,0 +1,37 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package brokers
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestDeleteDynamicConfig(t *testing.T) {
+	args := []string{"delete-dynamic-config", "--config", "dispatcherMinReadBatchSize"}
+	delOut, execErr, _, _ := TestBrokersCommands(deleteDynamicConfigCmd, args)
+	assert.Nil(t, execErr)
+	expectedOut := "Deleted dynamic config: dispatcherMinReadBatchSize successful."
+	assert.Equal(t, expectedOut, delOut.String())
+
+	failArgs := []string{"delete-dynamic-config", "--config", "errorName"}
+	_, nameErr, _, _ := TestBrokersCommands(deleteDynamicConfigCmd, failArgs)
+	assert.NotNil(t, nameErr)
+	assert.Equal(t, "code: 412 reason:  Can't update non-dynamic configuration", nameErr.Error())
+}

pkg/ctl/brokers/get_all_dynamic_config.go

@@ -0,0 +1,69 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package brokers
+
+import (
+	"github.com/streamnative/pulsarctl/pkg/cmdutils"
+	"github.com/streamnative/pulsarctl/pkg/pulsar"
+)
+
+func getAllDynamicConfigsCmd(vc *cmdutils.VerbCmd) {
+	desc := pulsar.LongDescription{}
+	desc.CommandUsedFor = "Get all overridden dynamic-configuration values"
+	desc.CommandPermission = "This command requires super-user permissions."
+
+	var examples []pulsar.Example
+	list := pulsar.Example{
+		Desc:    "Get all overridden dynamic-configuration values",
+		Command: "pulsarctl brokers get-all-dynamic-config",
+	}
+	examples = append(examples, list)
+	desc.CommandExamples = examples
+
+	var out []pulsar.Output
+	successOut := pulsar.Output{
+		Desc: "normal output",
+		Out: "{\n" +
+			"  \"dispatchThrottlingRatePerTopicInMsg\": \"true\"\n" +
+			"}",
+	}
+	out = append(out, successOut)
+	desc.CommandOutput = out
+
+	vc.SetDescription(
+		"get-all-dynamic-config",
+		"Get all overridden dynamic-configuration values",
+		desc.ToString(),
+		desc.ExampleToString(),
+		"get-all-dynamic-config")
+
+	vc.SetRunFunc(func() error {
+		return doGetAllDynamicConfigs(vc)
+	})
+}
+
+func doGetAllDynamicConfigs(vc *cmdutils.VerbCmd) error {
+	admin := cmdutils.NewPulsarClient()
+	brokersData, err := admin.Brokers().GetAllDynamicConfigurations()
+	if err != nil {
+		cmdutils.PrintError(vc.Command.OutOrStderr(), err)
+	} else {
+		cmdutils.PrintJSON(vc.Command.OutOrStdout(), brokersData)
+	}
+	return err
+}