diff --git a/pkg/cmdutils/cmdutils.go b/pkg/cmdutils/cmdutils.go index eb6204f3c..51c4e32c6 100644 --- a/pkg/cmdutils/cmdutils.go +++ b/pkg/cmdutils/cmdutils.go @@ -88,6 +88,10 @@ func NewPulsarClientWithAPIVersion(version pulsar.APIVersion) pulsar.Client { return PulsarCtlConfig.Client(version) } +func NewBookieClient() pulsar.BookieClient { + return PulsarCtlConfig.BookieClient() +} + func PrintJSON(w io.Writer, obj interface{}) { b, err := json.MarshalIndent(obj, "", " ") if err != nil { diff --git a/pkg/cmdutils/config.go b/pkg/cmdutils/config.go index 9befc52dd..ab9b0d4ba 100644 --- a/pkg/cmdutils/config.go +++ b/pkg/cmdutils/config.go @@ -41,6 +41,9 @@ type ClusterConfig struct { AuthParams string + // the bookkeeper web service url that pulsarctl connects to. + BookieWebServiceURL string + // Token and TokenFile is used to config the pulsarctl using token to authentication Token string TokenFile string @@ -77,6 +80,13 @@ func (c *ClusterConfig) FlagSet() *pflag.FlagSet { "", "Allow TLS trust cert file path") + flags.StringVar( + &c.BookieWebServiceURL, + "bookie-service-url", + pulsar.DefaultBookieWebServiceURL, + "The bookie web service url that pulsarctl connects to.", + ) + flags.StringVar( &c.Token, "token", @@ -148,3 +158,12 @@ func (c *ClusterConfig) Client(version pulsar.APIVersion) pulsar.Client { } return client } + +func (c *ClusterConfig) BookieClient() pulsar.BookieClient { + config := pulsar.DefaultConfig() + if len(c.BookieWebServiceURL) > 0 && c.BookieWebServiceURL != config.BookieWebServiceURL { + config.BookieWebServiceURL = c.BookieWebServiceURL + } + + return pulsar.NewBookieClient(config) +} diff --git a/pkg/ctl/bk/bk.go b/pkg/ctl/bk/bk.go new file mode 100644 index 000000000..976551ff0 --- /dev/null +++ b/pkg/ctl/bk/bk.go @@ -0,0 +1,38 @@ +// 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 bk + +import ( + "github.com/streamnative/pulsarctl/pkg/cmdutils" + "github.com/streamnative/pulsarctl/pkg/ctl/bk/ledger" + + "github.com/spf13/cobra" +) + +func Command(flagGrouping *cmdutils.FlagGrouping) *cobra.Command { + resourceCmd := cmdutils.NewResourceCmd( + "bk", + "Operations about bookKeeper", + "", + "", + ) + + resourceCmd.AddCommand(ledger.Command(flagGrouping)) + + return resourceCmd +} diff --git a/pkg/ctl/bk/ledger/delete.go b/pkg/ctl/bk/ledger/delete.go new file mode 100644 index 000000000..6f3e809d5 --- /dev/null +++ b/pkg/ctl/bk/ledger/delete.go @@ -0,0 +1,74 @@ +// 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 ledger + +import ( + "strconv" + + "github.com/streamnative/pulsarctl/pkg/cmdutils" + "github.com/streamnative/pulsarctl/pkg/pulsar" + + "github.com/pkg/errors" +) + +func DeleteCmd(vc *cmdutils.VerbCmd) { + var desc pulsar.LongDescription + desc.CommandUsedFor = "This command is used for deleting a ledger." + desc.CommandPermission = "none" + + var examples []pulsar.Example + deleteLedger := pulsar.Example{ + Desc: "Delete the specified ledger", + Command: "pulsarctl bookies ledger delete --ledger-id (ledger-id)", + } + examples = append(examples, deleteLedger) + desc.CommandExamples = examples + + var out []pulsar.Output + successOut := pulsar.Output{ + Desc: "normal output", + Out: "Successfully delete the ledger (ledger-id)", + } + out = append(out, successOut, argError) + desc.CommandOutput = out + + vc.SetDescription( + "delete", + "Delete a ledger", + desc.ToString(), + desc.ExampleToString()) + + vc.SetRunFuncWithNameArg(func() error { + return doDeleteCmd(vc) + }, "the ledger id is not specified or the ledger id is specified more than one") +} + +func doDeleteCmd(vc *cmdutils.VerbCmd) error { + id, err := strconv.ParseInt(vc.NameArg, 10, 64) + if err != nil || id < 0 { + return errors.Errorf("invalid ledger id %s", vc.NameArg) + } + + admin := cmdutils.NewBookieClient() + err = admin.Ledger().Delete(id) + if err == nil { + vc.Command.Printf("Successfully delete the ledger %d", id) + } + + return err +} diff --git a/pkg/ctl/bk/ledger/delete_test.go b/pkg/ctl/bk/ledger/delete_test.go new file mode 100644 index 000000000..807b67d89 --- /dev/null +++ b/pkg/ctl/bk/ledger/delete_test.go @@ -0,0 +1,42 @@ +// 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 ledger + +import ( + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestDeleteArgError(t *testing.T) { + args := []string{"delete"} + _, _, nameErr, _ := TestLedgerCommands(DeleteCmd, args) + assert.NotNil(t, nameErr) + assert.Equal(t, "the ledger id is not specified or the ledger id is specified more than one", + nameErr.Error()) + + args = []string{"delete", "a"} + _, execErr, _, _ := TestLedgerCommands(DeleteCmd, args) + assert.NotNil(t, execErr) + assert.Equal(t, "invalid ledger id a", execErr.Error()) + + args = []string{"delete", "--", "-1"} + _, execErr, _, _ = TestLedgerCommands(DeleteCmd, args) + assert.NotNil(t, execErr) + assert.Equal(t, "invalid ledger id -1", execErr.Error()) +} diff --git a/pkg/ctl/bk/ledger/errors.go b/pkg/ctl/bk/ledger/errors.go new file mode 100644 index 000000000..6edeb808d --- /dev/null +++ b/pkg/ctl/bk/ledger/errors.go @@ -0,0 +1,25 @@ +// 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 ledger + +import "github.com/streamnative/pulsarctl/pkg/pulsar" + +var argError = pulsar.Output{ + Desc: "the ledger id is not specified or the ledger id is specified more than one", + Out: "[✖] the ledger id is not specified or the ledger id is specified more than one", +} diff --git a/pkg/ctl/bk/ledger/get.go b/pkg/ctl/bk/ledger/get.go new file mode 100644 index 000000000..0e4324b52 --- /dev/null +++ b/pkg/ctl/bk/ledger/get.go @@ -0,0 +1,105 @@ +// 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 ledger + +import ( + "encoding/json" + "strconv" + + "github.com/streamnative/pulsarctl/pkg/cmdutils" + "github.com/streamnative/pulsarctl/pkg/pulsar" + + "github.com/pkg/errors" +) + +func GetCmd(vc *cmdutils.VerbCmd) { + var desc pulsar.LongDescription + desc.CommandUsedFor = "This command is used for getting the metadata of a ledger." + desc.CommandPermission = "none" + + var examples []pulsar.Example + get := pulsar.Example{ + Desc: "Get the metadata of the specified ledger", + Command: "pulsarctl bookies ledger get (ledger-i)", + } + examples = append(examples, get) + desc.CommandExamples = examples + + metadata := pulsar.LedgerMetadata{ + MetadataFormatVersion: 1, + Ensemble: 1, + WriteQuorum: 1, + AckQuorum: 1, + Length: 1, + LastEntryID: 1, + Ctime: 1, + CToken: 0, + State: "CLOSED", + DigestType: "MAC", + Ensembles: map[int64][]pulsar.BookieSocketAddress{ + 1: { + pulsar.BookieSocketAddress{ + HostName: "www.examples.com", + Port: 8080, + }, + }, + }, + CurrentEnsemble: []pulsar.BookieSocketAddress{ + { + HostName: "www.example.com", + Port: 8080, + }, + }, + Password: make([]byte, 0), + CustomMetadata: map[string][]byte{}, + } + meta, _ := json.MarshalIndent(metadata, "", " ") + + var out []pulsar.Output + successOut := pulsar.Output{ + Desc: "normal output", + Out: string(meta), + } + out = append(out, successOut, argError) + desc.CommandOutput = out + + vc.SetDescription( + "get", + "Get the metadata of a ledger", + desc.ToString(), + desc.ExampleToString()) + + vc.SetRunFuncWithNameArg(func() error { + return doGet(vc) + }, "the ledger id is not specified or the ledger id is specified more than one") +} + +func doGet(vc *cmdutils.VerbCmd) error { + id, err := strconv.ParseInt(vc.NameArg, 10, 64) + if err != nil || id < 0 { + return errors.Errorf("invalid ledger id %s", vc.NameArg) + } + + admin := cmdutils.NewBookieClient() + metadata, err := admin.Ledger().Get(id) + if err == nil { + cmdutils.PrintJSON(vc.Command.OutOrStdout(), metadata) + } + + return err +} diff --git a/pkg/ctl/bk/ledger/get_test.go b/pkg/ctl/bk/ledger/get_test.go new file mode 100644 index 000000000..c381efb07 --- /dev/null +++ b/pkg/ctl/bk/ledger/get_test.go @@ -0,0 +1,42 @@ +// 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 ledger + +import ( + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestGetArgError(t *testing.T) { + args := []string{"get"} + _, _, nameErr, _ := TestLedgerCommands(GetCmd, args) + assert.NotNil(t, nameErr) + assert.Equal(t, "the ledger id is not specified or the ledger id is specified more than one", + nameErr.Error()) + + args = []string{"get", "a"} + _, execErr, _, _ := TestLedgerCommands(GetCmd, args) + assert.NotNil(t, execErr) + assert.Equal(t, "invalid ledger id a", execErr.Error()) + + args = []string{"get", "--", "-1"} + _, execErr, _, _ = TestLedgerCommands(GetCmd, args) + assert.NotNil(t, execErr) + assert.Equal(t, "invalid ledger id -1", execErr.Error()) +} diff --git a/pkg/ctl/bk/ledger/ledger.go b/pkg/ctl/bk/ledger/ledger.go new file mode 100644 index 000000000..36c445fb6 --- /dev/null +++ b/pkg/ctl/bk/ledger/ledger.go @@ -0,0 +1,43 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package ledger + +import ( + "github.com/streamnative/pulsarctl/pkg/cmdutils" + + "github.com/spf13/cobra" +) + +func Command(flagGrouping *cmdutils.FlagGrouping) *cobra.Command { + resourceCmd := cmdutils.NewResourceCmd( + "ledger", + "Operations about ledger", + "", + "") + + commands := []func(*cmdutils.VerbCmd){ + DeleteCmd, + GetCmd, + ListCmd, + ReadCmd, + } + + cmdutils.AddVerbCmds(flagGrouping, resourceCmd, commands...) + + return resourceCmd +} diff --git a/pkg/ctl/bk/ledger/list.go b/pkg/ctl/bk/ledger/list.go new file mode 100644 index 000000000..a868c54bc --- /dev/null +++ b/pkg/ctl/bk/ledger/list.go @@ -0,0 +1,91 @@ +// 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 ledger + +import ( + "sort" + + "github.com/streamnative/pulsarctl/pkg/cmdutils" + "github.com/streamnative/pulsarctl/pkg/pulsar" + + "github.com/spf13/pflag" +) + +func ListCmd(vc *cmdutils.VerbCmd) { + var desc pulsar.LongDescription + desc.CommandUsedFor = "This command is used for listing all the ledgers." + desc.CommandPermission = "none" + + var examples []pulsar.Example + list := pulsar.Example{ + Desc: "List all the ledgers", + Command: "pulsarctl bookies ledger list", + } + + showMeta := pulsar.Example{ + Desc: "List all the ledgers and the metadata of the ledger", + Command: "pulsarctl bookies ledger list --show-metadata", + } + examples = append(examples, list, showMeta) + desc.CommandExamples = examples + + var out []pulsar.Output + successOut := pulsar.Output{ + Desc: "normal output", + Out: "[1,2,3,4]", + } + out = append(out, successOut) + desc.CommandOutput = out + + vc.SetDescription( + "list", + "list all the ledgers", + desc.ToString(), + desc.ExampleToString()) + + var show bool + + vc.SetRunFunc(func() error { + return doListCmd(vc, show) + }) + + vc.FlagSetGroup.InFlagSet("List Ledgers", func(set *pflag.FlagSet) { + set.BoolVarP(&show, "show-metadata", "p", false, + "Show the metadata of the ledgers") + }) +} + +func doListCmd(vc *cmdutils.VerbCmd, showMeta bool) error { + admin := cmdutils.NewBookieClient() + ledgers, err := admin.Ledger().List(showMeta) + if err == nil { + if !showMeta { + ledgerList := make([]int64, 0) + for k := range ledgers { + ledgerList = append(ledgerList, k) + } + sort.Slice(ledgerList, func(i, j int) bool { + return ledgerList[i] < ledgerList[j] + }) + vc.Command.Println(ledgerList) + } else { + cmdutils.PrintJSON(vc.Command.OutOrStdout(), ledgers) + } + } + return err +} diff --git a/pkg/ctl/bk/ledger/read.go b/pkg/ctl/bk/ledger/read.go new file mode 100644 index 000000000..14dc60cdc --- /dev/null +++ b/pkg/ctl/bk/ledger/read.go @@ -0,0 +1,106 @@ +// 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 ledger + +import ( + "strconv" + + "github.com/streamnative/pulsarctl/pkg/cmdutils" + "github.com/streamnative/pulsarctl/pkg/pulsar" + + "github.com/pkg/errors" + "github.com/spf13/pflag" +) + +func ReadCmd(vc *cmdutils.VerbCmd) { + var desc pulsar.LongDescription + desc.CommandUsedFor = "This command is used for reading a range of entries of a ledger." + desc.CommandPermission = "none" + + var examples []pulsar.Example + r := pulsar.Example{ + Desc: "Read a range of entries of the specified ledger", + Command: "pulsar bookies ledger read (ledger-id)", + } + + rs := pulsar.Example{ + Desc: "Read the entries of the specified ledger started from the given entry id", + Command: "pulsar bookies ledger --start (entry-id) (ledger-id)", + } + + rse := pulsar.Example{ + Desc: "Read the specified range of entries of the specified ledger", + Command: "pulsar bookies ledger --start (entry-id) --end (entry-id) (ledger-id)", + } + + examples = append(examples, r, rs, rse) + desc.CommandExamples = examples + + var out []pulsar.Output + successOut := pulsar.Output{ + Desc: "normal output", + Out: `{ + "ledger-id", "message" +}`, + } + out = append(out, successOut, argError) + desc.CommandOutput = out + + vc.SetDescription( + "read", + "Read a range of entries of a ledger", + desc.ToString(), + desc.ExampleToString()) + + var start int64 + var end int64 + + vc.SetRunFuncWithNameArg(func() error { + return doRead(vc, start, end) + }, "the ledger id is not specified or the ledger id is specified more than one") + + vc.FlagSetGroup.InFlagSet("Read Ledger", func(set *pflag.FlagSet) { + set.Int64VarP(&start, "start", "b", -1, + "") + set.Int64VarP(&end, "end", "e", -1, "") + }) + +} + +func doRead(vc *cmdutils.VerbCmd, start, end int64) error { + id, err := strconv.ParseInt(vc.NameArg, 10, 64) + if err != nil || id < 0 { + return errors.Errorf("invalid ledger id %s", vc.NameArg) + } + + if start != -1 && start < 0 { + return errors.Errorf("invalid start ledger id %d", start) + } + + if end != -1 && end < 0 { + return errors.Errorf("invalid end ledger id %d", end) + } + + admin := cmdutils.NewBookieClient() + info, err := admin.Ledger().Read(id, start, end) + if err == nil { + cmdutils.PrintJSON(vc.Command.OutOrStdout(), info) + } + + return err +} diff --git a/pkg/ctl/bk/ledger/read_test.go b/pkg/ctl/bk/ledger/read_test.go new file mode 100644 index 000000000..8e2f472fb --- /dev/null +++ b/pkg/ctl/bk/ledger/read_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 ledger + +import ( + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestReadArgError(t *testing.T) { + args := []string{"read"} + _, _, nameErr, _ := TestLedgerCommands(ReadCmd, args) + assert.NotNil(t, nameErr) + assert.Equal(t, "the ledger id is not specified or the ledger id is specified more than one", + nameErr.Error()) + + args = []string{"read", "a"} + _, execErr, _, _ := TestLedgerCommands(ReadCmd, args) + assert.NotNil(t, execErr) + assert.Equal(t, "invalid ledger id a", execErr.Error()) + + args = []string{"read", "--", "-1"} + _, execErr, _, _ = TestLedgerCommands(ReadCmd, args) + assert.NotNil(t, execErr) + assert.Equal(t, "invalid ledger id -1", execErr.Error()) + + args = []string{"read", "--start", "-2", "1"} + _, execErr, _, _ = TestLedgerCommands(ReadCmd, args) + assert.NotNil(t, execErr) + assert.Equal(t, "invalid start ledger id -2", execErr.Error()) + + args = []string{"read", "--end", "-2", "1"} + _, execErr, _, _ = TestLedgerCommands(ReadCmd, args) + assert.NotNil(t, execErr) + assert.Equal(t, "invalid end ledger id -2", execErr.Error()) +} diff --git a/pkg/ctl/bk/ledger/test_help.go b/pkg/ctl/bk/ledger/test_help.go new file mode 100644 index 000000000..418785041 --- /dev/null +++ b/pkg/ctl/bk/ledger/test_help.go @@ -0,0 +1,67 @@ +// 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 ledger + +import ( + "bytes" + + "github.com/streamnative/pulsarctl/pkg/cmdutils" + + "github.com/kris-nova/logger" + "github.com/spf13/cobra" +) + +func TestLedgerCommands(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{"ledger"}, args...)) + + resourceCmd := cmdutils.NewResourceCmd( + "ledger", + "Operations about bookie(s)", + "", + "") + flagGrouping := cmdutils.NewGrouping() + cmdutils.AddVerbCmd(flagGrouping, resourceCmd, newVerb) + rootCmd.AddCommand(resourceCmd) + err = rootCmd.Execute() + + return buf, execError, nameError, err +} diff --git a/pkg/ctl/topic/topic.go b/pkg/ctl/topic/topic.go index 368f8aee2..79c1b21cd 100644 --- a/pkg/ctl/topic/topic.go +++ b/pkg/ctl/topic/topic.go @@ -49,6 +49,7 @@ func Command(flagGrouping *cmdutils.FlagGrouping) *cobra.Command { GetLastMessageIDCmd, GetStatsCmd, GetInternalStatsCmd, + GetInternalInfoCmd, } cmdutils.AddVerbCmds(flagGrouping, resourceCmd, commands...) diff --git a/pkg/pulsar/Tenant.go b/pkg/pulsar/Tenant.go index 53d9ba557..d99d11381 100644 --- a/pkg/pulsar/Tenant.go +++ b/pkg/pulsar/Tenant.go @@ -26,42 +26,44 @@ type Tenants interface { } type tenants struct { - client *client + client *pulsarClient + request *client basePath string } -func (c *client) Tenants() Tenants { +func (c *pulsarClient) Tenants() Tenants { return &tenants{ client: c, + request: c.client, basePath: "/tenants", } } func (c *tenants) Create(data TenantData) error { endpoint := c.client.endpoint(c.basePath, data.Name) - return c.client.put(endpoint, &data) + return c.request.put(endpoint, &data) } func (c *tenants) Delete(name string) error { endpoint := c.client.endpoint(c.basePath, name) - return c.client.delete(endpoint) + return c.request.delete(endpoint) } func (c *tenants) Update(data TenantData) error { endpoint := c.client.endpoint(c.basePath, data.Name) - return c.client.post(endpoint, &data) + return c.request.post(endpoint, &data) } func (c *tenants) List() ([]string, error) { var tenantList []string endpoint := c.client.endpoint(c.basePath, "") - err := c.client.get(endpoint, &tenantList) + err := c.request.get(endpoint, &tenantList) return tenantList, err } func (c *tenants) Get(name string) (TenantData, error) { var data TenantData endpoint := c.client.endpoint(c.basePath, name) - err := c.client.get(endpoint, &data) + err := c.request.get(endpoint, &data) return data, err } diff --git a/pkg/pulsar/admin.go b/pkg/pulsar/admin.go index ef13b9e60..02630d7a6 100644 --- a/pkg/pulsar/admin.go +++ b/pkg/pulsar/admin.go @@ -34,7 +34,8 @@ import ( ) const ( - DefaultWebServiceURL = "http://localhost:8080" + DefaultWebServiceURL = "http://localhost:8080" + DefaultBookieWebServiceURL = "http://localhost:8081" ) var ReleaseVersion = "None" @@ -48,7 +49,11 @@ type Config struct { Auth *auth.TLSAuthProvider AuthParams string TLSOptions *TLSOptions - TokenAuth *auth.TokenAuthProvider + + BookieWebServiceURL string + BookieAPIVersion APIVersion + + TokenAuth *auth.TokenAuthProvider } type TLSOptions struct { @@ -65,6 +70,9 @@ func DefaultConfig() *Config { TLSOptions: &TLSOptions{ AllowInsecureConnection: false, }, + + BookieWebServiceURL: DefaultBookieWebServiceURL, + BookieAPIVersion: BV1, } return config } @@ -84,19 +92,32 @@ type Client interface { BrokerStats() BrokerStats } +// BookieClient provides a client to the BookKeeper Restful API +type BookieClient interface { + Ledger() Ledger +} + type client struct { webServiceURL string - apiVersion string httpClient *http.Client + transport *http.Transport versionInfo string + tokenAuth *auth.TokenAuthProvider +} + +type pulsarClient struct { + client *client + apiVersion string // TLS config auth *auth.TLSAuthProvider authParams string tlsOptions *TLSOptions - transport *http.Transport +} - tokenAuth *auth.TokenAuthProvider +type bookieClient struct { + client *client + apiVersion string } // New returns a new client @@ -105,14 +126,19 @@ func New(config *Config) (Client, error) { config.WebServiceURL = DefaultWebServiceURL } - c := &client{ - apiVersion: config.APIVersion.String(), - webServiceURL: config.WebServiceURL, - versionInfo: ReleaseVersion, - tokenAuth: config.TokenAuth, + c := &pulsarClient{ + apiVersion: config.APIVersion.String(), + client: &client{ + webServiceURL: config.WebServiceURL, + versionInfo: ReleaseVersion, + }, + } + + if config.TokenAuth != nil { + c.client.tokenAuth = config.TokenAuth } - if strings.HasPrefix(c.webServiceURL, "https://") { + if strings.HasPrefix(c.client.webServiceURL, "https://") { c.authParams = config.AuthParams c.tlsOptions = config.TLSOptions mapAuthParams := make(map[string]string) @@ -128,7 +154,7 @@ func New(config *Config) (Client, error) { return nil, err } - c.transport = &http.Transport{ + c.client.transport = &http.Transport{ MaxIdleConnsPerHost: 10, TLSClientConfig: tlsConf, } @@ -137,7 +163,22 @@ func New(config *Config) (Client, error) { return c, nil } -func (c *client) getTLSConfig() (*tls.Config, error) { +func NewBookieClient(config *Config) BookieClient { + if len(config.BookieWebServiceURL) == 0 { + config.BookieWebServiceURL = DefaultBookieWebServiceURL + } + + c := &bookieClient{ + apiVersion: config.BookieAPIVersion.String(), + client: &client{ + webServiceURL: config.BookieWebServiceURL, + }, + } + + return c +} + +func (c *pulsarClient) getTLSConfig() (*tls.Config, error) { tlsConfig := &tls.Config{ InsecureSkipVerify: c.tlsOptions.AllowInsecureConnection, } @@ -167,8 +208,12 @@ func (c *client) getTLSConfig() (*tls.Config, error) { return tlsConfig, nil } -func (c *client) endpoint(componentPath string, parts ...string) string { - return path.Join(makeHTTPPath(c.apiVersion, componentPath), endpoint(parts...)) +func (c *pulsarClient) endpoint(componentPath string, parts ...string) string { + return path.Join(makeHTTPPath("admin", c.apiVersion, componentPath), endpoint(parts...)) +} + +func (c *bookieClient) bookieEndpoint(componentPath string, parts ...string) string { + return path.Join(makeHTTPPath("api", c.apiVersion, componentPath+"/"), endpoint(parts...)) } // get is used to do a GET request against an endpoint @@ -251,10 +296,10 @@ func (c *client) putWithQueryParams(endpoint string, in, obj interface{}, params } func (c *client) delete(endpoint string) error { - return c.deleteWithQueryParams(endpoint, nil, nil) + return c.deleteWithQueryParams(endpoint, nil) } -func (c *client) deleteWithQueryParams(endpoint string, obj interface{}, params map[string]string) error { +func (c *client) deleteWithQueryParams(endpoint string, params map[string]string) error { req, err := c.newRequest(http.MethodDelete, endpoint) if err != nil { return err @@ -268,18 +313,13 @@ func (c *client) deleteWithQueryParams(endpoint string, obj interface{}, params req.params = query } + // nolint resp, err := checkSuccessful(c.doRequest(req)) if err != nil { return err } defer safeRespClose(resp) - if obj != nil { - if err := decodeJSONBody(resp, &obj); err != nil { - return err - } - } - return nil } diff --git a/pkg/pulsar/api_version.go b/pkg/pulsar/api_version.go index 70963d943..a916b136f 100644 --- a/pkg/pulsar/api_version.go +++ b/pkg/pulsar/api_version.go @@ -23,6 +23,7 @@ const ( V1 APIVersion = iota V2 V3 + BV1 ) const DefaultAPIVersion = "v2" @@ -35,6 +36,8 @@ func (v APIVersion) String() string { return "v2" case V3: return "v3" + case BV1: + return "v1" } return DefaultAPIVersion diff --git a/pkg/pulsar/bookie_data.go b/pkg/pulsar/bookie_data.go new file mode 100644 index 000000000..c9cab8b9a --- /dev/null +++ b/pkg/pulsar/bookie_data.go @@ -0,0 +1,42 @@ +// 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 BookieSocketAddress struct { + Port int `json:"port"` + HostName string `json:"hostname"` +} + +type LedgerMetadata struct { + StoreCtime bool `json:"storeCtime"` + HasPassword bool `json:"hasPassword"` + MetadataFormatVersion int `json:"metadataFormatVersion"` + Ensemble int `json:"ensembleSize"` + WriteQuorum int `json:"writeQuorumSize"` + AckQuorum int `json:"ackQuorumSize"` + Length int64 `json:"length"` + LastEntryID int64 `json:"lastEntryId"` + Ctime int64 `json:"ctime"` + CToken int64 `json:"cToken"` + State string `json:"state"` + DigestType string `json:"digestType"` + Ensembles map[int64][]BookieSocketAddress `json:"allEnsembles"` + CurrentEnsemble []BookieSocketAddress `json:"currentEnsemble"` + Password []byte `json:"password"` + CustomMetadata map[string][]byte `json:"customMetadata"` +} diff --git a/pkg/pulsar/broker_stats.go b/pkg/pulsar/broker_stats.go index b28974437..c73cdf1d9 100644 --- a/pkg/pulsar/broker_stats.go +++ b/pkg/pulsar/broker_stats.go @@ -33,13 +33,15 @@ type BrokerStats interface { } type brokerStats struct { - client *client + client *pulsarClient + request *client basePath string } -func (c *client) BrokerStats() BrokerStats { +func (c *pulsarClient) BrokerStats() BrokerStats { return &brokerStats{ client: c, + request: c.client, basePath: "/broker-stats", } } @@ -47,7 +49,7 @@ func (c *client) BrokerStats() BrokerStats { func (bs *brokerStats) GetMetrics() ([]Metrics, error) { endpoint := bs.client.endpoint(bs.basePath, "/metrics") var response []Metrics - err := bs.client.get(endpoint, &response) + err := bs.request.get(endpoint, &response) if err != nil { return nil, err } @@ -58,7 +60,7 @@ func (bs *brokerStats) GetMetrics() ([]Metrics, error) { func (bs *brokerStats) GetMBeans() ([]Metrics, error) { endpoint := bs.client.endpoint(bs.basePath, "/mbeans") var response []Metrics - err := bs.client.get(endpoint, &response) + err := bs.request.get(endpoint, &response) if err != nil { return nil, err } @@ -68,7 +70,7 @@ func (bs *brokerStats) GetMBeans() ([]Metrics, error) { func (bs *brokerStats) GetTopics() (string, error) { endpoint := bs.client.endpoint(bs.basePath, "/topics") - buf, err := bs.client.getWithQueryParams(endpoint, nil, nil, false) + buf, err := bs.request.getWithQueryParams(endpoint, nil, nil, false) if err != nil { return "", err } @@ -79,7 +81,7 @@ func (bs *brokerStats) GetTopics() (string, error) { func (bs *brokerStats) GetLoadReport() (*LocalBrokerData, error) { endpoint := bs.client.endpoint(bs.basePath, "/load-report") response := NewLocalBrokerData() - err := bs.client.get(endpoint, &response) + err := bs.request.get(endpoint, &response) if err != nil { return nil, nil } @@ -89,7 +91,7 @@ func (bs *brokerStats) GetLoadReport() (*LocalBrokerData, error) { func (bs *brokerStats) GetAllocatorStats(allocatorName string) (*AllocatorStats, error) { endpoint := bs.client.endpoint(bs.basePath, "/allocator-stats", allocatorName) var allocatorStats AllocatorStats - err := bs.client.get(endpoint, &allocatorStats) + err := bs.request.get(endpoint, &allocatorStats) if err != nil { return nil, err } diff --git a/pkg/pulsar/brokers.go b/pkg/pulsar/brokers.go index 9c720d548..e8564b350 100644 --- a/pkg/pulsar/brokers.go +++ b/pkg/pulsar/brokers.go @@ -55,13 +55,15 @@ type Brokers interface { } type broker struct { - client *client + client *pulsarClient + request *client basePath string } -func (c *client) Brokers() Brokers { +func (c *pulsarClient) Brokers() Brokers { return &broker{ client: c, + request: c.client, basePath: "/brokers", } } @@ -69,7 +71,7 @@ func (c *client) Brokers() Brokers { func (b *broker) GetActiveBrokers(cluster string) ([]string, error) { endpoint := b.client.endpoint(b.basePath, cluster) var res []string - err := b.client.get(endpoint, &res) + err := b.request.get(endpoint, &res) if err != nil { return nil, err } @@ -79,7 +81,7 @@ func (b *broker) GetActiveBrokers(cluster string) ([]string, error) { func (b *broker) GetDynamicConfigurationNames() ([]string, error) { endpoint := b.client.endpoint(b.basePath, "/configuration/") var res []string - err := b.client.get(endpoint, &res) + err := b.request.get(endpoint, &res) if err != nil { return nil, err } @@ -89,7 +91,7 @@ func (b *broker) GetDynamicConfigurationNames() ([]string, error) { func (b *broker) GetOwnedNamespaces(cluster, brokerURL string) (map[string]NamespaceOwnershipStatus, error) { endpoint := b.client.endpoint(b.basePath, cluster, brokerURL, "ownedNamespaces") var res map[string]NamespaceOwnershipStatus - err := b.client.get(endpoint, &res) + err := b.request.get(endpoint, &res) if err != nil { return nil, err } @@ -99,18 +101,18 @@ func (b *broker) GetOwnedNamespaces(cluster, brokerURL string) (map[string]Names func (b *broker) UpdateDynamicConfiguration(configName, configValue string) error { value := url.QueryEscape(configValue) endpoint := b.client.endpoint(b.basePath, "/configuration/", configName, value) - return b.client.post(endpoint, nil) + return b.request.post(endpoint, nil) } func (b *broker) DeleteDynamicConfiguration(configName string) error { endpoint := b.client.endpoint(b.basePath, "/configuration/", configName) - return b.client.delete(endpoint) + return b.request.delete(endpoint) } func (b *broker) GetRuntimeConfigurations() (map[string]string, error) { endpoint := b.client.endpoint(b.basePath, "/configuration/", "runtime") var res map[string]string - err := b.client.get(endpoint, &res) + err := b.request.get(endpoint, &res) if err != nil { return nil, err } @@ -120,7 +122,7 @@ func (b *broker) GetRuntimeConfigurations() (map[string]string, error) { func (b *broker) GetInternalConfigurationData() (*InternalConfigurationData, error) { endpoint := b.client.endpoint(b.basePath, "/internal-configuration") var res InternalConfigurationData - err := b.client.get(endpoint, &res) + err := b.request.get(endpoint, &res) if err != nil { return nil, err } @@ -130,7 +132,7 @@ func (b *broker) GetInternalConfigurationData() (*InternalConfigurationData, err func (b *broker) GetAllDynamicConfigurations() (map[string]string, error) { endpoint := b.client.endpoint(b.basePath, "/configuration/", "values") var res map[string]string - err := b.client.get(endpoint, &res) + err := b.request.get(endpoint, &res) if err != nil { return nil, err } @@ -140,7 +142,7 @@ func (b *broker) GetAllDynamicConfigurations() (map[string]string, error) { func (b *broker) HealthCheck() error { endpoint := b.client.endpoint(b.basePath, "/health") - buf, err := b.client.getWithQueryParams(endpoint, nil, nil, false) + buf, err := b.request.getWithQueryParams(endpoint, nil, nil, false) if err != nil { return err } diff --git a/pkg/pulsar/cluster.go b/pkg/pulsar/cluster.go index 9b486825c..e69b00ff6 100644 --- a/pkg/pulsar/cluster.go +++ b/pkg/pulsar/cluster.go @@ -35,81 +35,83 @@ type Clusters interface { } type clusters struct { - client *client + client *pulsarClient + request *client basePath string } -func (c *client) Clusters() Clusters { +func (c *pulsarClient) Clusters() Clusters { return &clusters{ client: c, + request: c.client, basePath: "/clusters", } } func (c *clusters) List() ([]string, error) { var clusters []string - err := c.client.get(c.client.endpoint(c.basePath), &clusters) + err := c.request.get(c.client.endpoint(c.basePath), &clusters) return clusters, err } func (c *clusters) Get(name string) (ClusterData, error) { cdata := ClusterData{} endpoint := c.client.endpoint(c.basePath, name) - err := c.client.get(endpoint, &cdata) + err := c.request.get(endpoint, &cdata) return cdata, err } func (c *clusters) Create(cdata ClusterData) error { endpoint := c.client.endpoint(c.basePath, cdata.Name) - return c.client.put(endpoint, &cdata) + return c.request.put(endpoint, &cdata) } func (c *clusters) Delete(name string) error { endpoint := c.client.endpoint(c.basePath, name) - return c.client.delete(endpoint) + return c.request.delete(endpoint) } func (c *clusters) Update(cdata ClusterData) error { endpoint := c.client.endpoint(c.basePath, cdata.Name) - return c.client.post(endpoint, &cdata) + return c.request.post(endpoint, &cdata) } func (c *clusters) GetPeerClusters(name string) ([]string, error) { var peerClusters []string endpoint := c.client.endpoint(c.basePath, name, "peers") - err := c.client.get(endpoint, &peerClusters) + err := c.request.get(endpoint, &peerClusters) return peerClusters, err } func (c *clusters) UpdatePeerClusters(cluster string, peerClusters []string) error { endpoint := c.client.endpoint(c.basePath, cluster, "peers") - return c.client.post(endpoint, peerClusters) + return c.request.post(endpoint, peerClusters) } func (c *clusters) CreateFailureDomain(data FailureDomainData) error { endpoint := c.client.endpoint(c.basePath, data.ClusterName, "failureDomains", data.DomainName) - return c.client.post(endpoint, &data) + return c.request.post(endpoint, &data) } func (c *clusters) GetFailureDomain(clusterName string, domainName string) (FailureDomainData, error) { var res FailureDomainData endpoint := c.client.endpoint(c.basePath, clusterName, "failureDomains", domainName) - err := c.client.get(endpoint, &res) + err := c.request.get(endpoint, &res) return res, err } func (c *clusters) ListFailureDomains(clusterName string) (FailureDomainMap, error) { var domainData FailureDomainMap endpoint := c.client.endpoint(c.basePath, clusterName, "failureDomains") - err := c.client.get(endpoint, &domainData) + err := c.request.get(endpoint, &domainData) return domainData, err } func (c *clusters) DeleteFailureDomain(data FailureDomainData) error { endpoint := c.client.endpoint(c.basePath, data.ClusterName, "failureDomains", data.DomainName) - return c.client.delete(endpoint) + return c.request.delete(endpoint) } func (c *clusters) UpdateFailureDomain(data FailureDomainData) error { endpoint := c.client.endpoint(c.basePath, data.ClusterName, "failureDomains", data.DomainName) - return c.client.post(endpoint, &data) + return c.request.post(endpoint, &data) } diff --git a/pkg/pulsar/functions.go b/pkg/pulsar/functions.go index 514fd8e63..2f5a37f3e 100644 --- a/pkg/pulsar/functions.go +++ b/pkg/pulsar/functions.go @@ -108,13 +108,15 @@ type Functions interface { } type functions struct { - client *client + client *pulsarClient + request *client basePath string } -func (c *client) Functions() Functions { +func (c *pulsarClient) Functions() Functions { return &functions{ client: c, + request: c.client, basePath: "/functions", } } @@ -184,7 +186,7 @@ func (f *functions) CreateFunc(funcConf *FunctionConfig, fileName string) error } contentType := multiPartWriter.FormDataContentType() - err = f.client.postWithMultiPart(endpoint, nil, bodyBuf, contentType) + err = f.request.postWithMultiPart(endpoint, nil, bodyBuf, contentType) if err != nil { return err } @@ -229,7 +231,7 @@ func (f *functions) CreateFuncWithURL(funcConf *FunctionConfig, pkgURL string) e } contentType := multiPartWriter.FormDataContentType() - err = f.client.postWithMultiPart(endpoint, nil, bodyBuf, contentType) + err = f.request.postWithMultiPart(endpoint, nil, bodyBuf, contentType) if err != nil { return err } @@ -239,56 +241,56 @@ func (f *functions) CreateFuncWithURL(funcConf *FunctionConfig, pkgURL string) e func (f *functions) StopFunction(tenant, namespace, name string) error { endpoint := f.client.endpoint(f.basePath, tenant, namespace, name) - return f.client.post(endpoint+"/stop", "") + return f.request.post(endpoint+"/stop", "") } func (f *functions) StopFunctionWithID(tenant, namespace, name string, instanceID int) error { id := fmt.Sprintf("%d", instanceID) endpoint := f.client.endpoint(f.basePath, tenant, namespace, name, id) - return f.client.post(endpoint+"/stop", "") + return f.request.post(endpoint+"/stop", "") } func (f *functions) DeleteFunction(tenant, namespace, name string) error { endpoint := f.client.endpoint(f.basePath, tenant, namespace, name) - return f.client.delete(endpoint) + return f.request.delete(endpoint) } func (f *functions) StartFunction(tenant, namespace, name string) error { endpoint := f.client.endpoint(f.basePath, tenant, namespace, name) - return f.client.post(endpoint+"/start", "") + return f.request.post(endpoint+"/start", "") } func (f *functions) StartFunctionWithID(tenant, namespace, name string, instanceID int) error { id := fmt.Sprintf("%d", instanceID) endpoint := f.client.endpoint(f.basePath, tenant, namespace, name, id) - return f.client.post(endpoint+"/start", "") + return f.request.post(endpoint+"/start", "") } func (f *functions) RestartFunction(tenant, namespace, name string) error { endpoint := f.client.endpoint(f.basePath, tenant, namespace, name) - return f.client.post(endpoint+"/restart", "") + return f.request.post(endpoint+"/restart", "") } func (f *functions) RestartFunctionWithID(tenant, namespace, name string, instanceID int) error { id := fmt.Sprintf("%d", instanceID) endpoint := f.client.endpoint(f.basePath, tenant, namespace, name, id) - return f.client.post(endpoint+"/restart", "") + return f.request.post(endpoint+"/restart", "") } func (f *functions) GetFunctions(tenant, namespace string) ([]string, error) { var functions []string endpoint := f.client.endpoint(f.basePath, tenant, namespace) - err := f.client.get(endpoint, &functions) + err := f.request.get(endpoint, &functions) return functions, err } func (f *functions) GetFunction(tenant, namespace, name string) (FunctionConfig, error) { var functionConfig FunctionConfig endpoint := f.client.endpoint(f.basePath, tenant, namespace, name) - err := f.client.get(endpoint, &functionConfig) + err := f.request.get(endpoint, &functionConfig) return functionConfig, err } @@ -360,7 +362,7 @@ func (f *functions) UpdateFunction(functionConfig *FunctionConfig, fileName stri } contentType := multiPartWriter.FormDataContentType() - err = f.client.putWithMultiPart(endpoint, bodyBuf, contentType) + err = f.request.putWithMultiPart(endpoint, bodyBuf, contentType) if err != nil { return err } @@ -425,7 +427,7 @@ func (f *functions) UpdateFunctionWithURL(functionConfig *FunctionConfig, pkgURL } contentType := multiPartWriter.FormDataContentType() - err = f.client.putWithMultiPart(endpoint, bodyBuf, contentType) + err = f.request.putWithMultiPart(endpoint, bodyBuf, contentType) if err != nil { return err } @@ -436,7 +438,7 @@ func (f *functions) UpdateFunctionWithURL(functionConfig *FunctionConfig, pkgURL func (f *functions) GetFunctionStatus(tenant, namespace, name string) (FunctionStatus, error) { var functionStatus FunctionStatus endpoint := f.client.endpoint(f.basePath, tenant, namespace, name) - err := f.client.get(endpoint+"/status", &functionStatus) + err := f.request.get(endpoint+"/status", &functionStatus) return functionStatus, err } @@ -445,14 +447,14 @@ func (f *functions) GetFunctionStatusWithInstanceID(tenant, namespace, name stri var functionInstanceStatusData FunctionInstanceStatusData id := fmt.Sprintf("%d", instanceID) endpoint := f.client.endpoint(f.basePath, tenant, namespace, name, id) - err := f.client.get(endpoint+"/status", &functionInstanceStatusData) + err := f.request.get(endpoint+"/status", &functionInstanceStatusData) return functionInstanceStatusData, err } func (f *functions) GetFunctionStats(tenant, namespace, name string) (FunctionStats, error) { var functionStats FunctionStats endpoint := f.client.endpoint(f.basePath, tenant, namespace, name) - err := f.client.get(endpoint+"/stats", &functionStats) + err := f.request.get(endpoint+"/stats", &functionStats) return functionStats, err } @@ -461,14 +463,14 @@ func (f *functions) GetFunctionStatsWithInstanceID(tenant, namespace, name strin var functionInstanceStatsData FunctionInstanceStatsData id := fmt.Sprintf("%d", instanceID) endpoint := f.client.endpoint(f.basePath, tenant, namespace, name, id) - err := f.client.get(endpoint+"/stats", &functionInstanceStatsData) + err := f.request.get(endpoint+"/stats", &functionInstanceStatsData) return functionInstanceStatsData, err } func (f *functions) GetFunctionState(tenant, namespace, name, key string) (FunctionState, error) { var functionState FunctionState endpoint := f.client.endpoint(f.basePath, tenant, namespace, name, "state", key) - err := f.client.get(endpoint, &functionState) + err := f.request.get(endpoint, &functionState) return functionState, err } @@ -505,7 +507,7 @@ func (f *functions) PutFunctionState(tenant, namespace, name string, state Funct contentType := multiPartWriter.FormDataContentType() - err = f.client.postWithMultiPart(endpoint, nil, bodyBuf, contentType) + err = f.request.postWithMultiPart(endpoint, nil, bodyBuf, contentType) if err != nil { return err @@ -574,7 +576,7 @@ func (f *functions) TriggerFunction(tenant, namespace, name, topic, triggerValue contentType := multiPartWriter.FormDataContentType() var str string - err := f.client.postWithMultiPart(endpoint, &str, bodyBuf, contentType) + err := f.request.postWithMultiPart(endpoint, &str, bodyBuf, contentType) if err != nil { return "", err } diff --git a/pkg/pulsar/ledger.go b/pkg/pulsar/ledger.go new file mode 100644 index 000000000..0ca7ea874 --- /dev/null +++ b/pkg/pulsar/ledger.go @@ -0,0 +1,88 @@ +// 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 + +import ( + "strconv" +) + +type Ledger interface { + // Delete the specified ledger + Delete(int64) error + + // List all the ledgers and get the metadata + List(bool) (map[int64]string, error) + + // Get the metadata of a ledger + Get(int64) (map[int64]LedgerMetadata, error) + + // Read a range of entries from a ledger + Read(int64, int64, int64) (map[string]string, error) +} + +type ledger struct { + client *bookieClient + request *client + basePath string + params map[string]string +} + +func (c *bookieClient) Ledger() Ledger { + return &ledger{ + client: c, + request: c.client, + basePath: "/ledger", + params: make(map[string]string), + } +} + +func (c *ledger) Delete(ledgerID int64) error { + endpoint := c.client.bookieEndpoint(c.basePath, "/delete") + c.params["ledger_id"] = strconv.FormatInt(ledgerID, 10) + return c.request.deleteWithQueryParams(endpoint, c.params) +} + +func (c *ledger) List(showMeta bool) (map[int64]string, error) { + endpoint := c.client.bookieEndpoint(c.basePath, "list") + c.params["print_metadata"] = strconv.FormatBool(showMeta) + var metadata map[int64]string + _, err := c.request.getWithQueryParams(endpoint, &metadata, c.params, true) + return metadata, err +} + +func (c *ledger) Get(ledgerID int64) (map[int64]LedgerMetadata, error) { + endpoint := c.client.bookieEndpoint(c.basePath, "metadata") + c.params["ledger_id"] = strconv.FormatInt(ledgerID, 10) + var metadata map[int64]LedgerMetadata + _, err := c.request.getWithQueryParams(endpoint, &metadata, c.params, true) + return metadata, err +} + +func (c *ledger) Read(ledgerID int64, start int64, end int64) (map[string]string, error) { + endpoint := c.client.bookieEndpoint(c.basePath, "read") + c.params["ledger_id"] = strconv.FormatInt(ledgerID, 10) + if start >= 0 { + c.params["start_entry_id"] = strconv.FormatInt(start, 10) + } + if end >= 0 { + c.params["end_entry_id"] = strconv.FormatInt(end, 10) + } + info := make(map[string]string) + _, err := c.request.getWithQueryParams(endpoint, &info, c.params, true) + return info, err +} diff --git a/pkg/pulsar/namespace.go b/pkg/pulsar/namespace.go index 9ec2700a0..faf9bfcee 100644 --- a/pkg/pulsar/namespace.go +++ b/pkg/pulsar/namespace.go @@ -232,13 +232,15 @@ type Namespaces interface { } type namespaces struct { - client *client + client *pulsarClient + request *client basePath string } -func (c *client) Namespaces() Namespaces { +func (c *pulsarClient) Namespaces() Namespaces { return &namespaces{ client: c, + request: c.client, basePath: "/namespaces", } } @@ -246,7 +248,7 @@ func (c *client) Namespaces() 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) + err := n.request.get(endpoint, &namespaces) return namespaces, err } @@ -257,7 +259,7 @@ func (n *namespaces) GetTopics(namespace string) ([]string, error) { return nil, err } endpoint := n.client.endpoint(n.basePath, ns.String(), "topics") - err = n.client.get(endpoint, &topics) + err = n.request.get(endpoint, &topics) return topics, err } @@ -268,7 +270,7 @@ func (n *namespaces) GetPolicies(namespace string) (*Policies, error) { return nil, err } endpoint := n.client.endpoint(n.basePath, ns.String()) - err = n.client.get(endpoint, &police) + err = n.request.get(endpoint, &police) return &police, err } @@ -282,7 +284,7 @@ func (n *namespaces) CreateNsWithPolices(namespace string, policies Policies) er return err } endpoint := n.client.endpoint(n.basePath, ns.String()) - return n.client.put(endpoint, &policies) + return n.request.put(endpoint, &policies) } func (n *namespaces) CreateNsWithBundlesData(namespace string, bundleData *BundlesData) error { @@ -294,7 +296,7 @@ func (n *namespaces) CreateNsWithBundlesData(namespace string, bundleData *Bundl polices := new(Policies) polices.Bundles = bundleData - return n.client.put(endpoint, &polices) + return n.request.put(endpoint, &polices) } func (n *namespaces) CreateNamespace(namespace string) error { @@ -303,7 +305,7 @@ func (n *namespaces) CreateNamespace(namespace string) error { return err } endpoint := n.client.endpoint(n.basePath, ns.String()) - return n.client.put(endpoint, nil) + return n.request.put(endpoint, nil) } func (n *namespaces) DeleteNamespace(namespace string) error { @@ -312,7 +314,7 @@ func (n *namespaces) DeleteNamespace(namespace string) error { return err } endpoint := n.client.endpoint(n.basePath, ns.String()) - return n.client.delete(endpoint) + return n.request.delete(endpoint) } func (n *namespaces) DeleteNamespaceBundle(namespace string, bundleRange string) error { @@ -321,7 +323,7 @@ func (n *namespaces) DeleteNamespaceBundle(namespace string, bundleRange string) return err } endpoint := n.client.endpoint(n.basePath, ns.String(), bundleRange) - return n.client.delete(endpoint) + return n.request.delete(endpoint) } func (n *namespaces) GetNamespaceMessageTTL(namespace string) (int, error) { @@ -331,7 +333,7 @@ func (n *namespaces) GetNamespaceMessageTTL(namespace string) (int, error) { return 0, err } endpoint := n.client.endpoint(n.basePath, nsName.String(), "messageTTL") - err = n.client.get(endpoint, &ttl) + err = n.request.get(endpoint, &ttl) return ttl, err } @@ -342,7 +344,7 @@ func (n *namespaces) SetNamespaceMessageTTL(namespace string, ttlInSeconds int) } endpoint := n.client.endpoint(n.basePath, nsName.String(), "messageTTL") - return n.client.post(endpoint, &ttlInSeconds) + return n.request.post(endpoint, &ttlInSeconds) } func (n *namespaces) SetRetention(namespace string, policy RetentionPolicies) error { @@ -351,7 +353,7 @@ func (n *namespaces) SetRetention(namespace string, policy RetentionPolicies) er return err } endpoint := n.client.endpoint(n.basePath, nsName.String(), "retention") - return n.client.post(endpoint, &policy) + return n.request.post(endpoint, &policy) } func (n *namespaces) GetRetention(namespace string) (*RetentionPolicies, error) { @@ -361,7 +363,7 @@ func (n *namespaces) GetRetention(namespace string) (*RetentionPolicies, error) return nil, err } endpoint := n.client.endpoint(n.basePath, nsName.String(), "retention") - err = n.client.get(endpoint, &policy) + err = n.request.get(endpoint, &policy) return &policy, err } @@ -372,7 +374,7 @@ func (n *namespaces) GetBacklogQuotaMap(namespace string) (map[BacklogQuotaType] return nil, err } endpoint := n.client.endpoint(n.basePath, nsName.String(), "backlogQuotaMap") - err = n.client.get(endpoint, &backlogQuotaMap) + err = n.request.get(endpoint, &backlogQuotaMap) return backlogQuotaMap, err } @@ -382,7 +384,7 @@ func (n *namespaces) SetBacklogQuota(namespace string, backlogQuota BacklogQuota return err } endpoint := n.client.endpoint(n.basePath, nsName.String(), "backlogQuota") - return n.client.post(endpoint, &backlogQuota) + return n.request.post(endpoint, &backlogQuota) } func (n *namespaces) RemoveBacklogQuota(namespace string) error { @@ -394,17 +396,17 @@ func (n *namespaces) RemoveBacklogQuota(namespace string) error { params := map[string]string{ "backlogQuotaType": string(DestinationStorage), } - return n.client.deleteWithQueryParams(endpoint, nil, params) + return n.request.deleteWithQueryParams(endpoint, params) } func (n *namespaces) SetSchemaValidationEnforced(namespace NameSpaceName, schemaValidationEnforced bool) error { endpoint := n.client.endpoint(n.basePath, namespace.String(), "schemaValidationEnforced") - return n.client.post(endpoint, schemaValidationEnforced) + return n.request.post(endpoint, schemaValidationEnforced) } func (n *namespaces) GetSchemaValidationEnforced(namespace NameSpaceName) (bool, error) { endpoint := n.client.endpoint(n.basePath, namespace.String(), "schemaValidationEnforced") - r, err := n.client.getWithQueryParams(endpoint, nil, nil, false) + r, err := n.request.getWithQueryParams(endpoint, nil, nil, false) if err != nil { return false, err } @@ -414,14 +416,14 @@ func (n *namespaces) GetSchemaValidationEnforced(namespace NameSpaceName) (bool, func (n *namespaces) SetSchemaAutoUpdateCompatibilityStrategy(namespace NameSpaceName, strategy SchemaCompatibilityStrategy) error { endpoint := n.client.endpoint(n.basePath, namespace.String(), "schemaAutoUpdateCompatibilityStrategy") - return n.client.put(endpoint, strategy.String()) + return n.request.put(endpoint, strategy.String()) } func (n *namespaces) GetSchemaAutoUpdateCompatibilityStrategy(namespace NameSpaceName) (SchemaCompatibilityStrategy, error) { endpoint := n.client.endpoint(n.basePath, namespace.String(), "schemaAutoUpdateCompatibilityStrategy") - b, err := n.client.getWithQueryParams(endpoint, nil, nil, false) + b, err := n.request.getWithQueryParams(endpoint, nil, nil, false) if err != nil { return "", err } @@ -434,17 +436,17 @@ func (n *namespaces) GetSchemaAutoUpdateCompatibilityStrategy(namespace NameSpac func (n *namespaces) ClearOffloadDeleteLag(namespace NameSpaceName) error { endpoint := n.client.endpoint(n.basePath, namespace.String(), "offloadDeletionLagMs") - return n.client.delete(endpoint) + return n.request.delete(endpoint) } func (n *namespaces) SetOffloadDeleteLag(namespace NameSpaceName, timeMs int64) error { endpoint := n.client.endpoint(n.basePath, namespace.String(), "offloadDeletionLagMs") - return n.client.put(endpoint, timeMs) + return n.request.put(endpoint, timeMs) } func (n *namespaces) GetOffloadDeleteLag(namespace NameSpaceName) (int64, error) { endpoint := n.client.endpoint(n.basePath, namespace.String(), "offloadDeletionLagMs") - b, err := n.client.getWithQueryParams(endpoint, nil, nil, false) + b, err := n.request.getWithQueryParams(endpoint, nil, nil, false) if err != nil { return -1, err } @@ -453,12 +455,12 @@ func (n *namespaces) GetOffloadDeleteLag(namespace NameSpaceName) (int64, error) func (n *namespaces) SetMaxConsumersPerSubscription(namespace NameSpaceName, max int) error { endpoint := n.client.endpoint(n.basePath, namespace.String(), "maxConsumersPerSubscription") - return n.client.post(endpoint, max) + return n.request.post(endpoint, max) } func (n *namespaces) GetMaxConsumersPerSubscription(namespace NameSpaceName) (int, error) { endpoint := n.client.endpoint(n.basePath, namespace.String(), "maxConsumersPerSubscription") - b, err := n.client.getWithQueryParams(endpoint, nil, nil, false) + b, err := n.request.getWithQueryParams(endpoint, nil, nil, false) if err != nil { return -1, err } @@ -467,12 +469,12 @@ func (n *namespaces) GetMaxConsumersPerSubscription(namespace NameSpaceName) (in func (n *namespaces) SetOffloadThreshold(namespace NameSpaceName, threshold int64) error { endpoint := n.client.endpoint(n.basePath, namespace.String(), "offloadThreshold") - return n.client.put(endpoint, threshold) + return n.request.put(endpoint, threshold) } func (n *namespaces) GetOffloadThreshold(namespace NameSpaceName) (int64, error) { endpoint := n.client.endpoint(n.basePath, namespace.String(), "offloadThreshold") - b, err := n.client.getWithQueryParams(endpoint, nil, nil, false) + b, err := n.request.getWithQueryParams(endpoint, nil, nil, false) if err != nil { return -1, err } @@ -481,12 +483,12 @@ func (n *namespaces) GetOffloadThreshold(namespace NameSpaceName) (int64, error) func (n *namespaces) SetMaxConsumersPerTopic(namespace NameSpaceName, max int) error { endpoint := n.client.endpoint(n.basePath, namespace.String(), "maxConsumersPerTopic") - return n.client.post(endpoint, max) + return n.request.post(endpoint, max) } func (n *namespaces) GetMaxConsumersPerTopic(namespace NameSpaceName) (int, error) { endpoint := n.client.endpoint(n.basePath, namespace.String(), "maxConsumersPerTopic") - b, err := n.client.getWithQueryParams(endpoint, nil, nil, false) + b, err := n.request.getWithQueryParams(endpoint, nil, nil, false) if err != nil { return -1, err } @@ -495,12 +497,12 @@ func (n *namespaces) GetMaxConsumersPerTopic(namespace NameSpaceName) (int, erro func (n *namespaces) SetCompactionThreshold(namespace NameSpaceName, threshold int64) error { endpoint := n.client.endpoint(n.basePath, namespace.String(), "compactionThreshold") - return n.client.put(endpoint, threshold) + return n.request.put(endpoint, threshold) } func (n *namespaces) GetCompactionThreshold(namespace NameSpaceName) (int64, error) { endpoint := n.client.endpoint(n.basePath, namespace.String(), "compactionThreshold") - b, err := n.client.getWithQueryParams(endpoint, nil, nil, false) + b, err := n.request.getWithQueryParams(endpoint, nil, nil, false) if err != nil { return -1, err } @@ -509,12 +511,12 @@ func (n *namespaces) GetCompactionThreshold(namespace NameSpaceName) (int64, err func (n *namespaces) SetMaxProducersPerTopic(namespace NameSpaceName, max int) error { endpoint := n.client.endpoint(n.basePath, namespace.String(), "maxProducersPerTopic") - return n.client.post(endpoint, max) + return n.request.post(endpoint, max) } func (n *namespaces) GetMaxProducersPerTopic(namespace NameSpaceName) (int, error) { endpoint := n.client.endpoint(n.basePath, namespace.String(), "maxProducersPerTopic") - b, err := n.client.getWithQueryParams(endpoint, nil, nil, false) + b, err := n.request.getWithQueryParams(endpoint, nil, nil, false) if err != nil { return -1, err } @@ -528,7 +530,7 @@ func (n *namespaces) GetNamespaceReplicationClusters(namespace string) ([]string return nil, err } endpoint := n.client.endpoint(n.basePath, nsName.String(), "replication") - err = n.client.get(endpoint, &data) + err = n.request.get(endpoint, &data) return data, err } @@ -538,7 +540,7 @@ func (n *namespaces) SetNamespaceReplicationClusters(namespace string, clusterId return err } endpoint := n.client.endpoint(n.basePath, nsName.String(), "replication") - return n.client.post(endpoint, &clusterIds) + return n.request.post(endpoint, &clusterIds) } func (n *namespaces) SetNamespaceAntiAffinityGroup(namespace string, namespaceAntiAffinityGroup string) error { @@ -547,7 +549,7 @@ func (n *namespaces) SetNamespaceAntiAffinityGroup(namespace string, namespaceAn return err } endpoint := n.client.endpoint(n.basePath, nsName.String(), "antiAffinity") - return n.client.post(endpoint, namespaceAntiAffinityGroup) + return n.request.post(endpoint, namespaceAntiAffinityGroup) } func (n *namespaces) GetAntiAffinityNamespaces(tenant, cluster, namespaceAntiAffinityGroup string) ([]string, error) { @@ -556,7 +558,7 @@ func (n *namespaces) GetAntiAffinityNamespaces(tenant, cluster, namespaceAntiAff params := map[string]string{ "property": tenant, } - _, err := n.client.getWithQueryParams(endpoint, &data, params, false) + _, err := n.request.getWithQueryParams(endpoint, &data, params, false) return data, err } @@ -566,7 +568,7 @@ func (n *namespaces) GetNamespaceAntiAffinityGroup(namespace string) (string, er return "", err } endpoint := n.client.endpoint(n.basePath, nsName.String(), "antiAffinity") - data, err := n.client.getWithQueryParams(endpoint, nil, nil, false) + data, err := n.request.getWithQueryParams(endpoint, nil, nil, false) return string(data), err } @@ -576,7 +578,7 @@ func (n *namespaces) DeleteNamespaceAntiAffinityGroup(namespace string) error { return err } endpoint := n.client.endpoint(n.basePath, nsName.String(), "antiAffinity") - return n.client.delete(endpoint) + return n.request.delete(endpoint) } func (n *namespaces) SetDeduplicationStatus(namespace string, enableDeduplication bool) error { @@ -585,7 +587,7 @@ func (n *namespaces) SetDeduplicationStatus(namespace string, enableDeduplicatio return err } endpoint := n.client.endpoint(n.basePath, nsName.String(), "deduplication") - return n.client.post(endpoint, enableDeduplication) + return n.request.post(endpoint, enableDeduplication) } func (n *namespaces) SetPersistence(namespace string, persistence PersistencePolicies) error { @@ -594,7 +596,7 @@ func (n *namespaces) SetPersistence(namespace string, persistence PersistencePol return err } endpoint := n.client.endpoint(n.basePath, nsName.String(), "persistence") - return n.client.post(endpoint, &persistence) + return n.request.post(endpoint, &persistence) } func (n *namespaces) SetBookieAffinityGroup(namespace string, bookieAffinityGroup BookieAffinityGroupData) error { @@ -603,7 +605,7 @@ func (n *namespaces) SetBookieAffinityGroup(namespace string, bookieAffinityGrou return err } endpoint := n.client.endpoint(n.basePath, nsName.String(), "persistence", "bookieAffinity") - return n.client.post(endpoint, &bookieAffinityGroup) + return n.request.post(endpoint, &bookieAffinityGroup) } func (n *namespaces) DeleteBookieAffinityGroup(namespace string) error { @@ -612,7 +614,7 @@ func (n *namespaces) DeleteBookieAffinityGroup(namespace string) error { return err } endpoint := n.client.endpoint(n.basePath, nsName.String(), "persistence", "bookieAffinity") - return n.client.delete(endpoint) + return n.request.delete(endpoint) } func (n *namespaces) GetBookieAffinityGroup(namespace string) (*BookieAffinityGroupData, error) { @@ -622,7 +624,7 @@ func (n *namespaces) GetBookieAffinityGroup(namespace string) (*BookieAffinityGr return nil, err } endpoint := n.client.endpoint(n.basePath, nsName.String(), "persistence", "bookieAffinity") - err = n.client.get(endpoint, &data) + err = n.request.get(endpoint, &data) return &data, err } @@ -633,7 +635,7 @@ func (n *namespaces) GetPersistence(namespace string) (*PersistencePolicies, err return nil, err } endpoint := n.client.endpoint(n.basePath, nsName.String(), "persistence") - err = n.client.get(endpoint, &persistence) + err = n.request.get(endpoint, &persistence) return &persistence, err } @@ -643,7 +645,7 @@ func (n *namespaces) Unload(namespace string) error { return err } endpoint := n.client.endpoint(n.basePath, nsName.String(), "unload") - return n.client.put(endpoint, "") + return n.request.put(endpoint, "") } func (n *namespaces) UnloadNamespaceBundle(namespace, bundle string) error { @@ -652,7 +654,7 @@ func (n *namespaces) UnloadNamespaceBundle(namespace, bundle string) error { return err } endpoint := n.client.endpoint(n.basePath, nsName.String(), bundle, "unload") - return n.client.put(endpoint, "") + return n.request.put(endpoint, "") } func (n *namespaces) SplitNamespaceBundle(namespace, bundle string, unloadSplitBundles bool) error { @@ -664,13 +666,13 @@ func (n *namespaces) SplitNamespaceBundle(namespace, bundle string, unloadSplitB params := map[string]string{ "unload": strconv.FormatBool(unloadSplitBundles), } - return n.client.putWithQueryParams(endpoint, "", nil, params) + return n.request.putWithQueryParams(endpoint, "", nil, params) } func (n *namespaces) GetNamespacePermissions(namespace NameSpaceName) (map[string][]AuthAction, error) { endpoint := n.client.endpoint(n.basePath, namespace.String(), "permissions") var permissions map[string][]AuthAction - err := n.client.get(endpoint, &permissions) + err := n.request.get(endpoint, &permissions) return permissions, err } @@ -680,110 +682,110 @@ func (n *namespaces) GrantNamespacePermission(namespace NameSpaceName, role stri for _, v := range action { s = append(s, v.String()) } - return n.client.post(endpoint, s) + return n.request.post(endpoint, s) } func (n *namespaces) RevokeNamespacePermission(namespace NameSpaceName, role string) error { endpoint := n.client.endpoint(n.basePath, namespace.String(), "permissions", role) - return n.client.delete(endpoint) + return n.request.delete(endpoint) } func (n *namespaces) GrantSubPermission(namespace NameSpaceName, sName string, roles []string) error { endpoint := n.client.endpoint(n.basePath, namespace.String(), "permissions", "subscription", sName) - return n.client.post(endpoint, roles) + return n.request.post(endpoint, roles) } func (n *namespaces) RevokeSubPermission(namespace NameSpaceName, sName, role string) error { endpoint := n.client.endpoint(n.basePath, namespace.String(), "permissions", "subscription", sName, role) - return n.client.delete(endpoint) + return n.request.delete(endpoint) } func (n *namespaces) SetSubscriptionAuthMode(namespace NameSpaceName, mode SubscriptionAuthMode) error { endpoint := n.client.endpoint(n.basePath, namespace.String(), "subscriptionAuthMode") - return n.client.post(endpoint, mode.String()) + return n.request.post(endpoint, mode.String()) } func (n *namespaces) SetEncryptionRequiredStatus(namespace NameSpaceName, encrypt bool) error { endpoint := n.client.endpoint(n.basePath, namespace.String(), "encryptionRequired") - return n.client.post(endpoint, strconv.FormatBool(encrypt)) + return n.request.post(endpoint, strconv.FormatBool(encrypt)) } func (n *namespaces) UnsubscribeNamespace(namespace NameSpaceName, sName string) error { endpoint := n.client.endpoint(n.basePath, namespace.String(), "unsubscribe", url.QueryEscape(sName)) - return n.client.post(endpoint, "") + return n.request.post(endpoint, "") } func (n *namespaces) UnsubscribeNamespaceBundle(namespace NameSpaceName, bundle, sName string) error { endpoint := n.client.endpoint(n.basePath, namespace.String(), bundle, "unsubscribe", url.QueryEscape(sName)) - return n.client.post(endpoint, "") + return n.request.post(endpoint, "") } func (n *namespaces) ClearNamespaceBundleBacklogForSubscription(namespace NameSpaceName, bundle, sName string) error { endpoint := n.client.endpoint(n.basePath, namespace.String(), bundle, "clearBacklog", url.QueryEscape(sName)) - return n.client.post(endpoint, "") + return n.request.post(endpoint, "") } func (n *namespaces) ClearNamespaceBundleBacklog(namespace NameSpaceName, bundle string) error { endpoint := n.client.endpoint(n.basePath, namespace.String(), bundle, "clearBacklog") - return n.client.post(endpoint, "") + return n.request.post(endpoint, "") } func (n *namespaces) ClearNamespaceBacklogForSubscription(namespace NameSpaceName, sName string) error { endpoint := n.client.endpoint(n.basePath, namespace.String(), "clearBacklog", url.QueryEscape(sName)) - return n.client.post(endpoint, "") + return n.request.post(endpoint, "") } func (n *namespaces) ClearNamespaceBacklog(namespace NameSpaceName) error { endpoint := n.client.endpoint(n.basePath, namespace.String(), "clearBacklog") - return n.client.post(endpoint, "") + return n.request.post(endpoint, "") } func (n *namespaces) SetReplicatorDispatchRate(namespace NameSpaceName, rate DispatchRate) error { endpoint := n.client.endpoint(n.basePath, namespace.String(), "replicatorDispatchRate") - return n.client.post(endpoint, rate) + return n.request.post(endpoint, rate) } func (n *namespaces) GetReplicatorDispatchRate(namespace NameSpaceName) (DispatchRate, error) { endpoint := n.client.endpoint(n.basePath, namespace.String(), "replicatorDispatchRate") var rate DispatchRate - err := n.client.get(endpoint, &rate) + err := n.request.get(endpoint, &rate) return rate, err } func (n *namespaces) SetSubscriptionDispatchRate(namespace NameSpaceName, rate DispatchRate) error { endpoint := n.client.endpoint(n.basePath, namespace.String(), "subscriptionDispatchRate") - return n.client.post(endpoint, rate) + return n.request.post(endpoint, rate) } func (n *namespaces) GetSubscriptionDispatchRate(namespace NameSpaceName) (DispatchRate, error) { endpoint := n.client.endpoint(n.basePath, namespace.String(), "subscriptionDispatchRate") var rate DispatchRate - err := n.client.get(endpoint, &rate) + err := n.request.get(endpoint, &rate) return rate, err } func (n *namespaces) SetSubscribeRate(namespace NameSpaceName, rate SubscribeRate) error { endpoint := n.client.endpoint(n.basePath, namespace.String(), "subscribeRate") - return n.client.post(endpoint, rate) + return n.request.post(endpoint, rate) } func (n *namespaces) GetSubscribeRate(namespace NameSpaceName) (SubscribeRate, error) { endpoint := n.client.endpoint(n.basePath, namespace.String(), "subscribeRate") var rate SubscribeRate - err := n.client.get(endpoint, &rate) + err := n.request.get(endpoint, &rate) return rate, err } func (n *namespaces) SetDispatchRate(namespace NameSpaceName, rate DispatchRate) error { endpoint := n.client.endpoint(n.basePath, namespace.String(), "dispatchRate") - return n.client.post(endpoint, rate) + return n.request.post(endpoint, rate) } func (n *namespaces) GetDispatchRate(namespace NameSpaceName) (DispatchRate, error) { endpoint := n.client.endpoint(n.basePath, namespace.String(), "dispatchRate") var rate DispatchRate - err := n.client.get(endpoint, &rate) + err := n.request.get(endpoint, &rate) return rate, err } diff --git a/pkg/pulsar/schema.go b/pkg/pulsar/schema.go index 880992e73..9e192dd0d 100644 --- a/pkg/pulsar/schema.go +++ b/pkg/pulsar/schema.go @@ -40,13 +40,15 @@ type Schema interface { } type schemas struct { - client *client + client *pulsarClient + request *client basePath string } -func (c *client) Schemas() Schema { +func (c *pulsarClient) Schemas() Schema { return &schemas{ client: c, + request: c.client, basePath: "/schemas", } } @@ -59,7 +61,7 @@ func (s *schemas) GetSchemaInfo(topic string) (*SchemaInfo, error) { var response GetSchemaResponse endpoint := s.client.endpoint(s.basePath, topicName.tenant, topicName.namespace, topicName.GetEncodedTopic(), "schema") - err = s.client.get(endpoint, &response) + err = s.request.get(endpoint, &response) if err != nil { return nil, err } @@ -77,7 +79,7 @@ func (s *schemas) GetSchemaInfoWithVersion(topic string) (*SchemaInfoWithVersion endpoint := s.client.endpoint(s.basePath, topicName.tenant, topicName.namespace, topicName.GetEncodedTopic(), "schema") - err = s.client.get(endpoint, &response) + err = s.request.get(endpoint, &response) if err != nil { fmt.Println("err:", err.Error()) return nil, err @@ -97,7 +99,7 @@ func (s *schemas) GetSchemaInfoByVersion(topic string, version int64) (*SchemaIn endpoint := s.client.endpoint(s.basePath, topicName.tenant, topicName.namespace, topicName.GetEncodedTopic(), "schema", strconv.FormatInt(version, 10)) - err = s.client.get(endpoint, &response) + err = s.request.get(endpoint, &response) if err != nil { return nil, err } @@ -117,7 +119,7 @@ func (s *schemas) DeleteSchema(topic string) error { fmt.Println(endpoint) - return s.client.delete(endpoint) + return s.request.delete(endpoint) } func (s *schemas) CreateSchemaByPayload(topic string, schemaPayload PostSchemaPayload) error { @@ -129,5 +131,5 @@ func (s *schemas) CreateSchemaByPayload(topic string, schemaPayload PostSchemaPa endpoint := s.client.endpoint(s.basePath, topicName.tenant, topicName.namespace, topicName.GetEncodedTopic(), "schema") - return s.client.post(endpoint, &schemaPayload) + return s.request.post(endpoint, &schemaPayload) } diff --git a/pkg/pulsar/sinks.go b/pkg/pulsar/sinks.go index 936bd4ace..85132f914 100644 --- a/pkg/pulsar/sinks.go +++ b/pkg/pulsar/sinks.go @@ -83,13 +83,15 @@ type Sinks interface { } type sinks struct { - client *client + client *pulsarClient + request *client basePath string } -func (c *client) Sinks() Sinks { +func (c *pulsarClient) Sinks() Sinks { return &sinks{ client: c, + request: c.client, basePath: "/sinks", } } @@ -111,14 +113,14 @@ func (s *sinks) createTextFromFiled(w *multipart.Writer, value string) (io.Write func (s *sinks) ListSinks(tenant, namespace string) ([]string, error) { var sinks []string endpoint := s.client.endpoint(s.basePath, tenant, namespace) - err := s.client.get(endpoint, &sinks) + err := s.request.get(endpoint, &sinks) return sinks, err } func (s *sinks) GetSink(tenant, namespace, sink string) (SinkConfig, error) { var sinkConfig SinkConfig endpoint := s.client.endpoint(s.basePath, tenant, namespace, sink) - err := s.client.get(endpoint, &sinkConfig) + err := s.request.get(endpoint, &sinkConfig) return sinkConfig, err } @@ -172,7 +174,7 @@ func (s *sinks) CreateSink(config *SinkConfig, fileName string) error { } contentType := multiPartWriter.FormDataContentType() - err = s.client.postWithMultiPart(endpoint, nil, bodyBuf, contentType) + err = s.request.postWithMultiPart(endpoint, nil, bodyBuf, contentType) if err != nil { return err } @@ -217,7 +219,7 @@ func (s *sinks) CreateSinkWithURL(config *SinkConfig, pkgURL string) error { } contentType := multiPartWriter.FormDataContentType() - err = s.client.postWithMultiPart(endpoint, nil, bodyBuf, contentType) + err = s.request.postWithMultiPart(endpoint, nil, bodyBuf, contentType) if err != nil { return err } @@ -292,7 +294,7 @@ func (s *sinks) UpdateSink(config *SinkConfig, fileName string, updateOptions *U } contentType := multiPartWriter.FormDataContentType() - err = s.client.putWithMultiPart(endpoint, bodyBuf, contentType) + err = s.request.putWithMultiPart(endpoint, bodyBuf, contentType) if err != nil { return err } @@ -356,7 +358,7 @@ func (s *sinks) UpdateSinkWithURL(config *SinkConfig, pkgURL string, updateOptio } contentType := multiPartWriter.FormDataContentType() - err = s.client.putWithMultiPart(endpoint, bodyBuf, contentType) + err = s.request.putWithMultiPart(endpoint, bodyBuf, contentType) if err != nil { return err } @@ -366,13 +368,13 @@ func (s *sinks) UpdateSinkWithURL(config *SinkConfig, pkgURL string, updateOptio func (s *sinks) DeleteSink(tenant, namespace, sink string) error { endpoint := s.client.endpoint(s.basePath, tenant, namespace, sink) - return s.client.delete(endpoint) + return s.request.delete(endpoint) } func (s *sinks) GetSinkStatus(tenant, namespace, sink string) (SinkStatus, error) { var sinkStatus SinkStatus endpoint := s.client.endpoint(s.basePath, tenant, namespace, sink) - err := s.client.get(endpoint+"/status", &sinkStatus) + err := s.request.get(endpoint+"/status", &sinkStatus) return sinkStatus, err } @@ -380,54 +382,54 @@ func (s *sinks) GetSinkStatusWithID(tenant, namespace, sink string, id int) (Sin var sinkInstanceStatusData SinkInstanceStatusData instanceID := fmt.Sprintf("%d", id) endpoint := s.client.endpoint(s.basePath, tenant, namespace, sink, instanceID) - err := s.client.get(endpoint+"/status", &sinkInstanceStatusData) + err := s.request.get(endpoint+"/status", &sinkInstanceStatusData) return sinkInstanceStatusData, err } func (s *sinks) RestartSink(tenant, namespace, sink string) error { endpoint := s.client.endpoint(s.basePath, tenant, namespace, sink) - return s.client.post(endpoint+"/restart", "") + return s.request.post(endpoint+"/restart", "") } func (s *sinks) RestartSinkWithID(tenant, namespace, sink string, instanceID int) error { id := fmt.Sprintf("%d", instanceID) endpoint := s.client.endpoint(s.basePath, tenant, namespace, sink, id) - return s.client.post(endpoint+"/restart", "") + return s.request.post(endpoint+"/restart", "") } func (s *sinks) StopSink(tenant, namespace, sink string) error { endpoint := s.client.endpoint(s.basePath, tenant, namespace, sink) - return s.client.post(endpoint+"/stop", "") + return s.request.post(endpoint+"/stop", "") } func (s *sinks) StopSinkWithID(tenant, namespace, sink string, instanceID int) error { id := fmt.Sprintf("%d", instanceID) endpoint := s.client.endpoint(s.basePath, tenant, namespace, sink, id) - return s.client.post(endpoint+"/stop", "") + return s.request.post(endpoint+"/stop", "") } func (s *sinks) StartSink(tenant, namespace, sink string) error { endpoint := s.client.endpoint(s.basePath, tenant, namespace, sink) - return s.client.post(endpoint+"/start", "") + return s.request.post(endpoint+"/start", "") } func (s *sinks) StartSinkWithID(tenant, namespace, sink string, instanceID int) error { id := fmt.Sprintf("%d", instanceID) endpoint := s.client.endpoint(s.basePath, tenant, namespace, sink, id) - return s.client.post(endpoint+"/start", "") + return s.request.post(endpoint+"/start", "") } func (s *sinks) GetBuiltInSinks() ([]*ConnectorDefinition, error) { var connectorDefinition []*ConnectorDefinition endpoint := s.client.endpoint(s.basePath, "builtinSinks") - err := s.client.get(endpoint, &connectorDefinition) + err := s.request.get(endpoint, &connectorDefinition) return connectorDefinition, err } func (s *sinks) ReloadBuiltInSinks() error { endpoint := s.client.endpoint(s.basePath, "reloadBuiltInSinks") - return s.client.post(endpoint, "") + return s.request.post(endpoint, "") } diff --git a/pkg/pulsar/sources.go b/pkg/pulsar/sources.go index ece4b5468..800c92b1c 100644 --- a/pkg/pulsar/sources.go +++ b/pkg/pulsar/sources.go @@ -83,13 +83,15 @@ type Sources interface { } type sources struct { - client *client + client *pulsarClient + request *client basePath string } -func (c *client) Sources() Sources { +func (c *pulsarClient) Sources() Sources { return &sources{ client: c, + request: c.client, basePath: "/sources", } } @@ -111,14 +113,14 @@ func (s *sources) createTextFromFiled(w *multipart.Writer, value string) (io.Wri func (s *sources) ListSources(tenant, namespace string) ([]string, error) { var sources []string endpoint := s.client.endpoint(s.basePath, tenant, namespace) - err := s.client.get(endpoint, &sources) + err := s.request.get(endpoint, &sources) return sources, err } func (s *sources) GetSource(tenant, namespace, source string) (SourceConfig, error) { var sourceConfig SourceConfig endpoint := s.client.endpoint(s.basePath, tenant, namespace, source) - err := s.client.get(endpoint, &sourceConfig) + err := s.request.get(endpoint, &sourceConfig) return sourceConfig, err } @@ -172,7 +174,7 @@ func (s *sources) CreateSource(config *SourceConfig, fileName string) error { } contentType := multiPartWriter.FormDataContentType() - err = s.client.postWithMultiPart(endpoint, nil, bodyBuf, contentType) + err = s.request.postWithMultiPart(endpoint, nil, bodyBuf, contentType) if err != nil { return err } @@ -217,7 +219,7 @@ func (s *sources) CreateSourceWithURL(config *SourceConfig, pkgURL string) error } contentType := multiPartWriter.FormDataContentType() - err = s.client.postWithMultiPart(endpoint, nil, bodyBuf, contentType) + err = s.request.postWithMultiPart(endpoint, nil, bodyBuf, contentType) if err != nil { return err } @@ -292,7 +294,7 @@ func (s *sources) UpdateSource(config *SourceConfig, fileName string, updateOpti } contentType := multiPartWriter.FormDataContentType() - err = s.client.putWithMultiPart(endpoint, bodyBuf, contentType) + err = s.request.putWithMultiPart(endpoint, bodyBuf, contentType) if err != nil { return err } @@ -356,7 +358,7 @@ func (s *sources) UpdateSourceWithURL(config *SourceConfig, pkgURL string, updat } contentType := multiPartWriter.FormDataContentType() - err = s.client.putWithMultiPart(endpoint, bodyBuf, contentType) + err = s.request.putWithMultiPart(endpoint, bodyBuf, contentType) if err != nil { return err } @@ -366,13 +368,13 @@ func (s *sources) UpdateSourceWithURL(config *SourceConfig, pkgURL string, updat func (s *sources) DeleteSource(tenant, namespace, source string) error { endpoint := s.client.endpoint(s.basePath, tenant, namespace, source) - return s.client.delete(endpoint) + return s.request.delete(endpoint) } func (s *sources) GetSourceStatus(tenant, namespace, source string) (SourceStatus, error) { var sourceStatus SourceStatus endpoint := s.client.endpoint(s.basePath, tenant, namespace, source) - err := s.client.get(endpoint+"/status", &sourceStatus) + err := s.request.get(endpoint+"/status", &sourceStatus) return sourceStatus, err } @@ -380,54 +382,54 @@ func (s *sources) GetSourceStatusWithID(tenant, namespace, source string, id int var sourceInstanceStatusData SourceInstanceStatusData instanceID := fmt.Sprintf("%d", id) endpoint := s.client.endpoint(s.basePath, tenant, namespace, source, instanceID) - err := s.client.get(endpoint+"/status", &sourceInstanceStatusData) + err := s.request.get(endpoint+"/status", &sourceInstanceStatusData) return sourceInstanceStatusData, err } func (s *sources) RestartSource(tenant, namespace, source string) error { endpoint := s.client.endpoint(s.basePath, tenant, namespace, source) - return s.client.post(endpoint+"/restart", "") + return s.request.post(endpoint+"/restart", "") } func (s *sources) RestartSourceWithID(tenant, namespace, source string, instanceID int) error { id := fmt.Sprintf("%d", instanceID) endpoint := s.client.endpoint(s.basePath, tenant, namespace, source, id) - return s.client.post(endpoint+"/restart", "") + return s.request.post(endpoint+"/restart", "") } func (s *sources) StopSource(tenant, namespace, source string) error { endpoint := s.client.endpoint(s.basePath, tenant, namespace, source) - return s.client.post(endpoint+"/stop", "") + return s.request.post(endpoint+"/stop", "") } func (s *sources) StopSourceWithID(tenant, namespace, source string, instanceID int) error { id := fmt.Sprintf("%d", instanceID) endpoint := s.client.endpoint(s.basePath, tenant, namespace, source, id) - return s.client.post(endpoint+"/stop", "") + return s.request.post(endpoint+"/stop", "") } func (s *sources) StartSource(tenant, namespace, source string) error { endpoint := s.client.endpoint(s.basePath, tenant, namespace, source) - return s.client.post(endpoint+"/start", "") + return s.request.post(endpoint+"/start", "") } func (s *sources) StartSourceWithID(tenant, namespace, source string, instanceID int) error { id := fmt.Sprintf("%d", instanceID) endpoint := s.client.endpoint(s.basePath, tenant, namespace, source, id) - return s.client.post(endpoint+"/start", "") + return s.request.post(endpoint+"/start", "") } func (s *sources) GetBuiltInSources() ([]*ConnectorDefinition, error) { var connectorDefinition []*ConnectorDefinition endpoint := s.client.endpoint(s.basePath, "builtinsources") - err := s.client.get(endpoint, &connectorDefinition) + err := s.request.get(endpoint, &connectorDefinition) return connectorDefinition, err } func (s *sources) ReloadBuiltInSources() error { endpoint := s.client.endpoint(s.basePath, "reloadBuiltInSources") - return s.client.post(endpoint, "") + return s.request.post(endpoint, "") } diff --git a/pkg/pulsar/subscription.go b/pkg/pulsar/subscription.go index 8b8e4d143..4be843ac8 100644 --- a/pkg/pulsar/subscription.go +++ b/pkg/pulsar/subscription.go @@ -44,14 +44,16 @@ type Subscriptions interface { } type subscriptions struct { - client *client + client *pulsarClient + request *client basePath string SubPath string } -func (c *client) Subscriptions() Subscriptions { +func (c *pulsarClient) Subscriptions() Subscriptions { return &subscriptions{ client: c, + request: c.client, basePath: "", SubPath: "subscription", } @@ -59,57 +61,57 @@ func (c *client) Subscriptions() Subscriptions { func (s *subscriptions) Create(topic TopicName, sName string, messageID MessageID) error { endpoint := s.client.endpoint(s.basePath, topic.GetRestPath(), s.SubPath, url.QueryEscape(sName)) - return s.client.put(endpoint, messageID) + return s.request.put(endpoint, messageID) } func (s *subscriptions) Delete(topic TopicName, sName string) error { endpoint := s.client.endpoint(s.basePath, topic.GetRestPath(), s.SubPath, url.QueryEscape(sName)) - return s.client.delete(endpoint) + return s.request.delete(endpoint) } func (s *subscriptions) List(topic TopicName) ([]string, error) { endpoint := s.client.endpoint(s.basePath, topic.GetRestPath(), "subscriptions") var list []string - return list, s.client.get(endpoint, &list) + return list, s.request.get(endpoint, &list) } func (s *subscriptions) ResetCursorToMessageID(topic TopicName, sName string, id MessageID) error { endpoint := s.client.endpoint(s.basePath, topic.GetRestPath(), s.SubPath, url.QueryEscape(sName), "resetcursor") - return s.client.post(endpoint, id) + return s.request.post(endpoint, id) } func (s *subscriptions) ResetCursorToTimestamp(topic TopicName, sName string, timestamp int64) error { endpoint := s.client.endpoint( s.basePath, topic.GetRestPath(), s.SubPath, url.QueryEscape(sName), "resetcursor", strconv.FormatInt(timestamp, 10)) - return s.client.post(endpoint, "") + return s.request.post(endpoint, "") } func (s *subscriptions) ClearBacklog(topic TopicName, sName string) error { endpoint := s.client.endpoint( s.basePath, topic.GetRestPath(), s.SubPath, url.QueryEscape(sName), "skip_all") - return s.client.post(endpoint, "") + return s.request.post(endpoint, "") } func (s *subscriptions) SkipMessages(topic TopicName, sName string, n int64) error { endpoint := s.client.endpoint( s.basePath, topic.GetRestPath(), s.SubPath, url.QueryEscape(sName), "skip", strconv.FormatInt(n, 10)) - return s.client.post(endpoint, "") + return s.request.post(endpoint, "") } func (s *subscriptions) ExpireMessages(topic TopicName, sName string, expire int64) error { endpoint := s.client.endpoint( s.basePath, topic.GetRestPath(), s.SubPath, url.QueryEscape(sName), "expireMessages", strconv.FormatInt(expire, 10)) - return s.client.post(endpoint, "") + return s.request.post(endpoint, "") } func (s *subscriptions) ExpireAllMessages(topic TopicName, expire int64) error { endpoint := s.client.endpoint( s.basePath, topic.GetRestPath(), "all_subscription", "expireMessages", strconv.FormatInt(expire, 10)) - return s.client.post(endpoint, "") + return s.request.post(endpoint, "") } func (s *subscriptions) PeekMessages(topic TopicName, sName string, n int) ([]*Message, error) { @@ -132,12 +134,12 @@ func (s *subscriptions) PeekMessages(topic TopicName, sName string, n int) ([]*M func (s *subscriptions) peekNthMessage(topic TopicName, sName string, pos int) ([]*Message, error) { endpoint := s.client.endpoint(s.basePath, topic.GetRestPath(), "subscription", url.QueryEscape(sName), "position", strconv.Itoa(pos)) - req, err := s.client.newRequest(http.MethodGet, endpoint) + req, err := s.request.newRequest(http.MethodGet, endpoint) if err != nil { return nil, err } - resp, err := checkSuccessful(s.client.doRequest(req)) + resp, err := checkSuccessful(s.request.doRequest(req)) if err != nil { return nil, err } diff --git a/pkg/pulsar/topic.go b/pkg/pulsar/topic.go index ec00a4353..62f883c05 100644 --- a/pkg/pulsar/topic.go +++ b/pkg/pulsar/topic.go @@ -47,16 +47,18 @@ type Topics interface { } type topics struct { - client *client + client *pulsarClient + request *client basePath string persistentPath string nonPersistentPath string lookupPath string } -func (c *client) Topics() Topics { +func (c *pulsarClient) Topics() Topics { return &topics{ client: c, + request: c.client, basePath: "", persistentPath: "/persistent", nonPersistentPath: "/non-persistent", @@ -69,7 +71,7 @@ func (t *topics) Create(topic TopicName, partitions int) error { if partitions == 0 { endpoint = t.client.endpoint(t.basePath, topic.GetRestPath()) } - return t.client.put(endpoint, partitions) + return t.request.put(endpoint, partitions) } func (t *topics) Delete(topic TopicName, force bool, nonPartitioned bool) error { @@ -80,18 +82,18 @@ func (t *topics) Delete(topic TopicName, force bool, nonPartitioned bool) error params := map[string]string{ "force": strconv.FormatBool(force), } - return t.client.deleteWithQueryParams(endpoint, nil, params) + return t.request.deleteWithQueryParams(endpoint, params) } func (t *topics) Update(topic TopicName, partitions int) error { endpoint := t.client.endpoint(t.basePath, topic.GetRestPath(), "partitions") - return t.client.post(endpoint, partitions) + return t.request.post(endpoint, partitions) } func (t *topics) GetMetadata(topic TopicName) (PartitionedTopicMetadata, error) { endpoint := t.client.endpoint(t.basePath, topic.GetRestPath(), "partitions") var partitionedMeta PartitionedTopicMetadata - err := t.client.get(endpoint, &partitionedMeta) + err := t.request.get(endpoint, &partitionedMeta) return partitionedMeta, err } @@ -135,21 +137,21 @@ func (t *topics) List(namespace NameSpaceName) ([]string, []string, error) { func (t *topics) getTopics(endpoint string, out chan<- []string, err chan<- error) { var topics []string - err <- t.client.get(endpoint, &topics) + err <- t.request.get(endpoint, &topics) out <- topics } func (t *topics) GetInternalInfo(topic TopicName) (ManagedLedgerInfo, error) { endpoint := t.client.endpoint(t.basePath, topic.GetRestPath(), "internal-info") var info ManagedLedgerInfo - err := t.client.get(endpoint, &info) + err := t.request.get(endpoint, &info) return info, err } func (t *topics) GetPermissions(topic TopicName) (map[string][]AuthAction, error) { var permissions map[string][]AuthAction endpoint := t.client.endpoint(t.basePath, topic.GetRestPath(), "permissions") - err := t.client.get(endpoint, &permissions) + err := t.request.get(endpoint, &permissions) return permissions, err } @@ -159,45 +161,45 @@ func (t *topics) GrantPermission(topic TopicName, role string, action []AuthActi for _, v := range action { s = append(s, v.String()) } - return t.client.post(endpoint, s) + return t.request.post(endpoint, s) } func (t *topics) RevokePermission(topic TopicName, role string) error { endpoint := t.client.endpoint(t.basePath, topic.GetRestPath(), "permissions", role) - return t.client.delete(endpoint) + return t.request.delete(endpoint) } func (t *topics) Lookup(topic TopicName) (LookupData, error) { var lookup LookupData endpoint := fmt.Sprintf("%s/%s", t.lookupPath, topic.GetRestPath()) - err := t.client.get(endpoint, &lookup) + err := t.request.get(endpoint, &lookup) return lookup, err } func (t *topics) GetBundleRange(topic TopicName) (string, error) { endpoint := fmt.Sprintf("%s/%s/%s", t.lookupPath, topic.GetRestPath(), "bundle") - data, err := t.client.getWithQueryParams(endpoint, nil, nil, false) + data, err := t.request.getWithQueryParams(endpoint, nil, nil, false) return string(data), err } func (t *topics) GetLastMessageID(topic TopicName) (MessageID, error) { var messageID MessageID endpoint := t.client.endpoint(t.basePath, topic.GetRestPath(), "lastMessageId") - err := t.client.get(endpoint, &messageID) + err := t.request.get(endpoint, &messageID) return messageID, err } func (t *topics) GetStats(topic TopicName) (TopicStats, error) { var stats TopicStats endpoint := t.client.endpoint(t.basePath, topic.GetRestPath(), "stats") - err := t.client.get(endpoint, &stats) + err := t.request.get(endpoint, &stats) return stats, err } func (t *topics) GetInternalStats(topic TopicName) (PersistentTopicInternalStats, error) { var stats PersistentTopicInternalStats endpoint := t.client.endpoint(t.basePath, topic.GetRestPath(), "internalStats") - err := t.client.get(endpoint, &stats) + err := t.request.get(endpoint, &stats) return stats, err } @@ -207,42 +209,42 @@ func (t *topics) GetPartitionedStats(topic TopicName, perPartition bool) (Partit params := map[string]string{ "perPartition": strconv.FormatBool(perPartition), } - _, err := t.client.getWithQueryParams(endpoint, &stats, params, true) + _, err := t.request.getWithQueryParams(endpoint, &stats, params, true) return stats, err } func (t *topics) Terminate(topic TopicName) (MessageID, error) { endpoint := t.client.endpoint(t.basePath, topic.GetRestPath(), "terminate") var messageID MessageID - err := t.client.postWithObj(endpoint, "", &messageID) + err := t.request.postWithObj(endpoint, "", &messageID) return messageID, err } func (t *topics) Offload(topic TopicName, messageID MessageID) error { endpoint := t.client.endpoint(t.basePath, topic.GetRestPath(), "offload") - return t.client.put(endpoint, messageID) + return t.request.put(endpoint, messageID) } func (t *topics) OffloadStatus(topic TopicName) (OffloadProcessStatus, error) { endpoint := t.client.endpoint(t.basePath, topic.GetRestPath(), "offload") var status OffloadProcessStatus - err := t.client.get(endpoint, &status) + err := t.request.get(endpoint, &status) return status, err } func (t *topics) Unload(topic TopicName) error { endpoint := t.client.endpoint(t.basePath, topic.GetRestPath(), "unload") - return t.client.put(endpoint, "") + return t.request.put(endpoint, "") } func (t *topics) Compact(topic TopicName) error { endpoint := t.client.endpoint(t.basePath, topic.GetRestPath(), "compaction") - return t.client.put(endpoint, "") + return t.request.put(endpoint, "") } func (t *topics) CompactStatus(topic TopicName) (LongRunningProcessStatus, error) { endpoint := t.client.endpoint(t.basePath, topic.GetRestPath(), "compaction") var status LongRunningProcessStatus - err := t.client.get(endpoint, &status) + err := t.request.get(endpoint, &status) return status, err } diff --git a/pkg/pulsar/utils.go b/pkg/pulsar/utils.go index 249f18d23..f8207300c 100644 --- a/pkg/pulsar/utils.go +++ b/pkg/pulsar/utils.go @@ -21,6 +21,6 @@ import ( "fmt" ) -func makeHTTPPath(apiVersion string, componentPath string) string { - return fmt.Sprintf("/admin/%s%s", apiVersion, componentPath) +func makeHTTPPath(service, apiVersion string, componentPath string) string { + return fmt.Sprintf("/%s/%s%s", service, apiVersion, componentPath) } diff --git a/pkg/pulsarctl.go b/pkg/pulsarctl.go index 8a1f0d2b2..4224facfd 100644 --- a/pkg/pulsarctl.go +++ b/pkg/pulsarctl.go @@ -19,6 +19,7 @@ package pkg import ( "github.com/streamnative/pulsarctl/pkg/cmdutils" + "github.com/streamnative/pulsarctl/pkg/ctl/bk" "github.com/streamnative/pulsarctl/pkg/ctl/brokers" "github.com/streamnative/pulsarctl/pkg/ctl/brokerstats" "github.com/streamnative/pulsarctl/pkg/ctl/cluster" @@ -89,6 +90,7 @@ func NewPulsarctlCmd() *cobra.Command { rootCmd.AddCommand(topic.Command(flagGrouping)) rootCmd.AddCommand(namespace.Command(flagGrouping)) rootCmd.AddCommand(schema.Command(flagGrouping)) + rootCmd.AddCommand(bk.Command(flagGrouping)) rootCmd.AddCommand(subscription.Command(flagGrouping)) rootCmd.AddCommand(brokers.Command(flagGrouping)) rootCmd.AddCommand(brokerstats.Command(flagGrouping))