Merge branch 'main' of github.com:linkall-labs/vanus into segment

Signed-off-by: jyjiangkai <[email protected]>

Author: hwjiangkai

Makefile

@@ -4,7 +4,7 @@ GIT_COMMIT=$(shell git log -1 --format='%h' | awk '{print $0}')
 DATE=$(shell date +%Y-%m-%d_%H:%M:%S%z)
 GO_VERSION=$(shell go version)
 
-export VANUS_LOG_LEVEL=debug
+export VANUS_LOG_LEVEL=info
 
 DOCKER_REGISTRY ?= public.ecr.aws
 DOCKER_REPO ?= ${DOCKER_REGISTRY}/vanus

client/client.go

@@ -40,7 +40,7 @@ type client struct {
 }
 
 func (c *client) Eventbus(ctx context.Context, ebName string) api.Eventbus {
-	_, span := c.tracer.Start(ctx, "Eventbus")
+	_, span := c.tracer.Start(ctx, "EventbusService")
 	defer span.End()
 
 	bus := func() api.Eventbus {

client/go.mod

@@ -6,9 +6,9 @@ require (
 	cloudevents.io/genproto v1.0.2
 	github.com/cloudevents/sdk-go/v2 v2.11.0
 	github.com/golang/mock v1.6.0
-	github.com/linkall-labs/vanus/observability v0.5.0
-	github.com/linkall-labs/vanus/pkg v0.5.0
-	github.com/linkall-labs/vanus/proto v0.5.0
+	github.com/linkall-labs/vanus/observability v0.5.1
+	github.com/linkall-labs/vanus/pkg v0.5.1
+	github.com/linkall-labs/vanus/proto v0.5.1
 	github.com/scylladb/go-set v1.0.2
 	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.36.4
 	go.opentelemetry.io/otel/trace v1.11.1

client/internal/vanus/eventbus/name_service.go

@@ -17,12 +17,11 @@ package eventbus
 import (
 	// standard libraries
 	"context"
+	"github.com/linkall-labs/vanus/pkg/cluster"
 
 	"github.com/linkall-labs/vanus/observability/tracing"
 	"go.opentelemetry.io/otel/trace"
 
-	// third-party libraries
-	"github.com/linkall-labs/vanus/pkg/controller"
 	"google.golang.org/grpc/credentials/insecure"
 
 	// first-party libraries
@@ -33,7 +32,7 @@ import (
 
 func NewNameService(endpoints []string) *NameService {
 	return &NameService{
-		client: controller.NewEventbusClient(endpoints, insecure.NewCredentials()),
+		client: cluster.NewClusterController(endpoints, insecure.NewCredentials()).EventbusService().RawClient(),
 		tracer: tracing.NewTracer("internal.discovery.eventbus", trace.SpanKindClient),
 	}
 }

client/internal/vanus/eventlog/name_service.go

@@ -25,19 +25,20 @@ import (
 	"google.golang.org/grpc/credentials/insecure"
 
 	// first-party libraries.
+	"github.com/linkall-labs/vanus/observability/log"
 	"github.com/linkall-labs/vanus/observability/tracing"
-	"github.com/linkall-labs/vanus/pkg/controller"
+	"github.com/linkall-labs/vanus/pkg/cluster"
 	ctrlpb "github.com/linkall-labs/vanus/proto/pkg/controller"
 	metapb "github.com/linkall-labs/vanus/proto/pkg/meta"
 
 	// this project.
-	"github.com/linkall-labs/vanus/client/pkg/errors"
 	"github.com/linkall-labs/vanus/client/pkg/record"
+	"github.com/linkall-labs/vanus/pkg/errors"
 )
 
 func NewNameService(endpoints []string) *NameService {
 	return &NameService{
-		client: controller.NewEventlogClient(endpoints, insecure.NewCredentials()),
+		client: cluster.NewClusterController(endpoints, insecure.NewCredentials()).EventlogService().RawClient(),
 		tracer: tracing.NewTracer("internal.discovery.eventlog", trace.SpanKindClient),
 	}
 }
@@ -58,9 +59,18 @@ func (ns *NameService) LookupWritableSegment(ctx context.Context, logID uint64)
 
 	resp, err := ns.client.GetAppendableSegment(ctx, req)
 	if err != nil {
+		log.Warning(ctx, "failed to GetAppendableSegment", map[string]interface{}{
+			"req":        req,
+			"res":        resp,
+			log.KeyError: err,
+		})
 		return nil, err
 	}
 
+	log.Debug(ctx, "GetAppendableSegment result", map[string]interface{}{
+		"req": req,
+		"res": resp,
+	})
 	segments := toSegments(resp.GetSegments())
 	if len(segments) == 0 {
 		return nil, errors.ErrNotWritable

client/internal/vanus/net/rpc/bare/client.go

@@ -17,20 +17,20 @@ package bare
 import (
 	// standard libraries
 	"context"
-	"github.com/linkall-labs/vanus/observability/tracing"
-	"go.opentelemetry.io/otel/trace"
 	"sync"
 	"time"
 
 	// third-party libraries
+	"go.opentelemetry.io/otel/trace"
 	"go.uber.org/atomic"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/connectivity"
 
 	// this project
 	"github.com/linkall-labs/vanus/client/internal/vanus/net/connection"
 	"github.com/linkall-labs/vanus/client/internal/vanus/net/rpc"
-	"github.com/linkall-labs/vanus/client/pkg/errors"
+	"github.com/linkall-labs/vanus/observability/tracing"
+	"github.com/linkall-labs/vanus/pkg/errors"
 )
 
 const (

client/internal/vanus/store/allocator.go

@@ -16,14 +16,15 @@ package store
 
 import (
 	"context"
+
 	"github.com/linkall-labs/vanus/observability/tracing"
 	"go.opentelemetry.io/otel/trace"
 
 	// standard libraries.
 	"sync"
 
 	// this project.
-	"github.com/linkall-labs/vanus/client/pkg/errors"
+	"github.com/linkall-labs/vanus/pkg/errors"
 )
 
 func NewAllocator() *Allocator {

client/internal/vanus/store/block_store.go

@@ -17,7 +17,6 @@ package store
 import (
 	// standard libraries
 	"context"
-	"strings"
 	"time"
 
 	"github.com/linkall-labs/vanus/observability/tracing"
@@ -27,18 +26,14 @@ import (
 	cepb "cloudevents.io/genproto/v1"
 	ce "github.com/cloudevents/sdk-go/v2"
 	"google.golang.org/grpc"
-	"google.golang.org/grpc/codes"
-	"google.golang.org/grpc/status"
 
 	// first-party libraries
-
 	segpb "github.com/linkall-labs/vanus/proto/pkg/segment"
 
 	// this project
 	"github.com/linkall-labs/vanus/client/internal/vanus/codec"
 	"github.com/linkall-labs/vanus/client/internal/vanus/net/rpc"
 	"github.com/linkall-labs/vanus/client/internal/vanus/net/rpc/bare"
-	"github.com/linkall-labs/vanus/client/pkg/errors"
 	"github.com/linkall-labs/vanus/client/pkg/primitive"
 )
 
@@ -94,12 +89,7 @@ func (s *BlockStore) Append(ctx context.Context, block uint64, event *ce.Event)
 
 	res, err := client.(segpb.SegmentServerClient).AppendToBlock(_ctx, req)
 	if err != nil {
-		sts := status.Convert(err)
-		// TODO: temporary scheme, wait for error code reconstruction
-		if strings.Contains(sts.Message(), "SEGMENT_FULL") {
-			return -1, errors.ErrNoSpace
-		}
-		return -1, errors.ErrNotWritable
+		return -1, err
 	}
 	// TODO(Y. F. Zhang): batch events
 	return res.GetOffsets()[0], nil
@@ -125,17 +115,6 @@ func (s *BlockStore) Read(
 
 	resp, err := client.(segpb.SegmentServerClient).ReadFromBlock(ctx, req)
 	if err != nil {
-		// TODO: convert error
-		if errStatus, ok := status.FromError(err); ok {
-			errMsg := errStatus.Message()
-			if strings.Contains(errMsg, "the offset on end") {
-				err = errors.ErrOnEnd
-			} else if strings.Contains(errMsg, "the offset exceeded") {
-				err = errors.ErrOverflow
-			} else if errStatus.Code() == codes.DeadlineExceeded {
-				err = errors.ErrTimeout
-			}
-		}
 		return nil, err
 	}