From 794bdda4562fda0edbcea9860ad11c5814c7df6b Mon Sep 17 00:00:00 2001 From: "xiaolong.ran" Date: Sat, 14 Sep 2019 17:10:25 +0800 Subject: [PATCH 1/4] Add create, delete, list, police, topic commands for namespace ctl Signed-off-by: xiaolong.ran --- main.go | 8 +- pkg/ctl/namespace/create.go | 111 +++++++++++++++++++++++ pkg/ctl/namespace/create_test.go | 55 +++++++++++ pkg/ctl/namespace/delete.go | 77 ++++++++++++++++ pkg/ctl/namespace/delete_test.go | 52 +++++++++++ pkg/ctl/namespace/list.go | 90 ++++++++++++++++++ pkg/ctl/namespace/namespace.go | 40 ++++++++ pkg/ctl/namespace/police.go | 141 +++++++++++++++++++++++++++++ pkg/ctl/namespace/police_test.go | 66 ++++++++++++++ pkg/ctl/namespace/test_help.go | 65 +++++++++++++ pkg/ctl/namespace/topics.go | 92 +++++++++++++++++++ pkg/ctl/namespace/topics_test.go | 50 ++++++++++ pkg/pulsar/admin.go | 15 +-- pkg/pulsar/auth_polices.go | 21 +++++ pkg/pulsar/backlog_quota.go | 50 ++++++++++ pkg/pulsar/bundles_data.go | 37 ++++++++ pkg/pulsar/data.go | 4 + pkg/pulsar/dispatch_rate.go | 44 +++++++++ pkg/pulsar/namespace.go | 140 ++++++++++++++++++++++++++++ pkg/pulsar/persistence_policies.go | 34 +++++++ pkg/pulsar/polices.go | 58 ++++++++++++ pkg/pulsar/retention_policies.go | 23 +++++ pkg/pulsar/schema_strategy.go | 31 +++++++ 23 files changed, 1294 insertions(+), 10 deletions(-) create mode 100644 pkg/ctl/namespace/create.go create mode 100644 pkg/ctl/namespace/create_test.go create mode 100644 pkg/ctl/namespace/delete.go create mode 100644 pkg/ctl/namespace/delete_test.go create mode 100644 pkg/ctl/namespace/list.go create mode 100644 pkg/ctl/namespace/namespace.go create mode 100644 pkg/ctl/namespace/police.go create mode 100644 pkg/ctl/namespace/police_test.go create mode 100644 pkg/ctl/namespace/test_help.go create mode 100644 pkg/ctl/namespace/topics.go create mode 100644 pkg/ctl/namespace/topics_test.go create mode 100644 pkg/pulsar/auth_polices.go create mode 100644 pkg/pulsar/backlog_quota.go create mode 100644 pkg/pulsar/bundles_data.go create mode 100644 pkg/pulsar/dispatch_rate.go create mode 100644 pkg/pulsar/namespace.go create mode 100644 pkg/pulsar/persistence_policies.go create mode 100644 pkg/pulsar/polices.go create mode 100644 pkg/pulsar/retention_policies.go create mode 100644 pkg/pulsar/schema_strategy.go diff --git a/main.go b/main.go index 5488ebef3..f8b9c250f 100644 --- a/main.go +++ b/main.go @@ -7,10 +7,11 @@ import ( "github.com/streamnative/pulsarctl/pkg/cmdutils" "github.com/streamnative/pulsarctl/pkg/ctl/cluster" "github.com/streamnative/pulsarctl/pkg/ctl/completion" - "github.com/streamnative/pulsarctl/pkg/ctl/topic" "github.com/streamnative/pulsarctl/pkg/ctl/functions" - "github.com/streamnative/pulsarctl/pkg/ctl/sources" - "github.com/streamnative/pulsarctl/pkg/ctl/tenant" + "github.com/streamnative/pulsarctl/pkg/ctl/namespace" + "github.com/streamnative/pulsarctl/pkg/ctl/sources" + "github.com/streamnative/pulsarctl/pkg/ctl/tenant" + "github.com/streamnative/pulsarctl/pkg/ctl/topic" "os" ) @@ -69,6 +70,7 @@ func addCommands(flagGrouping *cmdutils.FlagGrouping) { rootCmd.AddCommand(functions.Command(flagGrouping)) rootCmd.AddCommand(sources.Command(flagGrouping)) rootCmd.AddCommand(topic.Command(flagGrouping)) + rootCmd.AddCommand(namespace.Command(flagGrouping)) } func main() { diff --git a/pkg/ctl/namespace/create.go b/pkg/ctl/namespace/create.go new file mode 100644 index 000000000..360758c19 --- /dev/null +++ b/pkg/ctl/namespace/create.go @@ -0,0 +1,111 @@ +// 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/pkg/errors" + "github.com/spf13/pflag" + "github.com/streamnative/pulsarctl/pkg/cmdutils" + "github.com/streamnative/pulsarctl/pkg/pulsar" +) + +const MaxBundles = int64(1) << 32 + +func createNs(vc *cmdutils.VerbCmd) { + desc := pulsar.LongDescription{} + desc.CommandUsedFor = "Creates a new namespace" + desc.CommandPermission = "This command requires namespace admin permissions." + + var examples []pulsar.Example + create := pulsar.Example{ + Desc: "create a namespace named ", + Command: "pulsarctl namespaces create ", + } + examples = append(examples, create) + desc.CommandExamples = examples + + var out []pulsar.Output + successOut := pulsar.Output{ + Desc: "normal output", + Out: "Created successfully", + } + + 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, notExistTenantName, notTenantName, notExistNsName) + desc.CommandOutput = out + + vc.SetDescription( + "create", + "Create a new namespace", + desc.ToString(), + "create", + ) + + var namespaceData pulsar.NamespacesData + + vc.SetRunFuncWithNameArg(func() error { + return doCreate(vc, namespaceData) + }) + + vc.FlagSetGroup.InFlagSet("Namespaces", func(flagSet *pflag.FlagSet) { + flagSet.IntVarP( + &namespaceData.NumBundles, + "bundles", + "b", + 0, + "number of bundles to activate") + }) +} + +func doCreate(vc *cmdutils.VerbCmd, data pulsar.NamespacesData) error { + tenantAndNamespace := vc.NameArg + admin := cmdutils.NewPulsarClient() + + if data.NumBundles < 0 || data.NumBundles > int(MaxBundles) { + return errors.New("Invalid number of bundles. Number of numBundles has to be in the range of (0, 2^32].") + } + ns, err := pulsar.GetNamespaceName(tenantAndNamespace) + if err != nil { + return err + } + polices := new(pulsar.Polices) + if data.NumBundles > 0 { + polices.Bundles = pulsar.NewBundlesDataWithNumBundles(data.NumBundles) + } else { + polices.Bundles = nil + } + err = admin.Namespaces().CreateNsWithPolices(ns.String(), *polices) + if err == nil { + vc.Command.Printf("Created %s successfully", ns.String()) + } + return err +} diff --git a/pkg/ctl/namespace/create_test.go b/pkg/ctl/namespace/create_test.go new file mode 100644 index 000000000..b85a68690 --- /dev/null +++ b/pkg/ctl/namespace/create_test.go @@ -0,0 +1,55 @@ +// 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/stretchr/testify/assert" + "strings" + "testing" +) + +func TestCreateNs(t *testing.T) { + args := []string{"create", "public/test-namespace"} + createOut, _, _, err := TestNamespaceCommands(createNs, args) + assert.Nil(t, err) + assert.Equal(t, createOut.String(), "Created public/test-namespace successfully") + + args = []string{"list", "public"} + out, _, _, _ := TestNamespaceCommands(getNamespacesPerProperty, args) + fmt.Println(out.String()) + assert.True(t, strings.Contains(out.String(), "public/test-namespace")) +} + +func TestCreateNsArgsError(t *testing.T) { + args := []string{"create"} + _, _, nameErr, _ := TestNamespaceCommands(createNs, args) + + assert.Equal(t, "only one argument is allowed to be used as a name", nameErr.Error()) +} + +func TestCreateNsAlreadyExistError(t *testing.T) { + args := []string{"create", "public/test-ns-duplicate"} + _, execErr, _, _ := TestNamespaceCommands(createNs, args) + assert.Nil(t, execErr) + + args = []string{"create", "public/test-ns-duplicate"} + _, execErr, _, _ = TestNamespaceCommands(createNs, args) + assert.NotNil(t, execErr) + assert.Equal(t, "code: 409 reason: Namespace already exists", execErr.Error()) +} diff --git a/pkg/ctl/namespace/delete.go b/pkg/ctl/namespace/delete.go new file mode 100644 index 000000000..e6f6d5f17 --- /dev/null +++ b/pkg/ctl/namespace/delete.go @@ -0,0 +1,77 @@ +// 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 deleteNs(vc *cmdutils.VerbCmd) { + desc := pulsar.LongDescription{} + desc.CommandUsedFor = "Deletes a namespace. The namespace needs to be empty" + desc.CommandPermission = "This command requires namespace admin permissions." + + var examples []pulsar.Example + del := pulsar.Example{ + Desc: "Deletes a namespace", + Command: "pulsarctl namespaces delete ", + } + examples = append(examples, del) + desc.CommandExamples = examples + + var out []pulsar.Output + successOut := pulsar.Output{ + Desc: "normal output", + Out: "Created successfully", + } + + 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", + } + + out = append(out, successOut, notExistTenantName, notTenantName) + desc.CommandOutput = out + + vc.SetDescription( + "delete", + "Deletes a namespace. The namespace needs to be empty", + desc.ToString(), + "delete", + ) + + vc.SetRunFuncWithNameArg(func() error { + return doDeleteNs(vc) + }) +} + +func doDeleteNs(vc *cmdutils.VerbCmd) error { + ns := vc.NameArg + admin := cmdutils.NewPulsarClient() + err := admin.Namespaces().DeleteNamespace(ns) + if err == nil { + vc.Command.Printf("Deleted %s successfully", ns) + } + return err +} diff --git a/pkg/ctl/namespace/delete_test.go b/pkg/ctl/namespace/delete_test.go new file mode 100644 index 000000000..1e7e27e4b --- /dev/null +++ b/pkg/ctl/namespace/delete_test.go @@ -0,0 +1,52 @@ +// 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" + "strings" + "testing" +) + +func TestDeleteNsCmd(t *testing.T) { + args := []string{"create", "public/test-delete-namespace"} + createOut, _, _, err := TestNamespaceCommands(createNs, args) + assert.Nil(t, err) + assert.Equal(t, createOut.String(), "Created public/test-delete-namespace successfully") + + args = []string{"delete", "public/test-delete-namespace"} + delOut, _, _, _ := TestNamespaceCommands(deleteNs, args) + assert.Equal(t, delOut.String(), "Deleted public/test-delete-namespace successfully") + + args = []string{"list", "public"} + listOut, _, _, _ := TestNamespaceCommands(getNamespacesPerProperty, args) + assert.False(t, strings.Contains(listOut.String(), "public/test-delete-namespace")) +} + +func TestDeleteNsArgsError(t *testing.T) { + args := []string{"delete"} + _, _, nameErr, _ := TestNamespaceCommands(deleteNs, args) + assert.Equal(t, "only one argument is allowed to be used as a name", nameErr.Error()) +} + +func TestDeleteNonExistTenant(t *testing.T) { + args := []string{"delete", "non-existent-tenant/test-delete-namespace"} + _, execErr, _, _ := TestNamespaceCommands(deleteNs, args) + assert.NotNil(t, execErr) + assert.Equal(t, "code: 404 reason: Tenant does not exist", execErr.Error()) +} diff --git a/pkg/ctl/namespace/list.go b/pkg/ctl/namespace/list.go new file mode 100644 index 000000000..f534746ea --- /dev/null +++ b/pkg/ctl/namespace/list.go @@ -0,0 +1,90 @@ +// 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/olekukonko/tablewriter" + "github.com/streamnative/pulsarctl/pkg/cmdutils" + "github.com/streamnative/pulsarctl/pkg/pulsar" +) + +func getNamespacesPerProperty(vc *cmdutils.VerbCmd) { + desc := pulsar.LongDescription{} + desc.CommandUsedFor = "Get the namespaces for a tenant" + desc.CommandPermission = "This command requires namespace admin permissions." + + var examples []pulsar.Example + + list := pulsar.Example{ + Desc: "Get the namespaces for a tenant", + Command: "pulsarctl namespaces list ", + } + + examples = append(examples, list) + desc.CommandExamples = examples + + var out []pulsar.Output + successOut := pulsar.Output{ + Desc: "normal output", + Out: "+------------------+\n" + + "| NAMESPACE NAME |\n" + + "+------------------+\n" + + "| public/default |\n" + + "| public/functions |\n" + + "+------------------+", + } + + notTenantName := pulsar.Output{ + Desc: "you must specify a tenant name, please check if the tenant 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 if tenant name exists", + Out: "[✖] code: 404 reason: Tenant does not exist", + } + + out = append(out, successOut, notTenantName, notExistTenantName) + desc.CommandOutput = out + + vc.SetDescription( + "list", + "Get the namespaces for a tenant", + desc.ToString(), + "list", + ) + + vc.SetRunFuncWithNameArg(func() error { + return doListNamespaces(vc) + }) +} + +func doListNamespaces(vc *cmdutils.VerbCmd) error { + tenant := vc.NameArg + admin := cmdutils.NewPulsarClient() + listNamespaces, err := admin.Namespaces().GetNamespaces(tenant) + if err == nil { + table := tablewriter.NewWriter(vc.Command.OutOrStdout()) + table.SetHeader([]string{"Namespace Name"}) + for _, ns := range listNamespaces { + table.Append([]string{ns}) + } + table.Render() + } + return err +} diff --git a/pkg/ctl/namespace/namespace.go b/pkg/ctl/namespace/namespace.go new file mode 100644 index 000000000..f8805808f --- /dev/null +++ b/pkg/ctl/namespace/namespace.go @@ -0,0 +1,40 @@ +// 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/streamnative/pulsarctl/pkg/cmdutils" +) + +func Command(flagGrouping *cmdutils.FlagGrouping) *cobra.Command { + resourceCmd := cmdutils.NewResourceCmd( + "namespaces", + "Operations about namespaces", + "", + "namespace", + ) + + cmdutils.AddVerbCmd(flagGrouping, resourceCmd, getNamespacesPerProperty) + cmdutils.AddVerbCmd(flagGrouping, resourceCmd, getTopics) + cmdutils.AddVerbCmd(flagGrouping, resourceCmd, getPolicies) + cmdutils.AddVerbCmd(flagGrouping, resourceCmd, createNs) + cmdutils.AddVerbCmd(flagGrouping, resourceCmd, deleteNs) + + return resourceCmd +} diff --git a/pkg/ctl/namespace/police.go b/pkg/ctl/namespace/police.go new file mode 100644 index 000000000..bccc0c571 --- /dev/null +++ b/pkg/ctl/namespace/police.go @@ -0,0 +1,141 @@ +// 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 getPolicies(vc *cmdutils.VerbCmd) { + desc := pulsar.LongDescription{} + desc.CommandUsedFor = "Get the configuration policies of a namespace" + desc.CommandPermission = "This command requires namespace admin permissions." + + var examples []pulsar.Example + police := pulsar.Example{ + Desc: "Get the configuration policies of a namespace", + Command: "pulsarctl namespaces polices ", + } + examples = append(examples, police) + desc.CommandExamples = examples + + var out []pulsar.Output + successOut := pulsar.Output{ + Desc: "normal output", + Out: "{\n" + + " \"AuthPolicies\": {},\n" + + " \"ReplicationClusters\": null,\n" + + " \"Bundles\": {\n" + + " \"boundaries\": [\n" + + " \"0x00000000\",\n" + + " \"0x40000000\",\n" + + " \"0x80000000\",\n" + + " \"0xc0000000\",\n" + + " \"0xffffffff\"\n" + + " ],\n" + + " \"numBundles\": 4\n" + + " },\n" + + " \"BacklogQuotaMap\": null,\n" + + " \"TopicDispatchRate\": {\n" + + " \"standalone\": {\n" + + " \"DispatchThrottlingRateInMsg\": 0,\n" + + " \"DispatchThrottlingRateInByte\": 0,\n" + + " \"RatePeriodInSecond\": 1\n" + + " }\n" + + " },\n" + + " \"SubscriptionDispatchRate\": {\n" + + " \"standalone\": {\n" + + " \"DispatchThrottlingRateInMsg\": 0,\n" + + " \"DispatchThrottlingRateInByte\": 0,\n" + + " \"RatePeriodInSecond\": 1\n" + + " }\n" + + " },\n" + + " \"ClusterSubscribeRate\": {\n" + + " \"standalone\": {\n" + + " \"SubscribeThrottlingRatePerConsumer\": 0,\n" + + " \"RatePeriodInSecond\": 30\n" + + " }\n" + + " },\n" + + " \"Persistence\": {\n" + + " \"BookkeeperEnsemble\": 0,\n" + + " \"BookkeeperWriteQuorum\": 0,\n" + + " \"BookkeeperAckQuorum\": 0,\n" + + " \"ManagedLedgerMaxMarkDeleteRate\": 0\n" + + " },\n" + + " \"DeduplicationEnabled\": false,\n" + + " \"LatencyStatsSampleRate\": null,\n" + + " \"MessageTtlInSeconds\": 0,\n" + + " \"RetentionPolicies\": {\n" + + " \"RetentionTimeInMinutes\": 0,\n" + + " \"RetentionSizeInMB\": 0\n" + + " },\n" + + " \"Deleted\": false,\n" + + " \"AntiAffinityGroup\": \"\",\n" + + " \"EncryptionRequired\": false,\n" + + " \"SubscriptionAuthMode\": \"\",\n" + + " \"MaxProducersPerTopic\": 0,\n" + + " \"MaxConsumersPerTopic\": 0,\n" + + " \"MaxConsumersPerSubscription\": 0,\n" + + " \"CompactionThreshold\": 0,\n" + + " \"OffloadThreshold\": 0,\n" + + " \"OffloadDeletionLagMs\": 0,\n" + + " \"SchemaAutoUpdateCompatibilityStrategy\": \"\",\n" + + " \"SchemaValidationEnforced\": false\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, notExistTenantName, notTenantName, notExistNsName) + desc.CommandOutput = out + + vc.SetDescription( + "polices", + "Get the configuration policies of a namespace", + desc.ToString(), + "police", + ) + + vc.SetRunFuncWithNameArg(func() error { + return doGetPolices(vc) + }) +} + +func doGetPolices(vc *cmdutils.VerbCmd) error { + namespace := vc.NameArg + admin := cmdutils.NewPulsarClient() + police, err := admin.Namespaces().GetPolicies(namespace) + if err == nil { + cmdutils.PrintJson(vc.Command.OutOrStdout(), police) + } + return err +} diff --git a/pkg/ctl/namespace/police_test.go b/pkg/ctl/namespace/police_test.go new file mode 100644 index 000000000..ce4535c8a --- /dev/null +++ b/pkg/ctl/namespace/police_test.go @@ -0,0 +1,66 @@ +// 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 TestPolicesCommand(t *testing.T) { + args := []string{"polices", "public/default"} + out, execErr, _, _ := TestNamespaceCommands(getPolicies, args) + assert.Nil(t, execErr) + + var police pulsar.Polices + err := json.Unmarshal(out.Bytes(), &police) + assert.Nil(t, err) + + assert.Equal(t, police.DeduplicationEnabled, false) + assert.Equal(t, police.Deleted, false) + for key, value := range police.ClusterSubscribeRate { + exceptedValue := pulsar.SubscribeRate{ + SubscribeThrottlingRatePerConsumer: 0, + RatePeriodInSecond: 30, + } + assert.Equal(t, key, "standalone") + assert.Equal(t, exceptedValue, value) + } +} + +func TestPolicesNsArgsError(t *testing.T) { + args := []string{"polices"} + _, _, nameErr, _ := TestNamespaceCommands(getPolicies, args) + assert.Equal(t, "only one argument is allowed to be used as a name", nameErr.Error()) +} + +func TestPolicesNonExistTenant(t *testing.T) { + args := []string{"polices", "non-existent-tenant/default"} + _, execErr, _, _ := TestNamespaceCommands(getPolicies, args) + assert.NotNil(t, execErr) + assert.Equal(t, "code: 404 reason: Tenant does not exist", execErr.Error()) +} + +func TestPolicesNonExistNs(t *testing.T) { + args := []string{"polices", "public/test-not-exist-ns"} + _, execErr, _, _ := TestNamespaceCommands(getPolicies, args) + assert.NotNil(t, execErr) + assert.Equal(t, "code: 404 reason: Namespace does not exist", execErr.Error()) +} diff --git a/pkg/ctl/namespace/test_help.go b/pkg/ctl/namespace/test_help.go new file mode 100644 index 000000000..775bbb9ab --- /dev/null +++ b/pkg/ctl/namespace/test_help.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 ( + "bytes" + "github.com/kris-nova/logger" + "github.com/spf13/cobra" + "github.com/streamnative/pulsarctl/pkg/cmdutils" +) + +func TestNamespaceCommands(newVerb func(cmd *cmdutils.VerbCmd), args []string) (out *bytes.Buffer, execErr, nameErr, err error) { + var execError error + cmdutils.ExecErrorHandler = func(err error) { + execError = err + } + + var nameError error + cmdutils.CheckNameArgError = func(err error) { + nameError = err + + } + + var rootCmd = &cobra.Command{ + Use: "pulsarctl [command]", + Short: "a CLI for Apache Pulsar", + Run: func(cmd *cobra.Command, _ []string) { + if err := cmd.Help(); err != nil { + logger.Debug("ignoring error %q", err.Error()) + } + }, + } + + buf := new(bytes.Buffer) + rootCmd.SetOut(buf) + rootCmd.SetArgs(append([]string{"namespaces"}, args...)) + + resourceCmd := cmdutils.NewResourceCmd( + "namespaces", + "Operations about namespace(s)", + "", + "namespace") + flagGrouping := cmdutils.NewGrouping() + cmdutils.AddVerbCmd(flagGrouping, resourceCmd, newVerb) + rootCmd.AddCommand(resourceCmd) + err = rootCmd.Execute() + + return buf, execError, nameError, err + +} diff --git a/pkg/ctl/namespace/topics.go b/pkg/ctl/namespace/topics.go new file mode 100644 index 000000000..dde6f1fe9 --- /dev/null +++ b/pkg/ctl/namespace/topics.go @@ -0,0 +1,92 @@ +// 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/olekukonko/tablewriter" + "github.com/streamnative/pulsarctl/pkg/cmdutils" + "github.com/streamnative/pulsarctl/pkg/pulsar" +) + +func getTopics(vc *cmdutils.VerbCmd) { + desc := pulsar.LongDescription{} + desc.CommandUsedFor = "Get the list of topics for a namespace" + desc.CommandPermission = "This command requires namespace admin permissions." + + var examples []pulsar.Example + + topics := pulsar.Example{ + Desc: "Get the list of topics for a namespace", + Command: "pulsarctl namespaces topics ", + } + + examples = append(examples, topics) + desc.CommandExamples = examples + + var out []pulsar.Output + successOut := pulsar.Output{ + Desc: "normal output", + Out: "+-------------+\n" + + "| TOPICS NAME |\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 name not exist, please check the namespace name", + Out: "[✖] code: 404 reason: Namespace does not exist", + } + + out = append(out, successOut, notTenantName, notExistNsName, notExistTenantName) + + vc.SetDescription( + "topics", + "Get the list of topics for a namespace", + desc.ToString(), + "topics", + ) + + vc.SetRunFuncWithNameArg(func() error { + return doListTopics(vc) + }) +} + +func doListTopics(vc *cmdutils.VerbCmd) error { + tenantAndNamespace := vc.NameArg + admin := cmdutils.NewPulsarClient() + listTopics, err := admin.Namespaces().GetTopics(tenantAndNamespace) + if err == nil { + table := tablewriter.NewWriter(vc.Command.OutOrStdout()) + table.SetHeader([]string{"Topics Name"}) + for _, topic := range listTopics { + table.Append([]string{topic}) + } + table.Render() + } + return err +} diff --git a/pkg/ctl/namespace/topics_test.go b/pkg/ctl/namespace/topics_test.go new file mode 100644 index 000000000..d88c93008 --- /dev/null +++ b/pkg/ctl/namespace/topics_test.go @@ -0,0 +1,50 @@ +// 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 TestListNsTopicsCmd(t *testing.T) { + args := []string{"topics", "public/default"} + _, execErr, _, _ := TestNamespaceCommands(getTopics, args) + assert.Nil(t, execErr) +} + +func TestListTopicArgError(t *testing.T) { + args := []string{"topics"} + _, _, nameErr, _ := TestNamespaceCommands(getTopics, args) + assert.NotNil(t, nameErr) + assert.Equal(t, "only one argument is allowed to be used as a name", nameErr.Error()) +} + +func TestListNonExistNamespace(t *testing.T) { + args := []string{"topics", "public/non-exist-namespace"} + _, execErr, _, _ := TestNamespaceCommands(getTopics, args) + assert.NotNil(t, execErr) + assert.Equal(t, "code: 404 reason: Namespace does not exist", execErr.Error()) +} + +func TestListNonExistTenant(t *testing.T) { + args := []string{"topics", "non-exist-tenant/default"} + _, execErr, _, _ := TestNamespaceCommands(getTopics, args) + assert.NotNil(t, execErr) + assert.Equal(t, "code: 404 reason: Tenant does not exist", execErr.Error()) +} diff --git a/pkg/pulsar/admin.go b/pkg/pulsar/admin.go index f78c64914..1859ec278 100644 --- a/pkg/pulsar/admin.go +++ b/pkg/pulsar/admin.go @@ -27,9 +27,9 @@ type Config struct { HttpClient *http.Client ApiVersion ApiVersion - Auth *auth.TlsAuthProvider - AuthParams string - TlsOptions *TLSOptions + Auth *auth.TlsAuthProvider + AuthParams string + TlsOptions *TLSOptions } type TLSOptions struct { @@ -57,6 +57,7 @@ type Client interface { Tenants() Tenants Sources() Sources Topics() Topics + Namespaces() Namespaces } type client struct { @@ -407,10 +408,10 @@ func responseError(resp *http.Response) error { json.Unmarshal(body, &e) - e.Code = resp.StatusCode - if e.Reason == "" { - e.Reason = unknownErrorReason - } + e.Code = resp.StatusCode + if e.Reason == "" { + e.Reason = unknownErrorReason + } return e } diff --git a/pkg/pulsar/auth_polices.go b/pkg/pulsar/auth_polices.go new file mode 100644 index 000000000..3ede13025 --- /dev/null +++ b/pkg/pulsar/auth_polices.go @@ -0,0 +1,21 @@ +// 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 pulsar + +type AuthPolicies struct { +} diff --git a/pkg/pulsar/backlog_quota.go b/pkg/pulsar/backlog_quota.go new file mode 100644 index 000000000..712e947a3 --- /dev/null +++ b/pkg/pulsar/backlog_quota.go @@ -0,0 +1,50 @@ +// 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 pulsar + +type BacklogQuota struct { + Limit int64 + Police RetentionPolicy + BacklogQuotaType BacklogQuotaType +} + +type RetentionPolicy int + +type BacklogQuotaType string + +const DestinationStorage BacklogQuotaType = "destination_storage" + +const ( + 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/bundles_data.go b/pkg/pulsar/bundles_data.go new file mode 100644 index 000000000..fe141854b --- /dev/null +++ b/pkg/pulsar/bundles_data.go @@ -0,0 +1,37 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package pulsar + +type BundlesData struct { + Boundaries []string `json:"boundaries"` + NumBundles int `json:"numBundles"` +} + +func NewBundlesData(boundaries []string) *BundlesData { + return &BundlesData{ + Boundaries: boundaries, + NumBundles: len(boundaries) - 1, + } +} + +func NewBundlesDataWithNumBundles(numBundles int) *BundlesData { + return &BundlesData{ + Boundaries: nil, + NumBundles: numBundles, + } +} diff --git a/pkg/pulsar/data.go b/pkg/pulsar/data.go index abd124904..f7b0b9b9a 100644 --- a/pkg/pulsar/data.go +++ b/pkg/pulsar/data.go @@ -121,3 +121,7 @@ type SourceData struct { type PartitionedTopicMetadata struct { Partitions int `json:"partitions"` } + +type NamespacesData struct { + NumBundles int `json:"numBundles"` +} diff --git a/pkg/pulsar/dispatch_rate.go b/pkg/pulsar/dispatch_rate.go new file mode 100644 index 000000000..3813b03b2 --- /dev/null +++ b/pkg/pulsar/dispatch_rate.go @@ -0,0 +1,44 @@ +// 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 pulsar + +type DispatchRate struct { + DispatchThrottlingRateInMsg int + DispatchThrottlingRateInByte int64 + RatePeriodInSecond int +} + +func NewDispatchRate() *DispatchRate { + return &DispatchRate{ + DispatchThrottlingRateInMsg: -1, + DispatchThrottlingRateInByte: -1, + RatePeriodInSecond: 1, + } +} + +type SubscribeRate struct { + SubscribeThrottlingRatePerConsumer int + RatePeriodInSecond int +} + +func NewSubscribeRate() *SubscribeRate { + return &SubscribeRate{ + SubscribeThrottlingRatePerConsumer: -1, + RatePeriodInSecond: 30, + } +} diff --git a/pkg/pulsar/namespace.go b/pkg/pulsar/namespace.go new file mode 100644 index 000000000..a41c595ce --- /dev/null +++ b/pkg/pulsar/namespace.go @@ -0,0 +1,140 @@ +// 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 pulsar + +type Namespaces interface { + // Get the list of all the namespaces for a certain tenant + GetNamespaces(tenant string) ([]string, error) + + // Get the list of all the topics under a certain namespace + GetTopics(namespace string) ([]string, error) + + // Get the dump all the policies specified for a namespace + GetPolicies(namespace string) (*Polices, error) + + // Creates a new empty namespace with no policies attached + CreateNamespace(namespace string) error + + // Creates a new empty namespace with no policies attached + CreateNsWithNumBundles(namespace string, numBundles int) error + + // Creates a new namespace with the specified policies + CreateNsWithPolices(namespace string, polices Polices) error + + // Creates a new empty namespace with no policies attached + CreateNsWithBundlesData(namespace string, bundleData *BundlesData) error + + // Delete an existing namespace + DeleteNamespace(namespace string) error + + // Delete an existing bundle in a namespace + DeleteNamespaceBundle(namespace string, bundleRange string) error +} + +type namespaces struct { + client *client + basePath string +} + +func (c *client) Namespaces() Namespaces { + return &namespaces{ + client: c, + basePath: "/namespaces", + } +} + +func (n *namespaces) GetNamespaces(tenant string) ([]string, error) { + var namespaces []string + endpoint := n.client.endpoint(n.basePath, tenant) + err := n.client.get(endpoint, &namespaces) + return namespaces, err +} + +func (n *namespaces) GetTopics(namespace string) ([]string, error) { + var topics []string + ns, err := GetNamespaceName(namespace) + if err != nil { + return nil, err + } + endpoint := n.client.endpoint(n.basePath, ns.String(), "topics") + err = n.client.get(endpoint, &topics) + return topics, err +} + +func (n *namespaces) GetPolicies(namespace string) (*Polices, error) { + var police Polices + ns, err := GetNamespaceName(namespace) + if err != nil { + return nil, err + } + endpoint := n.client.endpoint(n.basePath, ns.String()) + err = n.client.get(endpoint, &police) + return &police, err +} + +func (n *namespaces) CreateNsWithNumBundles(namespace string, numBundles int) error { + return n.CreateNsWithBundlesData(namespace, NewBundlesDataWithNumBundles(numBundles)) +} + +func (n *namespaces) CreateNsWithPolices(namespace string, polices Polices) error { + ns, err := GetNamespaceName(namespace) + if err != nil { + return err + } + endpoint := n.client.endpoint(n.basePath, ns.String()) + return n.client.put(endpoint, &polices, nil) +} + +func (n *namespaces) CreateNsWithBundlesData(namespace string, bundleData *BundlesData) error { + ns, err := GetNamespaceName(namespace) + if err != nil { + return err + } + endpoint := n.client.endpoint(n.basePath, ns.String()) + polices := new(Polices) + polices.Bundles = bundleData + + return n.client.put(endpoint, &polices, nil) +} + +func (n *namespaces) CreateNamespace(namespace string) error { + ns, err := GetNamespaceName(namespace) + if err != nil { + return err + } + endpoint := n.client.endpoint(n.basePath, ns.String()) + return n.client.put(endpoint, nil, nil) +} + +func (n *namespaces) DeleteNamespace(namespace string) error { + ns, err := GetNamespaceName(namespace) + if err != nil { + return err + } + endpoint := n.client.endpoint(n.basePath, ns.String()) + return n.client.delete(endpoint, nil) +} + +func (n *namespaces) DeleteNamespaceBundle(namespace string, bundleRange string) error { + ns, err := GetNamespaceName(namespace) + if err != nil { + return err + } + endpoint := n.client.endpoint(n.basePath, ns.String(), bundleRange) + return n.client.delete(endpoint, nil) +} diff --git a/pkg/pulsar/persistence_policies.go b/pkg/pulsar/persistence_policies.go new file mode 100644 index 000000000..e6dfb7ddb --- /dev/null +++ b/pkg/pulsar/persistence_policies.go @@ -0,0 +1,34 @@ +// 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 pulsar + +type PersistencePolicies struct { + BookkeeperEnsemble int + BookkeeperWriteQuorum int + BookkeeperAckQuorum int + ManagedLedgerMaxMarkDeleteRate float64 +} + +func NewPersistencePolicies() *PersistencePolicies { + return &PersistencePolicies{ + BookkeeperEnsemble: 2, + BookkeeperWriteQuorum: 2, + BookkeeperAckQuorum: 2, + ManagedLedgerMaxMarkDeleteRate: 0.0, + } +} diff --git a/pkg/pulsar/polices.go b/pkg/pulsar/polices.go new file mode 100644 index 000000000..33600be8b --- /dev/null +++ b/pkg/pulsar/polices.go @@ -0,0 +1,58 @@ +// 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 pulsar + +const ( + FirstBoundary string = "0x00000000" + LastBoundary string = "0xffffffff" +) + +type Polices struct { + AuthPolicies AuthPolicies + ReplicationClusters []string + Bundles *BundlesData + BacklogQuotaMap map[BacklogQuotaType]BacklogQuota + TopicDispatchRate map[string]DispatchRate + SubscriptionDispatchRate map[string]DispatchRate + replicatorDispatchRate map[string]DispatchRate + ClusterSubscribeRate map[string]SubscribeRate + Persistence PersistencePolicies + DeduplicationEnabled bool + LatencyStatsSampleRate map[string]int + MessageTtlInSeconds int + RetentionPolicies RetentionPolicies + Deleted bool + AntiAffinityGroup string + EncryptionRequired bool + SubscriptionAuthMode SubscriptionAuthMode + MaxProducersPerTopic int + MaxConsumersPerTopic int + MaxConsumersPerSubscription int + CompactionThreshold int64 + OffloadThreshold int64 + OffloadDeletionLagMs int64 + SchemaAutoUpdateCompatibilityStrategy SchemaAutoUpdateCompatibilityStrategy + SchemaValidationEnforced bool +} + +type SubscriptionAuthMode string + +const ( + None SubscriptionAuthMode = "None" + Prefix SubscriptionAuthMode = "Prefix" +) diff --git a/pkg/pulsar/retention_policies.go b/pkg/pulsar/retention_policies.go new file mode 100644 index 000000000..c25214c08 --- /dev/null +++ b/pkg/pulsar/retention_policies.go @@ -0,0 +1,23 @@ +// 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 pulsar + +type RetentionPolicies struct { + RetentionTimeInMinutes int + RetentionSizeInMB int64 +} diff --git a/pkg/pulsar/schema_strategy.go b/pkg/pulsar/schema_strategy.go new file mode 100644 index 000000000..66b475259 --- /dev/null +++ b/pkg/pulsar/schema_strategy.go @@ -0,0 +1,31 @@ +// 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 pulsar + +type SchemaAutoUpdateCompatibilityStrategy string + +const ( + AutoUpdateDisabled SchemaAutoUpdateCompatibilityStrategy = "AutoUpdateDisabled" + Backward SchemaAutoUpdateCompatibilityStrategy = "Backward" + Forward SchemaAutoUpdateCompatibilityStrategy = "Forward" + Full SchemaAutoUpdateCompatibilityStrategy = "Full" + AlwaysCompatible SchemaAutoUpdateCompatibilityStrategy = "AlwaysCompatible" + BackwardTransitive SchemaAutoUpdateCompatibilityStrategy = "BackwardTransitive" + ForwardTransitive SchemaAutoUpdateCompatibilityStrategy = "ForwardTransitive" + FullTransitive SchemaAutoUpdateCompatibilityStrategy = "FullTransitive" +) From d759037d83269eba7858cd3db0bf2a6ae11e83db Mon Sep 17 00:00:00 2001 From: "xiaolong.ran" Date: Sat, 14 Sep 2019 17:25:31 +0800 Subject: [PATCH 2/4] fix a little Signed-off-by: xiaolong.ran --- pkg/ctl/namespace/create.go | 13 +++++++++++++ pkg/pulsar/data.go | 9 +++++---- 2 files changed, 18 insertions(+), 4 deletions(-) diff --git a/pkg/ctl/namespace/create.go b/pkg/ctl/namespace/create.go index 360758c19..b0430143b 100644 --- a/pkg/ctl/namespace/create.go +++ b/pkg/ctl/namespace/create.go @@ -83,6 +83,13 @@ func createNs(vc *cmdutils.VerbCmd) { "b", 0, "number of bundles to activate") + + flagSet.StringSliceVarP( + &namespaceData.Clusters, + "clusters", + "c", + nil, + "List of clusters this namespace will be assigned") }) } @@ -93,6 +100,7 @@ func doCreate(vc *cmdutils.VerbCmd, data pulsar.NamespacesData) error { if data.NumBundles < 0 || data.NumBundles > int(MaxBundles) { return errors.New("Invalid number of bundles. Number of numBundles has to be in the range of (0, 2^32].") } + ns, err := pulsar.GetNamespaceName(tenantAndNamespace) if err != nil { return err @@ -103,6 +111,11 @@ func doCreate(vc *cmdutils.VerbCmd, data pulsar.NamespacesData) error { } else { polices.Bundles = nil } + + if data.Clusters != nil { + polices.ReplicationClusters = data.Clusters + } + err = admin.Namespaces().CreateNsWithPolices(ns.String(), *polices) if err == nil { vc.Command.Printf("Created %s successfully", ns.String()) diff --git a/pkg/pulsar/data.go b/pkg/pulsar/data.go index 1ef7425f9..919b45edf 100644 --- a/pkg/pulsar/data.go +++ b/pkg/pulsar/data.go @@ -123,12 +123,13 @@ type PartitionedTopicMetadata struct { } type NamespacesData struct { - NumBundles int `json:"numBundles"` + NumBundles int `json:"numBundles"` + Clusters []string `json:"clusters"` } type LookupData struct { - BrokerUrl string `json:"brokerUrl"` + BrokerUrl string `json:"brokerUrl"` BrokerUrlTls string `json:"brokerUrlTls"` - HttpUrl string `json:"httpUrl"` - HttpUrlTls string `json:"httpUrlTls"` + HttpUrl string `json:"httpUrl"` + HttpUrlTls string `json:"httpUrlTls"` } From 199d5ab6b906216813feef9e5b70760cbded7685 Mon Sep 17 00:00:00 2001 From: "xiaolong.ran" Date: Mon, 16 Sep 2019 12:09:58 +0800 Subject: [PATCH 3/4] fix comments Signed-off-by: xiaolong.ran --- pkg/ctl/cluster/cluster.go | 4 +- pkg/ctl/cluster/create.go | 2 +- pkg/ctl/cluster/delete_failure_domain_test.go | 2 +- pkg/ctl/cluster/delete_test.go | 2 +- pkg/ctl/cluster/get_failure_domain_test.go | 4 +- pkg/ctl/cluster/get_peer_clusters_test.go | 3 +- pkg/ctl/cluster/list_failure_domain_test.go | 6 +- pkg/ctl/cluster/tls_test.go | 39 ++++---- pkg/ctl/cluster/update.go | 2 +- pkg/ctl/cluster/update_peer_clusters_test.go | 2 +- pkg/ctl/cluster/update_test.go | 2 +- pkg/ctl/functions/putstate.go | 3 - pkg/ctl/namespace/create.go | 21 +++-- pkg/ctl/namespace/create_test.go | 92 +++++++++++++++++-- pkg/ctl/namespace/delete.go | 4 +- pkg/ctl/namespace/delete_test.go | 14 ++- pkg/ctl/namespace/list.go | 10 +- pkg/ctl/namespace/namespace.go | 2 +- pkg/ctl/namespace/{police.go => policies.go} | 16 ++-- .../{police_test.go => policies_test.go} | 10 +- pkg/ctl/namespace/test_help.go | 1 - pkg/ctl/sources/delete_test.go | 2 - pkg/ctl/tenant/tenant.go | 2 +- pkg/ctl/tenant/update.go | 2 +- pkg/ctl/tenant/update_test.go | 6 +- pkg/ctl/utils/util.go | 1 - pkg/pulsar/auth_polices.go | 19 ++++ pkg/pulsar/bundles_data.go | 10 +- pkg/pulsar/namespace.go | 14 +-- pkg/pulsar/polices.go | 58 ------------ pkg/pulsar/policies.go | 82 +++++++++++++++++ 31 files changed, 283 insertions(+), 154 deletions(-) rename pkg/ctl/namespace/{police.go => policies.go} (92%) rename pkg/ctl/namespace/{police_test.go => policies_test.go} (90%) delete mode 100644 pkg/pulsar/polices.go create mode 100644 pkg/pulsar/policies.go diff --git a/pkg/ctl/cluster/cluster.go b/pkg/ctl/cluster/cluster.go index 1122b92ed..57f156fee 100644 --- a/pkg/ctl/cluster/cluster.go +++ b/pkg/ctl/cluster/cluster.go @@ -36,11 +36,11 @@ func Command(flagGrouping *cmdutils.FlagGrouping) *cobra.Command { "", "cluster") - cmdutils.AddVerbCmd(flagGrouping, resourceCmd, createClusterCmd) + cmdutils.AddVerbCmd(flagGrouping, resourceCmd, CreateClusterCmd) cmdutils.AddVerbCmd(flagGrouping, resourceCmd, listClustersCmd) cmdutils.AddVerbCmd(flagGrouping, resourceCmd, getClusterDataCmd) cmdutils.AddVerbCmd(flagGrouping, resourceCmd, deleteClusterCmd) - cmdutils.AddVerbCmd(flagGrouping, resourceCmd, updateClusterCmd) + cmdutils.AddVerbCmd(flagGrouping, resourceCmd, UpdateClusterCmd) cmdutils.AddVerbCmd(flagGrouping, resourceCmd, updatePeerClustersCmd) cmdutils.AddVerbCmd(flagGrouping, resourceCmd, getPeerClustersCmd) cmdutils.AddVerbCmd(flagGrouping, resourceCmd, createFailureDomainCmd) diff --git a/pkg/ctl/cluster/create.go b/pkg/ctl/cluster/create.go index c25e1bcab..51f27ae43 100644 --- a/pkg/ctl/cluster/create.go +++ b/pkg/ctl/cluster/create.go @@ -6,7 +6,7 @@ import ( "github.com/streamnative/pulsarctl/pkg/pulsar" ) -func createClusterCmd(vc *cmdutils.VerbCmd) { +func CreateClusterCmd(vc *cmdutils.VerbCmd) { // update the description vc.SetDescription( "add", diff --git a/pkg/ctl/cluster/delete_failure_domain_test.go b/pkg/ctl/cluster/delete_failure_domain_test.go index dd7c188ed..0ed56b85d 100644 --- a/pkg/ctl/cluster/delete_failure_domain_test.go +++ b/pkg/ctl/cluster/delete_failure_domain_test.go @@ -7,7 +7,7 @@ import ( func TestDeleteFailureDomainCmd(t *testing.T) { args := []string{"create", "delete-failure-test"} - _, _, _, err := TestClusterCommands(createClusterCmd, args) + _, _, _, err := TestClusterCommands(CreateClusterCmd, args) assert.Nil(t, err) args = []string{"create-failure-domain", "-b", "127.0.0.1:6650", "delete-failure-test", "delete-failure-domain"} diff --git a/pkg/ctl/cluster/delete_test.go b/pkg/ctl/cluster/delete_test.go index e7768120c..0db1a5198 100644 --- a/pkg/ctl/cluster/delete_test.go +++ b/pkg/ctl/cluster/delete_test.go @@ -8,7 +8,7 @@ import ( func TestDeleteClusterCmd(t *testing.T) { args := []string{"add", "delete-test"} - _, _, _, err := TestClusterCommands(createClusterCmd, args) + _, _, _, err := TestClusterCommands(CreateClusterCmd, args) assert.Nil(t, err) args = []string{"list"} diff --git a/pkg/ctl/cluster/get_failure_domain_test.go b/pkg/ctl/cluster/get_failure_domain_test.go index 053177a8c..ac02a5745 100644 --- a/pkg/ctl/cluster/get_failure_domain_test.go +++ b/pkg/ctl/cluster/get_failure_domain_test.go @@ -9,11 +9,11 @@ import ( func TestGetFailureDomainSuccess(t *testing.T) { args := []string{"create", "failure-broker-A"} - _, _, _, err := TestClusterCommands(createClusterCmd, args) + _, _, _, err := TestClusterCommands(CreateClusterCmd, args) assert.Nil(t, err) args = []string{"create", "failure-broker-B"} - _, _, _, err = TestClusterCommands(createClusterCmd, args) + _, _, _, err = TestClusterCommands(CreateClusterCmd, args) assert.Nil(t, err) args = []string{"create-failure-domain", "-b", "failure-broker-A", "-b", "failure-broker-B", "standalone", "failure-domain"} diff --git a/pkg/ctl/cluster/get_peer_clusters_test.go b/pkg/ctl/cluster/get_peer_clusters_test.go index d5008cd9d..986485939 100644 --- a/pkg/ctl/cluster/get_peer_clusters_test.go +++ b/pkg/ctl/cluster/get_peer_clusters_test.go @@ -8,7 +8,7 @@ import ( func TestGetPeerClustersCmd(t *testing.T) { args := []string{"add", "test_get_peer", "--peer-cluster", "standalone"} - _, _, _, err := TestClusterCommands(createClusterCmd, args) + _, _, _, err := TestClusterCommands(CreateClusterCmd, args) if err != nil { t.Fatal(err) } @@ -22,4 +22,3 @@ func TestGetPeerClustersCmd(t *testing.T) { res := out.String() assert.True(t, strings.Contains(res, "standalone")) } - diff --git a/pkg/ctl/cluster/list_failure_domain_test.go b/pkg/ctl/cluster/list_failure_domain_test.go index d48cf757b..5ebd6e0b5 100644 --- a/pkg/ctl/cluster/list_failure_domain_test.go +++ b/pkg/ctl/cluster/list_failure_domain_test.go @@ -9,15 +9,15 @@ import ( func TestListFailureDomainsCmd(t *testing.T) { args := []string{"create", "list-failure-test"} - _, _, _, err := TestClusterCommands(createClusterCmd, args) + _, _, _, err := TestClusterCommands(CreateClusterCmd, args) assert.Nil(t, err) args = []string{"create", "list-failure-broker-A"} - _, _, _, err = TestClusterCommands(createClusterCmd, args) + _, _, _, err = TestClusterCommands(CreateClusterCmd, args) assert.Nil(t, err) args = []string{"create", "list-failure-broker-B"} - _, _, _, err = TestClusterCommands(createClusterCmd, args) + _, _, _, err = TestClusterCommands(CreateClusterCmd, args) assert.Nil(t, err) args = []string{"create-failure-domain", "--brokers", "list-failure-broker-A", "list-failure-test", "list-failure-A"} diff --git a/pkg/ctl/cluster/tls_test.go b/pkg/ctl/cluster/tls_test.go index 8c06fc587..f8100c7e2 100644 --- a/pkg/ctl/cluster/tls_test.go +++ b/pkg/ctl/cluster/tls_test.go @@ -3,30 +3,29 @@ package cluster import ( - `github.com/stretchr/testify/assert` - `strings` - `testing` + "github.com/stretchr/testify/assert" + "strings" + "testing" ) func TestTLS(t *testing.T) { - args := []string{"clusters","add", "tls"} - _, err := TestTlsHelp(createClusterCmd, args) - assert.Nil(t, err) + args := []string{"clusters", "add", "tls"} + _, err := TestTlsHelp(CreateClusterCmd, args) + assert.Nil(t, err) - args = []string{"clusters","list"} - out, err := TestTlsHelp(listClustersCmd, args) - assert.Nil(t, err) - clusters := out.String() - assert.True(t, strings.Contains(clusters, "tls")) + args = []string{"clusters", "list"} + out, err := TestTlsHelp(listClustersCmd, args) + assert.Nil(t, err) + clusters := out.String() + assert.True(t, strings.Contains(clusters, "tls")) - args = []string{"clusters","delete", "tls"} - _, err = TestTlsHelp(deleteClusterCmd, args) - assert.Nil(t, err) + args = []string{"clusters", "delete", "tls"} + _, err = TestTlsHelp(deleteClusterCmd, args) + assert.Nil(t, err) - args = []string{"clusters","list"} - out, err = TestTlsHelp(listClustersCmd, args) - assert.Nil(t, err) - clusters = out.String() - assert.False(t, strings.Contains(clusters, "tls")) + args = []string{"clusters", "list"} + out, err = TestTlsHelp(listClustersCmd, args) + assert.Nil(t, err) + clusters = out.String() + assert.False(t, strings.Contains(clusters, "tls")) } - diff --git a/pkg/ctl/cluster/update.go b/pkg/ctl/cluster/update.go index e72482736..ad023221f 100644 --- a/pkg/ctl/cluster/update.go +++ b/pkg/ctl/cluster/update.go @@ -6,7 +6,7 @@ import ( "github.com/streamnative/pulsarctl/pkg/pulsar" ) -func updateClusterCmd(vc *cmdutils.VerbCmd) { +func UpdateClusterCmd(vc *cmdutils.VerbCmd) { var desc pulsar.LongDescription desc.CommandUsedFor = "This command is used for updating the cluster data of the specified cluster." desc.CommandPermission = "This command requires super-user permissions." diff --git a/pkg/ctl/cluster/update_peer_clusters_test.go b/pkg/ctl/cluster/update_peer_clusters_test.go index 1bd81a9b7..c822d1149 100644 --- a/pkg/ctl/cluster/update_peer_clusters_test.go +++ b/pkg/ctl/cluster/update_peer_clusters_test.go @@ -9,7 +9,7 @@ import ( func TestUpdatePeerClusters(t *testing.T) { args := []string{"add", "test_peer_cluster"} - _, _, _, err := TestClusterCommands(createClusterCmd, args) + _, _, _, err := TestClusterCommands(CreateClusterCmd, args) if err != nil { t.Fatal(err) } diff --git a/pkg/ctl/cluster/update_test.go b/pkg/ctl/cluster/update_test.go index 89c4208c4..4552fe2c3 100644 --- a/pkg/ctl/cluster/update_test.go +++ b/pkg/ctl/cluster/update_test.go @@ -19,7 +19,7 @@ func TestUpdateCluster(t *testing.T) { "standalone", } - _, _, _, err := TestClusterCommands(updateClusterCmd, args) + _, _, _, err := TestClusterCommands(UpdateClusterCmd, args) if err != nil { t.Error(err) } diff --git a/pkg/ctl/functions/putstate.go b/pkg/ctl/functions/putstate.go index 9220ddf73..4655125d4 100644 --- a/pkg/ctl/functions/putstate.go +++ b/pkg/ctl/functions/putstate.go @@ -18,7 +18,6 @@ package functions import ( - "fmt" "github.com/pkg/errors" "github.com/spf13/pflag" "github.com/streamnative/pulsarctl/pkg/cmdutils" @@ -146,8 +145,6 @@ func doPutStateFunction(vc *cmdutils.VerbCmd, funcData *pulsar.FunctionData) err state.Key = vc.NameArgs[0] value := vc.NameArgs[1] - fmt.Println("value:", value) - if value == "-" { state.StringValue = strings.Join(vc.NameArgs[2:], " ") } else if value == "=" { diff --git a/pkg/ctl/namespace/create.go b/pkg/ctl/namespace/create.go index b0430143b..91a2e7e54 100644 --- a/pkg/ctl/namespace/create.go +++ b/pkg/ctl/namespace/create.go @@ -29,11 +29,11 @@ const MaxBundles = int64(1) << 32 func createNs(vc *cmdutils.VerbCmd) { desc := pulsar.LongDescription{} desc.CommandUsedFor = "Creates a new namespace" - desc.CommandPermission = "This command requires namespace admin permissions." + desc.CommandPermission = "This command requires tenant admin permissions." var examples []pulsar.Example create := pulsar.Example{ - Desc: "create a namespace named ", + Desc: "creates a namespace named ", Command: "pulsarctl namespaces create ", } examples = append(examples, create) @@ -60,7 +60,12 @@ func createNs(vc *cmdutils.VerbCmd) { Out: "[✖] code: 404 reason: Namespace does not exist", } - out = append(out, successOut, notExistTenantName, notTenantName, notExistNsName) + positiveBundleErr := pulsar.Output{ + Desc: "Invalid number of bundles, please check --bundles value", + Out: "Invalid number of bundles. Number of numBundles has to be in the range of (0, 2^32].", + } + + out = append(out, successOut, notExistTenantName, notTenantName, notExistNsName, positiveBundleErr) desc.CommandOutput = out vc.SetDescription( @@ -105,18 +110,16 @@ func doCreate(vc *cmdutils.VerbCmd, data pulsar.NamespacesData) error { if err != nil { return err } - polices := new(pulsar.Polices) + policies := pulsar.NewDefaultPolicies() if data.NumBundles > 0 { - polices.Bundles = pulsar.NewBundlesDataWithNumBundles(data.NumBundles) - } else { - polices.Bundles = nil + policies.Bundles = pulsar.NewBundlesDataWithNumBundles(data.NumBundles) } if data.Clusters != nil { - polices.ReplicationClusters = data.Clusters + policies.ReplicationClusters = data.Clusters } - err = admin.Namespaces().CreateNsWithPolices(ns.String(), *polices) + err = admin.Namespaces().CreateNsWithPolices(ns.String(), *policies) if err == nil { vc.Command.Printf("Created %s successfully", ns.String()) } diff --git a/pkg/ctl/namespace/create_test.go b/pkg/ctl/namespace/create_test.go index b85a68690..f952a6b38 100644 --- a/pkg/ctl/namespace/create_test.go +++ b/pkg/ctl/namespace/create_test.go @@ -18,7 +18,10 @@ package namespace import ( - "fmt" + "encoding/json" + "github.com/streamnative/pulsarctl/pkg/ctl/cluster" + "github.com/streamnative/pulsarctl/pkg/ctl/tenant" + "github.com/streamnative/pulsarctl/pkg/pulsar" "github.com/stretchr/testify/assert" "strings" "testing" @@ -31,16 +34,61 @@ func TestCreateNs(t *testing.T) { assert.Equal(t, createOut.String(), "Created public/test-namespace successfully") args = []string{"list", "public"} - out, _, _, _ := TestNamespaceCommands(getNamespacesPerProperty, args) - fmt.Println(out.String()) + out, _, _, _ := TestNamespaceCommands(getNamespacesFromTenant, args) assert.True(t, strings.Contains(out.String(), "public/test-namespace")) + + policiesArgs := []string{"policies", "public/test-namespace"} + out, execErr, _, _ := TestNamespaceCommands(getPolicies, policiesArgs) + assert.Nil(t, execErr) + + var police pulsar.Policies + err = json.Unmarshal(out.Bytes(), &police) + assert.Nil(t, err) + + for cluster := range police.ClusterSubscribeRate { + assert.Equal(t, cluster, "standalone") + } + + assert.Equal(t, police.Bundles.NumBundles, 4) } -func TestCreateNsArgsError(t *testing.T) { - args := []string{"create"} - _, _, nameErr, _ := TestNamespaceCommands(createNs, args) +func TestCreateNsForBundles(t *testing.T) { + args := []string{"create", "public/test-namespace-bundles", "--bundles", "0"} + createOut, _, _, err := TestNamespaceCommands(createNs, args) + assert.Nil(t, err) + t.Log(createOut.String()) - assert.Equal(t, "only one argument is allowed to be used as a name", nameErr.Error()) + policiesArgs := []string{"policies", "public/test-namespace-bundles"} + out, execErr, _, _ := TestNamespaceCommands(getPolicies, policiesArgs) + assert.Nil(t, execErr) + + var police pulsar.Policies + err = json.Unmarshal(out.Bytes(), &police) + assert.Nil(t, err) + assert.Equal(t, 4, police.Bundles.NumBundles) +} + +func TestCreateNsForNegativeBundles(t *testing.T) { + args := []string{"create", "public/test-namespace-negative-bundles", "--bundles", "-1"} + createOut, execErr, _, err := TestNamespaceCommands(createNs, args) + assert.Nil(t, err) + exceptedErr := "Invalid number of bundles. Number of numBundles has to be in the range of (0, 2^32]." + t.Log(createOut.String()) + assert.Equal(t, exceptedErr, execErr.Error()) +} + +func TestCreateNsForPositiveBundles(t *testing.T) { + args := []string{"create", "public/test-namespace-positive-bundles", "--bundles", "12"} + _, _, _, err := TestNamespaceCommands(createNs, args) + + policiesArgs := []string{"policies", "public/test-namespace-positive-bundles"} + out, execErr, _, _ := TestNamespaceCommands(getPolicies, policiesArgs) + assert.Nil(t, execErr) + + var police pulsar.Policies + err = json.Unmarshal(out.Bytes(), &police) + assert.Nil(t, err) + assert.Equal(t, 12, police.Bundles.NumBundles) } func TestCreateNsAlreadyExistError(t *testing.T) { @@ -53,3 +101,33 @@ func TestCreateNsAlreadyExistError(t *testing.T) { assert.NotNil(t, execErr) assert.Equal(t, "code: 409 reason: Namespace already exists", execErr.Error()) } + +func TestCreateNsForCluster(t *testing.T) { + clusterArgs := []string{"create", "test-cluster", "--url", "192.168.12.11"} + _, _, _, err := cluster.TestClusterCommands(cluster.CreateClusterCmd, clusterArgs) + assert.Nil(t, err) + + updateTenantArgs := []string{"update", "--allowed-clusters", "test-cluster", "public"} + _, _, _, err = tenant.TestTenantCommands(tenant.UpdateTenantCmd, updateTenantArgs) + assert.Nil(t, err) + + nsArgs := []string{"create", "public/test-namespace-cluster", "--clusters", "test-cluster"} + nsOut, _, _, err := TestNamespaceCommands(createNs, nsArgs) + assert.Equal(t, "Created public/test-namespace-cluster successfully", nsOut.String()) + + policiesArgs := []string{"policies", "public/test-namespace-cluster"} + out, execErr, _, _ := TestNamespaceCommands(getPolicies, policiesArgs) + assert.Nil(t, execErr) + + var police pulsar.Policies + err = json.Unmarshal(out.Bytes(), &police) + assert.Nil(t, err) + assert.Equal(t, "test-cluster", police.ReplicationClusters[0]) +} + +func TestCreateNsArgsError(t *testing.T) { + args := []string{"create"} + _, _, nameErr, _ := TestNamespaceCommands(createNs, args) + + assert.Equal(t, "only one argument is allowed to be used as a name", nameErr.Error()) +} diff --git a/pkg/ctl/namespace/delete.go b/pkg/ctl/namespace/delete.go index e6f6d5f17..075217bec 100644 --- a/pkg/ctl/namespace/delete.go +++ b/pkg/ctl/namespace/delete.go @@ -25,7 +25,7 @@ import ( func deleteNs(vc *cmdutils.VerbCmd) { desc := pulsar.LongDescription{} desc.CommandUsedFor = "Deletes a namespace. The namespace needs to be empty" - desc.CommandPermission = "This command requires namespace admin permissions." + desc.CommandPermission = "This command requires tenant admin permissions." var examples []pulsar.Example del := pulsar.Example{ @@ -38,7 +38,7 @@ func deleteNs(vc *cmdutils.VerbCmd) { var out []pulsar.Output successOut := pulsar.Output{ Desc: "normal output", - Out: "Created successfully", + Out: "Deleted successfully", } notTenantName := pulsar.Output{ diff --git a/pkg/ctl/namespace/delete_test.go b/pkg/ctl/namespace/delete_test.go index 1e7e27e4b..f98739e7f 100644 --- a/pkg/ctl/namespace/delete_test.go +++ b/pkg/ctl/namespace/delete_test.go @@ -34,7 +34,7 @@ func TestDeleteNsCmd(t *testing.T) { assert.Equal(t, delOut.String(), "Deleted public/test-delete-namespace successfully") args = []string{"list", "public"} - listOut, _, _, _ := TestNamespaceCommands(getNamespacesPerProperty, args) + listOut, _, _, _ := TestNamespaceCommands(getNamespacesFromTenant, args) assert.False(t, strings.Contains(listOut.String(), "public/test-delete-namespace")) } @@ -44,9 +44,17 @@ func TestDeleteNsArgsError(t *testing.T) { assert.Equal(t, "only one argument is allowed to be used as a name", nameErr.Error()) } -func TestDeleteNonExistTenant(t *testing.T) { - args := []string{"delete", "non-existent-tenant/test-delete-namespace"} +func TestDeleteNonExistentTenant(t *testing.T) { + args := []string{"delete", "non-existent-tenant/default"} _, execErr, _, _ := TestNamespaceCommands(deleteNs, args) assert.NotNil(t, execErr) assert.Equal(t, "code: 404 reason: Tenant does not exist", execErr.Error()) } + +func TestDeleteNonExistentNamespace(t *testing.T) { + args := []string{"delete", "public/non-existent-namespace"} + _, execErr, _, _ := TestNamespaceCommands(deleteNs, args) + assert.NotNil(t, execErr) + nonExistentNsErr := "code: 404 reason: Namespace public/non-existent-namespace does not exist." + assert.Equal(t, nonExistentNsErr, execErr.Error()) +} diff --git a/pkg/ctl/namespace/list.go b/pkg/ctl/namespace/list.go index f534746ea..fe9e75aa2 100644 --- a/pkg/ctl/namespace/list.go +++ b/pkg/ctl/namespace/list.go @@ -23,15 +23,15 @@ import ( "github.com/streamnative/pulsarctl/pkg/pulsar" ) -func getNamespacesPerProperty(vc *cmdutils.VerbCmd) { +func getNamespacesFromTenant(vc *cmdutils.VerbCmd) { desc := pulsar.LongDescription{} - desc.CommandUsedFor = "Get the namespaces for a tenant" - desc.CommandPermission = "This command requires namespace admin permissions." + desc.CommandUsedFor = "Get the list of namespaces of a tenant" + desc.CommandPermission = "This command requires tenant admin permissions." var examples []pulsar.Example list := pulsar.Example{ - Desc: "Get the namespaces for a tenant", + Desc: "Get the list of namespaces of a tenant", Command: "pulsarctl namespaces list ", } @@ -64,7 +64,7 @@ func getNamespacesPerProperty(vc *cmdutils.VerbCmd) { vc.SetDescription( "list", - "Get the namespaces for a tenant", + "Get the list of namespaces of a tenant", desc.ToString(), "list", ) diff --git a/pkg/ctl/namespace/namespace.go b/pkg/ctl/namespace/namespace.go index f8805808f..287063cc6 100644 --- a/pkg/ctl/namespace/namespace.go +++ b/pkg/ctl/namespace/namespace.go @@ -30,7 +30,7 @@ func Command(flagGrouping *cmdutils.FlagGrouping) *cobra.Command { "namespace", ) - cmdutils.AddVerbCmd(flagGrouping, resourceCmd, getNamespacesPerProperty) + cmdutils.AddVerbCmd(flagGrouping, resourceCmd, getNamespacesFromTenant) cmdutils.AddVerbCmd(flagGrouping, resourceCmd, getTopics) cmdutils.AddVerbCmd(flagGrouping, resourceCmd, getPolicies) cmdutils.AddVerbCmd(flagGrouping, resourceCmd, createNs) diff --git a/pkg/ctl/namespace/police.go b/pkg/ctl/namespace/policies.go similarity index 92% rename from pkg/ctl/namespace/police.go rename to pkg/ctl/namespace/policies.go index bccc0c571..ae2c3c2b8 100644 --- a/pkg/ctl/namespace/police.go +++ b/pkg/ctl/namespace/policies.go @@ -25,12 +25,12 @@ import ( func getPolicies(vc *cmdutils.VerbCmd) { desc := pulsar.LongDescription{} desc.CommandUsedFor = "Get the configuration policies of a namespace" - desc.CommandPermission = "This command requires namespace admin permissions." + desc.CommandPermission = "This command requires tenant admin permissions." var examples []pulsar.Example police := pulsar.Example{ Desc: "Get the configuration policies of a namespace", - Command: "pulsarctl namespaces polices ", + Command: "pulsarctl namespaces policies ", } examples = append(examples, police) desc.CommandExamples = examples @@ -119,23 +119,23 @@ func getPolicies(vc *cmdutils.VerbCmd) { desc.CommandOutput = out vc.SetDescription( - "polices", + "policies", "Get the configuration policies of a namespace", desc.ToString(), - "police", + "policies", ) vc.SetRunFuncWithNameArg(func() error { - return doGetPolices(vc) + return doGetPolicies(vc) }) } -func doGetPolices(vc *cmdutils.VerbCmd) error { +func doGetPolicies(vc *cmdutils.VerbCmd) error { namespace := vc.NameArg admin := cmdutils.NewPulsarClient() - police, err := admin.Namespaces().GetPolicies(namespace) + policies, err := admin.Namespaces().GetPolicies(namespace) if err == nil { - cmdutils.PrintJson(vc.Command.OutOrStdout(), police) + cmdutils.PrintJson(vc.Command.OutOrStdout(), policies) } return err } diff --git a/pkg/ctl/namespace/police_test.go b/pkg/ctl/namespace/policies_test.go similarity index 90% rename from pkg/ctl/namespace/police_test.go rename to pkg/ctl/namespace/policies_test.go index ce4535c8a..6ebdf8dc6 100644 --- a/pkg/ctl/namespace/police_test.go +++ b/pkg/ctl/namespace/policies_test.go @@ -25,11 +25,11 @@ import ( ) func TestPolicesCommand(t *testing.T) { - args := []string{"polices", "public/default"} + args := []string{"policies", "public/default"} out, execErr, _, _ := TestNamespaceCommands(getPolicies, args) assert.Nil(t, execErr) - var police pulsar.Polices + var police pulsar.Policies err := json.Unmarshal(out.Bytes(), &police) assert.Nil(t, err) @@ -46,20 +46,20 @@ func TestPolicesCommand(t *testing.T) { } func TestPolicesNsArgsError(t *testing.T) { - args := []string{"polices"} + args := []string{"policies"} _, _, nameErr, _ := TestNamespaceCommands(getPolicies, args) assert.Equal(t, "only one argument is allowed to be used as a name", nameErr.Error()) } func TestPolicesNonExistTenant(t *testing.T) { - args := []string{"polices", "non-existent-tenant/default"} + args := []string{"policies", "non-existent-tenant/default"} _, execErr, _, _ := TestNamespaceCommands(getPolicies, args) assert.NotNil(t, execErr) assert.Equal(t, "code: 404 reason: Tenant does not exist", execErr.Error()) } func TestPolicesNonExistNs(t *testing.T) { - args := []string{"polices", "public/test-not-exist-ns"} + args := []string{"policies", "public/test-not-exist-ns"} _, execErr, _, _ := TestNamespaceCommands(getPolicies, args) assert.NotNil(t, execErr) assert.Equal(t, "code: 404 reason: Namespace does not exist", execErr.Error()) diff --git a/pkg/ctl/namespace/test_help.go b/pkg/ctl/namespace/test_help.go index 775bbb9ab..805dd827b 100644 --- a/pkg/ctl/namespace/test_help.go +++ b/pkg/ctl/namespace/test_help.go @@ -61,5 +61,4 @@ func TestNamespaceCommands(newVerb func(cmd *cmdutils.VerbCmd), args []string) ( err = rootCmd.Execute() return buf, execError, nameError, err - } diff --git a/pkg/ctl/sources/delete_test.go b/pkg/ctl/sources/delete_test.go index 0c3416f74..4e2e3a4ae 100644 --- a/pkg/ctl/sources/delete_test.go +++ b/pkg/ctl/sources/delete_test.go @@ -18,7 +18,6 @@ package sources import ( - "fmt" "github.com/stretchr/testify/assert" "strings" "testing" @@ -63,7 +62,6 @@ func TestFailureDeleteSource(t *testing.T) { exceptedErr := "Source test-source-delete doesn't exist" _, execErrMsg, _ := TestSourcesCommands(deleteSourcesCmd, failureDeleteArgs) - fmt.Println(execErrMsg.Error()) assert.True(t, strings.Contains(execErrMsg.Error(), exceptedErr)) assert.NotNil(t, execErrMsg) diff --git a/pkg/ctl/tenant/tenant.go b/pkg/ctl/tenant/tenant.go index a26a73ff2..9c017a2a1 100644 --- a/pkg/ctl/tenant/tenant.go +++ b/pkg/ctl/tenant/tenant.go @@ -30,7 +30,7 @@ func Command(flagGrouping *cmdutils.FlagGrouping) *cobra.Command { cmdutils.AddVerbCmd(flagGrouping, resourceCmd, createTenantCmd) cmdutils.AddVerbCmd(flagGrouping, resourceCmd, deleteTenantCmd) - cmdutils.AddVerbCmd(flagGrouping, resourceCmd, updateTenantCmd) + cmdutils.AddVerbCmd(flagGrouping, resourceCmd, UpdateTenantCmd) cmdutils.AddVerbCmd(flagGrouping, resourceCmd, listTenantCmd) cmdutils.AddVerbCmd(flagGrouping, resourceCmd, getTenantCmd) diff --git a/pkg/ctl/tenant/update.go b/pkg/ctl/tenant/update.go index 263bc27f6..69a0140c6 100644 --- a/pkg/ctl/tenant/update.go +++ b/pkg/ctl/tenant/update.go @@ -6,7 +6,7 @@ import ( "github.com/streamnative/pulsarctl/pkg/pulsar" ) -func updateTenantCmd(vc *cmdutils.VerbCmd) { +func UpdateTenantCmd(vc *cmdutils.VerbCmd) { var desc pulsar.LongDescription desc.CommandUsedFor = "This command is used for updating the configuration of a tenant." desc.CommandPermission = "This command requires super-user permissions." diff --git a/pkg/ctl/tenant/update_test.go b/pkg/ctl/tenant/update_test.go index 419708120..f9e3f2c65 100644 --- a/pkg/ctl/tenant/update_test.go +++ b/pkg/ctl/tenant/update_test.go @@ -29,7 +29,7 @@ func TestUpdateTenantCmd(t *testing.T) { assert.Equal(t, "standalone", tenantData.AllowedClusters[0]) args = []string{"update", "--admin-roles", "new-role", "--allowed-clusters", "standalone", "update-tenant-test"} - _, execErr, _, _ = TestTenantCommands(updateTenantCmd, args) + _, execErr, _, _ = TestTenantCommands(UpdateTenantCmd, args) assert.Nil(t, execErr) args = []string{"get", "update-tenant-test"} @@ -48,14 +48,14 @@ func TestUpdateTenantCmd(t *testing.T) { func TestUpdateArgsError(t *testing.T) { args := []string{"update"} - _, _, nameErr, _ := TestTenantCommands(updateTenantCmd, args) + _, _, nameErr, _ := TestTenantCommands(UpdateTenantCmd, args) assert.NotNil(t, nameErr) assert.Equal(t, "only one argument is allowed to be used as a name", nameErr.Error()) } func TestUpdateNonExistTenantError(t *testing.T) { args := []string{"update", "--admin-roles", "update-role", "--allowed-clusters", "standalone", "non-existent-topic"} - _, execErr, _, _ := TestTenantCommands(updateTenantCmd, args) + _, execErr, _, _ := TestTenantCommands(UpdateTenantCmd, args) assert.NotNil(t, execErr) assert.Equal(t, "code: 404 reason: Tenant does not exist", execErr.Error()) } diff --git a/pkg/ctl/utils/util.go b/pkg/ctl/utils/util.go index 2f9f3030a..3010c72d0 100644 --- a/pkg/ctl/utils/util.go +++ b/pkg/ctl/utils/util.go @@ -88,7 +88,6 @@ func InferMissingSinkeArguments(sinkConf *pulsar.SinkConfig) { func IsFileExist(filename string) bool { info, err := os.Stat(filename) if os.IsNotExist(err) { - fmt.Println(info) return false } fmt.Println("exists", info.Name(), info.Size(), info.ModTime()) diff --git a/pkg/pulsar/auth_polices.go b/pkg/pulsar/auth_polices.go index 3ede13025..0200f1146 100644 --- a/pkg/pulsar/auth_polices.go +++ b/pkg/pulsar/auth_polices.go @@ -18,4 +18,23 @@ package pulsar type AuthPolicies struct { + NamespaceAuth map[string]AuthAction `json:"namespace_auth"` + DestinationAuth map[string]map[string]AuthAction `json:"destination_auth"` + SubscriptionAuthRoles map[string][]string `json:"subscription_auth_roles"` } + +func NewAuthPolicies() *AuthPolicies { + return &AuthPolicies{ + NamespaceAuth: make(map[string]AuthAction), + DestinationAuth: make(map[string]map[string]AuthAction), + SubscriptionAuthRoles: make(map[string][]string), + } +} + +type AuthAction string + +const ( + produce AuthAction = "produce" + consume AuthAction = "consume" + function AuthAction = "functions" +) diff --git a/pkg/pulsar/bundles_data.go b/pkg/pulsar/bundles_data.go index fe141854b..5da16ae44 100644 --- a/pkg/pulsar/bundles_data.go +++ b/pkg/pulsar/bundles_data.go @@ -22,8 +22,8 @@ type BundlesData struct { NumBundles int `json:"numBundles"` } -func NewBundlesData(boundaries []string) *BundlesData { - return &BundlesData{ +func NewBundlesData(boundaries []string) BundlesData { + return BundlesData{ Boundaries: boundaries, NumBundles: len(boundaries) - 1, } @@ -35,3 +35,9 @@ func NewBundlesDataWithNumBundles(numBundles int) *BundlesData { NumBundles: numBundles, } } + +func NewDefaultBoundle() *BundlesData { + bundleData := NewBundlesDataWithNumBundles(1) + bundleData.Boundaries = append(bundleData.Boundaries, FirstBoundary, LastBoundary) + return bundleData +} diff --git a/pkg/pulsar/namespace.go b/pkg/pulsar/namespace.go index a41c595ce..f6c9b432c 100644 --- a/pkg/pulsar/namespace.go +++ b/pkg/pulsar/namespace.go @@ -25,7 +25,7 @@ type Namespaces interface { GetTopics(namespace string) ([]string, error) // Get the dump all the policies specified for a namespace - GetPolicies(namespace string) (*Polices, error) + GetPolicies(namespace string) (*Policies, error) // Creates a new empty namespace with no policies attached CreateNamespace(namespace string) error @@ -34,7 +34,7 @@ type Namespaces interface { CreateNsWithNumBundles(namespace string, numBundles int) error // Creates a new namespace with the specified policies - CreateNsWithPolices(namespace string, polices Polices) error + CreateNsWithPolices(namespace string, polices Policies) error // Creates a new empty namespace with no policies attached CreateNsWithBundlesData(namespace string, bundleData *BundlesData) error @@ -76,8 +76,8 @@ func (n *namespaces) GetTopics(namespace string) ([]string, error) { return topics, err } -func (n *namespaces) GetPolicies(namespace string) (*Polices, error) { - var police Polices +func (n *namespaces) GetPolicies(namespace string) (*Policies, error) { + var police Policies ns, err := GetNamespaceName(namespace) if err != nil { return nil, err @@ -91,13 +91,13 @@ func (n *namespaces) CreateNsWithNumBundles(namespace string, numBundles int) er return n.CreateNsWithBundlesData(namespace, NewBundlesDataWithNumBundles(numBundles)) } -func (n *namespaces) CreateNsWithPolices(namespace string, polices Polices) error { +func (n *namespaces) CreateNsWithPolices(namespace string, policies Policies) error { ns, err := GetNamespaceName(namespace) if err != nil { return err } endpoint := n.client.endpoint(n.basePath, ns.String()) - return n.client.put(endpoint, &polices, nil) + return n.client.put(endpoint, &policies, nil) } func (n *namespaces) CreateNsWithBundlesData(namespace string, bundleData *BundlesData) error { @@ -106,7 +106,7 @@ func (n *namespaces) CreateNsWithBundlesData(namespace string, bundleData *Bundl return err } endpoint := n.client.endpoint(n.basePath, ns.String()) - polices := new(Polices) + polices := new(Policies) polices.Bundles = bundleData return n.client.put(endpoint, &polices, nil) diff --git a/pkg/pulsar/polices.go b/pkg/pulsar/polices.go deleted file mode 100644 index 33600be8b..000000000 --- a/pkg/pulsar/polices.go +++ /dev/null @@ -1,58 +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 pulsar - -const ( - FirstBoundary string = "0x00000000" - LastBoundary string = "0xffffffff" -) - -type Polices struct { - AuthPolicies AuthPolicies - ReplicationClusters []string - Bundles *BundlesData - BacklogQuotaMap map[BacklogQuotaType]BacklogQuota - TopicDispatchRate map[string]DispatchRate - SubscriptionDispatchRate map[string]DispatchRate - replicatorDispatchRate map[string]DispatchRate - ClusterSubscribeRate map[string]SubscribeRate - Persistence PersistencePolicies - DeduplicationEnabled bool - LatencyStatsSampleRate map[string]int - MessageTtlInSeconds int - RetentionPolicies RetentionPolicies - Deleted bool - AntiAffinityGroup string - EncryptionRequired bool - SubscriptionAuthMode SubscriptionAuthMode - MaxProducersPerTopic int - MaxConsumersPerTopic int - MaxConsumersPerSubscription int - CompactionThreshold int64 - OffloadThreshold int64 - OffloadDeletionLagMs int64 - SchemaAutoUpdateCompatibilityStrategy SchemaAutoUpdateCompatibilityStrategy - SchemaValidationEnforced bool -} - -type SubscriptionAuthMode string - -const ( - None SubscriptionAuthMode = "None" - Prefix SubscriptionAuthMode = "Prefix" -) diff --git a/pkg/pulsar/policies.go b/pkg/pulsar/policies.go new file mode 100644 index 000000000..d8336db52 --- /dev/null +++ b/pkg/pulsar/policies.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 pulsar + +const ( + FirstBoundary string = "0x00000000" + LastBoundary string = "0xffffffff" +) + +type Policies struct { + AuthPolicies AuthPolicies `json:"auth_policies"` + ReplicationClusters []string `json:"replication_clusters"` + Bundles *BundlesData `json:"bundles"` + BacklogQuotaMap map[BacklogQuotaType]BacklogQuota `json:"backlog_quota_map"` + TopicDispatchRate map[string]DispatchRate `json:"topicDispatchRate"` + SubscriptionDispatchRate map[string]DispatchRate `json:"subscriptionDispatchRate"` + ReplicatorDispatchRate map[string]DispatchRate `json:"replicatorDispatchRate"` + ClusterSubscribeRate map[string]SubscribeRate `json:"clusterSubscribeRate"` + Persistence *PersistencePolicies `json:"persistence"` + DeduplicationEnabled bool `json:"deduplicationEnabled"` + LatencyStatsSampleRate map[string]int `json:"latency_stats_sample_rate"` + MessageTtlInSeconds int `json:"message_ttl_in_seconds"` + RetentionPolicies *RetentionPolicies `json:"retention_policies"` + Deleted bool `json:"deleted"` + AntiAffinityGroup string `json:"antiAffinityGroup"` + EncryptionRequired bool `json:"encryption_required"` + SubscriptionAuthMode SubscriptionAuthMode `json:"subscription_auth_mode"` + MaxProducersPerTopic int `json:"max_producers_per_topic"` + MaxConsumersPerTopic int `json:"max_consumers_per_topic"` + MaxConsumersPerSubscription int `json:"max_consumers_per_subscription"` + CompactionThreshold int64 `json:"compaction_threshold"` + OffloadThreshold int64 `json:"offload_threshold"` + OffloadDeletionLagMs int64 `json:"offload_deletion_lag_ms"` + SchemaAutoUpdateCompatibilityStrategy SchemaAutoUpdateCompatibilityStrategy `json:"schema_auto_update_compatibility_strategy"` + SchemaValidationEnforced bool `json:"schema_validation_enforced"` +} + +func NewDefaultPolicies() *Policies { + return &Policies{ + AuthPolicies: *NewAuthPolicies(), + ReplicationClusters: make([]string, 0, 10), + BacklogQuotaMap: make(map[BacklogQuotaType]BacklogQuota), + TopicDispatchRate: make(map[string]DispatchRate), + SubscriptionDispatchRate: make(map[string]DispatchRate), + ReplicatorDispatchRate: make(map[string]DispatchRate), + ClusterSubscribeRate: make(map[string]SubscribeRate), + LatencyStatsSampleRate: make(map[string]int), + MessageTtlInSeconds: 0, + Deleted: false, + EncryptionRequired: false, + SubscriptionAuthMode: None, + MaxProducersPerTopic: 0, + MaxConsumersPerSubscription: 0, + MaxConsumersPerTopic: 0, + CompactionThreshold: 0, + OffloadThreshold: -1, + SchemaAutoUpdateCompatibilityStrategy: Full, + SchemaValidationEnforced: false, + } +} + +type SubscriptionAuthMode string + +const ( + None SubscriptionAuthMode = "None" + Prefix SubscriptionAuthMode = "Prefix" +) From ba9faff6b382a7d79122c87831646963fb1b6c9e Mon Sep 17 00:00:00 2001 From: "xiaolong.ran" Date: Mon, 16 Sep 2019 20:34:14 +0800 Subject: [PATCH 4/4] Add backlog, retention and ttl commands for pulsarctl Signed-off-by: xiaolong.ran --- 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, 1071 insertions(+), 25 deletions(-) create mode 100644 pkg/ctl/namespace/backlog_quota_test.go create mode 100644 pkg/ctl/namespace/get_backlog_quota.go create mode 100644 pkg/ctl/namespace/get_message_ttl.go create mode 100644 pkg/ctl/namespace/get_retention.go create mode 100644 pkg/ctl/namespace/message_ttl_test.go create mode 100644 pkg/ctl/namespace/remove_backlog_quota.go create mode 100644 pkg/ctl/namespace/retention_test.go create mode 100644 pkg/ctl/namespace/set_backlog_quota.go create mode 100644 pkg/ctl/namespace/set_message_ttl.go create mode 100644 pkg/ctl/namespace/set_retention.go create 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 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 b1ec01cae..2edd91b61 100644 --- a/pkg/pulsar/data.go +++ b/pkg/pulsar/data.go @@ -123,8 +123,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), + } }