chore: add more tests (#3888)

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

.codecov.yml

@@ -1,3 +1,7 @@
+coverage:
+  status:
+    patch: true
+    project: false # disabled because project coverage is not stable
 comment:
   layout: "flags, files"
   behavior: once

.github/workflows/codeql-analysis.yml

@@ -35,7 +35,7 @@ jobs:
 
     steps:
     - name: Checkout repository
-      uses: actions/checkout@v3
+      uses: actions/checkout@v4
 
     # Initializes the CodeQL tools for scanning.
     - name: Initialize CodeQL

.github/workflows/go.yml

@@ -12,12 +12,12 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - name: Check out code into the Go module directory
-        uses: actions/checkout@v3
+        uses: actions/checkout@v4
 
       - name: Set up Go 1.x
-        uses: actions/setup-go@v3
+        uses: actions/setup-go@v4
         with:
-          go-version: 1.18
+          go-version: 1.19
           check-latest: true
           cache: true
         id: go
@@ -47,13 +47,13 @@ jobs:
     runs-on: windows-latest
     steps:
       - name: Checkout codebase
-        uses: actions/checkout@v3
+        uses: actions/checkout@v4
 
       - name: Set up Go 1.x
-        uses: actions/setup-go@v3
+        uses: actions/setup-go@v4
         with:
-          # use 1.18 to guarantee Go 1.18 compatibility
-          go-version: 1.18
+          # use 1.19 to guarantee Go 1.19 compatibility
+          go-version: 1.19
           check-latest: true
           cache: true
 

.github/workflows/issues.yml

@@ -7,7 +7,7 @@ jobs:
   close-issues:
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/stale@v6
+      - uses: actions/stale@v8
         with:
           days-before-issue-stale: 365
           days-before-issue-close: 90

.github/workflows/release.yaml

@@ -16,13 +16,13 @@ jobs:
           - goarch: "386"
             goos: darwin
     steps:
-      - uses: actions/checkout@v3
+      - uses: actions/checkout@v4
       - uses: zeromicro/go-zero-release-action@master
         with:
           github_token: ${{ secrets.GITHUB_TOKEN }}
           goos: ${{ matrix.goos }}
           goarch: ${{ matrix.goarch }}
-          goversion: "https://dl.google.com/go/go1.18.10.linux-amd64.tar.gz"
+          goversion: "https://dl.google.com/go/go1.19.13.linux-amd64.tar.gz"
           project_path: "tools/goctl"
           binary_name: "goctl"
           extra_files: tools/goctl/readme.md tools/goctl/readme-cn.md

.github/workflows/reviewdog.yml

@@ -5,7 +5,7 @@ jobs:
     name: runner / staticcheck
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v3
+      - uses: actions/checkout@v4
       - uses: reviewdog/action-staticcheck@v1
         with:
           github_token: ${{ secrets.github_token }}

.gitignore

@@ -12,6 +12,7 @@
 
 # ignore
 **/.idea
+**/.vscode
 **/.DS_Store
 **/logs
 **/adhoc

CONTRIBUTING.md

@@ -1,102 +1,76 @@
-# Contributing
+# 🚀 Contributing to go-zero
 
-Welcome to go-zero!
+Welcome to the go-zero community! We're thrilled to have you here. Contributing to our project is a fantastic way to be a part of the go-zero journey. Let's make this guide exciting and fun!
 
--   [Before you get started](#before-you-get-started)
-    -   [Code of Conduct](#code-of-conduct)
-    -   [Community Expectations](#community-expectations)
--   [Getting started](#getting-started)
--   [Your First Contribution](#your-first-contribution)
-    -   [Find something to work on](#find-something-to-work-on)
-        -   [Find a good first topic](#find-a-good-first-topic)
-        -   [Work on an Issue](#work-on-an-issue)
-        -   [File an Issue](#file-an-issue)
--   [Contributor Workflow](#contributor-workflow)
-    -   [Creating Pull Requests](#creating-pull-requests)
-    -   [Code Review](#code-review)
-    -   [Testing](#testing)
+## 📜 Before You Dive In
 
-# Before you get started
+### 🤝 Code of Conduct
 
-## Code of Conduct
+Let's start on the right foot. Please take a moment to read and embrace our [Code of Conduct](/code-of-conduct.md). We're all about creating a welcoming and respectful environment.
 
-Please make sure to read and observe our [Code of Conduct](/code-of-conduct.md).
+### 🌟 Community Expectations
 
-## Community Expectations
+At go-zero, we're like a close-knit family, and we believe in creating a healthy, friendly, and productive atmosphere. It's all about sharing knowledge and building amazing things together.
 
-go-zero is a community project driven by its community which strives to promote a healthy, friendly and productive environment.
-go-zero is a web and rpc framework written in Go. It's born to ensure the stability of the busy sites with resilient design. Builtin goctl greatly improves the development productivity.
+## 🚀 Getting Started
 
-# Getting started
+Get your adventure rolling! Here's how to begin:
 
-- Fork the repository on GitHub.
-- Make your changes on your fork repository.
-- Submit a PR.
+1. 🍴 **Fork the Repository**: Head over to the GitHub repository and fork it to your own space.
 
+2. 🛠️ **Make Your Magic**: Work your magic in your forked repository. Create new features, squash bugs, or improve documentation - it's your world to conquer!
 
-# Your First Contribution
+3. 🚀 **Submit a PR (Pull Request)**: When you're ready to unveil your creation, submit a Pull Request. We can't wait to see your awesome work!
 
-We will help you to contribute in different areas like filing issues, developing features, fixing critical bugs and
-getting your work reviewed and merged.
+## 🌟 Your First Contribution
 
-If you have questions about the development process,
-feel free to [file an issue](https://github.com/zeromicro/go-zero/issues/new/choose).
+We're here to guide you on your quest to become a go-zero contributor. Whether you want to file issues, develop features, or tame some critical bugs, we've got you covered.
 
-## Find something to work on
+If you have questions or need guidance at any stage, don't hesitate to [open an issue](https://github.com/zeromicro/go-zero/issues/new/choose).
 
-We are always in need of help, be it fixing documentation, reporting bugs or writing some code.
-Look at places where you feel best coding practices aren't followed, code refactoring is needed or tests are missing.
-Here is how you get started.
+## 🔍 Find Something to Work On
 
-### Find a good first topic
+Ready to dive into the action? There are several ways to contribute:
 
-[go-zero](https://github.com/zeromicro/go-zero) has beginner-friendly issues that provide a good first issue.
-For example, [go-zero](https://github.com/zeromicro/go-zero) has
-[help wanted](https://github.com/zeromicro/go-zero/issues?q=is%3Aopen+is%3Aissue+label%3A%22help+wanted%22) and
-[good first issue](https://github.com/zeromicro/go-zero/issues?q=is%3Aopen+is%3Aissue+label%3A%22good+first+issue%22)
-labels for issues that should not need deep knowledge of the system.
-We can help new contributors who wish to work on such issues.
+### 💼 Find a Good First Topic
 
-Another good way to contribute is to find a documentation improvement, such as a missing/broken link.
-Please see [Contributing](#contributing) below for the workflow.
+Discover easy-entry issues labeled as [help wanted](https://github.com/zeromicro/go-zero/issues?q=is%3Aopen+is%3Aissue+label%3A%22help+wanted%22) or [good first issue](https://github.com/zeromicro/go-zero/issues?q=is%3Aopen+is%3Aissue+label%3A%22good+first+issue%22). These issues are perfect for newcomers and don't require deep knowledge of the system. We're here to assist you with these tasks.
 
-#### Work on an issue
+### 🪄 Work on an Issue
 
-When you are willing to take on an issue, just reply on the issue. The maintainer will assign it to you.
+Once you've picked an issue that excites you, let us know by commenting on it. Our maintainers will assign it to you, and you can embark on your mission!
 
-### File an Issue
+### 📢 File an Issue
 
-While we encourage everyone to contribute code, it is also appreciated when someone reports an issue.
+Reporting an issue is just as valuable as code contributions. If you discover a problem, don't hesitate to [open an issue](https://github.com/zeromicro/go-zero/issues/new/choose). Be sure to follow our guidelines when submitting an issue.
 
-Please follow the prompted submission guidelines while opening an issue.
+## 🎯 Contributor Workflow
 
-# Contributor Workflow
+Here's a rough guide to your contributor journey:
 
-Please do not ever hesitate to ask a question or send a pull request.
+1. 🌱 Create a New Branch: Start by creating a topic branch, usually based on the 'master' branch. This is where your contribution will grow.
 
-This is a rough outline of what a contributor's workflow looks like:
+2. 💡 Make Commits: Commit your work in logical units. Each commit should tell a story.
 
-- Create a topic branch from where to base the contribution. This is usually master.
-- Make commits of logical units.
-- Push changes in a topic branch to a personal fork of the repository.
-- Submit a pull request to [go-zero](https://github.com/zeromicro/go-zero).
+3. 🚀 Push Changes: Push the changes in your topic branch to your personal fork of the repository.
 
-## Creating Pull Requests
+4. 📦 Submit a Pull Request: When your creation is complete, submit a Pull Request to the [go-zero repository](https://github.com/zeromicro/go-zero).
 
-Pull requests are often called simply "PR".
-go-zero generally follows the standard [github pull request](https://help.github.com/articles/about-pull-requests/) process.
-To submit a proposed change, please develop the code/fix and add new test cases.
-After that, run these local verifications before submitting pull request to predict the pass or
-fail of continuous integration.
+## 🌠 Creating Pull Requests
 
-* Format the code with `gofmt`
-* Run the test with data race enabled `go test -race ./...`
+Pull Requests (PRs) are your way of making a grand entrance with your contribution. Here's how to do it:
 
-## Code Review
+- 💼 Format Your Code: Ensure your code is beautifully formatted with `gofmt`.
+- 🏃 Run Tests: Verify that your changes pass all the tests, including data race tests. Run `go test -race ./...` for the ultimate validation.
 
-To make it easier for your PR to receive reviews, consider the reviewers will need you to:
+## 👁️‍🗨️ Code Review
 
-* follow [good coding guidelines](https://github.com/golang/go/wiki/CodeReviewComments).
-* write [good commit messages](https://chris.beams.io/posts/git-commit/).
-* break large changes into a logical series of smaller patches which individually make easily understandable changes, and in aggregate solve a broader issue.
+Getting your PR reviewed is the final step before your contribution becomes part of go-zero's magical world. To make the process smooth, keep these things in mind:
 
+- 🧙‍♀️ Follow Good Coding Practices: Stick to [good coding guidelines](https://github.com/golang/go/wiki/CodeReviewComments).
+- 📝 Write Awesome Commit Messages: Craft [impressive commit messages](https://chris.beams.io/posts/git-commit/) - they're like spells in the wizard's book!
+- 🔍 Break It Down: For larger changes, consider breaking them into a series of smaller, logical patches. Each patch should make an understandable and meaningful improvement.
+
+Congratulations on your contribution journey! We're thrilled to have you as part of our go-zero community. Let's make amazing things together! 🌟
+
+Now, go out there and start your adventure! If you have any more magical ideas to enhance this guide, please share them. 🔥

code-of-conduct.md

@@ -1,76 +1,127 @@
+
 # Contributor Covenant Code of Conduct
 
 ## Our Pledge
 
-In the interest of fostering an open and welcoming environment, we as
-contributors and maintainers pledge to make participation in our project and
-our community a harassment-free experience for everyone, regardless of age, body
-size, disability, ethnicity, sex characteristics, gender identity and expression,
-level of experience, education, socio-economic status, nationality, personal
-appearance, race, religion, or sexual identity and orientation.
+We as members, contributors, and leaders pledge to make participation in our
+community a harassment-free experience for everyone, regardless of age, body
+size, visible or invisible disability, ethnicity, sex characteristics, gender
+identity and expression, level of experience, education, socio-economic status,
+nationality, personal appearance, race, caste, color, religion, or sexual
+identity and orientation.
+
+We pledge to act and interact in ways that contribute to an open, welcoming,
+diverse, inclusive, and healthy community.
 
 ## Our Standards
 
-Examples of behavior that contributes to creating a positive environment
-include:
+Examples of behavior that contributes to a positive environment for our
+community include:
 
-* Using welcoming and inclusive language
-* Being respectful of differing viewpoints and experiences
-* Gracefully accepting constructive criticism
-* Focusing on what is best for the community
-* Showing empathy towards other community members
+* Demonstrating empathy and kindness toward other people
+* Being respectful of differing opinions, viewpoints, and experiences
+* Giving and gracefully accepting constructive feedback
+* Accepting responsibility and apologizing to those affected by our mistakes,
+  and learning from the experience
+* Focusing on what is best not just for us as individuals, but for the overall
+  community
 
-Examples of unacceptable behavior by participants include:
+Examples of unacceptable behavior include:
 
-* The use of sexualized language or imagery and unwelcome sexual attention or
-  advances
-* Trolling, insulting/derogatory comments, and personal or political attacks
+* The use of sexualized language or imagery, and sexual attention or advances of
+  any kind
+* Trolling, insulting or derogatory comments, and personal or political attacks
 * Public or private harassment
-* Publishing others' private information, such as a physical or electronic
-  address, without explicit permission
+* Publishing others' private information, such as a physical or email address,
+  without their explicit permission
 * Other conduct which could reasonably be considered inappropriate in a
   professional setting
 
-## Our Responsibilities
+## Enforcement Responsibilities
 
-Project maintainers are responsible for clarifying the standards of acceptable
-behavior and are expected to take appropriate and fair corrective action in
-response to any instances of unacceptable behavior.
+Community leaders are responsible for clarifying and enforcing our standards of
+acceptable behavior and will take appropriate and fair corrective action in
+response to any behavior that they deem inappropriate, threatening, offensive,
+or harmful.
 
-Project maintainers have the right and responsibility to remove, edit, or
-reject comments, commits, code, wiki edits, issues, and other contributions
-that are not aligned to this Code of Conduct, or to ban temporarily or
-permanently any contributor for other behaviors that they deem inappropriate,
-threatening, offensive, or harmful.
+Community leaders have the right and responsibility to remove, edit, or reject
+comments, commits, code, wiki edits, issues, and other contributions that are
+not aligned to this Code of Conduct, and will communicate reasons for moderation
+decisions when appropriate.
 
 ## Scope
 
-This Code of Conduct applies within all project spaces, and it also applies when
-an individual is representing the project or its community in public spaces.
-Examples of representing a project or community include using an official
-project e-mail address, posting via an official social media account, or acting
-as an appointed representative at an online or offline event. Representation of
-a project may be further defined and clarified by project maintainers.
+This Code of Conduct applies within all community spaces, and also applies when
+an individual is officially representing the community in public spaces.
+Examples of representing our community include using an official e-mail address,
+posting via an official social media account, or acting as an appointed
+representative at an online or offline event.
 
 ## Enforcement
 
 Instances of abusive, harassing, or otherwise unacceptable behavior may be
-reported by contacting the project team at [INSERT EMAIL ADDRESS]. All
-complaints will be reviewed and investigated and will result in a response that
-is deemed necessary and appropriate to the circumstances. The project team is
-obligated to maintain confidentiality with regard to the reporter of an incident.
-Further details of specific enforcement policies may be posted separately.
+reported to the community leaders responsible for enforcement at
+[INSERT CONTACT METHOD].
+All complaints will be reviewed and investigated promptly and fairly.
 
-Project maintainers who do not follow or enforce the Code of Conduct in good
-faith may face temporary or permanent repercussions as determined by other
-members of the project's leadership.
+All community leaders are obligated to respect the privacy and security of the
+reporter of any incident.
+
+## Enforcement Guidelines
+
+Community leaders will follow these Community Impact Guidelines in determining
+the consequences for any action they deem in violation of this Code of Conduct:
+
+### 1. Correction
+
+**Community Impact**: Use of inappropriate language or other behavior deemed
+unprofessional or unwelcome in the community.
+
+**Consequence**: A private, written warning from community leaders, providing
+clarity around the nature of the violation and an explanation of why the
+behavior was inappropriate. A public apology may be requested.
+
+### 2. Warning
+
+**Community Impact**: A violation through a single incident or series of
+actions.
+
+**Consequence**: A warning with consequences for continued behavior. No
+interaction with the people involved, including unsolicited interaction with
+those enforcing the Code of Conduct, for a specified period of time. This
+includes avoiding interactions in community spaces as well as external channels
+like social media. Violating these terms may lead to a temporary or permanent
+ban.
+
+### 3. Temporary Ban
+
+**Community Impact**: A serious violation of community standards, including
+sustained inappropriate behavior.
+
+**Consequence**: A temporary ban from any sort of interaction or public
+communication with the community for a specified period of time. No public or
+private interaction with the people involved, including unsolicited interaction
+with those enforcing the Code of Conduct, is allowed during this period.
+Violating these terms may lead to a permanent ban.
+
+### 4. Permanent Ban
+
+**Community Impact**: Demonstrating a pattern of violation of community
+standards, including sustained inappropriate behavior, harassment of an
+individual, or aggression toward or disparagement of classes of individuals.
+
+**Consequence**: A permanent ban from any sort of public interaction within the
+community.
 
 ## Attribution
 
-This Code of Conduct is adapted from the [Contributor Covenant][homepage], version 1.4,
-available at https://www.contributor-covenant.org/version/1/4/code-of-conduct.html
+This Code of Conduct is adapted from the [Contributor Covenant][homepage],
+version 2.1, available at
+[https://www.contributor-covenant.org/version/2/1/code_of_conduct.html][v2.1].
 
-[homepage]: https://www.contributor-covenant.org
+Community Impact Guidelines were inspired by
+[Mozilla's code of conduct enforcement ladder][Mozilla CoC].
 
-For answers to common questions about this code of conduct, see
-https://www.contributor-covenant.org/faq
+For answers to common questions about this code of conduct, see the FAQ at
+[https://www.contributor-covenant.org/faq][FAQ]. Translations are available at
+[https://www.contributor-covenant.org/translations][translations].

core/breaker/breaker.go

@@ -46,7 +46,7 @@ type (
 		// DoWithAcceptable returns an error instantly if the Breaker rejects the request.
 		// If a panic occurs in the request, the Breaker handles it as an error
 		// and causes the same panic again.
-		// acceptable checks if it's a successful call, even if the err is not nil.
+		// acceptable checks if it's a successful call, even if the error is not nil.
 		DoWithAcceptable(req func() error, acceptable Acceptable) error
 
 		// DoWithFallback runs the given request if the Breaker accepts it.
@@ -59,7 +59,7 @@ type (
 		// DoWithFallbackAcceptable runs the fallback if the Breaker rejects the request.
 		// If a panic occurs in the request, the Breaker handles it as an error
 		// and causes the same panic again.
-		// acceptable checks if it's a successful call, even if the err is not nil.
+		// acceptable checks if it's a successful call, even if the error is not nil.
 		DoWithFallbackAcceptable(req func() error, fallback func(err error) error, acceptable Acceptable) error
 	}
 
@@ -179,7 +179,7 @@ func (lt loggedThrottle) doReq(req func() error, fallback func(err error) error,
 }
 
 func (lt loggedThrottle) logError(err error) error {
-	if err == ErrServiceUnavailable {
+	if errors.Is(err, ErrServiceUnavailable) {
 		// if circuit open, not possible to have empty error window
 		stat.Report(fmt.Sprintf(
 			"proc(%s/%d), callee: %s, breaker is open and requests dropped\nlast errors:\n%s",

core/breaker/breakers.go

@@ -59,7 +59,7 @@ func GetBreaker(name string) Breaker {
 // NoBreakerFor disables the circuit breaker for the given name.
 func NoBreakerFor(name string) {
 	lock.Lock()
-	breakers[name] = newNoOpBreaker()
+	breakers[name] = newNopBreaker()
 	lock.Unlock()
 }
 

core/breaker/breakers_test.go

@@ -30,7 +30,7 @@ func TestBreakersDoWithAcceptable(t *testing.T) {
 		assert.Equal(t, errDummy, GetBreaker("anyone").DoWithAcceptable(func() error {
 			return errDummy
 		}, func(err error) bool {
-			return err == nil || err == errDummy
+			return err == nil || errors.Is(err, errDummy)
 		}))
 	}
 	verify(t, func() bool {
@@ -45,12 +45,12 @@ func TestBreakersDoWithAcceptable(t *testing.T) {
 		}, func(err error) bool {
 			return err == nil
 		})
-		assert.True(t, err == errDummy || err == ErrServiceUnavailable)
+		assert.True(t, errors.Is(err, errDummy) || errors.Is(err, ErrServiceUnavailable))
 	}
 	verify(t, func() bool {
-		return ErrServiceUnavailable == Do("another", func() error {
+		return errors.Is(Do("another", func() error {
 			return nil
-		})
+		}), ErrServiceUnavailable)
 	})
 }
 
@@ -75,12 +75,12 @@ func TestBreakersFallback(t *testing.T) {
 		}, func(err error) error {
 			return nil
 		})
-		assert.True(t, err == nil || err == errDummy)
+		assert.True(t, err == nil || errors.Is(err, errDummy))
 	}
 	verify(t, func() bool {
-		return ErrServiceUnavailable == Do("fallback", func() error {
+		return errors.Is(Do("fallback", func() error {
 			return nil
-		})
+		}), ErrServiceUnavailable)
 	})
 }
 
@@ -94,12 +94,12 @@ func TestBreakersAcceptableFallback(t *testing.T) {
 		}, func(err error) bool {
 			return err == nil
 		})
-		assert.True(t, err == nil || err == errDummy)
+		assert.True(t, err == nil || errors.Is(err, errDummy))
 	}
 	verify(t, func() bool {
-		return ErrServiceUnavailable == Do("acceptablefallback", func() error {
+		return errors.Is(Do("acceptablefallback", func() error {
 			return nil
-		})
+		}), ErrServiceUnavailable)
 	})
 }
 

core/breaker/googlebreaker_test.go

@@ -95,7 +95,7 @@ func TestGoogleBreakerAcceptable(t *testing.T) {
 	assert.Equal(t, errAcceptable, b.doReq(func() error {
 		return errAcceptable
 	}, nil, func(err error) bool {
-		return err == errAcceptable
+		return errors.Is(err, errAcceptable)
 	}))
 }
 
@@ -105,7 +105,7 @@ func TestGoogleBreakerNotAcceptable(t *testing.T) {
 	assert.Equal(t, errAcceptable, b.doReq(func() error {
 		return errAcceptable
 	}, nil, func(err error) bool {
-		return err != errAcceptable
+		return !errors.Is(err, errAcceptable)
 	}))
 }
 

core/breaker/nopbreaker.go

@@ -1,34 +1,34 @@
 package breaker
 
-const noOpBreakerName = "nopBreaker"
+const nopBreakerName = "nopBreaker"
 
-type noOpBreaker struct{}
+type nopBreaker struct{}
 
-func newNoOpBreaker() Breaker {
-	return noOpBreaker{}
+func newNopBreaker() Breaker {
+	return nopBreaker{}
 }
 
-func (b noOpBreaker) Name() string {
-	return noOpBreakerName
+func (b nopBreaker) Name() string {
+	return nopBreakerName
 }
 
-func (b noOpBreaker) Allow() (Promise, error) {
+func (b nopBreaker) Allow() (Promise, error) {
 	return nopPromise{}, nil
 }
 
-func (b noOpBreaker) Do(req func() error) error {
+func (b nopBreaker) Do(req func() error) error {
 	return req()
 }
 
-func (b noOpBreaker) DoWithAcceptable(req func() error, _ Acceptable) error {
+func (b nopBreaker) DoWithAcceptable(req func() error, _ Acceptable) error {
 	return req()
 }
 
-func (b noOpBreaker) DoWithFallback(req func() error, _ func(err error) error) error {
+func (b nopBreaker) DoWithFallback(req func() error, _ func(err error) error) error {
 	return req()
 }
 
-func (b noOpBreaker) DoWithFallbackAcceptable(req func() error, _ func(err error) error,
+func (b nopBreaker) DoWithFallbackAcceptable(req func() error, _ func(err error) error,
 	_ Acceptable) error {
 	return req()
 }

core/breaker/nopbreaker_test.go

@@ -8,8 +8,8 @@ import (
 )
 
 func TestNopBreaker(t *testing.T) {
-	b := newNoOpBreaker()
-	assert.Equal(t, noOpBreakerName, b.Name())
+	b := newNopBreaker()
+	assert.Equal(t, nopBreakerName, b.Name())
 	p, err := b.Allow()
 	assert.Nil(t, err)
 	p.Accept()

core/collection/ring.go

@@ -25,8 +25,14 @@ func (r *Ring) Add(v any) {
 	r.lock.Lock()
 	defer r.lock.Unlock()
 
-	r.elements[r.index%len(r.elements)] = v
+	rlen := len(r.elements)
+	r.elements[r.index%rlen] = v
 	r.index++
+
+	// prevent ring index overflow
+	if r.index >= rlen<<1 {
+		r.index -= rlen
+	}
 }
 
 // Take takes all items from r.
@@ -36,16 +42,18 @@ func (r *Ring) Take() []any {
 
 	var size int
 	var start int
-	if r.index > len(r.elements) {
-		size = len(r.elements)
-		start = r.index % len(r.elements)
+	rlen := len(r.elements)
+
+	if r.index > rlen {
+		size = rlen
+		start = r.index % rlen
 	} else {
 		size = r.index
 	}
 
 	elements := make([]any, size)
 	for i := 0; i < size; i++ {
-		elements[i] = r.elements[(start+i)%len(r.elements)]
+		elements[i] = r.elements[(start+i)%rlen]
 	}
 
 	return elements

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/discov/internal/registry.go

@@ -2,6 +2,7 @@ package internal
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"io"
 	"sort"
@@ -14,6 +15,7 @@ import (
 	"github.com/zeromicro/go-zero/core/logx"
 	"github.com/zeromicro/go-zero/core/syncx"
 	"github.com/zeromicro/go-zero/core/threading"
+	"go.etcd.io/etcd/api/v3/v3rpc/rpctypes"
 	clientv3 "go.etcd.io/etcd/client/v3"
 )
 
@@ -22,6 +24,7 @@ var (
 		clusters: make(map[string]*cluster),
 	}
 	connManager = syncx.NewResourceManager()
+	errClosed   = errors.New("etcd monitor chan has been closed")
 )
 
 // A Registry is a registry that manages the etcd client connections.
@@ -288,40 +291,47 @@ func (c *cluster) reload(cli EtcdClient) {
 
 func (c *cluster) watch(cli EtcdClient, key string, rev int64) {
 	for {
-		if c.watchStream(cli, key, rev) {
+		err := c.watchStream(cli, key, rev)
+		if err == nil {
 			return
 		}
+
+		if rev != 0 && errors.Is(err, rpctypes.ErrCompacted) {
+			logx.Errorf("etcd watch stream has been compacted, try to reload, rev %d", rev)
+			rev = c.load(cli, key)
+		}
+
+		// log the error and retry
+		logx.Error(err)
 	}
 }
 
-func (c *cluster) watchStream(cli EtcdClient, key string, rev int64) bool {
+func (c *cluster) watchStream(cli EtcdClient, key string, rev int64) error {
 	var rch clientv3.WatchChan
 	if rev != 0 {
-		rch = cli.Watch(clientv3.WithRequireLeader(c.context(cli)), makeKeyPrefix(key), clientv3.WithPrefix(),
-			clientv3.WithRev(rev+1))
+		rch = cli.Watch(clientv3.WithRequireLeader(c.context(cli)), makeKeyPrefix(key),
+			clientv3.WithPrefix(), clientv3.WithRev(rev+1))
 	} else {
-		rch = cli.Watch(clientv3.WithRequireLeader(c.context(cli)), makeKeyPrefix(key), clientv3.WithPrefix())
+		rch = cli.Watch(clientv3.WithRequireLeader(c.context(cli)), makeKeyPrefix(key),
+			clientv3.WithPrefix())
 	}
 
 	for {
 		select {
 		case wresp, ok := <-rch:
 			if !ok {
-				logx.Error("etcd monitor chan has been closed")
-				return false
+				return errClosed
 			}
 			if wresp.Canceled {
-				logx.Errorf("etcd monitor chan has been canceled, error: %v", wresp.Err())
-				return false
+				return fmt.Errorf("etcd monitor chan has been canceled, error: %w", wresp.Err())
 			}
 			if wresp.Err() != nil {
-				logx.Error(fmt.Sprintf("etcd monitor chan error: %v", wresp.Err()))
-				return false
+				return fmt.Errorf("etcd monitor chan error: %w", wresp.Err())
 			}
 
 			c.handleWatchEvents(key, wresp.Events)
 		case <-c.done:
-			return true
+			return nil
 		}
 	}
 }

core/fx/retry.go

@@ -2,7 +2,6 @@ package fx
 
 import (
 	"context"
-	"errors"
 	"time"
 
 	"github.com/zeromicro/go-zero/core/errorx"
@@ -10,8 +9,6 @@ import (
 
 const defaultRetryTimes = 3
 
-var errTimeout = errors.New("retry timeout")
-
 type (
 	// RetryOption defines the method to customize DoWithRetry.
 	RetryOption func(*retryOptions)
@@ -28,7 +25,7 @@ type (
 // 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) {
+	return retry(context.Background(), func(errChan chan error, retryCount int) {
 		errChan <- fn()
 	}, opts...)
 }
@@ -40,12 +37,12 @@ func DoWithRetry(fn func() error, opts ...RetryOption) error {
 // 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) {
+	return retry(ctx, func(errChan chan error, retryCount int) {
 		errChan <- fn(ctx, retryCount)
 	}, opts...)
 }
 
-func retry(fn func(errChan chan error, retryCount int), opts ...RetryOption) error {
+func retry(ctx context.Context, fn func(errChan chan error, retryCount int), opts ...RetryOption) error {
 	options := newRetryOptions()
 	for _, opt := range opts {
 		opt(options)
@@ -53,7 +50,6 @@ func retry(fn func(errChan chan error, retryCount int), opts ...RetryOption) err
 
 	var berr errorx.BatchError
 	var cancelFunc context.CancelFunc
-	ctx := context.Background()
 	if options.timeout > 0 {
 		ctx, cancelFunc = context.WithTimeout(ctx, options.timeout)
 		defer cancelFunc()
@@ -71,14 +67,14 @@ func retry(fn func(errChan chan error, retryCount int), opts ...RetryOption) err
 				return nil
 			}
 		case <-ctx.Done():
-			berr.Add(errTimeout)
+			berr.Add(ctx.Err())
 			return berr.Err()
 		}
 
 		if options.interval > 0 {
 			select {
 			case <-ctx.Done():
-				berr.Add(errTimeout)
+				berr.Add(ctx.Err())
 				return berr.Err()
 			case <-time.After(options.interval):
 			}

core/fx/retry_test.go

@@ -98,19 +98,51 @@ func TestRetryWithInterval(t *testing.T) {
 }
 
 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 {
+	t.Run("with timeout", func(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
-		}
-		time.Sleep(time.Millisecond * 150)
-		return errors.New("any ")
-	}, WithTimeout(time.Millisecond*250), WithInterval(time.Millisecond*150)))
+		}, 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)))
+	})
+
+	t.Run("with deadline exceeded", func(t *testing.T) {
+		ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(time.Millisecond*250))
+		defer cancel()
+
+		var times int
+		assert.Error(t, DoWithRetryCtx(ctx, func(ctx context.Context, retryCount int) error {
+			times++
+			time.Sleep(time.Millisecond * 150)
+			return errors.New("any")
+		}, WithInterval(time.Millisecond*150)))
+		assert.Equal(t, 1, times)
+	})
+
+	t.Run("with deadline not exceeded", func(t *testing.T) {
+		ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(time.Millisecond*250))
+		defer cancel()
+
+		var times int
+		assert.NoError(t, DoWithRetryCtx(ctx, func(ctx context.Context, retryCount int) error {
+			times++
+			if times == defaultRetryTimes {
+				return nil
+			}
+
+			time.Sleep(time.Millisecond * 50)
+			return errors.New("any")
+		}))
+		assert.Equal(t, defaultRetryTimes, times)
+	})
 }

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

@@ -51,6 +51,11 @@ b`,
 	}
 }
 
+func TestReadTextError(t *testing.T) {
+	_, err := ReadText("not-exist")
+	assert.NotNil(t, err)
+}
+
 func TestReadTextLines(t *testing.T) {
 	text := `1
 
@@ -94,6 +99,11 @@ func TestReadTextLines(t *testing.T) {
 	}
 }
 
+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/load/adaptiveshedder.go

@@ -136,7 +136,7 @@ func (as *adaptiveShedder) addFlying(delta int64) {
 	// update avgFlying when the request is finished.
 	// this strategy makes avgFlying have a little bit lag against flying, and smoother.
 	// when the flying requests increase rapidly, avgFlying increase slower, accept more requests.
-	// when the flying requests drop rapidly, avgFlying drop slower, accept less requests.
+	// when the flying requests drop rapidly, avgFlying drop slower, accept fewer requests.
 	// it makes the service to serve as more requests as possible.
 	if delta < 0 {
 		as.avgFlyingLock.Lock()

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
-	writeStack(fmt.Sprint(v...))
+	if shallLog(ErrorLevel) {
+		// 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
-	writeStack(fmt.Sprintf(format, v...))
+	if shallLog(ErrorLevel) {
+		// 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,7 +222,7 @@ 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)
 
@@ -269,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.
@@ -429,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/readme.md

@@ -40,7 +40,7 @@ type LogConf struct {
 - `Compress`: whether or not to compress log files, only works with `file` mode.
 - `KeepDays`: how many days that the log files are kept, after the given days, the outdated files will be deleted automatically. It has no effect on `console` mode.
 - `StackCooldownMillis`: how many milliseconds to rewrite stacktrace again. It’s used to avoid stacktrace flooding.
-- `MaxBackups`: represents how many backup log files will be kept. 0 means all files will be kept forever. Only take effect when `Rotation` is `size`. NOTE: the level of option `KeepDays` will be higher. Even thougth `MaxBackups` sets 0, log files will still be removed if the `KeepDays` limitation is reached.
+- `MaxBackups`: represents how many backup log files will be kept. 0 means all files will be kept forever. Only take effect when `Rotation` is `size`. NOTE: the level of option `KeepDays` will be higher. Even though `MaxBackups` sets 0, log files will still be removed if the `KeepDays` limitation is reached.
 - `MaxSize`: represents how much space the writing log file takes up. 0 means no limit. The unit is `MB`. Only take effect when `Rotation` is `size`.
 - `Rotation`: represents the type of log rotation rule. Default is `daily`.
   - `daily` rotate the logs by day.

core/logx/richlogger.go

@@ -41,67 +41,99 @@ type richLogger struct {
 }
 
 func (l *richLogger) Debug(v ...any) {
-	l.debug(fmt.Sprint(v...))
+	if shallLog(DebugLevel) {
+		l.debug(fmt.Sprint(v...))
+	}
 }
 
 func (l *richLogger) Debugf(format string, v ...any) {
-	l.debug(fmt.Sprintf(format, v...))
+	if shallLog(DebugLevel) {
+		l.debug(fmt.Sprintf(format, v...))
+	}
 }
 
 func (l *richLogger) Debugv(v any) {
-	l.debug(v)
+	if shallLog(DebugLevel) {
+		l.debug(v)
+	}
 }
 
 func (l *richLogger) Debugw(msg string, fields ...LogField) {
-	l.debug(msg, fields...)
+	if shallLog(DebugLevel) {
+		l.debug(msg, fields...)
+	}
 }
 
 func (l *richLogger) Error(v ...any) {
-	l.err(fmt.Sprint(v...))
+	if shallLog(ErrorLevel) {
+		l.err(fmt.Sprint(v...))
+	}
 }
 
 func (l *richLogger) Errorf(format string, v ...any) {
-	l.err(fmt.Sprintf(format, v...))
+	if shallLog(ErrorLevel) {
+		l.err(fmt.Sprintf(format, v...))
+	}
 }
 
 func (l *richLogger) Errorv(v any) {
-	l.err(v)
+	if shallLog(ErrorLevel) {
+		l.err(v)
+	}
 }
 
 func (l *richLogger) Errorw(msg string, fields ...LogField) {
-	l.err(msg, fields...)
+	if shallLog(ErrorLevel) {
+		l.err(msg, fields...)
+	}
 }
 
 func (l *richLogger) Info(v ...any) {
-	l.info(fmt.Sprint(v...))
+	if shallLog(InfoLevel) {
+		l.info(fmt.Sprint(v...))
+	}
 }
 
 func (l *richLogger) Infof(format string, v ...any) {
-	l.info(fmt.Sprintf(format, v...))
+	if shallLog(InfoLevel) {
+		l.info(fmt.Sprintf(format, v...))
+	}
 }
 
 func (l *richLogger) Infov(v any) {
-	l.info(v)
+	if shallLog(InfoLevel) {
+		l.info(v)
+	}
 }
 
 func (l *richLogger) Infow(msg string, fields ...LogField) {
-	l.info(msg, fields...)
+	if shallLog(InfoLevel) {
+		l.info(msg, fields...)
+	}
 }
 
 func (l *richLogger) Slow(v ...any) {
-	l.slow(fmt.Sprint(v...))
+	if shallLog(ErrorLevel) {
+		l.slow(fmt.Sprint(v...))
+	}
 }
 
 func (l *richLogger) Slowf(format string, v ...any) {
-	l.slow(fmt.Sprintf(format, v...))
+	if shallLog(ErrorLevel) {
+		l.slow(fmt.Sprintf(format, v...))
+	}
 }
 
 func (l *richLogger) Slowv(v any) {
-	l.slow(v)
+	if shallLog(ErrorLevel) {
+		l.slow(v)
+	}
 }
 
 func (l *richLogger) Sloww(msg string, fields ...LogField) {
-	l.slow(msg, fields...)
+	if shallLog(ErrorLevel) {
+		l.slow(msg, fields...)
+	}
 }
 
 func (l *richLogger) WithCallerSkip(skip int) Logger {

core/logx/rotatelogger.go

@@ -4,7 +4,6 @@ import (
 	"compress/gzip"
 	"errors"
 	"fmt"
-	"io"
 	"log"
 	"os"
 	"path"
@@ -299,6 +298,7 @@ func (l *RotateLogger) initialize() 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 {
-	in, err := os.Open(file)
+func gzipFile(file string, fsys fileSystem) (err error) {
+	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 {
-		return err
-	} else if err = w.Close(); err != nil {
+	if _, err = fsys.Copy(w, in); err != nil {
+		// failed to copy, no need to close w
 		return err
 	}
 
-	return os.Remove(file)
+	return fsys.Close(w)
 }

core/logx/rotatelogger_test.go

@@ -1,9 +1,12 @@
 package logx
 
 import (
+	"errors"
+	"io"
 	"os"
 	"path"
 	"path/filepath"
+	"sync/atomic"
 	"syscall"
 	"testing"
 	"time"
@@ -203,6 +206,27 @@ func TestRotateLoggerClose(t *testing.T) {
 		_, 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) {
@@ -429,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"
@@ -480,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/writer.go

@@ -7,6 +7,7 @@ import (
 	"io"
 	"log"
 	"path"
+	"runtime/debug"
 	"sync"
 	"sync/atomic"
 
@@ -332,11 +333,13 @@ func wrapLevelWithColor(level string) string {
 
 func writeJson(writer io.Writer, info any) {
 	if content, err := json.Marshal(info); err != nil {
-		log.Println(err.Error())
+		log.Printf("err: %s\n\n%s", err.Error(), debug.Stack())
 	} else if writer == nil {
 		log.Println(string(content))
 	} else {
-		writer.Write(append(content, '\n'))
+		if _, err := writer.Write(append(content, '\n')); err != nil {
+			log.Println(err.Error())
+		}
 	}
 }
 
@@ -384,7 +387,7 @@ func writePlainValue(writer io.Writer, level string, val any, fields ...string)
 	buf.WriteString(level)
 	buf.WriteByte(plainEncodingSep)
 	if err := json.NewEncoder(&buf).Encode(val); err != nil {
-		log.Println(err.Error())
+		log.Printf("err: %s\n\n%s", err.Error(), debug.Stack())
 		return
 	}
 

core/logx/writer_test.go

@@ -126,9 +126,23 @@ func TestWriteJson(t *testing.T) {
 	log.SetOutput(&buf)
 	writeJson(nil, "foo")
 	assert.Contains(t, buf.String(), "foo")
+
+	buf.Reset()
+	writeJson(hardToWriteWriter{}, "foo")
+	assert.Contains(t, buf.String(), "write error")
+
 	buf.Reset()
 	writeJson(nil, make(chan int))
 	assert.Contains(t, buf.String(), "unsupported type")
+
+	buf.Reset()
+	type C struct {
+		RC func()
+	}
+	writeJson(nil, C{
+		RC: func() {},
+	})
+	assert.Contains(t, buf.String(), "runtime/debug.Stack")
 }
 
 func TestWritePlainAny(t *testing.T) {
@@ -165,6 +179,14 @@ func TestWritePlainAny(t *testing.T) {
 	writePlainAny(hardToWriteWriter{}, levelFatal, "foo")
 	assert.Contains(t, buf.String(), "write error")
 
+	buf.Reset()
+	type C struct {
+		RC func()
+	}
+	writePlainAny(nil, levelError, C{
+		RC: func() {},
+	})
+	assert.Contains(t, buf.String(), "runtime/debug.Stack")
 }
 
 func TestLogWithLimitContentLength(t *testing.T) {

core/mapping/unmarshaler.go

@@ -18,8 +18,10 @@ import (
 )
 
 const (
-	defaultKeyName = "key"
-	delimiter      = '.'
+	defaultKeyName   = "key"
+	delimiter        = '.'
+	ignoreKey        = "-"
+	numberTypeString = "number"
 )
 
 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,14 +154,14 @@ 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 errTypeMismatch
+		return newTypeMismatchErrorWithHint(fullName, reflect.Slice.String(), refValue.Type().String())
 	}
 	if refValue.IsNil() {
 		return nil
@@ -173,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
 			}
 		}
@@ -200,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:
@@ -220,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
 		}
 	}
@@ -230,7 +248,11 @@ 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 {
+	if value == nil {
+		return errNilSliceElement
+	}
+
 	ithVal := slice.Index(index)
 	switch v := value.(type) {
 	case fmt.Stringer:
@@ -238,7 +260,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.
@@ -264,7 +286,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()
@@ -282,10 +304,10 @@ 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 {
@@ -304,11 +326,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
 			}
 
@@ -320,7 +343,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
 			}
 
@@ -331,7 +354,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
 			}
@@ -420,6 +443,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)
 	}
@@ -478,7 +505,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
@@ -536,13 +563,13 @@ func (u *Unmarshaler) processFieldNotFromString(fieldType reflect.Type, value re
 			parent:  vp.parent,
 		}, fullName)
 	case typeKind == reflect.Slice && valueKind == reflect.Slice:
-		return u.fillSlice(fieldType, value, mapValue)
+		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:
@@ -587,25 +614,28 @@ 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:
-		iValue, err := v.Int64()
+	case reflect.Int, reflect.Int8, reflect.Int16, reflect.Int32, reflect.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)
-	case reflect.Uint, reflect.Uint8, reflect.Uint16, reflect.Uint32, reflect.Uint64:
-		iValue, err := v.Int64()
-		if err != nil {
-			return err
+		// if value is a pointer, we need to check overflow with the pointer's value.
+		derefedValue := value
+		for derefedValue.Type().Kind() == reflect.Ptr {
+			derefedValue = derefedValue.Elem()
+		}
+		if derefedValue.CanFloat() && derefedValue.OverflowFloat(fValue) {
+			return fmt.Errorf("parsing %q as float32: value out of range", v.String())
 		}
 
-		if iValue < 0 {
-			return fmt.Errorf("unmarshal %q with bad value %q", fullName, v.String())
-		}
-
-		target.SetUint(uint64(iValue))
-	case reflect.Float32, reflect.Float64:
+		target.SetFloat(fValue)
+	case reflect.Float64:
 		fValue, err := v.Float64()
 		if err != nil {
 			return err
@@ -613,7 +643,7 @@ func (u *Unmarshaler) processFieldPrimitiveWithJSONNumber(fieldType reflect.Type
 
 		target.SetFloat(fValue)
 	default:
-		return newTypeMismatchError(fullName)
+		return newTypeMismatchErrorWithHint(fullName, typeKind.String(), numberTypeString)
 	}
 
 	SetValue(fieldType, value, target)
@@ -707,6 +737,10 @@ func (u *Unmarshaler) processNamedField(field reflect.StructField, value reflect
 		return err
 	}
 
+	if key == ignoreKey {
+		return nil
+	}
+
 	fullName = join(fullName, key)
 	if opts != nil && len(opts.EnvVar) > 0 {
 		envVal := proc.Env(opts.EnvVar)
@@ -721,7 +755,7 @@ 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 {
@@ -814,7 +848,7 @@ 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)
 		}
@@ -862,7 +896,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
 	}
 
@@ -884,11 +918,6 @@ func (u *Unmarshaler) unmarshalWithFullName(m valuerWithParent, v any, fullName
 		typeField := baseType.Field(i)
 		valueField := valElem.Field(i)
 		if err := u.processField(typeField, valueField, m, fullName); err != nil {
-			if len(fullName) > 0 {
-				err = fmt.Errorf("%w, fullName: %s, field: %s, type: %s",
-					err, fullName, typeField.Name, valueField.Type().Name())
-			}
-
 			return err
 		}
 	}
@@ -917,6 +946,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{
@@ -994,8 +1031,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) {
-	keys := readKeys(key)
+func getValue(m valuerWithParent, key string, opaque bool) (any, bool) {
+	keys := readKeys(key, opaque)
 	return getValueWithChainedKeys(m, keys)
 }
 
@@ -1049,7 +1086,16 @@ func newTypeMismatchError(name string) error {
 	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

@@ -30,11 +30,13 @@ const (
 	leftSquareBracket  = '['
 	rightSquareBracket = ']'
 	segmentSeparator   = ','
+	intSize            = 32 << (^uint(0) >> 63) // 32 or 64
 )
 
 var (
 	errUnsupportedType  = errors.New("unsupported type on setting field value")
 	errNumberRange      = errors.New("wrong number range setting")
+	errNilSliceElement  = errors.New("null element for slice")
 	optionsCache        = make(map[string]optionsCacheValue)
 	cacheLock           sync.RWMutex
 	structRequiredCache = make(map[reflect.Type]requiredCacheValue)
@@ -79,7 +81,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() {
@@ -100,27 +102,30 @@ func convertTypeFromString(kind reflect.Kind, str string) (any, error) {
 		default:
 			return false, errTypeMismatch
 		}
-	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 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 uintValue, nil
-	case reflect.Float32, reflect.Float64:
-		floatValue, err := strconv.ParseFloat(str, 64)
-		if err != nil {
-			return 0, fmt.Errorf("the value %q cannot parsed as float", str)
-		}
-
-		return floatValue, nil
+	case reflect.Int:
+		return strconv.ParseInt(str, 10, intSize)
+	case reflect.Int8:
+		return strconv.ParseInt(str, 10, 8)
+	case reflect.Int16:
+		return strconv.ParseInt(str, 10, 16)
+	case reflect.Int32:
+		return strconv.ParseInt(str, 10, 32)
+	case reflect.Int64:
+		return strconv.ParseInt(str, 10, 64)
+	case reflect.Uint:
+		return strconv.ParseUint(str, 10, intSize)
+	case reflect.Uint8:
+		return strconv.ParseUint(str, 10, 8)
+	case reflect.Uint16:
+		return strconv.ParseUint(str, 10, 16)
+	case reflect.Uint32:
+		return strconv.ParseUint(str, 10, 32)
+	case reflect.Uint64:
+		return strconv.ParseUint(str, 10, 64)
+	case reflect.Float32:
+		return strconv.ParseFloat(str, 32)
+	case reflect.Float64:
+		return strconv.ParseFloat(str, 64)
 	case reflect.String:
 		return str, nil
 	default:
@@ -372,8 +377,6 @@ func parseOption(fieldOpts *fieldOptions, fieldName, option string) error {
 		default:
 			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 {
@@ -488,19 +491,22 @@ func setMatchedPrimitiveValue(kind reflect.Kind, value reflect.Value, v any) err
 	switch kind {
 	case reflect.Bool:
 		value.SetBool(v.(bool))
+		return nil
 	case reflect.Int, reflect.Int8, reflect.Int16, reflect.Int32, reflect.Int64:
 		value.SetInt(v.(int64))
+		return nil
 	case reflect.Uint, reflect.Uint8, reflect.Uint16, reflect.Uint32, 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
 	}
-
-	return nil
 }
 
 func setValueFromString(kind reflect.Kind, value reflect.Value, str string) error {
@@ -577,7 +583,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
 	}

core/mapping/utils_test.go

@@ -218,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) {
@@ -300,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/yamlunmarshaler_test.go

@@ -1011,6 +1011,15 @@ func TestUnmarshalYamlMapRune(t *testing.T) {
 	assert.Equal(t, rune(3), v.Machine["node3"])
 }
 
+func TestUnmarshalYamlStringOfInt(t *testing.T) {
+	text := `password: 123456`
+	var v struct {
+		Password string `json:"password"`
+	}
+	reader := strings.NewReader(text)
+	assert.Error(t, UnmarshalYamlReader(reader, &v))
+}
+
 func TestUnmarshalYamlBadInput(t *testing.T) {
 	var v struct {
 		Any string

core/metric/counter.go

@@ -3,7 +3,6 @@ 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 (
@@ -47,20 +46,16 @@ func NewCounterVec(cfg *CounterVecOpts) CounterVec {
 	return cv
 }
 
-func (cv *promCounterVec) Inc(labels ...string) {
-	if !prometheus.Enabled() {
-		return
-	}
-
-	cv.counter.WithLabelValues(labels...).Inc()
+func (cv *promCounterVec) Add(v float64, labels ...string) {
+	update(func() {
+		cv.counter.WithLabelValues(labels...).Add(v)
+	})
 }
 
-func (cv *promCounterVec) Add(v float64, labels ...string) {
-	if !prometheus.Enabled() {
-		return
-	}
-
-	cv.counter.WithLabelValues(labels...).Add(v)
+func (cv *promCounterVec) Inc(labels ...string) {
+	update(func() {
+		cv.counter.WithLabelValues(labels...).Inc()
+	})
 }
 
 func (cv *promCounterVec) close() bool {

core/metric/gauge.go

@@ -3,7 +3,6 @@ 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 (
@@ -16,8 +15,12 @@ type (
 		Set(v float64, labels ...string)
 		// Inc increments labels.
 		Inc(labels ...string)
+		// Dec decrements labels.
+		Dec(labels ...string)
 		// Add adds v to labels.
 		Add(v float64, labels ...string)
+		// Sub subtracts v to labels.
+		Sub(v float64, labels ...string)
 		close() bool
 	}
 
@@ -32,13 +35,12 @@ func NewGaugeVec(cfg *GaugeVecOpts) GaugeVec {
 		return nil
 	}
 
-	vec := prom.NewGaugeVec(
-		prom.GaugeOpts{
-			Namespace: cfg.Namespace,
-			Subsystem: cfg.Subsystem,
-			Name:      cfg.Name,
-			Help:      cfg.Help,
-		}, cfg.Labels)
+	vec := prom.NewGaugeVec(prom.GaugeOpts{
+		Namespace: cfg.Namespace,
+		Subsystem: cfg.Subsystem,
+		Name:      cfg.Name,
+		Help:      cfg.Help,
+	}, cfg.Labels)
 	prom.MustRegister(vec)
 	gv := &promGaugeVec{
 		gauge: vec,
@@ -50,28 +52,34 @@ func NewGaugeVec(cfg *GaugeVecOpts) GaugeVec {
 	return gv
 }
 
-func (gv *promGaugeVec) Inc(labels ...string) {
-	if !prometheus.Enabled() {
-		return
-	}
-
-	gv.gauge.WithLabelValues(labels...).Inc()
+func (gv *promGaugeVec) Add(v float64, labels ...string) {
+	update(func() {
+		gv.gauge.WithLabelValues(labels...).Add(v)
+	})
 }
 
-func (gv *promGaugeVec) Add(v float64, labels ...string) {
-	if !prometheus.Enabled() {
-		return
-	}
+func (gv *promGaugeVec) Dec(labels ...string) {
+	update(func() {
+		gv.gauge.WithLabelValues(labels...).Dec()
+	})
+}
 
-	gv.gauge.WithLabelValues(labels...).Add(v)
+func (gv *promGaugeVec) Inc(labels ...string) {
+	update(func() {
+		gv.gauge.WithLabelValues(labels...).Inc()
+	})
 }
 
 func (gv *promGaugeVec) Set(v float64, labels ...string) {
-	if !prometheus.Enabled() {
-		return
-	}
+	update(func() {
+		gv.gauge.WithLabelValues(labels...).Set(v)
+	})
+}
 
-	gv.gauge.WithLabelValues(labels...).Set(v)
+func (gv *promGaugeVec) Sub(v float64, labels ...string) {
+	update(func() {
+		gv.gauge.WithLabelValues(labels...).Sub(v)
+	})
 }
 
 func (gv *promGaugeVec) close() bool {

core/metric/gauge_test.go

@@ -40,6 +40,23 @@ func TestGaugeInc(t *testing.T) {
 	assert.Equal(t, float64(2), r)
 }
 
+func TestGaugeDec(t *testing.T) {
+	startAgent()
+	gaugeVec := NewGaugeVec(&GaugeVecOpts{
+		Namespace: "rpc_client",
+		Subsystem: "requests",
+		Name:      "duration_ms",
+		Help:      "rpc server requests duration(ms).",
+		Labels:    []string{"path"},
+	})
+	defer gaugeVec.close()
+	gv, _ := gaugeVec.(*promGaugeVec)
+	gv.Dec("/users")
+	gv.Dec("/users")
+	r := testutil.ToFloat64(gv.gauge)
+	assert.Equal(t, float64(-2), r)
+}
+
 func TestGaugeAdd(t *testing.T) {
 	startAgent()
 	gaugeVec := NewGaugeVec(&GaugeVecOpts{
@@ -57,6 +74,23 @@ func TestGaugeAdd(t *testing.T) {
 	assert.Equal(t, float64(20), r)
 }
 
+func TestGaugeSub(t *testing.T) {
+	startAgent()
+	gaugeVec := NewGaugeVec(&GaugeVecOpts{
+		Namespace: "rpc_client",
+		Subsystem: "request",
+		Name:      "duration_ms",
+		Help:      "rpc server requests duration(ms).",
+		Labels:    []string{"path"},
+	})
+	defer gaugeVec.close()
+	gv, _ := gaugeVec.(*promGaugeVec)
+	gv.Sub(-100, "/classroom")
+	gv.Sub(30, "/classroom")
+	r := testutil.ToFloat64(gv.gauge)
+	assert.Equal(t, float64(70), r)
+}
+
 func TestGaugeSet(t *testing.T) {
 	startAgent()
 	gaugeVec := NewGaugeVec(&GaugeVecOpts{

core/metric/histogram.go

@@ -3,24 +3,26 @@ 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 HistogramVecOpts is a histogram vector options.
 	HistogramVecOpts struct {
-		Namespace string
-		Subsystem string
-		Name      string
-		Help      string
-		Labels    []string
-		Buckets   []float64
+		Namespace   string
+		Subsystem   string
+		Name        string
+		Help        string
+		Labels      []string
+		Buckets     []float64
+		ConstLabels map[string]string
 	}
 
 	// A HistogramVec interface represents a histogram vector.
 	HistogramVec interface {
 		// Observe adds observation v to labels.
 		Observe(v int64, labels ...string)
+		// ObserveFloat allow to observe float64 values.
+		ObserveFloat(v float64, labels ...string)
 		close() bool
 	}
 
@@ -36,11 +38,12 @@ func NewHistogramVec(cfg *HistogramVecOpts) HistogramVec {
 	}
 
 	vec := prom.NewHistogramVec(prom.HistogramOpts{
-		Namespace: cfg.Namespace,
-		Subsystem: cfg.Subsystem,
-		Name:      cfg.Name,
-		Help:      cfg.Help,
-		Buckets:   cfg.Buckets,
+		Namespace:   cfg.Namespace,
+		Subsystem:   cfg.Subsystem,
+		Name:        cfg.Name,
+		Help:        cfg.Help,
+		Buckets:     cfg.Buckets,
+		ConstLabels: cfg.ConstLabels,
 	}, cfg.Labels)
 	prom.MustRegister(vec)
 	hv := &promHistogramVec{
@@ -54,11 +57,15 @@ func NewHistogramVec(cfg *HistogramVecOpts) HistogramVec {
 }
 
 func (hv *promHistogramVec) Observe(v int64, labels ...string) {
-	if !prometheus.Enabled() {
-		return
-	}
+	update(func() {
+		hv.histogram.WithLabelValues(labels...).Observe(float64(v))
+	})
+}
 
-	hv.histogram.WithLabelValues(labels...).Observe(float64(v))
+func (hv *promHistogramVec) ObserveFloat(v float64, labels ...string) {
+	update(func() {
+		hv.histogram.WithLabelValues(labels...).Observe(v)
+	})
 }
 
 func (hv *promHistogramVec) close() bool {

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) {
@@ -15,7 +14,7 @@ func TestNewHistogramVec(t *testing.T) {
 		Help:    "rpc server requests duration(ms).",
 		Buckets: []float64{1, 2, 3},
 	})
-	defer histogramVec.close()
+	defer histogramVec.(*promHistogramVec).close()
 	histogramVecNil := NewHistogramVec(nil)
 	assert.NotNil(t, histogramVec)
 	assert.Nil(t, histogramVecNil)
@@ -29,9 +28,10 @@ func TestHistogramObserve(t *testing.T) {
 		Buckets: []float64{1, 2, 3},
 		Labels:  []string{"method"},
 	})
-	defer histogramVec.close()
+	defer histogramVec.(*promHistogramVec).close()
 	hv, _ := histogramVec.(*promHistogramVec)
 	hv.Observe(2, "/Users")
+	hv.ObserveFloat(1.1, "/Users")
 
 	metadata := `
 		# HELP counts rpc server requests duration(ms).
@@ -39,15 +39,13 @@ func TestHistogramObserve(t *testing.T) {
 `
 	val := `
 		counts_bucket{method="/Users",le="1"} 0
-		counts_bucket{method="/Users",le="2"} 1
-		counts_bucket{method="/Users",le="3"} 1
-		counts_bucket{method="/Users",le="+Inf"} 1
-		counts_sum{method="/Users"} 2
-        counts_count{method="/Users"} 1
+		counts_bucket{method="/Users",le="2"} 2
+		counts_bucket{method="/Users",le="3"} 2
+		counts_bucket{method="/Users",le="+Inf"} 2
+		counts_sum{method="/Users"} 3.1
+        counts_count{method="/Users"} 2
 `
 
 	err := testutil.CollectAndCompare(hv.histogram, strings.NewReader(metadata+val))
 	assert.Nil(t, err)
-
-	proc.Shutdown()
 }

core/metric/metric.go

@@ -1,5 +1,7 @@
 package metric
 
+import "github.com/zeromicro/go-zero/core/prometheus"
+
 // A VectorOpts is a general configuration.
 type VectorOpts struct {
 	Namespace string
@@ -8,3 +10,11 @@ type VectorOpts struct {
 	Help      string
 	Labels    []string
 }
+
+func update(fn func()) {
+	if !prometheus.Enabled() {
+		return
+	}
+
+	fn()
+}

core/metric/summary.go

@@ -0,0 +1,62 @@
+package metric
+
+import (
+	prom "github.com/prometheus/client_golang/prometheus"
+	"github.com/zeromicro/go-zero/core/proc"
+)
+
+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) {
+	update(func() {
+		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_test.go

@@ -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/goroutines+polyfill.go

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

core/proc/goroutines.go

@@ -18,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",
@@ -26,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,6 +1,10 @@
+//go:build linux || darwin
+
 package proc
 
 import (
+	"errors"
+	"os"
 	"strings"
 	"testing"
 
@@ -9,7 +13,29 @@ import (
 )
 
 func TestDumpGoroutines(t *testing.T) {
-	buf := logtest.NewCollector(t)
-	dumpGoroutines()
-	assert.True(t, strings.Contains(buf.String(), ".dump"))
+	t.Run("real file", func(t *testing.T) {
+		buf := logtest.NewCollector(t)
+		dumpGoroutines(fileCreator{})
+		assert.True(t, strings.Contains(buf.String(), ".dump"))
+	})
+
+	t.Run("fake file", func(t *testing.T) {
+		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/shutdown.go

@@ -52,10 +52,10 @@ func WrapUp() {
 	wrapUpListeners.notifyListeners()
 }
 
-func gracefulStop(signals chan os.Signal) {
+func gracefulStop(signals chan os.Signal, sig syscall.Signal) {
 	signal.Stop(signals)
 
-	logx.Info("Got signal SIGTERM, shutting down...")
+	logx.Infof("Got signal %d, shutting down...", sig)
 	go wrapUpListeners.notifyListeners()
 
 	time.Sleep(wrapUpTime)
@@ -63,7 +63,7 @@ func gracefulStop(signals chan os.Signal) {
 
 	time.Sleep(delayTimeBeforeForceQuit - wrapUpTime)
 	logx.Infof("Still alive after %v, going to force kill the process...", delayTimeBeforeForceQuit)
-	syscall.Kill(syscall.Getpid(), syscall.SIGTERM)
+	_ = syscall.Kill(syscall.Getpid(), sig)
 }
 
 type listenerManager struct {
@@ -96,4 +96,6 @@ func (lm *listenerManager) notifyListeners() {
 		group.RunSafe(listener)
 	}
 	group.Wait()
+
+	lm.listeners = nil
 }

core/proc/shutdown_test.go

@@ -28,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.go

@@ -6,43 +6,38 @@ import (
 	"os"
 	"os/signal"
 	"syscall"
+	"time"
 
 	"github.com/zeromicro/go-zero/core/logx"
 )
 
-const timeFormat = "0102150405"
+const (
+	profileDuration = time.Minute
+	timeFormat      = "0102150405"
+)
 
 var done = make(chan struct{})
 
 func init() {
 	go func() {
-		var profiler Stopper
-
 		// https://golang.org/pkg/os/signal/#Notify
 		signals := make(chan os.Signal, 1)
-		signal.Notify(signals, syscall.SIGUSR1, syscall.SIGUSR2, syscall.SIGTERM)
+		signal.Notify(signals, syscall.SIGUSR1, syscall.SIGUSR2, syscall.SIGTERM, syscall.SIGINT)
 
 		for {
 			v := <-signals
 			switch v {
 			case syscall.SIGUSR1:
-				dumpGoroutines()
+				dumpGoroutines(fileCreator{})
 			case syscall.SIGUSR2:
-				if profiler == nil {
-					profiler = StartProfile()
-				} else {
-					profiler.Stop()
-					profiler = nil
-				}
+				profiler := StartProfile()
+				time.AfterFunc(profileDuration, profiler.Stop)
 			case syscall.SIGTERM:
-				select {
-				case <-done:
-					// already closed
-				default:
-					close(done)
-				}
-
-				gracefulStop(signals)
+				stopOnSignal()
+				gracefulStop(signals, syscall.SIGTERM)
+			case syscall.SIGINT:
+				stopOnSignal()
+				gracefulStop(signals, syscall.SIGINT)
 			default:
 				logx.Error("Got unregistered signal:", v)
 			}
@@ -54,3 +49,12 @@ func init() {
 func Done() <-chan struct{} {
 	return done
 }
+
+func stopOnSignal() {
+	select {
+	case <-done:
+		// already closed
+	default:
+		close(done)
+	}
+}

core/proc/signals_test.go

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

core/queue/queue_test.go

@@ -2,6 +2,7 @@ package queue
 
 import (
 	"errors"
+	"math"
 	"sync"
 	"sync/atomic"
 	"testing"
@@ -39,7 +40,7 @@ func TestQueue(t *testing.T) {
 }
 
 func TestQueue_Broadcast(t *testing.T) {
-	producer := newMockedProducer(rounds)
+	producer := newMockedProducer(math.MaxInt32)
 	consumer := newMockedConsumer()
 	consumer.wait.Add(consumers)
 	q := NewQueue(func() (Producer, error) {
@@ -51,14 +52,14 @@ func TestQueue_Broadcast(t *testing.T) {
 	q.SetName("mockqueue")
 	q.SetNumConsumer(consumers)
 	q.SetNumProducer(1)
-	q.Broadcast("message")
 	go func() {
-		producer.wait.Wait()
+		time.Sleep(time.Millisecond * 100)
 		q.Stop()
 	}()
-	q.Start()
+	go q.Start()
+	time.Sleep(time.Millisecond * 50)
+	q.Broadcast("message")
 	consumer.wait.Wait()
-	assert.Equal(t, int32(rounds), atomic.LoadInt32(&consumer.count))
 	assert.Equal(t, int32(consumers), atomic.LoadInt32(&consumer.events))
 }
 

core/search/tree.go

@@ -69,10 +69,10 @@ func (t *Tree) Add(route string, item any) error {
 	}
 
 	err := add(t.root, route[1:], item)
-	switch err {
-	case errDupItem:
+	switch {
+	case errors.Is(err, errDupItem):
 		return duplicatedItem(route)
-	case errDupSlash:
+	case errors.Is(err, errDupSlash):
 		return duplicatedSlash(route)
 	default:
 		return err
@@ -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 {
-				return add(child, route[i+1:], item)
+			if child == nil {
+				return errInvalidState
 			}
 
-			return errInvalidState
+			return add(child, route[i+1:], item)
 		}
 
 		child := newNode(nil)

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

@@ -23,17 +23,22 @@ const (
 	ProMode = "pro"
 )
 
-// A ServiceConf is a service config.
-type ServiceConf struct {
-	Name       string
-	Log        logx.LogConf
-	Mode       string `json:",default=pro,options=dev|test|rt|pre|pro"`
-	MetricsUrl string `json:",optional"`
-	// Deprecated: please use DevServer
-	Prometheus prometheus.Config `json:",optional"`
-	Telemetry  trace.Config      `json:",optional"`
-	DevServer  devserver.Config  `json:",optional"`
-}
+type (
+	// DevServerConfig is type alias for devserver.Config
+	DevServerConfig = devserver.Config
+
+	// A ServiceConf is a service config.
+	ServiceConf struct {
+		Name       string
+		Log        logx.LogConf
+		Mode       string `json:",default=pro,options=dev|test|rt|pre|pro"`
+		MetricsUrl string `json:",optional"`
+		// Deprecated: please use DevServer
+		Prometheus prometheus.Config `json:",optional"`
+		Telemetry  trace.Config      `json:",optional"`
+		DevServer  DevServerConfig   `json:",optional"`
+	}
+)
 
 // MustSetUp sets up the service, exits on error.
 func (sc ServiceConf) MustSetUp() {

core/service/serviceconf_test.go

@@ -5,6 +5,7 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/zeromicro/go-zero/core/logx"
+	"github.com/zeromicro/go-zero/internal/devserver"
 )
 
 func TestServiceConf(t *testing.T) {
@@ -14,6 +15,10 @@ func TestServiceConf(t *testing.T) {
 			Mode: "console",
 		},
 		Mode: "dev",
+		DevServer: devserver.Config{
+			Port:       6470,
+			HealthPath: "/healthz",
+		},
 	}
 	c.MustSetUp()
 }

core/service/servicegroup.go

@@ -1,8 +1,7 @@
 package service
 
 import (
-	"log"
-
+	"github.com/zeromicro/go-zero/core/logx"
 	"github.com/zeromicro/go-zero/core/proc"
 	"github.com/zeromicro/go-zero/core/syncx"
 	"github.com/zeromicro/go-zero/core/threading"
@@ -51,7 +50,7 @@ func (sg *ServiceGroup) Add(service Service) {
 // Also, quitting this method will close the logx output.
 func (sg *ServiceGroup) Start() {
 	proc.AddShutdownListener(func() {
-		log.Println("Shutting down...")
+		logx.Info("Shutting down services in group")
 		sg.stopOnce()
 	})
 
@@ -68,7 +67,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/internal/cgroup_linux.go

@@ -2,6 +2,7 @@ package internal
 
 import (
 	"bufio"
+	"errors"
 	"fmt"
 	"math"
 	"os"
@@ -18,6 +19,7 @@ import (
 
 const (
 	cgroupDir   = "/sys/fs/cgroup"
+	cpuMaxFile  = cgroupDir + "/cpu.max"
 	cpuStatFile = cgroupDir + "/cpu.stat"
 	cpusetFile  = cgroupDir + "/cpuset.cpus.effective"
 )
@@ -30,10 +32,9 @@ var (
 )
 
 type cgroup interface {
-	cpuQuotaUs() (int64, error)
-	cpuPeriodUs() (uint64, error)
-	cpus() ([]uint64, error)
-	usageAllCpus() (uint64, error)
+	cpuQuota() (float64, error)
+	cpuUsage() (uint64, error)
+	effectiveCpus() (int, error)
 }
 
 func currentCgroup() (cgroup, error) {
@@ -48,13 +49,22 @@ type cgroupV1 struct {
 	cgroups map[string]string
 }
 
-func (c *cgroupV1) cpuQuotaUs() (int64, error) {
-	data, err := iox.ReadText(path.Join(c.cgroups["cpu"], "cpu.cfs_quota_us"))
+func (c *cgroupV1) cpuQuota() (float64, error) {
+	quotaUs, err := c.cpuQuotaUs()
 	if err != nil {
 		return 0, err
 	}
 
-	return strconv.ParseInt(data, 10, 64)
+	if quotaUs == -1 {
+		return -1, nil
+	}
+
+	periodUs, err := c.cpuPeriodUs()
+	if err != nil {
+		return 0, err
+	}
+
+	return float64(quotaUs) / float64(periodUs), nil
 }
 
 func (c *cgroupV1) cpuPeriodUs() (uint64, error) {
@@ -66,16 +76,16 @@ func (c *cgroupV1) cpuPeriodUs() (uint64, error) {
 	return parseUint(data)
 }
 
-func (c *cgroupV1) cpus() ([]uint64, error) {
-	data, err := iox.ReadText(path.Join(c.cgroups["cpuset"], "cpuset.cpus"))
+func (c *cgroupV1) cpuQuotaUs() (int64, error) {
+	data, err := iox.ReadText(path.Join(c.cgroups["cpu"], "cpu.cfs_quota_us"))
 	if err != nil {
-		return nil, err
+		return 0, err
 	}
 
-	return parseUints(data)
+	return strconv.ParseInt(data, 10, 64)
 }
 
-func (c *cgroupV1) usageAllCpus() (uint64, error) {
+func (c *cgroupV1) cpuUsage() (uint64, error) {
 	data, err := iox.ReadText(path.Join(c.cgroups["cpuacct"], "cpuacct.usage"))
 	if err != nil {
 		return 0, err
@@ -84,38 +94,53 @@ func (c *cgroupV1) usageAllCpus() (uint64, error) {
 	return parseUint(data)
 }
 
+func (c *cgroupV1) effectiveCpus() (int, error) {
+	data, err := iox.ReadText(path.Join(c.cgroups["cpuset"], "cpuset.cpus"))
+	if err != nil {
+		return 0, err
+	}
+
+	cpus, err := parseUints(data)
+	if err != nil {
+		return 0, err
+	}
+
+	return len(cpus), nil
+}
+
 type cgroupV2 struct {
 	cgroups map[string]string
 }
 
-func (c *cgroupV2) cpuQuotaUs() (int64, error) {
-	data, err := iox.ReadText(path.Join(cgroupDir, "cpu.cfs_quota_us"))
+func (c *cgroupV2) cpuQuota() (float64, error) {
+	data, err := iox.ReadText(cpuMaxFile)
 	if err != nil {
 		return 0, err
 	}
 
-	return strconv.ParseInt(data, 10, 64)
-}
+	fields := strings.Fields(data)
+	if len(fields) != 2 {
+		return 0, fmt.Errorf("cgroup: bad /sys/fs/cgroup/cpu.max file: %s", data)
+	}
 
-func (c *cgroupV2) cpuPeriodUs() (uint64, error) {
-	data, err := iox.ReadText(path.Join(cgroupDir, "cpu.cfs_period_us"))
+	if fields[0] == "max" {
+		return -1, nil
+	}
+
+	quotaUs, err := strconv.ParseInt(fields[0], 10, 64)
 	if err != nil {
 		return 0, err
 	}
 
-	return parseUint(data)
-}
-
-func (c *cgroupV2) cpus() ([]uint64, error) {
-	data, err := iox.ReadText(cpusetFile)
+	periodUs, err := strconv.ParseUint(fields[1], 10, 64)
 	if err != nil {
-		return nil, err
+		return 0, err
 	}
 
-	return parseUints(data)
+	return float64(quotaUs) / float64(periodUs), nil
 }
 
-func (c *cgroupV2) usageAllCpus() (uint64, error) {
+func (c *cgroupV2) cpuUsage() (uint64, error) {
 	usec, err := parseUint(c.cgroups["usage_usec"])
 	if err != nil {
 		return 0, err
@@ -124,6 +149,20 @@ func (c *cgroupV2) usageAllCpus() (uint64, error) {
 	return usec * uint64(time.Microsecond), nil
 }
 
+func (c *cgroupV2) effectiveCpus() (int, error) {
+	data, err := iox.ReadText(cpusetFile)
+	if err != nil {
+		return 0, err
+	}
+
+	cpus, err := parseUints(data)
+	if err != nil {
+		return 0, err
+	}
+
+	return len(cpus), nil
+}
+
 func currentCgroupV1() (cgroup, error) {
 	cgroupFile := fmt.Sprintf("/proc/%d/cgroup", os.Getpid())
 	lines, err := iox.ReadTextLines(cgroupFile, iox.WithoutBlank())
@@ -200,7 +239,7 @@ func isCgroup2UnifiedMode() bool {
 func parseUint(s string) (uint64, error) {
 	v, err := strconv.ParseInt(s, 10, 64)
 	if err != nil {
-		if err.(*strconv.NumError).Err == strconv.ErrRange {
+		if errors.Is(err, strconv.ErrRange) {
 			return 0, nil
 		}
 
@@ -219,27 +258,31 @@ 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 {
 		if strings.Contains(r, "-") {
 			fields := strings.SplitN(r, "-", 2)
-			min, err := parseUint(fields[0])
+			minimum, err := parseUint(fields[0])
 			if err != nil {
 				return nil, fmt.Errorf("cgroup: bad int list format: %s", val)
 			}
 
-			max, err := parseUint(fields[1])
+			maximum, err := parseUint(fields[1])
 			if err != nil {
 				return nil, fmt.Errorf("cgroup: bad int list format: %s", val)
 			}
 
-			if max < min {
+			if maximum < minimum {
 				return nil, fmt.Errorf("cgroup: bad int list format: %s", val)
 			}
 
-			for i := min; i <= max; i++ {
-				ints[i] = lang.Placeholder
+			for i := minimum; i <= maximum; i++ {
+				if _, ok := ints[i]; !ok {
+					ints[i] = lang.Placeholder
+					sets = append(sets, i)
+				}
 			}
 		} else {
 			v, err := parseUint(r)
@@ -247,19 +290,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")

core/stat/internal/cgroup_linux_test.go

@@ -1,6 +1,7 @@
 package internal
 
 import (
+	"fmt"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -11,17 +12,71 @@ func TestRunningInUserNS(t *testing.T) {
 	assert.False(t, runningInUserNS())
 }
 
-func TestCgroupV1(t *testing.T) {
-	if isCgroup2UnifiedMode() {
+func TestCgroups(t *testing.T) {
+	// test cgroup legacy(v1) & hybrid
+	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()
+		_, err = cg.effectiveCpus()
+		assert.NoError(t, err)
+		_, err = cg.cpuQuota()
+		assert.NoError(t, err)
+		_, err = cg.cpuUsage()
+		assert.NoError(t, err)
+	}
+
+	// test cgroup v2
+	if isCgroup2UnifiedMode() {
+		cg, err := currentCgroupV2()
+		assert.NoError(t, err)
+		_, err = cg.effectiveCpus()
+		assert.NoError(t, err)
+		_, err = cg.cpuQuota()
 		assert.Error(t, err)
+		_, err = cg.cpuUsage()
+		assert.NoError(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)
 	}
 }

core/stat/internal/cpu_linux.go

@@ -14,39 +14,32 @@ import (
 const (
 	cpuTicks  = 100
 	cpuFields = 8
+	cpuMax    = 1000
+	statFile  = "/proc/stat"
 )
 
 var (
 	preSystem uint64
 	preTotal  uint64
-	quota     float64
+	limit     float64
 	cores     uint64
 	initOnce  sync.Once
 )
 
 // if /proc not present, ignore the cpu calculation, like wsl linux
 func initialize() {
-	cpus, err := cpuSets()
+	cpus, err := effectiveCpus()
 	if err != nil {
 		logx.Error(err)
 		return
 	}
 
-	cores = uint64(len(cpus))
-	quota = float64(len(cpus))
-	cq, err := cpuQuota()
-	if err == nil {
-		if cq != -1 {
-			period, err := cpuPeriod()
-			if err != nil {
-				logx.Error(err)
-				return
-			}
-
-			limit := float64(cq) / float64(period)
-			if limit < quota {
-				quota = limit
-			}
+	cores = uint64(cpus)
+	limit = float64(cpus)
+	quota, err := cpuQuota()
+	if err == nil && quota > 0 {
+		if quota < limit {
+			limit = quota
 		}
 	}
 
@@ -56,7 +49,7 @@ func initialize() {
 		return
 	}
 
-	preTotal, err = totalCpuUsage()
+	preTotal, err = cpuUsage()
 	if err != nil {
 		logx.Error(err)
 		return
@@ -67,7 +60,7 @@ func initialize() {
 func RefreshCpu() uint64 {
 	initOnce.Do(initialize)
 
-	total, err := totalCpuUsage()
+	total, err := cpuUsage()
 	if err != nil {
 		return 0
 	}
@@ -81,7 +74,10 @@ func RefreshCpu() uint64 {
 	cpuDelta := total - preTotal
 	systemDelta := system - preSystem
 	if cpuDelta > 0 && systemDelta > 0 {
-		usage = uint64(float64(cpuDelta*cores*1e3) / (float64(systemDelta) * quota))
+		usage = uint64(float64(cpuDelta*cores*cpuMax) / (float64(systemDelta) * limit))
+		if usage > cpuMax {
+			usage = cpuMax
+		}
 	}
 	preSystem = system
 	preTotal = total
@@ -89,35 +85,35 @@ func RefreshCpu() uint64 {
 	return usage
 }
 
-func cpuQuota() (int64, error) {
+func cpuQuota() (float64, error) {
 	cg, err := currentCgroup()
 	if err != nil {
 		return 0, err
 	}
 
-	return cg.cpuQuotaUs()
+	return cg.cpuQuota()
 }
 
-func cpuPeriod() (uint64, error) {
+func cpuUsage() (uint64, error) {
 	cg, err := currentCgroup()
 	if err != nil {
 		return 0, err
 	}
 
-	return cg.cpuPeriodUs()
+	return cg.cpuUsage()
 }
 
-func cpuSets() ([]uint64, error) {
+func effectiveCpus() (int, error) {
 	cg, err := currentCgroup()
 	if err != nil {
-		return nil, err
+		return 0, err
 	}
 
-	return cg.cpus()
+	return cg.effectiveCpus()
 }
 
 func systemCpuUsage() (uint64, error) {
-	lines, err := iox.ReadTextLines("/proc/stat", iox.WithoutBlank())
+	lines, err := iox.ReadTextLines(statFile, iox.WithoutBlank())
 	if err != nil {
 		return 0, err
 	}
@@ -145,12 +141,3 @@ func systemCpuUsage() (uint64, error) {
 
 	return 0, errors.New("bad stats format")
 }
-
-func totalCpuUsage() (usage uint64, err error) {
-	var cg cgroup
-	if cg, err = currentCgroup(); err != nil {
-		return
-	}
-
-	return cg.usageAllCpus()
-}

core/stores/builder/builder.go

@@ -9,7 +9,7 @@ import (
 const dbTag = "db"
 
 // RawFieldNames converts golang struct field into slice string.
-func RawFieldNames(in any, postgresSql ...bool) []string {
+func RawFieldNames(in any, postgreSql ...bool) []string {
 	out := make([]string, 0)
 	v := reflect.ValueOf(in)
 	if v.Kind() == reflect.Ptr {
@@ -17,8 +17,8 @@ func RawFieldNames(in any, postgresSql ...bool) []string {
 	}
 
 	var pg bool
-	if len(postgresSql) > 0 {
-		pg = postgresSql[0]
+	if len(postgreSql) > 0 {
+		pg = postgreSql[0]
 	}
 
 	// we only accept structs

core/stores/cache/cachenode.go

@@ -96,7 +96,7 @@ func (c cacheNode) Get(key string, val any) error {
 // GetCtx gets the cache with key and fills into v.
 func (c cacheNode) GetCtx(ctx context.Context, key string, val any) error {
 	err := c.doGetCache(ctx, key, val)
-	if err == errPlaceholder {
+	if errors.Is(err, errPlaceholder) {
 		return c.errNotFound
 	}
 
@@ -210,16 +210,16 @@ func (c cacheNode) doTake(ctx context.Context, v any, key string,
 	logger := logx.WithContext(ctx)
 	val, fresh, err := c.barrier.DoEx(key, func() (any, error) {
 		if err := c.doGetCache(ctx, key, v); err != nil {
-			if err == errPlaceholder {
+			if errors.Is(err, errPlaceholder) {
 				return nil, c.errNotFound
-			} else if err != c.errNotFound {
+			} else if !errors.Is(err, c.errNotFound) {
 				// why we just return the error instead of query from db,
 				// because we don't allow the disaster pass to the dbs.
 				// fail fast, in case we bring down the dbs.
 				return nil, err
 			}
 
-			if err = query(v); err == c.errNotFound {
+			if err = query(v); errors.Is(err, c.errNotFound) {
 				if err = c.setCacheWithNotFound(ctx, key); err != nil {
 					logger.Error(err)
 				}

core/stores/mon/bulkinserter_test.go

@@ -11,8 +11,6 @@ import (
 
 func TestBulkInserter(t *testing.T) {
 	mt := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
-	defer mt.Close()
-
 	mt.Run("test", func(mt *mtest.T) {
 		mt.AddMockResponses(mtest.CreateSuccessResponse(bson.D{{Key: "ok", Value: 1}}...))
 		bulk, err := NewBulkInserter(createModel(mt).Collection)

core/stores/mon/clientmanager_test.go

@@ -9,8 +9,6 @@ import (
 
 func TestClientManger_getClient(t *testing.T) {
 	mt := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
-	defer mt.Close()
-
 	mt.Run("test", func(mt *mtest.T) {
 		Inject(mtest.ClusterURI(), mt.Client)
 		cli, err := getClient(mtest.ClusterURI())

core/stores/mon/collection.go

@@ -2,11 +2,10 @@ package mon
 
 import (
 	"context"
-	"encoding/json"
+	"errors"
 	"time"
 
 	"github.com/zeromicro/go-zero/core/breaker"
-	"github.com/zeromicro/go-zero/core/logx"
 	"github.com/zeromicro/go-zero/core/timex"
 	"go.mongodb.org/mongo-driver/mongo"
 	mopt "go.mongodb.org/mongo-driver/mongo/options"
@@ -502,45 +501,11 @@ func (c *decoratedCollection) UpdateOne(ctx context.Context, filter, update any,
 
 func (c *decoratedCollection) logDuration(ctx context.Context, method string,
 	startTime time.Duration, err error, docs ...any) {
-	duration := timex.Since(startTime)
-	logger := logx.WithContext(ctx).WithDuration(duration)
-
-	content, jerr := json.Marshal(docs)
-	// jerr should not be non-nil, but we don't care much on this,
-	// if non-nil, we just log without docs.
-	if jerr != nil {
-		if err != nil {
-			if duration > slowThreshold.Load() {
-				logger.Slowf("[MONGO] mongo(%s) - slowcall - %s - fail(%s)", c.name, method, err.Error())
-			} else {
-				logger.Infof("mongo(%s) - %s - fail(%s)", c.name, method, err.Error())
-			}
-		} else {
-			if duration > slowThreshold.Load() {
-				logger.Slowf("[MONGO] mongo(%s) - slowcall - %s - ok", c.name, method)
-			} else {
-				logger.Infof("mongo(%s) - %s - ok", c.name, method)
-			}
-		}
-	} else if err != nil {
-		if duration > slowThreshold.Load() {
-			logger.Slowf("[MONGO] mongo(%s) - slowcall - %s - fail(%s) - %s",
-				c.name, method, err.Error(), string(content))
-		} else {
-			logger.Infof("mongo(%s) - %s - fail(%s) - %s",
-				c.name, method, err.Error(), string(content))
-		}
-	} else {
-		if duration > slowThreshold.Load() {
-			logger.Slowf("[MONGO] mongo(%s) - slowcall - %s - ok - %s",
-				c.name, method, string(content))
-		} else {
-			logger.Infof("mongo(%s) - %s - ok - %s", c.name, method, string(content))
-		}
-	}
+	logDurationWithDocs(ctx, c.name, method, startTime, err, docs...)
 }
 
-func (c *decoratedCollection) logDurationSimple(ctx context.Context, method string, startTime time.Duration, err error) {
+func (c *decoratedCollection) logDurationSimple(ctx context.Context, method string,
+	startTime time.Duration, err error) {
 	logDuration(ctx, c.name, method, startTime, err)
 }
 
@@ -562,11 +527,19 @@ func (p keepablePromise) keep(err error) error {
 }
 
 func acceptable(err error) bool {
-	return err == nil || err == mongo.ErrNoDocuments || err == mongo.ErrNilValue ||
-		err == mongo.ErrNilDocument || err == mongo.ErrNilCursor || err == mongo.ErrEmptySlice ||
+	return err == nil ||
+		errors.Is(err, mongo.ErrNoDocuments) ||
+		errors.Is(err, mongo.ErrNilValue) ||
+		errors.Is(err, mongo.ErrNilDocument) ||
+		errors.Is(err, mongo.ErrNilCursor) ||
+		errors.Is(err, mongo.ErrEmptySlice) ||
 		// session errors
-		err == session.ErrSessionEnded || err == session.ErrNoTransactStarted ||
-		err == session.ErrTransactInProgress || err == session.ErrAbortAfterCommit ||
-		err == session.ErrAbortTwice || err == session.ErrCommitAfterAbort ||
-		err == session.ErrUnackWCUnsupported || err == session.ErrSnapshotTransaction
+		errors.Is(err, session.ErrSessionEnded) ||
+		errors.Is(err, session.ErrNoTransactStarted) ||
+		errors.Is(err, session.ErrTransactInProgress) ||
+		errors.Is(err, session.ErrAbortAfterCommit) ||
+		errors.Is(err, session.ErrAbortTwice) ||
+		errors.Is(err, session.ErrCommitAfterAbort) ||
+		errors.Is(err, session.ErrUnackWCUnsupported) ||
+		errors.Is(err, session.ErrSnapshotTransaction)
 }

core/stores/mon/collection_test.go

@@ -68,7 +68,6 @@ func TestKeepPromise_keep(t *testing.T) {
 
 func TestNewCollection(t *testing.T) {
 	mt := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
-	defer mt.Close()
 	mt.Run("test", func(mt *mtest.T) {
 		coll := mt.Coll
 		assert.NotNil(t, coll)
@@ -79,7 +78,6 @@ func TestNewCollection(t *testing.T) {
 
 func TestCollection_Aggregate(t *testing.T) {
 	mt := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
-	defer mt.Close()
 	mt.Run("test", func(mt *mtest.T) {
 		coll := mt.Coll
 		assert.NotNil(t, coll)
@@ -96,8 +94,6 @@ func TestCollection_Aggregate(t *testing.T) {
 
 func TestCollection_BulkWrite(t *testing.T) {
 	mt := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
-	defer mt.Close()
-
 	mt.Run("test", func(mt *mtest.T) {
 		c := decoratedCollection{
 			Collection: mt.Coll,
@@ -119,8 +115,6 @@ func TestCollection_BulkWrite(t *testing.T) {
 
 func TestCollection_CountDocuments(t *testing.T) {
 	mt := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
-	defer mt.Close()
-
 	mt.Run("test", func(mt *mtest.T) {
 		c := decoratedCollection{
 			Collection: mt.Coll,
@@ -145,8 +139,6 @@ func TestCollection_CountDocuments(t *testing.T) {
 
 func TestDecoratedCollection_DeleteMany(t *testing.T) {
 	mt := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
-	defer mt.Close()
-
 	mt.Run("test", func(mt *mtest.T) {
 		c := decoratedCollection{
 			Collection: mt.Coll,
@@ -165,8 +157,6 @@ func TestDecoratedCollection_DeleteMany(t *testing.T) {
 
 func TestCollection_Distinct(t *testing.T) {
 	mt := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
-	defer mt.Close()
-
 	mt.Run("test", func(mt *mtest.T) {
 		c := decoratedCollection{
 			Collection: mt.Coll,
@@ -185,8 +175,6 @@ func TestCollection_Distinct(t *testing.T) {
 
 func TestCollection_EstimatedDocumentCount(t *testing.T) {
 	mt := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
-	defer mt.Close()
-
 	mt.Run("test", func(mt *mtest.T) {
 		c := decoratedCollection{
 			Collection: mt.Coll,
@@ -205,8 +193,6 @@ func TestCollection_EstimatedDocumentCount(t *testing.T) {
 
 func TestCollection_Find(t *testing.T) {
 	mt := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
-	defer mt.Close()
-
 	mt.Run("test", func(mt *mtest.T) {
 		c := decoratedCollection{
 			Collection: mt.Coll,
@@ -253,8 +239,6 @@ func TestCollection_Find(t *testing.T) {
 
 func TestCollection_FindOne(t *testing.T) {
 	mt := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
-	defer mt.Close()
-
 	mt.Run("test", func(mt *mtest.T) {
 		c := decoratedCollection{
 			Collection: mt.Coll,
@@ -297,8 +281,6 @@ func TestCollection_FindOne(t *testing.T) {
 
 func TestCollection_FindOneAndDelete(t *testing.T) {
 	mt := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
-	defer mt.Close()
-
 	mt.Run("test", func(mt *mtest.T) {
 		c := decoratedCollection{
 			Collection: mt.Coll,
@@ -328,8 +310,6 @@ func TestCollection_FindOneAndDelete(t *testing.T) {
 
 func TestCollection_FindOneAndReplace(t *testing.T) {
 	mt := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
-	defer mt.Close()
-
 	mt.Run("test", func(mt *mtest.T) {
 		c := decoratedCollection{
 			Collection: mt.Coll,
@@ -360,8 +340,6 @@ func TestCollection_FindOneAndReplace(t *testing.T) {
 
 func TestCollection_FindOneAndUpdate(t *testing.T) {
 	mt := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
-	defer mt.Close()
-
 	mt.Run("test", func(mt *mtest.T) {
 		c := decoratedCollection{
 			Collection: mt.Coll,
@@ -393,8 +371,6 @@ func TestCollection_FindOneAndUpdate(t *testing.T) {
 
 func TestCollection_InsertOne(t *testing.T) {
 	mt := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
-	defer mt.Close()
-
 	mt.Run("test", func(mt *mtest.T) {
 		c := decoratedCollection{
 			Collection: mt.Coll,
@@ -413,8 +389,6 @@ func TestCollection_InsertOne(t *testing.T) {
 
 func TestCollection_InsertMany(t *testing.T) {
 	mt := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
-	defer mt.Close()
-
 	mt.Run("test", func(mt *mtest.T) {
 		c := decoratedCollection{
 			Collection: mt.Coll,
@@ -437,8 +411,6 @@ func TestCollection_InsertMany(t *testing.T) {
 
 func TestCollection_DeleteOne(t *testing.T) {
 	mt := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
-	defer mt.Close()
-
 	mt.Run("test", func(mt *mtest.T) {
 		c := decoratedCollection{
 			Collection: mt.Coll,
@@ -457,8 +429,6 @@ func TestCollection_DeleteOne(t *testing.T) {
 
 func TestCollection_DeleteMany(t *testing.T) {
 	mt := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
-	defer mt.Close()
-
 	mt.Run("test", func(mt *mtest.T) {
 		c := decoratedCollection{
 			Collection: mt.Coll,
@@ -477,8 +447,6 @@ func TestCollection_DeleteMany(t *testing.T) {
 
 func TestCollection_ReplaceOne(t *testing.T) {
 	mt := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
-	defer mt.Close()
-
 	mt.Run("test", func(mt *mtest.T) {
 		c := decoratedCollection{
 			Collection: mt.Coll,
@@ -500,8 +468,6 @@ func TestCollection_ReplaceOne(t *testing.T) {
 
 func TestCollection_UpdateOne(t *testing.T) {
 	mt := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
-	defer mt.Close()
-
 	mt.Run("test", func(mt *mtest.T) {
 		c := decoratedCollection{
 			Collection: mt.Coll,
@@ -522,8 +488,6 @@ func TestCollection_UpdateOne(t *testing.T) {
 
 func TestCollection_UpdateByID(t *testing.T) {
 	mt := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
-	defer mt.Close()
-
 	mt.Run("test", func(mt *mtest.T) {
 		c := decoratedCollection{
 			Collection: mt.Coll,
@@ -544,8 +508,6 @@ func TestCollection_UpdateByID(t *testing.T) {
 
 func TestCollection_UpdateMany(t *testing.T) {
 	mt := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
-	defer mt.Close()
-
 	mt.Run("test", func(mt *mtest.T) {
 		c := decoratedCollection{
 			Collection: mt.Coll,
@@ -566,7 +528,6 @@ func TestCollection_UpdateMany(t *testing.T) {
 
 func TestDecoratedCollection_LogDuration(t *testing.T) {
 	mt := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
-	defer mt.Close()
 	c := decoratedCollection{
 		Collection: mt.Coll,
 		brk:        breaker.NewBreaker(),
@@ -599,13 +560,11 @@ func TestDecoratedCollection_LogDuration(t *testing.T) {
 		errors.New("bar"), make(chan int))
 	assert.Contains(t, buf.String(), "foo")
 	assert.Contains(t, buf.String(), "bar")
-	assert.Contains(t, buf.String(), "slowcall")
 
 	buf.Reset()
 	c.logDuration(context.Background(), "foo", timex.Now()-slowThreshold.Load()*2,
 		errors.New("bar"))
 	assert.Contains(t, buf.String(), "foo")
-	assert.Contains(t, buf.String(), "slowcall")
 
 	buf.Reset()
 	c.logDuration(context.Background(), "foo", timex.Now()-slowThreshold.Load()*2, nil)

core/stores/mon/model_test.go

@@ -12,8 +12,6 @@ import (
 
 func TestModel_StartSession(t *testing.T) {
 	mt := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
-	defer mt.Close()
-
 	mt.Run("test", func(mt *mtest.T) {
 		m := createModel(mt)
 		sess, err := m.StartSession()
@@ -34,8 +32,6 @@ func TestModel_StartSession(t *testing.T) {
 
 func TestModel_Aggregate(t *testing.T) {
 	mt := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
-	defer mt.Close()
-
 	mt.Run("test", func(mt *mtest.T) {
 		m := createModel(mt)
 		find := mtest.CreateCursorResponse(
@@ -71,8 +67,6 @@ func TestModel_Aggregate(t *testing.T) {
 
 func TestModel_DeleteMany(t *testing.T) {
 	mt := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
-	defer mt.Close()
-
 	mt.Run("test", func(mt *mtest.T) {
 		m := createModel(mt)
 		mt.AddMockResponses(mtest.CreateSuccessResponse(bson.D{{Key: "n", Value: 1}}...))
@@ -88,8 +82,6 @@ func TestModel_DeleteMany(t *testing.T) {
 
 func TestModel_DeleteOne(t *testing.T) {
 	mt := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
-	defer mt.Close()
-
 	mt.Run("test", func(mt *mtest.T) {
 		m := createModel(mt)
 		mt.AddMockResponses(mtest.CreateSuccessResponse(bson.D{{Key: "n", Value: 1}}...))
@@ -105,8 +97,6 @@ func TestModel_DeleteOne(t *testing.T) {
 
 func TestModel_Find(t *testing.T) {
 	mt := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
-	defer mt.Close()
-
 	mt.Run("test", func(mt *mtest.T) {
 		m := createModel(mt)
 		find := mtest.CreateCursorResponse(
@@ -142,8 +132,6 @@ func TestModel_Find(t *testing.T) {
 
 func TestModel_FindOne(t *testing.T) {
 	mt := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
-	defer mt.Close()
-
 	mt.Run("test", func(mt *mtest.T) {
 		m := createModel(mt)
 		find := mtest.CreateCursorResponse(
@@ -170,8 +158,6 @@ func TestModel_FindOne(t *testing.T) {
 
 func TestModel_FindOneAndDelete(t *testing.T) {
 	mt := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
-	defer mt.Close()
-
 	mt.Run("test", func(mt *mtest.T) {
 		m := createModel(mt)
 		mt.AddMockResponses(mtest.CreateSuccessResponse(bson.D{
@@ -189,8 +175,6 @@ func TestModel_FindOneAndDelete(t *testing.T) {
 
 func TestModel_FindOneAndReplace(t *testing.T) {
 	mt := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
-	defer mt.Close()
-
 	mt.Run("test", func(mt *mtest.T) {
 		m := createModel(mt)
 		mt.AddMockResponses(mtest.CreateSuccessResponse(bson.D{
@@ -212,8 +196,6 @@ func TestModel_FindOneAndReplace(t *testing.T) {
 
 func TestModel_FindOneAndUpdate(t *testing.T) {
 	mt := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
-	defer mt.Close()
-
 	mt.Run("test", func(mt *mtest.T) {
 		m := createModel(mt)
 		mt.AddMockResponses(mtest.CreateSuccessResponse(bson.D{

core/stores/mon/options.go

@@ -9,7 +9,11 @@ import (
 
 const defaultTimeout = time.Second * 3
 
-var slowThreshold = syncx.ForAtomicDuration(defaultSlowThreshold)
+var (
+	slowThreshold = syncx.ForAtomicDuration(defaultSlowThreshold)
+	logMon        = syncx.ForAtomicBool(true)
+	logSlowMon    = syncx.ForAtomicBool(true)
+)
 
 type (
 	options = mopt.ClientOptions
@@ -18,6 +22,17 @@ type (
 	Option func(opts *options)
 )
 
+// DisableLog disables logging of mongo commands, includes info and slow logs.
+func DisableLog() {
+	logMon.Set(false)
+	logSlowMon.Set(false)
+}
+
+// DisableInfoLog disables info logging of mongo commands, but keeps slow logs.
+func DisableInfoLog() {
+	logMon.Set(false)
+}
+
 // SetSlowThreshold sets the slow threshold.
 func SetSlowThreshold(threshold time.Duration) {
 	slowThreshold.Set(threshold)

core/stores/mon/options_test.go

@@ -25,3 +25,29 @@ func TestWithTimeout(t *testing.T) {
 	WithTimeout(time.Second)(opts)
 	assert.Equal(t, time.Second, *opts.Timeout)
 }
+
+func TestDisableLog(t *testing.T) {
+	assert.True(t, logMon.True())
+	assert.True(t, logSlowMon.True())
+	defer func() {
+		logMon.Set(true)
+		logSlowMon.Set(true)
+	}()
+
+	DisableLog()
+	assert.False(t, logMon.True())
+	assert.False(t, logSlowMon.True())
+}
+
+func TestDisableInfoLog(t *testing.T) {
+	assert.True(t, logMon.True())
+	assert.True(t, logSlowMon.True())
+	defer func() {
+		logMon.Set(true)
+		logSlowMon.Set(true)
+	}()
+
+	DisableInfoLog()
+	assert.False(t, logMon.True())
+	assert.True(t, logSlowMon.True())
+}

core/stores/mon/trace.go

@@ -2,6 +2,7 @@ package mon
 
 import (
 	"context"
+	"errors"
 
 	"github.com/zeromicro/go-zero/core/trace"
 	"go.mongodb.org/mongo-driver/mongo"
@@ -23,8 +24,8 @@ func startSpan(ctx context.Context, cmd string) (context.Context, oteltrace.Span
 func endSpan(span oteltrace.Span, err error) {
 	defer span.End()
 
-	if err == nil || err == mongo.ErrNoDocuments ||
-		err == mongo.ErrNilValue || err == mongo.ErrNilDocument {
+	if err == nil || errors.Is(err, mongo.ErrNoDocuments) ||
+		errors.Is(err, mongo.ErrNilValue) || errors.Is(err, mongo.ErrNilDocument) {
 		span.SetStatus(codes.Ok, "")
 		return
 	}

core/stores/mon/util.go

@@ -2,6 +2,7 @@ package mon
 
 import (
 	"context"
+	"encoding/json"
 	"strings"
 	"time"
 
@@ -20,8 +21,41 @@ func logDuration(ctx context.Context, name, method string, startTime time.Durati
 	duration := timex.Since(startTime)
 	logger := logx.WithContext(ctx).WithDuration(duration)
 	if err != nil {
-		logger.Infof("mongo(%s) - %s - fail(%s)", name, method, err.Error())
-	} else {
+		logger.Errorf("mongo(%s) - %s - fail(%s)", name, method, err.Error())
+		return
+	}
+
+	if logSlowMon.True() && duration > slowThreshold.Load() {
+		logger.Slowf("[MONGO] mongo(%s) - slowcall - %s - ok", name, method)
+	} else if logMon.True() {
 		logger.Infof("mongo(%s) - %s - ok", name, method)
 	}
 }
+
+func logDurationWithDocs(ctx context.Context, name, method string, startTime time.Duration,
+	err error, docs ...any) {
+	duration := timex.Since(startTime)
+	logger := logx.WithContext(ctx).WithDuration(duration)
+
+	content, jerr := json.Marshal(docs)
+	// jerr should not be non-nil, but we don't care much on this,
+	// if non-nil, we just log without docs.
+	if jerr != nil {
+		if err != nil {
+			logger.Errorf("mongo(%s) - %s - fail(%s)", name, method, err.Error())
+		} else if logSlowMon.True() && duration > slowThreshold.Load() {
+			logger.Slowf("[MONGO] mongo(%s) - slowcall - %s - ok", name, method)
+		} else if logMon.True() {
+			logger.Infof("mongo(%s) - %s - ok", name, method)
+		}
+		return
+	}
+
+	if err != nil {
+		logger.Errorf("mongo(%s) - %s - fail(%s) - %s", name, method, err.Error(), string(content))
+	} else if logSlowMon.True() && duration > slowThreshold.Load() {
+		logger.Slowf("[MONGO] mongo(%s) - slowcall - %s - ok - %s", name, method, string(content))
+	} else if logMon.True() {
+		logger.Infof("mongo(%s) - %s - ok - %s", name, method, string(content))
+	}
+}

core/stores/mon/util_test.go

@@ -4,10 +4,10 @@ import (
 	"context"
 	"errors"
 	"testing"
-	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/zeromicro/go-zero/core/logx/logtest"
+	"github.com/zeromicro/go-zero/core/timex"
 )
 
 func TestFormatAddrs(t *testing.T) {
@@ -42,13 +42,148 @@ func Test_logDuration(t *testing.T) {
 	buf := logtest.NewCollector(t)
 
 	buf.Reset()
-	logDuration(context.Background(), "foo", "bar", time.Millisecond, nil)
+	logDuration(context.Background(), "foo", "bar", timex.Now()-slowThreshold.Load()*2, nil)
+	assert.Contains(t, buf.String(), "foo")
+	assert.Contains(t, buf.String(), "bar")
+	assert.Contains(t, buf.String(), "slow")
+
+	buf.Reset()
+	logDuration(context.Background(), "foo", "bar", timex.Now(), nil)
 	assert.Contains(t, buf.String(), "foo")
 	assert.Contains(t, buf.String(), "bar")
 
 	buf.Reset()
-	logDuration(context.Background(), "foo", "bar", time.Millisecond, errors.New("bar"))
+	logDuration(context.Background(), "foo", "bar", timex.Now(), errors.New("bar"))
+	assert.Contains(t, buf.String(), "foo")
+	assert.Contains(t, buf.String(), "bar")
+	assert.Contains(t, buf.String(), "fail")
+
+	defer func() {
+		logMon.Set(true)
+		logSlowMon.Set(true)
+	}()
+
+	buf.Reset()
+	DisableInfoLog()
+	logDuration(context.Background(), "foo", "bar", timex.Now(), nil)
+	assert.Empty(t, buf.String())
+
+	buf.Reset()
+	logDuration(context.Background(), "foo", "bar", timex.Now()-slowThreshold.Load()*2, nil)
+	assert.Contains(t, buf.String(), "foo")
+	assert.Contains(t, buf.String(), "bar")
+	assert.Contains(t, buf.String(), "slow")
+
+	buf.Reset()
+	DisableLog()
+	logDuration(context.Background(), "foo", "bar", timex.Now(), nil)
+	assert.Empty(t, buf.String())
+
+	buf.Reset()
+	logDuration(context.Background(), "foo", "bar", timex.Now()-slowThreshold.Load()*2, nil)
+	assert.Empty(t, buf.String())
+
+	buf.Reset()
+	logDuration(context.Background(), "foo", "bar", timex.Now(), errors.New("bar"))
 	assert.Contains(t, buf.String(), "foo")
 	assert.Contains(t, buf.String(), "bar")
 	assert.Contains(t, buf.String(), "fail")
 }
+
+func Test_logDurationWithDoc(t *testing.T) {
+	buf := logtest.NewCollector(t)
+	buf.Reset()
+
+	logDurationWithDocs(context.Background(), "foo", "bar", timex.Now()-slowThreshold.Load()*2, nil, make(chan int))
+	assert.Contains(t, buf.String(), "foo")
+	assert.Contains(t, buf.String(), "bar")
+	assert.Contains(t, buf.String(), "slow")
+
+	buf.Reset()
+	logDurationWithDocs(context.Background(), "foo", "bar", timex.Now()-slowThreshold.Load()*2, nil, "{'json': ''}")
+	assert.Contains(t, buf.String(), "foo")
+	assert.Contains(t, buf.String(), "bar")
+	assert.Contains(t, buf.String(), "slow")
+	assert.Contains(t, buf.String(), "json")
+
+	buf.Reset()
+	logDurationWithDocs(context.Background(), "foo", "bar", timex.Now(), nil, make(chan int))
+	assert.Contains(t, buf.String(), "foo")
+	assert.Contains(t, buf.String(), "bar")
+
+	buf.Reset()
+	logDurationWithDocs(context.Background(), "foo", "bar", timex.Now(), nil, "{'json': ''}")
+	assert.Contains(t, buf.String(), "foo")
+	assert.Contains(t, buf.String(), "bar")
+	assert.Contains(t, buf.String(), "json")
+
+	buf.Reset()
+	logDurationWithDocs(context.Background(), "foo", "bar", timex.Now(), errors.New("bar"), make(chan int))
+	assert.Contains(t, buf.String(), "foo")
+	assert.Contains(t, buf.String(), "bar")
+	assert.Contains(t, buf.String(), "fail")
+
+	buf.Reset()
+	logDurationWithDocs(context.Background(), "foo", "bar", timex.Now(), errors.New("bar"), "{'json': ''}")
+	assert.Contains(t, buf.String(), "foo")
+	assert.Contains(t, buf.String(), "bar")
+	assert.Contains(t, buf.String(), "fail")
+	assert.Contains(t, buf.String(), "json")
+
+	defer func() {
+		logMon.Set(true)
+		logSlowMon.Set(true)
+	}()
+
+	buf.Reset()
+	DisableInfoLog()
+	logDurationWithDocs(context.Background(), "foo", "bar", timex.Now(), nil, make(chan int))
+	assert.Empty(t, buf.String())
+
+	buf.Reset()
+	logDurationWithDocs(context.Background(), "foo", "bar", timex.Now(), nil, "{'json': ''}")
+	assert.Empty(t, buf.String())
+
+	buf.Reset()
+	logDurationWithDocs(context.Background(), "foo", "bar", timex.Now()-slowThreshold.Load()*2, nil, make(chan int))
+	assert.Contains(t, buf.String(), "foo")
+	assert.Contains(t, buf.String(), "bar")
+	assert.Contains(t, buf.String(), "slow")
+
+	buf.Reset()
+	logDurationWithDocs(context.Background(), "foo", "bar", timex.Now()-slowThreshold.Load()*2, nil, "{'json': ''}")
+	assert.Contains(t, buf.String(), "foo")
+	assert.Contains(t, buf.String(), "bar")
+	assert.Contains(t, buf.String(), "slow")
+	assert.Contains(t, buf.String(), "json")
+
+	buf.Reset()
+	DisableLog()
+	logDurationWithDocs(context.Background(), "foo", "bar", timex.Now(), nil, make(chan int))
+	assert.Empty(t, buf.String())
+
+	buf.Reset()
+	logDurationWithDocs(context.Background(), "foo", "bar", timex.Now(), nil, "{'json': ''}")
+	assert.Empty(t, buf.String())
+
+	buf.Reset()
+	logDurationWithDocs(context.Background(), "foo", "bar", timex.Now()-slowThreshold.Load()*2, nil, make(chan int))
+	assert.Empty(t, buf.String())
+
+	buf.Reset()
+	logDurationWithDocs(context.Background(), "foo", "bar", timex.Now()-slowThreshold.Load()*2, nil, "{'json': ''}")
+	assert.Empty(t, buf.String())
+
+	buf.Reset()
+	logDurationWithDocs(context.Background(), "foo", "bar", timex.Now(), errors.New("bar"), make(chan int))
+	assert.Contains(t, buf.String(), "foo")
+	assert.Contains(t, buf.String(), "bar")
+	assert.Contains(t, buf.String(), "fail")
+
+	buf.Reset()
+	logDurationWithDocs(context.Background(), "foo", "bar", timex.Now(), errors.New("bar"), "{'json': ''}")
+	assert.Contains(t, buf.String(), "foo")
+	assert.Contains(t, buf.String(), "bar")
+	assert.Contains(t, buf.String(), "fail")
+	assert.Contains(t, buf.String(), "json")
+}

core/stores/monc/cachedmodel_test.go

@@ -17,8 +17,6 @@ import (
 
 func TestNewModel(t *testing.T) {
 	mt := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
-	defer mt.Close()
-
 	mt.Run("test", func(mt *mtest.T) {
 		_, err := newModel("foo", mt.DB.Name(), mt.Coll.Name(), nil)
 		assert.NotNil(mt, err)
@@ -27,8 +25,6 @@ func TestNewModel(t *testing.T) {
 
 func TestModel_DelCache(t *testing.T) {
 	mt := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
-	defer mt.Close()
-
 	mt.Run("test", func(mt *mtest.T) {
 		m := createModel(t, mt)
 		assert.Nil(t, m.cache.Set("foo", "bar"))
@@ -42,8 +38,6 @@ func TestModel_DelCache(t *testing.T) {
 
 func TestModel_DeleteOne(t *testing.T) {
 	mt := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
-	defer mt.Close()
-
 	mt.Run("test", func(mt *mtest.T) {
 		mt.AddMockResponses(mtest.CreateSuccessResponse(bson.D{{Key: "n", Value: 1}}...))
 		m := createModel(t, mt)
@@ -65,8 +59,6 @@ func TestModel_DeleteOne(t *testing.T) {
 
 func TestModel_DeleteOneNoCache(t *testing.T) {
 	mt := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
-	defer mt.Close()
-
 	mt.Run("test", func(mt *mtest.T) {
 		mt.AddMockResponses(mtest.CreateSuccessResponse(bson.D{{Key: "n", Value: 1}}...))
 		m := createModel(t, mt)
@@ -81,8 +73,6 @@ func TestModel_DeleteOneNoCache(t *testing.T) {
 
 func TestModel_FindOne(t *testing.T) {
 	mt := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
-	defer mt.Close()
-
 	mt.Run("test", func(mt *mtest.T) {
 		resp := mtest.CreateCursorResponse(
 			1,
@@ -104,8 +94,6 @@ func TestModel_FindOne(t *testing.T) {
 
 func TestModel_FindOneNoCache(t *testing.T) {
 	mt := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
-	defer mt.Close()
-
 	mt.Run("test", func(mt *mtest.T) {
 		resp := mtest.CreateCursorResponse(
 			1,
@@ -126,8 +114,6 @@ func TestModel_FindOneNoCache(t *testing.T) {
 
 func TestModel_FindOneAndDelete(t *testing.T) {
 	mt := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
-	defer mt.Close()
-
 	mt.Run("test", func(mt *mtest.T) {
 		mt.AddMockResponses(mtest.CreateSuccessResponse(bson.D{
 			{Key: "value", Value: bson.D{{Key: "foo", Value: "bar"}}},
@@ -152,8 +138,6 @@ func TestModel_FindOneAndDelete(t *testing.T) {
 
 func TestModel_FindOneAndDeleteNoCache(t *testing.T) {
 	mt := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
-	defer mt.Close()
-
 	mt.Run("test", func(mt *mtest.T) {
 		mt.AddMockResponses(mtest.CreateSuccessResponse(bson.D{
 			{Key: "value", Value: bson.D{{Key: "foo", Value: "bar"}}},
@@ -169,8 +153,6 @@ func TestModel_FindOneAndDeleteNoCache(t *testing.T) {
 
 func TestModel_FindOneAndReplace(t *testing.T) {
 	mt := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
-	defer mt.Close()
-
 	mt.Run("test", func(mt *mtest.T) {
 		mt.AddMockResponses(mtest.CreateSuccessResponse(bson.D{
 			{Key: "value", Value: bson.D{{Key: "foo", Value: "bar"}}},
@@ -201,8 +183,6 @@ func TestModel_FindOneAndReplace(t *testing.T) {
 
 func TestModel_FindOneAndReplaceNoCache(t *testing.T) {
 	mt := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
-	defer mt.Close()
-
 	mt.Run("test", func(mt *mtest.T) {
 		mt.AddMockResponses(mtest.CreateSuccessResponse(bson.D{
 			{Key: "value", Value: bson.D{{Key: "foo", Value: "bar"}}},
@@ -220,8 +200,6 @@ func TestModel_FindOneAndReplaceNoCache(t *testing.T) {
 
 func TestModel_FindOneAndUpdate(t *testing.T) {
 	mt := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
-	defer mt.Close()
-
 	mt.Run("test", func(mt *mtest.T) {
 		mt.AddMockResponses(mtest.CreateSuccessResponse(bson.D{
 			{Key: "value", Value: bson.D{{Key: "foo", Value: "bar"}}},
@@ -252,8 +230,6 @@ func TestModel_FindOneAndUpdate(t *testing.T) {
 
 func TestModel_FindOneAndUpdateNoCache(t *testing.T) {
 	mt := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
-	defer mt.Close()
-
 	mt.Run("test", func(mt *mtest.T) {
 		mt.AddMockResponses(mtest.CreateSuccessResponse(bson.D{
 			{Key: "value", Value: bson.D{{Key: "foo", Value: "bar"}}},
@@ -271,8 +247,6 @@ func TestModel_FindOneAndUpdateNoCache(t *testing.T) {
 
 func TestModel_GetCache(t *testing.T) {
 	mt := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
-	defer mt.Close()
-
 	mt.Run("test", func(mt *mtest.T) {
 		m := createModel(t, mt)
 		assert.NotNil(t, m.cache)
@@ -285,8 +259,6 @@ func TestModel_GetCache(t *testing.T) {
 
 func TestModel_InsertOne(t *testing.T) {
 	mt := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
-	defer mt.Close()
-
 	mt.Run("test", func(mt *mtest.T) {
 		mt.AddMockResponses(mtest.CreateSuccessResponse(bson.D{
 			{Key: "value", Value: bson.D{{Key: "foo", Value: "bar"}}},
@@ -318,8 +290,6 @@ func TestModel_InsertOne(t *testing.T) {
 
 func TestModel_InsertOneNoCache(t *testing.T) {
 	mt := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
-	defer mt.Close()
-
 	mt.Run("test", func(mt *mtest.T) {
 		mt.AddMockResponses(mtest.CreateSuccessResponse(bson.D{
 			{Key: "value", Value: bson.D{{Key: "foo", Value: "bar"}}},
@@ -335,8 +305,6 @@ func TestModel_InsertOneNoCache(t *testing.T) {
 
 func TestModel_ReplaceOne(t *testing.T) {
 	mt := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
-	defer mt.Close()
-
 	mt.Run("test", func(mt *mtest.T) {
 		mt.AddMockResponses(mtest.CreateSuccessResponse(bson.D{
 			{Key: "value", Value: bson.D{{Key: "foo", Value: "bar"}}},
@@ -368,8 +336,6 @@ func TestModel_ReplaceOne(t *testing.T) {
 
 func TestModel_ReplaceOneNoCache(t *testing.T) {
 	mt := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
-	defer mt.Close()
-
 	mt.Run("test", func(mt *mtest.T) {
 		mt.AddMockResponses(mtest.CreateSuccessResponse(bson.D{
 			{Key: "value", Value: bson.D{{Key: "foo", Value: "bar"}}},
@@ -385,8 +351,6 @@ func TestModel_ReplaceOneNoCache(t *testing.T) {
 
 func TestModel_SetCache(t *testing.T) {
 	mt := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
-	defer mt.Close()
-
 	mt.Run("test", func(mt *mtest.T) {
 		m := createModel(t, mt)
 		assert.Nil(t, m.SetCache("foo", "bar"))
@@ -398,8 +362,6 @@ func TestModel_SetCache(t *testing.T) {
 
 func TestModel_UpdateByID(t *testing.T) {
 	mt := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
-	defer mt.Close()
-
 	mt.Run("test", func(mt *mtest.T) {
 		mt.AddMockResponses(mtest.CreateSuccessResponse(bson.D{
 			{Key: "value", Value: bson.D{{Key: "foo", Value: "bar"}}},
@@ -431,8 +393,6 @@ func TestModel_UpdateByID(t *testing.T) {
 
 func TestModel_UpdateByIDNoCache(t *testing.T) {
 	mt := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
-	defer mt.Close()
-
 	mt.Run("test", func(mt *mtest.T) {
 		mt.AddMockResponses(mtest.CreateSuccessResponse(bson.D{
 			{Key: "value", Value: bson.D{{Key: "foo", Value: "bar"}}},
@@ -448,8 +408,6 @@ func TestModel_UpdateByIDNoCache(t *testing.T) {
 
 func TestModel_UpdateMany(t *testing.T) {
 	mt := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
-	defer mt.Close()
-
 	mt.Run("test", func(mt *mtest.T) {
 		mt.AddMockResponses(mtest.CreateSuccessResponse(bson.D{
 			{Key: "value", Value: bson.D{{Key: "foo", Value: "bar"}}},
@@ -483,8 +441,6 @@ func TestModel_UpdateMany(t *testing.T) {
 
 func TestModel_UpdateManyNoCache(t *testing.T) {
 	mt := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
-	defer mt.Close()
-
 	mt.Run("test", func(mt *mtest.T) {
 		mt.AddMockResponses(mtest.CreateSuccessResponse(bson.D{
 			{Key: "value", Value: bson.D{{Key: "foo", Value: "bar"}}},
@@ -500,8 +456,6 @@ func TestModel_UpdateManyNoCache(t *testing.T) {
 
 func TestModel_UpdateOne(t *testing.T) {
 	mt := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
-	defer mt.Close()
-
 	mt.Run("test", func(mt *mtest.T) {
 		mt.AddMockResponses(mtest.CreateSuccessResponse(bson.D{
 			{Key: "value", Value: bson.D{{Key: "foo", Value: "bar"}}},
@@ -533,8 +487,6 @@ func TestModel_UpdateOne(t *testing.T) {
 
 func TestModel_UpdateOneNoCache(t *testing.T) {
 	mt := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
-	defer mt.Close()
-
 	mt.Run("test", func(mt *mtest.T) {
 		mt.AddMockResponses(mtest.CreateSuccessResponse(bson.D{
 			{Key: "value", Value: bson.D{{Key: "foo", Value: "bar"}}},

core/stores/redis/hook.go

@@ -2,14 +2,14 @@ package redis
 
 import (
 	"context"
+	"errors"
 	"io"
 	"net"
 	"strings"
 	"time"
 
-	red "github.com/go-redis/redis/v8"
+	red "github.com/redis/go-redis/v9"
 	"github.com/zeromicro/go-zero/core/breaker"
-	"github.com/zeromicro/go-zero/core/errorx"
 	"github.com/zeromicro/go-zero/core/logx"
 	"github.com/zeromicro/go-zero/core/mapping"
 	"github.com/zeromicro/go-zero/core/timex"
@@ -23,111 +23,84 @@ import (
 const spanName = "redis"
 
 var (
-	startTimeKey          = contextKey("startTime")
 	durationHook          = hook{}
 	redisCmdsAttributeKey = attribute.Key("redis.cmds")
 )
 
-type (
-	contextKey string
-	hook       struct{}
-)
+type hook struct{}
 
-func (h hook) BeforeProcess(ctx context.Context, cmd red.Cmder) (context.Context, error) {
-	return h.startSpan(context.WithValue(ctx, startTimeKey, timex.Now()), cmd), nil
+func (h hook) DialHook(next red.DialHook) red.DialHook {
+	return next
 }
 
-func (h hook) AfterProcess(ctx context.Context, cmd red.Cmder) error {
-	err := cmd.Err()
-	h.endSpan(ctx, err)
+func (h hook) ProcessHook(next red.ProcessHook) red.ProcessHook {
+	return func(ctx context.Context, cmd red.Cmder) error {
+		start := timex.Now()
+		ctx, endSpan := h.startSpan(ctx, cmd)
 
-	val := ctx.Value(startTimeKey)
-	if val == nil {
-		return nil
-	}
+		err := next(ctx, cmd)
 
-	start, ok := val.(time.Duration)
-	if !ok {
-		return nil
-	}
+		endSpan(err)
+		duration := timex.Since(start)
 
-	duration := timex.Since(start)
-	if duration > slowThreshold.Load() {
-		logDuration(ctx, []red.Cmder{cmd}, duration)
-	}
-
-	metricReqDur.Observe(duration.Milliseconds(), cmd.Name())
-	if msg := formatError(err); len(msg) > 0 {
-		metricReqErr.Inc(cmd.Name(), msg)
-	}
-
-	return nil
-}
-
-func (h hook) BeforeProcessPipeline(ctx context.Context, cmds []red.Cmder) (context.Context, error) {
-	if len(cmds) == 0 {
-		return ctx, nil
-	}
-
-	return h.startSpan(context.WithValue(ctx, startTimeKey, timex.Now()), cmds...), nil
-}
-
-func (h hook) AfterProcessPipeline(ctx context.Context, cmds []red.Cmder) error {
-	if len(cmds) == 0 {
-		return nil
-	}
-
-	batchError := errorx.BatchError{}
-	for _, cmd := range cmds {
-		err := cmd.Err()
-		if err == nil {
-			continue
+		if duration > slowThreshold.Load() {
+			logDuration(ctx, []red.Cmder{cmd}, duration)
+			metricSlowCount.Inc(cmd.Name())
 		}
 
-		batchError.Add(err)
-	}
-	h.endSpan(ctx, batchError.Err())
+		metricReqDur.Observe(duration.Milliseconds(), cmd.Name())
+		if msg := formatError(err); len(msg) > 0 {
+			metricReqErr.Inc(cmd.Name(), msg)
+		}
 
-	val := ctx.Value(startTimeKey)
-	if val == nil {
-		return nil
+		return err
 	}
+}
 
-	start, ok := val.(time.Duration)
-	if !ok {
-		return nil
+func (h hook) ProcessPipelineHook(next red.ProcessPipelineHook) red.ProcessPipelineHook {
+	return func(ctx context.Context, cmds []red.Cmder) error {
+		if len(cmds) == 0 {
+			return next(ctx, cmds)
+		}
+
+		start := timex.Now()
+		ctx, endSpan := h.startSpan(ctx, cmds...)
+
+		err := next(ctx, cmds)
+
+		endSpan(err)
+		duration := timex.Since(start)
+		if duration > slowThreshold.Load()*time.Duration(len(cmds)) {
+			logDuration(ctx, cmds, duration)
+		}
+
+		metricReqDur.Observe(duration.Milliseconds(), "Pipeline")
+		if msg := formatError(err); len(msg) > 0 {
+			metricReqErr.Inc("Pipeline", msg)
+		}
+
+		return err
 	}
-
-	duration := timex.Since(start)
-	if duration > slowThreshold.Load()*time.Duration(len(cmds)) {
-		logDuration(ctx, cmds, duration)
-	}
-
-	metricReqDur.Observe(duration.Milliseconds(), "Pipeline")
-	if msg := formatError(batchError.Err()); len(msg) > 0 {
-		metricReqErr.Inc("Pipeline", msg)
-	}
-
-	return nil
 }
 
 func formatError(err error) string {
-	if err == nil || err == red.Nil {
+	if err == nil || errors.Is(err, red.Nil) {
 		return ""
 	}
 
-	opErr, ok := err.(*net.OpError)
+	var opErr *net.OpError
+	ok := errors.As(err, &opErr)
 	if ok && opErr.Timeout() {
 		return "timeout"
 	}
 
-	switch err {
-	case io.EOF:
+	switch {
+	case err == io.EOF:
 		return "eof"
-	case context.DeadlineExceeded:
+	case errors.Is(err, context.DeadlineExceeded):
 		return "context deadline"
-	case breaker.ErrServiceUnavailable:
-		return "breaker"
+	case errors.Is(err, breaker.ErrServiceUnavailable):
+		return "breaker open"
 	default:
 		return "unexpected error"
 	}
@@ -151,7 +124,7 @@ func logDuration(ctx context.Context, cmds []red.Cmder, duration time.Duration)
 	logx.WithContext(ctx).WithDuration(duration).Slowf("[REDIS] slowcall on executing: %s", buf.String())
 }
 
-func (h hook) startSpan(ctx context.Context, cmds ...red.Cmder) context.Context {
+func (h hook) startSpan(ctx context.Context, cmds ...red.Cmder) (context.Context, func(err error)) {
 	tracer := trace.TracerFromContext(ctx)
 
 	ctx, span := tracer.Start(ctx,
@@ -165,18 +138,15 @@ func (h hook) startSpan(ctx context.Context, cmds ...red.Cmder) context.Context
 	}
 	span.SetAttributes(redisCmdsAttributeKey.StringSlice(cmdStrs))
 
-	return ctx
-}
+	return ctx, func(err error) {
+		defer span.End()
 
-func (h hook) endSpan(ctx context.Context, err error) {
-	span := oteltrace.SpanFromContext(ctx)
-	defer span.End()
+		if err == nil || errors.Is(err, red.Nil) {
+			span.SetStatus(codes.Ok, "")
+			return
+		}
 
-	if err == nil || err == red.Nil {
-		span.SetStatus(codes.Ok, "")
-		return
+		span.SetStatus(codes.Error, err.Error())
+		span.RecordError(err)
 	}
-
-	span.SetStatus(codes.Error, err.Error())
-	span.RecordError(err)
 }

core/stores/redis/hook_test.go

@@ -4,168 +4,103 @@ import (
 	"context"
 	"errors"
 	"io"
-	"log"
 	"net"
 	"strings"
 	"testing"
 	"time"
 
-	red "github.com/go-redis/redis/v8"
+	red "github.com/redis/go-redis/v9"
 	"github.com/stretchr/testify/assert"
 	"github.com/zeromicro/go-zero/core/breaker"
 	"github.com/zeromicro/go-zero/core/logx/logtest"
-	ztrace "github.com/zeromicro/go-zero/core/trace"
+	"github.com/zeromicro/go-zero/core/trace/tracetest"
 	tracesdk "go.opentelemetry.io/otel/trace"
 )
 
 func TestHookProcessCase1(t *testing.T) {
-	ztrace.StartAgent(ztrace.Config{
-		Name:     "go-zero-test",
-		Endpoint: "http://localhost:14268/api/traces",
-		Batcher:  "jaeger",
-		Sampler:  1.0,
-	})
-	defer ztrace.StopAgent()
+	tracetest.NewInMemoryExporter(t)
+	w := logtest.NewCollector(t)
 
-	writer := log.Writer()
-	var buf strings.Builder
-	log.SetOutput(&buf)
-	defer log.SetOutput(writer)
-
-	ctx, err := durationHook.BeforeProcess(context.Background(), red.NewCmd(context.Background()))
+	err := durationHook.ProcessHook(func(ctx context.Context, cmd red.Cmder) error {
+		assert.Equal(t, "redis", tracesdk.SpanFromContext(ctx).(interface{ Name() string }).Name())
+		return nil
+	})(context.Background(), red.NewCmd(context.Background()))
 	if err != nil {
 		t.Fatal(err)
 	}
 
-	assert.Nil(t, durationHook.AfterProcess(ctx, red.NewCmd(context.Background())))
-	assert.False(t, strings.Contains(buf.String(), "slow"))
-	assert.Equal(t, "redis", tracesdk.SpanFromContext(ctx).(interface{ Name() string }).Name())
+	assert.False(t, strings.Contains(w.String(), "slow"))
 }
 
 func TestHookProcessCase2(t *testing.T) {
-	ztrace.StartAgent(ztrace.Config{
-		Name:     "go-zero-test",
-		Endpoint: "http://localhost:14268/api/traces",
-		Batcher:  "jaeger",
-		Sampler:  1.0,
-	})
-	defer ztrace.StopAgent()
-
+	tracetest.NewInMemoryExporter(t)
 	w := logtest.NewCollector(t)
 
-	ctx, err := durationHook.BeforeProcess(context.Background(), red.NewCmd(context.Background()))
+	err := durationHook.ProcessHook(func(ctx context.Context, cmd red.Cmder) error {
+		assert.Equal(t, "redis", tracesdk.SpanFromContext(ctx).(interface{ Name() string }).Name())
+		time.Sleep(slowThreshold.Load() + time.Millisecond)
+		return nil
+	})(context.Background(), red.NewCmd(context.Background()))
 	if err != nil {
 		t.Fatal(err)
 	}
-	assert.Equal(t, "redis", tracesdk.SpanFromContext(ctx).(interface{ Name() string }).Name())
 
-	time.Sleep(slowThreshold.Load() + time.Millisecond)
-
-	assert.Nil(t, durationHook.AfterProcess(ctx, red.NewCmd(context.Background(), "foo", "bar")))
 	assert.True(t, strings.Contains(w.String(), "slow"))
 	assert.True(t, strings.Contains(w.String(), "trace"))
 	assert.True(t, strings.Contains(w.String(), "span"))
 }
 
-func TestHookProcessCase3(t *testing.T) {
-	writer := log.Writer()
-	var buf strings.Builder
-	log.SetOutput(&buf)
-	defer log.SetOutput(writer)
-
-	assert.Nil(t, durationHook.AfterProcess(context.Background(), red.NewCmd(context.Background())))
-	assert.True(t, buf.Len() == 0)
-}
-
-func TestHookProcessCase4(t *testing.T) {
-	writer := log.Writer()
-	var buf strings.Builder
-	log.SetOutput(&buf)
-	defer log.SetOutput(writer)
-
-	ctx := context.WithValue(context.Background(), startTimeKey, "foo")
-	assert.Nil(t, durationHook.AfterProcess(ctx, red.NewCmd(context.Background())))
-	assert.True(t, buf.Len() == 0)
-}
-
 func TestHookProcessPipelineCase1(t *testing.T) {
-	writer := log.Writer()
-	var buf strings.Builder
-	log.SetOutput(&buf)
-	defer log.SetOutput(writer)
+	tracetest.NewInMemoryExporter(t)
+	w := logtest.NewCollector(t)
 
-	_, err := durationHook.BeforeProcessPipeline(context.Background(), []red.Cmder{})
+	err := durationHook.ProcessPipelineHook(func(ctx context.Context, cmds []red.Cmder) error {
+		return nil
+	})(context.Background(), nil)
 	assert.NoError(t, err)
-	ctx, err := durationHook.BeforeProcessPipeline(context.Background(), []red.Cmder{
+
+	err = durationHook.ProcessPipelineHook(func(ctx context.Context, cmds []red.Cmder) error {
+		assert.Equal(t, "redis", tracesdk.SpanFromContext(ctx).(interface{ Name() string }).Name())
+		return nil
+	})(context.Background(), []red.Cmder{
 		red.NewCmd(context.Background()),
 	})
 	assert.NoError(t, err)
-	assert.Equal(t, "redis", tracesdk.SpanFromContext(ctx).(interface{ Name() string }).Name())
 
-	assert.NoError(t, durationHook.AfterProcessPipeline(ctx, []red.Cmder{}))
-	assert.NoError(t, durationHook.AfterProcessPipeline(ctx, []red.Cmder{
-		red.NewCmd(context.Background()),
-	}))
-	assert.False(t, strings.Contains(buf.String(), "slow"))
+	assert.False(t, strings.Contains(w.String(), "slow"))
 }
 
 func TestHookProcessPipelineCase2(t *testing.T) {
-	ztrace.StartAgent(ztrace.Config{
-		Name:     "go-zero-test",
-		Endpoint: "http://localhost:14268/api/traces",
-		Batcher:  "jaeger",
-		Sampler:  1.0,
-	})
-	defer ztrace.StopAgent()
-
+	tracetest.NewInMemoryExporter(t)
 	w := logtest.NewCollector(t)
 
-	ctx, err := durationHook.BeforeProcessPipeline(context.Background(), []red.Cmder{
+	err := durationHook.ProcessPipelineHook(func(ctx context.Context, cmds []red.Cmder) error {
+		assert.Equal(t, "redis", tracesdk.SpanFromContext(ctx).(interface{ Name() string }).Name())
+		time.Sleep(slowThreshold.Load() + time.Millisecond)
+		return nil
+	})(context.Background(), []red.Cmder{
 		red.NewCmd(context.Background()),
 	})
 	assert.NoError(t, err)
-	assert.Equal(t, "redis", tracesdk.SpanFromContext(ctx).(interface{ Name() string }).Name())
 
-	time.Sleep(slowThreshold.Load() + time.Millisecond)
-
-	assert.Nil(t, durationHook.AfterProcessPipeline(ctx, []red.Cmder{
-		red.NewCmd(context.Background(), "foo", "bar"),
-	}))
 	assert.True(t, strings.Contains(w.String(), "slow"))
 	assert.True(t, strings.Contains(w.String(), "trace"))
 	assert.True(t, strings.Contains(w.String(), "span"))
 }
 
 func TestHookProcessPipelineCase3(t *testing.T) {
-	w := logtest.NewCollector(t)
+	te := tracetest.NewInMemoryExporter(t)
 
-	assert.Nil(t, durationHook.AfterProcessPipeline(context.Background(), []red.Cmder{
+	err := durationHook.ProcessPipelineHook(func(ctx context.Context, cmds []red.Cmder) error {
+		assert.Equal(t, "redis", tracesdk.SpanFromContext(ctx).(interface{ Name() string }).Name())
+		return assert.AnError
+	})(context.Background(), []red.Cmder{
 		red.NewCmd(context.Background()),
-	}))
-	assert.True(t, len(w.String()) == 0)
-}
-
-func TestHookProcessPipelineCase4(t *testing.T) {
-	w := logtest.NewCollector(t)
-
-	ctx := context.WithValue(context.Background(), startTimeKey, "foo")
-	assert.Nil(t, durationHook.AfterProcessPipeline(ctx, []red.Cmder{
-		red.NewCmd(context.Background()),
-	}))
-	assert.True(t, len(w.String()) == 0)
-}
-
-func TestHookProcessPipelineCase5(t *testing.T) {
-	writer := log.Writer()
-	var buf strings.Builder
-	log.SetOutput(&buf)
-	defer log.SetOutput(writer)
-
-	ctx := context.WithValue(context.Background(), startTimeKey, "foo")
-	assert.Nil(t, durationHook.AfterProcessPipeline(ctx, []red.Cmder{
-		red.NewCmd(context.Background()),
-	}))
-	assert.True(t, buf.Len() == 0)
+	})
+	assert.ErrorIs(t, err, assert.AnError)
+	traceLogs := te.GetSpans().Snapshots()[0]
+	assert.Equal(t, "redis", traceLogs.Name())
+	assert.Equal(t, assert.AnError.Error(), traceLogs.Events()[0].Attributes[1].Value.AsString(), "trace should record error")
 }
 
 func TestLogDuration(t *testing.T) {
@@ -203,7 +138,7 @@ func TestFormatError(t *testing.T) {
 	assert.Equal(t, "context deadline", formatError(context.DeadlineExceeded))
 
 	// Test case: err is breaker.ErrServiceUnavailable
-	assert.Equal(t, "breaker", formatError(breaker.ErrServiceUnavailable))
+	assert.Equal(t, "breaker open", formatError(breaker.ErrServiceUnavailable))
 
 	// Test case: err is unknown
 	assert.Equal(t, "unexpected error", formatError(errors.New("some error")))

core/stores/redis/metrics.go

@@ -1,6 +1,12 @@
 package redis
 
-import "github.com/zeromicro/go-zero/core/metric"
+import (
+	"sync"
+
+	"github.com/prometheus/client_golang/prometheus"
+	red "github.com/redis/go-redis/v9"
+	"github.com/zeromicro/go-zero/core/metric"
+)
 
 const namespace = "redis_client"
 
@@ -11,7 +17,7 @@ var (
 		Name:      "duration_ms",
 		Help:      "redis client requests duration(ms).",
 		Labels:    []string{"command"},
-		Buckets:   []float64{5, 10, 25, 50, 100, 250, 500, 1000, 2500},
+		Buckets:   []float64{0.25, 0.5, 1, 1.5, 2, 3, 5, 10, 25, 50, 100, 250, 500, 1000, 2000, 5000, 10000, 15000},
 	})
 	metricReqErr = metric.NewCounterVec(&metric.CounterVecOpts{
 		Namespace: namespace,
@@ -20,4 +26,162 @@ var (
 		Help:      "redis client requests error count.",
 		Labels:    []string{"command", "error"},
 	})
+	metricSlowCount = metric.NewCounterVec(&metric.CounterVecOpts{
+		Namespace: namespace,
+		Subsystem: "requests",
+		Name:      "slow_total",
+		Help:      "redis client requests slow count.",
+		Labels:    []string{"command"},
+	})
+
+	connLabels                         = []string{"key", "client_type"}
+	connCollector                      = newCollector()
+	_             prometheus.Collector = (*collector)(nil)
 )
+
+type (
+	statGetter struct {
+		clientType string
+		key        string
+		poolSize   int
+		poolStats  func() *red.PoolStats
+	}
+
+	// collector collects statistics from a redis client.
+	// It implements the prometheus.Collector interface.
+	collector struct {
+		hitDesc     *prometheus.Desc
+		missDesc    *prometheus.Desc
+		timeoutDesc *prometheus.Desc
+		totalDesc   *prometheus.Desc
+		idleDesc    *prometheus.Desc
+		staleDesc   *prometheus.Desc
+		maxDesc     *prometheus.Desc
+
+		clients []*statGetter
+		lock    sync.Mutex
+	}
+)
+
+func newCollector() *collector {
+	c := &collector{
+		hitDesc: prometheus.NewDesc(
+			prometheus.BuildFQName(namespace, "", "pool_hit_total"),
+			"Number of times a connection was found in the pool",
+			connLabels, nil,
+		),
+		missDesc: prometheus.NewDesc(
+			prometheus.BuildFQName(namespace, "", "pool_miss_total"),
+			"Number of times a connection was not found in the pool",
+			connLabels, nil,
+		),
+		timeoutDesc: prometheus.NewDesc(
+			prometheus.BuildFQName(namespace, "", "pool_timeout_total"),
+			"Number of times a timeout occurred when looking for a connection in the pool",
+			connLabels, nil,
+		),
+		totalDesc: prometheus.NewDesc(
+			prometheus.BuildFQName(namespace, "", "pool_conn_total_current"),
+			"Current number of connections in the pool",
+			connLabels, nil,
+		),
+		idleDesc: prometheus.NewDesc(
+			prometheus.BuildFQName(namespace, "", "pool_conn_idle_current"),
+			"Current number of idle connections in the pool",
+			connLabels, nil,
+		),
+		staleDesc: prometheus.NewDesc(
+			prometheus.BuildFQName(namespace, "", "pool_conn_stale_total"),
+			"Number of times a connection was removed from the pool because it was stale",
+			connLabels, nil,
+		),
+		maxDesc: prometheus.NewDesc(
+			prometheus.BuildFQName(namespace, "", "pool_conn_max"),
+			"Max number of connections in the pool",
+			connLabels, nil,
+		),
+	}
+
+	prometheus.MustRegister(c)
+
+	return c
+}
+
+// Describe implements the prometheus.Collector interface.
+func (s *collector) Describe(descs chan<- *prometheus.Desc) {
+	descs <- s.hitDesc
+	descs <- s.missDesc
+	descs <- s.timeoutDesc
+	descs <- s.totalDesc
+	descs <- s.idleDesc
+	descs <- s.staleDesc
+	descs <- s.maxDesc
+}
+
+// Collect implements the prometheus.Collector interface.
+func (s *collector) Collect(metrics chan<- prometheus.Metric) {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+
+	for _, client := range s.clients {
+		key, clientType := client.key, client.clientType
+		stats := client.poolStats()
+
+		metrics <- prometheus.MustNewConstMetric(
+			s.hitDesc,
+			prometheus.CounterValue,
+			float64(stats.Hits),
+			key,
+			clientType,
+		)
+		metrics <- prometheus.MustNewConstMetric(
+			s.missDesc,
+			prometheus.CounterValue,
+			float64(stats.Misses),
+			key,
+			clientType,
+		)
+		metrics <- prometheus.MustNewConstMetric(
+			s.timeoutDesc,
+			prometheus.CounterValue,
+			float64(stats.Timeouts),
+			key,
+			clientType,
+		)
+		metrics <- prometheus.MustNewConstMetric(
+			s.totalDesc,
+			prometheus.GaugeValue,
+			float64(stats.TotalConns),
+			key,
+			clientType,
+		)
+		metrics <- prometheus.MustNewConstMetric(
+			s.idleDesc,
+			prometheus.GaugeValue,
+			float64(stats.IdleConns),
+			key,
+			clientType,
+		)
+		metrics <- prometheus.MustNewConstMetric(
+			s.staleDesc,
+			prometheus.CounterValue,
+			float64(stats.StaleConns),
+			key,
+			clientType,
+		)
+		metrics <- prometheus.MustNewConstMetric(
+			s.maxDesc,
+			prometheus.CounterValue,
+			float64(client.poolSize),
+			key,
+			clientType,
+		)
+	}
+}
+
+func (s *collector) registerClient(client *statGetter) {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+
+	s.clients = append(s.clients, client)
+}

core/stores/redis/metrics_test.go

@@ -0,0 +1,130 @@
+package redis
+
+import (
+	"io"
+	"net/http"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/testutil"
+	red "github.com/redis/go-redis/v9"
+	"github.com/stretchr/testify/assert"
+	"github.com/zeromicro/go-zero/core/conf"
+	"github.com/zeromicro/go-zero/internal/devserver"
+)
+
+func TestRedisMetric(t *testing.T) {
+	cfg := devserver.Config{}
+	_ = conf.FillDefault(&cfg)
+	server := devserver.NewServer(cfg)
+	server.StartAsync()
+	time.Sleep(time.Second)
+
+	metricReqDur.Observe(8, "test-cmd")
+	metricReqErr.Inc("test-cmd", "internal-error")
+	metricSlowCount.Inc("test-cmd")
+
+	url := "http://127.0.0.1:6060/metrics"
+	resp, err := http.Get(url)
+	assert.Nil(t, err)
+	defer resp.Body.Close()
+	s, err := io.ReadAll(resp.Body)
+	assert.Nil(t, err)
+	content := string(s)
+	assert.Contains(t, content, "redis_client_requests_duration_ms_sum{command=\"test-cmd\"} 8\n")
+	assert.Contains(t, content, "redis_client_requests_duration_ms_count{command=\"test-cmd\"} 1\n")
+	assert.Contains(t, content, "redis_client_requests_error_total{command=\"test-cmd\",error=\"internal-error\"} 1\n")
+	assert.Contains(t, content, "redis_client_requests_slow_total{command=\"test-cmd\"} 1\n")
+}
+
+func Test_newCollector(t *testing.T) {
+	prometheus.Unregister(connCollector)
+	c := newCollector()
+	c.registerClient(&statGetter{
+		clientType: "node",
+		key:        "test1",
+		poolSize:   10,
+		poolStats: func() *red.PoolStats {
+			return &red.PoolStats{
+				Hits:       10000,
+				Misses:     10,
+				Timeouts:   5,
+				TotalConns: 100,
+				IdleConns:  20,
+				StaleConns: 1,
+			}
+		},
+	})
+	c.registerClient(&statGetter{
+		clientType: "node",
+		key:        "test2",
+		poolSize:   11,
+		poolStats: func() *red.PoolStats {
+			return &red.PoolStats{
+				Hits:       10001,
+				Misses:     11,
+				Timeouts:   6,
+				TotalConns: 101,
+				IdleConns:  21,
+				StaleConns: 2,
+			}
+		},
+	})
+	c.registerClient(&statGetter{
+		clientType: "cluster",
+		key:        "test3",
+		poolSize:   5,
+		poolStats: func() *red.PoolStats {
+			return &red.PoolStats{
+				Hits:       20000,
+				Misses:     20,
+				Timeouts:   10,
+				TotalConns: 200,
+				IdleConns:  40,
+				StaleConns: 2,
+			}
+		},
+	})
+	val := `
+		# HELP redis_client_pool_conn_idle_current Current number of idle connections in the pool
+		# TYPE redis_client_pool_conn_idle_current gauge
+		redis_client_pool_conn_idle_current{client_type="cluster",key="test3"} 40
+		redis_client_pool_conn_idle_current{client_type="node",key="test1"} 20
+		redis_client_pool_conn_idle_current{client_type="node",key="test2"} 21
+	 	# HELP redis_client_pool_conn_max Max number of connections in the pool
+	 	# TYPE redis_client_pool_conn_max counter
+		redis_client_pool_conn_max{client_type="cluster",key="test3"} 5
+	 	redis_client_pool_conn_max{client_type="node",key="test1"} 10
+		redis_client_pool_conn_max{client_type="node",key="test2"} 11
+	 	# HELP redis_client_pool_conn_stale_total Number of times a connection was removed from the pool because it was stale
+	 	# TYPE redis_client_pool_conn_stale_total counter
+		redis_client_pool_conn_stale_total{client_type="cluster",key="test3"} 2
+	 	redis_client_pool_conn_stale_total{client_type="node",key="test1"} 1
+		redis_client_pool_conn_stale_total{client_type="node",key="test2"} 2
+	 	# HELP redis_client_pool_conn_total_current Current number of connections in the pool
+	 	# TYPE redis_client_pool_conn_total_current gauge
+		redis_client_pool_conn_total_current{client_type="cluster",key="test3"} 200
+	 	redis_client_pool_conn_total_current{client_type="node",key="test1"} 100
+		redis_client_pool_conn_total_current{client_type="node",key="test2"} 101
+	 	# HELP redis_client_pool_hit_total Number of times a connection was found in the pool
+	 	# TYPE redis_client_pool_hit_total counter
+		redis_client_pool_hit_total{client_type="cluster",key="test3"} 20000
+	 	redis_client_pool_hit_total{client_type="node",key="test1"} 10000
+		redis_client_pool_hit_total{client_type="node",key="test2"} 10001
+	 	# HELP redis_client_pool_miss_total Number of times a connection was not found in the pool
+	 	# TYPE redis_client_pool_miss_total counter
+		redis_client_pool_miss_total{client_type="cluster",key="test3"} 20
+	 	redis_client_pool_miss_total{client_type="node",key="test1"} 10
+		redis_client_pool_miss_total{client_type="node",key="test2"} 11
+	 	# HELP redis_client_pool_timeout_total Number of times a timeout occurred when looking for a connection in the pool
+	 	# TYPE redis_client_pool_timeout_total counter
+		redis_client_pool_timeout_total{client_type="cluster",key="test3"} 10
+	 	redis_client_pool_timeout_total{client_type="node",key="test1"} 5
+		redis_client_pool_timeout_total{client_type="node",key="test2"} 6
+`
+
+	err := testutil.CollectAndCompare(c, strings.NewReader(val))
+	assert.NoError(t, err)
+}

core/stores/redis/redis.go

@@ -7,7 +7,7 @@ import (
 	"strconv"
 	"time"
 
-	red "github.com/go-redis/redis/v8"
+	red "github.com/redis/go-redis/v9"
 	"github.com/zeromicro/go-zero/core/breaker"
 	"github.com/zeromicro/go-zero/core/errorx"
 	"github.com/zeromicro/go-zero/core/logx"
@@ -64,6 +64,7 @@ type (
 	// RedisNode interface represents a redis node.
 	RedisNode interface {
 		red.Cmdable
+		red.BitMapCmdable
 	}
 
 	// GeoLocation is used with GeoAdd to add geospatial location.
@@ -467,6 +468,33 @@ func (s *Redis) ExistsCtx(ctx context.Context, key string) (val bool, err error)
 	return
 }
 
+// ExistsMany is the implementation of redis exists command.
+// checks the existence of multiple keys in Redis using the EXISTS command.
+func (s *Redis) ExistsMany(keys ...string) (int64, error) {
+	return s.ExistsManyCtx(context.Background(), keys...)
+}
+
+// ExistsManyCtx is the implementation of redis exists command.
+// checks the existence of multiple keys in Redis using the EXISTS command.
+func (s *Redis) ExistsManyCtx(ctx context.Context, keys ...string) (val int64, err error) {
+	err = s.brk.DoWithAcceptable(func() error {
+		conn, err := getRedis(s)
+		if err != nil {
+			return err
+		}
+
+		v, err := conn.Exists(ctx, keys...).Result()
+		if err != nil {
+			return err
+		}
+
+		val = v
+		return nil
+	}, acceptable)
+
+	return
+}
+
 // Expire is the implementation of redis expire command.
 func (s *Redis) Expire(key string, seconds int) error {
 	return s.ExpireCtx(context.Background(), key, seconds)
@@ -1993,7 +2021,13 @@ func (s *Redis) TtlCtx(ctx context.Context, key string) (val int, err error) {
 			return err
 		}
 
-		val = int(duration / time.Second)
+		if duration >= 0 {
+			val = int(duration / time.Second)
+		} else {
+			// -2 means key does not exist
+			// -1 means key exists but has no expire
+			val = int(duration)
+		}
 		return nil
 	}, acceptable)
 
@@ -2025,7 +2059,7 @@ func (s *Redis) ZaddFloatCtx(ctx context.Context, key string, score float64, val
 			return err
 		}
 
-		v, err := conn.ZAdd(ctx, key, &red.Z{
+		v, err := conn.ZAdd(ctx, key, red.Z{
 			Score:  score,
 			Member: value,
 		}).Result()
@@ -2053,9 +2087,9 @@ func (s *Redis) ZaddsCtx(ctx context.Context, key string, ps ...Pair) (val int64
 			return err
 		}
 
-		var zs []*red.Z
+		var zs []red.Z
 		for _, p := range ps {
-			z := &red.Z{Score: float64(p.Score), Member: p.Key}
+			z := red.Z{Score: float64(p.Score), Member: p.Key}
 			zs = append(zs, z)
 		}
 
@@ -2849,7 +2883,7 @@ func withHook(hook red.Hook) Option {
 }
 
 func acceptable(err error) bool {
-	return err == nil || err == red.Nil || err == context.Canceled
+	return err == nil || err == red.Nil || errors.Is(err, context.Canceled)
 }
 
 func getRedis(r *Redis) (RedisNode, error) {

core/stores/redis/redis_test.go

@@ -10,29 +10,34 @@ import (
 	"time"
 
 	"github.com/alicebob/miniredis/v2"
-	red "github.com/go-redis/redis/v8"
+	red "github.com/redis/go-redis/v9"
 	"github.com/stretchr/testify/assert"
 	"github.com/zeromicro/go-zero/core/logx"
 	"github.com/zeromicro/go-zero/core/stringx"
 )
 
 type myHook struct {
-	red.Hook
 	includePing bool
 }
 
 var _ red.Hook = myHook{}
 
-func (m myHook) BeforeProcess(ctx context.Context, cmd red.Cmder) (context.Context, error) {
-	return ctx, nil
+func (m myHook) DialHook(next red.DialHook) red.DialHook {
+	return next
 }
 
-func (m myHook) AfterProcess(ctx context.Context, cmd red.Cmder) error {
-	// skip ping cmd
-	if cmd.Name() == "ping" && !m.includePing {
-		return nil
+func (m myHook) ProcessPipelineHook(next red.ProcessPipelineHook) red.ProcessPipelineHook {
+	return next
+}
+
+func (m myHook) ProcessHook(next red.ProcessHook) red.ProcessHook {
+	return func(ctx context.Context, cmd red.Cmder) error {
+		// skip ping cmd
+		if cmd.Name() == "ping" && !m.includePing {
+			return next(ctx, cmd)
+		}
+		return errors.New("hook error")
 	}
-	return errors.New("hook error")
 }
 
 func TestNewRedis(t *testing.T) {
@@ -224,6 +229,36 @@ func TestRedisTLS_Exists(t *testing.T) {
 	})
 }
 
+func TestRedis_ExistsMany(t *testing.T) {
+	runOnRedis(t, func(client *Redis) {
+		// Attempt to create a new Redis instance with an incorrect type and call ExistsMany
+		_, err := New(client.Addr, badType()).ExistsMany("key1", "key2")
+		assert.NotNil(t, err)
+
+		// Check if key1 and key2 exist, expecting that they do not
+		val, err := client.ExistsMany("key1", "key2")
+		assert.Nil(t, err)
+		assert.Equal(t, int64(0), val)
+
+		// Set the value for key1 and check if key1 exists
+		assert.Nil(t, client.Set("key1", "value1"))
+		val, err = client.ExistsMany("key1")
+		assert.Nil(t, err)
+		assert.Equal(t, int64(1), val)
+
+		// Set the value for key2 and check if key1 and key2 exist
+		assert.Nil(t, client.Set("key2", "value2"))
+		val, err = client.ExistsMany("key1", "key2")
+		assert.Nil(t, err)
+		assert.Equal(t, int64(2), val)
+
+		// Check if key1, key2, and a non-existent key3 exist, expecting that key1 and key2 do
+		val, err = client.ExistsMany("key1", "key2", "key3")
+		assert.Nil(t, err)
+		assert.Equal(t, int64(2), val)
+	})
+}
+
 func TestRedis_Eval(t *testing.T) {
 	runOnRedis(t, func(client *Redis) {
 		_, err := New(client.Addr, badType()).Eval(`redis.call("EXISTS", KEYS[1])`, []string{"notexist"})
@@ -1579,7 +1614,7 @@ func TestRedis_Pipelined(t *testing.T) {
 			func(pipe Pipeliner) error {
 				pipe.Incr(context.Background(), "pipelined_counter")
 				pipe.Expire(context.Background(), "pipelined_counter", time.Hour)
-				pipe.ZAdd(context.Background(), "zadd", &Z{Score: 12, Member: "zadd"})
+				pipe.ZAdd(context.Background(), "zadd", Z{Score: 12, Member: "zadd"})
 				return nil
 			},
 		)

core/stores/redis/redisblockingnode.go

@@ -3,7 +3,7 @@ package redis
 import (
 	"fmt"
 
-	red "github.com/go-redis/redis/v8"
+	red "github.com/redis/go-redis/v9"
 	"github.com/zeromicro/go-zero/core/logx"
 )
 

core/stores/redis/redisclientmanager.go

@@ -3,8 +3,9 @@ package redis
 import (
 	"crypto/tls"
 	"io"
+	"runtime"
 
-	red "github.com/go-redis/redis/v8"
+	red "github.com/redis/go-redis/v9"
 	"github.com/zeromicro/go-zero/core/syncx"
 )
 
@@ -14,7 +15,11 @@ const (
 	idleConns       = 8
 )
 
-var clientManager = syncx.NewResourceManager()
+var (
+	clientManager = syncx.NewResourceManager()
+	// nodePoolSize is default pool size for node type of redis.
+	nodePoolSize = 10 * runtime.GOMAXPROCS(0)
+)
 
 func getClient(r *Redis) (*red.Client, error) {
 	val, err := clientManager.GetResource(r.Addr, func() (io.Closer, error) {
@@ -37,6 +42,15 @@ func getClient(r *Redis) (*red.Client, error) {
 			store.AddHook(hook)
 		}
 
+		connCollector.registerClient(&statGetter{
+			clientType: NodeType,
+			key:        r.Addr,
+			poolSize:   nodePoolSize,
+			poolStats: func() *red.PoolStats {
+				return store.PoolStats()
+			},
+		})
+
 		return store, nil
 	})
 	if err != nil {

core/stores/redis/redisclustermanager.go

@@ -3,15 +3,20 @@ package redis
 import (
 	"crypto/tls"
 	"io"
+	"runtime"
 	"strings"
 
-	red "github.com/go-redis/redis/v8"
+	red "github.com/redis/go-redis/v9"
 	"github.com/zeromicro/go-zero/core/syncx"
 )
 
 const addrSep = ","
 
-var clusterManager = syncx.NewResourceManager()
+var (
+	clusterManager = syncx.NewResourceManager()
+	// clusterPoolSize is default pool size for cluster type of redis.
+	clusterPoolSize = 5 * runtime.GOMAXPROCS(0)
+)
 
 func getCluster(r *Redis) (*red.ClusterClient, error) {
 	val, err := clusterManager.GetResource(r.Addr, func() (io.Closer, error) {
@@ -33,6 +38,15 @@ func getCluster(r *Redis) (*red.ClusterClient, error) {
 			store.AddHook(hook)
 		}
 
+		connCollector.registerClient(&statGetter{
+			clientType: ClusterType,
+			key:        r.Addr,
+			poolSize:   clusterPoolSize,
+			poolStats: func() *red.PoolStats {
+				return store.PoolStats()
+			},
+		})
+
 		return store, nil
 	})
 	if err != nil {

core/stores/redis/redisclustermanager_test.go

@@ -3,6 +3,8 @@ package redis
 import (
 	"testing"
 
+	"github.com/alicebob/miniredis/v2"
+	red "github.com/redis/go-redis/v9"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -41,3 +43,17 @@ func TestSplitClusterAddrs(t *testing.T) {
 		})
 	}
 }
+
+func TestGetCluster(t *testing.T) {
+	r := miniredis.RunT(t)
+	defer r.Close()
+	c, err := getCluster(&Redis{
+		Addr:  r.Addr(),
+		Type:  ClusterType,
+		tls:   true,
+		hooks: []red.Hook{durationHook},
+	})
+	if assert.NoError(t, err) {
+		assert.NotNil(t, c)
+	}
+}

core/stores/redis/redislock.go

@@ -7,8 +7,7 @@ import (
 	"sync/atomic"
 	"time"
 
-	red "github.com/go-redis/redis/v8"
-
+	red "github.com/redis/go-redis/v9"
 	"github.com/zeromicro/go-zero/core/logx"
 	"github.com/zeromicro/go-zero/core/stringx"
 )

core/stores/sqlc/cachedsql.go

@@ -97,6 +97,9 @@ func (cc CachedConn) Exec(exec ExecFn, keys ...string) (sql.Result, error) {
 }
 
 // ExecCtx runs given exec on given keys, and returns execution result.
+// If DB operation succeeds, it will delete cache with given keys,
+// if DB operation fails, it will return nil result and non-nil error,
+// if DB operation succeeds but cache deletion fails, it will return result and non-nil error.
 func (cc CachedConn) ExecCtx(ctx context.Context, exec ExecCtxFn, keys ...string) (
 	sql.Result, error) {
 	res, err := exec(ctx, cc.db)
@@ -104,11 +107,7 @@ func (cc CachedConn) ExecCtx(ctx context.Context, exec ExecCtxFn, keys ...string
 		return nil, err
 	}
 
-	if err := cc.DelCacheCtx(ctx, keys...); err != nil {
-		return nil, err
-	}
-
-	return res, nil
+	return res, cc.DelCacheCtx(ctx, keys...)
 }
 
 // ExecNoCache runs exec with given sql statement, without affecting cache.

core/stores/sqlc/cachedsql_test.go

@@ -22,11 +22,11 @@ import (
 	"github.com/zeromicro/go-zero/core/logx"
 	"github.com/zeromicro/go-zero/core/stat"
 	"github.com/zeromicro/go-zero/core/stores/cache"
+	"github.com/zeromicro/go-zero/core/stores/dbtest"
 	"github.com/zeromicro/go-zero/core/stores/redis"
 	"github.com/zeromicro/go-zero/core/stores/redis/redistest"
 	"github.com/zeromicro/go-zero/core/stores/sqlx"
 	"github.com/zeromicro/go-zero/core/syncx"
-	"github.com/zeromicro/go-zero/internal/dbtest"
 )
 
 func init() {
@@ -471,31 +471,33 @@ func TestCachedConnExec(t *testing.T) {
 }
 
 func TestCachedConnExecDropCache(t *testing.T) {
-	r, err := miniredis.Run()
-	assert.Nil(t, err)
-	defer fx.DoWithTimeout(func() error {
-		r.Close()
-		return nil
-	}, time.Second)
+	t.Run("drop cache", func(t *testing.T) {
+		r, err := miniredis.Run()
+		assert.Nil(t, err)
+		defer fx.DoWithTimeout(func() error {
+			r.Close()
+			return nil
+		}, time.Second)
 
-	const (
-		key   = "user"
-		value = "any"
-	)
-	var conn trackedConn
-	c := NewNodeConn(&conn, redis.New(r.Addr()), cache.WithExpiry(time.Second*30))
-	assert.Nil(t, c.SetCache(key, value))
-	_, err = c.Exec(func(conn sqlx.SqlConn) (result sql.Result, e error) {
-		return conn.Exec("delete from user_table where id='kevin'")
-	}, key)
-	assert.Nil(t, err)
-	assert.True(t, conn.execValue)
-	_, err = r.Get(key)
-	assert.Exactly(t, miniredis.ErrKeyNotFound, err)
-	_, err = c.Exec(func(conn sqlx.SqlConn) (result sql.Result, e error) {
-		return nil, errors.New("foo")
-	}, key)
-	assert.NotNil(t, err)
+		const (
+			key   = "user"
+			value = "any"
+		)
+		var conn trackedConn
+		c := NewNodeConn(&conn, redis.New(r.Addr()), cache.WithExpiry(time.Second*30))
+		assert.Nil(t, c.SetCache(key, value))
+		_, err = c.Exec(func(conn sqlx.SqlConn) (result sql.Result, e error) {
+			return conn.Exec("delete from user_table where id='kevin'")
+		}, key)
+		assert.Nil(t, err)
+		assert.True(t, conn.execValue)
+		_, err = r.Get(key)
+		assert.Exactly(t, miniredis.ErrKeyNotFound, err)
+		_, err = c.Exec(func(conn sqlx.SqlConn) (result sql.Result, e error) {
+			return nil, errors.New("foo")
+		}, key)
+		assert.NotNil(t, err)
+	})
 }
 
 func TestCachedConn_SetCacheWithExpire(t *testing.T) {

core/stores/sqlx/bulkinserter.go

@@ -4,6 +4,7 @@ import (
 	"database/sql"
 	"fmt"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/zeromicro/go-zero/core/executors"
@@ -30,6 +31,7 @@ type (
 		executor *executors.PeriodicalExecutor
 		inserter *dbInserter
 		stmt     bulkStmt
+		lock     sync.RWMutex // guards stmt
 	}
 
 	bulkStmt struct {
@@ -65,6 +67,9 @@ func (bi *BulkInserter) Flush() {
 
 // Insert inserts given args.
 func (bi *BulkInserter) Insert(args ...any) error {
+	bi.lock.RLock()
+	defer bi.lock.RUnlock()
+
 	value, err := format(bi.stmt.valueFormat, args...)
 	if err != nil {
 		return err
@@ -95,6 +100,11 @@ func (bi *BulkInserter) UpdateStmt(stmt string) error {
 		return err
 	}
 
+	bi.lock.Lock()
+	defer bi.lock.Unlock()
+
+	// with write lock, it doesn't matter what's the order of setting bi.stmt and calling flush.
+	bi.stmt = bkStmt
 	bi.executor.Flush()
 	bi.executor.Sync(func() {
 		bi.inserter.stmt = bkStmt

core/stores/sqlx/bulkinserter_test.go

@@ -5,22 +5,30 @@ import (
 	"database/sql"
 	"errors"
 	"strconv"
+	"strings"
+	"sync"
+	"sync/atomic"
 	"testing"
 
 	"github.com/DATA-DOG/go-sqlmock"
 	"github.com/stretchr/testify/assert"
-	"github.com/zeromicro/go-zero/internal/dbtest"
+	"github.com/zeromicro/go-zero/core/stores/dbtest"
 )
 
 type mockedConn struct {
-	query   string
-	args    []any
-	execErr error
+	query          string
+	args           []any
+	execErr        error
+	updateCallback func(query string, args []any)
 }
 
 func (c *mockedConn) ExecCtx(_ context.Context, query string, args ...any) (sql.Result, error) {
 	c.query = query
 	c.args = args
+	if c.updateCallback != nil {
+		c.updateCallback(query, args)
+	}
+
 	return nil, c.execErr
 }
 
@@ -144,3 +152,50 @@ func TestBulkInserter_Update(t *testing.T) {
 	assert.NotNil(t, inserter.UpdateStmt("foo"))
 	assert.NotNil(t, inserter.Insert("foo", "bar"))
 }
+
+func TestBulkInserter_UpdateStmt(t *testing.T) {
+	var updated int32
+	conn := mockedConn{
+		execErr: errors.New("foo"),
+		updateCallback: func(query string, args []any) {
+			count := atomic.AddInt32(&updated, 1)
+			assert.Empty(t, args)
+			assert.Equal(t, 100, strings.Count(query, "foo"))
+			if count == 1 {
+				assert.Equal(t, 0, strings.Count(query, "bar"))
+			} else {
+				assert.Equal(t, 100, strings.Count(query, "bar"))
+			}
+		},
+	}
+
+	inserter, err := NewBulkInserter(&conn, `INSERT INTO classroom_dau(classroom) VALUES(?)`)
+	assert.NoError(t, err)
+
+	var wg1 sync.WaitGroup
+	wg1.Add(2)
+	for i := 0; i < 2; i++ {
+		go func() {
+			defer wg1.Done()
+			for i := 0; i < 50; i++ {
+				assert.NoError(t, inserter.Insert("foo"))
+			}
+		}()
+	}
+	wg1.Wait()
+
+	assert.NoError(t, inserter.UpdateStmt(`INSERT INTO classroom_dau(classroom, user) VALUES(?, ?)`))
+
+	var wg2 sync.WaitGroup
+	wg2.Add(1)
+	go func() {
+		defer wg2.Done()
+		for i := 0; i < 100; i++ {
+			assert.NoError(t, inserter.Insert("foo", "bar"))
+		}
+		inserter.Flush()
+	}()
+	wg2.Wait()
+
+	assert.Equal(t, int32(2), atomic.LoadInt32(&updated))
+}