From a26b18d6ac91450c14036d30896147fe0fc6e609 Mon Sep 17 00:00:00 2001 From: Yong Zhang Date: Mon, 14 Oct 2019 14:30:23 +0800 Subject: [PATCH 1/8] Add bookie ledger commands --- *Motivation* Introduce a admin client for bookKeeper. Add ledger commands to the admin client *Modifications* - adjust the `admin` to provide the bookKeeper service - Add ledger commands --- pkg/cmdutils/cmdutils.go | 4 + pkg/cmdutils/config.go | 19 +++++ pkg/ctl/bookie/bookie.go | 38 +++++++++ pkg/ctl/bookie/ledger/delete.go | 81 ++++++++++++++++++ pkg/ctl/bookie/ledger/get.go | 19 +++++ pkg/ctl/bookie/ledger/ledger.go | 41 +++++++++ pkg/ctl/bookie/ledger/list.go | 19 +++++ pkg/ctl/bookie/ledger/read.go | 19 +++++ pkg/pulsar/Tenant.go | 16 ++-- pkg/pulsar/admin.go | 64 +++++++++++--- pkg/pulsar/api_version.go | 2 +- pkg/pulsar/cluster.go | 30 +++---- pkg/pulsar/functions.go | 46 ++++++----- pkg/pulsar/ledger.go | 50 +++++++++++ pkg/pulsar/namespace.go | 142 ++++++++++++++++---------------- pkg/pulsar/schema.go | 16 ++-- pkg/pulsar/sinks.go | 40 ++++----- pkg/pulsar/sources.go | 40 ++++----- pkg/pulsar/topic.go | 48 +++++------ pkg/pulsar/utils.go | 4 +- pkg/pulsarctl.go | 2 + 21 files changed, 545 insertions(+), 195 deletions(-) create mode 100644 pkg/ctl/bookie/bookie.go create mode 100644 pkg/ctl/bookie/ledger/delete.go create mode 100644 pkg/ctl/bookie/ledger/get.go create mode 100644 pkg/ctl/bookie/ledger/ledger.go create mode 100644 pkg/ctl/bookie/ledger/list.go create mode 100644 pkg/ctl/bookie/ledger/read.go create mode 100644 pkg/pulsar/ledger.go diff --git a/pkg/cmdutils/cmdutils.go b/pkg/cmdutils/cmdutils.go index a30831612..9879c6718 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 80130aaf9..0225128ca 100644 --- a/pkg/cmdutils/config.go +++ b/pkg/cmdutils/config.go @@ -37,6 +37,9 @@ type ClusterConfig struct { TLSAllowInsecureConnection bool AuthParams string + + // the bookkeeper web service url that pulsarctl connects to. + BookieWebServiceURL string } func (c *ClusterConfig) FlagSet() *pflag.FlagSet { @@ -70,6 +73,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.", + ) + return flags } @@ -100,3 +110,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.WebServiceURL != config.BookieWebServiceURL { + config.BookieWebServiceURL = c.BookieWebServiceURL + } + + return pulsar.NewBookieClient(config) +} diff --git a/pkg/ctl/bookie/bookie.go b/pkg/ctl/bookie/bookie.go new file mode 100644 index 000000000..70bbe2c1e --- /dev/null +++ b/pkg/ctl/bookie/bookie.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 bookie + +import ( + "github.com/spf13/cobra" + "github.com/streamnative/pulsarctl/pkg/cmdutils" + "github.com/streamnative/pulsarctl/pkg/ctl/bookie/ledger" +) + +func Command(flagGrouping *cmdutils.FlagGrouping) *cobra.Command { + resourceCmd := cmdutils.NewResourceCmd( + "bookies", + "Operations about bookies", + "", + "bookie", + ) + + resourceCmd.AddCommand(ledger.Command(flagGrouping)) + + return resourceCmd +} \ No newline at end of file diff --git a/pkg/ctl/bookie/ledger/delete.go b/pkg/ctl/bookie/ledger/delete.go new file mode 100644 index 000000000..7b252bd7a --- /dev/null +++ b/pkg/ctl/bookie/ledger/delete.go @@ -0,0 +1,81 @@ +// 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/pkg/errors" + "github.com/spf13/cobra" + "github.com/spf13/pflag" + "github.com/streamnative/pulsarctl/pkg/cmdutils" + "github.com/streamnative/pulsarctl/pkg/pulsar" +) + +func DeleteLedgerCmd(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) + desc.CommandOutput = out + + vc.SetDescription( + "delete", + "d", + desc.ToString(), + desc.ExampleToString()) + + var id int64 + + vc.SetRunFuncWithNameArg(func() error { + return doDeleteLedgerCmd(vc, id) + }, "the ledger id is not specified or the ledger id is specified more than one") + + vc.FlagSetGroup.InFlagSet("Ledger", func(set *pflag.FlagSet) { + set.Int64Var(&id, "ledger-id", -1, "the delete ledger id") + cobra.MarkFlagRequired(set, "ledger-id") + }) +} + +func doDeleteLedgerCmd(vc *cmdutils.VerbCmd, id int64) error { + + if id <= 0 { + return errors.Errorf("invalid ledger id %d", id) + } + + admin := cmdutils.NewBookieClient() + err := admin.Ledger().DeleteLedger(id) + if err == nil { + vc.Command.Println("Successfully delete the ledger %d", id) + } + + return err +} diff --git a/pkg/ctl/bookie/ledger/get.go b/pkg/ctl/bookie/ledger/get.go new file mode 100644 index 000000000..5e4587d75 --- /dev/null +++ b/pkg/ctl/bookie/ledger/get.go @@ -0,0 +1,19 @@ +// 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 diff --git a/pkg/ctl/bookie/ledger/ledger.go b/pkg/ctl/bookie/ledger/ledger.go new file mode 100644 index 000000000..f1c6f7065 --- /dev/null +++ b/pkg/ctl/bookie/ledger/ledger.go @@ -0,0 +1,41 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. +// + +package 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){ + DeleteLedgerCmd, + } + + cmdutils.AddVerbCmds(flagGrouping, resourceCmd, commands...) + + return resourceCmd +} diff --git a/pkg/ctl/bookie/ledger/list.go b/pkg/ctl/bookie/ledger/list.go new file mode 100644 index 000000000..5e4587d75 --- /dev/null +++ b/pkg/ctl/bookie/ledger/list.go @@ -0,0 +1,19 @@ +// 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 diff --git a/pkg/ctl/bookie/ledger/read.go b/pkg/ctl/bookie/ledger/read.go new file mode 100644 index 000000000..5e4587d75 --- /dev/null +++ b/pkg/ctl/bookie/ledger/read.go @@ -0,0 +1,19 @@ +// 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 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 4fdc22c3d..4e871617b 100644 --- a/pkg/pulsar/admin.go +++ b/pkg/pulsar/admin.go @@ -35,7 +35,8 @@ import ( ) const ( - DefaultWebServiceURL = "http://localhost:8080" + DefaultWebServiceURL = "http://localhost:8080" + DefaultBookieWebServiceURL = "http://localhost:8081" ) // Config is used to configure the admin client @@ -47,6 +48,9 @@ type Config struct { Auth *auth.TLSAuthProvider AuthParams string TLSOptions *TLSOptions + + BookieWebServiceURL string + BookieAPIVersion APIVersion } type TLSOptions struct { @@ -63,6 +67,9 @@ func DefaultConfig() *Config { TLSOptions: &TLSOptions{ AllowInsecureConnection: false, }, + + BookieWebServiceURL: DefaultBookieWebServiceURL, + BookieAPIVersion: V1, } return config } @@ -79,16 +86,30 @@ type Client interface { Schemas() Schema } +// 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 +} + +type pulsarClient struct { + client *client + apiVersion string // TLS config auth *auth.TLSAuthProvider authParams string tlsOptions *TLSOptions - transport *http.Transport +} + +type bookieClient struct { + client *client + apiVersion string } // New returns a new client @@ -97,12 +118,14 @@ func New(config *Config) (Client, error) { config.WebServiceURL = DefaultWebServiceURL } - c := &client{ - apiVersion: config.APIVersion.String(), - webServiceURL: config.WebServiceURL, + c := &pulsarClient{ + apiVersion: config.APIVersion.String(), + client: &client{ + webServiceURL: config.WebServiceURL, + }, } - 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) @@ -118,7 +141,7 @@ func New(config *Config) (Client, error) { return nil, err } - c.transport = &http.Transport{ + c.client.transport = &http.Transport{ MaxIdleConnsPerHost: 10, TLSClientConfig: tlsConf, } @@ -127,7 +150,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.WebServiceURL) == 0 { + config.WebServiceURL = DefaultWebServiceURL + } + + c := &bookieClient{ + apiVersion: config.BookieAPIVersion.String(), + client: &client{ + webServiceURL: config.WebServiceURL, + }, + } + + return c +} + +func (c *pulsarClient) getTLSConfig() (*tls.Config, error) { tlsConfig := &tls.Config{ InsecureSkipVerify: c.tlsOptions.AllowInsecureConnection, } @@ -157,8 +195,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 diff --git a/pkg/pulsar/api_version.go b/pkg/pulsar/api_version.go index 70963d943..32e90025a 100644 --- a/pkg/pulsar/api_version.go +++ b/pkg/pulsar/api_version.go @@ -30,7 +30,7 @@ const DefaultAPIVersion = "v2" func (v APIVersion) String() string { switch v { case V1: - return "" + return "v1" case V2: return "v2" case V3: 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..9d319f0d7 --- /dev/null +++ b/pkg/pulsar/ledger.go @@ -0,0 +1,50 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. +// + +package pulsar + +import ( + "strconv" +) + +type Ledger interface { + // Delete the specified ledger + DeleteLedger(int64) 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: "/delete", + params: make(map[string]string), + } +} + +func (c *ledger) DeleteLedger(ledgerID int64) error { + endpoint := c.client.bookieEndpoint(c.basePath) + c.params["ledger_id"] = strconv.FormatInt(ledgerID, 10) + return c.request.deleteWithQueryParams(endpoint, nil, c.params) +} diff --git a/pkg/pulsar/namespace.go b/pkg/pulsar/namespace.go index a1f96efcf..fc9395347 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, nil, 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/topic.go b/pkg/pulsar/topic.go index ec00a4353..8fa1734ef 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, nil, 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 9341371ec..a61ffa355 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/bookie" "github.com/streamnative/pulsarctl/pkg/ctl/cluster" "github.com/streamnative/pulsarctl/pkg/ctl/completion" "github.com/streamnative/pulsarctl/pkg/ctl/namespace" @@ -86,6 +87,7 @@ func NewPulsarctlCmd() *cobra.Command { rootCmd.AddCommand(topic.Command(flagGrouping)) rootCmd.AddCommand(namespace.Command(flagGrouping)) rootCmd.AddCommand(schema.Command(flagGrouping)) + rootCmd.AddCommand(bookie.Command(flagGrouping)) return rootCmd } From 36dc68c120559660aab15758148805a5e9c79e6e Mon Sep 17 00:00:00 2001 From: Yong Zhang Date: Mon, 14 Oct 2019 22:51:46 +0800 Subject: [PATCH 2/8] Add bookie ledger commands --- *Modifications* - Add list, get, read commands for bookie ledger --- pkg/cmdutils/config.go | 2 +- pkg/ctl/bookie/bookie.go | 6 +-- pkg/ctl/bookie/ledger/delete.go | 35 ++++++------- pkg/ctl/bookie/ledger/errors.go | 25 ++++++++++ pkg/ctl/bookie/ledger/get.go | 88 ++++++++++++++++++++++++++++++++- pkg/ctl/bookie/ledger/ledger.go | 6 ++- pkg/ctl/bookie/ledger/list.go | 74 ++++++++++++++++++++++++++- pkg/ctl/bookie/ledger/read.go | 81 +++++++++++++++++++++++++++++- pkg/pulsar/admin.go | 21 +++----- pkg/pulsar/api_version.go | 5 +- pkg/pulsar/bookie_data.go | 42 ++++++++++++++++ pkg/pulsar/ledger.go | 50 ++++++++++++++++--- pkg/pulsar/namespace.go | 2 +- pkg/pulsar/topic.go | 2 +- 14 files changed, 387 insertions(+), 52 deletions(-) create mode 100644 pkg/ctl/bookie/ledger/errors.go create mode 100644 pkg/pulsar/bookie_data.go diff --git a/pkg/cmdutils/config.go b/pkg/cmdutils/config.go index 0225128ca..11b7b1aed 100644 --- a/pkg/cmdutils/config.go +++ b/pkg/cmdutils/config.go @@ -113,7 +113,7 @@ func (c *ClusterConfig) Client(version pulsar.APIVersion) pulsar.Client { func (c *ClusterConfig) BookieClient() pulsar.BookieClient { config := pulsar.DefaultConfig() - if len(c.BookieWebServiceURL) > 0 && c.WebServiceURL != config.BookieWebServiceURL { + if len(c.BookieWebServiceURL) > 0 && c.BookieWebServiceURL != config.BookieWebServiceURL { config.BookieWebServiceURL = c.BookieWebServiceURL } diff --git a/pkg/ctl/bookie/bookie.go b/pkg/ctl/bookie/bookie.go index 70bbe2c1e..d510ad900 100644 --- a/pkg/ctl/bookie/bookie.go +++ b/pkg/ctl/bookie/bookie.go @@ -14,14 +14,14 @@ // KIND, either express or implied. See the License for the // specific language governing permissions and limitations // under the License. -// package bookie import ( - "github.com/spf13/cobra" "github.com/streamnative/pulsarctl/pkg/cmdutils" "github.com/streamnative/pulsarctl/pkg/ctl/bookie/ledger" + + "github.com/spf13/cobra" ) func Command(flagGrouping *cmdutils.FlagGrouping) *cobra.Command { @@ -35,4 +35,4 @@ func Command(flagGrouping *cmdutils.FlagGrouping) *cobra.Command { resourceCmd.AddCommand(ledger.Command(flagGrouping)) return resourceCmd -} \ No newline at end of file +} diff --git a/pkg/ctl/bookie/ledger/delete.go b/pkg/ctl/bookie/ledger/delete.go index 7b252bd7a..6f3e809d5 100644 --- a/pkg/ctl/bookie/ledger/delete.go +++ b/pkg/ctl/bookie/ledger/delete.go @@ -14,19 +14,19 @@ // KIND, either express or implied. See the License for the // specific language governing permissions and limitations // under the License. -// package ledger import ( - "github.com/pkg/errors" - "github.com/spf13/cobra" - "github.com/spf13/pflag" + "strconv" + "github.com/streamnative/pulsarctl/pkg/cmdutils" "github.com/streamnative/pulsarctl/pkg/pulsar" + + "github.com/pkg/errors" ) -func DeleteLedgerCmd(vc *cmdutils.VerbCmd) { +func DeleteCmd(vc *cmdutils.VerbCmd) { var desc pulsar.LongDescription desc.CommandUsedFor = "This command is used for deleting a ledger." desc.CommandPermission = "none" @@ -44,37 +44,30 @@ func DeleteLedgerCmd(vc *cmdutils.VerbCmd) { Desc: "normal output", Out: "Successfully delete the ledger (ledger-id)", } - out = append(out, successOut) + out = append(out, successOut, argError) desc.CommandOutput = out vc.SetDescription( "delete", - "d", + "Delete a ledger", desc.ToString(), desc.ExampleToString()) - var id int64 - vc.SetRunFuncWithNameArg(func() error { - return doDeleteLedgerCmd(vc, id) + return doDeleteCmd(vc) }, "the ledger id is not specified or the ledger id is specified more than one") - - vc.FlagSetGroup.InFlagSet("Ledger", func(set *pflag.FlagSet) { - set.Int64Var(&id, "ledger-id", -1, "the delete ledger id") - cobra.MarkFlagRequired(set, "ledger-id") - }) } -func doDeleteLedgerCmd(vc *cmdutils.VerbCmd, id int64) error { - - if id <= 0 { - return errors.Errorf("invalid ledger id %d", id) +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().DeleteLedger(id) + err = admin.Ledger().Delete(id) if err == nil { - vc.Command.Println("Successfully delete the ledger %d", id) + vc.Command.Printf("Successfully delete the ledger %d", id) } return err diff --git a/pkg/ctl/bookie/ledger/errors.go b/pkg/ctl/bookie/ledger/errors.go new file mode 100644 index 000000000..6edeb808d --- /dev/null +++ b/pkg/ctl/bookie/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/bookie/ledger/get.go b/pkg/ctl/bookie/ledger/get.go index 5e4587d75..59937451d 100644 --- a/pkg/ctl/bookie/ledger/get.go +++ b/pkg/ctl/bookie/ledger/get.go @@ -14,6 +14,92 @@ // 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 { + 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/bookie/ledger/ledger.go b/pkg/ctl/bookie/ledger/ledger.go index f1c6f7065..36c445fb6 100644 --- a/pkg/ctl/bookie/ledger/ledger.go +++ b/pkg/ctl/bookie/ledger/ledger.go @@ -14,7 +14,6 @@ // KIND, either express or implied. See the License for the // specific language governing permissions and limitations // under the License. -// package ledger @@ -32,7 +31,10 @@ func Command(flagGrouping *cmdutils.FlagGrouping) *cobra.Command { "") commands := []func(*cmdutils.VerbCmd){ - DeleteLedgerCmd, + DeleteCmd, + GetCmd, + ListCmd, + ReadCmd, } cmdutils.AddVerbCmds(flagGrouping, resourceCmd, commands...) diff --git a/pkg/ctl/bookie/ledger/list.go b/pkg/ctl/bookie/ledger/list.go index 5e4587d75..5717c88d6 100644 --- a/pkg/ctl/bookie/ledger/list.go +++ b/pkg/ctl/bookie/ledger/list.go @@ -14,6 +14,78 @@ // 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", + 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/bookie/ledger/read.go b/pkg/ctl/bookie/ledger/read.go index 5e4587d75..525dc4254 100644 --- a/pkg/ctl/bookie/ledger/read.go +++ b/pkg/ctl/bookie/ledger/read.go @@ -14,6 +14,85 @@ // 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", + desc.ToString(), + desc.ExampleToString()) + + var start int64 + var end int64 + + vc.SetRunFuncWithNameArg(func() error { + return doRead(vc, start, end) + }, "the ledger is not specified or the ledger 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 { + return errors.Errorf("invalid ledger id %s", vc.NameArg) + } + + 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/pulsar/admin.go b/pkg/pulsar/admin.go index 4e871617b..e8172fb7a 100644 --- a/pkg/pulsar/admin.go +++ b/pkg/pulsar/admin.go @@ -69,7 +69,7 @@ func DefaultConfig() *Config { }, BookieWebServiceURL: DefaultBookieWebServiceURL, - BookieAPIVersion: V1, + BookieAPIVersion: BV1, } return config } @@ -151,14 +151,14 @@ func New(config *Config) (Client, error) { } func NewBookieClient(config *Config) BookieClient { - if len(config.WebServiceURL) == 0 { - config.WebServiceURL = DefaultWebServiceURL + if len(config.BookieWebServiceURL) == 0 { + config.BookieWebServiceURL = DefaultBookieWebServiceURL } c := &bookieClient{ apiVersion: config.BookieAPIVersion.String(), client: &client{ - webServiceURL: config.WebServiceURL, + webServiceURL: config.BookieWebServiceURL, }, } @@ -200,7 +200,7 @@ func (c *pulsarClient) endpoint(componentPath string, parts ...string) string { } func (c *bookieClient) bookieEndpoint(componentPath string, parts ...string) string { - return path.Join(makeHTTPPath("api", c.apiVersion, componentPath), endpoint(parts...)) + return path.Join(makeHTTPPath("api", c.apiVersion, componentPath+"/"), endpoint(parts...)) } // get is used to do a GET request against an endpoint @@ -283,10 +283,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 @@ -300,18 +300,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 32e90025a..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" @@ -30,11 +31,13 @@ const DefaultAPIVersion = "v2" func (v APIVersion) String() string { switch v { case V1: - return "v1" + return "" case V2: 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/ledger.go b/pkg/pulsar/ledger.go index 9d319f0d7..0ca7ea874 100644 --- a/pkg/pulsar/ledger.go +++ b/pkg/pulsar/ledger.go @@ -14,7 +14,6 @@ // KIND, either express or implied. See the License for the // specific language governing permissions and limitations // under the License. -// package pulsar @@ -24,7 +23,16 @@ import ( type Ledger interface { // Delete the specified ledger - DeleteLedger(int64) error + 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 { @@ -38,13 +46,43 @@ func (c *bookieClient) Ledger() Ledger { return &ledger{ client: c, request: c.client, - basePath: "/delete", + basePath: "/ledger", params: make(map[string]string), } } -func (c *ledger) DeleteLedger(ledgerID int64) error { - endpoint := c.client.bookieEndpoint(c.basePath) +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) - return c.request.deleteWithQueryParams(endpoint, nil, c.params) + 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 fc9395347..eec05b110 100644 --- a/pkg/pulsar/namespace.go +++ b/pkg/pulsar/namespace.go @@ -396,7 +396,7 @@ func (n *namespaces) RemoveBacklogQuota(namespace string) error { params := map[string]string{ "backlogQuotaType": string(DestinationStorage), } - return n.request.deleteWithQueryParams(endpoint, nil, params) + return n.request.deleteWithQueryParams(endpoint, params) } func (n *namespaces) SetSchemaValidationEnforced(namespace NameSpaceName, schemaValidationEnforced bool) error { diff --git a/pkg/pulsar/topic.go b/pkg/pulsar/topic.go index 8fa1734ef..62f883c05 100644 --- a/pkg/pulsar/topic.go +++ b/pkg/pulsar/topic.go @@ -82,7 +82,7 @@ func (t *topics) Delete(topic TopicName, force bool, nonPartitioned bool) error params := map[string]string{ "force": strconv.FormatBool(force), } - return t.request.deleteWithQueryParams(endpoint, nil, params) + return t.request.deleteWithQueryParams(endpoint, params) } func (t *topics) Update(topic TopicName, partitions int) error { From c9ac7f1badf0bde1ec10d869ef948b5dbd618bdb Mon Sep 17 00:00:00 2001 From: Yong Zhang Date: Mon, 14 Oct 2019 22:59:18 +0800 Subject: [PATCH 3/8] Fix errors --- pkg/pulsar/subscription.go | 28 +++++++++++++++------------- 1 file changed, 15 insertions(+), 13 deletions(-) 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 } From 551e069a84ca7fd116445fd07e97e6a231224209 Mon Sep 17 00:00:00 2001 From: Yong Zhang Date: Tue, 15 Oct 2019 11:09:18 +0800 Subject: [PATCH 4/8] Add tests to test arg error --- pkg/ctl/bookie/ledger/delete_test.go | 42 +++++++++++++++++ pkg/ctl/bookie/ledger/get.go | 2 +- pkg/ctl/bookie/ledger/get_test.go | 42 +++++++++++++++++ pkg/ctl/bookie/ledger/list.go | 2 +- pkg/ctl/bookie/ledger/read.go | 14 ++++-- pkg/ctl/bookie/ledger/read_test.go | 52 +++++++++++++++++++++ pkg/ctl/bookie/ledger/test_help.go | 67 ++++++++++++++++++++++++++++ 7 files changed, 216 insertions(+), 5 deletions(-) create mode 100644 pkg/ctl/bookie/ledger/delete_test.go create mode 100644 pkg/ctl/bookie/ledger/get_test.go create mode 100644 pkg/ctl/bookie/ledger/read_test.go create mode 100644 pkg/ctl/bookie/ledger/test_help.go diff --git a/pkg/ctl/bookie/ledger/delete_test.go b/pkg/ctl/bookie/ledger/delete_test.go new file mode 100644 index 000000000..807b67d89 --- /dev/null +++ b/pkg/ctl/bookie/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/bookie/ledger/get.go b/pkg/ctl/bookie/ledger/get.go index 59937451d..0e4324b52 100644 --- a/pkg/ctl/bookie/ledger/get.go +++ b/pkg/ctl/bookie/ledger/get.go @@ -91,7 +91,7 @@ func GetCmd(vc *cmdutils.VerbCmd) { func doGet(vc *cmdutils.VerbCmd) error { id, err := strconv.ParseInt(vc.NameArg, 10, 64) - if err != nil { + if err != nil || id < 0 { return errors.Errorf("invalid ledger id %s", vc.NameArg) } diff --git a/pkg/ctl/bookie/ledger/get_test.go b/pkg/ctl/bookie/ledger/get_test.go new file mode 100644 index 000000000..c381efb07 --- /dev/null +++ b/pkg/ctl/bookie/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/bookie/ledger/list.go b/pkg/ctl/bookie/ledger/list.go index 5717c88d6..a868c54bc 100644 --- a/pkg/ctl/bookie/ledger/list.go +++ b/pkg/ctl/bookie/ledger/list.go @@ -54,7 +54,7 @@ func ListCmd(vc *cmdutils.VerbCmd) { vc.SetDescription( "list", - "list", + "list all the ledgers", desc.ToString(), desc.ExampleToString()) diff --git a/pkg/ctl/bookie/ledger/read.go b/pkg/ctl/bookie/ledger/read.go index 525dc4254..14dc60cdc 100644 --- a/pkg/ctl/bookie/ledger/read.go +++ b/pkg/ctl/bookie/ledger/read.go @@ -63,7 +63,7 @@ func ReadCmd(vc *cmdutils.VerbCmd) { vc.SetDescription( "read", - "read", + "Read a range of entries of a ledger", desc.ToString(), desc.ExampleToString()) @@ -72,7 +72,7 @@ func ReadCmd(vc *cmdutils.VerbCmd) { vc.SetRunFuncWithNameArg(func() error { return doRead(vc, start, end) - }, "the ledger is not specified or the ledger is specified more than one") + }, "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, @@ -84,10 +84,18 @@ func ReadCmd(vc *cmdutils.VerbCmd) { func doRead(vc *cmdutils.VerbCmd, start, end int64) error { id, err := strconv.ParseInt(vc.NameArg, 10, 64) - if err != nil { + 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 { diff --git a/pkg/ctl/bookie/ledger/read_test.go b/pkg/ctl/bookie/ledger/read_test.go new file mode 100644 index 000000000..8e2f472fb --- /dev/null +++ b/pkg/ctl/bookie/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/bookie/ledger/test_help.go b/pkg/ctl/bookie/ledger/test_help.go new file mode 100644 index 000000000..418785041 --- /dev/null +++ b/pkg/ctl/bookie/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 +} From 7cf8c860295feac2e7146206b47b4e55b63993ab Mon Sep 17 00:00:00 2001 From: Yong Zhang Date: Tue, 15 Oct 2019 12:08:40 +0800 Subject: [PATCH 5/8] Add command --- pkg/ctl/topic/topic.go | 1 + 1 file changed, 1 insertion(+) 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...) From 0472bce859eb2c41026e75ffd23521d4e8d16408 Mon Sep 17 00:00:00 2001 From: Yong Zhang Date: Tue, 15 Oct 2019 16:23:57 +0800 Subject: [PATCH 6/8] Rename directory --- pkg/ctl/{bookie/bookie.go => bk/bk.go} | 10 +++++----- pkg/ctl/{bookie => bk}/ledger/delete.go | 0 pkg/ctl/{bookie => bk}/ledger/delete_test.go | 0 pkg/ctl/{bookie => bk}/ledger/errors.go | 0 pkg/ctl/{bookie => bk}/ledger/get.go | 0 pkg/ctl/{bookie => bk}/ledger/get_test.go | 0 pkg/ctl/{bookie => bk}/ledger/ledger.go | 0 pkg/ctl/{bookie => bk}/ledger/list.go | 0 pkg/ctl/{bookie => bk}/ledger/read.go | 0 pkg/ctl/{bookie => bk}/ledger/read_test.go | 0 pkg/ctl/{bookie => bk}/ledger/test_help.go | 0 pkg/pulsarctl.go | 4 ++-- 12 files changed, 7 insertions(+), 7 deletions(-) rename pkg/ctl/{bookie/bookie.go => bk/bk.go} (89%) rename pkg/ctl/{bookie => bk}/ledger/delete.go (100%) rename pkg/ctl/{bookie => bk}/ledger/delete_test.go (100%) rename pkg/ctl/{bookie => bk}/ledger/errors.go (100%) rename pkg/ctl/{bookie => bk}/ledger/get.go (100%) rename pkg/ctl/{bookie => bk}/ledger/get_test.go (100%) rename pkg/ctl/{bookie => bk}/ledger/ledger.go (100%) rename pkg/ctl/{bookie => bk}/ledger/list.go (100%) rename pkg/ctl/{bookie => bk}/ledger/read.go (100%) rename pkg/ctl/{bookie => bk}/ledger/read_test.go (100%) rename pkg/ctl/{bookie => bk}/ledger/test_help.go (100%) diff --git a/pkg/ctl/bookie/bookie.go b/pkg/ctl/bk/bk.go similarity index 89% rename from pkg/ctl/bookie/bookie.go rename to pkg/ctl/bk/bk.go index d510ad900..976551ff0 100644 --- a/pkg/ctl/bookie/bookie.go +++ b/pkg/ctl/bk/bk.go @@ -15,21 +15,21 @@ // specific language governing permissions and limitations // under the License. -package bookie +package bk import ( "github.com/streamnative/pulsarctl/pkg/cmdutils" - "github.com/streamnative/pulsarctl/pkg/ctl/bookie/ledger" + "github.com/streamnative/pulsarctl/pkg/ctl/bk/ledger" "github.com/spf13/cobra" ) func Command(flagGrouping *cmdutils.FlagGrouping) *cobra.Command { resourceCmd := cmdutils.NewResourceCmd( - "bookies", - "Operations about bookies", + "bk", + "Operations about bookKeeper", + "", "", - "bookie", ) resourceCmd.AddCommand(ledger.Command(flagGrouping)) diff --git a/pkg/ctl/bookie/ledger/delete.go b/pkg/ctl/bk/ledger/delete.go similarity index 100% rename from pkg/ctl/bookie/ledger/delete.go rename to pkg/ctl/bk/ledger/delete.go diff --git a/pkg/ctl/bookie/ledger/delete_test.go b/pkg/ctl/bk/ledger/delete_test.go similarity index 100% rename from pkg/ctl/bookie/ledger/delete_test.go rename to pkg/ctl/bk/ledger/delete_test.go diff --git a/pkg/ctl/bookie/ledger/errors.go b/pkg/ctl/bk/ledger/errors.go similarity index 100% rename from pkg/ctl/bookie/ledger/errors.go rename to pkg/ctl/bk/ledger/errors.go diff --git a/pkg/ctl/bookie/ledger/get.go b/pkg/ctl/bk/ledger/get.go similarity index 100% rename from pkg/ctl/bookie/ledger/get.go rename to pkg/ctl/bk/ledger/get.go diff --git a/pkg/ctl/bookie/ledger/get_test.go b/pkg/ctl/bk/ledger/get_test.go similarity index 100% rename from pkg/ctl/bookie/ledger/get_test.go rename to pkg/ctl/bk/ledger/get_test.go diff --git a/pkg/ctl/bookie/ledger/ledger.go b/pkg/ctl/bk/ledger/ledger.go similarity index 100% rename from pkg/ctl/bookie/ledger/ledger.go rename to pkg/ctl/bk/ledger/ledger.go diff --git a/pkg/ctl/bookie/ledger/list.go b/pkg/ctl/bk/ledger/list.go similarity index 100% rename from pkg/ctl/bookie/ledger/list.go rename to pkg/ctl/bk/ledger/list.go diff --git a/pkg/ctl/bookie/ledger/read.go b/pkg/ctl/bk/ledger/read.go similarity index 100% rename from pkg/ctl/bookie/ledger/read.go rename to pkg/ctl/bk/ledger/read.go diff --git a/pkg/ctl/bookie/ledger/read_test.go b/pkg/ctl/bk/ledger/read_test.go similarity index 100% rename from pkg/ctl/bookie/ledger/read_test.go rename to pkg/ctl/bk/ledger/read_test.go diff --git a/pkg/ctl/bookie/ledger/test_help.go b/pkg/ctl/bk/ledger/test_help.go similarity index 100% rename from pkg/ctl/bookie/ledger/test_help.go rename to pkg/ctl/bk/ledger/test_help.go diff --git a/pkg/pulsarctl.go b/pkg/pulsarctl.go index 0dc8ae009..15a0eadb7 100644 --- a/pkg/pulsarctl.go +++ b/pkg/pulsarctl.go @@ -19,7 +19,7 @@ package pkg import ( "github.com/streamnative/pulsarctl/pkg/cmdutils" - "github.com/streamnative/pulsarctl/pkg/ctl/bookie" + "github.com/streamnative/pulsarctl/pkg/ctl/bk" "github.com/streamnative/pulsarctl/pkg/ctl/cluster" "github.com/streamnative/pulsarctl/pkg/ctl/completion" "github.com/streamnative/pulsarctl/pkg/ctl/namespace" @@ -88,7 +88,7 @@ func NewPulsarctlCmd() *cobra.Command { rootCmd.AddCommand(topic.Command(flagGrouping)) rootCmd.AddCommand(namespace.Command(flagGrouping)) rootCmd.AddCommand(schema.Command(flagGrouping)) - rootCmd.AddCommand(bookie.Command(flagGrouping)) + rootCmd.AddCommand(bk.Command(flagGrouping)) rootCmd.AddCommand(subscription.Command(flagGrouping)) return rootCmd From 42b12216913e3cef08249e3a5cce02d1c369050c Mon Sep 17 00:00:00 2001 From: Yong Zhang Date: Thu, 17 Oct 2019 14:29:08 +0800 Subject: [PATCH 7/8] Merge code --- pkg/pulsar/broker_stats.go | 16 +++++++++------- pkg/pulsar/brokers.go | 24 +++++++++++++----------- 2 files changed, 22 insertions(+), 18 deletions(-) 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 } From 82986b42d032d26c1336e9cf56d0140f1721d77d Mon Sep 17 00:00:00 2001 From: Yong Zhang Date: Fri, 18 Oct 2019 20:46:47 +0800 Subject: [PATCH 8/8] Fix bulid error --- pkg/cmdutils/config.go | 4 ++-- pkg/pulsar/admin.go | 15 ++++++++------- 2 files changed, 10 insertions(+), 9 deletions(-) diff --git a/pkg/cmdutils/config.go b/pkg/cmdutils/config.go index 4affc22b3..ab9b0d4ba 100644 --- a/pkg/cmdutils/config.go +++ b/pkg/cmdutils/config.go @@ -44,7 +44,7 @@ type ClusterConfig struct { // 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 and TokenFile is used to config the pulsarctl using token to authentication Token string TokenFile string } @@ -87,7 +87,7 @@ func (c *ClusterConfig) FlagSet() *pflag.FlagSet { "The bookie web service url that pulsarctl connects to.", ) - flags.StringVar( + flags.StringVar( &c.Token, "token", "", diff --git a/pkg/pulsar/admin.go b/pkg/pulsar/admin.go index 19cb26ec9..02630d7a6 100644 --- a/pkg/pulsar/admin.go +++ b/pkg/pulsar/admin.go @@ -53,7 +53,7 @@ type Config struct { BookieWebServiceURL string BookieAPIVersion APIVersion - TokenAuth *auth.TokenAuthProvider + TokenAuth *auth.TokenAuthProvider } type TLSOptions struct { @@ -101,12 +101,13 @@ type client struct { webServiceURL string httpClient *http.Client transport *http.Transport + versionInfo string + tokenAuth *auth.TokenAuthProvider } type pulsarClient struct { client *client apiVersion string - versionInfo string // TLS config auth *auth.TLSAuthProvider @@ -117,9 +118,6 @@ type pulsarClient struct { type bookieClient struct { client *client apiVersion string - transport *http.Transport - - tokenAuth *auth.TokenAuthProvider } // New returns a new client @@ -132,9 +130,12 @@ func New(config *Config) (Client, error) { apiVersion: config.APIVersion.String(), client: &client{ webServiceURL: config.WebServiceURL, + versionInfo: ReleaseVersion, }, - versionInfo: ReleaseVersion, - tokenAuth: config.TokenAuth, + } + + if config.TokenAuth != nil { + c.client.tokenAuth = config.TokenAuth } if strings.HasPrefix(c.client.webServiceURL, "https://") {