RBAC built in privilege groups

Signed-off-by: shaoting-huang <[email protected]>

configs/milvus.yaml

@@ -19,7 +19,7 @@ etcd:
   # Endpoints used to access etcd service. You can change this parameter as the endpoints of your own etcd cluster.
   # Environment variable: ETCD_ENDPOINTS
   # etcd preferentially acquires valid address from environment variable ETCD_ENDPOINTS when Milvus is started.
-  endpoints: localhost:2379
+  endpoints: etcd:2379
   # Root prefix of the key to where Milvus stores data in etcd.
   # It is recommended to change this parameter before starting Milvus for the first time.
   # To share an etcd instance among multiple Milvus instances, consider changing this to a different value for each Milvus instance before you start them.
@@ -98,7 +98,7 @@ minio:
   # minio.address and minio.port together generate the valid access to MinIO or S3 service.
   # MinIO preferentially acquires the valid IP address from the environment variable MINIO_ADDRESS when Milvus is started.
   # Default value applies when MinIO or S3 is running on the same network with Milvus.
-  address: localhost
+  address: minio:9000
   port: 9000 # Port of MinIO or S3 service.
   # Access key ID that MinIO or S3 issues to user for authorized access.
   # Environment variable: MINIO_ACCESS_KEY_ID or minio.accessKeyID
@@ -184,7 +184,7 @@ pulsar:
   # pulsar.address and pulsar.port together generate the valid access to Pulsar.
   # Pulsar preferentially acquires the valid IP address from the environment variable PULSAR_ADDRESS when Milvus is started.
   # Default value applies when Pulsar is running on the same network with Milvus.
-  address: localhost
+  address: pulsar://pulsar:6650
   port: 6650 # Port of Pulsar service.
   webport: 80 # Web port of of Pulsar service. If you connect direcly without proxy, should use 8080.
   # The maximum size of each message in Pulsar. Unit: Byte.
@@ -811,6 +811,30 @@ common:
     # like the old password verification when updating the credential
     superUsers: 
     defaultRootPassword: Milvus # default password for root user
+    rbac:
+      overrideBuiltInPrivilgeGroups:
+        enabled: false # Whether to override build-in privilege groups
+      cluster:
+        readonly:
+          privileges: SelectOwnership,SelectUser,DescribeResourceGroup,ListResourceGroups # Cluster level readonly privileges
+        readwrite:
+          privileges: SelectOwnership,SelectUser,DescribeResourceGroup,ListResourceGroups,CreateOwnership,UpdateUser,DropOwnership,ManageOwnership,BackupRBAC,RestoreRBAC,CreateResourceGroup,UpdateResourceGroups,DropResourceGroup,TransferNode,TransferReplica # Cluster level readwrite privileges
+        admin:
+          privileges: SelectOwnership,SelectUser,DescribeResourceGroup,ListResourceGroups,CreateOwnership,UpdateUser,DropOwnership,ManageOwnership,BackupRBAC,RestoreRBAC,CreateResourceGroup,UpdateResourceGroups,DropResourceGroup,TransferNode,TransferReplica # Cluster level admin privileges
+      database:
+        readonly:
+          privileges: ListDatabases,DescribeDatabase # Database level readonly privileges
+        readwrite:
+          privileges: ListDatabases,DescribeDatabase,CreateDatabase,DropDatabase,AlterDatabase # Database level readwrite privileges
+        admin:
+          privileges: ListDatabases,DescribeDatabase,CreateDatabase,DropDatabase,AlterDatabase # Database level admin privileges
+      collection:
+        readonly:
+          privileges: Query,Search,IndexDetail,GetFlushState,GetLoadState,GetLoadingProgress,HasPartition,ShowPartitions,ShowCollections,ListAliases,DescribeCollection,DescribeAlias,GetStatistics # Collection level readonly privileges
+        readwrite:
+          privileges: Query,Search,IndexDetail,GetFlushState,GetLoadState,GetLoadingProgress,HasPartition,ShowPartitions,ShowCollections,ListAliases,DescribeCollection,DescribeAlias,GetStatistics,CreateIndex,DropIndex,CreatePartition,DropPartition,Load,Release,Insert,Delete,Upsert,Import,Flush,Compaction,LoadBalance,RenameCollection,CreateAlias,DropAlias,CreateCollection,DropCollection,FlushAll # Collection level readwrite privileges
+        admin:
+          privileges: Query,Search,IndexDetail,GetFlushState,GetLoadState,GetLoadingProgress,HasPartition,ShowPartitions,ShowCollections,ListAliases,DescribeCollection,DescribeAlias,GetStatistics,CreateIndex,DropIndex,CreatePartition,DropPartition,Load,Release,Insert,Delete,Upsert,Import,Flush,Compaction,LoadBalance,RenameCollection,CreateAlias,DropAlias,CreateCollection,DropCollection,FlushAll # Collection level admin privileges
     tlsMode: 0
   session:
     ttl: 30 # ttl value when session granting a lease to register service

go.mod

@@ -23,7 +23,7 @@ require (
 	github.com/grpc-ecosystem/go-grpc-middleware v1.3.0
 	github.com/klauspost/compress v1.17.9
 	github.com/mgutz/ansi v0.0.0-20200706080929-d51e80ef957d
-	github.com/milvus-io/milvus-proto/go-api/v2 v2.3.4-0.20241111062829-6de3d96f664f
+	github.com/milvus-io/milvus-proto/go-api/v2 v2.3.4-0.20241114133823-d3506c6f465c
 	github.com/minio/minio-go/v7 v7.0.73
 	github.com/pingcap/log v1.1.1-0.20221015072633-39906604fb81
 	github.com/prometheus/client_golang v1.14.0

go.sum

@@ -628,8 +628,8 @@ github.com/milvus-io/cgosymbolizer v0.0.0-20240722103217-b7dee0e50119 h1:9VXijWu
 github.com/milvus-io/cgosymbolizer v0.0.0-20240722103217-b7dee0e50119/go.mod h1:DvXTE/K/RtHehxU8/GtDs4vFtfw64jJ3PaCnFri8CRg=
 github.com/milvus-io/gorocksdb v0.0.0-20220624081344-8c5f4212846b h1:TfeY0NxYxZzUfIfYe5qYDBzt4ZYRqzUjTR6CvUzjat8=
 github.com/milvus-io/gorocksdb v0.0.0-20220624081344-8c5f4212846b/go.mod h1:iwW+9cWfIzzDseEBCCeDSN5SD16Tidvy8cwQ7ZY8Qj4=
-github.com/milvus-io/milvus-proto/go-api/v2 v2.3.4-0.20241111062829-6de3d96f664f h1:yLxT8NH0ixUOJMqJuk0xvGf0cKsr+N2xibyTat256PI=
-github.com/milvus-io/milvus-proto/go-api/v2 v2.3.4-0.20241111062829-6de3d96f664f/go.mod h1:/6UT4zZl6awVeXLeE7UGDWZvXj3IWkRsh3mqsn0DiAs=
+github.com/milvus-io/milvus-proto/go-api/v2 v2.3.4-0.20241114133823-d3506c6f465c h1:Ay5w6sTE1QxCydCqqW5N44EcJrMqaqbL5zcp2vclkOw=
+github.com/milvus-io/milvus-proto/go-api/v2 v2.3.4-0.20241114133823-d3506c6f465c/go.mod h1:/6UT4zZl6awVeXLeE7UGDWZvXj3IWkRsh3mqsn0DiAs=
 github.com/milvus-io/pulsar-client-go v0.12.1 h1:O2JZp1tsYiO7C0MQ4hrUY/aJXnn2Gry6hpm7UodghmE=
 github.com/milvus-io/pulsar-client-go v0.12.1/go.mod h1:dkutuH4oS2pXiGm+Ti7fQZ4MRjrMPZ8IJeEGAWMeckk=
 github.com/minio/asm2plan9s v0.0.0-20200509001527-cdd76441f9d8 h1:AMFGa4R4MiIpspGNG7Z948v4n35fFGB3RR3G/ry4FWs=

internal/rootcoord/root_coord.go

@@ -576,7 +576,12 @@ func (c *Core) initRbac() error {
 	}
 
 	if Params.RoleCfg.Enabled.GetAsBool() {
-		return c.initBuiltinRoles()
+		if err := c.initBuiltinRoles(); err != nil {
+			return err
+		}
+	}
+	if err := c.initBuiltinPrivilegeGroups(); err != nil {
+		return err
 	}
 	return nil
 }
@@ -624,6 +629,51 @@ func (c *Core) initPublicRolePrivilege() error {
 	return nil
 }
 
+func (c *Core) initBuiltinPrivilegeGroups() error {
+	// init built in privilege groups, override by config if rbac config enabled
+	for groupName, privileges := range util.BuiltinPrivilegeGroups {
+		if err := c.meta.CreatePrivilegeGroup(groupName); err != nil {
+			return err
+		}
+		if Params.RbacConfig.Enabled.GetAsBool() {
+			var confPrivs []string
+			switch groupName {
+			case "ClusterReadOnly":
+				confPrivs = Params.RbacConfig.ClusterReadOnlyPrivileges.GetAsStrings()
+			case "ClusterReadWrite":
+				confPrivs = Params.RbacConfig.ClusterReadWritePrivileges.GetAsStrings()
+			case "ClusterAdmin":
+				confPrivs = Params.RbacConfig.ClusterAdminPrivileges.GetAsStrings()
+			case "DatabaseReadOnly":
+				confPrivs = Params.RbacConfig.DBReadOnlyPrivileges.GetAsStrings()
+			case "DatabaseReadWrite":
+				confPrivs = Params.RbacConfig.DBReadWritePrivileges.GetAsStrings()
+			case "DatabaseAdmin":
+				confPrivs = Params.RbacConfig.DBAdminPrivileges.GetAsStrings()
+			case "CollectionReadOnly":
+				confPrivs = Params.RbacConfig.CollectionReadOnlyPrivileges.GetAsStrings()
+			case "CollectionReadWrite":
+				confPrivs = Params.RbacConfig.CollectionReadWritePrivileges.GetAsStrings()
+			case "CollectionAdmin":
+				confPrivs = Params.RbacConfig.CollectionAdminPrivileges.GetAsStrings()
+			default:
+				return nil
+			}
+			if len(confPrivs) > 0 {
+				privileges = confPrivs
+			}
+		}
+
+		privs := lo.Map(privileges, func(name string, _ int) *milvuspb.PrivilegeEntity {
+			return &milvuspb.PrivilegeEntity{Name: name}
+		})
+		if err := c.meta.OperatePrivilegeGroup(groupName, privs, milvuspb.OperatePrivilegeGroupType_AddPrivilegesToGroup); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 func (c *Core) initBuiltinRoles() error {
 	rolePrivilegesMap := Params.RoleCfg.Roles.GetAsRoleDetails()
 	for role, privilegesJSON := range rolePrivilegesMap {
@@ -2583,24 +2633,24 @@ func (c *Core) OperatePrivilege(ctx context.Context, in *milvuspb.OperatePrivile
 		return merr.StatusWithErrorCode(err, commonpb.ErrorCode_OperatePrivilegeFailure), nil
 	}
 
-	// set up privilege name for metastore
-	privName := in.Entity.Grantor.Privilege.Name
-	ctxLog.Debug("before PrivilegeNameForMetastore", zap.String("privilege", privName))
-	if !util.IsAnyWord(privName) {
-		dbPrivName, err := c.getMetastorePrivilegeName(privName)
-		if err != nil {
-			return merr.StatusWithErrorCode(err, commonpb.ErrorCode_OperatePrivilegeFailure), nil
-		}
-		in.Entity.Grantor.Privilege.Name = dbPrivName
-	}
-	ctxLog.Debug("after PrivilegeNameForMetastore", zap.String("privilege", privName))
-
+	// set up object name if it is global object type
 	if in.Entity.Object.Name == commonpb.ObjectType_Global.String() {
 		in.Entity.ObjectName = util.AnyWord
 	}
 
+	privName := in.Entity.Grantor.Privilege.Name
+
 	redoTask := newBaseRedoTask(c.stepExecutor)
 	redoTask.AddSyncStep(NewSimpleStep("operate privilege meta data", func(ctx context.Context) ([]nestedStep, error) {
+		if !util.IsAnyWord(privName) {
+			// set up privilege name for metastore
+			dbPrivName, err := c.getMetastorePrivilegeName(privName)
+			if err != nil {
+				return nil, err
+			}
+			in.Entity.Grantor.Privilege.Name = dbPrivName
+		}
+
 		err := c.meta.OperatePrivilege(util.DefaultTenant, in.Entity, in.Type)
 		if err != nil && !common.IsIgnorableError(err) {
 			log.Warn("fail to operate the privilege", zap.Any("in", in), zap.Error(err))
@@ -2609,6 +2659,8 @@ func (c *Core) OperatePrivilege(ctx context.Context, in *milvuspb.OperatePrivile
 		return nil, nil
 	}))
 	redoTask.AddAsyncStep(NewSimpleStep("operate privilege cache", func(ctx context.Context) ([]nestedStep, error) {
+		// set back to expand privilege group
+		in.Entity.Grantor.Privilege.Name = privName
 		var opType int32
 		switch in.Type {
 		case milvuspb.OperatePrivilegeType_Grant:
@@ -2619,9 +2671,22 @@ func (c *Core) OperatePrivilege(ctx context.Context, in *milvuspb.OperatePrivile
 			log.Warn("invalid operate type for the OperatePrivilege api", zap.Any("in", in))
 			return nil, nil
 		}
+		grants := []*milvuspb.GrantEntity{in.Entity}
+
+		allGroups, err := c.meta.ListPrivilegeGroups()
+		if err != nil {
+			return nil, err
+		}
+		groups := lo.SliceToMap(allGroups, func(group *milvuspb.PrivilegeGroupInfo) (string, []*milvuspb.PrivilegeEntity) {
+			return group.GroupName, group.Privileges
+		})
+		expandGrants, err := c.expandPrivilegeGroups(grants, groups)
+		if err != nil {
+			return nil, err
+		}
 		if err := c.proxyClientManager.RefreshPolicyInfoCache(ctx, &proxypb.RefreshPolicyInfoCacheRequest{
 			OpType: opType,
-			OpKey:  funcutil.PolicyForPrivilege(in.Entity.Role.Name, in.Entity.Object.Name, in.Entity.ObjectName, in.Entity.Grantor.Privilege.Name, in.Entity.DbName),
+			OpKey:  funcutil.PolicyForPrivileges(expandGrants),
 		}); err != nil {
 			log.Warn("fail to refresh policy info cache", zap.Any("in", in), zap.Error(err))
 			return nil, err
@@ -3095,11 +3160,11 @@ func (c *Core) OperatePrivilegeGroup(ctx context.Context, in *milvuspb.OperatePr
 		rolesToRevoke := []*milvuspb.GrantEntity{}
 		rolesToGrant := []*milvuspb.GrantEntity{}
 		compareGrants := func(a, b *milvuspb.GrantEntity) bool {
-			return a.Role.GetName() == b.Role.GetName() &&
-				a.Object.GetName() == b.Object.GetName() &&
+			return a.Role.Name == b.Role.Name &&
+				a.Object.Name == b.Object.Name &&
 				a.ObjectName == b.ObjectName &&
-				a.Grantor.GetUser().GetName() == b.Grantor.GetUser().GetName() &&
-				a.Grantor.GetPrivilege().GetName() == b.Grantor.GetPrivilege().GetName() &&
+				a.Grantor.User.Name == b.Grantor.User.Name &&
+				a.Grantor.Privilege.Name == b.Grantor.Privilege.Name &&
 				a.DbName == b.DbName
 		}
 		for _, role := range roles {
@@ -3110,8 +3175,14 @@ func (c *Core) OperatePrivilegeGroup(ctx context.Context, in *milvuspb.OperatePr
 			if err != nil {
 				return nil, err
 			}
-			currGrants := c.expandPrivilegeGroups(grants, currGroups)
-			newGrants := c.expandPrivilegeGroups(grants, newGroups)
+			currGrants, err := c.expandPrivilegeGroups(grants, currGroups)
+			if err != nil {
+				return nil, err
+			}
+			newGrants, err := c.expandPrivilegeGroups(grants, newGroups)
+			if err != nil {
+				return nil, err
+			}
 
 			toRevoke := lo.Filter(currGrants, func(item *milvuspb.GrantEntity, _ int) bool {
 				return !lo.ContainsBy(newGrants, func(newItem *milvuspb.GrantEntity) bool {
@@ -3175,20 +3246,42 @@ func (c *Core) OperatePrivilegeGroup(ctx context.Context, in *milvuspb.OperatePr
 	return merr.Success(), nil
 }
 
-func (c *Core) expandPrivilegeGroups(grants []*milvuspb.GrantEntity, groups map[string][]*milvuspb.PrivilegeEntity) []*milvuspb.GrantEntity {
+func (c *Core) expandPrivilegeGroups(grants []*milvuspb.GrantEntity, groups map[string][]*milvuspb.PrivilegeEntity) ([]*milvuspb.GrantEntity, error) {
 	newGrants := []*milvuspb.GrantEntity{}
 	for _, grant := range grants {
-		if groups[grant.Grantor.Privilege.Name] == nil {
-			newGrants = append(newGrants, grant)
+		privName := grant.Grantor.Privilege.Name
+		if privGroup, exists := groups[privName]; !exists {
+			metaName, err := c.getMetastorePrivilegeName(privName)
+			if err != nil {
+				return nil, err
+			}
+			newGrants = append(newGrants, &milvuspb.GrantEntity{
+				Role:       grant.Role,
+				Object:     grant.Object,
+				ObjectName: grant.ObjectName,
+				Grantor: &milvuspb.GrantorEntity{
+					User: grant.Grantor.User,
+					Privilege: &milvuspb.PrivilegeEntity{
+						Name: metaName,
+					},
+				},
+				DbName: grant.DbName,
+			})
 		} else {
-			for _, priv := range groups[grant.Grantor.Privilege.Name] {
+			for _, priv := range privGroup {
+				metaName, err := c.getMetastorePrivilegeName(priv.Name)
+				if err != nil {
+					return nil, err
+				}
 				newGrants = append(newGrants, &milvuspb.GrantEntity{
 					Role:       grant.Role,
 					Object:     grant.Object,
 					ObjectName: grant.ObjectName,
 					Grantor: &milvuspb.GrantorEntity{
-						User:      grant.Grantor.User,
-						Privilege: priv,
+						User: grant.Grantor.User,
+						Privilege: &milvuspb.PrivilegeEntity{
+							Name: metaName,
+						},
 					},
 					DbName: grant.DbName,
 				})
@@ -3198,5 +3291,5 @@ func (c *Core) expandPrivilegeGroups(grants []*milvuspb.GrantEntity, groups map[
 	// uniq by role + object + object name + grantor user + privilege name + db name
 	return lo.UniqBy(newGrants, func(g *milvuspb.GrantEntity) string {
 		return fmt.Sprintf("%s-%s-%s-%s-%s-%s", g.Role, g.Object, g.ObjectName, g.Grantor.User, g.Grantor.Privilege.Name, g.DbName)
-	})
+	}), nil
 }

internal/rootcoord/root_coord_test.go

@@ -1976,6 +1976,8 @@ func TestCore_InitRBAC(t *testing.T) {
 		c := newTestCore(withHealthyCode(), withMeta(meta))
 		meta.EXPECT().CreateRole(mock.Anything, mock.Anything).Return(nil).Twice()
 		meta.EXPECT().OperatePrivilege(mock.Anything, mock.Anything, mock.Anything).Return(nil).Twice()
+		meta.EXPECT().CreatePrivilegeGroup(mock.Anything).Return(nil).Times(len(util.BuiltinPrivilegeGroups))
+		meta.EXPECT().OperatePrivilegeGroup(mock.Anything, mock.Anything, mock.Anything).Return(nil).Times(len(util.BuiltinPrivilegeGroups))
 
 		Params.Save(Params.RoleCfg.Enabled.Key, "false")
 		Params.Save(Params.ProxyCfg.EnablePublicPrivilege.Key, "true")
@@ -1995,6 +1997,8 @@ func TestCore_InitRBAC(t *testing.T) {
 		c := newTestCore(withHealthyCode(), withMeta(meta))
 		meta.EXPECT().CreateRole(mock.Anything, mock.Anything).Return(nil).Times(3)
 		meta.EXPECT().OperatePrivilege(mock.Anything, mock.Anything, mock.Anything).Return(nil).Once()
+		meta.EXPECT().CreatePrivilegeGroup(mock.Anything).Return(nil).Times(len(util.BuiltinPrivilegeGroups))
+		meta.EXPECT().OperatePrivilegeGroup(mock.Anything, mock.Anything, mock.Anything).Return(nil).Times(len(util.BuiltinPrivilegeGroups))
 
 		Params.Save(Params.RoleCfg.Enabled.Key, "true")
 		Params.Save(Params.RoleCfg.Roles.Key, builtinRoles)
@@ -2009,6 +2013,25 @@ func TestCore_InitRBAC(t *testing.T) {
 		err := c.initRbac()
 		assert.NoError(t, err)
 	})
+
+	t.Run("init default privilege groups", func(t *testing.T) {
+		clusterReadWrite := `SelectOwnership,SelectUser,DescribeResourceGroup`
+		meta := mockrootcoord.NewIMetaTable(t)
+		c := newTestCore(withHealthyCode(), withMeta(meta))
+		meta.EXPECT().CreatePrivilegeGroup(mock.Anything).Return(nil).Times(len(util.BuiltinPrivilegeGroups))
+		meta.EXPECT().OperatePrivilegeGroup(mock.Anything, mock.Anything, mock.Anything).Return(nil).Times(len(util.BuiltinPrivilegeGroups))
+
+		Params.Save(Params.RbacConfig.Enabled.Key, "true")
+		Params.Save(Params.RbacConfig.ClusterReadWritePrivileges.Key, clusterReadWrite)
+
+		defer func() {
+			Params.Reset(Params.RbacConfig.Enabled.Key)
+			Params.Reset(Params.RbacConfig.ClusterReadWritePrivileges.Key)
+		}()
+
+		err := c.initBuiltinPrivilegeGroups()
+		assert.NoError(t, err)
+	})
 }
 
 func TestCore_BackupRBAC(t *testing.T) {

pkg/go.mod

@@ -14,7 +14,7 @@ require (
 	github.com/grpc-ecosystem/go-grpc-middleware v1.3.0
 	github.com/json-iterator/go v1.1.12
 	github.com/klauspost/compress v1.17.7
-	github.com/milvus-io/milvus-proto/go-api/v2 v2.3.4-0.20241111062829-6de3d96f664f
+	github.com/milvus-io/milvus-proto/go-api/v2 v2.3.4-0.20241114133823-d3506c6f465c
 	github.com/nats-io/nats-server/v2 v2.10.12
 	github.com/nats-io/nats.go v1.34.1
 	github.com/panjf2000/ants/v2 v2.7.2

pkg/go.sum

@@ -488,8 +488,8 @@ github.com/milvus-io/cgosymbolizer v0.0.0-20240722103217-b7dee0e50119 h1:9VXijWu
 github.com/milvus-io/cgosymbolizer v0.0.0-20240722103217-b7dee0e50119/go.mod h1:DvXTE/K/RtHehxU8/GtDs4vFtfw64jJ3PaCnFri8CRg=
 github.com/milvus-io/gorocksdb v0.0.0-20220624081344-8c5f4212846b h1:TfeY0NxYxZzUfIfYe5qYDBzt4ZYRqzUjTR6CvUzjat8=
 github.com/milvus-io/gorocksdb v0.0.0-20220624081344-8c5f4212846b/go.mod h1:iwW+9cWfIzzDseEBCCeDSN5SD16Tidvy8cwQ7ZY8Qj4=
-github.com/milvus-io/milvus-proto/go-api/v2 v2.3.4-0.20241111062829-6de3d96f664f h1:yLxT8NH0ixUOJMqJuk0xvGf0cKsr+N2xibyTat256PI=
-github.com/milvus-io/milvus-proto/go-api/v2 v2.3.4-0.20241111062829-6de3d96f664f/go.mod h1:/6UT4zZl6awVeXLeE7UGDWZvXj3IWkRsh3mqsn0DiAs=
+github.com/milvus-io/milvus-proto/go-api/v2 v2.3.4-0.20241114133823-d3506c6f465c h1:Ay5w6sTE1QxCydCqqW5N44EcJrMqaqbL5zcp2vclkOw=
+github.com/milvus-io/milvus-proto/go-api/v2 v2.3.4-0.20241114133823-d3506c6f465c/go.mod h1:/6UT4zZl6awVeXLeE7UGDWZvXj3IWkRsh3mqsn0DiAs=
 github.com/milvus-io/pulsar-client-go v0.12.1 h1:O2JZp1tsYiO7C0MQ4hrUY/aJXnn2Gry6hpm7UodghmE=
 github.com/milvus-io/pulsar-client-go v0.12.1/go.mod h1:dkutuH4oS2pXiGm+Ti7fQZ4MRjrMPZ8IJeEGAWMeckk=
 github.com/minio/highwayhash v1.0.2 h1:Aak5U0nElisjDCfPSG79Tgzkn2gl66NxOMspRrKnA/g=