chore: add more tests (#3592)

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>

.codecov.yml

@@ -4,3 +4,6 @@ comment:
   require_changes: true
 ignore:
   - "tools"
+  - "**/mock"
+  - "**/*_mock.go"
+  - "**/*test"

.github/FUNDING.yml

@@ -10,4 +10,4 @@ liberapay: # Replace with a single Liberapay username
 issuehunt: # Replace with a single IssueHunt username
 otechie: # Replace with a single Otechie username
 custom: # https://gitee.com/kevwan/static/raw/master/images/sponsor.jpg
-ethereum: 0x5052b7f6B937B02563996D23feb69b38D06Ca150 | kevwan
+ethereum: # 0x5052b7f6B937B02563996D23feb69b38D06Ca150 | kevwan

.github/workflows/go.yml

@@ -17,7 +17,7 @@ jobs:
       - name: Set up Go 1.x
         uses: actions/setup-go@v3
         with:
-          go-version: ^1.18
+          go-version: 1.18
           check-latest: true
           cache: true
         id: go
@@ -29,8 +29,12 @@ jobs:
       - name: Lint
         run: |
           go vet -stdmethods=false $(go list ./...)
-          go install mvdan.cc/gofumpt@latest
+
-          test -z "$(gofumpt -l -extra .)" || echo "Please run 'gofumpt -l -w -extra .'"
+          go mod tidy
+          if ! test -z "$(git status --porcelain)"; then
+            echo "Please run 'go mod tidy'"
+            exit 1
+          fi
 
       - name: Test
         run: go test -race -coverprofile=coverage.txt -covermode=atomic ./...
@@ -57,5 +61,5 @@ jobs:
         run: |
           go mod verify
           go mod download
-          go test -v -race ./...
+          go test ./...
           cd tools/goctl && go build -v goctl.go

.github/workflows/release.yaml

@@ -22,7 +22,7 @@ jobs:
           github_token: ${{ secrets.GITHUB_TOKEN }}
           goos: ${{ matrix.goos }}
           goarch: ${{ matrix.goarch }}
-          goversion: "https://dl.google.com/go/go1.17.5.linux-amd64.tar.gz"
+          goversion: "https://dl.google.com/go/go1.18.10.linux-amd64.tar.gz"
           project_path: "tools/goctl"
           binary_name: "goctl"
           extra_files: tools/goctl/readme.md tools/goctl/readme-cn.md

.gitignore

@@ -12,11 +12,13 @@
 
 # ignore
 **/.idea
+**/.vscode
 **/.DS_Store
 **/logs
+**/adhoc
+**/coverage.txt
 
 # for test purpose
-**/adhoc
 go.work
 go.work.sum
 

core/bloom/bloom.go

@@ -1,6 +1,7 @@
 package bloom
 
 import (
+	"context"
 	"errors"
 	"strconv"
 
@@ -8,28 +9,29 @@ import (
 	"github.com/zeromicro/go-zero/core/stores/redis"
 )
 
-const (
+// for detailed error rate table, see http://pages.cs.wisc.edu/~cao/papers/summary-cache/node8.html
-	// for detailed error rate table, see http://pages.cs.wisc.edu/~cao/papers/summary-cache/node8.html
+// maps as k in the error rate table
-	// maps as k in the error rate table
+const maps = 14
-	maps      = 14
+
-	setScript = `
+var (
+	// ErrTooLargeOffset indicates the offset is too large in bitset.
+	ErrTooLargeOffset = errors.New("too large offset")
+
+	setScript = redis.NewScript(`
 for _, offset in ipairs(ARGV) do
 	redis.call("setbit", KEYS[1], offset, 1)
 end
-`
+`)
-	testScript = `
+	testScript = redis.NewScript(`
 for _, offset in ipairs(ARGV) do
 	if tonumber(redis.call("getbit", KEYS[1], offset)) == 0 then
 		return false
 	end
 end
 return true
-`
+`)
 )
 
-// ErrTooLargeOffset indicates the offset is too large in bitset.
-var ErrTooLargeOffset = errors.New("too large offset")
-
 type (
 	// A Filter is a bloom filter.
 	Filter struct {
@@ -38,8 +40,8 @@ type (
 	}
 
 	bitSetProvider interface {
-		check([]uint) (bool, error)
+		check(ctx context.Context, offsets []uint) (bool, error)
-		set([]uint) error
+		set(ctx context.Context, offsets []uint) error
 	}
 )
 
@@ -58,14 +60,24 @@ func New(store *redis.Redis, key string, bits uint) *Filter {
 
 // Add adds data into f.
 func (f *Filter) Add(data []byte) error {
+	return f.AddCtx(context.Background(), data)
+}
+
+// AddCtx adds data into f with context.
+func (f *Filter) AddCtx(ctx context.Context, data []byte) error {
 	locations := f.getLocations(data)
-	return f.bitSet.set(locations)
+	return f.bitSet.set(ctx, locations)
 }
 
 // Exists checks if data is in f.
 func (f *Filter) Exists(data []byte) (bool, error) {
+	return f.ExistsCtx(context.Background(), data)
+}
+
+// ExistsCtx checks if data is in f with context.
+func (f *Filter) ExistsCtx(ctx context.Context, data []byte) (bool, error) {
 	locations := f.getLocations(data)
-	isSet, err := f.bitSet.check(locations)
+	isSet, err := f.bitSet.check(ctx, locations)
 	if err != nil {
 		return false, err
 	}
@@ -111,13 +123,13 @@ func (r *redisBitSet) buildOffsetArgs(offsets []uint) ([]string, error) {
 	return args, nil
 }
 
-func (r *redisBitSet) check(offsets []uint) (bool, error) {
+func (r *redisBitSet) check(ctx context.Context, offsets []uint) (bool, error) {
 	args, err := r.buildOffsetArgs(offsets)
 	if err != nil {
 		return false, err
 	}
 
-	resp, err := r.store.Eval(testScript, []string{r.key}, args)
+	resp, err := r.store.ScriptRunCtx(ctx, testScript, []string{r.key}, args)
 	if err == redis.Nil {
 		return false, nil
 	} else if err != nil {
@@ -132,22 +144,24 @@ func (r *redisBitSet) check(offsets []uint) (bool, error) {
 	return exists == 1, nil
 }
 
+// del only use for testing.
 func (r *redisBitSet) del() error {
 	_, err := r.store.Del(r.key)
 	return err
 }
 
+// expire only use for testing.
 func (r *redisBitSet) expire(seconds int) error {
 	return r.store.Expire(r.key, seconds)
 }
 
-func (r *redisBitSet) set(offsets []uint) error {
+func (r *redisBitSet) set(ctx context.Context, offsets []uint) error {
 	args, err := r.buildOffsetArgs(offsets)
 	if err != nil {
 		return err
 	}
 
-	_, err = r.store.Eval(setScript, []string{r.key}, args)
+	_, err = r.store.ScriptRunCtx(ctx, setScript, []string{r.key}, args)
 	if err == redis.Nil {
 		return nil
 	}

core/bloom/bloom_test.go

@@ -1,30 +1,31 @@
 package bloom
 
 import (
+	"context"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/zeromicro/go-zero/core/logx"
 	"github.com/zeromicro/go-zero/core/stores/redis/redistest"
 )
 
 func TestRedisBitSet_New_Set_Test(t *testing.T) {
-	store, clean, err := redistest.CreateRedis()
+	store := redistest.CreateRedis(t)
-	assert.Nil(t, err)
+	ctx := context.Background()
-	defer clean()
 
 	bitSet := newRedisBitSet(store, "test_key", 1024)
-	isSetBefore, err := bitSet.check([]uint{0})
+	isSetBefore, err := bitSet.check(ctx, []uint{0})
 	if err != nil {
 		t.Fatal(err)
 	}
 	if isSetBefore {
 		t.Fatal("Bit should not be set")
 	}
-	err = bitSet.set([]uint{512})
+	err = bitSet.set(ctx, []uint{512})
 	if err != nil {
 		t.Fatal(err)
 	}
-	isSetAfter, err := bitSet.check([]uint{512})
+	isSetAfter, err := bitSet.check(ctx, []uint{512})
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -42,9 +43,7 @@ func TestRedisBitSet_New_Set_Test(t *testing.T) {
 }
 
 func TestRedisBitSet_Add(t *testing.T) {
-	store, clean, err := redistest.CreateRedis()
+	store := redistest.CreateRedis(t)
-	assert.Nil(t, err)
-	defer clean()
 
 	filter := New(store, "test_key", 64)
 	assert.Nil(t, filter.Add([]byte("hello")))
@@ -53,3 +52,51 @@ func TestRedisBitSet_Add(t *testing.T) {
 	assert.Nil(t, err)
 	assert.True(t, ok)
 }
+
+func TestFilter_Exists(t *testing.T) {
+	store, clean := redistest.CreateRedisWithClean(t)
+
+	rbs := New(store, "test", 64)
+	_, err := rbs.Exists([]byte{0, 1, 2})
+	assert.NoError(t, err)
+
+	clean()
+	rbs = New(store, "test", 64)
+	_, err = rbs.Exists([]byte{0, 1, 2})
+	assert.Error(t, err)
+}
+
+func TestRedisBitSet_check(t *testing.T) {
+	store, clean := redistest.CreateRedisWithClean(t)
+	ctx := context.Background()
+
+	rbs := newRedisBitSet(store, "test", 0)
+	assert.Error(t, rbs.set(ctx, []uint{0, 1, 2}))
+	_, err := rbs.check(ctx, []uint{0, 1, 2})
+	assert.Error(t, err)
+
+	rbs = newRedisBitSet(store, "test", 64)
+	_, err = rbs.check(ctx, []uint{0, 1, 2})
+	assert.NoError(t, err)
+
+	clean()
+	rbs = newRedisBitSet(store, "test", 64)
+	_, err = rbs.check(ctx, []uint{0, 1, 2})
+	assert.Error(t, err)
+}
+
+func TestRedisBitSet_set(t *testing.T) {
+	logx.Disable()
+	store, clean := redistest.CreateRedisWithClean(t)
+	ctx := context.Background()
+
+	rbs := newRedisBitSet(store, "test", 0)
+	assert.Error(t, rbs.set(ctx, []uint{0, 1, 2}))
+
+	rbs = newRedisBitSet(store, "test", 64)
+	assert.NoError(t, rbs.set(ctx, []uint{0, 1, 2}))
+
+	clean()
+	rbs = newRedisBitSet(store, "test", 64)
+	assert.Error(t, rbs.set(ctx, []uint{0, 1, 2}))
+}

core/codec/gzip_test.go

@@ -2,6 +2,8 @@ package codec
 
 import (
 	"bytes"
+	"compress/gzip"
+	"errors"
 	"fmt"
 	"testing"
 
@@ -21,3 +23,45 @@ func TestGzip(t *testing.T) {
 	assert.True(t, len(bs) < buf.Len())
 	assert.Equal(t, buf.Bytes(), actual)
 }
+
+func TestGunzip(t *testing.T) {
+	tests := []struct {
+		name        string
+		input       []byte
+		expected    []byte
+		expectedErr error
+	}{
+		{
+			name: "valid input",
+			input: func() []byte {
+				var buf bytes.Buffer
+				gz := gzip.NewWriter(&buf)
+				gz.Write([]byte("hello"))
+				gz.Close()
+				return buf.Bytes()
+			}(),
+			expected:    []byte("hello"),
+			expectedErr: nil,
+		},
+		{
+			name:        "invalid input",
+			input:       []byte("invalid input"),
+			expected:    nil,
+			expectedErr: gzip.ErrHeader,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			result, err := Gunzip(test.input)
+
+			if !bytes.Equal(result, test.expected) {
+				t.Errorf("unexpected result: %v", result)
+			}
+
+			if !errors.Is(err, test.expectedErr) {
+				t.Errorf("unexpected error: %v", err)
+			}
+		})
+	}
+}

core/codec/rsa_test.go

@@ -2,6 +2,7 @@ package codec
 
 import (
 	"encoding/base64"
+	"os"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -41,6 +42,7 @@ func TestCryption(t *testing.T) {
 
 	file, err := fs.TempFilenameWithText(priKey)
 	assert.Nil(t, err)
+	defer os.Remove(file)
 	dec, err := NewRsaDecrypter(file)
 	assert.Nil(t, err)
 	actual, err := dec.Decrypt(ret)

core/collection/safemap.go

@@ -29,6 +29,8 @@ func NewSafeMap() *SafeMap {
 // Del deletes the value with the given key from m.
 func (m *SafeMap) Del(key any) {
 	m.lock.Lock()
+	defer m.lock.Unlock()
+
 	if _, ok := m.dirtyOld[key]; ok {
 		delete(m.dirtyOld, key)
 		m.deletionOld++
@@ -52,7 +54,6 @@ func (m *SafeMap) Del(key any) {
 		m.dirtyNew = make(map[any]any)
 		m.deletionNew = 0
 	}
-	m.lock.Unlock()
 }
 
 // Get gets the value with the given key from m.
@@ -89,6 +90,8 @@ func (m *SafeMap) Range(f func(key, val any) bool) {
 // Set sets the value into m with the given key.
 func (m *SafeMap) Set(key, value any) {
 	m.lock.Lock()
+	defer m.lock.Unlock()
+
 	if m.deletionOld <= maxDeletion {
 		if _, ok := m.dirtyNew[key]; ok {
 			delete(m.dirtyNew, key)
@@ -102,7 +105,6 @@ func (m *SafeMap) Set(key, value any) {
 		}
 		m.dirtyNew[key] = value
 	}
-	m.lock.Unlock()
 }
 
 // Size returns the size of m.

core/collection/safemap_test.go

@@ -147,3 +147,65 @@ func TestSafeMap_Range(t *testing.T) {
 	assert.Equal(t, m.dirtyNew, newMap.dirtyNew)
 	assert.Equal(t, m.dirtyOld, newMap.dirtyOld)
 }
+
+func TestSetManyTimes(t *testing.T) {
+	const iteration = maxDeletion * 2
+	m := NewSafeMap()
+	for i := 0; i < iteration; i++ {
+		m.Set(i, i)
+		if i%3 == 0 {
+			m.Del(i / 2)
+		}
+	}
+	var count int
+	m.Range(func(k, v any) bool {
+		count++
+		return count < maxDeletion/2
+	})
+	assert.Equal(t, maxDeletion/2, count)
+	for i := 0; i < iteration; i++ {
+		m.Set(i, i)
+		if i%3 == 0 {
+			m.Del(i / 2)
+		}
+	}
+	for i := 0; i < iteration; i++ {
+		m.Set(i, i)
+		if i%3 == 0 {
+			m.Del(i / 2)
+		}
+	}
+	for i := 0; i < iteration; i++ {
+		m.Set(i, i)
+		if i%3 == 0 {
+			m.Del(i / 2)
+		}
+	}
+
+	count = 0
+	m.Range(func(k, v any) bool {
+		count++
+		return count < maxDeletion
+	})
+	assert.Equal(t, maxDeletion, count)
+}
+
+func TestSetManyTimesNew(t *testing.T) {
+	m := NewSafeMap()
+	for i := 0; i < maxDeletion*3; i++ {
+		m.Set(i, i)
+	}
+	for i := 0; i < maxDeletion*2; i++ {
+		m.Del(i)
+	}
+	for i := 0; i < maxDeletion*3; i++ {
+		m.Set(i+maxDeletion*3, i+maxDeletion*3)
+	}
+	for i := 0; i < maxDeletion*2; i++ {
+		m.Del(i + maxDeletion*2)
+	}
+	for i := 0; i < maxDeletion-copyThreshold+1; i++ {
+		m.Del(i + maxDeletion*2)
+	}
+	assert.Equal(t, 0, len(m.dirtyNew))
+}

core/conf/config.go

@@ -13,7 +13,10 @@ import (
 	"github.com/zeromicro/go-zero/internal/encoding"
 )
 
-const jsonTagKey = "json"
+const (
+	jsonTagKey = "json"
+	jsonTagSep = ','
+)
 
 var (
 	fillDefaultUnmarshaler = mapping.NewUnmarshaler(jsonTagKey, mapping.WithDefault())
@@ -70,13 +73,13 @@ func LoadConfig(file string, v any, opts ...Option) error {
 
 // LoadFromJsonBytes loads config into v from content json bytes.
 func LoadFromJsonBytes(content []byte, v any) error {
-	info, err := buildFieldsInfo(reflect.TypeOf(v))
+	info, err := buildFieldsInfo(reflect.TypeOf(v), "")
 	if err != nil {
 		return err
 	}
 
 	var m map[string]any
-	if err := jsonx.Unmarshal(content, &m); err != nil {
+	if err = jsonx.Unmarshal(content, &m); err != nil {
 		return err
 	}
 
@@ -124,13 +127,13 @@ func MustLoad(path string, v any, opts ...Option) {
 	}
 }
 
-func addOrMergeFields(info *fieldInfo, key string, child *fieldInfo) error {
+func addOrMergeFields(info *fieldInfo, key string, child *fieldInfo, fullName string) error {
 	if prev, ok := info.children[key]; ok {
 		if child.mapField != nil {
-			return newDupKeyError(key)
+			return newConflictKeyError(fullName)
 		}
 
-		if err := mergeFields(prev, key, child.children); err != nil {
+		if err := mergeFields(prev, key, child.children, fullName); err != nil {
 			return err
 		}
 	} else {
@@ -140,27 +143,27 @@ func addOrMergeFields(info *fieldInfo, key string, child *fieldInfo) error {
 	return nil
 }
 
-func buildAnonymousFieldInfo(info *fieldInfo, lowerCaseName string, ft reflect.Type) error {
+func buildAnonymousFieldInfo(info *fieldInfo, lowerCaseName string, ft reflect.Type, fullName string) error {
 	switch ft.Kind() {
 	case reflect.Struct:
-		fields, err := buildFieldsInfo(ft)
+		fields, err := buildFieldsInfo(ft, fullName)
 		if err != nil {
 			return err
 		}
 
 		for k, v := range fields.children {
-			if err = addOrMergeFields(info, k, v); err != nil {
+			if err = addOrMergeFields(info, k, v, fullName); err != nil {
 				return err
 			}
 		}
 	case reflect.Map:
-		elemField, err := buildFieldsInfo(mapping.Deref(ft.Elem()))
+		elemField, err := buildFieldsInfo(mapping.Deref(ft.Elem()), fullName)
 		if err != nil {
 			return err
 		}
 
 		if _, ok := info.children[lowerCaseName]; ok {
-			return newDupKeyError(lowerCaseName)
+			return newConflictKeyError(fullName)
 		}
 
 		info.children[lowerCaseName] = &fieldInfo{
@@ -169,7 +172,7 @@ func buildAnonymousFieldInfo(info *fieldInfo, lowerCaseName string, ft reflect.T
 		}
 	default:
 		if _, ok := info.children[lowerCaseName]; ok {
-			return newDupKeyError(lowerCaseName)
+			return newConflictKeyError(fullName)
 		}
 
 		info.children[lowerCaseName] = &fieldInfo{
@@ -180,14 +183,14 @@ func buildAnonymousFieldInfo(info *fieldInfo, lowerCaseName string, ft reflect.T
 	return nil
 }
 
-func buildFieldsInfo(tp reflect.Type) (*fieldInfo, error) {
+func buildFieldsInfo(tp reflect.Type, fullName string) (*fieldInfo, error) {
 	tp = mapping.Deref(tp)
 
 	switch tp.Kind() {
 	case reflect.Struct:
-		return buildStructFieldsInfo(tp)
+		return buildStructFieldsInfo(tp, fullName)
 	case reflect.Array, reflect.Slice:
-		return buildFieldsInfo(mapping.Deref(tp.Elem()))
+		return buildFieldsInfo(mapping.Deref(tp.Elem()), fullName)
 	case reflect.Chan, reflect.Func:
 		return nil, fmt.Errorf("unsupported type: %s", tp.Kind())
 	default:
@@ -197,23 +200,23 @@ func buildFieldsInfo(tp reflect.Type) (*fieldInfo, error) {
 	}
 }
 
-func buildNamedFieldInfo(info *fieldInfo, lowerCaseName string, ft reflect.Type) error {
+func buildNamedFieldInfo(info *fieldInfo, lowerCaseName string, ft reflect.Type, fullName string) error {
 	var finfo *fieldInfo
 	var err error
 
 	switch ft.Kind() {
 	case reflect.Struct:
-		finfo, err = buildFieldsInfo(ft)
+		finfo, err = buildFieldsInfo(ft, fullName)
 		if err != nil {
 			return err
 		}
 	case reflect.Array, reflect.Slice:
-		finfo, err = buildFieldsInfo(ft.Elem())
+		finfo, err = buildFieldsInfo(ft.Elem(), fullName)
 		if err != nil {
 			return err
 		}
 	case reflect.Map:
-		elemInfo, err := buildFieldsInfo(mapping.Deref(ft.Elem()))
+		elemInfo, err := buildFieldsInfo(mapping.Deref(ft.Elem()), fullName)
 		if err != nil {
 			return err
 		}
@@ -223,31 +226,37 @@ func buildNamedFieldInfo(info *fieldInfo, lowerCaseName string, ft reflect.Type)
 			mapField: elemInfo,
 		}
 	default:
-		finfo, err = buildFieldsInfo(ft)
+		finfo, err = buildFieldsInfo(ft, fullName)
 		if err != nil {
 			return err
 		}
 	}
 
-	return addOrMergeFields(info, lowerCaseName, finfo)
+	return addOrMergeFields(info, lowerCaseName, finfo, fullName)
 }
 
-func buildStructFieldsInfo(tp reflect.Type) (*fieldInfo, error) {
+func buildStructFieldsInfo(tp reflect.Type, fullName string) (*fieldInfo, error) {
 	info := &fieldInfo{
 		children: make(map[string]*fieldInfo),
 	}
 
 	for i := 0; i < tp.NumField(); i++ {
 		field := tp.Field(i)
-		name := field.Name
+		if !field.IsExported() {
+			continue
+		}
+
+		name := getTagName(field)
 		lowerCaseName := toLowerCase(name)
 		ft := mapping.Deref(field.Type)
 		// flatten anonymous fields
 		if field.Anonymous {
-			if err := buildAnonymousFieldInfo(info, lowerCaseName, ft); err != nil {
+			if err := buildAnonymousFieldInfo(info, lowerCaseName, ft,
+				getFullName(fullName, lowerCaseName)); err != nil {
 				return nil, err
 			}
-		} else if err := buildNamedFieldInfo(info, lowerCaseName, ft); err != nil {
+		} else if err := buildNamedFieldInfo(info, lowerCaseName, ft,
+			getFullName(fullName, lowerCaseName)); err != nil {
 			return nil, err
 		}
 	}
@@ -255,15 +264,32 @@ func buildStructFieldsInfo(tp reflect.Type) (*fieldInfo, error) {
 	return info, nil
 }
 
-func mergeFields(prev *fieldInfo, key string, children map[string]*fieldInfo) error {
+// getTagName get the tag name of the given field, if no tag name, use file.Name.
+// field.Name is returned on tags like `json:""` and `json:",optional"`.
+func getTagName(field reflect.StructField) string {
+	if tag, ok := field.Tag.Lookup(jsonTagKey); ok {
+		if pos := strings.IndexByte(tag, jsonTagSep); pos >= 0 {
+			tag = tag[:pos]
+		}
+
+		tag = strings.TrimSpace(tag)
+		if len(tag) > 0 {
+			return tag
+		}
+	}
+
+	return field.Name
+}
+
+func mergeFields(prev *fieldInfo, key string, children map[string]*fieldInfo, fullName string) error {
 	if len(prev.children) == 0 || len(children) == 0 {
-		return newDupKeyError(key)
+		return newConflictKeyError(fullName)
 	}
 
 	// merge fields
 	for k, v := range children {
 		if _, ok := prev.children[k]; ok {
-			return newDupKeyError(k)
+			return newConflictKeyError(fullName)
 		}
 
 		prev.children[k] = v
@@ -314,14 +340,22 @@ func toLowerCaseKeyMap(m map[string]any, info *fieldInfo) map[string]any {
 	return res
 }
 
-type dupKeyError struct {
+type conflictKeyError struct {
 	key string
 }
 
-func newDupKeyError(key string) dupKeyError {
+func newConflictKeyError(key string) conflictKeyError {
-	return dupKeyError{key: key}
+	return conflictKeyError{key: key}
 }
 
-func (e dupKeyError) Error() string {
+func (e conflictKeyError) Error() string {
-	return fmt.Sprintf("duplicated key %s", e.key)
+	return fmt.Sprintf("conflict key %s, pay attention to anonymous fields", e.key)
+}
+
+func getFullName(parent, child string) string {
+	if len(parent) == 0 {
+		return child
+	}
+
+	return strings.Join([]string{parent, child}, ".")
 }

core/conf/config_test.go

@@ -2,6 +2,7 @@ package conf
 
 import (
 	"os"
+	"reflect"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -9,7 +10,7 @@ import (
 	"github.com/zeromicro/go-zero/core/hash"
 )
 
-var dupErr dupKeyError
+var dupErr conflictKeyError
 
 func TestLoadConfig_notExists(t *testing.T) {
 	assert.NotNil(t, Load("not_a_file", nil))
@@ -34,11 +35,11 @@ func TestConfigJson(t *testing.T) {
 	"c": "${FOO}",
 	"d": "abcd!@#$112"
 }`
+	t.Setenv("FOO", "2")
+
 	for _, test := range tests {
 		test := test
 		t.Run(test, func(t *testing.T) {
-			os.Setenv("FOO", "2")
-			defer os.Unsetenv("FOO")
 			tmpfile, err := createTempFile(test, text)
 			assert.Nil(t, err)
 			defer os.Remove(tmpfile)
@@ -80,8 +81,7 @@ b = 1
 c = "${FOO}"
 d = "abcd!@#$112"
 `
-	os.Setenv("FOO", "2")
+	t.Setenv("FOO", "2")
-	defer os.Unsetenv("FOO")
 	tmpfile, err := createTempFile(".toml", text)
 	assert.Nil(t, err)
 	defer os.Remove(tmpfile)
@@ -123,6 +123,24 @@ d = "abcd"
 	}
 }
 
+func TestConfigWithLower(t *testing.T) {
+	text := `a = "foo"
+b = 1
+`
+	tmpfile, err := createTempFile(".toml", text)
+	assert.Nil(t, err)
+	defer os.Remove(tmpfile)
+
+	var val struct {
+		A string `json:"a"`
+		b int
+	}
+	if assert.NoError(t, Load(tmpfile, &val)) {
+		assert.Equal(t, "foo", val.A)
+		assert.Equal(t, 0, val.b)
+	}
+}
+
 func TestConfigJsonCanonical(t *testing.T) {
 	text := []byte(`{"a": "foo", "B": "bar"}`)
 
@@ -188,8 +206,7 @@ b = 1
 c = "${FOO}"
 d = "abcd!@#112"
 `
-	os.Setenv("FOO", "2")
+	t.Setenv("FOO", "2")
-	defer os.Unsetenv("FOO")
 	tmpfile, err := createTempFile(".toml", text)
 	assert.Nil(t, err)
 	defer os.Remove(tmpfile)
@@ -220,11 +237,10 @@ func TestConfigJsonEnv(t *testing.T) {
 	"c": "${FOO}",
 	"d": "abcd!@#$a12 3"
 }`
+	t.Setenv("FOO", "2")
 	for _, test := range tests {
 		test := test
 		t.Run(test, func(t *testing.T) {
-			os.Setenv("FOO", "2")
-			defer os.Unsetenv("FOO")
 			tmpfile, err := createTempFile(test, text)
 			assert.Nil(t, err)
 			defer os.Remove(tmpfile)
@@ -672,7 +688,7 @@ func Test_FieldOverwrite(t *testing.T) {
 			input := []byte(`{"Name": "hello"}`)
 			err := LoadFromJsonBytes(input, val)
 			assert.ErrorAs(t, err, &dupErr)
-			assert.Equal(t, newDupKeyError("name").Error(), err.Error())
+			assert.Equal(t, newConflictKeyError("name").Error(), err.Error())
 		}
 
 		validate(&St1{})
@@ -715,7 +731,7 @@ func Test_FieldOverwrite(t *testing.T) {
 			input := []byte(`{"Name": "hello"}`)
 			err := LoadFromJsonBytes(input, val)
 			assert.ErrorAs(t, err, &dupErr)
-			assert.Equal(t, newDupKeyError("name").Error(), err.Error())
+			assert.Error(t, err)
 		}
 
 		validate(&St0{})
@@ -1022,22 +1038,22 @@ func TestLoadNamedFieldOverwritten(t *testing.T) {
 	})
 }
 
-func createTempFile(ext, text string) (string, error) {
+func TestLoadLowerMemberShouldNotConflict(t *testing.T) {
-	tmpfile, err := os.CreateTemp(os.TempDir(), hash.Md5Hex([]byte(text))+"*"+ext)
+	type (
-	if err != nil {
+		Redis struct {
-		return "", err
+			db uint
-	}
+		}
 
-	if err := os.WriteFile(tmpfile.Name(), []byte(text), os.ModeTemporary); err != nil {
+		Config struct {
-		return "", err
+			db uint
-	}
+			Redis
+		}
+	)
 
-	filename := tmpfile.Name()
+	var c Config
-	if err = tmpfile.Close(); err != nil {
+	assert.NoError(t, LoadFromJsonBytes([]byte(`{}`), &c))
-		return "", err
+	assert.Zero(t, c.db)
-	}
+	assert.Zero(t, c.Redis.db)
-
-	return filename, nil
 }
 
 func TestFillDefaultUnmarshal(t *testing.T) {
@@ -1079,7 +1095,7 @@ func TestFillDefaultUnmarshal(t *testing.T) {
 		assert.Equal(t, st.C, "c")
 	})
 
-	t.Run("has vaue", func(t *testing.T) {
+	t.Run("has value", func(t *testing.T) {
 		type St struct {
 			A string `json:",default=a"`
 			B string
@@ -1091,3 +1107,201 @@ func TestFillDefaultUnmarshal(t *testing.T) {
 		assert.Error(t, err)
 	})
 }
+
+func TestConfigWithJsonTag(t *testing.T) {
+	t.Run("map with value", func(t *testing.T) {
+		var input = []byte(`[Value]
+[Value.first]
+Email = "foo"
+[Value.second]
+Email = "bar"`)
+
+		type Value struct {
+			Email string
+		}
+
+		type Config struct {
+			ValueMap map[string]Value `json:"Value"`
+		}
+
+		var c Config
+		if assert.NoError(t, LoadFromTomlBytes(input, &c)) {
+			assert.Len(t, c.ValueMap, 2)
+		}
+	})
+
+	t.Run("map with ptr value", func(t *testing.T) {
+		var input = []byte(`[Value]
+[Value.first]
+Email = "foo"
+[Value.second]
+Email = "bar"`)
+
+		type Value struct {
+			Email string
+		}
+
+		type Config struct {
+			ValueMap map[string]*Value `json:"Value"`
+		}
+
+		var c Config
+		if assert.NoError(t, LoadFromTomlBytes(input, &c)) {
+			assert.Len(t, c.ValueMap, 2)
+		}
+	})
+
+	t.Run("map with optional", func(t *testing.T) {
+		var input = []byte(`[Value]
+[Value.first]
+Email = "foo"
+[Value.second]
+Email = "bar"`)
+
+		type Value struct {
+			Email string
+		}
+
+		type Config struct {
+			Value map[string]Value `json:",optional"`
+		}
+
+		var c Config
+		if assert.NoError(t, LoadFromTomlBytes(input, &c)) {
+			assert.Len(t, c.Value, 2)
+		}
+	})
+
+	t.Run("map with empty tag", func(t *testing.T) {
+		var input = []byte(`[Value]
+[Value.first]
+Email = "foo"
+[Value.second]
+Email = "bar"`)
+
+		type Value struct {
+			Email string
+		}
+
+		type Config struct {
+			Value map[string]Value `json:"  "`
+		}
+
+		var c Config
+		if assert.NoError(t, LoadFromTomlBytes(input, &c)) {
+			assert.Len(t, c.Value, 2)
+		}
+	})
+}
+
+func Test_getFullName(t *testing.T) {
+	assert.Equal(t, "a.b", getFullName("a", "b"))
+	assert.Equal(t, "a", getFullName("", "a"))
+}
+
+func Test_buildFieldsInfo(t *testing.T) {
+	type ParentSt struct {
+		Name string
+		M    map[string]int
+	}
+	tests := []struct {
+		name        string
+		t           reflect.Type
+		ok          bool
+		containsKey string
+	}{
+		{
+			name: "normal",
+			t:    reflect.TypeOf(struct{ A string }{}),
+			ok:   true,
+		},
+		{
+			name: "struct anonymous",
+			t: reflect.TypeOf(struct {
+				ParentSt
+				Name string
+			}{}),
+			ok:          false,
+			containsKey: newConflictKeyError("name").Error(),
+		},
+		{
+			name: "struct ptr anonymous",
+			t: reflect.TypeOf(struct {
+				*ParentSt
+				Name string
+			}{}),
+			ok:          false,
+			containsKey: newConflictKeyError("name").Error(),
+		},
+		{
+			name: "more struct anonymous",
+			t: reflect.TypeOf(struct {
+				Value struct {
+					ParentSt
+					Name string
+				}
+			}{}),
+			ok:          false,
+			containsKey: newConflictKeyError("value.name").Error(),
+		},
+		{
+			name: "map anonymous",
+			t: reflect.TypeOf(struct {
+				ParentSt
+				M string
+			}{}),
+			ok:          false,
+			containsKey: newConflictKeyError("m").Error(),
+		},
+		{
+			name: "map more anonymous",
+			t: reflect.TypeOf(struct {
+				Value struct {
+					ParentSt
+					M string
+				}
+			}{}),
+			ok:          false,
+			containsKey: newConflictKeyError("value.m").Error(),
+		},
+		{
+			name: "struct slice anonymous",
+			t: reflect.TypeOf([]struct {
+				ParentSt
+				Name string
+			}{}),
+			ok:          false,
+			containsKey: newConflictKeyError("name").Error(),
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			_, err := buildFieldsInfo(tt.t, "")
+			if tt.ok {
+				assert.NoError(t, err)
+			} else {
+				assert.Error(t, err)
+				assert.Equal(t, err.Error(), tt.containsKey)
+			}
+		})
+	}
+}
+
+func createTempFile(ext, text string) (string, error) {
+	tmpFile, err := os.CreateTemp(os.TempDir(), hash.Md5Hex([]byte(text))+"*"+ext)
+	if err != nil {
+		return "", err
+	}
+
+	if err := os.WriteFile(tmpFile.Name(), []byte(text), os.ModeTemporary); err != nil {
+		return "", err
+	}
+
+	filename := tmpFile.Name()
+	if err = tmpFile.Close(); err != nil {
+		return "", err
+	}
+
+	return filename, nil
+}

core/conf/properties_test.go

@@ -45,8 +45,7 @@ func TestPropertiesEnv(t *testing.T) {
 	assert.Nil(t, err)
 	defer os.Remove(tmpfile)
 
-	os.Setenv("FOO", "2")
+	t.Setenv("FOO", "2")
-	defer os.Unsetenv("FOO")
 
 	props, err := LoadProperties(tmpfile, UseEnv())
 	assert.Nil(t, err)

core/discov/config.go

@@ -13,6 +13,7 @@ var (
 type EtcdConf struct {
 	Hosts              []string
 	Key                string
+	ID                 int64  `json:",optional"`
 	User               string `json:",optional"`
 	Pass               string `json:",optional"`
 	CertFile           string `json:",optional"`
@@ -26,6 +27,11 @@ func (c EtcdConf) HasAccount() bool {
 	return len(c.User) > 0 && len(c.Pass) > 0
 }
 
+// HasID returns if ID provided.
+func (c EtcdConf) HasID() bool {
+	return c.ID > 0
+}
+
 // HasTLS returns if TLS CertFile/CertKeyFile/CACertFile are provided.
 func (c EtcdConf) HasTLS() bool {
 	return len(c.CertFile) > 0 && len(c.CertKeyFile) > 0 && len(c.CACertFile) > 0

core/discov/config_test.go

@@ -80,3 +80,90 @@ func TestEtcdConf_HasAccount(t *testing.T) {
 		assert.Equal(t, test.hasAccount, test.EtcdConf.HasAccount())
 	}
 }
+
+func TestEtcdConf_HasID(t *testing.T) {
+	tests := []struct {
+		EtcdConf
+		hasServerID bool
+	}{
+		{
+			EtcdConf: EtcdConf{
+				Hosts: []string{"any"},
+				ID:    -1,
+			},
+			hasServerID: false,
+		},
+		{
+			EtcdConf: EtcdConf{
+				Hosts: []string{"any"},
+				ID:    0,
+			},
+			hasServerID: false,
+		},
+		{
+			EtcdConf: EtcdConf{
+				Hosts: []string{"any"},
+				ID:    10000,
+			},
+			hasServerID: true,
+		},
+	}
+
+	for _, test := range tests {
+		assert.Equal(t, test.hasServerID, test.EtcdConf.HasID())
+	}
+}
+
+func TestEtcdConf_HasTLS(t *testing.T) {
+	tests := []struct {
+		name string
+		conf EtcdConf
+		want bool
+	}{
+		{
+			name: "empty config",
+			conf: EtcdConf{},
+			want: false,
+		},
+		{
+			name: "missing CertFile",
+			conf: EtcdConf{
+				CertKeyFile: "key",
+				CACertFile:  "ca",
+			},
+			want: false,
+		},
+		{
+			name: "missing CertKeyFile",
+			conf: EtcdConf{
+				CertFile:   "cert",
+				CACertFile: "ca",
+			},
+			want: false,
+		},
+		{
+			name: "missing CACertFile",
+			conf: EtcdConf{
+				CertFile:    "cert",
+				CertKeyFile: "key",
+			},
+			want: false,
+		},
+		{
+			name: "valid config",
+			conf: EtcdConf{
+				CertFile:    "cert",
+				CertKeyFile: "key",
+				CACertFile:  "ca",
+			},
+			want: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := tt.conf.HasTLS()
+			assert.Equal(t, tt.want, got)
+		})
+	}
+}

core/discov/internal/accountmanager_test.go

@@ -1,12 +1,85 @@
 package internal
 
 import (
+	"os"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/zeromicro/go-zero/core/stringx"
 )
 
+const (
+	certContent = `-----BEGIN CERTIFICATE-----
+MIIDazCCAlOgAwIBAgIUEg9GVO2oaPn+YSmiqmFIuAo10WIwDQYJKoZIhvcNAQEM
+BQAwRTELMAkGA1UEBhMCQVUxEzARBgNVBAgMClNvbWUtU3RhdGUxITAfBgNVBAoM
+GEludGVybmV0IFdpZGdpdHMgUHR5IEx0ZDAgFw0yMzAzMTExMzIxMjNaGA8yMTIz
+MDIxNTEzMjEyM1owRTELMAkGA1UEBhMCQVUxEzARBgNVBAgMClNvbWUtU3RhdGUx
+ITAfBgNVBAoMGEludGVybmV0IFdpZGdpdHMgUHR5IEx0ZDCCASIwDQYJKoZIhvcN
+AQEBBQADggEPADCCAQoCggEBALplXlWsIf0O/IgnIplmiZHKGnxyfyufyE2FBRNk
+OofRqbKuPH8GNqbkvZm7N29fwTDAQ+mViAggCkDht4hOzoWJMA7KYJt8JnTSWL48
+M1lcrpc9DL2gszC/JF/FGvyANbBtLklkZPFBGdHUX14pjrT937wqPtm+SqUHSvRT
+B7bmwmm2drRcmhpVm98LSlV7uQ2EgnJgsLjBPITKUejLmVLHfgX0RwQ2xIpX9pS4
+FCe1BTacwl2gGp7Mje7y4Mfv3o0ArJW6Tuwbjx59ZXwb1KIP71b7bT04AVS8ZeYO
+UMLKKuB5UR9x9Rn6cLXOTWBpcMVyzDgrAFLZjnE9LPUolZMCAwEAAaNRME8wHwYD
+VR0jBBgwFoAUeW8w8pmhncbRgTsl48k4/7wnfx8wCQYDVR0TBAIwADALBgNVHQ8E
+BAMCBPAwFAYDVR0RBA0wC4IJbG9jYWxob3N0MA0GCSqGSIb3DQEBDAUAA4IBAQAI
+y9xaoS88CLPBsX6mxfcTAFVfGNTRW9VN9Ng1cCnUR+YGoXGM/l+qP4f7p8ocdGwK
+iYZErVTzXYIn+D27//wpY3klJk3gAnEUBT3QRkStBw7XnpbeZ2oPBK+cmDnCnZPS
+BIF1wxPX7vIgaxs5Zsdqwk3qvZ4Djr2wP7LabNWTLSBKgQoUY45Liw6pffLwcGF9
+UKlu54bvGze2SufISCR3ib+I+FLvqpvJhXToZWYb/pfI/HccuCL1oot1x8vx6DQy
+U+TYxlZsKS5mdNxAX3dqEkEMsgEi+g/tzDPXJImfeCGGBhIOXLm8SRypiuGdEbc9
+xkWYxRPegajuEZGvCqVs
+-----END CERTIFICATE-----`
+	keyContent = `-----BEGIN RSA PRIVATE KEY-----
+MIIEowIBAAKCAQEAumVeVawh/Q78iCcimWaJkcoafHJ/K5/ITYUFE2Q6h9Gpsq48
+fwY2puS9mbs3b1/BMMBD6ZWICCAKQOG3iE7OhYkwDspgm3wmdNJYvjwzWVyulz0M
+vaCzML8kX8Ua/IA1sG0uSWRk8UEZ0dRfXimOtP3fvCo+2b5KpQdK9FMHtubCabZ2
+tFyaGlWb3wtKVXu5DYSCcmCwuME8hMpR6MuZUsd+BfRHBDbEilf2lLgUJ7UFNpzC
+XaAansyN7vLgx+/ejQCslbpO7BuPHn1lfBvUog/vVvttPTgBVLxl5g5Qwsoq4HlR
+H3H1Gfpwtc5NYGlwxXLMOCsAUtmOcT0s9SiVkwIDAQABAoIBAD5meTJNMgO55Kjg
+ESExxpRcCIno+tHr5+6rvYtEXqPheOIsmmwb9Gfi4+Z3WpOaht5/Pz0Ppj6yGzyl
+U//6AgGKb+BDuBvVcDpjwPnOxZIBCSHwejdxeQu0scSuA97MPS0XIAvJ5FEv7ijk
+5Bht6SyGYURpECltHygoTNuGgGqmO+McCJRLE9L09lTBI6UQ/JQwWJqSr7wx6iPU
+M1Ze/srIV+7cyEPu6i0DGjS1gSQKkX68Lqn1w6oE290O+OZvleO0gZ02fLDWCZke
+aeD9+EU/Pw+rqm3H6o0szOFIpzhRp41FUdW9sybB3Yp3u7c/574E+04Z/e30LMKs
+TCtE1QECgYEA3K7KIpw0NH2HXL5C3RHcLmr204xeBfS70riBQQuVUgYdmxak2ima
+80RInskY8hRhSGTg0l+VYIH8cmjcUyqMSOELS5XfRH99r4QPiK8AguXg80T4VumY
+W3Pf+zEC2ssgP/gYthV0g0Xj5m2QxktOF9tRw5nkg739ZR4dI9lm/iECgYEA2Dnf
+uwEDGqHiQRF6/fh5BG/nGVMvrefkqx6WvTJQ3k/M/9WhxB+lr/8yH46TuS8N2b29
+FoTf3Mr9T7pr/PWkOPzoY3P56nYbKU8xSwCim9xMzhBMzj8/N9ukJvXy27/VOz56
+eQaKqnvdXNGtPJrIMDGHps2KKWlKLyAlapzjVTMCgYAA/W++tACv85g13EykfT4F
+n0k4LbsGP9DP4zABQLIMyiY72eAncmRVjwrcW36XJ2xATOONTgx3gF3HjZzfaqNy
+eD/6uNNllUTVEryXGmHgNHPL45VRnn6memCY2eFvZdXhM5W4y2PYaunY0MkDercA
++GTngbs6tBF88KOk04bYwQKBgFl68cRgsdkmnwwQYNaTKfmVGYzYaQXNzkqmWPko
+xmCJo6tHzC7ubdG8iRCYHzfmahPuuj6EdGPZuSRyYFgJi5Ftz/nAN+84OxtIQ3zn
+YWOgskQgaLh9YfsKsQ7Sf1NDOsnOnD5TX7UXl07fEpLe9vNCvAFiU8e5Y9LGudU5
+4bYTAoGBAMdX3a3bXp4cZvXNBJ/QLVyxC6fP1Q4haCR1Od3m+T00Jth2IX2dk/fl
+p6xiJT1av5JtYabv1dFKaXOS5s1kLGGuCCSKpkvFZm826aQ2AFm0XGqEQDLeei5b
+A52Kpy/YJ+RkG4BTFtAooFq6DmA0cnoP6oPvG2h6XtDJwDTPInJb
+-----END RSA PRIVATE KEY-----`
+	caContent = `-----BEGIN CERTIFICATE-----
+MIIDbTCCAlWgAwIBAgIUBJvFoCowKich7MMfseJ+DYzzirowDQYJKoZIhvcNAQEM
+BQAwRTELMAkGA1UEBhMCQVUxEzARBgNVBAgMClNvbWUtU3RhdGUxITAfBgNVBAoM
+GEludGVybmV0IFdpZGdpdHMgUHR5IEx0ZDAgFw0yMzAzMTExMzIxMDNaGA8yMTIz
+MDIxNTEzMjEwM1owRTELMAkGA1UEBhMCQVUxEzARBgNVBAgMClNvbWUtU3RhdGUx
+ITAfBgNVBAoMGEludGVybmV0IFdpZGdpdHMgUHR5IEx0ZDCCASIwDQYJKoZIhvcN
+AQEBBQADggEPADCCAQoCggEBAO4to2YMYj0bxgr2FCiweSTSFuPx33zSw2x/s9Wf
+OR41bm2DFsyYT5f3sOIKlXZEdLmOKty2e3ho3yC0EyNpVHdykkkHT3aDI17quZax
+kYi/URqqtl1Z08A22txolc04hAZisg2BypGi3vql81UW1t3zyloGnJoIAeXR9uca
+ljP6Bk3bwsxoVBLi1JtHrO0hHLQaeHmKhAyrys06X0LRdn7Px48yRZlt6FaLSa8X
+YiRM0G44bVy/h6BkoQjMYGwVmCVk6zjJ9U7ZPFqdnDMNxAfR+hjDnYodqdLDMTTR
+1NPVrnEnNwFx0AMLvgt/ba/45vZCEAmSZnFXFAJJcM7ai9ECAwEAAaNTMFEwHQYD
+VR0OBBYEFHlvMPKZoZ3G0YE7JePJOP+8J38fMB8GA1UdIwQYMBaAFHlvMPKZoZ3G
+0YE7JePJOP+8J38fMA8GA1UdEwEB/wQFMAMBAf8wDQYJKoZIhvcNAQEMBQADggEB
+AMX8dNulADOo9uQgBMyFb9TVra7iY0zZjzv4GY5XY7scd52n6CnfAPvYBBDnTr/O
+BgNp5jaujb4+9u/2qhV3f9n+/3WOb2CmPehBgVSzlXqHeQ9lshmgwZPeem2T+8Tm
+Nnc/xQnsUfCFszUDxpkr55+aLVM22j02RWqcZ4q7TAaVYL+kdFVMc8FoqG/0ro6A
+BjE/Qn0Nn7ciX1VUjDt8l+k7ummPJTmzdi6i6E4AwO9dzrGNgGJ4aWL8cC6xYcIX
+goVIRTFeONXSDno/oPjWHpIPt7L15heMpKBHNuzPkKx2YVqPHE5QZxWfS+Lzgx+Q
+E2oTTM0rYKOZ8p6000mhvKI=
+-----END CERTIFICATE-----`
+)
+
 func TestAccount(t *testing.T) {
 	endpoints := []string{
 		"192.168.0.2:2379",
@@ -32,3 +105,34 @@ func TestAccount(t *testing.T) {
 	assert.Equal(t, username, account.User)
 	assert.Equal(t, anotherPassword, account.Pass)
 }
+
+func TestTLSMethods(t *testing.T) {
+	certFile := createTempFile(t, []byte(certContent))
+	defer os.Remove(certFile)
+	keyFile := createTempFile(t, []byte(keyContent))
+	defer os.Remove(keyFile)
+	caFile := createTempFile(t, []byte(caContent))
+	defer os.Remove(caFile)
+
+	assert.NoError(t, AddTLS([]string{"foo"}, certFile, keyFile, caFile, false))
+	cfg, ok := GetTLS([]string{"foo"})
+	assert.True(t, ok)
+	assert.NotNil(t, cfg)
+
+	assert.Error(t, AddTLS([]string{"bar"}, "bad-file", keyFile, caFile, false))
+	assert.Error(t, AddTLS([]string{"bar"}, certFile, keyFile, "bad-file", false))
+}
+
+func createTempFile(t *testing.T, body []byte) string {
+	tmpFile, err := os.CreateTemp(os.TempDir(), "go-unit-*.tmp")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	tmpFile.Close()
+	if err = os.WriteFile(tmpFile.Name(), body, os.ModePerm); err != nil {
+		t.Fatal(err)
+	}
+
+	return tmpFile.Name()
+}

core/discov/internal/registry.go

@@ -337,13 +337,11 @@ func (c *cluster) watchConnState(cli EtcdClient) {
 // DialClient dials an etcd cluster with given endpoints.
 func DialClient(endpoints []string) (EtcdClient, error) {
 	cfg := clientv3.Config{
-		Endpoints:            endpoints,
+		Endpoints:           endpoints,
-		AutoSyncInterval:     autoSyncInterval,
+		AutoSyncInterval:    autoSyncInterval,
-		DialTimeout:          DialTimeout,
+		DialTimeout:         DialTimeout,
-		DialKeepAliveTime:    dialKeepAliveTime,
+		RejectOldCluster:    true,
-		DialKeepAliveTimeout: DialTimeout,
+		PermitWithoutStream: true,
-		RejectOldCluster:     true,
-		PermitWithoutStream:  true,
 	}
 	if account, ok := GetAccount(endpoints); ok {
 		cfg.Username = account.User

core/discov/internal/registry_test.go

@@ -2,8 +2,10 @@ package internal
 
 import (
 	"context"
+	"os"
 	"sync"
 	"testing"
+	"time"
 
 	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/assert"
@@ -14,6 +16,7 @@ import (
 	"go.etcd.io/etcd/api/v3/etcdserverpb"
 	"go.etcd.io/etcd/api/v3/mvccpb"
 	clientv3 "go.etcd.io/etcd/client/v3"
+	"go.etcd.io/etcd/client/v3/mock/mockserver"
 )
 
 var mockLock sync.Mutex
@@ -242,3 +245,58 @@ func TestValueOnlyContext(t *testing.T) {
 	ctx.Done()
 	assert.Nil(t, ctx.Err())
 }
+
+func TestDialClient(t *testing.T) {
+	svr, err := mockserver.StartMockServers(1)
+	assert.NoError(t, err)
+	svr.StartAt(0)
+
+	certFile := createTempFile(t, []byte(certContent))
+	defer os.Remove(certFile)
+	keyFile := createTempFile(t, []byte(keyContent))
+	defer os.Remove(keyFile)
+	caFile := createTempFile(t, []byte(caContent))
+	defer os.Remove(caFile)
+
+	endpoints := []string{svr.Servers[0].Address}
+	AddAccount(endpoints, "foo", "bar")
+	assert.NoError(t, AddTLS(endpoints, certFile, keyFile, caFile, false))
+
+	old := DialTimeout
+	DialTimeout = time.Millisecond
+	defer func() {
+		DialTimeout = old
+	}()
+	_, err = DialClient(endpoints)
+	assert.Error(t, err)
+}
+
+func TestRegistry_Monitor(t *testing.T) {
+	svr, err := mockserver.StartMockServers(1)
+	assert.NoError(t, err)
+	svr.StartAt(0)
+
+	endpoints := []string{svr.Servers[0].Address}
+	GetRegistry().lock.Lock()
+	GetRegistry().clusters = map[string]*cluster{
+		getClusterKey(endpoints): {
+			listeners: map[string][]UpdateListener{},
+			values: map[string]map[string]string{
+				"foo": {
+					"bar": "baz",
+				},
+			},
+		},
+	}
+	GetRegistry().lock.Unlock()
+	assert.Error(t, GetRegistry().Monitor(endpoints, "foo", new(mockListener)))
+}
+
+type mockListener struct {
+}
+
+func (m *mockListener) OnAdd(_ KV) {
+}
+
+func (m *mockListener) OnDelete(_ KV) {
+}

core/discov/internal/vars.go

@@ -9,7 +9,6 @@ const (
 	autoSyncInterval   = time.Minute
 	coolDownInterval   = time.Second
 	dialTimeout        = 5 * time.Second
-	dialKeepAliveTime  = 5 * time.Second
 	requestTimeout     = 3 * time.Second
 	endpointsSeparator = ","
 )

core/discov/publisher_test.go

@@ -1,7 +1,10 @@
 package discov
 
 import (
+	"context"
 	"errors"
+	"net"
+	"os"
 	"sync"
 	"testing"
 	"time"
@@ -13,6 +16,83 @@ import (
 	"github.com/zeromicro/go-zero/core/logx"
 	"github.com/zeromicro/go-zero/core/stringx"
 	clientv3 "go.etcd.io/etcd/client/v3"
+	"golang.org/x/net/http2"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/resolver"
+	"google.golang.org/grpc/resolver/manual"
+)
+
+const (
+	certContent = `-----BEGIN CERTIFICATE-----
+MIIDazCCAlOgAwIBAgIUEg9GVO2oaPn+YSmiqmFIuAo10WIwDQYJKoZIhvcNAQEM
+BQAwRTELMAkGA1UEBhMCQVUxEzARBgNVBAgMClNvbWUtU3RhdGUxITAfBgNVBAoM
+GEludGVybmV0IFdpZGdpdHMgUHR5IEx0ZDAgFw0yMzAzMTExMzIxMjNaGA8yMTIz
+MDIxNTEzMjEyM1owRTELMAkGA1UEBhMCQVUxEzARBgNVBAgMClNvbWUtU3RhdGUx
+ITAfBgNVBAoMGEludGVybmV0IFdpZGdpdHMgUHR5IEx0ZDCCASIwDQYJKoZIhvcN
+AQEBBQADggEPADCCAQoCggEBALplXlWsIf0O/IgnIplmiZHKGnxyfyufyE2FBRNk
+OofRqbKuPH8GNqbkvZm7N29fwTDAQ+mViAggCkDht4hOzoWJMA7KYJt8JnTSWL48
+M1lcrpc9DL2gszC/JF/FGvyANbBtLklkZPFBGdHUX14pjrT937wqPtm+SqUHSvRT
+B7bmwmm2drRcmhpVm98LSlV7uQ2EgnJgsLjBPITKUejLmVLHfgX0RwQ2xIpX9pS4
+FCe1BTacwl2gGp7Mje7y4Mfv3o0ArJW6Tuwbjx59ZXwb1KIP71b7bT04AVS8ZeYO
+UMLKKuB5UR9x9Rn6cLXOTWBpcMVyzDgrAFLZjnE9LPUolZMCAwEAAaNRME8wHwYD
+VR0jBBgwFoAUeW8w8pmhncbRgTsl48k4/7wnfx8wCQYDVR0TBAIwADALBgNVHQ8E
+BAMCBPAwFAYDVR0RBA0wC4IJbG9jYWxob3N0MA0GCSqGSIb3DQEBDAUAA4IBAQAI
+y9xaoS88CLPBsX6mxfcTAFVfGNTRW9VN9Ng1cCnUR+YGoXGM/l+qP4f7p8ocdGwK
+iYZErVTzXYIn+D27//wpY3klJk3gAnEUBT3QRkStBw7XnpbeZ2oPBK+cmDnCnZPS
+BIF1wxPX7vIgaxs5Zsdqwk3qvZ4Djr2wP7LabNWTLSBKgQoUY45Liw6pffLwcGF9
+UKlu54bvGze2SufISCR3ib+I+FLvqpvJhXToZWYb/pfI/HccuCL1oot1x8vx6DQy
+U+TYxlZsKS5mdNxAX3dqEkEMsgEi+g/tzDPXJImfeCGGBhIOXLm8SRypiuGdEbc9
+xkWYxRPegajuEZGvCqVs
+-----END CERTIFICATE-----`
+	keyContent = `-----BEGIN RSA PRIVATE KEY-----
+MIIEowIBAAKCAQEAumVeVawh/Q78iCcimWaJkcoafHJ/K5/ITYUFE2Q6h9Gpsq48
+fwY2puS9mbs3b1/BMMBD6ZWICCAKQOG3iE7OhYkwDspgm3wmdNJYvjwzWVyulz0M
+vaCzML8kX8Ua/IA1sG0uSWRk8UEZ0dRfXimOtP3fvCo+2b5KpQdK9FMHtubCabZ2
+tFyaGlWb3wtKVXu5DYSCcmCwuME8hMpR6MuZUsd+BfRHBDbEilf2lLgUJ7UFNpzC
+XaAansyN7vLgx+/ejQCslbpO7BuPHn1lfBvUog/vVvttPTgBVLxl5g5Qwsoq4HlR
+H3H1Gfpwtc5NYGlwxXLMOCsAUtmOcT0s9SiVkwIDAQABAoIBAD5meTJNMgO55Kjg
+ESExxpRcCIno+tHr5+6rvYtEXqPheOIsmmwb9Gfi4+Z3WpOaht5/Pz0Ppj6yGzyl
+U//6AgGKb+BDuBvVcDpjwPnOxZIBCSHwejdxeQu0scSuA97MPS0XIAvJ5FEv7ijk
+5Bht6SyGYURpECltHygoTNuGgGqmO+McCJRLE9L09lTBI6UQ/JQwWJqSr7wx6iPU
+M1Ze/srIV+7cyEPu6i0DGjS1gSQKkX68Lqn1w6oE290O+OZvleO0gZ02fLDWCZke
+aeD9+EU/Pw+rqm3H6o0szOFIpzhRp41FUdW9sybB3Yp3u7c/574E+04Z/e30LMKs
+TCtE1QECgYEA3K7KIpw0NH2HXL5C3RHcLmr204xeBfS70riBQQuVUgYdmxak2ima
+80RInskY8hRhSGTg0l+VYIH8cmjcUyqMSOELS5XfRH99r4QPiK8AguXg80T4VumY
+W3Pf+zEC2ssgP/gYthV0g0Xj5m2QxktOF9tRw5nkg739ZR4dI9lm/iECgYEA2Dnf
+uwEDGqHiQRF6/fh5BG/nGVMvrefkqx6WvTJQ3k/M/9WhxB+lr/8yH46TuS8N2b29
+FoTf3Mr9T7pr/PWkOPzoY3P56nYbKU8xSwCim9xMzhBMzj8/N9ukJvXy27/VOz56
+eQaKqnvdXNGtPJrIMDGHps2KKWlKLyAlapzjVTMCgYAA/W++tACv85g13EykfT4F
+n0k4LbsGP9DP4zABQLIMyiY72eAncmRVjwrcW36XJ2xATOONTgx3gF3HjZzfaqNy
+eD/6uNNllUTVEryXGmHgNHPL45VRnn6memCY2eFvZdXhM5W4y2PYaunY0MkDercA
++GTngbs6tBF88KOk04bYwQKBgFl68cRgsdkmnwwQYNaTKfmVGYzYaQXNzkqmWPko
+xmCJo6tHzC7ubdG8iRCYHzfmahPuuj6EdGPZuSRyYFgJi5Ftz/nAN+84OxtIQ3zn
+YWOgskQgaLh9YfsKsQ7Sf1NDOsnOnD5TX7UXl07fEpLe9vNCvAFiU8e5Y9LGudU5
+4bYTAoGBAMdX3a3bXp4cZvXNBJ/QLVyxC6fP1Q4haCR1Od3m+T00Jth2IX2dk/fl
+p6xiJT1av5JtYabv1dFKaXOS5s1kLGGuCCSKpkvFZm826aQ2AFm0XGqEQDLeei5b
+A52Kpy/YJ+RkG4BTFtAooFq6DmA0cnoP6oPvG2h6XtDJwDTPInJb
+-----END RSA PRIVATE KEY-----`
+	caContent = `-----BEGIN CERTIFICATE-----
+MIIDbTCCAlWgAwIBAgIUBJvFoCowKich7MMfseJ+DYzzirowDQYJKoZIhvcNAQEM
+BQAwRTELMAkGA1UEBhMCQVUxEzARBgNVBAgMClNvbWUtU3RhdGUxITAfBgNVBAoM
+GEludGVybmV0IFdpZGdpdHMgUHR5IEx0ZDAgFw0yMzAzMTExMzIxMDNaGA8yMTIz
+MDIxNTEzMjEwM1owRTELMAkGA1UEBhMCQVUxEzARBgNVBAgMClNvbWUtU3RhdGUx
+ITAfBgNVBAoMGEludGVybmV0IFdpZGdpdHMgUHR5IEx0ZDCCASIwDQYJKoZIhvcN
+AQEBBQADggEPADCCAQoCggEBAO4to2YMYj0bxgr2FCiweSTSFuPx33zSw2x/s9Wf
+OR41bm2DFsyYT5f3sOIKlXZEdLmOKty2e3ho3yC0EyNpVHdykkkHT3aDI17quZax
+kYi/URqqtl1Z08A22txolc04hAZisg2BypGi3vql81UW1t3zyloGnJoIAeXR9uca
+ljP6Bk3bwsxoVBLi1JtHrO0hHLQaeHmKhAyrys06X0LRdn7Px48yRZlt6FaLSa8X
+YiRM0G44bVy/h6BkoQjMYGwVmCVk6zjJ9U7ZPFqdnDMNxAfR+hjDnYodqdLDMTTR
+1NPVrnEnNwFx0AMLvgt/ba/45vZCEAmSZnFXFAJJcM7ai9ECAwEAAaNTMFEwHQYD
+VR0OBBYEFHlvMPKZoZ3G0YE7JePJOP+8J38fMB8GA1UdIwQYMBaAFHlvMPKZoZ3G
+0YE7JePJOP+8J38fMA8GA1UdEwEB/wQFMAMBAf8wDQYJKoZIhvcNAQEMBQADggEB
+AMX8dNulADOo9uQgBMyFb9TVra7iY0zZjzv4GY5XY7scd52n6CnfAPvYBBDnTr/O
+BgNp5jaujb4+9u/2qhV3f9n+/3WOb2CmPehBgVSzlXqHeQ9lshmgwZPeem2T+8Tm
+Nnc/xQnsUfCFszUDxpkr55+aLVM22j02RWqcZ4q7TAaVYL+kdFVMc8FoqG/0ro6A
+BjE/Qn0Nn7ciX1VUjDt8l+k7ummPJTmzdi6i6E4AwO9dzrGNgGJ4aWL8cC6xYcIX
+goVIRTFeONXSDno/oPjWHpIPt7L15heMpKBHNuzPkKx2YVqPHE5QZxWfS+Lzgx+Q
+E2oTTM0rYKOZ8p6000mhvKI=
+-----END CERTIFICATE-----`
 )
 
 func init() {
@@ -37,7 +117,7 @@ func TestPublisher_register(t *testing.T) {
 	assert.Nil(t, err)
 }
 
-func TestPublisher_registerWithId(t *testing.T) {
+func TestPublisher_registerWithOptions(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	defer ctrl.Finish()
 	const id = 2
@@ -49,7 +129,15 @@ func TestPublisher_registerWithId(t *testing.T) {
 		ID: 1,
 	}, nil)
 	cli.EXPECT().Put(gomock.Any(), makeEtcdKey("thekey", id), "thevalue", gomock.Any())
-	pub := NewPublisher(nil, "thekey", "thevalue", WithId(id))
+
+	certFile := createTempFile(t, []byte(certContent))
+	defer os.Remove(certFile)
+	keyFile := createTempFile(t, []byte(keyContent))
+	defer os.Remove(keyFile)
+	caFile := createTempFile(t, []byte(caContent))
+	defer os.Remove(caFile)
+	pub := NewPublisher(nil, "thekey", "thevalue", WithId(id),
+		WithPubEtcdTLS(certFile, keyFile, caFile, true))
 	_, err := pub.register(cli)
 	assert.Nil(t, err)
 }
@@ -169,3 +257,92 @@ func TestPublisher_Resume(t *testing.T) {
 	}()
 	<-publisher.resumeChan
 }
+
+func TestPublisher_keepAliveAsync(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+	const id clientv3.LeaseID = 1
+	conn := createMockConn(t)
+	defer conn.Close()
+	cli := internal.NewMockEtcdClient(ctrl)
+	cli.EXPECT().ActiveConnection().Return(conn).AnyTimes()
+	cli.EXPECT().Close()
+	defer cli.Close()
+	cli.ActiveConnection()
+	restore := setMockClient(cli)
+	defer restore()
+	cli.EXPECT().Ctx().AnyTimes()
+	cli.EXPECT().KeepAlive(gomock.Any(), id)
+	cli.EXPECT().Grant(gomock.Any(), timeToLive).Return(&clientv3.LeaseGrantResponse{
+		ID: 1,
+	}, nil)
+	cli.EXPECT().Put(gomock.Any(), makeEtcdKey("thekey", int64(id)), "thevalue", gomock.Any())
+	var wg sync.WaitGroup
+	wg.Add(1)
+	cli.EXPECT().Revoke(gomock.Any(), id).Do(func(_, _ any) {
+		wg.Done()
+	})
+	pub := NewPublisher([]string{"the-endpoint"}, "thekey", "thevalue")
+	pub.lease = id
+	assert.Nil(t, pub.KeepAlive())
+	pub.Stop()
+	wg.Wait()
+}
+
+func createMockConn(t *testing.T) *grpc.ClientConn {
+	lis, err := net.Listen("tcp", "localhost:0")
+	if err != nil {
+		t.Fatalf("Error while listening. Err: %v", err)
+	}
+	defer lis.Close()
+	lisAddr := resolver.Address{Addr: lis.Addr().String()}
+	lisDone := make(chan struct{})
+	dialDone := make(chan struct{})
+	// 1st listener accepts the connection and then does nothing
+	go func() {
+		defer close(lisDone)
+		conn, err := lis.Accept()
+		if err != nil {
+			t.Errorf("Error while accepting. Err: %v", err)
+			return
+		}
+		framer := http2.NewFramer(conn, conn)
+		if err := framer.WriteSettings(http2.Setting{}); err != nil {
+			t.Errorf("Error while writing settings. Err: %v", err)
+			return
+		}
+		<-dialDone // Close conn only after dial returns.
+	}()
+
+	r := manual.NewBuilderWithScheme("whatever")
+	r.InitialState(resolver.State{Addresses: []resolver.Address{lisAddr}})
+	client, err := grpc.DialContext(context.Background(), r.Scheme()+":///test.server",
+		grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithResolvers(r))
+	close(dialDone)
+	if err != nil {
+		t.Fatalf("Dial failed. Err: %v", err)
+	}
+
+	timeout := time.After(1 * time.Second)
+	select {
+	case <-timeout:
+		t.Fatal("timed out waiting for server to finish")
+	case <-lisDone:
+	}
+
+	return client
+}
+
+func createTempFile(t *testing.T, body []byte) string {
+	tmpFile, err := os.CreateTemp(os.TempDir(), "go-unit-*.tmp")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	tmpFile.Close()
+	if err = os.WriteFile(tmpFile.Name(), body, os.ModePerm); err != nil {
+		t.Fatal(err)
+	}
+
+	return tmpFile.Name()
+}

core/executors/bulkexecutor_test.go

@@ -78,7 +78,7 @@ func TestBulkExecutorFlush(t *testing.T) {
 	wait.Wait()
 }
 
-func TestBuldExecutorFlushSlowTasks(t *testing.T) {
+func TestBulkExecutorFlushSlowTasks(t *testing.T) {
 	const total = 1500
 	lock := new(sync.Mutex)
 	result := make([]any, 0, 10000)

core/executors/periodicalexecutor.go

@@ -81,7 +81,7 @@ func (pe *PeriodicalExecutor) Flush() bool {
 	}())
 }
 
-// Sync lets caller to run fn thread-safe with pe, especially for the underlying container.
+// Sync lets caller run fn thread-safe with pe, especially for the underlying container.
 func (pe *PeriodicalExecutor) Sync(fn func()) {
 	pe.lock.Lock()
 	defer pe.lock.Unlock()
@@ -116,7 +116,7 @@ func (pe *PeriodicalExecutor) addAndCheck(task any) (any, bool) {
 }
 
 func (pe *PeriodicalExecutor) backgroundFlush() {
-	threading.GoSafe(func() {
+	go func() {
 		// flush before quit goroutine to avoid missing tasks
 		defer pe.Flush()
 
@@ -144,7 +144,7 @@ func (pe *PeriodicalExecutor) backgroundFlush() {
 				}
 			}
 		}
-	})
+	}()
 }
 
 func (pe *PeriodicalExecutor) doneExecution() {
@@ -162,7 +162,9 @@ func (pe *PeriodicalExecutor) executeTasks(tasks any) bool {
 
 	ok := pe.hasTasks(tasks)
 	if ok {
-		pe.container.Execute(tasks)
+		threading.RunSafe(func() {
+			pe.container.Execute(tasks)
+		})
 	}
 
 	return ok

core/executors/periodicalexecutor_test.go

@@ -108,25 +108,83 @@ func TestPeriodicalExecutor_Bulk(t *testing.T) {
 	lock.Unlock()
 }
 
+func TestPeriodicalExecutor_Panic(t *testing.T) {
+	// avoid data race
+	var lock sync.Mutex
+	ticker := timex.NewFakeTicker()
+
+	var (
+		executedTasks []int
+		expected      []int
+	)
+	executor := NewPeriodicalExecutor(time.Millisecond, newContainer(time.Millisecond, func(tasks any) {
+		tt := tasks.([]int)
+		lock.Lock()
+		executedTasks = append(executedTasks, tt...)
+		lock.Unlock()
+		if tt[0] == 0 {
+			panic("test")
+		}
+	}))
+	executor.newTicker = func(duration time.Duration) timex.Ticker {
+		return ticker
+	}
+	for i := 0; i < 30; i++ {
+		executor.Add(i)
+		expected = append(expected, i)
+	}
+	ticker.Tick()
+	ticker.Tick()
+	time.Sleep(time.Millisecond)
+	lock.Lock()
+	assert.Equal(t, expected, executedTasks)
+	lock.Unlock()
+}
+
+func TestPeriodicalExecutor_FlushPanic(t *testing.T) {
+	var (
+		executedTasks []int
+		expected      []int
+		lock          sync.Mutex
+	)
+	executor := NewPeriodicalExecutor(time.Millisecond, newContainer(time.Millisecond, func(tasks any) {
+		tt := tasks.([]int)
+		lock.Lock()
+		executedTasks = append(executedTasks, tt...)
+		lock.Unlock()
+		if tt[0] == 0 {
+			panic("flush panic")
+		}
+	}))
+	for i := 0; i < 8; i++ {
+		executor.Add(i)
+		expected = append(expected, i)
+	}
+	executor.Flush()
+	lock.Lock()
+	assert.Equal(t, expected, executedTasks)
+	lock.Unlock()
+}
+
 func TestPeriodicalExecutor_Wait(t *testing.T) {
 	var lock sync.Mutex
-	executer := NewBulkExecutor(func(tasks []any) {
+	executor := NewBulkExecutor(func(tasks []any) {
 		lock.Lock()
 		defer lock.Unlock()
 		time.Sleep(10 * time.Millisecond)
 	}, WithBulkTasks(1), WithBulkInterval(time.Second))
 	for i := 0; i < 10; i++ {
-		executer.Add(1)
+		executor.Add(1)
 	}
-	executer.Flush()
+	executor.Flush()
-	executer.Wait()
+	executor.Wait()
 }
 
 func TestPeriodicalExecutor_WaitFast(t *testing.T) {
 	const total = 3
 	var cnt int
 	var lock sync.Mutex
-	executer := NewBulkExecutor(func(tasks []any) {
+	executor := NewBulkExecutor(func(tasks []any) {
 		defer func() {
 			cnt++
 		}()
@@ -135,10 +193,10 @@ func TestPeriodicalExecutor_WaitFast(t *testing.T) {
 		time.Sleep(10 * time.Millisecond)
 	}, WithBulkTasks(1), WithBulkInterval(10*time.Millisecond))
 	for i := 0; i < total; i++ {
-		executer.Add(2)
+		executor.Add(2)
 	}
-	executer.Flush()
+	executor.Flush()
-	executer.Wait()
+	executor.Wait()
 	assert.Equal(t, total, cnt)
 }
 
@@ -151,13 +209,7 @@ func TestPeriodicalExecutor_Deadlock(t *testing.T) {
 }
 
 func TestPeriodicalExecutor_hasTasks(t *testing.T) {
-	ticker := timex.NewFakeTicker()
-	defer ticker.Stop()
-
 	exec := NewPeriodicalExecutor(time.Millisecond, newContainer(time.Millisecond, nil))
-	exec.newTicker = func(d time.Duration) timex.Ticker {
-		return ticker
-	}
 	assert.False(t, exec.hasTasks(nil))
 	assert.True(t, exec.hasTasks(1))
 }

core/filex/file_test.go

@@ -74,6 +74,11 @@ func TestFirstLineShort(t *testing.T) {
 	assert.Equal(t, "first line", val)
 }
 
+func TestFirstLineError(t *testing.T) {
+	_, err := FirstLine("/tmp/does-not-exist")
+	assert.Error(t, err)
+}
+
 func TestLastLine(t *testing.T) {
 	filename, err := fs.TempFilenameWithText(text)
 	assert.Nil(t, err)
@@ -113,3 +118,8 @@ func TestLastLineWithLastNewlineShort(t *testing.T) {
 	assert.Nil(t, err)
 	assert.Equal(t, "last line", val)
 }
+
+func TestLastLineError(t *testing.T) {
+	_, err := LastLine("/tmp/does-not-exist")
+	assert.Error(t, err)
+}

core/fs/files+polyfill.go

@@ -1,5 +1,4 @@
 //go:build windows
-// +build windows
 
 package fs
 

core/fs/files.go

@@ -1,5 +1,4 @@
 //go:build linux || darwin
-// +build linux darwin
 
 package fs
 

core/fs/temps.go

@@ -11,29 +11,29 @@ import (
 // The file is kept as open, the caller should close the file handle,
 // and remove the file by name.
 func TempFileWithText(text string) (*os.File, error) {
-	tmpfile, err := os.CreateTemp(os.TempDir(), hash.Md5Hex([]byte(text)))
+	tmpFile, err := os.CreateTemp(os.TempDir(), hash.Md5Hex([]byte(text)))
 	if err != nil {
 		return nil, err
 	}
 
-	if err := os.WriteFile(tmpfile.Name(), []byte(text), os.ModeTemporary); err != nil {
+	if err := os.WriteFile(tmpFile.Name(), []byte(text), os.ModeTemporary); err != nil {
 		return nil, err
 	}
 
-	return tmpfile, nil
+	return tmpFile, nil
 }
 
 // TempFilenameWithText creates the file with the given content,
 // and returns the filename (full path).
 // The caller should remove the file after use.
 func TempFilenameWithText(text string) (string, error) {
-	tmpfile, err := TempFileWithText(text)
+	tmpFile, err := TempFileWithText(text)
 	if err != nil {
 		return "", err
 	}
 
-	filename := tmpfile.Name()
+	filename := tmpFile.Name()
-	if err = tmpfile.Close(); err != nil {
+	if err = tmpFile.Close(); err != nil {
 		return "", err
 	}
 

core/fx/retry.go

@@ -1,31 +1,87 @@
 package fx
 
-import "github.com/zeromicro/go-zero/core/errorx"
+import (
+	"context"
+	"errors"
+	"time"
+
+	"github.com/zeromicro/go-zero/core/errorx"
+)
 
 const defaultRetryTimes = 3
 
+var errTimeout = errors.New("retry timeout")
+
 type (
 	// RetryOption defines the method to customize DoWithRetry.
 	RetryOption func(*retryOptions)
 
 	retryOptions struct {
-		times int
+		times    int
+		interval time.Duration
+		timeout  time.Duration
 	}
 )
 
 // DoWithRetry runs fn, and retries if failed. Default to retry 3 times.
+// Note that if the fn function accesses global variables outside the function
+// and performs modification operations, it is best to lock them,
+// otherwise there may be data race issues
 func DoWithRetry(fn func() error, opts ...RetryOption) error {
+	return retry(func(errChan chan error, retryCount int) {
+		errChan <- fn()
+	}, opts...)
+}
+
+// DoWithRetryCtx runs fn, and retries if failed. Default to retry 3 times.
+// fn retryCount indicates the current number of retries, starting from 0
+// Note that if the fn function accesses global variables outside the function
+// and performs modification operations, it is best to lock them,
+// otherwise there may be data race issues
+func DoWithRetryCtx(ctx context.Context, fn func(ctx context.Context, retryCount int) error,
+	opts ...RetryOption) error {
+	return retry(func(errChan chan error, retryCount int) {
+		errChan <- fn(ctx, retryCount)
+	}, opts...)
+}
+
+func retry(fn func(errChan chan error, retryCount int), opts ...RetryOption) error {
 	options := newRetryOptions()
 	for _, opt := range opts {
 		opt(options)
 	}
 
 	var berr errorx.BatchError
+	var cancelFunc context.CancelFunc
+	ctx := context.Background()
+	if options.timeout > 0 {
+		ctx, cancelFunc = context.WithTimeout(ctx, options.timeout)
+		defer cancelFunc()
+	}
+
+	errChan := make(chan error, 1)
 	for i := 0; i < options.times; i++ {
-		if err := fn(); err != nil {
+		go fn(errChan, i)
-			berr.Add(err)
+
-		} else {
+		select {
-			return nil
+		case err := <-errChan:
+			if err != nil {
+				berr.Add(err)
+			} else {
+				return nil
+			}
+		case <-ctx.Done():
+			berr.Add(errTimeout)
+			return berr.Err()
+		}
+
+		if options.interval > 0 {
+			select {
+			case <-ctx.Done():
+				berr.Add(errTimeout)
+				return berr.Err()
+			case <-time.After(options.interval):
+			}
 		}
 	}
 
@@ -39,6 +95,18 @@ func WithRetry(times int) RetryOption {
 	}
 }
 
+func WithInterval(interval time.Duration) RetryOption {
+	return func(options *retryOptions) {
+		options.interval = interval
+	}
+}
+
+func WithTimeout(timeout time.Duration) RetryOption {
+	return func(options *retryOptions) {
+		options.timeout = timeout
+	}
+}
+
 func newRetryOptions() *retryOptions {
 	return &retryOptions{
 		times: defaultRetryTimes,

core/fx/retry_test.go

@@ -1,8 +1,10 @@
 package fx
 
 import (
+	"context"
 	"errors"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 )
@@ -12,31 +14,103 @@ func TestRetry(t *testing.T) {
 		return errors.New("any")
 	}))
 
-	var times int
+	times1 := 0
 	assert.Nil(t, DoWithRetry(func() error {
-		times++
+		times1++
-		if times == defaultRetryTimes {
+		if times1 == defaultRetryTimes {
 			return nil
 		}
 		return errors.New("any")
 	}))
 
-	times = 0
+	times2 := 0
 	assert.NotNil(t, DoWithRetry(func() error {
-		times++
+		times2++
-		if times == defaultRetryTimes+1 {
+		if times2 == defaultRetryTimes+1 {
 			return nil
 		}
 		return errors.New("any")
 	}))
 
 	total := 2 * defaultRetryTimes
-	times = 0
+	times3 := 0
 	assert.Nil(t, DoWithRetry(func() error {
-		times++
+		times3++
-		if times == total {
+		if times3 == total {
 			return nil
 		}
 		return errors.New("any")
 	}, WithRetry(total)))
 }
+
+func TestRetryWithTimeout(t *testing.T) {
+	assert.Nil(t, DoWithRetry(func() error {
+		return nil
+	}, WithTimeout(time.Millisecond*500)))
+
+	times1 := 0
+	assert.Nil(t, DoWithRetry(func() error {
+		times1++
+		if times1 == 1 {
+			return errors.New("any ")
+		}
+		time.Sleep(time.Millisecond * 150)
+		return nil
+	}, WithTimeout(time.Millisecond*250)))
+
+	total := defaultRetryTimes
+	times2 := 0
+	assert.Nil(t, DoWithRetry(func() error {
+		times2++
+		if times2 == total {
+			return nil
+		}
+		time.Sleep(time.Millisecond * 50)
+		return errors.New("any")
+	}, WithTimeout(time.Millisecond*50*(time.Duration(total)+2))))
+
+	assert.NotNil(t, DoWithRetry(func() error {
+		return errors.New("any")
+	}, WithTimeout(time.Millisecond*250)))
+}
+
+func TestRetryWithInterval(t *testing.T) {
+	times1 := 0
+	assert.NotNil(t, DoWithRetry(func() error {
+		times1++
+		if times1 == 1 {
+			return errors.New("any")
+		}
+		time.Sleep(time.Millisecond * 150)
+		return nil
+	}, WithTimeout(time.Millisecond*250), WithInterval(time.Millisecond*150)))
+
+	times2 := 0
+	assert.NotNil(t, DoWithRetry(func() error {
+		times2++
+		if times2 == 2 {
+			return nil
+		}
+		time.Sleep(time.Millisecond * 150)
+		return errors.New("any ")
+	}, WithTimeout(time.Millisecond*250), WithInterval(time.Millisecond*150)))
+
+}
+
+func TestRetryCtx(t *testing.T) {
+	assert.NotNil(t, DoWithRetryCtx(context.Background(), func(ctx context.Context, retryCount int) error {
+		if retryCount == 0 {
+			return errors.New("any")
+		}
+		time.Sleep(time.Millisecond * 150)
+		return nil
+	}, WithTimeout(time.Millisecond*250), WithInterval(time.Millisecond*150)))
+
+	assert.NotNil(t, DoWithRetryCtx(context.Background(), func(ctx context.Context, retryCount int) error {
+		if retryCount == 1 {
+			return nil
+		}
+		time.Sleep(time.Millisecond * 150)
+		return errors.New("any ")
+	}, WithTimeout(time.Millisecond*250), WithInterval(time.Millisecond*150)))
+}

core/fx/stream.go

@@ -292,6 +292,18 @@ func (s Stream) Map(fn MapFunc, opts ...Option) Stream {
 	}, opts...)
 }
 
+// Max returns the maximum item from the underlying source.
+func (s Stream) Max(less LessFunc) any {
+	var max any
+	for item := range s.source {
+		if max == nil || less(max, item) {
+			max = item
+		}
+	}
+
+	return max
+}
+
 // Merge merges all the items into a slice and generates a new stream.
 func (s Stream) Merge() Stream {
 	var items []any
@@ -306,6 +318,18 @@ func (s Stream) Merge() Stream {
 	return Range(source)
 }
 
+// Min returns the minimum item from the underlying source.
+func (s Stream) Min(less LessFunc) any {
+	var min any
+	for item := range s.source {
+		if min == nil || less(item, min) {
+			min = item
+		}
+	}
+
+	return min
+}
+
 // NoneMatch returns whether all elements of this stream don't match the provided predicate.
 // May not evaluate the predicate on all elements if not necessary for determining the result.
 // If the stream is empty then true is returned and the predicate is not evaluated.

core/fx/stream_test.go

@@ -503,6 +503,83 @@ func TestStream_Concat(t *testing.T) {
 	})
 }
 
+func TestStream_Max(t *testing.T) {
+	runCheckedTest(t, func(t *testing.T) {
+		tests := []struct {
+			name     string
+			elements []any
+			max      any
+		}{
+			{
+				name: "no elements with nil",
+			},
+			{
+				name:     "no elements",
+				elements: []any{},
+				max:      nil,
+			},
+			{
+				name:     "1 element",
+				elements: []any{1},
+				max:      1,
+			},
+			{
+				name:     "multiple elements",
+				elements: []any{1, 2, 9, 5, 8},
+				max:      9,
+			},
+		}
+
+		for _, test := range tests {
+			t.Run(test.name, func(t *testing.T) {
+				val := Just(test.elements...).Max(func(a, b any) bool {
+					return a.(int) < b.(int)
+				})
+				assetEqual(t, test.max, val)
+			})
+		}
+	})
+}
+
+func TestStream_Min(t *testing.T) {
+	runCheckedTest(t, func(t *testing.T) {
+		tests := []struct {
+			name     string
+			elements []any
+			min      any
+		}{
+			{
+				name: "no elements with nil",
+				min:  nil,
+			},
+			{
+				name:     "no elements",
+				elements: []any{},
+				min:      nil,
+			},
+			{
+				name:     "1 element",
+				elements: []any{1},
+				min:      1,
+			},
+			{
+				name:     "multiple elements",
+				elements: []any{-1, 1, 2, 9, 5, 8},
+				min:      -1,
+			},
+		}
+
+		for _, test := range tests {
+			t.Run(test.name, func(t *testing.T) {
+				val := Just(test.elements...).Min(func(a, b any) bool {
+					return a.(int) < b.(int)
+				})
+				assetEqual(t, test.min, val)
+			})
+		}
+	})
+}
+
 func BenchmarkParallelMapReduce(b *testing.B) {
 	b.ReportAllocs()
 

core/iox/bufferpool.go

@@ -32,6 +32,10 @@ func (bp *BufferPool) Get() *bytes.Buffer {
 
 // Put returns buf into bp.
 func (bp *BufferPool) Put(buf *bytes.Buffer) {
+	if buf == nil {
+		return
+	}
+
 	if buf.Cap() < bp.capability {
 		bp.pool.Put(buf)
 	}

core/iox/bufferpool_test.go

@@ -13,3 +13,26 @@ func TestBufferPool(t *testing.T) {
 	pool.Put(bytes.NewBuffer(make([]byte, 0, 2*capacity)))
 	assert.True(t, pool.Get().Cap() <= capacity)
 }
+
+func TestBufferPool_Put(t *testing.T) {
+	t.Run("with nil buf", func(t *testing.T) {
+		pool := NewBufferPool(1024)
+		pool.Put(nil)
+		val := pool.Get()
+		assert.IsType(t, new(bytes.Buffer), val)
+	})
+
+	t.Run("with less-cap buf", func(t *testing.T) {
+		pool := NewBufferPool(1024)
+		pool.Put(bytes.NewBuffer(make([]byte, 0, 512)))
+		val := pool.Get()
+		assert.IsType(t, new(bytes.Buffer), val)
+	})
+
+	t.Run("with more-cap buf", func(t *testing.T) {
+		pool := NewBufferPool(1024)
+		pool.Put(bytes.NewBuffer(make([]byte, 0, 1024<<1)))
+		val := pool.Get()
+		assert.IsType(t, new(bytes.Buffer), val)
+	})
+}

core/iox/nopcloser_test.go

@@ -0,0 +1,12 @@
+package iox
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestNopCloser(t *testing.T) {
+	closer := NopCloser(nil)
+	assert.NoError(t, closer.Close())
+}

core/iox/read.go

@@ -35,6 +35,16 @@ func KeepSpace() TextReadOption {
 	}
 }
 
+// LimitDupReadCloser returns two io.ReadCloser that read from the first will be written to the second.
+// But the second io.ReadCloser is limited to up to n bytes.
+// The first returned reader needs to be read first, because the content
+// read from it will be written to the underlying buffer of the second reader.
+func LimitDupReadCloser(reader io.ReadCloser, n int64) (io.ReadCloser, io.ReadCloser) {
+	var buf bytes.Buffer
+	tee := LimitTeeReader(reader, &buf, n)
+	return io.NopCloser(tee), io.NopCloser(&buf)
+}
+
 // ReadBytes reads exactly the bytes with the length of len(buf)
 func ReadBytes(reader io.Reader, buf []byte) error {
 	var got int

core/iox/read_test.go

@@ -40,17 +40,22 @@ b`,
 
 	for _, test := range tests {
 		t.Run(test.input, func(t *testing.T) {
-			tmpfile, err := fs.TempFilenameWithText(test.input)
+			tmpFile, err := fs.TempFilenameWithText(test.input)
 			assert.Nil(t, err)
-			defer os.Remove(tmpfile)
+			defer os.Remove(tmpFile)
 
-			content, err := ReadText(tmpfile)
+			content, err := ReadText(tmpFile)
 			assert.Nil(t, err)
 			assert.Equal(t, test.expect, content)
 		})
 	}
 }
 
+func TestReadTextError(t *testing.T) {
+	_, err := ReadText("not-exist")
+	assert.NotNil(t, err)
+}
+
 func TestReadTextLines(t *testing.T) {
 	text := `1
 
@@ -59,9 +64,9 @@ func TestReadTextLines(t *testing.T) {
     #a
     3`
 
-	tmpfile, err := fs.TempFilenameWithText(text)
+	tmpFile, err := fs.TempFilenameWithText(text)
 	assert.Nil(t, err)
-	defer os.Remove(tmpfile)
+	defer os.Remove(tmpFile)
 
 	tests := []struct {
 		options     []TextReadOption
@@ -87,13 +92,18 @@ func TestReadTextLines(t *testing.T) {
 
 	for _, test := range tests {
 		t.Run(stringx.Rand(), func(t *testing.T) {
-			lines, err := ReadTextLines(tmpfile, test.options...)
+			lines, err := ReadTextLines(tmpFile, test.options...)
 			assert.Nil(t, err)
 			assert.Equal(t, test.expectLines, len(lines))
 		})
 	}
 }
 
+func TestReadTextLinesError(t *testing.T) {
+	_, err := ReadTextLines("not-exist")
+	assert.NotNil(t, err)
+}
+
 func TestDupReadCloser(t *testing.T) {
 	input := "hello"
 	reader := io.NopCloser(bytes.NewBufferString(input))
@@ -108,6 +118,29 @@ func TestDupReadCloser(t *testing.T) {
 	verify(r2)
 }
 
+func TestLimitDupReadCloser(t *testing.T) {
+	input := "hello world"
+	limitBytes := int64(4)
+	reader := io.NopCloser(bytes.NewBufferString(input))
+	r1, r2 := LimitDupReadCloser(reader, limitBytes)
+	verify := func(r io.Reader) {
+		output, err := io.ReadAll(r)
+		assert.Nil(t, err)
+		assert.Equal(t, input, string(output))
+	}
+	verifyLimit := func(r io.Reader, limit int64) {
+		output, err := io.ReadAll(r)
+		if limit < int64(len(input)) {
+			input = input[:limit]
+		}
+		assert.Nil(t, err)
+		assert.Equal(t, input, string(output))
+	}
+
+	verify(r1)
+	verifyLimit(r2, limitBytes)
+}
+
 func TestReadBytes(t *testing.T) {
 	reader := io.NopCloser(bytes.NewBufferString("helloworld"))
 	buf := make([]byte, 5)

core/iox/tee.go

@@ -0,0 +1,35 @@
+package iox
+
+import "io"
+
+// LimitTeeReader returns a Reader that writes up to n bytes to w what it reads from r.
+// First n bytes reads from r performed through it are matched with
+// corresponding writes to w. There is no internal buffering -
+// the write must complete before the first n bytes read completes.
+// Any error encountered while writing is reported as a read error.
+func LimitTeeReader(r io.Reader, w io.Writer, n int64) io.Reader {
+	return &limitTeeReader{r, w, n}
+}
+
+type limitTeeReader struct {
+	r io.Reader
+	w io.Writer
+	n int64 // limit bytes remaining
+}
+
+func (t *limitTeeReader) Read(p []byte) (n int, err error) {
+	n, err = t.r.Read(p)
+	if n > 0 && t.n > 0 {
+		limit := int64(n)
+		if limit > t.n {
+			limit = t.n
+		}
+		if n, err := t.w.Write(p[:limit]); err != nil {
+			return n, err
+		}
+
+		t.n -= limit
+	}
+
+	return
+}

core/iox/tee_test.go

@@ -0,0 +1,40 @@
+package iox
+
+import (
+	"bytes"
+	"io"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestLimitTeeReader(t *testing.T) {
+	limit := int64(4)
+	src := []byte("hello, world")
+	dst := make([]byte, len(src))
+	rb := bytes.NewBuffer(src)
+	wb := new(bytes.Buffer)
+	r := LimitTeeReader(rb, wb, limit)
+	if n, err := io.ReadFull(r, dst); err != nil || n != len(src) {
+		t.Fatalf("ReadFull(r, dst) = %d, %v; want %d, nil", n, err, len(src))
+	}
+	if !bytes.Equal(dst, src) {
+		t.Errorf("bytes read = %q want %q", dst, src)
+	}
+	if !bytes.Equal(wb.Bytes(), src[:limit]) {
+		t.Errorf("bytes written = %q want %q", wb.Bytes(), src)
+	}
+
+	n, err := r.Read(dst)
+	assert.Equal(t, 0, n)
+	assert.Equal(t, io.EOF, err)
+
+	rb = bytes.NewBuffer(src)
+	pr, pw := io.Pipe()
+	if assert.NoError(t, pr.Close()) {
+		r = LimitTeeReader(rb, pw, limit)
+		n, err := io.ReadFull(r, dst)
+		assert.Equal(t, 0, n)
+		assert.Equal(t, io.ErrClosedPipe, err)
+	}
+}

core/iox/textfile.go

@@ -2,6 +2,7 @@ package iox
 
 import (
 	"bytes"
+	"errors"
 	"io"
 	"os"
 )
@@ -26,7 +27,7 @@ func CountLines(file string) (int, error) {
 		count += bytes.Count(buf[:c], lineSep)
 
 		switch {
-		case err == io.EOF:
+		case errors.Is(err, io.EOF):
 			if noEol {
 				count++
 			}

core/iox/textfile_test.go

@@ -24,3 +24,8 @@ func TestCountLines(t *testing.T) {
 	assert.Nil(t, err)
 	assert.Equal(t, 4, lines)
 }
+
+func TestCountLinesError(t *testing.T) {
+	_, err := CountLines("not-exist")
+	assert.NotNil(t, err)
+}

core/iox/textlinescanner_test.go

@@ -3,6 +3,7 @@ package iox
 import (
 	"strings"
 	"testing"
+	"testing/iotest"
 
 	"github.com/stretchr/testify/assert"
 )
@@ -22,3 +23,10 @@ func TestScanner(t *testing.T) {
 	}
 	assert.EqualValues(t, []string{"1", "2", "3", "4"}, lines)
 }
+
+func TestBadScanner(t *testing.T) {
+	scanner := NewTextLineScanner(iotest.ErrReader(iotest.ErrTimeout))
+	assert.False(t, scanner.Scan())
+	_, err := scanner.Line()
+	assert.ErrorIs(t, err, iotest.ErrTimeout)
+}

core/limit/periodlimit.go

@@ -9,21 +9,6 @@ import (
 	"github.com/zeromicro/go-zero/core/stores/redis"
 )
 
-// to be compatible with aliyun redis, we cannot use `local key = KEYS[1]` to reuse the key
-const periodScript = `local limit = tonumber(ARGV[1])
-local window = tonumber(ARGV[2])
-local current = redis.call("INCRBY", KEYS[1], 1)
-if current == 1 then
-    redis.call("expire", KEYS[1], window)
-end
-if current < limit then
-    return 1
-elseif current == limit then
-    return 2
-else
-    return 0
-end`
-
 const (
 	// Unknown means not initialized state.
 	Unknown = iota
@@ -39,8 +24,25 @@ const (
 	internalHitQuota  = 2
 )
 
-// ErrUnknownCode is an error that represents unknown status code.
+var (
-var ErrUnknownCode = errors.New("unknown status code")
+	// ErrUnknownCode is an error that represents unknown status code.
+	ErrUnknownCode = errors.New("unknown status code")
+
+	// to be compatible with aliyun redis, we cannot use `local key = KEYS[1]` to reuse the key
+	periodScript = redis.NewScript(`local limit = tonumber(ARGV[1])
+local window = tonumber(ARGV[2])
+local current = redis.call("INCRBY", KEYS[1], 1)
+if current == 1 then
+    redis.call("expire", KEYS[1], window)
+end
+if current < limit then
+    return 1
+elseif current == limit then
+    return 2
+else
+    return 0
+end`)
+)
 
 type (
 	// PeriodOption defines the method to customize a PeriodLimit.
@@ -80,7 +82,7 @@ func (h *PeriodLimit) Take(key string) (int, error) {
 
 // TakeCtx requests a permit with context, it returns the permit state.
 func (h *PeriodLimit) TakeCtx(ctx context.Context, key string) (int, error) {
-	resp, err := h.limitStore.EvalCtx(ctx, periodScript, []string{h.keyPrefix + key}, []string{
+	resp, err := h.limitStore.ScriptRunCtx(ctx, periodScript, []string{h.keyPrefix + key}, []string{
 		strconv.Itoa(h.quota),
 		strconv.Itoa(h.calcExpireSeconds()),
 	})

core/limit/periodlimit_test.go

@@ -33,9 +33,7 @@ func TestPeriodLimit_RedisUnavailable(t *testing.T) {
 }
 
 func testPeriodLimit(t *testing.T, opts ...PeriodOption) {
-	store, clean, err := redistest.CreateRedis()
+	store := redistest.CreateRedis(t)
-	assert.Nil(t, err)
-	defer clean()
 
 	const (
 		seconds = 1

core/limit/tokenlimit.go

@@ -15,10 +15,15 @@ import (
 )
 
 const (
-	// to be compatible with aliyun redis, we cannot use `local key = KEYS[1]` to reuse the key
+	tokenFormat     = "{%s}.tokens"
-	// KEYS[1] as tokens_key
+	timestampFormat = "{%s}.ts"
-	// KEYS[2] as timestamp_key
+	pingInterval    = time.Millisecond * 100
-	script = `local rate = tonumber(ARGV[1])
+)
+
+// to be compatible with aliyun redis, we cannot use `local key = KEYS[1]` to reuse the key
+// KEYS[1] as tokens_key
+// KEYS[2] as timestamp_key
+var script = redis.NewScript(`local rate = tonumber(ARGV[1])
 local capacity = tonumber(ARGV[2])
 local now = tonumber(ARGV[3])
 local requested = tonumber(ARGV[4])
@@ -45,11 +50,7 @@ end
 redis.call("setex", KEYS[1], ttl, new_tokens)
 redis.call("setex", KEYS[2], ttl, now)
 
-return allowed`
+return allowed`)
-	tokenFormat     = "{%s}.tokens"
-	timestampFormat = "{%s}.ts"
-	pingInterval    = time.Millisecond * 100
-)
 
 // A TokenLimiter controls how frequently events are allowed to happen with in one second.
 type TokenLimiter struct {
@@ -110,7 +111,7 @@ func (lim *TokenLimiter) reserveN(ctx context.Context, now time.Time, n int) boo
 		return lim.rescueLimiter.AllowN(now, n)
 	}
 
-	resp, err := lim.store.EvalCtx(ctx,
+	resp, err := lim.store.ScriptRunCtx(ctx,
 		script,
 		[]string{
 			lim.tokenKey,

core/limit/tokenlimit_test.go

@@ -70,9 +70,7 @@ func TestTokenLimit_Rescue(t *testing.T) {
 }
 
 func TestTokenLimit_Take(t *testing.T) {
-	store, clean, err := redistest.CreateRedis()
+	store := redistest.CreateRedis(t)
-	assert.Nil(t, err)
-	defer clean()
 
 	const (
 		total = 100
@@ -92,9 +90,7 @@ func TestTokenLimit_Take(t *testing.T) {
 }
 
 func TestTokenLimit_TakeBurst(t *testing.T) {
-	store, clean, err := redistest.CreateRedis()
+	store := redistest.CreateRedis(t)
-	assert.Nil(t, err)
-	defer clean()
 
 	const (
 		total = 100

core/logc/logs_test.go

@@ -1,7 +1,6 @@
 package logc
 
 import (
-	"bytes"
 	"context"
 	"encoding/json"
 	"fmt"
@@ -11,14 +10,11 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/zeromicro/go-zero/core/logx"
+	"github.com/zeromicro/go-zero/core/logx/logtest"
 )
 
 func TestAddGlobalFields(t *testing.T) {
-	var buf bytes.Buffer
+	buf := logtest.NewCollector(t)
-	writer := logx.NewWriter(&buf)
-	old := logx.Reset()
-	logx.SetWriter(writer)
-	defer logx.SetWriter(old)
 
 	Info(context.Background(), "hello")
 	buf.Reset()
@@ -34,155 +30,90 @@ func TestAddGlobalFields(t *testing.T) {
 }
 
 func TestAlert(t *testing.T) {
-	var buf strings.Builder
+	buf := logtest.NewCollector(t)
-	writer := logx.NewWriter(&buf)
-	old := logx.Reset()
-	logx.SetWriter(writer)
-	defer logx.SetWriter(old)
-
 	Alert(context.Background(), "foo")
 	assert.True(t, strings.Contains(buf.String(), "foo"), buf.String())
 }
 
 func TestError(t *testing.T) {
-	var buf strings.Builder
+	buf := logtest.NewCollector(t)
-	writer := logx.NewWriter(&buf)
-	old := logx.Reset()
-	logx.SetWriter(writer)
-	defer logx.SetWriter(old)
-
 	file, line := getFileLine()
 	Error(context.Background(), "foo")
 	assert.True(t, strings.Contains(buf.String(), fmt.Sprintf("%s:%d", file, line+1)))
 }
 
 func TestErrorf(t *testing.T) {
-	var buf strings.Builder
+	buf := logtest.NewCollector(t)
-	writer := logx.NewWriter(&buf)
-	old := logx.Reset()
-	logx.SetWriter(writer)
-	defer logx.SetWriter(old)
-
 	file, line := getFileLine()
 	Errorf(context.Background(), "foo %s", "bar")
 	assert.True(t, strings.Contains(buf.String(), fmt.Sprintf("%s:%d", file, line+1)))
 }
 
 func TestErrorv(t *testing.T) {
-	var buf strings.Builder
+	buf := logtest.NewCollector(t)
-	writer := logx.NewWriter(&buf)
-	old := logx.Reset()
-	logx.SetWriter(writer)
-	defer logx.SetWriter(old)
-
 	file, line := getFileLine()
 	Errorv(context.Background(), "foo")
 	assert.True(t, strings.Contains(buf.String(), fmt.Sprintf("%s:%d", file, line+1)))
 }
 
 func TestErrorw(t *testing.T) {
-	var buf strings.Builder
+	buf := logtest.NewCollector(t)
-	writer := logx.NewWriter(&buf)
-	old := logx.Reset()
-	logx.SetWriter(writer)
-	defer logx.SetWriter(old)
-
 	file, line := getFileLine()
 	Errorw(context.Background(), "foo", Field("a", "b"))
 	assert.True(t, strings.Contains(buf.String(), fmt.Sprintf("%s:%d", file, line+1)))
 }
 
 func TestInfo(t *testing.T) {
-	var buf strings.Builder
+	buf := logtest.NewCollector(t)
-	writer := logx.NewWriter(&buf)
-	old := logx.Reset()
-	logx.SetWriter(writer)
-	defer logx.SetWriter(old)
-
 	file, line := getFileLine()
 	Info(context.Background(), "foo")
 	assert.True(t, strings.Contains(buf.String(), fmt.Sprintf("%s:%d", file, line+1)))
 }
 
 func TestInfof(t *testing.T) {
-	var buf strings.Builder
+	buf := logtest.NewCollector(t)
-	writer := logx.NewWriter(&buf)
-	old := logx.Reset()
-	logx.SetWriter(writer)
-	defer logx.SetWriter(old)
-
 	file, line := getFileLine()
 	Infof(context.Background(), "foo %s", "bar")
 	assert.True(t, strings.Contains(buf.String(), fmt.Sprintf("%s:%d", file, line+1)))
 }
 
 func TestInfov(t *testing.T) {
-	var buf strings.Builder
+	buf := logtest.NewCollector(t)
-	writer := logx.NewWriter(&buf)
-	old := logx.Reset()
-	logx.SetWriter(writer)
-	defer logx.SetWriter(old)
-
 	file, line := getFileLine()
 	Infov(context.Background(), "foo")
 	assert.True(t, strings.Contains(buf.String(), fmt.Sprintf("%s:%d", file, line+1)))
 }
 
 func TestInfow(t *testing.T) {
-	var buf strings.Builder
+	buf := logtest.NewCollector(t)
-	writer := logx.NewWriter(&buf)
-	old := logx.Reset()
-	logx.SetWriter(writer)
-	defer logx.SetWriter(old)
-
 	file, line := getFileLine()
 	Infow(context.Background(), "foo", Field("a", "b"))
 	assert.True(t, strings.Contains(buf.String(), fmt.Sprintf("%s:%d", file, line+1)))
 }
 
 func TestDebug(t *testing.T) {
-	var buf strings.Builder
+	buf := logtest.NewCollector(t)
-	writer := logx.NewWriter(&buf)
-	old := logx.Reset()
-	logx.SetWriter(writer)
-	defer logx.SetWriter(old)
-
 	file, line := getFileLine()
 	Debug(context.Background(), "foo")
 	assert.True(t, strings.Contains(buf.String(), fmt.Sprintf("%s:%d", file, line+1)))
 }
 
 func TestDebugf(t *testing.T) {
-	var buf strings.Builder
+	buf := logtest.NewCollector(t)
-	writer := logx.NewWriter(&buf)
-	old := logx.Reset()
-	logx.SetWriter(writer)
-	defer logx.SetWriter(old)
-
 	file, line := getFileLine()
 	Debugf(context.Background(), "foo %s", "bar")
 	assert.True(t, strings.Contains(buf.String(), fmt.Sprintf("%s:%d", file, line+1)))
 }
 
 func TestDebugv(t *testing.T) {
-	var buf strings.Builder
+	buf := logtest.NewCollector(t)
-	writer := logx.NewWriter(&buf)
-	old := logx.Reset()
-	logx.SetWriter(writer)
-	defer logx.SetWriter(old)
-
 	file, line := getFileLine()
 	Debugv(context.Background(), "foo")
 	assert.True(t, strings.Contains(buf.String(), fmt.Sprintf("%s:%d", file, line+1)))
 }
 
 func TestDebugw(t *testing.T) {
-	var buf strings.Builder
+	buf := logtest.NewCollector(t)
-	writer := logx.NewWriter(&buf)
-	old := logx.Reset()
-	logx.SetWriter(writer)
-	defer logx.SetWriter(old)
-
 	file, line := getFileLine()
 	Debugw(context.Background(), "foo", Field("a", "b"))
 	assert.True(t, strings.Contains(buf.String(), fmt.Sprintf("%s:%d", file, line+1)))
@@ -204,48 +135,28 @@ func TestMisc(t *testing.T) {
 }
 
 func TestSlow(t *testing.T) {
-	var buf strings.Builder
+	buf := logtest.NewCollector(t)
-	writer := logx.NewWriter(&buf)
-	old := logx.Reset()
-	logx.SetWriter(writer)
-	defer logx.SetWriter(old)
-
 	file, line := getFileLine()
 	Slow(context.Background(), "foo")
 	assert.True(t, strings.Contains(buf.String(), fmt.Sprintf("%s:%d", file, line+1)), buf.String())
 }
 
 func TestSlowf(t *testing.T) {
-	var buf strings.Builder
+	buf := logtest.NewCollector(t)
-	writer := logx.NewWriter(&buf)
-	old := logx.Reset()
-	logx.SetWriter(writer)
-	defer logx.SetWriter(old)
-
 	file, line := getFileLine()
 	Slowf(context.Background(), "foo %s", "bar")
 	assert.True(t, strings.Contains(buf.String(), fmt.Sprintf("%s:%d", file, line+1)), buf.String())
 }
 
 func TestSlowv(t *testing.T) {
-	var buf strings.Builder
+	buf := logtest.NewCollector(t)
-	writer := logx.NewWriter(&buf)
-	old := logx.Reset()
-	logx.SetWriter(writer)
-	defer logx.SetWriter(old)
-
 	file, line := getFileLine()
 	Slowv(context.Background(), "foo")
 	assert.True(t, strings.Contains(buf.String(), fmt.Sprintf("%s:%d", file, line+1)), buf.String())
 }
 
 func TestSloww(t *testing.T) {
-	var buf strings.Builder
+	buf := logtest.NewCollector(t)
-	writer := logx.NewWriter(&buf)
-	old := logx.Reset()
-	logx.SetWriter(writer)
-	defer logx.SetWriter(old)
-
 	file, line := getFileLine()
 	Sloww(context.Background(), "foo", Field("a", "b"))
 	assert.True(t, strings.Contains(buf.String(), fmt.Sprintf("%s:%d", file, line+1)), buf.String())

core/logx/config.go

@@ -23,7 +23,7 @@ type LogConf struct {
 	MaxContentLength uint32 `json:",optional"`
 	// Compress represents whether to compress the log file, default is `false`.
 	Compress bool `json:",optional"`
-	// Stdout represents whether to log statistics, default is `true`.
+	// Stat represents whether to log statistics, default is `true`.
 	Stat bool `json:",default=true"`
 	// KeepDays represents how many days the log files will be kept. Default to keep all files.
 	// Only take effect when Mode is `file` or `volume`, both work when Rotation is `daily` or `size`.
@@ -32,13 +32,13 @@ type LogConf struct {
 	StackCooldownMillis int `json:",default=100"`
 	// MaxBackups represents how many backup log files will be kept. 0 means all files will be kept forever.
 	// Only take effect when RotationRuleType is `size`.
-	// Even thougth `MaxBackups` sets 0, log files will still be removed
+	// Even though `MaxBackups` sets 0, log files will still be removed
 	// if the `KeepDays` limitation is reached.
 	MaxBackups int `json:",default=0"`
 	// MaxSize represents how much space the writing log file takes up. 0 means no limit. The unit is `MB`.
 	// Only take effect when RotationRuleType is `size`
 	MaxSize int `json:",default=0"`
-	// RotationRuleType represents the type of log rotation rule. Default is `daily`.
+	// Rotation represents the type of log rotation rule. Default is `daily`.
 	// daily: daily rotation.
 	// size: size limited rotation.
 	Rotation string `json:",default=daily,options=[daily,size]"`

core/logx/fs.go

@@ -0,0 +1,40 @@
+package logx
+
+import (
+	"io"
+	"os"
+)
+
+var fileSys realFileSystem
+
+type (
+	fileSystem interface {
+		Close(closer io.Closer) error
+		Copy(writer io.Writer, reader io.Reader) (int64, error)
+		Create(name string) (*os.File, error)
+		Open(name string) (*os.File, error)
+		Remove(name string) error
+	}
+
+	realFileSystem struct{}
+)
+
+func (fs realFileSystem) Close(closer io.Closer) error {
+	return closer.Close()
+}
+
+func (fs realFileSystem) Copy(writer io.Writer, reader io.Reader) (int64, error) {
+	return io.Copy(writer, reader)
+}
+
+func (fs realFileSystem) Create(name string) (*os.File, error) {
+	return os.Create(name)
+}
+
+func (fs realFileSystem) Open(name string) (*os.File, error) {
+	return os.Open(name)
+}
+
+func (fs realFileSystem) Remove(name string) error {
+	return os.Remove(name)
+}

core/logx/logs.go

@@ -68,22 +68,30 @@ func Close() error {
 
 // Debug writes v into access log.
 func Debug(v ...any) {
-	writeDebug(fmt.Sprint(v...))
+	if shallLog(DebugLevel) {
+		writeDebug(fmt.Sprint(v...))
+	}
 }
 
 // Debugf writes v with format into access log.
 func Debugf(format string, v ...any) {
-	writeDebug(fmt.Sprintf(format, v...))
+	if shallLog(DebugLevel) {
+		writeDebug(fmt.Sprintf(format, v...))
+	}
 }
 
 // Debugv writes v into access log with json content.
 func Debugv(v any) {
-	writeDebug(v)
+	if shallLog(DebugLevel) {
+		writeDebug(v)
+	}
 }
 
 // Debugw writes msg along with fields into access log.
 func Debugw(msg string, fields ...LogField) {
-	writeDebug(msg, fields...)
+	if shallLog(DebugLevel) {
+		writeDebug(msg, fields...)
+	}
 }
 
 // Disable disables the logging.
@@ -99,35 +107,47 @@ func DisableStat() {
 
 // Error writes v into error log.
 func Error(v ...any) {
-	writeError(fmt.Sprint(v...))
+	if shallLog(ErrorLevel) {
+		writeError(fmt.Sprint(v...))
+	}
 }
 
 // Errorf writes v with format into error log.
 func Errorf(format string, v ...any) {
-	writeError(fmt.Errorf(format, v...).Error())
+	if shallLog(ErrorLevel) {
+		writeError(fmt.Errorf(format, v...).Error())
+	}
 }
 
 // ErrorStack writes v along with call stack into error log.
 func ErrorStack(v ...any) {
-	// there is newline in stack string
+	if shallLog(ErrorLevel) {
-	writeStack(fmt.Sprint(v...))
+		// there is newline in stack string
+		writeStack(fmt.Sprint(v...))
+	}
 }
 
 // ErrorStackf writes v along with call stack in format into error log.
 func ErrorStackf(format string, v ...any) {
-	// there is newline in stack string
+	if shallLog(ErrorLevel) {
-	writeStack(fmt.Sprintf(format, v...))
+		// there is newline in stack string
+		writeStack(fmt.Sprintf(format, v...))
+	}
 }
 
 // Errorv writes v into error log with json content.
 // No call stack attached, because not elegant to pack the messages.
 func Errorv(v any) {
-	writeError(v)
+	if shallLog(ErrorLevel) {
+		writeError(v)
+	}
 }
 
 // Errorw writes msg along with fields into error log.
 func Errorw(msg string, fields ...LogField) {
-	writeError(msg, fields...)
+	if shallLog(ErrorLevel) {
+		writeError(msg, fields...)
+	}
 }
 
 // Field returns a LogField for the given key and value.
@@ -170,22 +190,30 @@ func Field(key string, value any) LogField {
 
 // Info writes v into access log.
 func Info(v ...any) {
-	writeInfo(fmt.Sprint(v...))
+	if shallLog(InfoLevel) {
+		writeInfo(fmt.Sprint(v...))
+	}
 }
 
 // Infof writes v with format into access log.
 func Infof(format string, v ...any) {
-	writeInfo(fmt.Sprintf(format, v...))
+	if shallLog(InfoLevel) {
+		writeInfo(fmt.Sprintf(format, v...))
+	}
 }
 
 // Infov writes v into access log with json content.
 func Infov(v any) {
-	writeInfo(v)
+	if shallLog(InfoLevel) {
+		writeInfo(v)
+	}
 }
 
 // Infow writes msg along with fields into access log.
 func Infow(msg string, fields ...LogField) {
-	writeInfo(msg, fields...)
+	if shallLog(InfoLevel) {
+		writeInfo(msg, fields...)
+	}
 }
 
 // Must checks if err is nil, otherwise logs the error and exits.
@@ -194,10 +222,15 @@ func Must(err error) {
 		return
 	}
 
-	msg := err.Error()
+	msg := fmt.Sprintf("%+v\n\n%s", err.Error(), debug.Stack())
 	log.Print(msg)
 	getWriter().Severe(msg)
-	os.Exit(1)
+
+	if ExitOnFatal.True() {
+		os.Exit(1)
+	} else {
+		panic(msg)
+	}
 }
 
 // MustSetup sets up logging with given config c. It exits on error.
@@ -264,42 +297,58 @@ func SetUp(c LogConf) (err error) {
 
 // Severe writes v into severe log.
 func Severe(v ...any) {
-	writeSevere(fmt.Sprint(v...))
+	if shallLog(SevereLevel) {
+		writeSevere(fmt.Sprint(v...))
+	}
 }
 
 // Severef writes v with format into severe log.
 func Severef(format string, v ...any) {
-	writeSevere(fmt.Sprintf(format, v...))
+	if shallLog(SevereLevel) {
+		writeSevere(fmt.Sprintf(format, v...))
+	}
 }
 
 // Slow writes v into slow log.
 func Slow(v ...any) {
-	writeSlow(fmt.Sprint(v...))
+	if shallLog(ErrorLevel) {
+		writeSlow(fmt.Sprint(v...))
+	}
 }
 
 // Slowf writes v with format into slow log.
 func Slowf(format string, v ...any) {
-	writeSlow(fmt.Sprintf(format, v...))
+	if shallLog(ErrorLevel) {
+		writeSlow(fmt.Sprintf(format, v...))
+	}
 }
 
 // Slowv writes v into slow log with json content.
 func Slowv(v any) {
-	writeSlow(v)
+	if shallLog(ErrorLevel) {
+		writeSlow(v)
+	}
 }
 
 // Sloww writes msg along with fields into slow log.
 func Sloww(msg string, fields ...LogField) {
-	writeSlow(msg, fields...)
+	if shallLog(ErrorLevel) {
+		writeSlow(msg, fields...)
+	}
 }
 
 // Stat writes v into stat log.
 func Stat(v ...any) {
-	writeStat(fmt.Sprint(v...))
+	if shallLogStat() && shallLog(InfoLevel) {
+		writeStat(fmt.Sprint(v...))
+	}
 }
 
 // Statf writes v with format into stat log.
 func Statf(format string, v ...any) {
-	writeStat(fmt.Sprintf(format, v...))
+	if shallLogStat() && shallLog(InfoLevel) {
+		writeStat(fmt.Sprintf(format, v...))
+	}
 }
 
 // WithCooldownMillis customizes logging on writing call stack interval.
@@ -353,14 +402,16 @@ func createOutput(path string) (io.WriteCloser, error) {
 		return nil, ErrLogPathNotSet
 	}
 
+	var rule RotateRule
 	switch options.rotationRule {
 	case sizeRotationRule:
-		return NewLogger(path, NewSizeLimitRotateRule(path, backupFileDelimiter, options.keepDays,
+		rule = NewSizeLimitRotateRule(path, backupFileDelimiter, options.keepDays, options.maxSize,
-			options.maxSize, options.maxBackups, options.gzipEnabled), options.gzipEnabled)
+			options.maxBackups, options.gzipEnabled)
 	default:
-		return NewLogger(path, DefaultRotateRule(path, backupFileDelimiter, options.keepDays,
+		rule = DefaultRotateRule(path, backupFileDelimiter, options.keepDays, options.gzipEnabled)
-			options.gzipEnabled), options.gzipEnabled)
 	}
+
+	return NewLogger(path, rule, options.gzipEnabled)
 }
 
 func getWriter() Writer {
@@ -422,44 +473,58 @@ func shallLogStat() bool {
 	return atomic.LoadUint32(&disableStat) == 0
 }
 
+// writeDebug writes v into debug log.
+// Not checking shallLog here is for performance consideration.
+// If we check shallLog here, the fmt.Sprint might be called even if the log level is not enabled.
+// The caller should check shallLog before calling this function.
 func writeDebug(val any, fields ...LogField) {
-	if shallLog(DebugLevel) {
+	getWriter().Debug(val, addCaller(fields...)...)
-		getWriter().Debug(val, addCaller(fields...)...)
-	}
 }
 
+// writeError writes v into error log.
+// Not checking shallLog here is for performance consideration.
+// If we check shallLog here, the fmt.Sprint might be called even if the log level is not enabled.
+// The caller should check shallLog before calling this function.
 func writeError(val any, fields ...LogField) {
-	if shallLog(ErrorLevel) {
+	getWriter().Error(val, addCaller(fields...)...)
-		getWriter().Error(val, addCaller(fields...)...)
-	}
 }
 
+// writeInfo writes v into info log.
+// Not checking shallLog here is for performance consideration.
+// If we check shallLog here, the fmt.Sprint might be called even if the log level is not enabled.
+// The caller should check shallLog before calling this function.
 func writeInfo(val any, fields ...LogField) {
-	if shallLog(InfoLevel) {
+	getWriter().Info(val, addCaller(fields...)...)
-		getWriter().Info(val, addCaller(fields...)...)
-	}
 }
 
+// writeSevere writes v into severe log.
+// Not checking shallLog here is for performance consideration.
+// If we check shallLog here, the fmt.Sprint might be called even if the log level is not enabled.
+// The caller should check shallLog before calling this function.
 func writeSevere(msg string) {
-	if shallLog(SevereLevel) {
+	getWriter().Severe(fmt.Sprintf("%s\n%s", msg, string(debug.Stack())))
-		getWriter().Severe(fmt.Sprintf("%s\n%s", msg, string(debug.Stack())))
-	}
 }
 
+// writeSlow writes v into slow log.
+// Not checking shallLog here is for performance consideration.
+// If we check shallLog here, the fmt.Sprint might be called even if the log level is not enabled.
+// The caller should check shallLog before calling this function.
 func writeSlow(val any, fields ...LogField) {
-	if shallLog(ErrorLevel) {
+	getWriter().Slow(val, addCaller(fields...)...)
-		getWriter().Slow(val, addCaller(fields...)...)
-	}
 }
 
+// writeStack writes v into stack log.
+// Not checking shallLog here is for performance consideration.
+// If we check shallLog here, the fmt.Sprint might be called even if the log level is not enabled.
+// The caller should check shallLog before calling this function.
 func writeStack(msg string) {
-	if shallLog(ErrorLevel) {
+	getWriter().Stack(fmt.Sprintf("%s\n%s", msg, string(debug.Stack())))
-		getWriter().Stack(fmt.Sprintf("%s\n%s", msg, string(debug.Stack())))
-	}
 }
 
+// writeStat writes v into stat log.
+// Not checking shallLog here is for performance consideration.
+// If we check shallLog here, the fmt.Sprint might be called even if the log level is not enabled.
+// The caller should check shallLog before calling this function.
 func writeStat(msg string) {
-	if shallLogStat() && shallLog(InfoLevel) {
+	getWriter().Stat(msg, addCaller()...)
-		getWriter().Stat(msg, addCaller()...)
-	}
 }

core/logx/logs_test.go

@@ -24,6 +24,10 @@ var (
 	_    Writer = (*mockWriter)(nil)
 )
 
+func init() {
+	ExitOnFatal.Set(false)
+}
+
 type mockWriter struct {
 	lock    sync.Mutex
 	builder strings.Builder
@@ -208,6 +212,12 @@ func TestFileLineConsoleMode(t *testing.T) {
 	assert.True(t, w.Contains(fmt.Sprintf("%s:%d", file, line+1)))
 }
 
+func TestMust(t *testing.T) {
+	assert.Panics(t, func() {
+		Must(errors.New("foo"))
+	})
+}
+
 func TestStructedLogAlert(t *testing.T) {
 	w := new(mockWriter)
 	old := writer.Swap(w)
@@ -574,26 +584,38 @@ func TestSetup(t *testing.T) {
 		atomic.StoreUint32(&encoding, jsonEncodingType)
 	}()
 
+	setupOnce = sync.Once{}
+	MustSetup(LogConf{
+		ServiceName: "any",
+		Mode:        "console",
+		Encoding:    "json",
+		TimeFormat:  timeFormat,
+	})
+	setupOnce = sync.Once{}
 	MustSetup(LogConf{
 		ServiceName: "any",
 		Mode:        "console",
 		TimeFormat:  timeFormat,
 	})
+	setupOnce = sync.Once{}
 	MustSetup(LogConf{
 		ServiceName: "any",
 		Mode:        "file",
 		Path:        os.TempDir(),
 	})
+	setupOnce = sync.Once{}
 	MustSetup(LogConf{
 		ServiceName: "any",
 		Mode:        "volume",
 		Path:        os.TempDir(),
 	})
+	setupOnce = sync.Once{}
 	MustSetup(LogConf{
 		ServiceName: "any",
 		Mode:        "console",
 		TimeFormat:  timeFormat,
 	})
+	setupOnce = sync.Once{}
 	MustSetup(LogConf{
 		ServiceName: "any",
 		Mode:        "console",

core/logx/logtest/logtest.go

@@ -0,0 +1,84 @@
+package logtest
+
+import (
+	"bytes"
+	"encoding/json"
+	"io"
+	"testing"
+
+	"github.com/zeromicro/go-zero/core/logx"
+)
+
+type Buffer struct {
+	buf *bytes.Buffer
+	t   *testing.T
+}
+
+func Discard(t *testing.T) {
+	prev := logx.Reset()
+	logx.SetWriter(logx.NewWriter(io.Discard))
+
+	t.Cleanup(func() {
+		logx.SetWriter(prev)
+	})
+}
+
+func NewCollector(t *testing.T) *Buffer {
+	var buf bytes.Buffer
+	writer := logx.NewWriter(&buf)
+	prev := logx.Reset()
+	logx.SetWriter(writer)
+
+	t.Cleanup(func() {
+		logx.SetWriter(prev)
+	})
+
+	return &Buffer{
+		buf: &buf,
+		t:   t,
+	}
+}
+
+func (b *Buffer) Bytes() []byte {
+	return b.buf.Bytes()
+}
+
+func (b *Buffer) Content() string {
+	var m map[string]interface{}
+	if err := json.Unmarshal(b.buf.Bytes(), &m); err != nil {
+		return ""
+	}
+
+	content, ok := m["content"]
+	if !ok {
+		return ""
+	}
+
+	switch val := content.(type) {
+	case string:
+		return val
+	default:
+		// err is impossible to be not nil, unmarshaled from b.buf.Bytes()
+		bs, _ := json.Marshal(content)
+		return string(bs)
+	}
+}
+
+func (b *Buffer) Reset() {
+	b.buf.Reset()
+}
+
+func (b *Buffer) String() string {
+	return b.buf.String()
+}
+
+func PanicOnFatal(t *testing.T) {
+	ok := logx.ExitOnFatal.CompareAndSwap(true, false)
+	if !ok {
+		return
+	}
+
+	t.Cleanup(func() {
+		logx.ExitOnFatal.CompareAndSwap(false, true)
+	})
+}

core/logx/logtest/logtest_test.go

@@ -0,0 +1,44 @@
+package logtest
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/zeromicro/go-zero/core/logx"
+)
+
+func TestCollector(t *testing.T) {
+	const input = "hello"
+	c := NewCollector(t)
+	logx.Info(input)
+	assert.Equal(t, input, c.Content())
+	assert.Contains(t, c.String(), input)
+	c.Reset()
+	assert.Empty(t, c.Bytes())
+}
+
+func TestPanicOnFatal(t *testing.T) {
+	const input = "hello"
+	Discard(t)
+	logx.Info(input)
+
+	PanicOnFatal(t)
+	PanicOnFatal(t)
+	assert.Panics(t, func() {
+		logx.Must(errors.New("foo"))
+	})
+}
+
+func TestCollectorContent(t *testing.T) {
+	const input = "hello"
+	c := NewCollector(t)
+	c.buf.WriteString(input)
+	assert.Empty(t, c.Content())
+	c.Reset()
+	c.buf.WriteString(`{}`)
+	assert.Empty(t, c.Content())
+	c.Reset()
+	c.buf.WriteString(`{"content":1}`)
+	assert.Equal(t, "1", c.Content())
+}

core/logx/richlogger.go

@@ -65,7 +65,7 @@ func (l *richLogger) Errorf(format string, v ...any) {
 }
 
 func (l *richLogger) Errorv(v any) {
-	l.err(fmt.Sprint(v))
+	l.err(v)
 }
 
 func (l *richLogger) Errorw(msg string, fields ...LogField) {

core/logx/richlogger_test.go

@@ -66,6 +66,9 @@ func TestTraceDebug(t *testing.T) {
 	l.WithDuration(time.Second).Debugv(testlog)
 	validate(t, w.String(), true, true)
 	w.Reset()
+	l.WithDuration(time.Second).Debugv(testobj)
+	validateContentType(t, w.String(), map[string]any{}, true, true)
+	w.Reset()
 	l.WithDuration(time.Second).Debugw(testlog, Field("foo", "bar"))
 	validate(t, w.String(), true, true)
 	assert.True(t, strings.Contains(w.String(), "foo"), w.String())
@@ -103,6 +106,9 @@ func TestTraceError(t *testing.T) {
 	l.WithDuration(time.Second).Errorv(testlog)
 	validate(t, w.String(), true, true)
 	w.Reset()
+	l.WithDuration(time.Second).Errorv(testobj)
+	validateContentType(t, w.String(), map[string]any{}, true, true)
+	w.Reset()
 	l.WithDuration(time.Second).Errorw(testlog, Field("basket", "ball"))
 	validate(t, w.String(), true, true)
 	assert.True(t, strings.Contains(w.String(), "basket"), w.String())
@@ -137,6 +143,9 @@ func TestTraceInfo(t *testing.T) {
 	l.WithDuration(time.Second).Infov(testlog)
 	validate(t, w.String(), true, true)
 	w.Reset()
+	l.WithDuration(time.Second).Infov(testobj)
+	validateContentType(t, w.String(), map[string]any{}, true, true)
+	w.Reset()
 	l.WithDuration(time.Second).Infow(testlog, Field("basket", "ball"))
 	validate(t, w.String(), true, true)
 	assert.True(t, strings.Contains(w.String(), "basket"), w.String())
@@ -173,6 +182,9 @@ func TestTraceInfoConsole(t *testing.T) {
 	w.Reset()
 	l.WithDuration(time.Second).Infov(testlog)
 	validate(t, w.String(), true, true)
+	w.Reset()
+	l.WithDuration(time.Second).Infov(testobj)
+	validateContentType(t, w.String(), map[string]any{}, true, true)
 }
 
 func TestTraceSlow(t *testing.T) {
@@ -204,6 +216,9 @@ func TestTraceSlow(t *testing.T) {
 	l.WithDuration(time.Second).Slowv(testlog)
 	validate(t, w.String(), true, true)
 	w.Reset()
+	l.WithDuration(time.Second).Slowv(testobj)
+	validateContentType(t, w.String(), map[string]any{}, true, true)
+	w.Reset()
 	l.WithDuration(time.Second).Sloww(testlog, Field("basket", "ball"))
 	validate(t, w.String(), true, true)
 	assert.True(t, strings.Contains(w.String(), "basket"), w.String())
@@ -311,8 +326,32 @@ func validate(t *testing.T, body string, expectedTrace, expectedSpan bool) {
 	assert.Equal(t, expectedSpan, len(val.Span) > 0, body)
 }
 
-type mockValue struct {
+func validateContentType(t *testing.T, body string, expectedType any, expectedTrace, expectedSpan bool) {
-	Trace string `json:"trace"`
+	var val mockValue
-	Span  string `json:"span"`
+	dec := json.NewDecoder(strings.NewReader(body))
-	Foo   string `json:"foo"`
+
+	for {
+		var doc mockValue
+		err := dec.Decode(&doc)
+		if err == io.EOF {
+			// all done
+			break
+		}
+		if err != nil {
+			continue
+		}
+
+		val = doc
+	}
+
+	assert.IsType(t, expectedType, val.Content, body)
+	assert.Equal(t, expectedTrace, len(val.Trace) > 0, body)
+	assert.Equal(t, expectedSpan, len(val.Span) > 0, body)
+}
+
+type mockValue struct {
+	Trace   string `json:"trace"`
+	Span    string `json:"span"`
+	Foo     string `json:"foo"`
+	Content any    `json:"content"`
 }

core/logx/rotatelogger.go

@@ -4,7 +4,6 @@ import (
 	"compress/gzip"
 	"errors"
 	"fmt"
-	"io"
 	"log"
 	"os"
 	"path"
@@ -237,7 +236,7 @@ func NewLogger(filename string, rule RotateRule, compress bool) (*RotateLogger,
 		rule:     rule,
 		compress: compress,
 	}
-	if err := l.init(); err != nil {
+	if err := l.initialize(); err != nil {
 		return nil, err
 	}
 
@@ -281,7 +280,7 @@ func (l *RotateLogger) getBackupFilename() string {
 	return l.backup
 }
 
-func (l *RotateLogger) init() error {
+func (l *RotateLogger) initialize() error {
 	l.backup = l.rule.BackupFileName()
 
 	if fileInfo, err := os.Stat(l.filename); err != nil {
@@ -299,6 +298,7 @@ func (l *RotateLogger) init() error {
 		if l.fp, err = os.OpenFile(l.filename, os.O_APPEND|os.O_WRONLY, defaultFileMode); err != nil {
 			return err
 		}
+
 		l.currentSize = fileInfo.Size()
 	}
 
@@ -382,7 +382,15 @@ func (l *RotateLogger) startWorker() {
 			case event := <-l.channel:
 				l.write(event)
 			case <-l.done:
-				return
+				// avoid losing logs before closing.
+				for {
+					select {
+					case event := <-l.channel:
+						l.write(event)
+					default:
+						return
+					}
+				}
 			}
 		}
 	}()
@@ -406,7 +414,7 @@ func (l *RotateLogger) write(v []byte) {
 func compressLogFile(file string) {
 	start := time.Now()
 	Infof("compressing log file: %s", file)
-	if err := gzipFile(file); err != nil {
+	if err := gzipFile(file, fileSys); err != nil {
 		Errorf("compress error: %s", err)
 	} else {
 		Infof("compressed log file: %s, took %s", file, time.Since(start))
@@ -421,25 +429,37 @@ func getNowDateInRFC3339Format() string {
 	return time.Now().Format(fileTimeFormat)
 }
 
-func gzipFile(file string) error {
+func gzipFile(file string, fsys fileSystem) (err error) {
-	in, err := os.Open(file)
+	in, err := fsys.Open(file)
 	if err != nil {
 		return err
 	}
-	defer in.Close()
+	defer func() {
+		if e := fsys.Close(in); e != nil {
+			Errorf("failed to close file: %s, error: %v", file, e)
+		}
+		if err == nil {
+			// only remove the original file when compression is successful
+			err = fsys.Remove(file)
+		}
+	}()
 
-	out, err := os.Create(fmt.Sprintf("%s%s", file, gzipExt))
+	out, err := fsys.Create(fmt.Sprintf("%s%s", file, gzipExt))
 	if err != nil {
 		return err
 	}
-	defer out.Close()
+	defer func() {
+		e := fsys.Close(out)
+		if err == nil {
+			err = e
+		}
+	}()
 
 	w := gzip.NewWriter(out)
-	if _, err = io.Copy(w, in); err != nil {
+	if _, err = fsys.Copy(w, in); err != nil {
-		return err
+		// failed to copy, no need to close w
-	} else if err = w.Close(); err != nil {
 		return err
 	}
 
-	return os.Remove(file)
+	return fsys.Close(w)
 }

core/logx/rotatelogger_test.go

@@ -1,8 +1,12 @@
 package logx
 
 import (
+	"errors"
+	"io"
 	"os"
+	"path"
 	"path/filepath"
+	"sync/atomic"
 	"syscall"
 	"testing"
 	"time"
@@ -13,18 +17,58 @@ import (
 )
 
 func TestDailyRotateRuleMarkRotated(t *testing.T) {
-	var rule DailyRotateRule
+	t.Run("daily rule", func(t *testing.T) {
-	rule.MarkRotated()
+		var rule DailyRotateRule
-	assert.Equal(t, getNowDate(), rule.rotatedTime)
+		rule.MarkRotated()
+		assert.Equal(t, getNowDate(), rule.rotatedTime)
+	})
+
+	t.Run("daily rule", func(t *testing.T) {
+		rule := DefaultRotateRule("test", "-", 1, false)
+		_, ok := rule.(*DailyRotateRule)
+		assert.True(t, ok)
+	})
 }
 
 func TestDailyRotateRuleOutdatedFiles(t *testing.T) {
-	var rule DailyRotateRule
+	t.Run("no files", func(t *testing.T) {
-	assert.Empty(t, rule.OutdatedFiles())
+		var rule DailyRotateRule
-	rule.days = 1
+		assert.Empty(t, rule.OutdatedFiles())
-	assert.Empty(t, rule.OutdatedFiles())
+		rule.days = 1
-	rule.gzip = true
+		assert.Empty(t, rule.OutdatedFiles())
-	assert.Empty(t, rule.OutdatedFiles())
+		rule.gzip = true
+		assert.Empty(t, rule.OutdatedFiles())
+	})
+
+	t.Run("bad files", func(t *testing.T) {
+		rule := DailyRotateRule{
+			filename: "[a-z",
+		}
+		assert.Empty(t, rule.OutdatedFiles())
+		rule.days = 1
+		assert.Empty(t, rule.OutdatedFiles())
+		rule.gzip = true
+		assert.Empty(t, rule.OutdatedFiles())
+	})
+
+	t.Run("temp files", func(t *testing.T) {
+		boundary := time.Now().Add(-time.Hour * time.Duration(hoursPerDay) * 2).Format(dateFormat)
+		f1, err := os.CreateTemp(os.TempDir(), "go-zero-test-"+boundary)
+		assert.NoError(t, err)
+		_ = f1.Close()
+		f2, err := os.CreateTemp(os.TempDir(), "go-zero-test-"+boundary)
+		assert.NoError(t, err)
+		_ = f2.Close()
+		t.Cleanup(func() {
+			_ = os.Remove(f1.Name())
+			_ = os.Remove(f2.Name())
+		})
+		rule := DailyRotateRule{
+			filename: path.Join(os.TempDir(), "go-zero-test-"),
+			days:     1,
+		}
+		assert.NotEmpty(t, rule.OutdatedFiles())
+	})
 }
 
 func TestDailyRotateRuleShallRotate(t *testing.T) {
@@ -34,20 +78,101 @@ func TestDailyRotateRuleShallRotate(t *testing.T) {
 }
 
 func TestSizeLimitRotateRuleMarkRotated(t *testing.T) {
-	var rule SizeLimitRotateRule
+	t.Run("size limit rule", func(t *testing.T) {
-	rule.MarkRotated()
+		var rule SizeLimitRotateRule
-	assert.Equal(t, getNowDateInRFC3339Format(), rule.rotatedTime)
+		rule.MarkRotated()
+		assert.Equal(t, getNowDateInRFC3339Format(), rule.rotatedTime)
+	})
+
+	t.Run("size limit rule", func(t *testing.T) {
+		rule := NewSizeLimitRotateRule("foo", "-", 1, 1, 1, false)
+		rule.MarkRotated()
+		assert.Equal(t, getNowDateInRFC3339Format(), rule.(*SizeLimitRotateRule).rotatedTime)
+	})
 }
 
 func TestSizeLimitRotateRuleOutdatedFiles(t *testing.T) {
-	var rule SizeLimitRotateRule
+	t.Run("no files", func(t *testing.T) {
-	assert.Empty(t, rule.OutdatedFiles())
+		var rule SizeLimitRotateRule
-	rule.days = 1
+		assert.Empty(t, rule.OutdatedFiles())
-	assert.Empty(t, rule.OutdatedFiles())
+		rule.days = 1
-	rule.gzip = true
+		assert.Empty(t, rule.OutdatedFiles())
-	assert.Empty(t, rule.OutdatedFiles())
+		rule.gzip = true
-	rule.maxBackups = 0
+		assert.Empty(t, rule.OutdatedFiles())
-	assert.Empty(t, rule.OutdatedFiles())
+		rule.maxBackups = 0
+		assert.Empty(t, rule.OutdatedFiles())
+	})
+
+	t.Run("bad files", func(t *testing.T) {
+		rule := SizeLimitRotateRule{
+			DailyRotateRule: DailyRotateRule{
+				filename: "[a-z",
+			},
+		}
+		assert.Empty(t, rule.OutdatedFiles())
+		rule.days = 1
+		assert.Empty(t, rule.OutdatedFiles())
+		rule.gzip = true
+		assert.Empty(t, rule.OutdatedFiles())
+	})
+
+	t.Run("temp files", func(t *testing.T) {
+		boundary := time.Now().Add(-time.Hour * time.Duration(hoursPerDay) * 2).Format(dateFormat)
+		f1, err := os.CreateTemp(os.TempDir(), "go-zero-test-"+boundary)
+		assert.NoError(t, err)
+		f2, err := os.CreateTemp(os.TempDir(), "go-zero-test-"+boundary)
+		assert.NoError(t, err)
+		boundary1 := time.Now().Add(time.Hour * time.Duration(hoursPerDay) * 2).Format(dateFormat)
+		f3, err := os.CreateTemp(os.TempDir(), "go-zero-test-"+boundary1)
+		assert.NoError(t, err)
+		t.Cleanup(func() {
+			_ = f1.Close()
+			_ = os.Remove(f1.Name())
+			_ = f2.Close()
+			_ = os.Remove(f2.Name())
+			_ = f3.Close()
+			_ = os.Remove(f3.Name())
+		})
+		rule := SizeLimitRotateRule{
+			DailyRotateRule: DailyRotateRule{
+				filename: path.Join(os.TempDir(), "go-zero-test-"),
+				days:     1,
+			},
+			maxBackups: 3,
+		}
+		assert.NotEmpty(t, rule.OutdatedFiles())
+	})
+
+	t.Run("no backups", func(t *testing.T) {
+		boundary := time.Now().Add(-time.Hour * time.Duration(hoursPerDay) * 2).Format(dateFormat)
+		f1, err := os.CreateTemp(os.TempDir(), "go-zero-test-"+boundary)
+		assert.NoError(t, err)
+		f2, err := os.CreateTemp(os.TempDir(), "go-zero-test-"+boundary)
+		assert.NoError(t, err)
+		boundary1 := time.Now().Add(time.Hour * time.Duration(hoursPerDay) * 2).Format(dateFormat)
+		f3, err := os.CreateTemp(os.TempDir(), "go-zero-test-"+boundary1)
+		assert.NoError(t, err)
+		t.Cleanup(func() {
+			_ = f1.Close()
+			_ = os.Remove(f1.Name())
+			_ = f2.Close()
+			_ = os.Remove(f2.Name())
+			_ = f3.Close()
+			_ = os.Remove(f3.Name())
+		})
+		rule := SizeLimitRotateRule{
+			DailyRotateRule: DailyRotateRule{
+				filename: path.Join(os.TempDir(), "go-zero-test-"),
+				days:     1,
+			},
+		}
+		assert.NotEmpty(t, rule.OutdatedFiles())
+
+		logger := new(RotateLogger)
+		logger.rule = &rule
+		logger.maybeDeleteOutdatedFiles()
+		assert.Empty(t, rule.OutdatedFiles())
+	})
 }
 
 func TestSizeLimitRotateRuleShallRotate(t *testing.T) {
@@ -61,14 +186,47 @@ func TestSizeLimitRotateRuleShallRotate(t *testing.T) {
 }
 
 func TestRotateLoggerClose(t *testing.T) {
-	filename, err := fs.TempFilenameWithText("foo")
+	t.Run("close", func(t *testing.T) {
-	assert.Nil(t, err)
+		filename, err := fs.TempFilenameWithText("foo")
-	if len(filename) > 0 {
+		assert.Nil(t, err)
-		defer os.Remove(filename)
+		if len(filename) > 0 {
-	}
+			defer os.Remove(filename)
-	logger, err := NewLogger(filename, new(DailyRotateRule), false)
+		}
-	assert.Nil(t, err)
+		logger, err := NewLogger(filename, new(DailyRotateRule), false)
-	assert.Nil(t, logger.Close())
+		assert.Nil(t, err)
+		_, err = logger.Write([]byte("foo"))
+		assert.Nil(t, err)
+		assert.Nil(t, logger.Close())
+	})
+
+	t.Run("close and write", func(t *testing.T) {
+		logger := new(RotateLogger)
+		logger.done = make(chan struct{})
+		close(logger.done)
+		_, err := logger.Write([]byte("foo"))
+		assert.ErrorIs(t, err, ErrLogFileClosed)
+	})
+
+	t.Run("close without losing logs", func(t *testing.T) {
+		text := "foo"
+		filename, err := fs.TempFilenameWithText(text)
+		assert.Nil(t, err)
+		if len(filename) > 0 {
+			defer os.Remove(filename)
+		}
+		logger, err := NewLogger(filename, new(DailyRotateRule), false)
+		assert.Nil(t, err)
+		msg := []byte("foo")
+		n := 100
+		for i := 0; i < n; i++ {
+			_, err = logger.Write(msg)
+			assert.Nil(t, err)
+		}
+		assert.Nil(t, logger.Close())
+		bs, err := os.ReadFile(filename)
+		assert.Nil(t, err)
+		assert.Equal(t, len(msg)*n+len(text), len(bs))
+	})
 }
 
 func TestRotateLoggerGetBackupFilename(t *testing.T) {
@@ -179,7 +337,7 @@ func TestRotateLoggerWithSizeLimitRotateRuleClose(t *testing.T) {
 	}
 	logger, err := NewLogger(filename, new(SizeLimitRotateRule), false)
 	assert.Nil(t, err)
-	assert.Nil(t, logger.Close())
+	_ = logger.Close()
 }
 
 func TestRotateLoggerGetBackupWithSizeLimitRotateRuleFilename(t *testing.T) {
@@ -295,6 +453,85 @@ func TestRotateLoggerWithSizeLimitRotateRuleWrite(t *testing.T) {
 	logger.write([]byte(`baz`))
 }
 
+func TestGzipFile(t *testing.T) {
+	err := errors.New("any error")
+
+	t.Run("gzip file open failed", func(t *testing.T) {
+		fsys := &fakeFileSystem{
+			openFn: func(name string) (*os.File, error) {
+				return nil, err
+			},
+		}
+		assert.ErrorIs(t, err, gzipFile("any", fsys))
+		assert.False(t, fsys.Removed())
+	})
+
+	t.Run("gzip file create failed", func(t *testing.T) {
+		fsys := &fakeFileSystem{
+			createFn: func(name string) (*os.File, error) {
+				return nil, err
+			},
+		}
+		assert.ErrorIs(t, err, gzipFile("any", fsys))
+		assert.False(t, fsys.Removed())
+	})
+
+	t.Run("gzip file copy failed", func(t *testing.T) {
+		fsys := &fakeFileSystem{
+			copyFn: func(writer io.Writer, reader io.Reader) (int64, error) {
+				return 0, err
+			},
+		}
+		assert.ErrorIs(t, err, gzipFile("any", fsys))
+		assert.False(t, fsys.Removed())
+	})
+
+	t.Run("gzip file last close failed", func(t *testing.T) {
+		var called int32
+		fsys := &fakeFileSystem{
+			closeFn: func(closer io.Closer) error {
+				if atomic.AddInt32(&called, 1) > 2 {
+					return err
+				}
+				return nil
+			},
+		}
+		assert.NoError(t, gzipFile("any", fsys))
+		assert.True(t, fsys.Removed())
+	})
+
+	t.Run("gzip file remove failed", func(t *testing.T) {
+		fsys := &fakeFileSystem{
+			removeFn: func(name string) error {
+				return err
+			},
+		}
+		assert.Error(t, err, gzipFile("any", fsys))
+		assert.True(t, fsys.Removed())
+	})
+
+	t.Run("gzip file everything ok", func(t *testing.T) {
+		fsys := &fakeFileSystem{}
+		assert.NoError(t, gzipFile("any", fsys))
+		assert.True(t, fsys.Removed())
+	})
+}
+
+func TestRotateLogger_WithExistingFile(t *testing.T) {
+	const body = "foo"
+	filename, err := fs.TempFilenameWithText(body)
+	assert.Nil(t, err)
+	if len(filename) > 0 {
+		defer os.Remove(filename)
+	}
+
+	rule := NewSizeLimitRotateRule(filename, "-", 1, 100, 3, false)
+	logger, err := NewLogger(filename, rule, false)
+	assert.Nil(t, err)
+	assert.Equal(t, int64(len(body)), logger.currentSize)
+	assert.Nil(t, logger.Close())
+}
+
 func BenchmarkRotateLogger(b *testing.B) {
 	filename := "./test.log"
 	filename2 := "./test2.log"
@@ -346,3 +583,53 @@ func BenchmarkRotateLogger(b *testing.B) {
 		}
 	})
 }
+
+type fakeFileSystem struct {
+	removed  int32
+	closeFn  func(closer io.Closer) error
+	copyFn   func(writer io.Writer, reader io.Reader) (int64, error)
+	createFn func(name string) (*os.File, error)
+	openFn   func(name string) (*os.File, error)
+	removeFn func(name string) error
+}
+
+func (f *fakeFileSystem) Close(closer io.Closer) error {
+	if f.closeFn != nil {
+		return f.closeFn(closer)
+	}
+	return nil
+}
+
+func (f *fakeFileSystem) Copy(writer io.Writer, reader io.Reader) (int64, error) {
+	if f.copyFn != nil {
+		return f.copyFn(writer, reader)
+	}
+	return 0, nil
+}
+
+func (f *fakeFileSystem) Create(name string) (*os.File, error) {
+	if f.createFn != nil {
+		return f.createFn(name)
+	}
+	return nil, nil
+}
+
+func (f *fakeFileSystem) Open(name string) (*os.File, error) {
+	if f.openFn != nil {
+		return f.openFn(name)
+	}
+	return nil, nil
+}
+
+func (f *fakeFileSystem) Remove(name string) error {
+	atomic.AddInt32(&f.removed, 1)
+
+	if f.removeFn != nil {
+		return f.removeFn(name)
+	}
+	return nil
+}
+
+func (f *fakeFileSystem) Removed() bool {
+	return atomic.LoadInt32(&f.removed) > 0
+}

core/logx/syslog_test.go

@@ -12,6 +12,8 @@ import (
 
 const testlog = "Stay hungry, stay foolish."
 
+var testobj = map[string]any{"foo": "bar"}
+
 func TestCollectSysLog(t *testing.T) {
 	CollectSysLog()
 	content := getContent(captureOutput(func() {

core/logx/vars.go

@@ -1,6 +1,10 @@
 package logx
 
-import "errors"
+import (
+	"errors"
+
+	"github.com/zeromicro/go-zero/core/syncx"
+)
 
 const (
 	// DebugLevel logs everything
@@ -61,6 +65,8 @@ var (
 	ErrLogPathNotSet = errors.New("log path must be set")
 	// ErrLogServiceNameNotSet is an error that indicates that the service name is not set.
 	ErrLogServiceNameNotSet = errors.New("log service name must be set")
+	// ExitOnFatal defines whether to exit on fatal errors, defined here to make it easier to test.
+	ExitOnFatal = syncx.ForAtomicBool(true)
 
 	truncatedField = Field(truncatedKey, true)
 )

core/logx/writer_test.go

@@ -97,6 +97,15 @@ func TestConsoleWriter(t *testing.T) {
 	w.(*concreteWriter).statLog = easyToCloseWriter{}
 }
 
+func TestNewFileWriter(t *testing.T) {
+	t.Run("access", func(t *testing.T) {
+		_, err := newFileWriter(LogConf{
+			Path: "/not-exists",
+		})
+		assert.Error(t, err)
+	})
+}
+
 func TestNopWriter(t *testing.T) {
 	assert.NotPanics(t, func() {
 		var w nopWriter

core/mapping/unmarshaler.go

@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"math"
 	"reflect"
 	"strconv"
 	"strings"
@@ -20,6 +21,7 @@ import (
 const (
 	defaultKeyName = "key"
 	delimiter      = '.'
+	ignoreKey      = "-"
 )
 
 var (
@@ -49,6 +51,7 @@ type (
 	unmarshalOptions struct {
 		fillDefault  bool
 		fromString   bool
+		opaqueKeys   bool
 		canonicalKey func(key string) string
 	}
 )
@@ -72,7 +75,11 @@ func UnmarshalKey(m map[string]any, v any) error {
 }
 
 // Unmarshal unmarshals m into v.
-func (u *Unmarshaler) Unmarshal(i any, v any) error {
+func (u *Unmarshaler) Unmarshal(i, v any) error {
+	return u.unmarshal(i, v, "")
+}
+
+func (u *Unmarshaler) unmarshal(i, v any, fullName string) error {
 	valueType := reflect.TypeOf(v)
 	if valueType.Kind() != reflect.Ptr {
 		return errValueNotSettable
@@ -85,13 +92,13 @@ func (u *Unmarshaler) Unmarshal(i any, v any) error {
 			return errTypeMismatch
 		}
 
-		return u.UnmarshalValuer(mapValuer(iv), v)
+		return u.unmarshalValuer(mapValuer(iv), v, fullName)
 	case []any:
 		if elemType.Kind() != reflect.Slice {
 			return errTypeMismatch
 		}
 
-		return u.fillSlice(elemType, reflect.ValueOf(v).Elem(), iv)
+		return u.fillSlice(elemType, reflect.ValueOf(v).Elem(), iv, fullName)
 	default:
 		return errUnsupportedType
 	}
@@ -99,17 +106,21 @@ func (u *Unmarshaler) Unmarshal(i any, v any) error {
 
 // UnmarshalValuer unmarshals m into v.
 func (u *Unmarshaler) UnmarshalValuer(m Valuer, v any) error {
-	return u.unmarshalWithFullName(simpleValuer{current: m}, v, "")
+	return u.unmarshalValuer(simpleValuer{current: m}, v, "")
 }
 
-func (u *Unmarshaler) fillMap(fieldType reflect.Type, value reflect.Value, mapValue any) error {
+func (u *Unmarshaler) unmarshalValuer(m Valuer, v any, fullName string) error {
+	return u.unmarshalWithFullName(simpleValuer{current: m}, v, fullName)
+}
+
+func (u *Unmarshaler) fillMap(fieldType reflect.Type, value reflect.Value, mapValue any, fullName string) error {
 	if !value.CanSet() {
 		return errValueNotSettable
 	}
 
 	fieldKeyType := fieldType.Key()
 	fieldElemType := fieldType.Elem()
-	targetValue, err := u.generateMap(fieldKeyType, fieldElemType, mapValue)
+	targetValue, err := u.generateMap(fieldKeyType, fieldElemType, mapValue, fullName)
 	if err != nil {
 		return err
 	}
@@ -143,19 +154,22 @@ func (u *Unmarshaler) fillMapFromString(value reflect.Value, mapValue any) error
 	return nil
 }
 
-func (u *Unmarshaler) fillSlice(fieldType reflect.Type, value reflect.Value, mapValue any) error {
+func (u *Unmarshaler) fillSlice(fieldType reflect.Type, value reflect.Value, mapValue any, fullName string) error {
 	if !value.CanSet() {
 		return errValueNotSettable
 	}
 
+	refValue := reflect.ValueOf(mapValue)
+	if refValue.Kind() != reflect.Slice {
+		return newTypeMismatchErrorWithHint(fullName, reflect.Slice.String(), refValue.Type().String())
+	}
+	if refValue.IsNil() {
+		return nil
+	}
+
 	baseType := fieldType.Elem()
 	dereffedBaseType := Deref(baseType)
 	dereffedBaseKind := dereffedBaseType.Kind()
-	refValue := reflect.ValueOf(mapValue)
-	if refValue.IsNil() {
-		return nil
-	}
-
 	conv := reflect.MakeSlice(reflect.SliceOf(baseType), refValue.Len(), refValue.Cap())
 	if refValue.Len() == 0 {
 		value.Set(conv)
@@ -170,20 +184,27 @@ func (u *Unmarshaler) fillSlice(fieldType reflect.Type, value reflect.Value, map
 		}
 
 		valid = true
+		sliceFullName := fmt.Sprintf("%s[%d]", fullName, i)
+
 		switch dereffedBaseKind {
 		case reflect.Struct:
 			target := reflect.New(dereffedBaseType)
-			if err := u.Unmarshal(ithValue.(map[string]any), target.Interface()); err != nil {
+			val, ok := ithValue.(map[string]any)
+			if !ok {
+				return errTypeMismatch
+			}
+
+			if err := u.unmarshal(val, target.Interface(), sliceFullName); err != nil {
 				return err
 			}
 
 			SetValue(fieldType.Elem(), conv.Index(i), target.Elem())
 		case reflect.Slice:
-			if err := u.fillSlice(dereffedBaseType, conv.Index(i), ithValue); err != nil {
+			if err := u.fillSlice(dereffedBaseType, conv.Index(i), ithValue, sliceFullName); err != nil {
 				return err
 			}
 		default:
-			if err := u.fillSliceValue(conv, i, dereffedBaseKind, ithValue); err != nil {
+			if err := u.fillSliceValue(conv, i, dereffedBaseKind, ithValue, sliceFullName); err != nil {
 				return err
 			}
 		}
@@ -197,7 +218,7 @@ func (u *Unmarshaler) fillSlice(fieldType reflect.Type, value reflect.Value, map
 }
 
 func (u *Unmarshaler) fillSliceFromString(fieldType reflect.Type, value reflect.Value,
-	mapValue any) error {
+	mapValue any, fullName string) error {
 	var slice []any
 	switch v := mapValue.(type) {
 	case fmt.Stringer:
@@ -217,7 +238,7 @@ func (u *Unmarshaler) fillSliceFromString(fieldType reflect.Type, value reflect.
 	conv := reflect.MakeSlice(reflect.SliceOf(baseFieldType), len(slice), cap(slice))
 
 	for i := 0; i < len(slice); i++ {
-		if err := u.fillSliceValue(conv, i, baseFieldKind, slice[i]); err != nil {
+		if err := u.fillSliceValue(conv, i, baseFieldKind, slice[i], fullName); err != nil {
 			return err
 		}
 	}
@@ -227,7 +248,7 @@ func (u *Unmarshaler) fillSliceFromString(fieldType reflect.Type, value reflect.
 }
 
 func (u *Unmarshaler) fillSliceValue(slice reflect.Value, index int,
-	baseKind reflect.Kind, value any) error {
+	baseKind reflect.Kind, value any, fullName string) error {
 	ithVal := slice.Index(index)
 	switch v := value.(type) {
 	case fmt.Stringer:
@@ -235,7 +256,7 @@ func (u *Unmarshaler) fillSliceValue(slice reflect.Value, index int,
 	case string:
 		return setValueFromString(baseKind, ithVal, v)
 	case map[string]any:
-		return u.fillMap(ithVal.Type(), ithVal, value)
+		return u.fillMap(ithVal.Type(), ithVal, value, fullName)
 	default:
 		// don't need to consider the difference between int, int8, int16, int32, int64,
 		// uint, uint8, uint16, uint32, uint64, because they're handled as json.Number.
@@ -261,7 +282,7 @@ func (u *Unmarshaler) fillSliceValue(slice reflect.Value, index int,
 }
 
 func (u *Unmarshaler) fillSliceWithDefault(derefedType reflect.Type, value reflect.Value,
-	defaultValue string) error {
+	defaultValue, fullName string) error {
 	baseFieldType := Deref(derefedType.Elem())
 	baseFieldKind := baseFieldType.Kind()
 	defaultCacheLock.Lock()
@@ -279,16 +300,20 @@ func (u *Unmarshaler) fillSliceWithDefault(derefedType reflect.Type, value refle
 		defaultCacheLock.Unlock()
 	}
 
-	return u.fillSlice(derefedType, value, slice)
+	return u.fillSlice(derefedType, value, slice, fullName)
 }
 
-func (u *Unmarshaler) generateMap(keyType, elemType reflect.Type, mapValue any) (reflect.Value, error) {
+func (u *Unmarshaler) generateMap(keyType, elemType reflect.Type, mapValue any, fullName string) (reflect.Value, error) {
 	mapType := reflect.MapOf(keyType, elemType)
 	valueType := reflect.TypeOf(mapValue)
 	if mapType == valueType {
 		return reflect.ValueOf(mapValue), nil
 	}
 
+	if keyType != valueType.Key() {
+		return emptyValue, errTypeMismatch
+	}
+
 	refValue := reflect.ValueOf(mapValue)
 	targetValue := reflect.MakeMapWithSize(mapType, refValue.Len())
 	dereffedElemType := Deref(elemType)
@@ -297,11 +322,12 @@ func (u *Unmarshaler) generateMap(keyType, elemType reflect.Type, mapValue any)
 	for _, key := range refValue.MapKeys() {
 		keythValue := refValue.MapIndex(key)
 		keythData := keythValue.Interface()
+		mapFullName := fmt.Sprintf("%s[%s]", fullName, key.String())
 
 		switch dereffedElemKind {
 		case reflect.Slice:
 			target := reflect.New(dereffedElemType)
-			if err := u.fillSlice(elemType, target.Elem(), keythData); err != nil {
+			if err := u.fillSlice(elemType, target.Elem(), keythData, mapFullName); err != nil {
 				return emptyValue, err
 			}
 
@@ -313,7 +339,7 @@ func (u *Unmarshaler) generateMap(keyType, elemType reflect.Type, mapValue any)
 			}
 
 			target := reflect.New(dereffedElemType)
-			if err := u.Unmarshal(keythMap, target.Interface()); err != nil {
+			if err := u.unmarshal(keythMap, target.Interface(), mapFullName); err != nil {
 				return emptyValue, err
 			}
 
@@ -324,7 +350,7 @@ func (u *Unmarshaler) generateMap(keyType, elemType reflect.Type, mapValue any)
 				return emptyValue, errTypeMismatch
 			}
 
-			innerValue, err := u.generateMap(elemType.Key(), elemType.Elem(), keythMap)
+			innerValue, err := u.generateMap(elemType.Key(), elemType.Elem(), keythMap, mapFullName)
 			if err != nil {
 				return emptyValue, err
 			}
@@ -343,7 +369,12 @@ func (u *Unmarshaler) generateMap(keyType, elemType reflect.Type, mapValue any)
 					return emptyValue, errTypeMismatch
 				}
 
-				targetValue.SetMapIndex(key, reflect.ValueOf(v))
+				val := reflect.ValueOf(v)
+				if !val.Type().AssignableTo(dereffedElemType) {
+					return emptyValue, errTypeMismatch
+				}
+
+				targetValue.SetMapIndex(key, val)
 			case json.Number:
 				target := reflect.New(dereffedElemType)
 				if err := setValueFromString(dereffedElemKind, target.Elem(), v.String()); err != nil {
@@ -408,6 +439,10 @@ func (u *Unmarshaler) processAnonymousField(field reflect.StructField, value ref
 		return err
 	}
 
+	if key == ignoreKey {
+		return nil
+	}
+
 	if options.optional() {
 		return u.processAnonymousFieldOptional(field, value, key, m, fullName)
 	}
@@ -466,7 +501,7 @@ func (u *Unmarshaler) processAnonymousStructFieldOptional(fieldType reflect.Type
 			return err
 		}
 
-		_, hasValue := getValue(m, fieldKey)
+		_, hasValue := getValue(m, fieldKey, u.opts.opaqueKeys)
 		if hasValue {
 			if !filled {
 				filled = true
@@ -486,7 +521,7 @@ func (u *Unmarshaler) processAnonymousStructFieldOptional(fieldType reflect.Type
 	}
 
 	if filled && required != requiredFilled {
-		return fmt.Errorf("%s is not fully set", key)
+		return fmt.Errorf("%q is not fully set", key)
 	}
 
 	return nil
@@ -509,8 +544,8 @@ func (u *Unmarshaler) processFieldNotFromString(fieldType reflect.Type, value re
 	vp valueWithParent, opts *fieldOptionsWithContext, fullName string) error {
 	derefedFieldType := Deref(fieldType)
 	typeKind := derefedFieldType.Kind()
-	valueKind := reflect.TypeOf(vp.value).Kind()
 	mapValue := vp.value
+	valueKind := reflect.TypeOf(mapValue).Kind()
 
 	switch {
 	case valueKind == reflect.Map && typeKind == reflect.Struct:
@@ -523,12 +558,14 @@ func (u *Unmarshaler) processFieldNotFromString(fieldType reflect.Type, value re
 			current: mapValuer(mv),
 			parent:  vp.parent,
 		}, fullName)
+	case typeKind == reflect.Slice && valueKind == reflect.Slice:
+		return u.fillSlice(fieldType, value, mapValue, fullName)
 	case valueKind == reflect.Map && typeKind == reflect.Map:
-		return u.fillMap(fieldType, value, mapValue)
+		return u.fillMap(fieldType, value, mapValue, fullName)
 	case valueKind == reflect.String && typeKind == reflect.Map:
 		return u.fillMapFromString(value, mapValue)
 	case valueKind == reflect.String && typeKind == reflect.Slice:
-		return u.fillSliceFromString(fieldType, value, mapValue)
+		return u.fillSliceFromString(fieldType, value, mapValue, fullName)
 	case valueKind == reflect.String && derefedFieldType == durationType:
 		return fillDurationValue(fieldType, value, mapValue.(string))
 	default:
@@ -541,23 +578,16 @@ func (u *Unmarshaler) processFieldPrimitive(fieldType reflect.Type, value reflec
 	typeKind := Deref(fieldType).Kind()
 	valueKind := reflect.TypeOf(mapValue).Kind()
 
-	switch {
+	switch v := mapValue.(type) {
-	case typeKind == reflect.Slice && valueKind == reflect.Slice:
+	case json.Number:
-		return u.fillSlice(fieldType, value, mapValue)
+		return u.processFieldPrimitiveWithJSONNumber(fieldType, value, v, opts, fullName)
-	case typeKind == reflect.Map && valueKind == reflect.Map:
-		return u.fillMap(fieldType, value, mapValue)
 	default:
-		switch v := mapValue.(type) {
+		if typeKind == valueKind {
-		case json.Number:
+			if err := validateValueInOptions(mapValue, opts.options()); err != nil {
-			return u.processFieldPrimitiveWithJSONNumber(fieldType, value, v, opts, fullName)
+				return err
-		default:
-			if typeKind == valueKind {
-				if err := validateValueInOptions(mapValue, opts.options()); err != nil {
-					return err
-				}
-
-				return fillWithSameType(fieldType, value, mapValue, opts)
 			}
+
+			return fillWithSameType(fieldType, value, mapValue, opts)
 		}
 	}
 
@@ -580,25 +610,23 @@ func (u *Unmarshaler) processFieldPrimitiveWithJSONNumber(fieldType reflect.Type
 	target := reflect.New(Deref(fieldType)).Elem()
 
 	switch typeKind {
-	case reflect.Int, reflect.Int8, reflect.Int16, reflect.Int32, reflect.Int64:
+	case reflect.Int, reflect.Int8, reflect.Int16, reflect.Int32, reflect.Int64,
-		iValue, err := v.Int64()
+		reflect.Uint, reflect.Uint8, reflect.Uint16, reflect.Uint32, reflect.Uint64:
+		if err := setValueFromString(typeKind, target, v.String()); err != nil {
+			return err
+		}
+	case reflect.Float32:
+		fValue, err := v.Float64()
 		if err != nil {
 			return err
 		}
 
-		target.SetInt(iValue)
+		if fValue > math.MaxFloat32 {
-	case reflect.Uint, reflect.Uint8, reflect.Uint16, reflect.Uint32, reflect.Uint64:
+			return float32OverflowError(v.String())
-		iValue, err := v.Int64()
-		if err != nil {
-			return err
 		}
 
-		if iValue < 0 {
+		target.SetFloat(fValue)
-			return fmt.Errorf("unmarshal %q with bad value %q", fullName, v.String())
+	case reflect.Float64:
-		}
-
-		target.SetUint(uint64(iValue))
-	case reflect.Float32, reflect.Float64:
 		fValue, err := v.Float64()
 		if err != nil {
 			return err
@@ -606,7 +634,7 @@ func (u *Unmarshaler) processFieldPrimitiveWithJSONNumber(fieldType reflect.Type
 
 		target.SetFloat(fValue)
 	default:
-		return newTypeMismatchError(fullName)
+		return newTypeMismatchErrorWithHint(fullName, typeKind.String(), value.Type().String())
 	}
 
 	SetValue(fieldType, value, target)
@@ -691,11 +719,19 @@ func (u *Unmarshaler) processFieldWithEnvValue(fieldType reflect.Type, value ref
 
 func (u *Unmarshaler) processNamedField(field reflect.StructField, value reflect.Value,
 	m valuerWithParent, fullName string) error {
+	if !field.IsExported() {
+		return nil
+	}
+
 	key, opts, err := u.parseOptionsWithContext(field, m, fullName)
 	if err != nil {
 		return err
 	}
 
+	if key == ignoreKey {
+		return nil
+	}
+
 	fullName = join(fullName, key)
 	if opts != nil && len(opts.EnvVar) > 0 {
 		envVal := proc.Env(opts.EnvVar)
@@ -710,12 +746,12 @@ func (u *Unmarshaler) processNamedField(field reflect.StructField, value reflect
 	}
 
 	valuer := createValuer(m, opts)
-	mapValue, hasValue := getValue(valuer, canonicalKey)
+	mapValue, hasValue := getValue(valuer, canonicalKey, u.opts.opaqueKeys)
 
 	// When fillDefault is used, m is a null value, hasValue must be false, all priority judgments fillDefault.
 	if u.opts.fillDefault {
 		if !value.IsZero() {
-			return fmt.Errorf("set the default value, %s must be zero", fullName)
+			return fmt.Errorf("set the default value, %q must be zero", fullName)
 		}
 		return u.processNamedFieldWithoutValue(field.Type, value, opts, fullName)
 	} else if !hasValue {
@@ -736,11 +772,11 @@ func (u *Unmarshaler) processNamedFieldWithValue(fieldType reflect.Type, value r
 			return nil
 		}
 
-		return fmt.Errorf("field %s mustn't be nil", key)
+		return fmt.Errorf("field %q mustn't be nil", key)
 	}
 
 	if !value.CanSet() {
-		return fmt.Errorf("field %s is not settable", key)
+		return fmt.Errorf("field %q is not settable", key)
 	}
 
 	maybeNewValue(fieldType, value)
@@ -784,7 +820,7 @@ func (u *Unmarshaler) processNamedFieldWithValueFromString(fieldType reflect.Typ
 		}
 
 		if !stringx.Contains(options, checkValue) {
-			return fmt.Errorf(`value "%s" for field "%s" is not defined in options "%v"`,
+			return fmt.Errorf(`value "%s" for field %q is not defined in options "%v"`,
 				mapValue, key, options)
 		}
 	}
@@ -803,13 +839,18 @@ func (u *Unmarshaler) processNamedFieldWithoutValue(fieldType reflect.Type, valu
 
 		switch fieldKind {
 		case reflect.Array, reflect.Slice:
-			return u.fillSliceWithDefault(derefedType, value, defaultValue)
+			return u.fillSliceWithDefault(derefedType, value, defaultValue, fullName)
 		default:
 			return setValueFromString(fieldKind, value, defaultValue)
 		}
 	}
 
 	if u.opts.fillDefault {
+		if fieldType.Kind() != reflect.Ptr && fieldKind == reflect.Struct {
+			return u.processFieldNotFromString(fieldType, value, valueWithParent{
+				value: emptyMap,
+			}, opts, fullName)
+		}
 		return nil
 	}
 
@@ -846,7 +887,7 @@ func (u *Unmarshaler) processNamedFieldWithoutValue(fieldType reflect.Type, valu
 
 func (u *Unmarshaler) unmarshalWithFullName(m valuerWithParent, v any, fullName string) error {
 	rv := reflect.ValueOf(v)
-	if err := ValidatePtr(&rv); err != nil {
+	if err := ValidatePtr(rv); err != nil {
 		return err
 	}
 
@@ -865,12 +906,9 @@ func (u *Unmarshaler) unmarshalWithFullName(m valuerWithParent, v any, fullName
 
 	numFields := baseType.NumField()
 	for i := 0; i < numFields; i++ {
-		field := baseType.Field(i)
+		typeField := baseType.Field(i)
-		if !field.IsExported() {
+		valueField := valElem.Field(i)
-			continue
+		if err := u.processField(typeField, valueField, m, fullName); err != nil {
-		}
-
-		if err := u.processField(field, valElem.Field(i), m, fullName); err != nil {
 			return err
 		}
 	}
@@ -899,6 +937,14 @@ func WithDefault() UnmarshalOption {
 	}
 }
 
+// WithOpaqueKeys customizes an Unmarshaler with opaque keys.
+// Opaque keys are keys that are not processed by the unmarshaler.
+func WithOpaqueKeys() UnmarshalOption {
+	return func(opt *unmarshalOptions) {
+		opt.opaqueKeys = true
+	}
+}
+
 func createValuer(v valuerWithParent, opts *fieldOptionsWithContext) valuerWithParent {
 	if opts.inherit() {
 		return recursiveValuer{
@@ -976,8 +1022,8 @@ func fillWithSameType(fieldType reflect.Type, value reflect.Value, mapValue any,
 }
 
 // getValue gets the value for the specific key, the key can be in the format of parentKey.childKey
-func getValue(m valuerWithParent, key string) (any, bool) {
+func getValue(m valuerWithParent, key string, opaque bool) (any, bool) {
-	keys := readKeys(key)
+	keys := readKeys(key, opaque)
 	return getValueWithChainedKeys(m, keys)
 }
 
@@ -1024,14 +1070,23 @@ func join(elem ...string) string {
 }
 
 func newInitError(name string) error {
-	return fmt.Errorf("field %s is not set", name)
+	return fmt.Errorf("field %q is not set", name)
 }
 
 func newTypeMismatchError(name string) error {
-	return fmt.Errorf("type mismatch for field %s", name)
+	return fmt.Errorf("type mismatch for field %q", name)
 }
 
-func readKeys(key string) []string {
+func newTypeMismatchErrorWithHint(name, expectType, actualType string) error {
+	return fmt.Errorf("type mismatch for field %q, expect %q, actual %q",
+		name, expectType, actualType)
+}
+
+func readKeys(key string, opaque bool) []string {
+	if opaque {
+		return []string{key}
+	}
+
 	cacheKeysLock.Lock()
 	keys, ok := cacheKeys[key]
 	cacheKeysLock.Unlock()

core/mapping/utils.go

@@ -42,6 +42,10 @@ var (
 )
 
 type (
+	integer interface {
+		~int | ~int8 | ~int16 | ~int32 | ~int64 | ~uint | ~uint8 | ~uint16 | ~uint32 | ~uint64
+	}
+
 	optionsCacheValue struct {
 		key     string
 		options *fieldOptions
@@ -79,7 +83,7 @@ func SetMapIndexValue(tp reflect.Type, value, key, target reflect.Value) {
 }
 
 // ValidatePtr validates v if it's a valid pointer.
-func ValidatePtr(v *reflect.Value) error {
+func ValidatePtr(v reflect.Value) error {
 	// sequence is very important, IsNil must be called after checking Kind() with reflect.Ptr,
 	// panic otherwise
 	if !v.IsValid() || v.Kind() != reflect.Ptr || v.IsNil() {
@@ -103,21 +107,32 @@ func convertTypeFromString(kind reflect.Kind, str string) (any, error) {
 	case reflect.Int, reflect.Int8, reflect.Int16, reflect.Int32, reflect.Int64:
 		intValue, err := strconv.ParseInt(str, 10, 64)
 		if err != nil {
-			return 0, fmt.Errorf("the value %q cannot parsed as int", str)
+			return 0, err
 		}
 
 		return intValue, nil
 	case reflect.Uint, reflect.Uint8, reflect.Uint16, reflect.Uint32, reflect.Uint64:
 		uintValue, err := strconv.ParseUint(str, 10, 64)
 		if err != nil {
-			return 0, fmt.Errorf("the value %q cannot parsed as uint", str)
+			return 0, err
 		}
 
 		return uintValue, nil
-	case reflect.Float32, reflect.Float64:
+	case reflect.Float32:
 		floatValue, err := strconv.ParseFloat(str, 64)
 		if err != nil {
-			return 0, fmt.Errorf("the value %q cannot parsed as float", str)
+			return 0, err
+		}
+
+		if floatValue > math.MaxFloat32 {
+			return 0, float32OverflowError(str)
+		}
+
+		return floatValue, nil
+	case reflect.Float64:
+		floatValue, err := strconv.ParseFloat(str, 64)
+		if err != nil {
+			return 0, err
 		}
 
 		return floatValue, nil
@@ -215,6 +230,10 @@ func implicitValueRequiredStruct(tag string, tp reflect.Type) (bool, error) {
 	return false, nil
 }
 
+func intOverflowError[T integer](v T, kind reflect.Kind) error {
+	return fmt.Errorf("parsing \"%d\" as %s: value out of range", v, kind.String())
+}
+
 func isLeftInclude(b byte) (bool, error) {
 	switch b {
 	case '[':
@@ -237,6 +256,10 @@ func isRightInclude(b byte) (bool, error) {
 	}
 }
 
+func float32OverflowError(str string) error {
+	return fmt.Errorf("parsing %q as float32: value out of range", str)
+}
+
 func maybeNewValue(fieldType reflect.Type, value reflect.Value) {
 	if fieldType.Kind() == reflect.Ptr && value.IsNil() {
 		value.Set(reflect.New(value.Type().Elem()))
@@ -255,7 +278,7 @@ func parseGroupedSegments(val string) []string {
 
 // don't modify returned fieldOptions, it's cached and shared among different calls.
 func parseKeyAndOptions(tagName string, field reflect.StructField) (string, *fieldOptions, error) {
-	value := field.Tag.Get(tagName)
+	value := strings.TrimSpace(field.Tag.Get(tagName))
 	if len(value) == 0 {
 		return field.Name, nil, nil
 	}
@@ -370,10 +393,8 @@ func parseOption(fieldOpts *fieldOptions, fieldName, option string) error {
 			fieldOpts.Optional = true
 			fieldOpts.OptionalDep = segs[1]
 		default:
-			return fmt.Errorf("field %s has wrong optional", fieldName)
+			return fmt.Errorf("field %q has wrong optional", fieldName)
 		}
-	case option == optionalOption:
-		fieldOpts.Optional = true
 	case strings.HasPrefix(option, optionsOption):
 		val, err := parseProperty(fieldName, optionsOption, option)
 		if err != nil {
@@ -429,7 +450,7 @@ func parseOptions(val string) []string {
 func parseProperty(field, tag, val string) (string, error) {
 	segs := strings.Split(val, equalToken)
 	if len(segs) != 2 {
-		return "", fmt.Errorf("field %s has wrong %s", field, tag)
+		return "", fmt.Errorf("field %q has wrong tag value %q", field, tag)
 	}
 
 	return strings.TrimSpace(segs[1]), nil
@@ -484,22 +505,61 @@ func parseSegments(val string) []string {
 	return segments
 }
 
+func setIntValue(value reflect.Value, v any, min, max int64) error {
+	iv := v.(int64)
+	if iv < min || iv > max {
+		return intOverflowError(iv, value.Kind())
+	}
+
+	value.SetInt(iv)
+	return nil
+}
+
 func setMatchedPrimitiveValue(kind reflect.Kind, value reflect.Value, v any) error {
 	switch kind {
 	case reflect.Bool:
 		value.SetBool(v.(bool))
-	case reflect.Int, reflect.Int8, reflect.Int16, reflect.Int32, reflect.Int64:
+		return nil
+	case reflect.Int: // int depends on int size, 32 or 64
+		return setIntValue(value, v, math.MinInt, math.MaxInt)
+	case reflect.Int8:
+		return setIntValue(value, v, math.MinInt8, math.MaxInt8)
+	case reflect.Int16:
+		return setIntValue(value, v, math.MinInt16, math.MaxInt16)
+	case reflect.Int32:
+		return setIntValue(value, v, math.MinInt32, math.MaxInt32)
+	case reflect.Int64:
 		value.SetInt(v.(int64))
-	case reflect.Uint, reflect.Uint8, reflect.Uint16, reflect.Uint32, reflect.Uint64:
+		return nil
+	case reflect.Uint: // uint depends on int size, 32 or 64
+		return setUintValue(value, v, math.MaxUint)
+	case reflect.Uint8:
+		return setUintValue(value, v, math.MaxUint8)
+	case reflect.Uint16:
+		return setUintValue(value, v, math.MaxUint16)
+	case reflect.Uint32:
+		return setUintValue(value, v, math.MaxUint32)
+	case reflect.Uint64:
 		value.SetUint(v.(uint64))
+		return nil
 	case reflect.Float32, reflect.Float64:
 		value.SetFloat(v.(float64))
+		return nil
 	case reflect.String:
 		value.SetString(v.(string))
+		return nil
 	default:
 		return errUnsupportedType
 	}
+}
 
+func setUintValue(value reflect.Value, v any, boundary uint64) error {
+	iv := v.(uint64)
+	if iv > boundary {
+		return intOverflowError(iv, value.Kind())
+	}
+
+	value.SetUint(iv)
 	return nil
 }
 
@@ -577,7 +637,8 @@ func usingDifferentKeys(key string, field reflect.StructField) bool {
 	return false
 }
 
-func validateAndSetValue(kind reflect.Kind, value reflect.Value, str string, opts *fieldOptionsWithContext) error {
+func validateAndSetValue(kind reflect.Kind, value reflect.Value, str string,
+	opts *fieldOptionsWithContext) error {
 	if !value.CanSet() {
 		return errValueNotSettable
 	}
@@ -628,7 +689,7 @@ func validateValueInOptions(val any, options []string) error {
 		switch v := val.(type) {
 		case string:
 			if !stringx.Contains(options, v) {
-				return fmt.Errorf(`error: value "%s" is not defined in options "%v"`, v, options)
+				return fmt.Errorf(`error: value %q is not defined in options "%v"`, v, options)
 			}
 		default:
 			if !stringx.Contains(options, Repr(v)) {

core/mapping/utils_test.go

@@ -144,6 +144,10 @@ func TestParseSegments(t *testing.T) {
 			input:  "",
 			expect: []string{},
 		},
+		{
+			input:  "   ",
+			expect: []string{},
+		},
 		{
 			input:  ",",
 			expect: []string{""},
@@ -214,30 +218,31 @@ func TestParseSegments(t *testing.T) {
 func TestValidatePtrWithNonPtr(t *testing.T) {
 	var foo string
 	rve := reflect.ValueOf(foo)
-	assert.NotNil(t, ValidatePtr(&rve))
+	assert.NotNil(t, ValidatePtr(rve))
 }
 
 func TestValidatePtrWithPtr(t *testing.T) {
 	var foo string
 	rve := reflect.ValueOf(&foo)
-	assert.Nil(t, ValidatePtr(&rve))
+	assert.Nil(t, ValidatePtr(rve))
 }
 
 func TestValidatePtrWithNilPtr(t *testing.T) {
 	var foo *string
 	rve := reflect.ValueOf(foo)
-	assert.NotNil(t, ValidatePtr(&rve))
+	assert.NotNil(t, ValidatePtr(rve))
 }
 
 func TestValidatePtrWithZeroValue(t *testing.T) {
 	var s string
 	e := reflect.Zero(reflect.TypeOf(s))
-	assert.NotNil(t, ValidatePtr(&e))
+	assert.NotNil(t, ValidatePtr(e))
 }
 
 func TestSetValueNotSettable(t *testing.T) {
 	var i int
-	assert.NotNil(t, setValueFromString(reflect.Int, reflect.ValueOf(i), "1"))
+	assert.Error(t, setValueFromString(reflect.Int, reflect.ValueOf(i), "1"))
+	assert.Error(t, validateAndSetValue(reflect.Int, reflect.ValueOf(i), "1", nil))
 }
 
 func TestParseKeyAndOptionsErrors(t *testing.T) {
@@ -296,3 +301,36 @@ func TestSetValueFormatErrors(t *testing.T) {
 		})
 	}
 }
+
+func TestValidateValueRange(t *testing.T) {
+	t.Run("float", func(t *testing.T) {
+		assert.NoError(t, validateValueRange(1.2, nil))
+	})
+
+	t.Run("float number range", func(t *testing.T) {
+		assert.NoError(t, validateNumberRange(1.2, nil))
+	})
+
+	t.Run("bad float", func(t *testing.T) {
+		assert.Error(t, validateValueRange("a", &fieldOptionsWithContext{
+			Range: &numberRange{},
+		}))
+	})
+
+	t.Run("bad float validate", func(t *testing.T) {
+		var v struct {
+			Foo float32
+		}
+		assert.Error(t, validateAndSetValue(reflect.Int, reflect.ValueOf(&v).Elem().Field(0),
+			"1", &fieldOptionsWithContext{
+				Range: &numberRange{
+					left:  2,
+					right: 3,
+				},
+			}))
+	})
+}
+
+func TestSetMatchedPrimitiveValue(t *testing.T) {
+	assert.Error(t, setMatchedPrimitiveValue(reflect.Func, reflect.ValueOf(2), "1"))
+}

core/mapping/valuer.go

@@ -34,7 +34,7 @@ type (
 	recursiveValuer node
 )
 
-// Value gets the value assciated with the given key from mv.
+// Value gets the value associated with the given key from mv.
 func (mv mapValuer) Value(key string) (any, bool) {
 	v, ok := mv[key]
 	return v, ok

core/metric/histogram_test.go

@@ -6,7 +6,6 @@ import (
 
 	"github.com/prometheus/client_golang/prometheus/testutil"
 	"github.com/stretchr/testify/assert"
-	"github.com/zeromicro/go-zero/core/proc"
 )
 
 func TestNewHistogramVec(t *testing.T) {
@@ -48,6 +47,4 @@ func TestHistogramObserve(t *testing.T) {
 
 	err := testutil.CollectAndCompare(hv.histogram, strings.NewReader(metadata+val))
 	assert.Nil(t, err)
-
-	proc.Shutdown()
 }

core/metric/summary.go

@@ -0,0 +1,65 @@
+package metric
+
+import (
+	prom "github.com/prometheus/client_golang/prometheus"
+	"github.com/zeromicro/go-zero/core/proc"
+	"github.com/zeromicro/go-zero/core/prometheus"
+)
+
+type (
+	// A SummaryVecOpts is a summary vector options
+	SummaryVecOpts struct {
+		VecOpt     VectorOpts
+		Objectives map[float64]float64
+	}
+
+	// A SummaryVec interface represents a summary vector.
+	SummaryVec interface {
+		// Observe adds observation v to labels.
+		Observe(v float64, labels ...string)
+		close() bool
+	}
+
+	promSummaryVec struct {
+		summary *prom.SummaryVec
+	}
+)
+
+// NewSummaryVec return a SummaryVec
+func NewSummaryVec(cfg *SummaryVecOpts) SummaryVec {
+	if cfg == nil {
+		return nil
+	}
+
+	vec := prom.NewSummaryVec(
+		prom.SummaryOpts{
+			Namespace:  cfg.VecOpt.Namespace,
+			Subsystem:  cfg.VecOpt.Subsystem,
+			Name:       cfg.VecOpt.Name,
+			Help:       cfg.VecOpt.Help,
+			Objectives: cfg.Objectives,
+		},
+		cfg.VecOpt.Labels,
+	)
+	prom.MustRegister(vec)
+	sv := &promSummaryVec{
+		summary: vec,
+	}
+	proc.AddShutdownListener(func() {
+		sv.close()
+	})
+
+	return sv
+}
+
+func (sv *promSummaryVec) Observe(v float64, labels ...string) {
+	if !prometheus.Enabled() {
+		return
+	}
+
+	sv.summary.WithLabelValues(labels...).Observe(v)
+}
+
+func (sv *promSummaryVec) close() bool {
+	return prom.Unregister(sv.summary)
+}

core/metric/summary_test.go

@@ -0,0 +1,68 @@
+package metric
+
+import (
+	"strings"
+	"testing"
+
+	"github.com/prometheus/client_golang/prometheus/testutil"
+	"github.com/stretchr/testify/assert"
+	"github.com/zeromicro/go-zero/core/proc"
+)
+
+func TestNewSummaryVec(t *testing.T) {
+	summaryVec := NewSummaryVec(&SummaryVecOpts{
+		VecOpt: VectorOpts{
+			Namespace: "http_server",
+			Subsystem: "requests",
+			Name:      "duration_quantiles",
+			Help:      "rpc client requests duration(ms) φ quantiles ",
+			Labels:    []string{"method"},
+		},
+		Objectives: map[float64]float64{
+			0.5: 0.01,
+			0.9: 0.01,
+		},
+	})
+	defer summaryVec.close()
+	summaryVecNil := NewSummaryVec(nil)
+	assert.NotNil(t, summaryVec)
+	assert.Nil(t, summaryVecNil)
+}
+
+func TestSummaryObserve(t *testing.T) {
+	startAgent()
+	summaryVec := NewSummaryVec(&SummaryVecOpts{
+		VecOpt: VectorOpts{
+			Namespace: "http_server",
+			Subsystem: "requests",
+			Name:      "duration_quantiles",
+			Help:      "rpc client requests duration(ms) φ quantiles ",
+			Labels:    []string{"method"},
+		},
+		Objectives: map[float64]float64{
+			0.3: 0.01,
+			0.6: 0.01,
+			1:   0.01,
+		},
+	})
+	defer summaryVec.close()
+	sv := summaryVec.(*promSummaryVec)
+	sv.Observe(100, "GET")
+	sv.Observe(200, "GET")
+	sv.Observe(300, "GET")
+	metadata := `
+		# HELP http_server_requests_duration_quantiles rpc client requests duration(ms) φ quantiles 
+		# TYPE http_server_requests_duration_quantiles summary
+`
+	val := `
+		http_server_requests_duration_quantiles{method="GET",quantile="0.3"} 100
+		http_server_requests_duration_quantiles{method="GET",quantile="0.6"} 200
+		http_server_requests_duration_quantiles{method="GET",quantile="1"} 300
+		http_server_requests_duration_quantiles_sum{method="GET"} 600
+		http_server_requests_duration_quantiles_count{method="GET"} 3
+`
+
+	err := testutil.CollectAndCompare(sv.summary, strings.NewReader(metadata+val))
+	assert.Nil(t, err)
+	proc.Shutdown()
+}

core/mr/mapreduce_fuzzcase_test.go

@@ -1,5 +1,4 @@
 //go:build fuzz
-// +build fuzz
 
 package mr
 

core/mr/mapreduce_test.go

@@ -3,7 +3,7 @@ package mr
 import (
 	"context"
 	"errors"
-	"io/ioutil"
+	"io"
 	"log"
 	"runtime"
 	"sync/atomic"
@@ -17,7 +17,7 @@ import (
 var errDummy = errors.New("dummy")
 
 func init() {
-	log.SetOutput(ioutil.Discard)
+	log.SetOutput(io.Discard)
 }
 
 func TestFinish(t *testing.T) {
@@ -574,6 +574,7 @@ func TestMapReduceWithContext(t *testing.T) {
 			cancel()
 		}
 		writer.Write(i)
+		time.Sleep(time.Millisecond)
 	}, func(pipe <-chan int, cancel func(error)) {
 		for item := range pipe {
 			i := item

core/proc/env_test.go

@@ -1,7 +1,6 @@
 package proc
 
 import (
-	"os"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -21,13 +20,11 @@ func TestEnvInt(t *testing.T) {
 	val, ok := EnvInt("any")
 	assert.Equal(t, 0, val)
 	assert.False(t, ok)
-	err := os.Setenv("anyInt", "10")
+	t.Setenv("anyInt", "10")
-	assert.Nil(t, err)
 	val, ok = EnvInt("anyInt")
 	assert.Equal(t, 10, val)
 	assert.True(t, ok)
-	err = os.Setenv("anyString", "a")
+	t.Setenv("anyString", "a")
-	assert.Nil(t, err)
 	val, ok = EnvInt("anyString")
 	assert.Equal(t, 0, val)
 	assert.False(t, ok)

core/proc/goroutines+polyfill.go

@@ -1,7 +0,0 @@
-//go:build windows
-// +build windows
-
-package proc
-
-func dumpGoroutines() {
-}

core/proc/goroutines.go

@@ -1,5 +1,4 @@
 //go:build linux || darwin
-// +build linux darwin
 
 package proc
 
@@ -19,7 +18,11 @@ const (
 	debugLevel       = 2
 )
 
-func dumpGoroutines() {
+type creator interface {
+	Create(name string) (file *os.File, err error)
+}
+
+func dumpGoroutines(ctor creator) {
 	command := path.Base(os.Args[0])
 	pid := syscall.Getpid()
 	dumpFile := path.Join(os.TempDir(), fmt.Sprintf("%s-%d-goroutines-%s.dump",
@@ -27,10 +30,16 @@ func dumpGoroutines() {
 
 	logx.Infof("Got dump goroutine signal, printing goroutine profile to %s", dumpFile)
 
-	if f, err := os.Create(dumpFile); err != nil {
+	if f, err := ctor.Create(dumpFile); err != nil {
 		logx.Errorf("Failed to dump goroutine profile, error: %v", err)
 	} else {
 		defer f.Close()
 		pprof.Lookup(goroutineProfile).WriteTo(f, debugLevel)
 	}
 }
+
+type fileCreator struct{}
+
+func (fc fileCreator) Create(name string) (file *os.File, err error) {
+	return os.Create(name)
+}

core/proc/goroutines_test.go

@@ -1,23 +1,41 @@
+//go:build linux || darwin
+
 package proc
 
 import (
+	"errors"
+	"os"
 	"strings"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
-	"github.com/zeromicro/go-zero/core/logx"
+	"github.com/zeromicro/go-zero/core/logx/logtest"
 )
 
 func TestDumpGoroutines(t *testing.T) {
-	var buf strings.Builder
+	t.Run("real file", func(t *testing.T) {
-	w := logx.NewWriter(&buf)
+		buf := logtest.NewCollector(t)
-	o := logx.Reset()
+		dumpGoroutines(fileCreator{})
-	logx.SetWriter(w)
+		assert.True(t, strings.Contains(buf.String(), ".dump"))
-	defer func() {
+	})
-		logx.Reset()
-		logx.SetWriter(o)
-	}()
 
-	dumpGoroutines()
+	t.Run("fake file", func(t *testing.T) {
-	assert.True(t, strings.Contains(buf.String(), ".dump"))
+		const msg = "any message"
+		buf := logtest.NewCollector(t)
+		err := errors.New(msg)
+		dumpGoroutines(fakeCreator{
+			file: &os.File{},
+			err:  err,
+		})
+		assert.True(t, strings.Contains(buf.String(), msg))
+	})
+}
+
+type fakeCreator struct {
+	file *os.File
+	err  error
+}
+
+func (fc fakeCreator) Create(name string) (file *os.File, err error) {
+	return fc.file, fc.err
 }

core/proc/profile+polyfill.go

@@ -1,5 +1,4 @@
 //go:build windows
-// +build windows
 
 package proc
 

core/proc/profile.go

@@ -1,5 +1,4 @@
 //go:build linux || darwin
-// +build linux darwin
 
 package proc
 

core/proc/profile_test.go

@@ -5,25 +5,16 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
-	"github.com/zeromicro/go-zero/core/logx"
+	"github.com/zeromicro/go-zero/core/logx/logtest"
 )
 
 func TestProfile(t *testing.T) {
-	var buf strings.Builder
+	c := logtest.NewCollector(t)
-	w := logx.NewWriter(&buf)
-	o := logx.Reset()
-	logx.SetWriter(w)
-
-	defer func() {
-		logx.Reset()
-		logx.SetWriter(o)
-	}()
-
 	profiler := StartProfile()
 	// start again should not work
 	assert.NotNil(t, StartProfile())
 	profiler.Stop()
 	// stop twice
 	profiler.Stop()
-	assert.True(t, strings.Contains(buf.String(), ".pprof"))
+	assert.True(t, strings.Contains(c.String(), ".pprof"))
 }

core/proc/shutdown+polyfill.go

@@ -1,5 +1,4 @@
 //go:build windows
-// +build windows
 
 package proc
 

core/proc/shutdown.go

@@ -1,5 +1,4 @@
 //go:build linux || darwin
-// +build linux darwin
 
 package proc
 
@@ -97,4 +96,6 @@ func (lm *listenerManager) notifyListeners() {
 		group.RunSafe(listener)
 	}
 	group.Wait()
+
+	lm.listeners = nil
 }

core/proc/shutdown_test.go

@@ -1,5 +1,4 @@
 //go:build linux || darwin
-// +build linux darwin
 
 package proc
 
@@ -29,3 +28,33 @@ func TestShutdown(t *testing.T) {
 	called()
 	assert.Equal(t, 3, val)
 }
+
+func TestNotifyMoreThanOnce(t *testing.T) {
+	ch := make(chan struct{}, 1)
+
+	go func() {
+		var val int
+		called := AddWrapUpListener(func() {
+			val++
+		})
+		WrapUp()
+		WrapUp()
+		called()
+		assert.Equal(t, 1, val)
+
+		called = AddShutdownListener(func() {
+			val += 2
+		})
+		Shutdown()
+		Shutdown()
+		called()
+		assert.Equal(t, 3, val)
+		ch <- struct{}{}
+	}()
+
+	select {
+	case <-ch:
+	case <-time.After(time.Second):
+		t.Fatal("timeout, check error logs")
+	}
+}

core/proc/signals+polyfill.go

@@ -1,5 +1,4 @@
 //go:build windows
-// +build windows
 
 package proc
 

core/proc/signals.go

@@ -1,5 +1,4 @@
 //go:build linux || darwin
-// +build linux darwin
 
 package proc
 
@@ -27,7 +26,7 @@ func init() {
 			v := <-signals
 			switch v {
 			case syscall.SIGUSR1:
-				dumpGoroutines()
+				dumpGoroutines(fileCreator{})
 			case syscall.SIGUSR2:
 				if profiler == nil {
 					profiler = StartProfile()

core/proc/signals_test.go

@@ -1,3 +1,5 @@
+//go:build linux || darwin
+
 package proc
 
 import (

core/prof/runtime.go

@@ -2,6 +2,8 @@ package prof
 
 import (
 	"fmt"
+	"io"
+	"os"
 	"runtime"
 	"time"
 )
@@ -13,6 +15,10 @@ const (
 
 // DisplayStats prints the goroutine, memory, GC stats with given interval, default to 5 seconds.
 func DisplayStats(interval ...time.Duration) {
+	displayStatsWithWriter(os.Stdout, interval...)
+}
+
+func displayStatsWithWriter(writer io.Writer, interval ...time.Duration) {
 	duration := defaultInterval
 	for _, val := range interval {
 		duration = val
@@ -24,7 +30,7 @@ func DisplayStats(interval ...time.Duration) {
 		for range ticker.C {
 			var m runtime.MemStats
 			runtime.ReadMemStats(&m)
-			fmt.Printf("Goroutines: %d, Alloc: %vm, TotalAlloc: %vm, Sys: %vm, NumGC: %v\n",
+			fmt.Fprintf(writer, "Goroutines: %d, Alloc: %vm, TotalAlloc: %vm, Sys: %vm, NumGC: %v\n",
 				runtime.NumGoroutine(), m.Alloc/mega, m.TotalAlloc/mega, m.Sys/mega, m.NumGC)
 		}
 	}()

core/prof/runtime_test.go

@@ -0,0 +1,36 @@
+package prof
+
+import (
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestDisplayStats(t *testing.T) {
+	writer := &threadSafeBuffer{
+		buf: strings.Builder{},
+	}
+	displayStatsWithWriter(writer, time.Millisecond*10)
+	time.Sleep(time.Millisecond * 50)
+	assert.Contains(t, writer.String(), "Goroutines: ")
+}
+
+type threadSafeBuffer struct {
+	buf  strings.Builder
+	lock sync.Mutex
+}
+
+func (b *threadSafeBuffer) String() string {
+	b.lock.Lock()
+	defer b.lock.Unlock()
+	return b.buf.String()
+}
+
+func (b *threadSafeBuffer) Write(p []byte) (n int, err error) {
+	b.lock.Lock()
+	defer b.lock.Unlock()
+	return b.buf.Write(p)
+}

core/prometheus/agent.go

@@ -21,6 +21,11 @@ func Enabled() bool {
 	return enabled.True()
 }
 
+// Enable enables prometheus.
+func Enable() {
+	enabled.Set(true)
+}
+
 // StartAgent starts a prometheus agent.
 func StartAgent(c Config) {
 	if len(c.Host) == 0 {

core/queue/queue_test.go

@@ -1,6 +1,8 @@
 package queue
 
 import (
+	"errors"
+	"math"
 	"sync"
 	"sync/atomic"
 	"testing"
@@ -37,10 +39,82 @@ func TestQueue(t *testing.T) {
 	assert.Equal(t, int32(rounds), atomic.LoadInt32(&consumer.count))
 }
 
+func TestQueue_Broadcast(t *testing.T) {
+	producer := newMockedProducer(math.MaxInt32)
+	consumer := newMockedConsumer()
+	consumer.wait.Add(consumers)
+	q := NewQueue(func() (Producer, error) {
+		return producer, nil
+	}, func() (Consumer, error) {
+		return consumer, nil
+	})
+	q.AddListener(new(mockedListener))
+	q.SetName("mockqueue")
+	q.SetNumConsumer(consumers)
+	q.SetNumProducer(1)
+	go func() {
+		time.Sleep(time.Millisecond * 100)
+		q.Stop()
+	}()
+	go q.Start()
+	time.Sleep(time.Millisecond * 50)
+	q.Broadcast("message")
+	consumer.wait.Wait()
+	assert.Equal(t, int32(consumers), atomic.LoadInt32(&consumer.events))
+}
+
+func TestQueue_PauseResume(t *testing.T) {
+	producer := newMockedProducer(rounds)
+	consumer := newMockedConsumer()
+	consumer.wait.Add(consumers)
+	q := NewQueue(func() (Producer, error) {
+		return producer, nil
+	}, func() (Consumer, error) {
+		return consumer, nil
+	})
+	q.AddListener(new(mockedListener))
+	q.SetName("mockqueue")
+	q.SetNumConsumer(consumers)
+	q.SetNumProducer(1)
+	go func() {
+		producer.wait.Wait()
+		q.Stop()
+	}()
+	q.Start()
+	producer.listener.OnProducerPause()
+	assert.Equal(t, int32(0), atomic.LoadInt32(&q.active))
+	producer.listener.OnProducerResume()
+	assert.Equal(t, int32(1), atomic.LoadInt32(&q.active))
+	assert.Equal(t, int32(rounds), atomic.LoadInt32(&consumer.count))
+}
+
+func TestQueue_ConsumeError(t *testing.T) {
+	producer := newMockedProducer(rounds)
+	consumer := newMockedConsumer()
+	consumer.consumeErr = errors.New("consume error")
+	consumer.wait.Add(consumers)
+	q := NewQueue(func() (Producer, error) {
+		return producer, nil
+	}, func() (Consumer, error) {
+		return consumer, nil
+	})
+	q.AddListener(new(mockedListener))
+	q.SetName("mockqueue")
+	q.SetNumConsumer(consumers)
+	q.SetNumProducer(1)
+	go func() {
+		producer.wait.Wait()
+		q.Stop()
+	}()
+	q.Start()
+	assert.Equal(t, int32(rounds), atomic.LoadInt32(&consumer.count))
+}
+
 type mockedConsumer struct {
-	count  int32
+	count      int32
-	events int32
+	events     int32
-	wait   sync.WaitGroup
+	consumeErr error
+	wait       sync.WaitGroup
 }
 
 func newMockedConsumer() *mockedConsumer {
@@ -49,7 +123,7 @@ func newMockedConsumer() *mockedConsumer {
 
 func (c *mockedConsumer) Consume(string) error {
 	atomic.AddInt32(&c.count, 1)
-	return nil
+	return c.consumeErr
 }
 
 func (c *mockedConsumer) OnEvent(any) {
@@ -59,9 +133,10 @@ func (c *mockedConsumer) OnEvent(any) {
 }
 
 type mockedProducer struct {
-	total int32
+	total    int32
-	count int32
+	count    int32
-	wait  sync.WaitGroup
+	listener ProduceListener
+	wait     sync.WaitGroup
 }
 
 func newMockedProducer(total int32) *mockedProducer {
@@ -72,6 +147,7 @@ func newMockedProducer(total int32) *mockedProducer {
 }
 
 func (p *mockedProducer) AddListener(listener ProduceListener) {
+	p.listener = listener
 }
 
 func (p *mockedProducer) Produce() (string, bool) {

core/rescue/recover.go

@@ -1,6 +1,11 @@
 package rescue
 
-import "github.com/zeromicro/go-zero/core/logx"
+import (
+	"context"
+	"runtime/debug"
+
+	"github.com/zeromicro/go-zero/core/logx"
+)
 
 // Recover is used with defer to do cleanup on panics.
 // Use it like:
@@ -15,3 +20,14 @@ func Recover(cleanups ...func()) {
 		logx.ErrorStack(p)
 	}
 }
+
+// RecoverCtx is used with defer to do cleanup on panics.
+func RecoverCtx(ctx context.Context, cleanups ...func()) {
+	for _, cleanup := range cleanups {
+		cleanup()
+	}
+
+	if p := recover(); p != nil {
+		logx.WithContext(ctx).Errorf("%+v\n%s", p, debug.Stack())
+	}
+}

core/rescue/recover_test.go

@@ -1,6 +1,7 @@
 package rescue
 
 import (
+	"context"
 	"sync/atomic"
 	"testing"
 
@@ -25,3 +26,17 @@ func TestRescue(t *testing.T) {
 	})
 	assert.Equal(t, int32(5), atomic.LoadInt32(&count))
 }
+
+func TestRescueCtx(t *testing.T) {
+	var count int32
+	assert.NotPanics(t, func() {
+		defer RecoverCtx(context.Background(), func() {
+			atomic.AddInt32(&count, 2)
+		}, func() {
+			atomic.AddInt32(&count, 3)
+		})
+
+		panic("hello")
+	})
+	assert.Equal(t, int32(5), atomic.LoadInt32(&count))
+}

core/search/tree.go

@@ -171,11 +171,11 @@ func add(nd *node, route string, item any) error {
 		token := route[:i]
 		children := nd.getChildren(token)
 		if child, ok := children[token]; ok {
-			if child != nil {
+			if child == nil {
-				return add(child, route[i+1:], item)
+				return errInvalidState
 			}
 
-			return errInvalidState
+			return add(child, route[i+1:], item)
 		}
 
 		child := newNode(nil)

core/search/tree_debug.go

@@ -1,5 +1,4 @@
 //go:build debug
-// +build debug
 
 package search
 

core/search/tree_test.go

@@ -11,7 +11,7 @@ import (
 
 type mockedRoute struct {
 	route string
-	value int
+	value any
 }
 
 func TestSearch(t *testing.T) {
@@ -187,6 +187,12 @@ func TestSearchInvalidItem(t *testing.T) {
 	assert.Equal(t, errEmptyItem, err)
 }
 
+func TestSearchInvalidState(t *testing.T) {
+	nd := newNode("0")
+	nd.children[0]["1"] = nil
+	assert.Error(t, add(nd, "1/2", "2"))
+}
+
 func BenchmarkSearchTree(b *testing.B) {
 	const (
 		avgLen  = 1000

core/service/serviceconf.go

@@ -1,8 +1,6 @@
 package service
 
 import (
-	"log"
-
 	"github.com/zeromicro/go-zero/core/load"
 	"github.com/zeromicro/go-zero/core/logx"
 	"github.com/zeromicro/go-zero/core/proc"
@@ -39,9 +37,7 @@ type ServiceConf struct {
 
 // MustSetUp sets up the service, exits on error.
 func (sc ServiceConf) MustSetUp() {
-	if err := sc.SetUp(); err != nil {
+	logx.Must(sc.SetUp())
-		log.Fatal(err)
-	}
 }
 
 // SetUp sets up the service.

core/service/servicegroup.go

@@ -68,7 +68,7 @@ func (sg *ServiceGroup) doStart() {
 
 	for i := range sg.services {
 		service := sg.services[i]
-		routineGroup.RunSafe(func() {
+		routineGroup.Run(func() {
 			service.Start()
 		})
 	}

core/service/servicegroup_test.go

@@ -14,30 +14,6 @@ var (
 	done   = make(chan struct{})
 )
 
-type mockedService struct {
-	quit       chan struct{}
-	multiplier int
-}
-
-func newMockedService(multiplier int) *mockedService {
-	return &mockedService{
-		quit:       make(chan struct{}),
-		multiplier: multiplier,
-	}
-}
-
-func (s *mockedService) Start() {
-	mutex.Lock()
-	number *= s.multiplier
-	mutex.Unlock()
-	done <- struct{}{}
-	<-s.quit
-}
-
-func (s *mockedService) Stop() {
-	close(s.quit)
-}
-
 func TestServiceGroup(t *testing.T) {
 	multipliers := []int{2, 3, 5, 7}
 	want := 1
@@ -126,3 +102,27 @@ type mockedStarter struct {
 func (s mockedStarter) Start() {
 	s.fn()
 }
+
+type mockedService struct {
+	quit       chan struct{}
+	multiplier int
+}
+
+func newMockedService(multiplier int) *mockedService {
+	return &mockedService{
+		quit:       make(chan struct{}),
+		multiplier: multiplier,
+	}
+}
+
+func (s *mockedService) Start() {
+	mutex.Lock()
+	number *= s.multiplier
+	mutex.Unlock()
+	done <- struct{}{}
+	<-s.quit
+}
+
+func (s *mockedService) Stop() {
+	close(s.quit)
+}

core/stat/alert+polyfill.go

@@ -1,5 +1,4 @@
 //go:build !linux
-// +build !linux
 
 package stat
 

core/stat/alert.go

@@ -1,5 +1,4 @@
 //go:build linux
-// +build linux
 
 package stat
 

core/stat/alert_test.go

@@ -1,10 +1,8 @@
 //go:build linux
-// +build linux
 
 package stat
 
 import (
-	"os"
 	"strconv"
 	"sync/atomic"
 	"testing"
@@ -13,8 +11,7 @@ import (
 )
 
 func TestReport(t *testing.T) {
-	os.Setenv(clusterNameKey, "test-cluster")
+	t.Setenv(clusterNameKey, "test-cluster")
-	defer os.Unsetenv(clusterNameKey)
 
 	var count int32
 	SetReporter(func(s string) {

core/stat/internal/cgroup_linux.go

@@ -3,6 +3,7 @@ package internal
 import (
 	"bufio"
 	"fmt"
+	"math"
 	"os"
 	"path"
 	"strconv"
@@ -218,6 +219,7 @@ func parseUints(val string) ([]uint64, error) {
 		return nil, nil
 	}
 
+	var sets []uint64
 	ints := make(map[uint64]lang.PlaceholderType)
 	cols := strings.Split(val, ",")
 	for _, r := range cols {
@@ -238,7 +240,10 @@ func parseUints(val string) ([]uint64, error) {
 			}
 
 			for i := min; i <= max; i++ {
-				ints[i] = lang.Placeholder
+				if _, ok := ints[i]; !ok {
+					ints[i] = lang.Placeholder
+					sets = append(sets, i)
+				}
 			}
 		} else {
 			v, err := parseUint(r)
@@ -246,19 +251,17 @@ func parseUints(val string) ([]uint64, error) {
 				return nil, err
 			}
 
-			ints[v] = lang.Placeholder
+			if _, ok := ints[v]; !ok {
+				ints[v] = lang.Placeholder
+				sets = append(sets, v)
+			}
 		}
 	}
 
-	var sets []uint64
-	for k := range ints {
-		sets = append(sets, k)
-	}
-
 	return sets, nil
 }
 
-// runningInUserNS detects whether we are currently running in an user namespace.
+// runningInUserNS detects whether we are currently running in a user namespace.
 func runningInUserNS() bool {
 	nsOnce.Do(func() {
 		file, err := os.Open("/proc/self/uid_map")
@@ -278,13 +281,12 @@ func runningInUserNS() bool {
 		var a, b, c int64
 		fmt.Sscanf(line, "%d %d %d", &a, &b, &c)
 
-		/*
+		// We assume we are in the initial user namespace if we have a full
-		 * We assume we are in the initial user namespace if we have a full
+		// range - 4294967295 uids starting at uid 0.
-		 * range - 4294967295 uids starting at uid 0.
+		if a == 0 && b == 0 && c == math.MaxUint32 {
-		 */
-		if a == 0 && b == 0 && c == 4294967295 {
 			return
 		}
+
 		inUserNS = true
 	})
 

core/stat/internal/cgroup_linux_test.go

@@ -0,0 +1,71 @@
+package internal
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestRunningInUserNS(t *testing.T) {
+	// should be false in docker
+	assert.False(t, runningInUserNS())
+}
+
+func TestCgroupV1(t *testing.T) {
+	if isCgroup2UnifiedMode() {
+		cg, err := currentCgroupV1()
+		assert.NoError(t, err)
+		_, err = cg.cpus()
+		assert.Error(t, err)
+		_, err = cg.cpuPeriodUs()
+		assert.Error(t, err)
+		_, err = cg.cpuQuotaUs()
+		assert.Error(t, err)
+		_, err = cg.usageAllCpus()
+		assert.Error(t, err)
+	}
+}
+
+func TestParseUint(t *testing.T) {
+	tests := []struct {
+		input string
+		want  uint64
+		err   error
+	}{
+		{"0", 0, nil},
+		{"123", 123, nil},
+		{"-1", 0, nil},
+		{"-18446744073709551616", 0, nil},
+		{"foo", 0, fmt.Errorf("cgroup: bad int format: foo")},
+	}
+
+	for _, tt := range tests {
+		got, err := parseUint(tt.input)
+		assert.Equal(t, tt.err, err)
+		assert.Equal(t, tt.want, got)
+	}
+}
+
+func TestParseUints(t *testing.T) {
+	tests := []struct {
+		input string
+		want  []uint64
+		err   error
+	}{
+		{"", nil, nil},
+		{"1,2,3", []uint64{1, 2, 3}, nil},
+		{"1-3", []uint64{1, 2, 3}, nil},
+		{"1-3,5,7-9", []uint64{1, 2, 3, 5, 7, 8, 9}, nil},
+		{"foo", nil, fmt.Errorf("cgroup: bad int format: foo")},
+		{"1-bar", nil, fmt.Errorf("cgroup: bad int list format: 1-bar")},
+		{"bar-3", nil, fmt.Errorf("cgroup: bad int list format: bar-3")},
+		{"3-1", nil, fmt.Errorf("cgroup: bad int list format: 3-1")},
+	}
+
+	for _, tt := range tests {
+		got, err := parseUints(tt.input)
+		assert.Equal(t, tt.err, err)
+		assert.Equal(t, tt.want, got)
+	}
+}