Add a series of docs about operation (#523)

CHANGES.md

@@ -50,6 +50,7 @@ Release Notes.
 - Add web-ui interacting guide.
 - Add bydbctl interacting guide.
 - Add cluster management guide.
+- Add operation related documents: configuration, troubleshooting, system, upgrade, and observability.
 
 ### Chores
 

banyand/measure/merger_policy.go

@@ -20,27 +20,29 @@ package measure
 import (
 	"math"
 	"sort"
+
+	"github.com/apache/skywalking-banyandb/pkg/run"
 )
 
 // MergePolicy aims to choose an optimal combination
 // that has the lowest write amplification.
 type mergePolicy struct {
 	maxParts           int
 	minMergeMultiplier float64
-	maxFanOutSize      uint64
+	maxFanOutSize      run.Bytes
 }
 
 // NewDefaultMergePolicy create a MergePolicy with default parameters.
 func newDefaultMergePolicy() *mergePolicy {
-	return newMergePolicy(8, 1.7, math.MaxUint64)
+	return newMergePolicy(8, 1.7, run.Bytes(math.MaxInt64))
 }
 
 func newDefaultMergePolicyForTesting() *mergePolicy {
-	return newMergePolicy(4, 1.7, math.MaxUint64)
+	return newMergePolicy(4, 1.7, run.Bytes(math.MaxInt64))
 }
 
 // NewMergePolicy creates a MergePolicy with given parameters.
-func newMergePolicy(maxParts int, minMergeMul float64, maxFanOutSize uint64) *mergePolicy {
+func newMergePolicy(maxParts int, minMergeMul float64, maxFanOutSize run.Bytes) *mergePolicy {
 	return &mergePolicy{
 		maxParts:           maxParts,
 		minMergeMultiplier: minMergeMul,
@@ -53,7 +55,7 @@ func (l *mergePolicy) getPartsToMerge(dst, src []*partWrapper, freeDiskSize uint
 		return dst
 	}
 
-	maxFanOut := min(freeDiskSize, l.maxFanOutSize)
+	maxFanOut := min(freeDiskSize, uint64(l.maxFanOutSize))
 	// Filter out too big parts.
 	// This should reduce N for O(N^2) algorithm below.
 	maxInPartBytes := uint64(float64(maxFanOut) / l.minMergeMultiplier)

banyand/measure/service.go

@@ -19,7 +19,6 @@ package measure
 
 import (
 	"context"
-	"math"
 	"path"
 
 	"github.com/pkg/errors"
@@ -82,7 +81,7 @@ func (s *service) FlagSet() *run.FlagSet {
 	flagS.StringVar(&s.root, "measure-root-path", "/tmp", "the root path of database")
 	flagS.DurationVar(&s.option.flushTimeout, "measure-flush-timeout", defaultFlushTimeout, "the memory data timeout of measure")
 	s.option.mergePolicy = newDefaultMergePolicy()
-	flagS.Uint64Var(&s.option.mergePolicy.maxFanOutSize, "max-fan-out-size", math.MaxUint64, "the upper bound of a single file size after merge")
+	flagS.VarP(&s.option.mergePolicy.maxFanOutSize, "measure-max-fan-out-size", "", "the upper bound of a single file size after merge of measure")
 	return flagS
 }
 

banyand/observability/metrics_system.go

@@ -33,7 +33,7 @@ import (
 	"github.com/apache/skywalking-banyandb/pkg/meter"
 )
 
-var log = logger.GetLogger("observability", "metrics", "system")
+var log = logger.GetLogger("metrics")
 
 var (
 	cpuCount      = 0

banyand/queue/pub/pub.go

@@ -200,7 +200,7 @@ func (*pub) Name() string {
 }
 
 func (p *pub) PreRun(context.Context) error {
-	p.log = logger.GetLogger("server-queue")
+	p.log = logger.GetLogger("server-queue-pub")
 	p.metadata.RegisterHandler("queue-client", schema.KindNode, p)
 	return nil
 }

banyand/queue/sub/server.go

@@ -94,7 +94,7 @@ func NewServer(omr observability.MetricsRegistry) queue.Server {
 }
 
 func (s *server) PreRun(_ context.Context) error {
-	s.log = logger.GetLogger("server-queue")
+	s.log = logger.GetLogger("server-queue-sub")
 	s.metrics = newMetrics(s.omr.With(queueSubScope))
 	return nil
 }
@@ -129,7 +129,7 @@ func (s *server) Validate() error {
 	if s.addr == ":" {
 		return errNoAddr
 	}
-	s.healthAddr = net.JoinHostPort("127.0.0.1", strconv.FormatUint(uint64(s.healthPort), 10))
+	s.healthAddr = net.JoinHostPort(s.host, strconv.FormatUint(uint64(s.healthPort), 10))
 	if s.healthAddr == ":" {
 		return errNoAddr
 	}
@@ -194,10 +194,10 @@ func (s *server) Serve() run.StopNotify {
 	}
 	gwMux := runtime.NewServeMux(runtime.WithHealthzEndpoint(client))
 	mux := chi.NewRouter()
-	mux.Mount("/api", gwMux)
+	mux.Mount("/api", http.StripPrefix("/api", gwMux))
 	s.healthSrv = &http.Server{
 		Addr:              s.healthAddr,
-		Handler:           chi.NewRouter(),
+		Handler:           mux,
 		ReadHeaderTimeout: 3 * time.Second,
 	}
 	var wg sync.WaitGroup

banyand/stream/merger_policy.go

@@ -20,31 +20,33 @@ package stream
 import (
 	"math"
 	"sort"
+
+	"github.com/apache/skywalking-banyandb/pkg/run"
 )
 
 // MergePolicy aims to choose an optimal combination
 // that has the lowest write amplification.
 type mergePolicy struct {
 	maxParts           int
 	minMergeMultiplier float64
-	maxFanOutSize      uint64
+	maxFanOutSize      run.Bytes
 }
 
 // NewDefaultMergePolicy create a MergePolicy with default parameters.
 func newDefaultMergePolicy() *mergePolicy {
-	return newMergePolicy(15, 1.7, math.MaxUint64)
+	return newMergePolicy(15, 1.7, run.Bytes(math.MaxInt64))
 }
 
 func newDefaultMergePolicyForTesting() *mergePolicy {
-	return newMergePolicy(4, 1.7, math.MaxUint64)
+	return newMergePolicy(4, 1.7, run.Bytes(math.MaxInt64))
 }
 
 func newDisabledMergePolicyForTesting() *mergePolicy {
 	return newMergePolicy(0, 0, 0)
 }
 
 // NewMergePolicy creates a MergePolicy with given parameters.
-func newMergePolicy(maxParts int, minMergeMul float64, maxFanOutSize uint64) *mergePolicy {
+func newMergePolicy(maxParts int, minMergeMul float64, maxFanOutSize run.Bytes) *mergePolicy {
 	return &mergePolicy{
 		maxParts:           maxParts,
 		minMergeMultiplier: minMergeMul,
@@ -57,7 +59,7 @@ func (l *mergePolicy) getPartsToMerge(dst, src []*partWrapper, freeDiskSize uint
 		return dst
 	}
 
-	maxFanOut := min(freeDiskSize, l.maxFanOutSize)
+	maxFanOut := min(freeDiskSize, uint64(l.maxFanOutSize))
 	// Filter out too big parts.
 	// This should reduce N for O(N^2) algorithm below.
 	maxInPartBytes := uint64(float64(maxFanOut) / l.minMergeMultiplier)

banyand/stream/service.go

@@ -19,7 +19,6 @@ package stream
 
 import (
 	"context"
-	"math"
 	"path"
 
 	"github.com/pkg/errors"
@@ -82,7 +81,7 @@ func (s *service) FlagSet() *run.FlagSet {
 	flagS.DurationVar(&s.option.flushTimeout, "stream-flush-timeout", defaultFlushTimeout, "the memory data timeout of stream")
 	flagS.DurationVar(&s.option.elementIndexFlushTimeout, "element-index-flush-timeout", defaultFlushTimeout, "the elementIndex timeout of stream")
 	s.option.mergePolicy = newDefaultMergePolicy()
-	flagS.Uint64Var(&s.option.mergePolicy.maxFanOutSize, "max-fan-out-size", math.MaxUint64, "the upper bound of a single file size after merge")
+	flagS.VarP(&s.option.mergePolicy.maxFanOutSize, "stream-max-fan-out-size", "", "the upper bound of a single file size after merge of stream")
 	return flagS
 }
 

bydbctl/internal/cmd/group.go

@@ -61,7 +61,7 @@ func newGroupCmd() *cobra.Command {
 					fmt.Printf("group %s is created", reqBody.name)
 					fmt.Println()
 					return nil
-				}, enableTLS, insecure, grpcCert)
+				}, enableTLS, insecure, cert)
 		},
 	}
 
@@ -90,7 +90,7 @@ func newGroupCmd() *cobra.Command {
 					fmt.Printf("group %s is updated", reqBody.name)
 					fmt.Println()
 					return nil
-				}, enableTLS, insecure, grpcCert)
+				}, enableTLS, insecure, cert)
 		},
 	}
 	bindFileFlag(createCmd, updateCmd)
@@ -102,7 +102,7 @@ func newGroupCmd() *cobra.Command {
 		RunE: func(_ *cobra.Command, _ []string) (err error) {
 			return rest(parseFromFlags, func(request request) (*resty.Response, error) {
 				return request.req.SetPathParam("group", request.group).Get(getPath("/api/v1/group/schema/{group}"))
-			}, yamlPrinter, enableTLS, insecure, grpcCert)
+			}, yamlPrinter, enableTLS, insecure, cert)
 		},
 	}
 
@@ -118,7 +118,7 @@ func newGroupCmd() *cobra.Command {
 					fmt.Printf("group %s is deleted", reqBody.group)
 					fmt.Println()
 					return nil
-				}, enableTLS, insecure, grpcCert)
+				}, enableTLS, insecure, cert)
 		},
 	}
 
@@ -129,7 +129,7 @@ func newGroupCmd() *cobra.Command {
 		RunE: func(_ *cobra.Command, _ []string) (err error) {
 			return rest(nil, func(request request) (*resty.Response, error) {
 				return request.req.Get(getPath("/api/v1/group/schema/lists"))
-			}, yamlPrinter, enableTLS, insecure, grpcCert)
+			}, yamlPrinter, enableTLS, insecure, cert)
 		},
 	}
 

bydbctl/internal/cmd/health_check.go

@@ -49,15 +49,15 @@ func newHealthCheckCmd() *cobra.Command {
 			if grpcAddr == "" {
 				return rest(nil, func(request request) (*resty.Response, error) {
 					return request.req.Get(getPath("/api/healthz"))
-				}, yamlPrinter, enableTLS, insecure, grpcCert)
+				}, yamlPrinter, enableTLS, insecure, cert)
 			}
 			if enableTLS {
 				config := &tls.Config{
 					// #nosec G402
 					InsecureSkipVerify: insecure,
 				}
-				if grpcCert != "" {
-					cert, errRead := os.ReadFile(grpcCert)
+				if cert != "" {
+					cert, errRead := os.ReadFile(cert)
 					if errRead != nil {
 						return errRead
 					}

bydbctl/internal/cmd/health_check_test.go

@@ -30,7 +30,7 @@ import (
 	"github.com/apache/skywalking-banyandb/pkg/test/setup"
 )
 
-var _ = Describe("health check after launching banyandb server", func() {
+var _ = Describe("health check after launching banyandb server with gRPC and HTTP TLS", func() {
 	var deferFunc func()
 	var grpcAddr, httpAddr, certFile string
 	var rootCmd *cobra.Command
@@ -47,16 +47,31 @@ var _ = Describe("health check after launching banyandb server", func() {
 	})
 
 	It("http health check should pass", func() {
-		rootCmd.SetArgs([]string{"health", "--addr", "http://" + httpAddr})
+		rootCmd.SetArgs([]string{"health", "--addr", "https://" + httpAddr, "--cert", certFile, "--enable-tls", "true"})
 		out := capturer.CaptureStdout(func() {
 			err := rootCmd.Execute()
 			Expect(err).NotTo(HaveOccurred())
 		})
 		Expect(out).To(ContainSubstring("SERVING"))
 	})
 
+	It("http health check should pass with insecure flag set", func() {
+		rootCmd.SetArgs([]string{"health", "--addr", "https://" + httpAddr, "--insecure", "true", "--enable-tls", "true"})
+		out := capturer.CaptureStdout(func() {
+			err := rootCmd.Execute()
+			Expect(err).NotTo(HaveOccurred())
+		})
+		Expect(out).To(ContainSubstring("SERVING"))
+	})
+
+	It("http health check should fail without the proper cert", func() {
+		rootCmd.SetArgs([]string{"health", "--addr", "https://" + httpAddr, "--enable-tls", "true"})
+		err := rootCmd.Execute()
+		Expect(err).To(HaveOccurred())
+	})
+
 	It("grpc health check should pass", func() {
-		rootCmd.SetArgs([]string{"health", "--grpc-addr", grpcAddr, "--grpc-cert", certFile, "--enable-tls", "true"})
+		rootCmd.SetArgs([]string{"health", "--grpc-addr", grpcAddr, "--cert", certFile, "--enable-tls", "true"})
 		out := capturer.CaptureStdout(func() {
 			err := rootCmd.Execute()
 			Expect(err).NotTo(HaveOccurred())
@@ -84,6 +99,39 @@ var _ = Describe("health check after launching banyandb server", func() {
 	})
 })
 
+var _ = Describe("health check after launching banyandb server", func() {
+	var deferFunc func()
+	var grpcAddr, httpAddr string
+	var rootCmd *cobra.Command
+	BeforeEach(func() {
+		grpcAddr, httpAddr, deferFunc = setup.Standalone()
+		rootCmd = &cobra.Command{Use: "root"}
+		cmd.RootCmdFlags(rootCmd)
+	})
+
+	It("http health check should pass", func() {
+		rootCmd.SetArgs([]string{"health", "--addr", "http://" + httpAddr})
+		out := capturer.CaptureStdout(func() {
+			err := rootCmd.Execute()
+			Expect(err).NotTo(HaveOccurred())
+		})
+		Expect(out).To(ContainSubstring("SERVING"))
+	})
+
+	It("grpc health check should pass", func() {
+		rootCmd.SetArgs([]string{"health", "--grpc-addr", grpcAddr})
+		out := capturer.CaptureStdout(func() {
+			err := rootCmd.Execute()
+			Expect(err).NotTo(HaveOccurred())
+		})
+		Expect(out).To(ContainSubstring("connected"))
+	})
+
+	AfterEach(func() {
+		deferFunc()
+	})
+})
+
 var _ = Describe("health check without launching banyandb server", func() {
 	var rootCmd *cobra.Command
 	BeforeEach(func() {

bydbctl/internal/cmd/index_rule.go

@@ -64,7 +64,7 @@ func newIndexRuleCmd() *cobra.Command {
 					fmt.Printf("indexRule %s.%s is created", reqBody.group, reqBody.name)
 					fmt.Println()
 					return nil
-				}, enableTLS, insecure, grpcCert)
+				}, enableTLS, insecure, cert)
 		},
 	}
 
@@ -95,7 +95,7 @@ func newIndexRuleCmd() *cobra.Command {
 					fmt.Printf("indexRule %s.%s is updated", reqBody.group, reqBody.name)
 					fmt.Println()
 					return nil
-				}, enableTLS, insecure, grpcCert)
+				}, enableTLS, insecure, cert)
 		},
 	}
 
@@ -106,7 +106,7 @@ func newIndexRuleCmd() *cobra.Command {
 		RunE: func(_ *cobra.Command, _ []string) (err error) {
 			return rest(parseFromFlags, func(request request) (*resty.Response, error) {
 				return request.req.SetPathParam("name", request.name).SetPathParam("group", request.group).Get(getPath(indexRuleSchemaPathWithParams))
-			}, yamlPrinter, enableTLS, insecure, grpcCert)
+			}, yamlPrinter, enableTLS, insecure, cert)
 		},
 	}
 
@@ -121,7 +121,7 @@ func newIndexRuleCmd() *cobra.Command {
 				fmt.Printf("indexRule %s.%s is deleted", reqBody.group, reqBody.name)
 				fmt.Println()
 				return nil
-			}, enableTLS, insecure, grpcCert)
+			}, enableTLS, insecure, cert)
 		},
 	}
 	bindNameFlag(getCmd, deleteCmd)
@@ -133,7 +133,7 @@ func newIndexRuleCmd() *cobra.Command {
 		RunE: func(_ *cobra.Command, _ []string) (err error) {
 			return rest(parseFromFlags, func(request request) (*resty.Response, error) {
 				return request.req.SetPathParam("group", request.group).Get(getPath("/api/v1/index-rule/schema/lists/{group}"))
-			}, yamlPrinter, enableTLS, insecure, grpcCert)
+			}, yamlPrinter, enableTLS, insecure, cert)
 		},
 	}