项目文件夹

文件
wehub-resource-sync 498b235461
Build and test / Build and test AMD64 Ubuntu 22.04 (push) Failing after 0s
Publish Builder / amazonlinux2023 (push) Failing after 1s
Build and test / UT for Go (push) Has been skipped
Publish KRTE Images / KRTE (push) Failing after 1s
Build and test / Integration Test (push) Has been skipped
Build and test / Upload Code Coverage (push) Has been skipped
Publish Builder / rockylinux9 (push) Failing after 1s
Publish Builder / ubuntu22.04 (push) Failing after 0s
Publish Builder / ubuntu24.04 (push) Failing after 0s
Publish Gpu Builder / publish-gpu-builder (push) Failing after 1s
Publish Test Images / PyTest (push) Failing after 0s
Build and test / UT for Cpp (push) Has been cancelled
chore: import upstream snapshot with attribution
2026-07-13 12:31:17 +08:00

170 行
4.5 KiB
Go

package milvusclient
import (
"context"
"crypto/tls"
"fmt"
"math"
"net/url"
"regexp"
"strings"
"time"
"github.com/cockroachdb/errors"
"google.golang.org/grpc"
"google.golang.org/grpc/backoff"
"google.golang.org/grpc/keepalive"
)
const (
disableDatabase uint64 = 1 << iota
disableJSON
disableDynamicSchema
disableParitionKey
)
var regexValidScheme = regexp.MustCompile(`^https?:\/\/`)
// DefaultGrpcOpts is GRPC options for milvus client.
var DefaultGrpcOpts = []grpc.DialOption{
grpc.WithBlock(),
grpc.WithKeepaliveParams(keepalive.ClientParameters{
Time: 5 * time.Second,
Timeout: 10 * time.Second,
PermitWithoutStream: true,
}),
grpc.WithConnectParams(grpc.ConnectParams{
Backoff: backoff.Config{
BaseDelay: 100 * time.Millisecond,
Multiplier: 1.6,
Jitter: 0.2,
MaxDelay: 3 * time.Second,
},
MinConnectTimeout: 3 * time.Second,
}),
grpc.WithDefaultCallOptions(
grpc.MaxCallRecvMsgSize(math.MaxInt32), // math.MaxInt32 = 2147483647, 2GB - 1
),
}
// ClientConfig for milvus client.
type ClientConfig struct {
Address string // Remote address, "localhost:19530".
Username string // Username for auth.
Password string // Password for auth.
DBName string // DBName for this client.
EnableTLSAuth bool // Enable TLS Auth for transport security.
APIKey string // API key
tlsConfig *tls.Config // Custom TLS config, set via WithTLSConfig method.
DialOptions []grpc.DialOption // Dial options for GRPC.
RetryRateLimit *RetryRateLimitOption // option for retry on rate limit inteceptor
DisableConn bool
// TelemetryConfig for client telemetry settings
TelemetryConfig *TelemetryConfig
ServerVersion string // ServerVersion
parsedAddress *url.URL
flags uint64 // internal flags
}
type RetryRateLimitOption struct {
MaxRetry uint
MaxBackoff time.Duration
}
func (cfg *ClientConfig) parse() error {
// Prepend default fake tcp:// scheme for remote address.
address := cfg.Address
if !regexValidScheme.MatchString(address) {
address = fmt.Sprintf("tcp://%s", address)
}
remoteURL, err := url.Parse(address)
if err != nil {
return errors.Wrap(err, "milvus address parse fail")
}
// Remote Host should never be empty.
if remoteURL.Host == "" {
return errors.New("empty remote host of milvus address")
}
// Use DBName in remote url path.
if cfg.DBName == "" {
cfg.DBName = strings.TrimLeft(remoteURL.Path, "/")
}
// Always enable tls auth for https remote url.
if remoteURL.Scheme == "https" {
cfg.EnableTLSAuth = true
}
if remoteURL.Port() == "" && cfg.EnableTLSAuth {
remoteURL.Host += ":443"
}
cfg.parsedAddress = remoteURL
return nil
}
// Get parsed remote milvus address, should be called after parse was called.
func (c *ClientConfig) getParsedAddress() string {
return c.parsedAddress.Host
}
// useDatabase change the inner db name.
func (c *ClientConfig) useDatabase(dbName string) {
c.DBName = dbName
}
func (c *ClientConfig) setServerInfo(serverInfo string) {
c.ServerVersion = serverInfo
}
func (c *ClientConfig) getRetryOnRateLimitInterceptor() grpc.UnaryClientInterceptor {
if c.RetryRateLimit == nil {
c.RetryRateLimit = c.defaultRetryRateLimitOption()
}
return RetryOnRateLimitInterceptor(c.RetryRateLimit.MaxRetry, c.RetryRateLimit.MaxBackoff, func(ctx context.Context, attempt uint) time.Duration {
return 10 * time.Millisecond * time.Duration(math.Pow(3, float64(attempt)))
})
}
func (c *ClientConfig) defaultRetryRateLimitOption() *RetryRateLimitOption {
return &RetryRateLimitOption{
MaxRetry: 75,
MaxBackoff: 3 * time.Second,
}
}
// addFlags set internal flags
func (c *ClientConfig) addFlags(flags uint64) {
c.flags |= flags
}
// hasFlags check flags is set
func (c *ClientConfig) hasFlags(flags uint64) bool {
return (c.flags & flags) > 0
}
func (c *ClientConfig) resetFlags(flags uint64) {
c.flags &= ^flags
}
// WithTLSConfig sets the custom TLS configuration and enables TLS auth.
// This method should be used to configure custom TLS settings (e.g., mTLS, custom CA).
func (c *ClientConfig) WithTLSConfig(tlsConfig *tls.Config) *ClientConfig {
c.tlsConfig = tlsConfig
c.EnableTLSAuth = true
return c
}
// WithGrpcAuthority sets the gRPC :authority header, used for proxy-based routing.
// DefaultGrpcOpts are always applied by dialOptions(), so only the authority option is needed here.
func (c *ClientConfig) WithGrpcAuthority(authority string) *ClientConfig {
c.DialOptions = []grpc.DialOption{grpc.WithAuthority(authority)}
return c
}