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

Signed-off-by: jyjiangkai <jyjiangkai@163.com>

Author: hwjiangkai

.golangci.yaml

@@ -396,6 +396,9 @@ issues:
     - source: "strconv|make|len|math"
       linters:
         - gomnd
+    - path: "action"
+      linters:
+        - dupl
     - path: "convert.go"
       linters:
         - dupl
@@ -412,7 +415,7 @@ issues:
       linters:
         - gosec
         - gomnd
-    - path: "action.go"
+    - path: "init.go"
       linters:
         - gochecknoinits
     - path: "^vsctl"

client/go.mod

@@ -3,7 +3,6 @@ module github.com/linkall-labs/vanus/client
 go 1.18
 
 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.1
@@ -29,9 +28,7 @@ require (
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/sirupsen/logrus v1.9.0 // indirect
-	github.com/stretchr/objx v0.4.0 // indirect
 	go.opentelemetry.io/otel v1.11.1 // indirect
-	go.opentelemetry.io/otel/exporters/jaeger v1.9.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/internal/retry v1.9.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.9.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.9.0 // indirect
@@ -49,7 +46,7 @@ require (
 )
 
 replace (
-	cloudevents.io/genproto => ../proto/include/cloudevents/pkg
+	cloud.google.com/go => cloud.google.com/go v0.100.2
 	github.com/linkall-labs/vanus/observability => ../observability
 	github.com/linkall-labs/vanus/pkg => ../pkg
 	github.com/linkall-labs/vanus/proto => ../proto

client/go.sum

@@ -12,9 +12,9 @@ import (
 	stdtime "time"
 
 	// third-party libraries
-	cepb "cloudevents.io/genproto/v1"
 	"github.com/cloudevents/sdk-go/v2/event"
 	"github.com/cloudevents/sdk-go/v2/types"
+	cepb "github.com/linkall-labs/vanus/proto/pkg/cloudevents"
 	"google.golang.org/protobuf/types/known/anypb"
 	"google.golang.org/protobuf/types/known/timestamppb"
 )
@@ -35,14 +35,14 @@ var (
 	zeroTime = stdtime.Time{}
 )
 
-// convert an SDK event to a protobuf variant of the event that can be marshaled.
+// ToProto convert an SDK event to a protobuf variant of the event that can be marshaled.
 func ToProto(e *event.Event) (*cepb.CloudEvent, error) {
 	container := &cepb.CloudEvent{
 		Id:          e.ID(),
 		Source:      e.Source(),
 		SpecVersion: e.SpecVersion(),
 		Type:        e.Type(),
-		Attributes:  make(map[string]*cepb.CloudEventAttributeValue),
+		Attributes:  make(map[string]*cepb.CloudEvent_CloudEventAttributeValue),
 	}
 	if e.DataContentType() != "" {
 		container.Attributes[datacontenttype], _ = attributeFor(e.DataContentType())
@@ -78,39 +78,39 @@ func ToProto(e *event.Event) (*cepb.CloudEvent, error) {
 	return container, nil
 }
 
-func attributeFor(v interface{}) (*cepb.CloudEventAttributeValue, error) {
+func attributeFor(v interface{}) (*cepb.CloudEvent_CloudEventAttributeValue, error) {
 	vv, err := types.Validate(v)
 	if err != nil {
 		return nil, err
 	}
-	attr := &cepb.CloudEventAttributeValue{}
+	attr := &cepb.CloudEvent_CloudEventAttributeValue{}
 	switch vt := vv.(type) {
 	case bool:
-		attr.Attr = &cepb.CloudEventAttributeValue_CeBoolean{
+		attr.Attr = &cepb.CloudEvent_CloudEventAttributeValue_CeBoolean{
 			CeBoolean: vt,
 		}
 	case int32:
-		attr.Attr = &cepb.CloudEventAttributeValue_CeInteger{
+		attr.Attr = &cepb.CloudEvent_CloudEventAttributeValue_CeInteger{
 			CeInteger: vt,
 		}
 	case string:
-		attr.Attr = &cepb.CloudEventAttributeValue_CeString{
+		attr.Attr = &cepb.CloudEvent_CloudEventAttributeValue_CeString{
 			CeString: vt,
 		}
 	case []byte:
-		attr.Attr = &cepb.CloudEventAttributeValue_CeBytes{
+		attr.Attr = &cepb.CloudEvent_CloudEventAttributeValue_CeBytes{
 			CeBytes: vt,
 		}
 	case types.URI:
-		attr.Attr = &cepb.CloudEventAttributeValue_CeUri{
+		attr.Attr = &cepb.CloudEvent_CloudEventAttributeValue_CeUri{
 			CeUri: vt.String(),
 		}
 	case types.URIRef:
-		attr.Attr = &cepb.CloudEventAttributeValue_CeUriRef{
+		attr.Attr = &cepb.CloudEvent_CloudEventAttributeValue_CeUriRef{
 			CeUriRef: vt.String(),
 		}
 	case types.Timestamp:
-		attr.Attr = &cepb.CloudEventAttributeValue_CeTimestamp{
+		attr.Attr = &cepb.CloudEvent_CloudEventAttributeValue_CeTimestamp{
 			CeTimestamp: timestamppb.New(vt.Time),
 		}
 	default:
@@ -119,38 +119,38 @@ func attributeFor(v interface{}) (*cepb.CloudEventAttributeValue, error) {
 	return attr, nil
 }
 
-func valueFrom(attr *cepb.CloudEventAttributeValue) (interface{}, error) {
+func valueFrom(attr *cepb.CloudEvent_CloudEventAttributeValue) (interface{}, error) {
 	var v interface{}
 	switch vt := attr.Attr.(type) {
-	case *cepb.CloudEventAttributeValue_CeBoolean:
+	case *cepb.CloudEvent_CloudEventAttributeValue_CeBoolean:
 		v = vt.CeBoolean
-	case *cepb.CloudEventAttributeValue_CeInteger:
+	case *cepb.CloudEvent_CloudEventAttributeValue_CeInteger:
 		v = vt.CeInteger
-	case *cepb.CloudEventAttributeValue_CeString:
+	case *cepb.CloudEvent_CloudEventAttributeValue_CeString:
 		v = vt.CeString
-	case *cepb.CloudEventAttributeValue_CeBytes:
+	case *cepb.CloudEvent_CloudEventAttributeValue_CeBytes:
 		v = vt.CeBytes
-	case *cepb.CloudEventAttributeValue_CeUri:
+	case *cepb.CloudEvent_CloudEventAttributeValue_CeUri:
 		uri, err := url.Parse(vt.CeUri)
 		if err != nil {
 			return nil, fmt.Errorf("failed to parse URI value %s: %s", vt.CeUri, err.Error())
 		}
 		v = uri
-	case *cepb.CloudEventAttributeValue_CeUriRef:
+	case *cepb.CloudEvent_CloudEventAttributeValue_CeUriRef:
 		uri, err := url.Parse(vt.CeUriRef)
 		if err != nil {
 			return nil, fmt.Errorf("failed to parse URIRef value %s: %s", vt.CeUriRef, err.Error())
 		}
 		v = types.URIRef{URL: *uri}
-	case *cepb.CloudEventAttributeValue_CeTimestamp:
+	case *cepb.CloudEvent_CloudEventAttributeValue_CeTimestamp:
 		v = vt.CeTimestamp.AsTime()
 	default:
 		return nil, fmt.Errorf("unsupported attribute type: %T", vt)
 	}
 	return types.Validate(v)
 }
 
-// Convert from a protobuf variant into the generic, SDK event.
+// FromProto Convert from a protobuf variant into the generic, SDK event.
 func FromProto(container *cepb.CloudEvent) (*event.Event, error) {
 	e := event.New()
 	e.SetID(container.Id)
@@ -179,7 +179,7 @@ func FromProto(container *cepb.CloudEvent) (*event.Event, error) {
 	if container.Attributes != nil {
 		attr := container.Attributes[datacontenttype]
 		if attr != nil {
-			if stattr, ok := attr.Attr.(*cepb.CloudEventAttributeValue_CeString); ok {
+			if stattr, ok := attr.Attr.(*cepb.CloudEvent_CloudEventAttributeValue_CeString); ok {
 				contentType = stattr.CeString
 			}
 		}

client/internal/vanus/codec/protobuf.go

@@ -25,12 +25,13 @@ import (
 	"github.com/linkall-labs/vanus/observability/tracing"
 	"go.opentelemetry.io/otel/trace"
 
-	// third-party libraries
-	cepb "cloudevents.io/genproto/v1"
 	ce "github.com/cloudevents/sdk-go/v2"
 	"google.golang.org/grpc"
 
 	// first-party libraries
+
+	// third-party libraries
+	cepb "github.com/linkall-labs/vanus/proto/pkg/cloudevents"
 	errpb "github.com/linkall-labs/vanus/proto/pkg/errors"
 	segpb "github.com/linkall-labs/vanus/proto/pkg/segment"
 
@@ -560,3 +561,25 @@ func (s *BlockStore) LookupOffset(ctx context.Context, blockID uint64, t time.Ti
 	}
 	return res.Offset, nil
 }
+
+func (s *BlockStore) AppendBatch(ctx context.Context, block uint64, event *cepb.CloudEventBatch) (int64, error) {
+	_ctx, span := s.tracer.Start(ctx, "AppendBatch")
+	defer span.End()
+
+	req := &segpb.AppendToBlockRequest{
+		BlockId: block,
+		Events:  event,
+	}
+
+	client, err := s.client.Get(_ctx)
+	if err != nil {
+		return -1, err
+	}
+
+	res, err := client.(segpb.SegmentServerClient).AppendToBlock(_ctx, req)
+	if err != nil {
+		return -1, err
+	}
+	// TODO(Y. F. Zhang): batch events
+	return res.GetOffsets()[0], nil
+}

client/internal/vanus/store/block_store.go

@@ -19,6 +19,7 @@ import (
 	"context"
 
 	ce "github.com/cloudevents/sdk-go/v2"
+	"github.com/linkall-labs/vanus/proto/pkg/cloudevents"
 )
 
 type Eventbus interface {
@@ -33,6 +34,7 @@ type Eventbus interface {
 type BusWriter interface {
 	AppendOne(ctx context.Context, event *ce.Event, opts ...WriteOption) (eid string, err error)
 	AppendMany(ctx context.Context, events []*ce.Event, opts ...WriteOption) (eid []string, err error)
+	AppendBatch(ctx context.Context, events *cloudevents.CloudEventBatch, opts ...WriteOption) (err error)
 }
 
 type BusReader interface {

client/pkg/api/client.go

@@ -20,6 +20,7 @@ import (
 	"encoding/base64"
 	"encoding/binary"
 	stderrors "errors"
+	"github.com/linkall-labs/vanus/proto/pkg/cloudevents"
 	"io"
 	"sync"
 
@@ -444,6 +445,29 @@ type busWriter struct {
 	tracer *tracing.Tracer
 }
 
+func (w *busWriter) AppendBatch(ctx context.Context, events *cloudevents.CloudEventBatch, opts ...api.WriteOption) (err error) {
+	_ctx, span := w.tracer.Start(ctx, "CloudEventBatch")
+	defer span.End()
+
+	var writeOpts *api.WriteOptions = w.opts
+	if len(opts) > 0 {
+		writeOpts = w.opts.Copy()
+		for _, opt := range opts {
+			opt(writeOpts)
+		}
+	}
+
+	// 1. pick a writer of eventlog
+	lw, err := w.pickWritableLog(_ctx, writeOpts)
+	if err != nil {
+		return err
+	}
+
+	// 2. append the event to the eventlog
+	_, err = lw.AppendMany(_ctx, events)
+	return err
+}
+
 var _ api.BusWriter = (*busWriter)(nil)
 
 func (w *busWriter) AppendOne(ctx context.Context, event *ce.Event, opts ...api.WriteOption) (eid string, err error) {

client/pkg/api/mock_client.go

@@ -19,6 +19,8 @@ import (
 	// standard libraries.
 	"context"
 
+	"github.com/linkall-labs/vanus/proto/pkg/cloudevents"
+
 	// third-party libraries.
 	ce "github.com/cloudevents/sdk-go/v2"
 
@@ -51,6 +53,8 @@ type LogWriter interface {
 
 	Append(ctx context.Context, event *ce.Event) (off int64, err error)
 
+	AppendMany(ctx context.Context, events *cloudevents.CloudEventBatch) (off int64, err error)
+
 	AppendManyStream(ctx context.Context, events []*ce.Event) ([]int64, error)
 }
 

client/pkg/eventbus/eventbus.go

@@ -22,6 +22,8 @@ import (
 	"sync"
 	"time"
 
+	"github.com/linkall-labs/vanus/proto/pkg/cloudevents"
+
 	"github.com/linkall-labs/vanus/observability/tracing"
 	"go.opentelemetry.io/otel/trace"
 
@@ -329,6 +331,10 @@ func (l *eventlog) refreshReadableSegments(ctx context.Context) {
 	_ = l.readableWatcher.Refresh(ctx)
 }
 
+var (
+	_ LogWriter = &logWriter{}
+)
+
 // logWriter is the writer of eventlog.
 //
 // Append is thread-safety.
@@ -338,6 +344,28 @@ type logWriter struct {
 	mu   sync.RWMutex
 }
 
+func (w *logWriter) AppendMany(ctx context.Context, events *cloudevents.CloudEventBatch) (off int64, err error) {
+	retryTimes := defaultRetryTimes
+	for i := 1; i <= retryTimes; i++ {
+		offset, err := w.doAppendBatch(ctx, events)
+		if err == nil {
+			return offset, nil
+		}
+		vlog.Warning(ctx, "failed to Append", map[string]interface{}{
+			vlog.KeyError: err,
+			"offset":      offset,
+		})
+		if errors.Is(err, errors.ErrFull) {
+			if i < retryTimes {
+				continue
+			}
+		}
+		return -1, err
+	}
+
+	return -1, errors.ErrUnknown
+}
+
 func (w *logWriter) Log() Eventlog {
 	return w.elog
 }
@@ -428,6 +456,21 @@ func (w *logWriter) doSyncAppendStream(ctx context.Context, events []*ce.Event)
 	return offsets, nil
 }
 
+func (w *logWriter) doAppendBatch(ctx context.Context, event *cloudevents.CloudEventBatch) (int64, error) {
+	segment, err := w.selectWritableSegment(ctx)
+	if err != nil {
+		return -1, err
+	}
+	offset, err := segment.AppendBatch(ctx, event)
+	if err != nil {
+		if errors.Is(err, errors.ErrFull) {
+			segment.SetNotWritable()
+		}
+		return -1, err
+	}
+	return offset, nil
+}
+
 func (w *logWriter) selectWritableSegment(ctx context.Context) (*segment, error) {
 	segment := func() *segment {
 		w.mu.RLock()