Repository: matrixorigin/matrixone
Pull request #3839 (2022-07-09 16:06:14)

logservice: added helper function for testing
**What type of PR is this?**

- [ ] API-change
- [ ] BUG
- [ ] Improvement
- [ ] Documentation
- [ ] Feature
- [x] Test and CI
- [ ] Code Refactoring

**Which issue(s) this PR fixes:**

issue #

**What this PR does / why we need it:**

Added helper function that can make it easier for @zhangxu19830126 and @jiangxinmeng1 to use LogService for testing.

**Special notes for your reviewer:**

Not Available

**Additional documentation (e.g. design docs, usage docs, etc.):**

Not Available



Patch:
diff --git a/pkg/hakeeper/rsm.go b/pkg/hakeeper/rsm.go
index 9a1063b0f..03f9572d0 100644
--- a/pkg/hakeeper/rsm.go
+++ b/pkg/hakeeper/rsm.go
@@ -44,7 +44,7 @@ const (
 	// DefaultHAKeeperShardID is the shard ID assigned to the special HAKeeper
 	// shard.
 	DefaultHAKeeperShardID uint64 = 0
-	headerSize                    = 4
+	headerSize                    = pb.HeaderSize
 )
 
 type StateQuery struct{}
diff --git a/pkg/logservice/client.go b/pkg/logservice/client.go
index 9d6161b01..b06c99843 100644
--- a/pkg/logservice/client.go
+++ b/pkg/logservice/client.go
@@ -51,10 +51,10 @@ func IsTempError(err error) bool {
 type ClientConfig struct {
 	// ReadOnly indicates whether this is a read-only client.
 	ReadOnly bool
-	// ShardID is the shard ID of the log service shard to be used.
-	ShardID uint64
-	// ReplicaID is the replica ID of the DN that owns the created client.
-	ReplicaID uint64
+	// LogShardID is the shard ID of the log service shard to be used.
+	LogShardID uint64
+	// DNReplicaID is the replica ID of the DN that owns the created client.
+	DNReplicaID uint64
 	// LogService nodes service addresses. This field is provided for testing
 	// purposes only.
 	ServiceAddresses []string
@@ -62,12 +62,40 @@ type ClientConfig struct {
 
 // Client is the Log Service Client interface exposed to the DN.
 type Client interface {
+	// Close closes the client.
 	Close() error
+	// Config returns the specified configuration when creating the client.
 	Config() ClientConfig
+	// GetLogRecord returns a new LogRecord instance with its Data field enough
+	// to hold payloadLength bytes of payload. The layout of the Data field is
+	// 4 bytes of record type (pb.UserEntryUpdate) + 8 bytes DN replica ID +
+	// payloadLength bytes of actual payload.
+	GetLogRecord(payloadLength int) pb.LogRecord
+	// Append appends the specified LogRecrd into the Log Service. On success, the
+	// assigned Lsn will be returned. For the specified LogRecord, only its Data
+	// field is used with all other fields ignored by Append(). Once returned, the
+	// pb.LogRecord can be reused.
 	Append(ctx context.Context, rec pb.LogRecord) (Lsn, error)
+	// Read reads the Log Service from the specified Lsn position until the
+	// returned LogRecord set reachs the specified maxSize in bytes. The returned
+	// Lsn indicates the next Lsn to use to resume the read, or it means
+	// everything available has been read when it equals to the specified Lsn.
+	// The returned pb.LogRecord records will have their Lsn and Type fields set,
+	// the Lsn field is the Lsn assigned to the record while the Type field tells
+	// whether the record is an internal record generated by the Log Service itself
+	// or appended by the user.
 	Read(ctx context.Context, firstLsn Lsn, maxSize uint64) ([]pb.LogRecord, Lsn, error)
+	// Truncate truncates the Log Service log at the specified Lsn with Lsn
+	// itself included. This allows the Log Service to free up storage capacities
+	// for future appends, all future reads must start after the specified Lsn
+	// position.
 	Truncate(ctx context.Context, lsn Lsn) error
+	// GetTruncatedLsn returns the largest Lsn value that has been specified for
+	// truncation.
 	GetTruncatedLsn(ctx context.Context) (Lsn, error)
+	// GetTSOTimestamp requests a total of count unique timestamps from the TSO and
+	// return the first assigned such timestamp, that is TSO timestamps
+	// [returned value, returned value + count] will be owned by the caller.
 	GetTSOTimestamp(ctx context.Context, count uint64) (uint64, error)
 }
 
@@ -81,11 +109,11 @@ type client struct {
 
 var _ Client = (*client)(nil)
 
-// CreateClient creates a Log Service client. Each returned client can be used
+// NewClient creates a Log Service client. Each returned client can be used
 // to synchronously issue requests to the Log Service. To send multiple requests
 // to the Log Service in parallel, multiple clients should be created and used
 // to do so.
-func CreateClient(ctx context.Context,
+func NewClient(ctx context.Context,
 	cfg ClientConfig) (Client, error) {
 	pool := &sync.Pool{}
 	pool.New = func() interface{} {
@@ -126,19 +154,21 @@ func CreateClient(ctx context.Context,
 	return nil, e
 }
 
-// Close closes the client.
 func (c *client) Close() error {
 	return c.client.Close()
 }
 
-// Config returns the specified configuration when creating the client.
 func (c *client) Config() ClientConfig {
 	return c.cfg
 }
 
-// Append appends the specified LogRecrd into the Log Service. On success, the
-// assigned Lsn will be returned. For the specified LogRecord, only its Dsta
-// field is used with all other fields ignored by Append().
+func (c *client) GetLogRecord(payloadLength int) pb.LogRecord {
+	data := make([]byte, headerSize+8+payloadLength)
+	binaryEnc.PutUint32(data, uint32(pb.UserEntryUpdate))
+	binaryEnc.PutUint64(data[headerSize:], c.cfg.DNReplicaID)
+	return pb.LogRecord{Data: data}
+}
+
 func (c *client) Append(ctx context.Context, rec pb.LogRecord) (Lsn, error) {
 	if c.readOnly() {
 		return 0, ErrIncompatibleClient
@@ -147,19 +177,11 @@ func (c *client) Append(ctx context.Context, rec pb.LogRecord) (Lsn, error) {
 	return c.append(ctx, rec)
 }
 
-// Read reads the Log Service from the specified Lsn position until the
-// returned LogRecord set reachs the specified maxSize in bytes. The returned
-// Lsn indicates the next Lsn to use to resume the read, or it means
-// everything available has been read when it equals to the specified Lsn.
 func (c *client) Read(ctx context.Context,
 	firstLsn Lsn, maxSize uint64) ([]pb.LogRecord, Lsn, error) {
 	return c.read(ctx, firstLsn, maxSize)
 }
 
-// Truncate truncates the Log Service log at the specified Lsn with Lsn
-// itself included. This allows the Log Service to free up storage capacities
-// for future appends, all future reads must start after the specified Lsn
-// position.
 func (c *client) Truncate(ctx context.Context, lsn Lsn) error {
 	if c.readOnly() {
 		return ErrIncompatibleClient
@@ -167,15 +189,10 @@ func (c *client) Truncate(ctx context.Context, lsn Lsn) error {
 	return c.truncate(ctx, lsn)
 }
 
-// GetTruncatedLsn returns the largest Lsn value that has been specified for
-// truncation.
 func (c *client) GetTruncatedLsn(ctx context.Context) (Lsn, error) {
 	return c.getTruncatedLsn(ctx)
 }
 
-// GetTSOTimestamp requests a total of count unique timestamps from the TSO and
-// return the first assigned such timestamp, that is TSO timestamps
-// [returned value, returned value + count] will be owned by the caller.
 func (c *client) GetTSOTimestamp(ctx context.Context, count uint64) (uint64, error) {
 	return c.tsoRequest(ctx, count)
 }
@@ -206,8 +223,8 @@ func (c *client) request(ctx context.Context,
 		Method:  mt,
 		Timeout: int64(timeout),
 		LogRequest: pb.LogRequest{
-			ShardID: c.cfg.ShardID,
-			DNID:    c.cfg.ReplicaID,
+			ShardID: c.cfg.LogShardID,
+			DNID:    c.cfg.DNReplicaID,
 			Lsn:     lsn,
 			MaxSize: maxSize,
 		},
diff --git a/pkg/logservice/rsm.go b/pkg/logservice/rsm.go
index c4d865744..0588ddacb 100644
--- a/pkg/logservice/rsm.go
+++ b/pkg/logservice/rsm.go
@@ -29,7 +29,7 @@ var (
 
 const (
 	firstLogShardID uint64 = 1
-	headerSize             = 4
+	headerSize             = pb.HeaderSize
 )
 
 // used to indicate query types
diff --git a/pkg/pb/logservice/logservice.go b/pkg/pb/logservice/logservice.go
index 2ea6ccb22..24bc28275 100644
--- a/pkg/pb/logservice/logservice.go
+++ b/pkg/pb/logservice/logservice.go
@@ -22,9 +22,20 @@ import (
 
 const (
 	// NoLeader is the replica ID of the leader node.
-	NoLeader uint64 = 0
+	NoLeader   uint64 = 0
+	HeaderSize        = 4
 )
 
+// ResizePayload resizes the payload length to length bytes.
+func (m *LogRecord) ResizePayload(length int) {
+	m.Data = m.Data[HeaderSize+8+length:]
+}
+
+// Payload returns the payload byte slice.
+func (m *LogRecord) Payload() []byte {
+	return m.Data[HeaderSize+8:]
+}
+
 // NewRSMState creates a new HAKeeperRSMState instance.
 func NewRSMState() HAKeeperRSMState {
 	return HAKeeperRSMState{


Test patch:
diff --git a/pkg/logservice/client_test.go b/pkg/logservice/client_test.go
index 0b7fe237a..b8b1d897b 100644
--- a/pkg/logservice/client_test.go
+++ b/pkg/logservice/client_test.go
@@ -46,14 +46,14 @@ func runClientTest(t *testing.T,
 
 	scfg := ClientConfig{
 		ReadOnly:         readOnly,
-		ShardID:          1,
-		ReplicaID:        2,
+		LogShardID:       1,
+		DNReplicaID:      2,
 		ServiceAddresses: []string{testServiceAddress},
 	}
 
 	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
 	defer cancel()
-	c, err := CreateClient(ctx, scfg)
+	c, err := NewClient(ctx, scfg)
 	require.NoError(t, err)
 	defer func() {
 		assert.NoError(t, c.Close())
@@ -90,35 +90,35 @@ func TestClientGetTSOTimestamp(t *testing.T) {
 
 func TestClientAppend(t *testing.T) {
 	fn := func(t *testing.T, cfg ClientConfig, c Client) {
-		cmd := make([]byte, 16+headerSize+8)
-		cmd = getAppendCmd(cmd, cfg.ReplicaID)
-		rand.Read(cmd[headerSize+8:])
+		rec := c.GetLogRecord(16)
+		rand.Read(rec.Payload())
 		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
 		defer cancel()
-		lsn, err := c.Append(ctx, pb.LogRecord{Data: cmd})
+		lsn, err := c.Append(ctx, rec)
 		require.NoError(t, err)
 		assert.Equal(t, uint64(4), lsn)
 
-		lsn, err = c.Append(ctx, pb.LogRecord{Data: cmd})
+		lsn, err = c.Append(ctx, rec)
 		require.NoError(t, err)
 		assert.Equal(t, uint64(5), lsn)
 
-		cmd = getAppendCmd(cmd, cfg.ReplicaID+1)
+		cmd := make([]byte, 16+headerSize+8)
+		cmd = getAppendCmd(cmd, cfg.DNReplicaID+1)
 		_, err = c.Append(ctx, pb.LogRecord{Data: cmd})
 		assert.Equal(t, ErrNotLeaseHolder, err)
 	}
 	runClientTest(t, false, fn)
 }
 
+// FIXME: actually enforce allowed allocation
 func TestClientAppendAlloc(t *testing.T) {
 	fn := func(t *testing.T, cfg ClientConfig, c Client) {
-		cmd := make([]byte, 16+headerSize+8)
-		cmd = getAppendCmd(cmd, cfg.ReplicaID)
-		rand.Read(cmd[headerSize+8:])
+		rec := c.GetLogRecord(16)
+		rand.Read(rec.Payload())
 		ac := testing.AllocsPerRun(1000, func() {
 			ctx, cancel := context.WithTimeout(context.Background(), time.Second)
 			defer cancel()
-			_, err := c.Append(ctx, pb.LogRecord{Data: cmd})
+			_, err := c.Append(ctx, rec)
 			require.NoError(t, err)
 		})
 		plog.Infof("ac: %f", ac)
@@ -128,19 +128,17 @@ func TestClientAppendAlloc(t *testing.T) {
 
 func TestClientRead(t *testing.T) {
 	fn := func(t *testing.T, cfg ClientConfig, c Client) {
-		cmd := make([]byte, 16+headerSize+8)
-		cmd = getAppendCmd(cmd, cfg.ReplicaID)
-		rand.Read(cmd[headerSize+8:])
+		rec := c.GetLogRecord(16)
+		rand.Read(rec.Payload())
 		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
 		defer cancel()
-		lsn, err := c.Append(ctx, pb.LogRecord{Data: cmd})
+		lsn, err := c.Append(ctx, rec)
 		require.NoError(t, err)
 		assert.Equal(t, uint64(4), lsn)
 
-		cmd2 := make([]byte, 16+headerSize+8)
-		cmd2 = getAppendCmd(cmd2, cfg.ReplicaID)
-		rand.Read(cmd2[headerSize+8:])
-		lsn, err = c.Append(ctx, pb.LogRecord{Data: cmd2})
+		rec2 := c.GetLogRecord(16)
+		rand.Read(rec2.Payload())
+		lsn, err = c.Append(ctx, rec2)
 		require.NoError(t, err)
 		assert.Equal(t, uint64(5), lsn)
 
@@ -149,8 +147,8 @@ func TestClientRead(t *testing.T) {
 		require.NoError(t, err)
 		assert.Equal(t, uint64(4), lsn)
 		require.Equal(t, 2, len(recs))
-		assert.Equal(t, cmd, recs[0].Data)
-		assert.Equal(t, cmd2, recs[1].Data)
+		assert.Equal(t, rec.Data, recs[0].Data)
+		assert.Equal(t, rec2.Data, recs[1].Data)
 
 		_, _, err = c.Read(ctx, 6, math.MaxUint64)
 		assert.Equal(t, ErrOutOfRange, err)
@@ -160,12 +158,11 @@ func TestClientRead(t *testing.T) {
 
 func TestClientTruncate(t *testing.T) {
 	fn := func(t *testing.T, cfg ClientConfig, c Client) {
-		cmd := make([]byte, 16+headerSize+8)
-		cmd = getAppendCmd(cmd, cfg.ReplicaID)
-		rand.Read(cmd[headerSize+8:])
+		rec := c.GetLogRecord(16)
+		rand.Read(rec.Payload())
 		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
 		defer cancel()
-		lsn, err := c.Append(ctx, pb.LogRecord{Data: cmd})
+		lsn, err := c.Append(ctx, rec)
 		require.NoError(t, err)
 		assert.Equal(t, uint64(4), lsn)
 
@@ -181,12 +178,11 @@ func TestClientTruncate(t *testing.T) {
 
 func TestReadOnlyClientRejectWriteRequests(t *testing.T) {
 	fn := func(t *testing.T, cfg ClientConfig, c Client) {
-		cmd := make([]byte, 16+headerSize+8)
-		cmd = getAppendCmd(cmd, cfg.ReplicaID)
-		rand.Read(cmd[headerSize+8:])
+		rec := c.GetLogRecord(16)
+		rand.Read(rec.Payload())
 		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
 		defer cancel()
-		_, err := c.Append(ctx, pb.LogRecord{Data: cmd})
+		_, err := c.Append(ctx, rec)
 		require.Equal(t, ErrIncompatibleClient, err)
 		require.Equal(t, ErrIncompatibleClient, c.Truncate(ctx, 4))
 	}
diff --git a/pkg/logservice/testclient.go b/pkg/logservice/testclient.go
new file mode 100644
index 000000000..989f893cc
--- /dev/null
+++ b/pkg/logservice/testclient.go
@@ -0,0 +1,57 @@
+// Copyright 2021 - 2022 Matrix Origin
+//
+// Licensed 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 logservice
+
+import (
+	"github.com/lni/vfs"
+
+	pb "github.com/matrixorigin/matrixone/pkg/pb/logservice"
+)
+
+func NewTestService(fs vfs.FS) (*Service, ClientConfig, error) {
+	addr := "localhost:9000"
+	cfg := Config{
+		RTTMillisecond:       10,
+		GossipSeedAddresses:  []string{addr},
+		DeploymentID:         1,
+		FS:                   fs,
+		ServiceListenAddress: addr,
+		ServiceAddress:       addr,
+	}
+	cfg.Fill()
+	service, err := NewService(cfg)
+	if err != nil {
+		return nil, ClientConfig{}, err
+	}
+
+	cmd := pb.ScheduleCommand{
+		ConfigChange: &pb.ConfigChange{
+			ChangeType: pb.StartReplica,
+			Replica: pb.Replica{
+				ShardID:   1,
+				ReplicaID: 1,
+			},
+			InitialMembers: map[uint64]string{1: service.ID()},
+		},
+	}
+	service.handleCommands([]pb.ScheduleCommand{cmd})
+
+	ccfg := ClientConfig{
+		LogShardID:       1,
+		DNReplicaID:      10,
+		ServiceAddresses: []string{addr},
+	}
+	return service, ccfg, nil
+}
diff --git a/pkg/logservice/testclient_test.go b/pkg/logservice/testclient_test.go
new file mode 100644
index 000000000..05a097610
--- /dev/null
+++ b/pkg/logservice/testclient_test.go
@@ -0,0 +1,68 @@
+// Copyright 2021 - 2022 Matrix Origin
+//
+// Licensed 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 logservice
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/lni/vfs"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestTestClient(t *testing.T) {
+	fs := vfs.NewStrictMem()
+	service, ccfg, err := NewTestService(fs)
+	require.NoError(t, err)
+	defer service.Close()
+
+	// you need to decision what timeout value to use
+	// you also need to retry requests on timeout
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+
+	// get two clients, you can get many clients to be used concurrently
+	// but each client itself is not goroutine safe
+	client1, err := NewClient(ctx, ccfg)
+	require.NoError(t, err)
+	defer client1.Close()
+	client2, err := NewClient(ctx, ccfg)
+	require.NoError(t, err)
+	defer client2.Close()
+
+	// Don't use the Data field unless you know what you are doing
+	rec1 := client1.GetLogRecord(5)
+	rec2 := client2.GetLogRecord(5)
+	copy(rec1.Payload(), []byte("hello"))
+	copy(rec2.Payload(), []byte("world"))
+
+	// append the records
+	lsn1, err := client1.Append(ctx, rec1)
+	require.NoError(t, err)
+	lsn2, err := client2.Append(ctx, rec2)
+	require.NoError(t, err)
+	assert.Equal(t, lsn1+1, lsn2)
+
+	// read them back and check whether the returned data is expected
+	recs, lsn, err := client1.Read(ctx, lsn1, 1024)
+	require.NoError(t, err)
+	assert.Equal(t, lsn1, lsn)
+	assert.Equal(t, 2, len(recs))
+	assert.Equal(t, rec1.Payload(), recs[0].Payload())
+	assert.Equal(t, rec2.Payload(), recs[1].Payload())
+}