fix: simplify go ut (#46606)

issue: #46500

- simplify the run_go_codecov.sh to make sure the set -e to protect any
sub command failure.
- remove all embed etcd in test to make full test can be run at local.

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->

## PR Summary: Simplify Go Unit Tests by Removing Embedded etcd and
Async Startup Scaffolding

**Core Invariant:**
This PR assumes that unit tests can be simplified by running without
embedded etcd servers (delegating to environment-based or external etcd
instances via `kvfactory.GetEtcdAndPath()` or `ETCD_ENDPOINTS`) and by
removing goroutine-based async startup scaffolding in favor of
synchronous component initialization. Tests remain functionally
equivalent while becoming simpler to run and debug locally.

**What is Removed or Simplified:**

1. **Embedded etcd test infrastructure deleted**: Removes
`EmbedEtcdUtil` type and its public methods (SetupEtcd,
TearDownEmbedEtcd) from `pkg/util/testutils/embed_etcd.go`, removes the
`StartTestEmbedEtcdServer()` helper from `pkg/util/etcd/etcd_util.go`,
and removes etcd embedding from test suites (e.g., `TaskSuite`,
`EtcdSourceSuite`, `mixcoord/client_test.go`). Tests now either skip
etcd-dependent tests (via `MILVUS_UT_WITHOUT_KAFKA=1` environment flag
in `kafka_test.go`) or source etcd from external configuration (via
`kvfactory.GetEtcdAndPath()` in `task_test.go`, or `ETCD_ENDPOINTS`
environment variable in `etcd_source_test.go`). This eliminates the
overhead of spinning up temporary etcd servers for unit tests.

2. **Async startup scaffolding replaced with synchronous
initialization**: In `internal/proxy/proxy_test.go` and
`proxy_rpc_test.go`, the `startGrpc()` method signature removes the
`sync.WaitGroup` parameter; components are now created, prepared, and
run synchronously in-place rather than in goroutines (e.g., `go
testServer.startGrpc(ctx, &p)` becomes `testServer.startGrpc(ctx, &p)`
running synchronously). Readiness checks (e.g., `waitForGrpcReady()`)
remain in place to ensure startup safety without concurrency constructs.
This simplifies control flow and reduces debugging complexity.

3. **Shell script orchestration unified with proper error handling**: In
`scripts/run_go_codecov.sh` and `scripts/run_intergration_test.sh`,
per-package inline test invocations are consolidated into a single
`test_cmd()` function with unified `TEST_CMD_WITH_ARGS` array containing
race, coverage, verbose, and other flags. The problematic `set -ex` is
replaced with `set -e` alone (removing debug output noise while
preserving strict error semantics), ensuring the scripts fail fast on
any command failure.

**Why No Regression:**
- Test assertions and code paths remain unchanged; only deployment
source of etcd (embedded → external) and startup orchestration (async →
sync) change.
- Readiness verification (e.g., `waitForGrpcReady()`) is retained,
ensuring components are initialized before test execution.
- Test flags (race detection, coverage, verbosity) are uniformly applied
across all packages via unified `TEST_CMD_WITH_ARGS`, preserving test
coverage and quality.
- `set -e` alone is sufficient for strict failure detection without the
`-x` flag's verbose output.

<!-- end of auto-generated comment: release notes by coderabbit.ai -->

---------

Signed-off-by: chyezh <chyezh@outlook.com>

internal/distributed/mixcoord/client/client_test.go

@@ -20,25 +20,21 @@ import (
 	"context"
 	"math/rand"
 	"os"
-	"strings"
 	"testing"
 	"time"
 
 	"github.com/cockroachdb/errors"
 	"github.com/stretchr/testify/assert"
 	mock1 "github.com/stretchr/testify/mock"
-	"go.uber.org/zap"
 	"google.golang.org/grpc"
 
 	"github.com/milvus-io/milvus-proto/go-api/v2/commonpb"
 	"github.com/milvus-io/milvus-proto/go-api/v2/milvuspb"
 	"github.com/milvus-io/milvus/internal/mocks"
 	"github.com/milvus-io/milvus/internal/util/mock"
-	"github.com/milvus-io/milvus/pkg/v2/log"
 	"github.com/milvus-io/milvus/pkg/v2/proto/datapb"
 	"github.com/milvus-io/milvus/pkg/v2/proto/indexpb"
 	"github.com/milvus-io/milvus/pkg/v2/proto/internalpb"
-	"github.com/milvus-io/milvus/pkg/v2/util/etcd"
 	"github.com/milvus-io/milvus/pkg/v2/util/merr"
 	"github.com/milvus-io/milvus/pkg/v2/util/paramtable"
 )
@@ -46,19 +42,6 @@ import (
 var mockErr = errors.New("mock grpc err")
 
 func TestMain(m *testing.M) {
-	// init embed etcd
-	embedetcdServer, tempDir, err := etcd.StartTestEmbedEtcdServer()
-	if err != nil {
-		log.Fatal("failed to start embed etcd server", zap.Error(err))
-	}
-	defer os.RemoveAll(tempDir)
-	defer embedetcdServer.Close()
-
-	addrs := etcd.GetEmbedEtcdEndpoints(embedetcdServer)
-
-	paramtable.Init()
-	paramtable.Get().Save(Params.EtcdCfg.Endpoints.Key, strings.Join(addrs, ","))
-
 	rand.Seed(time.Now().UnixNano())
 	os.Exit(m.Run())
 }

internal/proxy/proxy_rpc_test.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"os"
 	"strings"
-	"sync"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -23,7 +22,6 @@ import (
 
 func TestProxyRpcLimit(t *testing.T) {
 	var err error
-	var wg sync.WaitGroup
 
 	path := "/tmp/milvus/rocksmq" + funcutil.GenRandomStr()
 	t.Setenv("ROCKSMQ_PATH", path)
@@ -50,8 +48,7 @@ func TestProxyRpcLimit(t *testing.T) {
 
 	testServer := newProxyTestServer(proxy)
 	testServer.Proxy.SetAddress(p.GetAddress())
-	wg.Add(1)
-	go testServer.startGrpc(ctx, &wg, &p)
+	go testServer.startGrpc(ctx, &p)
 	assert.NoError(t, testServer.waitForGrpcReady())
 	defer testServer.grpcServer.Stop()
 	client, err := grpcproxyclient.NewClient(ctx, "localhost:"+p.Port.GetValue(), 1)

internal/querycoordv2/task/task_test.go

@@ -39,17 +39,16 @@ import (
 	. "github.com/milvus-io/milvus/internal/querycoordv2/params"
 	"github.com/milvus-io/milvus/internal/querycoordv2/session"
 	"github.com/milvus-io/milvus/internal/querycoordv2/utils"
+	kvfactory "github.com/milvus-io/milvus/internal/util/dependency/kv"
 	"github.com/milvus-io/milvus/pkg/v2/common"
 	"github.com/milvus-io/milvus/pkg/v2/kv"
 	"github.com/milvus-io/milvus/pkg/v2/proto/datapb"
 	"github.com/milvus-io/milvus/pkg/v2/proto/indexpb"
 	"github.com/milvus-io/milvus/pkg/v2/proto/querypb"
 	"github.com/milvus-io/milvus/pkg/v2/proto/rootcoordpb"
-	"github.com/milvus-io/milvus/pkg/v2/util/etcd"
 	"github.com/milvus-io/milvus/pkg/v2/util/merr"
 	"github.com/milvus-io/milvus/pkg/v2/util/metricsinfo"
 	"github.com/milvus-io/milvus/pkg/v2/util/paramtable"
-	"github.com/milvus-io/milvus/pkg/v2/util/testutils"
 	"github.com/milvus-io/milvus/pkg/v2/util/typeutil"
 )
 
@@ -61,7 +60,6 @@ type distribution struct {
 
 type TaskSuite struct {
 	suite.Suite
-	testutils.EmbedEtcdUtil
 
 	// Data
 	collection      int64
@@ -147,15 +145,7 @@ func (suite *TaskSuite) SetupTest() {
 
 	config := GenerateEtcdConfig()
 	suite.ctx = context.Background()
-	cli, err := etcd.GetEtcdClient(
-		config.UseEmbedEtcd.GetAsBool(),
-		config.EtcdUseSSL.GetAsBool(),
-		config.Endpoints.GetAsStrings(),
-		config.EtcdTLSCert.GetValue(),
-		config.EtcdTLSKey.GetValue(),
-		config.EtcdTLSCACert.GetValue(),
-		config.EtcdTLSMinVersion.GetValue())
-	suite.Require().NoError(err)
+	cli, _ := kvfactory.GetEtcdAndPath()
 
 	suite.kv = etcdkv.NewEtcdKV(cli, config.MetaRootPath.GetValue())
 	suite.store = querycoord.NewCatalog(suite.kv)

pkg/config/etcd_source_test.go

@@ -19,11 +19,11 @@ package config
 import (
 	"context"
 	"os"
+	"strings"
 	"testing"
 	"time"
 
 	"github.com/stretchr/testify/suite"
-	"go.etcd.io/etcd/server/v3/embed"
 	"go.uber.org/atomic"
 
 	"github.com/milvus-io/milvus/pkg/v2/util/etcd"
@@ -32,29 +32,18 @@ import (
 type EtcdSourceSuite struct {
 	suite.Suite
 
-	embedEtcdServer *embed.Etcd
-	tempDir         string
-	endpoints       []string
+	endpoints []string
 }
 
 func (s *EtcdSourceSuite) SetupSuite() {
-	// init embed etcd
-	embedServer, tempDir, err := etcd.StartTestEmbedEtcdServer()
-
-	s.Require().NoError(err)
-
-	s.embedEtcdServer = embedServer
-	s.tempDir = tempDir
-	s.endpoints = etcd.GetEmbedEtcdEndpoints(embedServer)
+	endpoints := os.Getenv("ETCD_ENDPOINTS")
+	if endpoints == "" {
+		endpoints = "localhost:2379"
+	}
+	s.endpoints = strings.Split(endpoints, ",")
 }
 
 func (s *EtcdSourceSuite) TearDownSuite() {
-	if s.embedEtcdServer != nil {
-		s.embedEtcdServer.Close()
-	}
-	if s.tempDir != "" {
-		os.RemoveAll(s.tempDir)
-	}
 }
 
 func (s *EtcdSourceSuite) TestNewSource() {

pkg/streaming/walimpls/impls/kafka/kafka_test.go

@@ -1,6 +1,7 @@
 package kafka
 
 import (
+	"os"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -28,6 +29,9 @@ func TestRegistry(t *testing.T) {
 }
 
 func TestKafka(t *testing.T) {
+	if os.Getenv("MILVUS_UT_WITHOUT_KAFKA") != "" {
+		t.Skip("there's no kafka broker available, skipping kafka test")
+	}
 	walimpls.NewWALImplsTestFramework(t, 100, &builderImpl{}).Run()
 }
 

pkg/util/etcd/etcd_util.go

@@ -253,20 +253,6 @@ func buildKvGroup(keys, values []string) (map[string]string, error) {
 	return ret, nil
 }
 
-// StartTestEmbedEtcdServer returns a newly created embed etcd server.
-// ### USED FOR UNIT TEST ONLY ###
-func StartTestEmbedEtcdServer() (*embed.Etcd, string, error) {
-	dir, err := os.MkdirTemp(os.TempDir(), "milvus_ut")
-	if err != nil {
-		return nil, "", err
-	}
-	config := embed.NewConfig()
-	config.Dir = dir
-	config.LogLevel = "warn"
-	server, err := embed.StartEtcd(config)
-	return server, dir, err
-}
-
 // GetEmbedEtcdEndpoints returns etcd listener address for endpoint config.
 func GetEmbedEtcdEndpoints(server *embed.Etcd) []string {
 	addrs := make([]string, 0, len(server.Clients))

pkg/util/testutils/embed_etcd.go

@@ -1,50 +0,0 @@
-// Licensed to the LF AI & Data foundation under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package testutils
-
-import (
-	"os"
-
-	"go.etcd.io/etcd/server/v3/embed"
-
-	"github.com/milvus-io/milvus/pkg/v2/util/etcd"
-)
-
-type EmbedEtcdUtil struct {
-	server  *embed.Etcd
-	tempDir string
-}
-
-func (util *EmbedEtcdUtil) SetupEtcd() ([]string, error) {
-	// init embed etcd
-	embedetcdServer, tempDir, err := etcd.StartTestEmbedEtcdServer()
-	if err != nil {
-		return nil, err
-	}
-	util.server, util.tempDir = embedetcdServer, tempDir
-
-	return etcd.GetEmbedEtcdEndpoints(embedetcdServer), nil
-}
-
-func (util *EmbedEtcdUtil) TearDownEmbedEtcd() {
-	if util.server != nil {
-		util.server.Close()
-	}
-	if util.tempDir != "" {
-		os.RemoveAll(util.tempDir)
-	}
-}

scripts/run_go_codecov.sh

@@ -16,61 +16,64 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-FILE_COVERAGE_INFO="go_coverage.txt"
-FILE_COVERAGE_HTML="go_coverage.html"
+FILE_COVERAGE_INFO="$PWD/go_coverage.txt"
+FILE_COVERAGE_HTML="$PWD/go_coverage.html"
+
 
 BASEDIR=$(dirname "$0")
 source $BASEDIR/setenv.sh
 
-set -ex
-echo "mode: atomic" > ${FILE_COVERAGE_INFO}
+set -e
 
-# run unittest
-echo "Running unittest under ./internal & ./pkg"
+echo "mode: atomic" > ${FILE_COVERAGE_INFO}
 
 TEST_CMD=$@
 if [ -z "$TEST_CMD" ]; then
    TEST_CMD="go test" 
 fi
 
-# starting the timer
-beginTime=`date +%s`
-pushd cmd/tools
-$TEST_CMD -gcflags="all=-N -l" -race -tags dynamic,test -v -buildvcs=false -coverpkg=./... -coverprofile=profile.out -covermode=atomic ./...
-if [ -f profile.out ]; then
-    grep -v kafka profile.out | grep -v planparserv2/generated | grep -v mocks | sed '1d' >> ../${FILE_COVERAGE_INFO}
-    rm profile.out
-fi
-popd
-for d in $(go list -buildvcs=false ./internal/... | grep -v -e vendor -e kafka -e planparserv2/generated -e mocks); do
-    $TEST_CMD -gcflags="all=-N -l" -race -tags dynamic,test -v -buildvcs=false -coverpkg=./... -coverprofile=profile.out -covermode=atomic "$d"
-    if [ -f profile.out ]; then
-        grep -v kafka profile.out | grep -v planparserv2/generated | grep -v mocks | sed '1d' >> ${FILE_COVERAGE_INFO}
-        rm profile.out
-    fi
-done
-pushd pkg
-for d in $(go list -buildvcs=false ./... | grep -v -e vendor -e kafka -e planparserv2/generated -e mocks); do
-    $TEST_CMD -gcflags="all=-N -l" -race -tags dynamic,test -v -buildvcs=false -coverpkg=./... -coverprofile=profile.out -covermode=atomic "$d"
-    if [ -f profile.out ]; then
-        grep -v kafka profile.out | grep -v planparserv2/generated | grep -v mocks | sed '1d' >> ../${FILE_COVERAGE_INFO}
-        rm profile.out
-    fi
-done
-popd
-# milvusclient
-pushd client
-for d in $(go list -buildvcs=false ./... | grep -v -e vendor -e kafka -e planparserv2/generated -e mocks); do
-    $TEST_CMD -gcflags="all=-N -l" -race -tags dynamic -v -buildvcs=false -coverpkg=./... -coverprofile=profile.out -covermode=atomic "$d"
-    if [ -f profile.out ]; then
-        grep -v kafka profile.out | grep -v planparserv2/generated | grep -v mocks | sed '1d' >> ../${FILE_COVERAGE_INFO}
-        rm profile.out
-    fi
-done
-popd
-endTime=`date +%s`
+TEST_CMD_WITH_ARGS=(
+    $TEST_CMD
+    "-gcflags=all=-N -l"
+    -race
+    -tags dynamic,test
+    -v
+    -failfast
+    -buildvcs=false
+    -coverpkg=./...
+    -coverprofile=profile.out
+    -covermode=atomic
+)
 
-echo "Total time for go unittest:" $(($endTime-$beginTime)) "s"
+function test_cmd() {
+    mapfile -t PKGS < <(go list -tags dynamic,test ./...)
+    for pkg in "${PKGS[@]}"; do
+        echo -e "-----------------------------------\nRunning test cases at $pkg ..." 
+        "${TEST_CMD_WITH_ARGS[@]}" "$pkg"
+        if [ -f profile.out ]; then
+            # Skip the per-profile header to keep a single global "mode:" line
+            # Skip the packages that are not covered by the test
+            sed '1{/^mode:/d}' profile.out | grep -vE '(planparserv2/generated|mocks)' >> "${FILE_COVERAGE_INFO}" || [ $? -eq 1 ] 
+            rm profile.out
+        fi
+        echo -e "-----------------------------------\n"
+    done
+}
+
+export MILVUS_UT_WITHOUT_KAFKA=1 # kafka is not available in the CI environment, so skip the kafka tests
+
+# starting the timer
+beginTime=$(date +%s)
+echo -e "=== Running go unittest ===\n\n"
+
+for d in cmd/tools internal pkg client; do
+    pushd "$d" 
+    test_cmd 
+    popd
+done
+
+endTime=$(date +%s)
+echo -e "=== Total time for go unittest: $(($endTime-$beginTime)) s ==="
 
 # generate html report
 # go tool cover -html=./${FILE_COVERAGE_INFO} -o ./${FILE_COVERAGE_HTML}

scripts/run_intergration_test.sh

@@ -23,36 +23,55 @@ FILE_COVERAGE_INFO="it_coverage.txt"
 BASEDIR=$(dirname "$0")
 source $BASEDIR/setenv.sh
 
-TEST_CMD=$@
-if [ -z "$TEST_CMD" ]; then
-   TEST_CMD="go test -failfast -count=1" 
-fi
-
 set -e
+
 echo "mode: atomic" > ${FILE_COVERAGE_INFO}
 echo "MILVUS_WORK_DIR: $MILVUS_WORK_DIR"
 
-# starting the timer
-beginTime=`date +%s`
+TEST_CMD=$@
+if [ -z "$TEST_CMD" ]; then
+   TEST_CMD="go test" 
+fi
 
-for d in $(go list ./tests/integration/...); do
-    echo "Start to run integration test under \"$d\" pkg"
-    if [[ $d == *"coordrecovery"* ]]; then
-        echo "running coordrecovery"
-        # simplified command to speed up coord init test since it is large.
-        $TEST_CMD -tags dynamic,test -v -coverprofile=profile.out -covermode=atomic "$d" -caseTimeout=20m -timeout=30m
-    elif [[ $d == *"import"* ]]; then
-        $TEST_CMD -tags dynamic,test -v -coverprofile=profile.out -covermode=atomic "$d" -caseTimeout=20m -timeout=60m
-    else
-        $TEST_CMD -race -tags dynamic,test -v -coverpkg=./... -coverprofile=profile.out -covermode=atomic "$d" -caseTimeout=15m -timeout=30m
-    fi
-    if [ -f profile.out ]; then
-        grep -v kafka profile.out | grep -v planparserv2/generated | grep -v mocks | sed '1d' >> ${FILE_COVERAGE_INFO}
-        rm profile.out
-    fi
+TEST_CMD_WITH_ARGS=(
+    $TEST_CMD
+    "-gcflags=all=-N -l"
+    -race
+    -tags dynamic,test
+    -v
+    -failfast
+    -count=1
+    -buildvcs=false
+    -coverpkg=./...
+    -coverprofile=profile.out
+    -covermode=atomic
+    -caseTimeout=20m 
+    -timeout=60m
+)
+
+function test_cmd() {
+    mapfile -t PKGS < <(go list -tags dynamic,test ./...)
+    for pkg in "${PKGS[@]}"; do
+        echo -e "-----------------------------------\nRunning test cases at $pkg ..." 
+        "${TEST_CMD_WITH_ARGS[@]}" "$pkg"
+        if [ -f profile.out ]; then
+            # Skip the per-profile header to keep a single global "mode:" line
+            # Skip the packages that are not covered by the test
+            sed '1{/^mode:/d}' profile.out | grep -vE '(planparserv2/generated|mocks)' >> "${FILE_COVERAGE_INFO}" || [ $? -eq 1 ] 
+            rm profile.out
+        fi
+        echo -e "-----------------------------------\n"
+    done
+}
+
+beginTime=`date +%s`
+printf "=== Running integration test ===\n\n"
+
+for d in tests/integration; do
+    pushd "$d"
+    test_cmd 
+    popd
 done
 
 endTime=`date +%s`
-
-echo "Total time for go integration test:" $(($endTime-$beginTime)) "s"
-
+printf "=== Total time for go integration test: $(($endTime-$beginTime)) s==="