From a6ad55eddab53a332011e80defa7584c4f5f6bb2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=86=89=E5=B0=8F=E9=BE=99?= Date: Tue, 17 Sep 2019 11:47:35 +0800 Subject: [PATCH] Revert "Add backlog, retention and ttl commands for pulsarctl (#61)" This reverts commit eba2a94859640ef4c5d6e94517c3e1d3d8fc7add. --- pkg/ctl/namespace/backlog_quota_test.go | 57 --------- pkg/ctl/namespace/get_backlog_quota.go | 86 ------------- pkg/ctl/namespace/get_message_ttl.go | 82 ------------ pkg/ctl/namespace/get_retention.go | 85 ------------- pkg/ctl/namespace/message_ttl_test.go | 41 ------ pkg/ctl/namespace/namespace.go | 7 -- pkg/ctl/namespace/remove_backlog_quota.go | 83 ------------- pkg/ctl/namespace/retention_test.go | 65 ---------- pkg/ctl/namespace/set_backlog_quota.go | 131 ------------------- pkg/ctl/namespace/set_message_ttl.go | 100 --------------- pkg/ctl/namespace/set_retention.go | 145 ---------------------- pkg/ctl/namespace/util.go | 67 ---------- pkg/pulsar/backlog_quota.go | 34 +++-- pkg/pulsar/data.go | 8 +- pkg/pulsar/namespace.go | 94 -------------- pkg/pulsar/retention_policies.go | 11 +- 16 files changed, 25 insertions(+), 1071 deletions(-) delete mode 100644 pkg/ctl/namespace/backlog_quota_test.go delete mode 100644 pkg/ctl/namespace/get_backlog_quota.go delete mode 100644 pkg/ctl/namespace/get_message_ttl.go delete mode 100644 pkg/ctl/namespace/get_retention.go delete mode 100644 pkg/ctl/namespace/message_ttl_test.go delete mode 100644 pkg/ctl/namespace/remove_backlog_quota.go delete mode 100644 pkg/ctl/namespace/retention_test.go delete mode 100644 pkg/ctl/namespace/set_backlog_quota.go delete mode 100644 pkg/ctl/namespace/set_message_ttl.go delete mode 100644 pkg/ctl/namespace/set_retention.go delete mode 100644 pkg/ctl/namespace/util.go diff --git a/pkg/ctl/namespace/backlog_quota_test.go b/pkg/ctl/namespace/backlog_quota_test.go deleted file mode 100644 index 8ebbbf629..000000000 --- a/pkg/ctl/namespace/backlog_quota_test.go +++ /dev/null @@ -1,57 +0,0 @@ -// 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") -} diff --git a/pkg/ctl/namespace/get_backlog_quota.go b/pkg/ctl/namespace/get_backlog_quota.go deleted file mode 100644 index 3bd9e9326..000000000 --- a/pkg/ctl/namespace/get_backlog_quota.go +++ /dev/null @@ -1,86 +0,0 @@ -// 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 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 -} diff --git a/pkg/ctl/namespace/get_message_ttl.go b/pkg/ctl/namespace/get_message_ttl.go deleted file mode 100644 index da9a856f4..000000000 --- a/pkg/ctl/namespace/get_message_ttl.go +++ /dev/null @@ -1,82 +0,0 @@ -// 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: "", - } - - 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 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 -} diff --git a/pkg/ctl/namespace/get_retention.go b/pkg/ctl/namespace/get_retention.go deleted file mode 100644 index a46f50827..000000000 --- a/pkg/ctl/namespace/get_retention.go +++ /dev/null @@ -1,85 +0,0 @@ -// 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 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 -} diff --git a/pkg/ctl/namespace/message_ttl_test.go b/pkg/ctl/namespace/message_ttl_test.go deleted file mode 100644 index 0d9552e5b..000000000 --- a/pkg/ctl/namespace/message_ttl_test.go +++ /dev/null @@ -1,41 +0,0 @@ -// 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") -} diff --git a/pkg/ctl/namespace/namespace.go b/pkg/ctl/namespace/namespace.go index 587c91091..287063cc6 100644 --- a/pkg/ctl/namespace/namespace.go +++ b/pkg/ctl/namespace/namespace.go @@ -35,13 +35,6 @@ 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 } diff --git a/pkg/ctl/namespace/remove_backlog_quota.go b/pkg/ctl/namespace/remove_backlog_quota.go deleted file mode 100644 index 90ce4170d..000000000 --- a/pkg/ctl/namespace/remove_backlog_quota.go +++ /dev/null @@ -1,83 +0,0 @@ -// 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 removeBacklogQuota(vc *cmdutils.VerbCmd) { - desc := pulsar.LongDescription{} - desc.CommandUsedFor = "Remove a backlog quota policy from a namespace" - desc.CommandPermission = "This command requires tenant admin permissions." - - var examples []pulsar.Example - removeBacklog := pulsar.Example{ - Desc: "Remove a backlog quota policy from a namespace", - Command: "pulsarctl namespaces remove-backlog-quota tenant/namespace", - } - examples = append(examples, removeBacklog) - desc.CommandExamples = examples - - var out []pulsar.Output - successOut := pulsar.Output{ - Desc: "normal output", - Out: "Remove backlog quota successfully for [tenant/namespace]", - } - - 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 does not exist", - } - - out = append(out, successOut, notTenantName, notExistTenantName, notExistNsName) - desc.CommandOutput = out - - vc.SetDescription( - "remove-backlog-quota", - "Remove a backlog quota policy from a namespace", - desc.ToString(), - "remove-backlog-quota", - ) - - vc.SetRunFuncWithNameArg(func() error { - return doRemoveBacklog(vc) - }) - -} - -func doRemoveBacklog(vc *cmdutils.VerbCmd) error { - ns := vc.NameArg - admin := cmdutils.NewPulsarClient() - err := admin.Namespaces().RemoveBacklogQuota(ns) - if err == nil { - vc.Command.Printf("Remove backlog quota successfully for [%s]", ns) - } - return err -} diff --git a/pkg/ctl/namespace/retention_test.go b/pkg/ctl/namespace/retention_test.go deleted file mode 100644 index c8dc36d58..000000000 --- a/pkg/ctl/namespace/retention_test.go +++ /dev/null @@ -1,65 +0,0 @@ -// 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 TestRetention(t *testing.T) { - getArgs := []string{"get-retention", "public/default"} - getOut, execErr, _, _ := TestNamespaceCommands(getRetention, getArgs) - assert.Nil(t, execErr) - - var retention pulsar.RetentionPolicies - err := json.Unmarshal(getOut.Bytes(), &retention) - assert.Nil(t, err) - assert.Equal(t, int64(0), retention.RetentionSizeInMB) - assert.Equal(t, 0, retention.RetentionTimeInMinutes) - - setArgs := []string{"set-retention", "public/default", "--time", "10m", "--size", "10M"} - setOut, execErr, _, _ := TestNamespaceCommands(setRetention, setArgs) - assert.Nil(t, execErr) - assert.Equal(t, setOut.String(), "Set retention successfully for [public/default]") - - getArgs = []string{"get-retention", "public/default"} - getOut, execErr, _, _ = TestNamespaceCommands(getRetention, getArgs) - assert.Nil(t, execErr) - - err = json.Unmarshal(getOut.Bytes(), &retention) - assert.Nil(t, err) - assert.Equal(t, int64(10), retention.RetentionSizeInMB) - assert.Equal(t, 10, retention.RetentionTimeInMinutes) - - // test negative value for time arg - setArgWithTime := []string{"set-retention", "public/default", "--time", "-10m", "--size", "10M"} - _, execErr, _, _ = TestNamespaceCommands(setRetention, setArgWithTime) - assert.Nil(t, execErr) - - getArgs = []string{"get-retention", "public/default"} - getOut, execErr, _, _ = TestNamespaceCommands(getRetention, getArgs) - assert.Nil(t, execErr) - - err = json.Unmarshal(getOut.Bytes(), &retention) - assert.Nil(t, err) - assert.Equal(t, int64(10), retention.RetentionSizeInMB) - assert.Equal(t, -10, retention.RetentionTimeInMinutes) -} diff --git a/pkg/ctl/namespace/set_backlog_quota.go b/pkg/ctl/namespace/set_backlog_quota.go deleted file mode 100644 index 1b7020622..000000000 --- a/pkg/ctl/namespace/set_backlog_quota.go +++ /dev/null @@ -1,131 +0,0 @@ -// 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 ( - "fmt" - "github.com/spf13/cobra" - "github.com/spf13/pflag" - "github.com/streamnative/pulsarctl/pkg/cmdutils" - "github.com/streamnative/pulsarctl/pkg/pulsar" -) - -func setBacklogQuota(vc *cmdutils.VerbCmd) { - desc := pulsar.LongDescription{} - desc.CommandUsedFor = "Set a backlog quota policy for a namespace" - desc.CommandPermission = "This command requires tenant admin permissions." - - var examples []pulsar.Example - setBacklog := pulsar.Example{ - Desc: "Set a backlog quota policy for a namespace", - Command: "pulsarctl namespaces set-backlog-quota tenant/namespace \n" + - "\t--limit 2G \n" + - "\t--policy producer_request_hold", - } - examples = append(examples, setBacklog) - desc.CommandExamples = examples - - var out []pulsar.Output - successOut := pulsar.Output{ - Desc: "normal output", - Out: "Set backlog quota successfully for [tenant/namespace]", - } - - 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 does not exist", - } - - noSupportPolicyType := pulsar.Output{ - Desc: "invalid retention policy type, please check --policy arg", - Out: "invalid retention policy type: ", - } - - out = append(out, successOut, notTenantName, notExistTenantName, notExistNsName, noSupportPolicyType) - desc.CommandOutput = out - - vc.SetDescription( - "set-backlog-quota", - "Set a backlog quota policy for a namespace", - desc.ToString(), - "set-backlog-quota", - ) - - var namespaceData pulsar.NamespacesData - - vc.SetRunFuncWithNameArg(func() error { - return doSetBacklogQuota(vc, namespaceData) - }) - - vc.FlagSetGroup.InFlagSet("Namespaces", func(flagSet *pflag.FlagSet) { - flagSet.StringVarP( - &namespaceData.LimitStr, - "limit", - "l", - "", - "Size limit (eg: 10M, 16G)") - - flagSet.StringVarP( - &namespaceData.PolicyStr, - "policy", - "p", - "", - "Retention policy to enforce when the limit is reached.\n"+ - "Valid options are: [producer_request_hold, producer_exception, consumer_backlog_eviction]") - cobra.MarkFlagRequired(flagSet, "limit") - cobra.MarkFlagRequired(flagSet, "policy") - }) -} - -func doSetBacklogQuota(vc *cmdutils.VerbCmd, data pulsar.NamespacesData) error { - ns := vc.NameArg - admin := cmdutils.NewPulsarClient() - - sizeLimit, err := validateSizeString(data.LimitStr) - if err != nil { - return err - } - - var policy pulsar.RetentionPolicy - switch data.PolicyStr { - case "producer_request_hold": - policy = pulsar.ProducerRequestHold - case "producer_exception": - policy = pulsar.ProducerException - case "consumer_backlog_eviction": - policy = pulsar.ConsumerBacklogEviction - default: - return fmt.Errorf("invalid retention policy type: %v", data.PolicyStr) - } - - err = admin.Namespaces().SetBacklogQuota(ns, pulsar.NewBacklogQuota(sizeLimit, policy)) - if err == nil { - vc.Command.Printf("Set backlog quota successfully for [%s]", ns) - } - return err -} diff --git a/pkg/ctl/namespace/set_message_ttl.go b/pkg/ctl/namespace/set_message_ttl.go deleted file mode 100644 index b882f28e7..000000000 --- a/pkg/ctl/namespace/set_message_ttl.go +++ /dev/null @@ -1,100 +0,0 @@ -// 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/spf13/cobra" - "github.com/spf13/pflag" - "github.com/streamnative/pulsarctl/pkg/cmdutils" - "github.com/streamnative/pulsarctl/pkg/pulsar" -) - -func setMessageTTL(vc *cmdutils.VerbCmd) { - desc := pulsar.LongDescription{} - desc.CommandUsedFor = "Set Message TTL for a namespace" - desc.CommandPermission = "This command requires tenant admin permissions." - - var examples []pulsar.Example - setMsgTTL := pulsar.Example{ - Desc: "Set Message TTL for a namespace", - Command: "pulsarctl namespaces set-message-ttl tenant/namespace -ttl 10", - } - examples = append(examples, setMsgTTL) - desc.CommandExamples = examples - - var out []pulsar.Output - successOut := pulsar.Output{ - Desc: "normal output", - Out: "Set message TTL successfully for [tenant/namespace]", - } - - 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 does not exist", - } - - failOut := pulsar.Output{ - Desc: "Invalid value for message TTL, please check -ttl arg", - Out: "code: 412 reason: Invalid value for message TTL", - } - out = append(out, successOut, failOut, notTenantName, notExistTenantName, notExistNsName) - desc.CommandOutput = out - - vc.SetDescription( - "set-message-ttl", - "Set Message TTL for a namespace", - desc.ToString(), - "set-message-ttl", - ) - - var namespaceData pulsar.NamespacesData - - vc.SetRunFuncWithNameArg(func() error { - return doSetMessageTTL(vc, namespaceData) - }) - - vc.FlagSetGroup.InFlagSet("Namespaces", func(flagSet *pflag.FlagSet) { - flagSet.IntVarP( - &namespaceData.MessageTTL, - "messageTTL", - "t", - 0, - "Message TTL in seconds") - cobra.MarkFlagRequired(flagSet, "messageTTL") - }) -} - -func doSetMessageTTL(vc *cmdutils.VerbCmd, data pulsar.NamespacesData) error { - ns := vc.NameArg - admin := cmdutils.NewPulsarClient() - err := admin.Namespaces().SetNamespaceMessageTTL(ns, data.MessageTTL) - if err == nil { - vc.Command.Printf("Set message TTL successfully for [%s]", ns) - } - return err -} diff --git a/pkg/ctl/namespace/set_retention.go b/pkg/ctl/namespace/set_retention.go deleted file mode 100644 index b1eda802c..000000000 --- a/pkg/ctl/namespace/set_retention.go +++ /dev/null @@ -1,145 +0,0 @@ -// 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 ( - "fmt" - "github.com/spf13/cobra" - "github.com/spf13/pflag" - "github.com/streamnative/pulsarctl/pkg/cmdutils" - "github.com/streamnative/pulsarctl/pkg/pulsar" -) - -func setRetention(vc *cmdutils.VerbCmd) { - desc := pulsar.LongDescription{} - desc.CommandUsedFor = "Set the retention policy for a namespace" - desc.CommandPermission = "This command requires tenant admin permissions." - - var examples []pulsar.Example - setRetentionWithTime := pulsar.Example{ - Desc: "Set the retention policy for a namespace", - Command: "pulsarctl namespaces set-retention tenant/namespace --time 100m", - } - - setRetentionWithSize := pulsar.Example{ - Desc: "Set the retention policy for a namespace", - Command: "pulsarctl namespaces set-retention tenant/namespace --size 1G", - } - examples = append(examples, setRetentionWithTime, setRetentionWithSize) - desc.CommandExamples = examples - - var out []pulsar.Output - successOut := pulsar.Output{ - Desc: "normal output", - Out: "Set retention successfully for [tenant/namespace]", - } - - 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 does not exist", - } - - notSetBacklog := pulsar.Output{ - Desc: "Retention Quota must exceed configured backlog quota for namespace", - Out: "Retention Quota must exceed configured backlog quota for namespace", - } - - //Retention Quota must exceed configured backlog quota for namespace - - out = append(out, successOut, notTenantName, notExistTenantName, notExistNsName, notSetBacklog) - desc.CommandOutput = out - - vc.SetDescription( - "set-retention", - "Set the retention policy for a namespace", - desc.ToString(), - "set-retention", - ) - - var data pulsar.NamespacesData - - vc.SetRunFuncWithNameArg(func() error { - return doSetRetention(vc, data) - }) - - vc.FlagSetGroup.InFlagSet("Namespaces", func(flagSet *pflag.FlagSet) { - flagSet.StringVar( - &data.RetentionTimeStr, - "time", - "", - "Retention time in minutes (or minutes, hours,days,weeks eg: 100m, 3h, 2d, 5w).\n"+ - "0 means no retention and -1 means infinite time retention") - - flagSet.StringVar( - &data.LimitStr, - "size", - "", - "Retention size limit (eg: 10M, 16G, 3T).\n"+ - "0 or less than 1MB means no retention and -1 means infinite size retention") - - cobra.MarkFlagRequired(flagSet, "time") - cobra.MarkFlagRequired(flagSet, "size") - }) -} - -func doSetRetention(vc *cmdutils.VerbCmd, data pulsar.NamespacesData) error { - ns := vc.NameArg - admin := cmdutils.NewPulsarClient() - sizeLimit, err := validateSizeString(data.LimitStr) - if err != nil { - return err - } - retentionTimeInSecond, err := parseRelativeTimeInSeconds(data.RetentionTimeStr) - if err != nil { - return err - } - - var ( - retentionTimeInMin int - retentionSizeInMB int - ) - - if retentionTimeInSecond != -1 { - fmt.Println("retentionTimeInSecond: ", retentionTimeInSecond) - retentionTimeInMin = int(retentionTimeInSecond.Minutes()) - } else { - retentionTimeInMin = -1 - } - - if sizeLimit != -1 { - retentionSizeInMB = int(sizeLimit / (1024 * 1024)) - } else { - retentionSizeInMB = -1 - } - err = admin.Namespaces().SetRetention(ns, pulsar.NewRetentionPolicies(retentionTimeInMin, retentionSizeInMB)) - if err == nil { - vc.Command.Printf("Set retention successfully for [%s]", ns) - } - - return err -} diff --git a/pkg/ctl/namespace/util.go b/pkg/ctl/namespace/util.go deleted file mode 100644 index 64f0369bd..000000000 --- a/pkg/ctl/namespace/util.go +++ /dev/null @@ -1,67 +0,0 @@ -package namespace - -import ( - "github.com/pkg/errors" - "strconv" - "strings" - "time" -) - -func validateSizeString(s string) (int64, error) { - end := s[len(s)-1:] - value := s[:len(s)-1] - switch end { - case "k": - fallthrough - case "K": - v, err := strconv.ParseInt(value, 10, 64) - return v * 1024, err - case "m": - fallthrough - case "M": - v, err := strconv.ParseInt(value, 10, 64) - return v * 1024 * 1024, err - case "g": - fallthrough - case "G": - v, err := strconv.ParseInt(value, 10, 64) - return v * 1024 * 1024 * 1024, err - case "t": - fallthrough - case "T": - v, err := strconv.ParseInt(value, 10, 64) - return v * 1024 * 1024 * 1024 * 1024, err - default: - return strconv.ParseInt(s, 10, 64) - } -} - -func parseRelativeTimeInSeconds(relativeTime string) (time.Duration, error) { - if relativeTime == "" { - return -1, errors.New("Time can not be empty.") - } - - unitTime := relativeTime[len(relativeTime)-1:] - t := relativeTime[:len(relativeTime)-1] - timeValue, err := strconv.ParseInt(t, 10, 64) - if err != nil { - return -1, errors.Errorf("Invalid time '%s'", t) - } - - switch strings.ToLower(unitTime) { - case "s": - return time.Duration(timeValue) * time.Second, nil - case "m": - return time.Duration(timeValue) * time.Minute, nil - case "h": - return time.Duration(timeValue) * time.Hour, nil - case "d": - return time.Duration(timeValue) * time.Hour * 24, nil - case "w": - return time.Duration(timeValue) * time.Hour * 24 * 7, nil - case "y": - return time.Duration(timeValue) * time.Hour * 24 * 7 * 365, nil - default: - return -1, errors.Errorf("Invalid time unit '%s'", unitTime) - } -} diff --git a/pkg/pulsar/backlog_quota.go b/pkg/pulsar/backlog_quota.go index 8d83b7871..712e947a3 100644 --- a/pkg/pulsar/backlog_quota.go +++ b/pkg/pulsar/backlog_quota.go @@ -18,25 +18,33 @@ package pulsar type BacklogQuota struct { - Limit int64 `json:"limit"` - Policy RetentionPolicy `json:"policy"` + Limit int64 + Police RetentionPolicy + BacklogQuotaType BacklogQuotaType } -func NewBacklogQuota(limit int64, policy RetentionPolicy) BacklogQuota { - return BacklogQuota{ - Limit: limit, - Policy: policy, - } -} - -type RetentionPolicy string +type RetentionPolicy int type BacklogQuotaType string const DestinationStorage BacklogQuotaType = "destination_storage" const ( - ProducerRequestHold RetentionPolicy = "producer_request_hold" - ProducerException RetentionPolicy = "producer_exception" - ConsumerBacklogEviction RetentionPolicy = "consumer_backlog_eviction" + ProducerRequestHold RetentionPolicy = iota + ProducerException + ConsumerBacklogEviction ) + +func (rp RetentionPolicy) String() string { + names := [...]string{ + "ProducerRequestHold", + "ProducerException", + "ConsumerBacklogEviction", + } + + if rp < ProducerRequestHold || rp > ConsumerBacklogEviction { + return "Unknown Retention Policy" + } + + return names[rp] +} diff --git a/pkg/pulsar/data.go b/pkg/pulsar/data.go index 083f88b71..c1453e812 100644 --- a/pkg/pulsar/data.go +++ b/pkg/pulsar/data.go @@ -152,12 +152,8 @@ type PartitionedTopicMetadata struct { } type NamespacesData struct { - NumBundles int `json:"numBundles"` - Clusters []string `json:"clusters"` - MessageTTL int `json:"messageTTL"` - RetentionTimeStr string `json:"retentionTimeStr"` - LimitStr string `json:"limitStr"` - PolicyStr string `json:"policyStr"` + NumBundles int `json:"numBundles"` + Clusters []string `json:"clusters"` } type SchemaData struct { diff --git a/pkg/pulsar/namespace.go b/pkg/pulsar/namespace.go index 15f10d93b..f6c9b432c 100644 --- a/pkg/pulsar/namespace.go +++ b/pkg/pulsar/namespace.go @@ -44,27 +44,6 @@ type Namespaces interface { // Delete an existing bundle in a namespace DeleteNamespaceBundle(namespace string, bundleRange string) error - - // Set the messages Time to Live for all the topics within a namespace - SetNamespaceMessageTTL(namespace string, ttlInSeconds int) error - - // Get the message TTL for a namespace - GetNamespaceMessageTTL(namespace string) (int, error) - - // Get the retention configuration for a namespace - GetRetention(namespace string) (*RetentionPolicies, error) - - // Set the retention configuration for all the topics on a namespace - SetRetention(namespace string, policy RetentionPolicies) error - - // Get backlog quota map on a namespace - GetBacklogQuotaMap(namespace string) (map[BacklogQuotaType]BacklogQuota, error) - - // Set a backlog quota for all the topics on a namespace - SetBacklogQuota(namespace string, backlogQuota BacklogQuota) error - - // Remove a backlog quota policy from a namespace - RemoveBacklogQuota(namespace string) error } type namespaces struct { @@ -159,76 +138,3 @@ func (n *namespaces) DeleteNamespaceBundle(namespace string, bundleRange string) endpoint := n.client.endpoint(n.basePath, ns.String(), bundleRange) return n.client.delete(endpoint, nil) } - -func (n *namespaces) GetNamespaceMessageTTL(namespace string) (int, error) { - var ttl int - nsName, err := GetNamespaceName(namespace) - if err != nil { - return 0, err - } - endpoint := n.client.endpoint(n.basePath, nsName.String(), "messageTTL") - err = n.client.get(endpoint, &ttl) - return ttl, err -} - -func (n *namespaces) SetNamespaceMessageTTL(namespace string, ttlInSeconds int) error { - nsName, err := GetNamespaceName(namespace) - if err != nil { - return err - } - - endpoint := n.client.endpoint(n.basePath, nsName.String(), "messageTTL") - return n.client.post(endpoint, &ttlInSeconds, nil) -} - -func (n *namespaces) SetRetention(namespace string, policy RetentionPolicies) error { - nsName, err := GetNamespaceName(namespace) - if err != nil { - return err - } - endpoint := n.client.endpoint(n.basePath, nsName.String(), "retention") - return n.client.post(endpoint, &policy, nil) -} - -func (n *namespaces) GetRetention(namespace string) (*RetentionPolicies, error) { - var policy RetentionPolicies - nsName, err := GetNamespaceName(namespace) - if err != nil { - return nil, err - } - endpoint := n.client.endpoint(n.basePath, nsName.String(), "retention") - err = n.client.get(endpoint, &policy) - return &policy, err -} - -func (n *namespaces) GetBacklogQuotaMap(namespace string) (map[BacklogQuotaType]BacklogQuota, error) { - var backlogQuotaMap map[BacklogQuotaType]BacklogQuota - nsName, err := GetNamespaceName(namespace) - if err != nil { - return nil, err - } - endpoint := n.client.endpoint(n.basePath, nsName.String(), "backlogQuotaMap") - err = n.client.get(endpoint, &backlogQuotaMap) - return backlogQuotaMap, err -} - -func (n *namespaces) SetBacklogQuota(namespace string, backlogQuota BacklogQuota) error { - nsName, err := GetNamespaceName(namespace) - if err != nil { - return err - } - endpoint := n.client.endpoint(n.basePath, nsName.String(), "backlogQuota") - return n.client.post(endpoint, &backlogQuota, nil) -} - -func (n *namespaces) RemoveBacklogQuota(namespace string) error { - nsName, err := GetNamespaceName(namespace) - if err != nil { - return err - } - endpoint := n.client.endpoint(n.basePath, nsName.String(), "backlogQuota") - params := map[string]string{ - "backlogQuotaType": string(DestinationStorage), - } - return n.client.deleteWithQueryParams(endpoint, nil, params) -} diff --git a/pkg/pulsar/retention_policies.go b/pkg/pulsar/retention_policies.go index 23e3d7c73..c25214c08 100644 --- a/pkg/pulsar/retention_policies.go +++ b/pkg/pulsar/retention_policies.go @@ -18,13 +18,6 @@ package pulsar type RetentionPolicies struct { - RetentionTimeInMinutes int `json:"retentionTimeInMinutes"` - RetentionSizeInMB int64 `json:"retentionSizeInMB"` -} - -func NewRetentionPolicies(retentionTimeInMinutes int, retentionSizeInMB int) RetentionPolicies { - return RetentionPolicies{ - RetentionTimeInMinutes: retentionTimeInMinutes, - RetentionSizeInMB: int64(retentionSizeInMB), - } + RetentionTimeInMinutes int + RetentionSizeInMB int64 }