more ut coverage

Author: gouhongshen

pkg/fileservice/get_test.go

@@ -19,6 +19,7 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	"iter"
 )
 
 func TestGetForBackup(t *testing.T) {
@@ -30,3 +31,56 @@ func TestGetForBackup(t *testing.T) {
 	assert.True(t, ok)
 	assert.Equal(t, dir, localFS.rootPath)
 }
+
+func TestGetForBackupS3Opts(t *testing.T) {
+	ctx := context.Background()
+	dir := t.TempDir()
+	spec := JoinPath("s3-opts,endpoint=disk,bucket="+dir+",prefix=backup-prefix,name=backup", "object")
+	fs, err := GetForBackup(ctx, spec)
+	assert.Nil(t, err)
+	s3fs, ok := fs.(*S3FS)
+	assert.True(t, ok)
+	assert.Equal(t, "backup", s3fs.name)
+	assert.Equal(t, "backup-prefix", s3fs.keyPrefix)
+}
+
+type dummyFileService struct{ name string }
+
+func (d dummyFileService) Delete(ctx context.Context, filePaths ...string) error { return nil }
+func (d dummyFileService) Name() string                                          { return d.name }
+func (d dummyFileService) Read(ctx context.Context, vector *IOVector) error      { return nil }
+func (d dummyFileService) ReadCache(ctx context.Context, vector *IOVector) error { return nil }
+func (d dummyFileService) Write(ctx context.Context, vector IOVector) error      { return nil }
+func (d dummyFileService) List(ctx context.Context, dirPath string) iter.Seq2[*DirEntry, error] {
+	return func(yield func(*DirEntry, error) bool) {
+		yield(&DirEntry{Name: "a"}, nil)
+	}
+}
+func (d dummyFileService) StatFile(ctx context.Context, filePath string) (*DirEntry, error) {
+	return &DirEntry{Name: filePath}, nil
+}
+func (d dummyFileService) PrefetchFile(ctx context.Context, filePath string) error { return nil }
+func (d dummyFileService) Cost() *CostAttr                                         { return nil }
+func (d dummyFileService) Close(ctx context.Context)                               {}
+
+func TestGetFromMappings(t *testing.T) {
+	fs1 := dummyFileService{name: "first"}
+	fs2 := dummyFileService{name: "second"}
+	mapping, err := NewFileServices("first", fs1, fs2)
+	assert.NoError(t, err)
+
+	var res FileService
+	res, err = Get[FileService](mapping, "second")
+	assert.NoError(t, err)
+	assert.Equal(t, "second", res.Name())
+
+	_, err = Get[FileService](mapping, "missing")
+	assert.Error(t, err)
+
+	res, err = Get[FileService](fs1, "first")
+	assert.NoError(t, err)
+	assert.Equal(t, "first", res.Name())
+
+	_, err = Get[FileService](fs1, "other")
+	assert.Error(t, err)
+}

pkg/fileservice/object_storage_arguments_test.go

@@ -185,6 +185,61 @@ func TestAWSRegion(t *testing.T) {
 	assert.NotNil(t, args.validate())
 }
 
+func TestSetFromStringParallelMode(t *testing.T) {
+	var args ObjectStorageArguments
+	assert.NoError(t, args.SetFromString([]string{"parallel-mode=force"}))
+	assert.Equal(t, ParallelForce, args.ParallelMode)
+
+	args = ObjectStorageArguments{
+		ParallelMode: ParallelAuto,
+	}
+	assert.NoError(t, args.SetFromString([]string{"parallel-mode=unknown"}))
+	assert.Equal(t, ParallelAuto, args.ParallelMode)
+}
+
+func TestObjectStorageArgumentsValidateDefaults(t *testing.T) {
+	args := ObjectStorageArguments{
+		Endpoint: "example.com",
+	}
+	assert.NoError(t, args.validate())
+	assert.Equal(t, "https://example.com", args.Endpoint)
+	assert.Equal(t, "mo-service", args.RoleSessionName)
+}
+
+func TestObjectStorageArgumentsShouldLoadDefaultCredentials(t *testing.T) {
+	t.Setenv("AWS_ACCESS_KEY_ID", "ak")
+	t.Setenv("AWS_SECRET_ACCESS_KEY", "sk")
+	args := ObjectStorageArguments{}
+	assert.True(t, args.shouldLoadDefaultCredentials())
+
+	args = ObjectStorageArguments{
+		NoDefaultCredentials: true,
+		KeyID:                "id",
+		KeySecret:            "secret",
+	}
+	assert.False(t, args.shouldLoadDefaultCredentials())
+
+	args = ObjectStorageArguments{
+		NoDefaultCredentials: true,
+		RoleARN:              "arn",
+	}
+	assert.True(t, args.shouldLoadDefaultCredentials())
+}
+
+func TestObjectStorageArgumentsString(t *testing.T) {
+	args := ObjectStorageArguments{
+		Name:        "foo",
+		KeyPrefix:   "bar",
+		Concurrency: 3,
+	}
+	s := args.String()
+	var decoded ObjectStorageArguments
+	assert.NoError(t, json.Unmarshal([]byte(s), &decoded))
+	assert.Equal(t, args.Name, decoded.Name)
+	assert.Equal(t, args.KeyPrefix, decoded.KeyPrefix)
+	assert.Equal(t, args.Concurrency, decoded.Concurrency)
+}
+
 func TestParseHDFSArgs(t *testing.T) {
 	var args ObjectStorageArguments
 	if err := args.SetFromString([]string{

pkg/fileservice/object_storage_http_trace_test.go

@@ -0,0 +1,78 @@
+// Copyright 2025 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 fileservice
+
+import (
+	"context"
+	"net/http/httptrace"
+	"strings"
+	"testing"
+
+	"github.com/matrixorigin/matrixone/pkg/common/moerr"
+	"github.com/stretchr/testify/require"
+)
+
+func TestObjectStorageHTTPTraceWriteMultipartParallel(t *testing.T) {
+	upstream := &mockParallelObjectStorage{supports: true}
+	wrapped := newObjectStorageHTTPTrace(upstream)
+
+	err := wrapped.WriteMultipartParallel(context.Background(), "key", strings.NewReader("data"), nil, nil)
+	require.NoError(t, err)
+	require.NotNil(t, upstream.ctx)
+	require.NotNil(t, httptrace.ContextClientTrace(upstream.ctx))
+	require.Equal(t, "key", upstream.key)
+}
+
+func TestObjectStorageHTTPTraceWriteMultipartParallelUnsupported(t *testing.T) {
+	wrapped := newObjectStorageHTTPTrace(dummyObjectStorage{})
+
+	err := wrapped.WriteMultipartParallel(context.Background(), "key", strings.NewReader("data"), nil, nil)
+	require.Error(t, err)
+	require.True(t, moerr.IsMoErrCode(err, moerr.ErrNotSupported))
+}
+
+func TestObjectStorageHTTPTraceDelegates(t *testing.T) {
+	upstream := &recordingObjectStorage{}
+	wrapped := newObjectStorageHTTPTrace(upstream)
+	ctx := context.Background()
+
+	require.NoError(t, wrapped.Delete(ctx, "a"))
+	exists, err := wrapped.Exists(ctx, "b")
+	require.NoError(t, err)
+	require.True(t, exists)
+	iterSeq := wrapped.List(ctx, "c")
+	var listed []string
+	iterSeq(func(entry *DirEntry, err error) bool {
+		require.NoError(t, err)
+		listed = append(listed, entry.Name)
+		return true
+	})
+	reader, err := wrapped.Read(ctx, "d", nil, nil)
+	require.NoError(t, err)
+	defer reader.Close()
+	buf := make([]byte, 4)
+	_, _ = reader.Read(buf)
+	size, err := wrapped.Stat(ctx, "e")
+	require.NoError(t, err)
+	require.Equal(t, int64(3), size)
+	require.NoError(t, wrapped.Write(ctx, "f", strings.NewReader("payload"), nil, nil))
+
+	require.Len(t, upstream.calls, 6)
+	for _, ctx := range upstream.ctxs {
+		require.NotNil(t, httptrace.ContextClientTrace(ctx))
+	}
+	require.ElementsMatch(t, []string{"delete", "exists", "list", "read", "stat", "write"}, upstream.calls)
+	require.Equal(t, []string{"one"}, listed)
+}

pkg/fileservice/object_storage_metrics_test.go

@@ -0,0 +1,93 @@
+// Copyright 2025 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 fileservice
+
+import (
+	"context"
+	"strings"
+	"testing"
+
+	"github.com/matrixorigin/matrixone/pkg/common/moerr"
+	metric "github.com/matrixorigin/matrixone/pkg/util/metric/v2"
+	"github.com/prometheus/client_golang/prometheus/testutil"
+	"github.com/stretchr/testify/require"
+)
+
+func TestObjectStorageMetricsWriteMultipartParallel(t *testing.T) {
+	name := t.Name()
+	upstream := &mockParallelObjectStorage{supports: true}
+	wrapped := newObjectStorageMetrics(upstream, name)
+	gauge := metric.FSObjectStorageOperations.WithLabelValues(name, "write")
+	before := testutil.ToFloat64(gauge)
+
+	err := wrapped.WriteMultipartParallel(context.Background(), "key", strings.NewReader("data"), nil, nil)
+	require.NoError(t, err)
+	require.Equal(t, before+1, testutil.ToFloat64(gauge))
+	require.Equal(t, "key", upstream.key)
+}
+
+func TestObjectStorageMetricsWriteMultipartParallelNotSupported(t *testing.T) {
+	name := t.Name()
+	wrapped := newObjectStorageMetrics(dummyObjectStorage{}, name)
+	gauge := metric.FSObjectStorageOperations.WithLabelValues(name, "write")
+	before := testutil.ToFloat64(gauge)
+
+	err := wrapped.WriteMultipartParallel(context.Background(), "key", strings.NewReader("data"), nil, nil)
+	require.Error(t, err)
+	require.True(t, moerr.IsMoErrCode(err, moerr.ErrNotSupported))
+	require.Equal(t, before+1, testutil.ToFloat64(gauge))
+}
+
+func TestObjectStorageMetricsDelegates(t *testing.T) {
+	name := t.Name()
+	upstream := &recordingObjectStorage{}
+	wrapped := newObjectStorageMetrics(upstream, name)
+
+	require.NoError(t, wrapped.Delete(context.Background(), "a"))
+	_, _ = wrapped.Exists(context.Background(), "b")
+	seq := wrapped.List(context.Background(), "c")
+	seq(func(_ *DirEntry, _ error) bool { return true })
+	rc, err := wrapped.Read(context.Background(), "d", nil, nil)
+	require.NoError(t, err)
+	require.NoError(t, rc.Close())
+	_, _ = wrapped.Stat(context.Background(), "e")
+	require.NoError(t, wrapped.Write(context.Background(), "f", strings.NewReader("x"), nil, nil))
+
+	require.ElementsMatch(t, []string{
+		"delete", "exists", "list", "read", "stat", "write",
+	}, upstream.calls)
+
+	// gauges incremented
+	require.True(t, testutil.ToFloat64(metric.FSObjectStorageOperations.WithLabelValues(name, "delete")) >= 1)
+	require.True(t, testutil.ToFloat64(metric.FSObjectStorageOperations.WithLabelValues(name, "exists")) >= 1)
+	require.True(t, testutil.ToFloat64(metric.FSObjectStorageOperations.WithLabelValues(name, "list")) >= 1)
+	require.True(t, testutil.ToFloat64(metric.FSObjectStorageOperations.WithLabelValues(name, "read")) >= 1)
+	require.True(t, testutil.ToFloat64(metric.FSObjectStorageOperations.WithLabelValues(name, "stat")) >= 1)
+	require.True(t, testutil.ToFloat64(metric.FSObjectStorageOperations.WithLabelValues(name, "write")) >= 1)
+}
+
+func TestObjectStorageMetricsReadCloseDecrementsActive(t *testing.T) {
+	name := t.Name()
+	upstream := &recordingObjectStorage{}
+	wrapped := newObjectStorageMetrics(upstream, name)
+	active := metric.FSObjectStorageOperations.WithLabelValues(name, "active-read")
+	before := testutil.ToFloat64(active)
+
+	r, err := wrapped.Read(context.Background(), "key", nil, nil)
+	require.NoError(t, err)
+	require.Equal(t, before+1, testutil.ToFloat64(active))
+	require.NoError(t, r.Close())
+	require.Equal(t, before, testutil.ToFloat64(active))
+}

pkg/fileservice/object_storage_semaphore_test.go

@@ -0,0 +1,87 @@
+// Copyright 2025 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 fileservice
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestObjectStorageSemaphoreSerializes(t *testing.T) {
+	start := make(chan struct{}, 2)
+	wait := make(chan struct{})
+	upstream := &blockingObjectStorage{
+		start: start,
+		wait:  wait,
+	}
+	sem := newObjectStorageSemaphore(upstream, 1)
+
+	done := make(chan struct{})
+	go func() {
+		require.NoError(t, sem.Write(context.Background(), "a", nil, nil, nil))
+		close(done)
+	}()
+
+	select {
+	case <-start:
+	case <-time.After(time.Second):
+		t.Fatal("first write did not start")
+	}
+
+	startSecond := make(chan struct{})
+	go func() {
+		defer close(startSecond)
+		require.NoError(t, sem.Write(context.Background(), "b", nil, nil, nil))
+	}()
+
+	select {
+	case <-startSecond:
+		t.Fatal("second write started before release")
+	case <-time.After(50 * time.Millisecond):
+	}
+
+	close(wait) // release first
+	select {
+	case <-startSecond:
+	case <-time.After(time.Second):
+		t.Fatal("second write not started after release")
+	}
+	<-done
+}
+
+func TestObjectStorageSemaphoreReleasesOnError(t *testing.T) {
+	start := make(chan struct{}, 1)
+	wait := make(chan struct{})
+	upstream := &blockingObjectStorage{
+		start: start,
+		wait:  wait,
+		err:   context.DeadlineExceeded,
+	}
+	sem := newObjectStorageSemaphore(upstream, 1)
+
+	err := sem.Write(context.Background(), "a", nil, nil, nil)
+	require.Error(t, err)
+	close(wait)
+
+	// another call should proceed after the failed one
+	select {
+	case start <- struct{}{}:
+	default:
+	}
+	require.NoError(t, sem.Delete(context.Background(), "x"))
+}