diff --git a/pkg/test/base_container.go b/pkg/test/base_container.go index 4476a0173..cafd38a57 100644 --- a/pkg/test/base_container.go +++ b/pkg/test/base_container.go @@ -61,6 +61,11 @@ func (bc *BaseContainer) WithNetworkAliases(aliases map[string][]string) *BaseCo return bc } +// GetANetworkAlias returns a network alias of the container. +func (bc *BaseContainer) GetANetworkAlias(network string) string { + return bc.containerRequest.NetworkAliases[network][0] +} + // WithCmd sets the containers start up commands. func (bc *BaseContainer) WithCmd(cmd []string) *BaseContainer { bc.containerRequest.Cmd = append(bc.containerRequest.Cmd, cmd...) diff --git a/pkg/test/bookkeeper/cluster.go b/pkg/test/bookkeeper/cluster.go new file mode 100644 index 000000000..493eae9cb --- /dev/null +++ b/pkg/test/bookkeeper/cluster.go @@ -0,0 +1,142 @@ +// 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 bookkeeper + +import ( + "context" + "fmt" + "strconv" + + "github.com/streamnative/pulsarctl/pkg/test" + "github.com/streamnative/pulsarctl/pkg/test/bookkeeper/containers" + + "github.com/pkg/errors" + "github.com/testcontainers/testcontainers-go" +) + +var ( + LatestImage = "apache/bookkeeper:latest" +) + +type ClusterDef struct { + clusterSpec *ClusterSpec + networkName string + network testcontainers.Network + zkContainer *test.BaseContainer + bookieContainers map[string]*test.BaseContainer +} + +func DefaultCluster() (test.Cluster, error) { + return NewBookieCluster(DefaultClusterSpec()) +} + +func NewBookieCluster(spec *ClusterSpec) (test.Cluster, error) { + c := &ClusterDef{clusterSpec: spec} + c.networkName = spec.ClusterName + test.RandomSuffix() + network, err := test.NewNetwork(c.networkName) + if err != nil { + return c, err + } + c.network = network + + c.zkContainer = containers.NewZookeeperContainer(spec.Image, c.networkName) + c.bookieContainers = getBookieContainers(spec, c.networkName, containers.DefaultZookeeperServiceString()) + + return c, nil +} + +func getBookieContainers(c *ClusterSpec, networkName, zkServers string) map[string]*test.BaseContainer { + bookies := make(map[string]*test.BaseContainer) + for i := 0; i < c.NumBookies; i++ { + name := fmt.Sprintf("%s-%d", containers.BookieName, i) + bookie := containers.NewBookieContainer(c.Image, networkName) + bookie.WithEnv(map[string]string{ + "BK_zkServers": zkServers, + "BK_httpServerEnabled": "true", + "BK_httpServerPort": strconv.Itoa(c.BookieHTTPServicePort), + "BK_ledgerDirectories": "bk/ledgers", + "BK_indexDirectories": "bk/ledgers", + "BK_journalDirectory": "bk/journal", + }) + bookies[name] = bookie + } + return bookies +} + +func (c *ClusterDef) Start(ctx context.Context) error { + err := c.zkContainer.Start(ctx) + if err != nil { + return errors.WithMessage(err, "encountering errors when starting the zookeeper") + } + fmt.Printf("Zookeeper %s:%s started.\n", + c.zkContainer.GetANetworkAlias(c.networkName), c.zkContainer.GetContainerID()) + zkName := c.zkContainer.GetANetworkAlias(c.networkName) + init := InitBookieCluster(c.clusterSpec.Image, c.networkName, fmt.Sprintf("%s:2181", zkName)) + err = init.Start(ctx) + if err != nil { + return errors.WithMessage(err, "encountering errors when formatting metadata for bookkeeper") + } + fmt.Println("BookKeeper metadata initialized successfully.") + + for k, v := range c.bookieContainers { + err = v.Start(ctx) + if err != nil { + return errors.WithMessagef(err, "encountering errors when starting the bookie %s\n", k) + } + fmt.Printf("Bookie %s:%s started.\n", k, v.GetContainerID()) + } + + return nil +} + +func (c *ClusterDef) Stop(ctx context.Context) error { + if c.bookieContainers != nil { + for _, v := range c.bookieContainers { + err := v.Stop(ctx) + if err != nil { + return err + } + } + } + return nil +} + +func (c *ClusterDef) GetPlainTextServiceURL(ctx context.Context) (string, error) { + return "", errors.New("unsupported operation") +} + +func (c *ClusterDef) GetHTTPServiceURL(ctx context.Context) (string, error) { + port, err := c.getABookie().MappedPort(ctx, strconv.Itoa(c.clusterSpec.BookieHTTPServicePort)) + if err != nil { + return "", err + } + return "http://localhost:" + port.Port(), nil +} + +func (c *ClusterDef) Close(ctx context.Context) { + if c.network != nil { + c.network.Remove(ctx) + } +} + +func (c *ClusterDef) getABookie() *test.BaseContainer { + for _, v := range c.bookieContainers { + return v + } + return nil +} diff --git a/pkg/test/bookkeeper/cluster_script.go b/pkg/test/bookkeeper/cluster_script.go new file mode 100644 index 000000000..deb92dd87 --- /dev/null +++ b/pkg/test/bookkeeper/cluster_script.go @@ -0,0 +1,29 @@ +// 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 bookkeeper + +import "github.com/streamnative/pulsarctl/pkg/test" + +func InitBookieCluster(image, network, zookeeper string) *test.BaseContainer { + bookieInit := test.NewContainer(image) + bookieInit.WithNetwork([]string{network}) + bookieInit.WithEnv(map[string]string{"BK_zkServers": zookeeper}) + bookieInit.WithCmd([]string{"/opt/bookkeeper/bin/bookkeeper", "shell", "metaformat"}) + bookieInit.WaitForLog("/opt/bookkeeper/bin/bookkeeper shell metaformat") + return bookieInit +} diff --git a/pkg/test/bookkeeper/cluster_spec.go b/pkg/test/bookkeeper/cluster_spec.go new file mode 100644 index 000000000..284a17621 --- /dev/null +++ b/pkg/test/bookkeeper/cluster_spec.go @@ -0,0 +1,40 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package bookkeeper + +import "github.com/streamnative/pulsarctl/pkg/test/bookkeeper/containers" + +type ClusterSpec struct { + Image string + ClusterName string + NumBookies int + BookieServicePort int + BookieHTTPServicePort int + ZookeeperServicePort int +} + +func DefaultClusterSpec() *ClusterSpec { + return &ClusterSpec{ + Image: LatestImage, + ClusterName: "default-bookie", + NumBookies: 1, + BookieServicePort: containers.DefaultBookieServicePort, + BookieHTTPServicePort: containers.DefaultBookieHTTPServicePort, + ZookeeperServicePort: containers.DefaultZookeeperServicePort, + } +} diff --git a/pkg/test/bookkeeper/cluster_test.go b/pkg/test/bookkeeper/cluster_test.go new file mode 100644 index 000000000..ab42e8358 --- /dev/null +++ b/pkg/test/bookkeeper/cluster_test.go @@ -0,0 +1,55 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package bookkeeper + +import ( + "context" + "net/http" + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestDefaultCluster(t *testing.T) { + ctx := context.Background() + bkCluster, err := DefaultCluster() + // nolint + defer bkCluster.Close(ctx) + if err != nil { + t.Fatal(err) + } + err = bkCluster.Start(ctx) + if err != nil { + t.Fatal(err) + } + defer bkCluster.Stop(ctx) + + path, err := bkCluster.GetHTTPServiceURL(ctx) + if err != nil { + t.Fatal(err) + } + + resp, err := http.Get(path + "/api/v1/bookie/list_bookie_info") + // nolint + defer resp.Body.Close() + if err != nil { + t.Fatal(err) + } + + assert.Equal(t, 200, resp.StatusCode) +} diff --git a/pkg/test/bookkeeper/containers/bookie.go b/pkg/test/bookkeeper/containers/bookie.go new file mode 100644 index 000000000..6d6a8af1e --- /dev/null +++ b/pkg/test/bookkeeper/containers/bookie.go @@ -0,0 +1,36 @@ +// 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 containers + +import "github.com/streamnative/pulsarctl/pkg/test" + +const ( + BookieName = "bookie" + DefaultBookieServicePort = 3181 + DefaultBookieHTTPServicePort = 8080 +) + +func NewBookieContainer(image, network string) *test.BaseContainer { + bk := test.NewContainer(image) + bk.WithNetwork([]string{network}) + bk.WithNetworkAliases(map[string][]string{network: {BookieName}}) + bk.ExposedPorts([]string{"8080"}) + bk.WithCmd([]string{"bookie"}) + bk.WaitForPort("8080") + return bk +} diff --git a/pkg/test/bookkeeper/containers/zookeeper.go b/pkg/test/bookkeeper/containers/zookeeper.go new file mode 100644 index 000000000..35732e0f3 --- /dev/null +++ b/pkg/test/bookkeeper/containers/zookeeper.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 containers + +import ( + "fmt" + + "github.com/streamnative/pulsarctl/pkg/test" +) + +const ( + ZookeeperName = "zookeeper" + DefaultZookeeperServicePort = 2181 +) + +func NewZookeeperContainer(image, network string) *test.BaseContainer { + zookeeper := test.NewContainer(image) + zookeeper.WithNetwork([]string{network}) + zookeeper.WithNetworkAliases(map[string][]string{network: {ZookeeperName}}) + zookeeper.WithCmd([]string{"zookeeper"}) + zookeeper.WaitForLog("binding to port 0.0.0.0/0.0.0.0:2181") + return zookeeper +} + +func DefaultZookeeperServiceString() string { + return fmt.Sprintf("%s:%d", ZookeeperName, DefaultZookeeperServicePort) +} diff --git a/pkg/test/cluster.go b/pkg/test/cluster.go new file mode 100644 index 000000000..33b1eb56f --- /dev/null +++ b/pkg/test/cluster.go @@ -0,0 +1,37 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package test + +import "context" + +type Cluster interface { + // Start a cluster. + Start(ctx context.Context) error + + // Stop a cluster. + Stop(ctx context.Context) error + + // GetPlainTextServiceURL gets the service connect string. + GetPlainTextServiceURL(ctx context.Context) (string, error) + + // GetHTTPServiceURL gets the HTTP service connect string. + GetHTTPServiceURL(ctx context.Context) (string, error) + + // Close closes the resources used for starting cluster. + Close(ctx context.Context) +} diff --git a/pkg/test/pulsar/cluster.go b/pkg/test/pulsar/cluster.go index 92e25c11d..f4f50768f 100644 --- a/pkg/test/pulsar/cluster.go +++ b/pkg/test/pulsar/cluster.go @@ -41,6 +41,7 @@ var ( type ClusterDef struct { clusterSpec *ClusterSpec + networkName string network testcontainers.Network zkContainer *test.BaseContainer proxyContainer *test.BaseContainer @@ -66,37 +67,33 @@ type Cluster interface { } // DefaultPulsarCluster creates a pulsar cluster using the default cluster spec. -func DefaultPulsarCluster() (Cluster, error) { +func DefaultPulsarCluster() (test.Cluster, error) { return NewPulsarCluster(DefaultClusterSpec()) } // NewPulsarCluster creates a pulsar cluster using the spec. -func NewPulsarCluster(spec *ClusterSpec) (Cluster, error) { - networkName := spec.ClusterName - network, err := test.NewNetwork(networkName) +func NewPulsarCluster(spec *ClusterSpec) (test.Cluster, error) { + c := &ClusterDef{clusterSpec: spec} + c.networkName = spec.ClusterName + test.RandomSuffix() + network, err := test.NewNetwork(c.networkName) if err != nil { - return nil, err + return c, err } + c.network = network - zookeeper := containers.NewZookeeperContainer(LatestImage, networkName) - bookies := getBookieContainers(networkName, spec.NumBookies) - brokers := getBrokerContainers(spec.ClusterName, networkName, spec.NumBrokers) + c.zkContainer = containers.NewZookeeperContainer(LatestImage, c.networkName) + c.bookieContainers = getBookieContainers(c.networkName, spec.NumBookies) + brokers := getBrokerContainers(spec.ClusterName, c.networkName, spec.NumBrokers) broker := getABrokerNetAlias(brokers) - proxy := containers.NewProxyContainer(LatestImage, networkName).WithEnv(map[string]string{ + c.brokerContainers = brokers + c.proxyContainer = containers.NewProxyContainer(LatestImage, c.networkName).WithEnv(map[string]string{ "webServicePort": strconv.Itoa(spec.ProxyHTTPServicePort), "servicePort": strconv.Itoa(spec.ProxyServicePort), "brokerServiceURL": fmt.Sprintf("pulsar://%s:%d", broker, spec.BrokerServicePort), "brokerWebServiceURL": fmt.Sprintf("http://%s:%d", broker, spec.BrokerHTTPServicePort), }) - return &ClusterDef{ - network: network, - clusterSpec: spec, - zkContainer: zookeeper, - proxyContainer: proxy, - bookieContainers: bookies, - brokerContainers: brokers, - }, nil + return c, nil } func getBookieContainers(network string, num int) map[string]*test.BaseContainer { @@ -146,7 +143,7 @@ func (c *ClusterDef) Start(ctx context.Context) error { Zookeeper: fmt.Sprintf("%s:%d", containers.ZookeeperName, DefaultZKPort), Broker: fmt.Sprintf("%s:%d", getABrokerNetAlias(c.brokerContainers), c.clusterSpec.BrokerHTTPServicePort), - }, LatestImage, c.clusterSpec.ClusterName) + }, LatestImage, c.networkName) err = init.Start(ctx) if err != nil { return errors.WithMessage(err, "encountered errors when initializing the pulsar cluster") @@ -231,5 +228,7 @@ func (c *ClusterDef) GetHTTPServiceURL(ctx context.Context) (string, error) { } func (c *ClusterDef) Close(ctx context.Context) { - c.network.Remove(ctx) + if c.network != nil { + c.network.Remove(ctx) + } } diff --git a/pkg/test/utils.go b/pkg/test/utils.go index 0d542555d..af2d1864b 100644 --- a/pkg/test/utils.go +++ b/pkg/test/utils.go @@ -19,6 +19,8 @@ package test import ( "context" + "strconv" + "time" "github.com/testcontainers/testcontainers-go" ) @@ -37,3 +39,7 @@ func NewNetwork(name string) (testcontainers.Network, error) { }) return net, err } + +func RandomSuffix() string { + return "-" + strconv.FormatInt(time.Now().Unix(), 10) +}