Add backlog, retention and ttl commands for pulsarctl

pkg/ctl/namespace/backlog_quota_test.go

@@ -0,0 +1,57 @@
+// 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 namespace
+
+import (
+	"encoding/json"
+	"github.com/streamnative/pulsarctl/pkg/pulsar"
+	"github.com/stretchr/testify/assert"
+	"testing"
+)
+
+func TestBacklogQuota(t *testing.T) {
+	args := []string{"set-backlog-quota", "public/default", "--limit", "12M", "--policy", "consumer_backlog_eviction"}
+	setOut, execErr, _, _ := TestNamespaceCommands(setBacklogQuota, args)
+	assert.Nil(t, execErr)
+	assert.Equal(t, setOut.String(), "Set backlog quota successfully for [public/default]")
+
+	getArgs := []string{"get-backlog-quotas", "public/default"}
+	getOut, execErr, _, _ := TestNamespaceCommands(getBacklogQuota, getArgs)
+	assert.Nil(t, execErr)
+	var backlogQuotaMap map[pulsar.BacklogQuotaType]pulsar.BacklogQuota
+	err := json.Unmarshal(getOut.Bytes(), &backlogQuotaMap)
+	assert.Nil(t, err)
+
+	for key, value := range backlogQuotaMap {
+		assert.Equal(t, key, pulsar.DestinationStorage)
+		assert.Equal(t, value.Limit, int64(12582912))
+		assert.Equal(t, value.Policy, pulsar.ConsumerBacklogEviction)
+	}
+
+	delArgs := []string{"remove-backlog-quota", "public/default"}
+	delOut, execErr, _, _ := TestNamespaceCommands(removeBacklogQuota, delArgs)
+	assert.Nil(t, execErr)
+	assert.Equal(t, delOut.String(), "Remove backlog quota successfully for [public/default]")
+}
+
+func TestFailureBacklogQuota(t *testing.T) {
+	args := []string{"set-backlog-quota", "public/default", "--limit", "12M", "--policy", "no-support-policy"}
+	_, execErr, _, _ := TestNamespaceCommands(setBacklogQuota, args)
+	assert.NotNil(t, execErr)
+	assert.Equal(t, execErr.Error(), "invalid retention policy type: no-support-policy")
+}

pkg/ctl/namespace/get_backlog_quota.go

@@ -0,0 +1,86 @@
+// 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 namespace
+
+import (
+	"github.com/streamnative/pulsarctl/pkg/cmdutils"
+	"github.com/streamnative/pulsarctl/pkg/pulsar"
+)
+
+func getBacklogQuota(vc *cmdutils.VerbCmd) {
+	desc := pulsar.LongDescription{}
+	desc.CommandUsedFor = "Get the backlog quota policies for a namespace"
+	desc.CommandPermission = "This command requires tenant admin permissions."
+
+	var examples []pulsar.Example
+	getBacklog := pulsar.Example{
+		Desc:    "Get the backlog quota policies for a namespace",
+		Command: "pulsarctl namespaces get-backlog-quotas tenant/namespace",
+	}
+	examples = append(examples, getBacklog)
+	desc.CommandExamples = examples
+
+	var out []pulsar.Output
+	successOut := pulsar.Output{
+		Desc: "normal output",
+		Out: "{\n" +
+			"  \"destination_storage\" : {\n" +
+			"    \"limit\" : 10737418240,\n" +
+			"    \"policy\" : \"producer_request_hold\"\n" +
+			"  }\n" +
+			"}",
+	}
+
+	notTenantName := pulsar.Output{
+		Desc: "you must specify a tenant/namespace name, please check if the tenant/namespace name is provided",
+		Out:  "[✖]  only one argument is allowed to be used as a name",
+	}
+
+	notExistTenantName := pulsar.Output{
+		Desc: "the tenant name not exist, please check the tenant name",
+		Out:  "[✖]  code: 404 reason: Tenant does not exist",
+	}
+
+	notExistNsName := pulsar.Output{
+		Desc: "the namespace not exist, please check namespace name",
+		Out:  "[✖]  code: 404 reason: Namespace <tenant/namespace> does not exist",
+	}
+	out = append(out, successOut, notTenantName, notExistTenantName, notExistNsName)
+	desc.CommandOutput = out
+
+	vc.SetDescription(
+		"get-backlog-quotas",
+		"Get the backlog quota policies for a namespace",
+		desc.ToString(),
+		"get-backlog-quotas",
+	)
+
+	vc.SetRunFuncWithNameArg(func() error {
+		return doGetBacklogQuotas(vc)
+	})
+}
+
+func doGetBacklogQuotas(vc *cmdutils.VerbCmd) error {
+	ns := vc.NameArg
+	admin := cmdutils.NewPulsarClient()
+	backlogQuotasMap, err := admin.Namespaces().GetBacklogQuotaMap(ns)
+	if err == nil {
+		cmdutils.PrintJson(vc.Command.OutOrStdout(), &backlogQuotasMap)
+	}
+	return err
+}

pkg/ctl/namespace/get_message_ttl.go

@@ -0,0 +1,82 @@
+// 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 namespace
+
+import (
+	"github.com/streamnative/pulsarctl/pkg/cmdutils"
+	"github.com/streamnative/pulsarctl/pkg/pulsar"
+)
+
+func getMessageTTL(vc *cmdutils.VerbCmd) {
+	desc := pulsar.LongDescription{}
+	desc.CommandUsedFor = "Get message TTL for a namespace"
+	desc.CommandPermission = "This command requires tenant admin permissions."
+
+	var examples []pulsar.Example
+	setMsgTTL := pulsar.Example{
+		Desc:    "Get message TTL for a namespace",
+		Command: "pulsarctl namespaces get-message-ttl tenant/namespace",
+	}
+	examples = append(examples, setMsgTTL)
+	desc.CommandExamples = examples
+
+	var out []pulsar.Output
+	successOut := pulsar.Output{
+		Desc: "normal output",
+		Out:  "<ttl-value>",
+	}
+
+	notTenantName := pulsar.Output{
+		Desc: "you must specify a tenant/namespace name, please check if the tenant/namespace name is provided",
+		Out:  "[✖]  only one argument is allowed to be used as a name",
+	}
+
+	notExistTenantName := pulsar.Output{
+		Desc: "the tenant name not exist, please check the tenant name",
+		Out:  "[✖]  code: 404 reason: Tenant does not exist",
+	}
+
+	notExistNsName := pulsar.Output{
+		Desc: "the namespace not exist, please check namespace name",
+		Out:  "[✖]  code: 404 reason: Namespace <tenant/namespace> does not exist",
+	}
+
+	out = append(out, successOut, notTenantName, notExistTenantName, notExistNsName)
+	desc.CommandOutput = out
+
+	vc.SetDescription(
+		"get-message-ttl",
+		"Get Message TTL for a namespace",
+		desc.ToString(),
+		"get-message-ttl",
+	)
+
+	vc.SetRunFuncWithNameArg(func() error {
+		return doGetMessageTTL(vc)
+	})
+}
+
+func doGetMessageTTL(vc *cmdutils.VerbCmd) error {
+	ns := vc.NameArg
+	admin := cmdutils.NewPulsarClient()
+	ttl, err := admin.Namespaces().GetNamespaceMessageTTL(ns)
+	if err == nil {
+		vc.Command.Print(ttl)
+	}
+	return err
+}

pkg/ctl/namespace/get_retention.go

@@ -0,0 +1,85 @@
+// 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 namespace
+
+import (
+	"github.com/streamnative/pulsarctl/pkg/cmdutils"
+	"github.com/streamnative/pulsarctl/pkg/pulsar"
+)
+
+func getRetention(vc *cmdutils.VerbCmd) {
+	desc := pulsar.LongDescription{}
+	desc.CommandUsedFor = "Get the retention policy for a namespace"
+	desc.CommandPermission = "This command requires tenant admin permissions."
+
+	var examples []pulsar.Example
+	getRetention := pulsar.Example{
+		Desc:    "Get the retention policy for a namespace",
+		Command: "pulsarctl namespaces get-retention tenant/namespace",
+	}
+	examples = append(examples, getRetention)
+	desc.CommandExamples = examples
+
+	var out []pulsar.Output
+	successOut := pulsar.Output{
+		Desc: "normal output",
+		Out: "{\n" +
+			"  \"RetentionTimeInMinutes\": 0,\n" +
+			"  \"RetentionSizeInMB\": 0\n" +
+			"}",
+	}
+
+	notTenantName := pulsar.Output{
+		Desc: "you must specify a tenant/namespace name, please check if the tenant/namespace name is provided",
+		Out:  "[✖]  only one argument is allowed to be used as a name",
+	}
+
+	notExistTenantName := pulsar.Output{
+		Desc: "the tenant name not exist, please check the tenant name",
+		Out:  "[✖]  code: 404 reason: Tenant does not exist",
+	}
+
+	notExistNsName := pulsar.Output{
+		Desc: "the namespace not exist, please check namespace name",
+		Out:  "[✖]  code: 404 reason: Namespace <tenant/namespace> does not exist",
+	}
+
+	out = append(out, successOut, notTenantName, notExistTenantName, notExistNsName)
+	desc.CommandOutput = out
+
+	vc.SetDescription(
+		"get-retention",
+		"Get the retention policy for a namespace",
+		desc.ToString(),
+		"get-retention",
+	)
+
+	vc.SetRunFuncWithNameArg(func() error {
+		return doGetRetention(vc)
+	})
+}
+
+func doGetRetention(vc *cmdutils.VerbCmd) error {
+	ns := vc.NameArg
+	admin := cmdutils.NewPulsarClient()
+	policy, err := admin.Namespaces().GetRetention(ns)
+	if err == nil {
+		cmdutils.PrintJson(vc.Command.OutOrStdout(), &policy)
+	}
+	return err
+}

pkg/ctl/namespace/message_ttl_test.go

@@ -0,0 +1,41 @@
+// 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 namespace
+
+import (
+	"github.com/stretchr/testify/assert"
+	"testing"
+)
+
+func TestMessageTTL(t *testing.T) {
+	setTTLArgs := []string{"set-message-ttl", "public/default", "-t", "20"}
+	setOut, execErr, _, _ := TestNamespaceCommands(setMessageTTL, setTTLArgs)
+	assert.Nil(t, execErr)
+	assert.Equal(t, setOut.String(), "Set message TTL successfully for [public/default]")
+
+	getTTLArgs := []string{"get-message-ttl", "public/default"}
+	getOut, execErr, _, _ := TestNamespaceCommands(getMessageTTL, getTTLArgs)
+	assert.Nil(t, execErr)
+	assert.Equal(t, getOut.String(), "20")
+
+	// test negative value for ttl arg
+	setTTLArgs = []string{"set-message-ttl", "public/default", "-t", "-2"}
+	_, execErr, _, _ = TestNamespaceCommands(setMessageTTL, setTTLArgs)
+	assert.NotNil(t, execErr)
+	assert.Equal(t, execErr.Error(), "code: 412 reason: Invalid value for message TTL")
+}

pkg/ctl/namespace/namespace.go

@@ -35,6 +35,13 @@ func Command(flagGrouping *cmdutils.FlagGrouping) *cobra.Command {
 	cmdutils.AddVerbCmd(flagGrouping, resourceCmd, getPolicies)
 	cmdutils.AddVerbCmd(flagGrouping, resourceCmd, createNs)
 	cmdutils.AddVerbCmd(flagGrouping, resourceCmd, deleteNs)
+	cmdutils.AddVerbCmd(flagGrouping, resourceCmd, setMessageTTL)
+	cmdutils.AddVerbCmd(flagGrouping, resourceCmd, getMessageTTL)
+	cmdutils.AddVerbCmd(flagGrouping, resourceCmd, getRetention)
+	cmdutils.AddVerbCmd(flagGrouping, resourceCmd, setRetention)
+	cmdutils.AddVerbCmd(flagGrouping, resourceCmd, getBacklogQuota)
+	cmdutils.AddVerbCmd(flagGrouping, resourceCmd, setBacklogQuota)
+	cmdutils.AddVerbCmd(flagGrouping, resourceCmd, removeBacklogQuota)
 
 	return resourceCmd
 }