diff --git a/pkg/ctl/namespace/backlog_quota_test.go b/pkg/ctl/namespace/backlog_quota_test.go new file mode 100644 index 000000000..8ebbbf629 --- /dev/null +++ b/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") +} diff --git a/pkg/ctl/namespace/get_backlog_quota.go b/pkg/ctl/namespace/get_backlog_quota.go new file mode 100644 index 000000000..3bd9e9326 --- /dev/null +++ b/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 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 new file mode 100644 index 000000000..da9a856f4 --- /dev/null +++ b/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: "", + } + + 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 new file mode 100644 index 000000000..a46f50827 --- /dev/null +++ b/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 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 new file mode 100644 index 000000000..0d9552e5b --- /dev/null +++ b/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") +} diff --git a/pkg/ctl/namespace/namespace.go b/pkg/ctl/namespace/namespace.go index 287063cc6..587c91091 100644 --- a/pkg/ctl/namespace/namespace.go +++ b/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 } diff --git a/pkg/ctl/namespace/remove_backlog_quota.go b/pkg/ctl/namespace/remove_backlog_quota.go new file mode 100644 index 000000000..90ce4170d --- /dev/null +++ b/pkg/ctl/namespace/remove_backlog_quota.go @@ -0,0 +1,83 @@ +// 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 new file mode 100644 index 000000000..c8dc36d58 --- /dev/null +++ b/pkg/ctl/namespace/retention_test.go @@ -0,0 +1,65 @@ +// 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 new file mode 100644 index 000000000..1b7020622 --- /dev/null +++ b/pkg/ctl/namespace/set_backlog_quota.go @@ -0,0 +1,131 @@ +// 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 new file mode 100644 index 000000000..b882f28e7 --- /dev/null +++ b/pkg/ctl/namespace/set_message_ttl.go @@ -0,0 +1,100 @@ +// 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 new file mode 100644 index 000000000..b1eda802c --- /dev/null +++ b/pkg/ctl/namespace/set_retention.go @@ -0,0 +1,145 @@ +// 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 new file mode 100644 index 000000000..64f0369bd --- /dev/null +++ b/pkg/ctl/namespace/util.go @@ -0,0 +1,67 @@ +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 712e947a3..8d83b7871 100644 --- a/pkg/pulsar/backlog_quota.go +++ b/pkg/pulsar/backlog_quota.go @@ -18,33 +18,25 @@ package pulsar type BacklogQuota struct { - Limit int64 - Police RetentionPolicy - BacklogQuotaType BacklogQuotaType + Limit int64 `json:"limit"` + Policy RetentionPolicy `json:"policy"` } -type RetentionPolicy int +func NewBacklogQuota(limit int64, policy RetentionPolicy) BacklogQuota { + return BacklogQuota{ + Limit: limit, + Policy: policy, + } +} + +type RetentionPolicy string type BacklogQuotaType string const DestinationStorage BacklogQuotaType = "destination_storage" const ( - ProducerRequestHold RetentionPolicy = iota - ProducerException - ConsumerBacklogEviction + ProducerRequestHold RetentionPolicy = "producer_request_hold" + ProducerException RetentionPolicy = "producer_exception" + ConsumerBacklogEviction RetentionPolicy = "consumer_backlog_eviction" ) - -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 c1453e812..083f88b71 100644 --- a/pkg/pulsar/data.go +++ b/pkg/pulsar/data.go @@ -152,8 +152,12 @@ type PartitionedTopicMetadata struct { } type NamespacesData struct { - NumBundles int `json:"numBundles"` - Clusters []string `json:"clusters"` + NumBundles int `json:"numBundles"` + Clusters []string `json:"clusters"` + MessageTTL int `json:"messageTTL"` + RetentionTimeStr string `json:"retentionTimeStr"` + LimitStr string `json:"limitStr"` + PolicyStr string `json:"policyStr"` } type SchemaData struct { diff --git a/pkg/pulsar/namespace.go b/pkg/pulsar/namespace.go index f6c9b432c..15f10d93b 100644 --- a/pkg/pulsar/namespace.go +++ b/pkg/pulsar/namespace.go @@ -44,6 +44,27 @@ 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 { @@ -138,3 +159,76 @@ 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 c25214c08..23e3d7c73 100644 --- a/pkg/pulsar/retention_policies.go +++ b/pkg/pulsar/retention_policies.go @@ -18,6 +18,13 @@ package pulsar type RetentionPolicies struct { - RetentionTimeInMinutes int - RetentionSizeInMB int64 + RetentionTimeInMinutes int `json:"retentionTimeInMinutes"` + RetentionSizeInMB int64 `json:"retentionSizeInMB"` +} + +func NewRetentionPolicies(retentionTimeInMinutes int, retentionSizeInMB int) RetentionPolicies { + return RetentionPolicies{ + RetentionTimeInMinutes: retentionTimeInMinutes, + RetentionSizeInMB: int64(retentionSizeInMB), + } }