You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by lu...@apache.org on 2022/06/02 06:13:41 UTC
[skywalking-banyandb] branch optimize-etcd-tests created (now 905c69c)
This is an automated email from the ASF dual-hosted git repository.
lujiajing pushed a change to branch optimize-etcd-tests
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
at 905c69c use random port instead of unix sockets
This branch includes the following new commits:
new 905c69c use random port instead of unix sockets
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[skywalking-banyandb] 01/01: use random port instead of unix sockets
Posted by lu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lujiajing pushed a commit to branch optimize-etcd-tests
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit 905c69c8c43c4a8c44f61374347aae00007bf20b
Author: Megrez Lu <lu...@gmail.com>
AuthorDate: Thu Jun 2 14:13:30 2022 +0800
use random port instead of unix sockets
---
banyand/liaison/grpc/measure_test.go | 11 ++++--
banyand/liaison/grpc/registry_test.go | 5 ++-
banyand/liaison/grpc/stream_test.go | 10 ++++--
banyand/measure/measure_suite_test.go | 4 ++-
banyand/metadata/metadata.go | 11 ++++--
banyand/metadata/schema/etcd.go | 11 +-----
banyand/metadata/schema/etcd_test.go | 17 +++++----
banyand/query/processor_test.go | 3 ++
banyand/stream/stream_suite_test.go | 3 ++
pkg/query/logical/common_test.go | 12 +++++--
pkg/query/logical/measure_analyzer_test.go | 9 ++++-
pkg/query/logical/stream_analyzer_test.go | 9 ++++-
pkg/test/network.go | 55 ++++++++++++++++++++++++++++++
13 files changed, 130 insertions(+), 30 deletions(-)
diff --git a/banyand/liaison/grpc/measure_test.go b/banyand/liaison/grpc/measure_test.go
index 94270cf..4c1c3c2 100644
--- a/banyand/liaison/grpc/measure_test.go
+++ b/banyand/liaison/grpc/measure_test.go
@@ -42,14 +42,18 @@ var _ = Describe("Measure", func() {
var path string
var gracefulStop, deferFunc func()
var conn *grpclib.ClientConn
+ var listenClientURL, listenPeerURL string
BeforeEach(func() {
var err error
path, deferFunc, err = test.NewSpace()
Expect(err).NotTo(HaveOccurred())
+ listenClientURL, listenPeerURL, err = test.NewEtcdListenUrls()
+ Expect(err).NotTo(HaveOccurred())
})
It("is a plain server", func() {
By("Verifying an empty server")
- flags := []string{"--stream-root-path=" + path, "--measure-root-path=" + path, "--metadata-root-path=" + path}
+ flags := []string{"--stream-root-path=" + path, "--measure-root-path=" + path, "--metadata-root-path=" + path,
+ "--etcd-listen-client-url=" + listenClientURL, "--etcd-listen-peer-url=" + listenPeerURL}
gracefulStop = setup(true, flags)
credentials := insecure.NewCredentials()
var err error
@@ -75,7 +79,10 @@ var _ = Describe("Measure", func() {
}, defaultEventallyTimeout).Should(Equal(1))
})
It("is a TLS server", func() {
- flags := []string{"--tls=true", "--stream-root-path=" + path, "--measure-root-path=" + path, "--metadata-root-path=" + path}
+ flags := []string{"--tls=true", "--stream-root-path=" + path, "--measure-root-path=" + path,
+ "--metadata-root-path=" + path, "--stream-root-path=" + path, "--measure-root-path=" + path,
+ "--metadata-root-path=" + path, "--etcd-listen-client-url=" + listenClientURL,
+ "--etcd-listen-peer-url=" + listenPeerURL}
_, currentFile, _, _ := runtime.Caller(0)
basePath := filepath.Dir(currentFile)
certFile := filepath.Join(basePath, "testdata/server_cert.pem")
diff --git a/banyand/liaison/grpc/registry_test.go b/banyand/liaison/grpc/registry_test.go
index 1626d7b..97552c1 100644
--- a/banyand/liaison/grpc/registry_test.go
+++ b/banyand/liaison/grpc/registry_test.go
@@ -160,7 +160,10 @@ func setupForRegistry() func() {
var flags []string
metaPath, metaDeferFunc, err := test.NewSpace()
Expect(err).NotTo(HaveOccurred())
- flags = append(flags, "--metadata-root-path="+metaPath)
+ listenClientURL, listenPeerURL, err := test.NewEtcdListenUrls()
+ Expect(err).NotTo(HaveOccurred())
+ flags = append(flags, "--metadata-root-path="+metaPath, "--etcd-listen-client-url="+listenClientURL,
+ "--etcd-listen-peer-url="+listenPeerURL)
deferFunc := test.SetUpModules(
flags,
repo,
diff --git a/banyand/liaison/grpc/stream_test.go b/banyand/liaison/grpc/stream_test.go
index 88c60c6..e620f54 100644
--- a/banyand/liaison/grpc/stream_test.go
+++ b/banyand/liaison/grpc/stream_test.go
@@ -49,14 +49,18 @@ var _ = Describe("Stream", func() {
var path string
var gracefulStop, deferFunc func()
var conn *grpclib.ClientConn
+ var listenClientURL, listenPeerURL string
BeforeEach(func() {
var err error
path, deferFunc, err = test.NewSpace()
Expect(err).NotTo(HaveOccurred())
+ listenClientURL, listenPeerURL, err = test.NewEtcdListenUrls()
+ Expect(err).NotTo(HaveOccurred())
})
It("is a plain server", func() {
By("Verifying an empty server")
- flags := []string{"--stream-root-path=" + path, "--measure-root-path=" + path, "--metadata-root-path=" + path}
+ flags := []string{"--stream-root-path=" + path, "--measure-root-path=" + path, "--metadata-root-path=" + path,
+ "--etcd-listen-client-url=" + listenClientURL, "--etcd-listen-peer-url=" + listenPeerURL}
gracefulStop = setup(true, flags)
var err error
conn, err = grpclib.Dial("localhost:17912", grpclib.WithTransportCredentials(insecure.NewCredentials()))
@@ -81,7 +85,9 @@ var _ = Describe("Stream", func() {
}, defaultEventallyTimeout).Should(Equal(1))
})
It("is a TLS server", func() {
- flags := []string{"--tls=true", "--stream-root-path=" + path, "--measure-root-path=" + path, "--metadata-root-path=" + path}
+ flags := []string{"--tls=true", "--stream-root-path=" + path, "--measure-root-path=" + path,
+ "--metadata-root-path=" + path, "--etcd-listen-client-url=" + listenClientURL,
+ "--etcd-listen-peer-url=" + listenPeerURL}
_, currentFile, _, _ := runtime.Caller(0)
basePath := filepath.Dir(currentFile)
certFile := filepath.Join(basePath, "testdata/server_cert.pem")
diff --git a/banyand/measure/measure_suite_test.go b/banyand/measure/measure_suite_test.go
index d24d12c..3f295d8 100644
--- a/banyand/measure/measure_suite_test.go
+++ b/banyand/measure/measure_suite_test.go
@@ -86,7 +86,6 @@ func setUp() (*services, func()) {
// Init Metadata Service
metadataService, err := metadata.NewService(context.TODO())
gomega.Expect(err).NotTo(gomega.HaveOccurred())
- gomega.Expect(err).NotTo(gomega.HaveOccurred())
// Init Measure Service
measureService, err := measure.NewService(context.TODO(), metadataService, repo, pipeline)
@@ -99,6 +98,9 @@ func setUp() (*services, func()) {
rootPath, deferFunc, err := test.NewSpace()
gomega.Expect(err).NotTo(gomega.HaveOccurred())
flags = append(flags, "--measure-root-path="+rootPath)
+ listenClientURL, listenPeerURL, err := test.NewEtcdListenUrls()
+ gomega.Expect(err).NotTo(gomega.HaveOccurred())
+ flags = append(flags, "--etcd-listen-client-url="+listenClientURL, "--etcd-listen-peer-url="+listenPeerURL)
moduleDeferFunc := test.SetUpModules(
flags,
repo,
diff --git a/banyand/metadata/metadata.go b/banyand/metadata/metadata.go
index a8524c7..8ecdb31 100644
--- a/banyand/metadata/metadata.go
+++ b/banyand/metadata/metadata.go
@@ -59,13 +59,17 @@ type Service interface {
}
type service struct {
- schemaRegistry schema.Registry
- rootDir string
+ schemaRegistry schema.Registry
+ rootDir string
+ listenClientURL string
+ listenPeerURL string
}
func (s *service) FlagSet() *run.FlagSet {
fs := run.NewFlagSet("metadata")
fs.StringVarP(&s.rootDir, "metadata-root-path", "", "/tmp", "the root path of metadata")
+ fs.StringVarP(&s.listenClientURL, "etcd-listen-client-url", "", "http://localhost:2379", "A URL to listen on for client traffic")
+ fs.StringVarP(&s.listenPeerURL, "etcd-listen-peer-url", "", "http://localhost:2380", "A URL to listen on for peer traffic")
return fs
}
@@ -78,7 +82,8 @@ func (s *service) Validate() error {
func (s *service) PreRun() error {
var err error
- s.schemaRegistry, err = schema.NewEtcdSchemaRegistry(schema.UseRandomListener(),
+ s.schemaRegistry, err = schema.NewEtcdSchemaRegistry(
+ schema.ConfigureListener(s.listenClientURL, s.listenPeerURL),
schema.RootDir(s.rootDir), schema.LoggerLevel(logger.GetLogger().GetLevel().String()))
if err != nil {
return err
diff --git a/banyand/metadata/schema/etcd.go b/banyand/metadata/schema/etcd.go
index 98a9466..2d57755 100644
--- a/banyand/metadata/schema/etcd.go
+++ b/banyand/metadata/schema/etcd.go
@@ -19,8 +19,6 @@ package schema
import (
"context"
- "fmt"
- "math/rand"
"net/url"
"os"
"path/filepath"
@@ -68,15 +66,8 @@ func LoggerLevel(loggerLevel string) RegistryOption {
}
}
-func randomUnixDomainListener() (string, string) {
- i := rand.Uint64()
- return fmt.Sprintf("%s://localhost:%d%06d", unixDomainSockScheme, os.Getpid(), i),
- fmt.Sprintf("%s://localhost:%d%06d", unixDomainSockScheme, os.Getpid(), i+1)
-}
-
-func UseRandomListener() RegistryOption {
+func ConfigureListener(lc, lp string) RegistryOption {
return func(config *etcdSchemaRegistryConfig) {
- lc, lp := randomUnixDomainListener()
config.listenerClientURL = lc
config.listenerPeerURL = lp
}
diff --git a/banyand/metadata/schema/etcd_test.go b/banyand/metadata/schema/etcd_test.go
index 2bdd32d..f919ffc 100644
--- a/banyand/metadata/schema/etcd_test.go
+++ b/banyand/metadata/schema/etcd_test.go
@@ -22,6 +22,7 @@ import (
"embed"
"errors"
"fmt"
+ "github.com/apache/skywalking-banyandb/pkg/test"
"os"
"path"
"testing"
@@ -123,15 +124,19 @@ func useRandomTempDir() RegistryOption {
}
}
-func useUnixDomain() RegistryOption {
+func useRandomPort() RegistryOption {
return func(config *etcdSchemaRegistryConfig) {
- config.listenerClientURL, config.listenerPeerURL = randomUnixDomainListener()
+ ports, err := test.AllocateFreePorts(2)
+ if err != nil {
+ panic("fail to find free ports")
+ }
+ config.listenerClientURL, config.listenerPeerURL = fmt.Sprintf("http://127.0.0.1:%d", ports[0]), fmt.Sprintf("http://127.0.0.1:%d", ports[1])
}
}
func Test_Etcd_Entity_Get(t *testing.T) {
tester := assert.New(t)
- registry, err := NewEtcdSchemaRegistry(useUnixDomain(), useRandomTempDir(), LoggerLevel("warn"))
+ registry, err := NewEtcdSchemaRegistry(useRandomPort(), useRandomTempDir(), LoggerLevel("warn"))
tester.NoError(err)
tester.NotNil(registry)
defer registry.Close()
@@ -223,7 +228,7 @@ func Test_Etcd_Entity_Get(t *testing.T) {
func Test_Etcd_Entity_List(t *testing.T) {
tester := assert.New(t)
- registry, err := NewEtcdSchemaRegistry(useUnixDomain(), useRandomTempDir(), LoggerLevel("warn"))
+ registry, err := NewEtcdSchemaRegistry(useRandomPort(), useRandomTempDir(), LoggerLevel("warn"))
tester.NoError(err)
tester.NotNil(registry)
defer registry.Close()
@@ -305,7 +310,7 @@ func Test_Etcd_Entity_List(t *testing.T) {
func Test_Etcd_Delete(t *testing.T) {
tester := assert.New(t)
- registry, err := NewEtcdSchemaRegistry(useUnixDomain(), useRandomTempDir(), LoggerLevel("warn"))
+ registry, err := NewEtcdSchemaRegistry(useRandomPort(), useRandomTempDir(), LoggerLevel("warn"))
tester.NoError(err)
tester.NotNil(registry)
defer registry.Close()
@@ -374,7 +379,7 @@ func Test_Etcd_Delete(t *testing.T) {
func Test_Notify(t *testing.T) {
req := require.New(t)
- registry, err := NewEtcdSchemaRegistry(useUnixDomain(), useRandomTempDir(), LoggerLevel("warn"))
+ registry, err := NewEtcdSchemaRegistry(useRandomPort(), useRandomTempDir(), LoggerLevel("warn"))
req.NoError(err)
req.NotNil(registry)
defer registry.Close()
diff --git a/banyand/query/processor_test.go b/banyand/query/processor_test.go
index fba13c4..b72908b 100644
--- a/banyand/query/processor_test.go
+++ b/banyand/query/processor_test.go
@@ -116,6 +116,9 @@ func setUpServices() (*services, func()) {
Expect(err).NotTo(HaveOccurred())
flags = append(flags, "--measure-root-path="+rootPath)
flags = append(flags, "--stream-root-path="+rootPath)
+ listenClientURL, listenPeerURL, err := test.NewEtcdListenUrls()
+ Expect(err).NotTo(HaveOccurred())
+ flags = append(flags, "--etcd-listen-client-url="+listenClientURL, "--etcd-listen-peer-url="+listenPeerURL)
executor, err := query.NewExecutor(context.TODO(), streamService, measureService, metadataService, repo, pipeline)
Expect(err).NotTo(HaveOccurred())
moduleDeferFunc := test.SetUpModules(
diff --git a/banyand/stream/stream_suite_test.go b/banyand/stream/stream_suite_test.go
index 86f3322..4f827eb 100644
--- a/banyand/stream/stream_suite_test.go
+++ b/banyand/stream/stream_suite_test.go
@@ -97,6 +97,9 @@ func setUp() (*services, func()) {
rootPath, deferFunc, err := test.NewSpace()
Expect(err).NotTo(HaveOccurred())
flags = append(flags, "--stream-root-path="+rootPath)
+ listenClientURL, listenPeerURL, err := test.NewEtcdListenUrls()
+ Expect(err).NotTo(HaveOccurred())
+ flags = append(flags, "--etcd-listen-client-url="+listenClientURL, "--etcd-listen-peer-url="+listenPeerURL)
moduleDeferFunc := test.SetUpModules(
flags,
repo,
diff --git a/pkg/query/logical/common_test.go b/pkg/query/logical/common_test.go
index 2588f58..c4c6132 100644
--- a/pkg/query/logical/common_test.go
+++ b/pkg/query/logical/common_test.go
@@ -102,9 +102,12 @@ func setup(t *require.Assertions) (stream.Stream, metadata.Service, func()) {
metadataSvc, err := metadata.NewService(context.TODO())
t.NoError(err)
+ listenClientURL, listenPeerURL, err := test.NewEtcdListenUrls()
+ t.NoError(err)
etcdRootDir := teststream.RandomTempDir()
- err = metadataSvc.FlagSet().Parse([]string{"--metadata-root-path=" + etcdRootDir})
+ err = metadataSvc.FlagSet().Parse([]string{"--metadata-root-path=" + etcdRootDir,
+ "--etcd-listen-client-url=" + listenClientURL, "--etcd-listen-peer-url=" + listenPeerURL})
t.NoError(err)
streamSvc, err := stream.NewService(context.TODO(), metadataSvc, repo, pipeline)
@@ -155,9 +158,12 @@ func setupMeasure(t *require.Assertions) (measure.Measure, metadata.Service, fun
metadataSvc, err := metadata.NewService(context.TODO())
t.NoError(err)
+ listenClientURL, listenPeerURL, err := test.NewEtcdListenUrls()
+ t.NoError(err)
- etcdRootDir := testmeasure.RandomTempDir()
- err = metadataSvc.FlagSet().Parse([]string{"--metadata-root-path=" + etcdRootDir})
+ etcdRootDir := teststream.RandomTempDir()
+ err = metadataSvc.FlagSet().Parse([]string{"--metadata-root-path=" + etcdRootDir,
+ "--etcd-listen-client-url=" + listenClientURL, "--etcd-listen-peer-url=" + listenPeerURL})
t.NoError(err)
measureSvc, err := measure.NewService(context.TODO(), metadataSvc, repo, pipeline)
diff --git a/pkg/query/logical/measure_analyzer_test.go b/pkg/query/logical/measure_analyzer_test.go
index 1d803e9..332688c 100644
--- a/pkg/query/logical/measure_analyzer_test.go
+++ b/pkg/query/logical/measure_analyzer_test.go
@@ -33,6 +33,7 @@ import (
"github.com/apache/skywalking-banyandb/pkg/logger"
pb "github.com/apache/skywalking-banyandb/pkg/pb/v1"
"github.com/apache/skywalking-banyandb/pkg/query/logical"
+ "github.com/apache/skywalking-banyandb/pkg/test"
testmeasure "github.com/apache/skywalking-banyandb/pkg/test/measure"
)
@@ -52,9 +53,15 @@ func setUpMeasureAnalyzer() (*logical.MeasureAnalyzer, func(), error) {
return nil, func() {
}, err
}
+ listenClientURL, listenPeerURL, err := test.NewEtcdListenUrls()
+ if err != nil {
+ return nil, func() {
+ }, err
+ }
rootDir := testmeasure.RandomTempDir()
- err = metadataService.FlagSet().Parse([]string{"--metadata-root-path=" + rootDir})
+ err = metadataService.FlagSet().Parse([]string{"--metadata-root-path=" + rootDir,
+ "--etcd-listen-client-url=" + listenClientURL, "--etcd-listen-peer-url=" + listenPeerURL})
if err != nil {
return nil, func() {
diff --git a/pkg/query/logical/stream_analyzer_test.go b/pkg/query/logical/stream_analyzer_test.go
index 6348642..a5d697f 100644
--- a/pkg/query/logical/stream_analyzer_test.go
+++ b/pkg/query/logical/stream_analyzer_test.go
@@ -32,6 +32,7 @@ import (
"github.com/apache/skywalking-banyandb/pkg/logger"
pb "github.com/apache/skywalking-banyandb/pkg/pb/v1"
"github.com/apache/skywalking-banyandb/pkg/query/logical"
+ "github.com/apache/skywalking-banyandb/pkg/test"
teststream "github.com/apache/skywalking-banyandb/pkg/test/stream"
)
@@ -51,9 +52,15 @@ func setUpStreamAnalyzer() (*logical.StreamAnalyzer, func(), error) {
return nil, func() {
}, err
}
+ listenClientURL, listenPeerURL, err := test.NewEtcdListenUrls()
+ if err != nil {
+ return nil, func() {
+ }, err
+ }
rootDir := teststream.RandomTempDir()
- err = metadataService.FlagSet().Parse([]string{"--metadata-root-path=" + rootDir})
+ err = metadataService.FlagSet().Parse([]string{"--metadata-root-path=" + rootDir,
+ "--etcd-listen-client-url=" + listenClientURL, "--etcd-listen-peer-url=" + listenPeerURL})
if err != nil {
return nil, func() {
diff --git a/pkg/test/network.go b/pkg/test/network.go
new file mode 100644
index 0000000..95145b1
--- /dev/null
+++ b/pkg/test/network.go
@@ -0,0 +1,55 @@
+// Licensed to 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. Apache Software Foundation (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 (
+ "fmt"
+ "net"
+)
+
+// AllocateFreePorts finds n available ports
+// Copies from https://github.com/tendermint/tendermint/issues/3682
+func AllocateFreePorts(n int) ([]int, error) {
+ ports := make([]int, n)
+
+ for k, _ := range ports {
+ addr, err := net.ResolveTCPAddr("tcp", "localhost:0")
+ if err != nil {
+ return ports, err
+ }
+
+ l, err := net.ListenTCP("tcp", addr)
+ if err != nil {
+ return ports, err
+ }
+ // This is done on purpose - we want to keep ports
+ // busy to avoid collisions when getting the next one
+ defer func() { _ = l.Close() }()
+ ports[k] = l.Addr().(*net.TCPAddr).Port
+ }
+
+ return ports, nil
+}
+
+func NewEtcdListenUrls() (string, string, error) {
+ ports, err := AllocateFreePorts(2)
+ if err != nil {
+ return "", "", err
+ }
+ return fmt.Sprintf("http://127.0.0.1:%d", ports[0]), fmt.Sprintf("http://127.0.0.1:%d", ports[1]), nil
+}