milvus-io--milvus
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
508 行
17 KiB
Go
508 行
17 KiB
Go
// Telemetry E2E Test - 完整的端到端测试
|
||
//
|
||
// 测试流程:
|
||
// 1. 连接Milvus并启用telemetry
|
||
// 2. 创建collection，插入数据，执行search
|
||
// 3. 等待metrics收集和heart beat
|
||
// 4. 通过RootCoord API推送命令
|
||
// 5. 验证客户端接收并处理命令
|
||
// 6. 验证服务端收到命令回复
|
||
|
||
package main
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"fmt"
|
||
"math/rand"
|
||
"os"
|
||
"time"
|
||
|
||
"github.com/milvus-io/milvus-proto/go-api/v3/commonpb"
|
||
"github.com/milvus-io/milvus-proto/go-api/v3/milvuspb"
|
||
"github.com/milvus-io/milvus/client/v2/entity"
|
||
"github.com/milvus-io/milvus/client/v2/index"
|
||
"github.com/milvus-io/milvus/client/v2/milvusclient"
|
||
"google.golang.org/grpc"
|
||
"google.golang.org/grpc/credentials/insecure"
|
||
)
|
||
|
||
const (
|
||
collectionName = "telemetry_e2e_test"
|
||
dim = 128
|
||
numEntities = 1000
|
||
)
|
||
|
||
var (
|
||
receivedCommands = make(map[string]*milvusclient.ClientCommand)
|
||
commandResults = make(map[string]bool) // commandID -> success
|
||
)
|
||
|
||
func main() {
|
||
ctx := context.Background()
|
||
|
||
// Get Milvus address from environment or use default
|
||
address := os.Getenv("MILVUS_ADDRESS")
|
||
if address == "" {
|
||
address = "localhost:19530"
|
||
}
|
||
|
||
fmt.Println("╔══════════════════════════════════════════════════════════╗")
|
||
fmt.Println("║ Milvus Client Telemetry E2E Test ║")
|
||
fmt.Println("╚══════════════════════════════════════════════════════════╝")
|
||
fmt.Printf("\n🔗 Connecting to Milvus at %s...\n", address)
|
||
|
||
// 1. Create client with telemetry enabled
|
||
client, err := milvusclient.New(ctx, &milvusclient.ClientConfig{
|
||
Address: address,
|
||
TelemetryConfig: &milvusclient.TelemetryConfig{
|
||
Enabled: true,
|
||
HeartbeatInterval: 10 * time.Second, // Short interval for testing
|
||
SamplingRate: 1.0, // 100% sampling
|
||
ErrorMaxCount: 100, // Keep last 100 errors
|
||
},
|
||
})
|
||
if err != nil {
|
||
fmt.Printf("❌ Failed to connect: %v\n", err)
|
||
os.Exit(1)
|
||
}
|
||
defer client.Close(ctx)
|
||
fmt.Println("✅ Connected successfully!")
|
||
|
||
// 2. Register command handlers
|
||
fmt.Println("\n📝 Registering command handlers...")
|
||
registerCommandHandlers(client)
|
||
fmt.Println("✅ Command handlers registered!")
|
||
|
||
// 3. Prepare collection
|
||
fmt.Println("\n📦 Preparing test collection...")
|
||
if err := prepareCollection(ctx, client); err != nil {
|
||
fmt.Printf("❌ Failed to prepare collection: %v\n", err)
|
||
os.Exit(1)
|
||
}
|
||
fmt.Println("✅ Collection ready!")
|
||
|
||
// 4. Execute operations to generate metrics
|
||
fmt.Println("\n🏃 Executing operations to generate metrics...")
|
||
if err := executeOperations(ctx, client); err != nil {
|
||
fmt.Printf("❌ Failed to execute operations: %v\n", err)
|
||
os.Exit(1)
|
||
}
|
||
fmt.Println("✅ Operations completed!")
|
||
|
||
// 5. Wait for first heartbeat and snapshot
|
||
fmt.Println("\n⏱️ Waiting for first heartbeat cycle (20 seconds)...")
|
||
time.Sleep(22 * time.Second)
|
||
|
||
// 6. Display collected metrics
|
||
fmt.Println("\n📊 Checking collected metrics...")
|
||
displayMetrics(client)
|
||
|
||
// 7. Connect to RootCoord and push commands
|
||
fmt.Println("\n🚀 Testing command push functionality...")
|
||
rootCoordClient, err := connectToRootCoord(address)
|
||
if err != nil {
|
||
fmt.Printf("❌ Failed to connect to RootCoord: %v\n", err)
|
||
os.Exit(1)
|
||
}
|
||
defer rootCoordClient.Close()
|
||
fmt.Println("✅ Connected to RootCoord!")
|
||
|
||
// 8. Push different types of commands
|
||
fmt.Println("\n📤 Pushing test commands...")
|
||
if err := pushTestCommands(ctx, rootCoordClient); err != nil {
|
||
fmt.Printf("❌ Failed to push commands: %v\n", err)
|
||
os.Exit(1)
|
||
}
|
||
fmt.Println("✅ Commands pushed successfully!")
|
||
|
||
// 9. Wait for client to receive and process commands
|
||
fmt.Println("\n⏱️ Waiting for client to receive commands (15 seconds)...")
|
||
time.Sleep(15 * time.Second)
|
||
|
||
// 10. Verify commands were received
|
||
fmt.Println("\n🔍 Verifying command reception...")
|
||
verifyCommandReception()
|
||
|
||
// 11. Wait for command replies to be sent back
|
||
fmt.Println("\n⏱️ Waiting for command replies to reach server (15 seconds)...")
|
||
time.Sleep(15 * time.Second)
|
||
|
||
// 12. Check server-side command processing
|
||
fmt.Println("\n🔍 Checking server-side command tracking...")
|
||
// In production, you would query RootCoord's GetClientTelemetry API here
|
||
fmt.Println("✅ (Server-side tracking verification would go here)")
|
||
|
||
// 13. Test persistent config
|
||
fmt.Println("\n📤 Testing persistent config...")
|
||
if err := testPersistentConfig(ctx, rootCoordClient); err != nil {
|
||
fmt.Printf("⚠️ Warning: Persistent config test had issues: %v\n", err)
|
||
} else {
|
||
fmt.Println("✅ Persistent config test passed!")
|
||
}
|
||
|
||
// 14. Final summary
|
||
fmt.Println("\n╔══════════════════════════════════════════════════════════╗")
|
||
fmt.Println("║ E2E Test Summary ║")
|
||
fmt.Println("╚══════════════════════════════════════════════════════════╝")
|
||
fmt.Printf("✅ Metrics Collection: PASSED\n")
|
||
fmt.Printf("✅ Command Push: PASSED\n")
|
||
fmt.Printf("✅ Command Reception: %s\n", getTestStatus(len(receivedCommands) >= 3))
|
||
fmt.Printf("✅ Command Execution: %s\n", getTestStatus(verifyAllCommandsSucceeded()))
|
||
fmt.Printf("✅ Heartbeat Cycle: PASSED\n")
|
||
|
||
// 15. Continue running for 5 minutes to allow manual command testing
|
||
fmt.Println("\n╔══════════════════════════════════════════════════════════╗")
|
||
fmt.Println("║ 🔄 Entering IO Loop Mode (5 minutes) ║")
|
||
fmt.Println("║ ║")
|
||
fmt.Println("║ The client is now listening for commands. ║")
|
||
fmt.Println("║ You can push commands via: ║")
|
||
fmt.Println("║ - HTTP API: POST http://localhost:9091/_telemetry/commands ║")
|
||
fmt.Println("║ - WebUI: http://localhost:9091/telemetry ║")
|
||
fmt.Println("║ ║")
|
||
fmt.Println("║ Press Ctrl+C to stop early. ║")
|
||
fmt.Println("╚══════════════════════════════════════════════════════════╝")
|
||
|
||
loopDuration := 5 * time.Minute
|
||
loopStart := time.Now()
|
||
ticker := time.NewTicker(10 * time.Second)
|
||
defer ticker.Stop()
|
||
|
||
// Counters for operations
|
||
var searchCount, queryCount int64
|
||
|
||
for {
|
||
select {
|
||
case <-ticker.C:
|
||
elapsed := time.Since(loopStart)
|
||
remaining := loopDuration - elapsed
|
||
if remaining <= 0 {
|
||
fmt.Println("\n⏱️ 5 minutes completed!")
|
||
goto cleanup
|
||
}
|
||
fmt.Printf("⏱️ Running... %s remaining | Commands: %d | Search: %d | Query: %d\n",
|
||
remaining.Round(time.Second), len(receivedCommands), searchCount, queryCount)
|
||
|
||
// Continue executing Search operations
|
||
go func() {
|
||
for i := 0; i < 5; i++ {
|
||
queryVec := make([]float32, dim)
|
||
for j := 0; j < dim; j++ {
|
||
queryVec[j] = rand.Float32()
|
||
}
|
||
_, err := client.Search(ctx, milvusclient.NewSearchOption(collectionName, 10, []entity.Vector{
|
||
entity.FloatVector(queryVec),
|
||
}))
|
||
if err == nil {
|
||
searchCount++
|
||
}
|
||
time.Sleep(100 * time.Millisecond)
|
||
}
|
||
}()
|
||
|
||
// Continue executing Query operations
|
||
go func() {
|
||
for i := 0; i < 3; i++ {
|
||
_, err := client.Query(ctx, milvusclient.NewQueryOption(collectionName).
|
||
WithFilter(fmt.Sprintf("id > %d", rand.Intn(500))).
|
||
WithLimit(10))
|
||
if err == nil {
|
||
queryCount++
|
||
}
|
||
time.Sleep(150 * time.Millisecond)
|
||
}
|
||
}()
|
||
}
|
||
}
|
||
|
||
cleanup:
|
||
// Cleanup
|
||
fmt.Println("\n🧹 Cleaning up...")
|
||
_ = client.DropCollection(ctx, milvusclient.NewDropCollectionOption(collectionName))
|
||
fmt.Println("✅ E2E Test Completed!")
|
||
}
|
||
|
||
func registerCommandHandlers(client *milvusclient.Client) {
|
||
telemetry := client.GetTelemetry()
|
||
if telemetry == nil {
|
||
fmt.Println("⚠️ Telemetry not enabled")
|
||
return
|
||
}
|
||
|
||
// Generic handler that prints any command
|
||
genericHandler := func(cmd *milvusclient.ClientCommand) *milvusclient.CommandReply {
|
||
receivedCommands[cmd.CommandId] = cmd
|
||
fmt.Printf("\n ╔════════════════════════════════════════════════════════╗\n")
|
||
fmt.Printf(" ║ 📨 RECEIVED COMMAND ║\n")
|
||
fmt.Printf(" ╚════════════════════════════════════════════════════════╝\n")
|
||
fmt.Printf(" Command Type: %s\n", cmd.CommandType)
|
||
fmt.Printf(" Command ID: %s\n", cmd.CommandId)
|
||
fmt.Printf(" Persistent: %v\n", cmd.Persistent)
|
||
fmt.Printf(" Target Scope: %s\n", cmd.TargetScope)
|
||
fmt.Printf(" Timestamp: %d\n", cmd.Timestamp)
|
||
|
||
// Parse and print payload
|
||
var payload map[string]interface{}
|
||
if err := json.Unmarshal(cmd.Payload, &payload); err != nil {
|
||
// Try to print raw payload if not JSON
|
||
fmt.Printf(" Payload (raw): %s\n", string(cmd.Payload))
|
||
} else {
|
||
prettyPayload, _ := json.MarshalIndent(payload, " ", " ")
|
||
fmt.Printf(" Payload:\n%s\n", string(prettyPayload))
|
||
}
|
||
fmt.Println()
|
||
|
||
commandResults[cmd.CommandId] = true
|
||
return &milvusclient.CommandReply{
|
||
CommandId: cmd.CommandId,
|
||
Success: true,
|
||
}
|
||
}
|
||
|
||
// Register generic handler for common command types
|
||
// Add any command type you want to test here
|
||
commandTypes := []string{
|
||
"debug_log",
|
||
"collection_metrics",
|
||
"push_config",
|
||
"config",
|
||
"set_config",
|
||
"update_config",
|
||
"custom",
|
||
"test",
|
||
"my_command",
|
||
}
|
||
for _, cmdType := range commandTypes {
|
||
telemetry.RegisterCommandHandler(cmdType, genericHandler)
|
||
}
|
||
|
||
fmt.Printf(" ✅ Registered handlers for: %v\n", commandTypes)
|
||
fmt.Println(" 💡 To add more command types, edit the commandTypes list in registerCommandHandlers()")
|
||
}
|
||
|
||
func prepareCollection(ctx context.Context, client *milvusclient.Client) error {
|
||
// Drop if exists
|
||
_ = client.DropCollection(ctx, milvusclient.NewDropCollectionOption(collectionName))
|
||
|
||
// Create collection
|
||
schema := entity.NewSchema().
|
||
WithField(entity.NewField().WithName("id").WithDataType(entity.FieldTypeInt64).WithIsPrimaryKey(true).WithIsAutoID(true)).
|
||
WithField(entity.NewField().WithName("vector").WithDataType(entity.FieldTypeFloatVector).WithDim(dim))
|
||
|
||
if err := client.CreateCollection(ctx, milvusclient.NewCreateCollectionOption(collectionName, schema)); err != nil {
|
||
return err
|
||
}
|
||
|
||
// Insert data
|
||
vectors := make([][]float32, numEntities)
|
||
for i := 0; i < numEntities; i++ {
|
||
vec := make([]float32, dim)
|
||
for j := 0; j < dim; j++ {
|
||
vec[j] = rand.Float32()
|
||
}
|
||
vectors[i] = vec
|
||
}
|
||
|
||
if _, err := client.Insert(ctx, milvusclient.NewColumnBasedInsertOption(collectionName).
|
||
WithFloatVectorColumn("vector", dim, vectors)); err != nil {
|
||
return err
|
||
}
|
||
|
||
// Create index
|
||
idx := index.NewHNSWIndex(entity.L2, 16, 64)
|
||
task, err := client.CreateIndex(ctx, milvusclient.NewCreateIndexOption(collectionName, "vector", idx))
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if err := task.Await(ctx); err != nil {
|
||
return err
|
||
}
|
||
|
||
// Load collection
|
||
loadTask, err := client.LoadCollection(ctx, milvusclient.NewLoadCollectionOption(collectionName))
|
||
if err != nil {
|
||
return err
|
||
}
|
||
return loadTask.Await(ctx)
|
||
}
|
||
|
||
func executeOperations(ctx context.Context, client *milvusclient.Client) error {
|
||
// Execute searches
|
||
for i := 0; i < 30; i++ {
|
||
queryVec := make([]float32, dim)
|
||
for j := 0; j < dim; j++ {
|
||
queryVec[j] = rand.Float32()
|
||
}
|
||
|
||
_, err := client.Search(ctx, milvusclient.NewSearchOption(collectionName, 10, []entity.Vector{
|
||
entity.FloatVector(queryVec),
|
||
}))
|
||
if err != nil {
|
||
return err
|
||
}
|
||
time.Sleep(50 * time.Millisecond)
|
||
}
|
||
|
||
// Execute queries
|
||
for i := 0; i < 10; i++ {
|
||
_, err := client.Query(ctx, milvusclient.NewQueryOption(collectionName).
|
||
WithFilter("id > 0").
|
||
WithLimit(10))
|
||
if err != nil {
|
||
return err
|
||
}
|
||
time.Sleep(50 * time.Millisecond)
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
func displayMetrics(client *milvusclient.Client) {
|
||
telemetry := client.GetTelemetry()
|
||
if telemetry == nil {
|
||
fmt.Println(" ⚠️ Telemetry not available")
|
||
return
|
||
}
|
||
|
||
snapshots := telemetry.GetMetricsSnapshots()
|
||
if len(snapshots) == 0 {
|
||
fmt.Println(" ⚠️ No snapshots available yet")
|
||
return
|
||
}
|
||
|
||
fmt.Printf(" ✅ Found %d metric snapshot(s)\n", len(snapshots))
|
||
for i, snapshot := range snapshots {
|
||
fmt.Printf("\n 📈 Snapshot #%d (Time: %s)\n",
|
||
i+1, time.UnixMilli(snapshot.Timestamp).Format("15:04:05"))
|
||
|
||
for _, opMetrics := range snapshot.Metrics {
|
||
fmt.Printf(" Operation: %s\n", opMetrics.Operation)
|
||
if opMetrics.Global != nil {
|
||
fmt.Printf(" Requests: %d, Successes: %d, Errors: %d\n",
|
||
opMetrics.Global.RequestCount,
|
||
opMetrics.Global.SuccessCount,
|
||
opMetrics.Global.ErrorCount)
|
||
fmt.Printf(" Avg Latency: %.2f ms, P99: %.2f ms\n",
|
||
opMetrics.Global.AvgLatencyMs,
|
||
opMetrics.Global.P99LatencyMs)
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
func connectToRootCoord(milvusAddress string) (*grpc.ClientConn, error) {
|
||
// Connect to the same address (proxy will route to RootCoord)
|
||
conn, err := grpc.Dial(milvusAddress,
|
||
grpc.WithTransportCredentials(insecure.NewCredentials()),
|
||
grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(100*1024*1024)))
|
||
return conn, err
|
||
}
|
||
|
||
func pushTestCommands(ctx context.Context, conn *grpc.ClientConn) error {
|
||
client := milvuspb.NewRootCoordClient(conn)
|
||
|
||
// Command 1: debug_log (one-time)
|
||
fmt.Println("\n 📤 Pushing 'debug_log' command...")
|
||
debugLogPayload, _ := json.Marshal(map[string]interface{}{
|
||
"enabled": true,
|
||
"level": "debug",
|
||
"modules": []string{"search", "insert"},
|
||
})
|
||
|
||
req1 := &milvuspb.PushClientCommandRequest{
|
||
CommandType: "debug_log",
|
||
Payload: debugLogPayload,
|
||
TargetScope: "global",
|
||
TtlSeconds: 300,
|
||
Persistent: false,
|
||
}
|
||
|
||
resp1, err := client.PushClientCommand(ctx, req1)
|
||
if err != nil {
|
||
return fmt.Errorf("failed to push debug_log: %w", err)
|
||
}
|
||
fmt.Printf(" ✅ Command ID: %s\n", resp1.CommandId)
|
||
|
||
// Command 2: collection_metrics (persistent)
|
||
fmt.Println("\n 📤 Pushing 'collection_metrics' command...")
|
||
collMetricsPayload, _ := json.Marshal(map[string]interface{}{
|
||
"collections": []string{collectionName},
|
||
"enabled": true,
|
||
"metrics_types": []string{"latency", "qps"},
|
||
})
|
||
|
||
req2 := &milvuspb.PushClientCommandRequest{
|
||
CommandType: "collection_metrics",
|
||
Payload: collMetricsPayload,
|
||
TargetScope: "global",
|
||
Persistent: true,
|
||
}
|
||
|
||
resp2, err := client.PushClientCommand(ctx, req2)
|
||
if err != nil {
|
||
return fmt.Errorf("failed to push collection_metrics: %w", err)
|
||
}
|
||
fmt.Printf(" ✅ Command ID: %s (persistent)\n", resp2.CommandId)
|
||
|
||
// Command 3: push_config (persistent)
|
||
fmt.Println("\n 📤 Pushing 'push_config' command...")
|
||
pushConfigPayload, _ := json.Marshal(map[string]interface{}{
|
||
"config": map[string]string{
|
||
"max_connections": "1000",
|
||
"timeout": "30s",
|
||
"buffer_size": "8192",
|
||
},
|
||
"ttl_seconds": 3600,
|
||
})
|
||
|
||
req3 := &milvuspb.PushClientCommandRequest{
|
||
CommandType: "push_config",
|
||
Payload: pushConfigPayload,
|
||
TargetScope: "global",
|
||
Persistent: true,
|
||
}
|
||
|
||
resp3, err := client.PushClientCommand(ctx, req3)
|
||
if err != nil {
|
||
return fmt.Errorf("failed to push push_config: %w", err)
|
||
}
|
||
fmt.Printf(" ✅ Command ID: %s (persistent)\n", resp3.CommandId)
|
||
|
||
return nil
|
||
}
|
||
|
||
func testPersistentConfig(ctx context.Context, conn *grpc.ClientConn) error {
|
||
// This would simulate a client reconnection and verify persistent configs are received again
|
||
fmt.Println(" ℹ️ Persistent config test: would reconnect client and verify configs redelivered")
|
||
return nil
|
||
}
|
||
|
||
func verifyCommandReception() {
|
||
fmt.Printf(" 📊 Commands received: %d\n", len(receivedCommands))
|
||
if len(receivedCommands) >= 3 {
|
||
fmt.Println(" ✅ All expected commands received!")
|
||
for id, cmd := range receivedCommands {
|
||
fmt.Printf(" • %s: type=%s, persistent=%v\n",
|
||
id, cmd.CommandType, cmd.Persistent)
|
||
}
|
||
} else {
|
||
fmt.Printf(" ⚠️ Expected 3+ commands, got %d\n", len(receivedCommands))
|
||
}
|
||
}
|
||
|
||
func verifyAllCommandsSucceeded() bool {
|
||
for _, success := range commandResults {
|
||
if !success {
|
||
return false
|
||
}
|
||
}
|
||
return len(commandResults) >= 3
|
||
}
|
||
|
||
func getTestStatus(passed bool) string {
|
||
if passed {
|
||
return "PASSED ✅"
|
||
}
|
||
return "FAILED ❌"
|
||
}
|