Add bookkeeper cluster for testing bkctl

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...)

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
+}

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
+}

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,
+	}
+}

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)
+}

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
+}

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)
+}

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)
+}