fix helm values

Signed-off-by: Cai Zhang <[email protected]>
milvus-io · Jan 9, 2025 · c941d12 · c941d12
1 parent 242bff0
commit c941d12
Show file tree

Hide file tree

Showing 4 changed files with 38 additions and 28 deletions.
diff --git a/internal/indexnode/task_stats.go b/internal/indexnode/task_stats.go
@@ -120,12 +120,13 @@ func (st *statsTask) PreExecute(ctx context.Context) error {
 	defer span.End()
 
 	st.queueDur = st.tr.RecordSpan()
-	log.Ctx(ctx).Info("Begin to prepare stats task",
+	log.Ctx(ctx).Info("Begin to PreExecute stats task",
 		zap.String("clusterID", st.req.GetClusterID()),
 		zap.Int64("taskID", st.req.GetTaskID()),
 		zap.Int64("collectionID", st.req.GetCollectionID()),
 		zap.Int64("partitionID", st.req.GetPartitionID()),
 		zap.Int64("segmentID", st.req.GetSegmentID()),
+		zap.Int64("queue duration(ms)", st.queueDur.Milliseconds()),
 	)
 
 	if err := binlog.DecompressBinLogWithRootPath(st.req.GetStorageConfig().GetRootPath(), storage.InsertBinlog, st.req.GetCollectionID(), st.req.GetPartitionID(),
@@ -156,6 +157,16 @@ func (st *statsTask) PreExecute(ctx context.Context) error {
 		}
 	}
 
+	preExecuteRecordSpan := st.tr.RecordSpan()
+
+	log.Ctx(ctx).Info("successfully PreExecute stats task",
+		zap.String("clusterID", st.req.GetClusterID()),
+		zap.Int64("taskID", st.req.GetTaskID()),
+		zap.Int64("collectionID", st.req.GetCollectionID()),
+		zap.Int64("partitionID", st.req.GetPartitionID()),
+		zap.Int64("segmentID", st.req.GetSegmentID()),
+		zap.Int64("preExecuteRecordSpan(ms)", preExecuteRecordSpan.Milliseconds()),
+	)
 	return nil
 }
 
@@ -175,25 +186,20 @@ func (st *statsTask) sortSegment(ctx context.Context) ([]*datapb.FieldBinlog, er
 
 		allBinlogs    = make(map[typeutil.UniqueID]*datapb.FieldBinlog) // All binlog meta of a segment
 		uploadFutures = make([]*conc.Future[any], 0)
-
-		downloadCost     time.Duration
-		serWriteTimeCost time.Duration
-		sortTimeCost     time.Duration
 	)
 
-	downloadStart := time.Now()
 	values, err := st.downloadData(ctx, numRows, writer.GetPkID(), bm25FieldIds)
 	if err != nil {
 		log.Ctx(ctx).Warn("download data failed", zap.Int64("taskID", st.req.GetTaskID()), zap.Error(err))
 		return nil, err
 	}
-	downloadCost = time.Since(downloadStart)
 
-	sortStart := time.Now()
+	downloadRecordSpan := st.tr.RecordSpan()
+
 	sort.Slice(values, func(i, j int) bool {
 		return values[i].PK.LT(values[j].PK)
 	})
-	sortTimeCost += time.Since(sortStart)
+	sortRecordSpan := st.tr.RecordSpan()
 
 	for i, v := range values {
 		err := writer.Write(v)
@@ -203,13 +209,11 @@ func (st *statsTask) sortSegment(ctx context.Context) ([]*datapb.FieldBinlog, er
 		}
 
 		if (i+1)%statsBatchSize == 0 && writer.IsFullWithBinlogMaxSize(st.req.GetBinlogMaxSize()) {
-			serWriteStart := time.Now()
 			binlogNum, kvs, partialBinlogs, err := serializeWrite(ctx, st.req.GetStorageConfig().GetRootPath(), st.req.GetStartLogID()+st.logIDOffset, writer)
 			if err != nil {
 				log.Ctx(ctx).Warn("stats wrong, failed to serialize writer", zap.Int64("taskID", st.req.GetTaskID()), zap.Error(err))
 				return nil, err
 			}
-			serWriteTimeCost += time.Since(serWriteStart)
 
 			uploadFutures = append(uploadFutures, st.binlogIO.AsyncUpload(ctx, kvs)...)
 			mergeFieldBinlogs(allBinlogs, partialBinlogs)
@@ -226,34 +230,35 @@ func (st *statsTask) sortSegment(ctx context.Context) ([]*datapb.FieldBinlog, er
 	}
 
 	if !writer.FlushAndIsEmpty() {
-		serWriteStart := time.Now()
 		binlogNum, kvs, partialBinlogs, err := serializeWrite(ctx, st.req.GetStorageConfig().GetRootPath(), st.req.GetStartLogID()+st.logIDOffset, writer)
 		if err != nil {
 			log.Ctx(ctx).Warn("stats wrong, failed to serialize writer", zap.Int64("taskID", st.req.GetTaskID()), zap.Error(err))
 			return nil, err
 		}
-		serWriteTimeCost += time.Since(serWriteStart)
 		st.logIDOffset += binlogNum
 
 		uploadFutures = append(uploadFutures, st.binlogIO.AsyncUpload(ctx, kvs)...)
 		mergeFieldBinlogs(allBinlogs, partialBinlogs)
 		flushBatchCount++
 	}
 
+	writeRecordSpan := st.tr.RecordSpan()
+
 	err = conc.AwaitAll(uploadFutures...)
 	if err != nil {
 		log.Ctx(ctx).Warn("stats wrong, failed to upload kvs", zap.Int64("taskID", st.req.GetTaskID()), zap.Error(err))
 		return nil, err
 	}
 
-	serWriteStart := time.Now()
+	waitUploadRecordSpan := st.tr.RecordSpan()
+
 	binlogNums, sPath, err := statSerializeWrite(ctx, st.req.GetStorageConfig().GetRootPath(), st.binlogIO, st.req.GetStartLogID()+st.logIDOffset, writer, numRows)
 	if err != nil {
 		log.Ctx(ctx).Warn("stats wrong, failed to serialize write segment stats", zap.Int64("taskID", st.req.GetTaskID()),
 			zap.Int64("remaining row count", numRows), zap.Error(err))
 		return nil, err
 	}
-	serWriteTimeCost += time.Since(serWriteStart)
+	statslogRecordSpan := st.tr.RecordSpan()
 
 	st.logIDOffset += binlogNums
 
@@ -271,7 +276,7 @@ func (st *statsTask) sortSegment(ctx context.Context) ([]*datapb.FieldBinlog, er
 		}
 	}
 
-	totalElapse := st.tr.RecordSpan()
+	bm25LogRecordSpan := st.tr.RecordSpan()
 
 	insertLogs := lo.Values(allBinlogs)
 	if err := binlog.CompressFieldBinlogs(insertLogs); err != nil {
@@ -291,6 +296,8 @@ func (st *statsTask) sortSegment(ctx context.Context) ([]*datapb.FieldBinlog, er
 		st.req.GetInsertChannel(),
 		int64(len(values)), insertLogs, statsLogs, bm25StatsLogs)
 
+	totalElapse := st.tr.ElapseSpan()
+
 	log.Ctx(ctx).Info("sort segment end",
 		zap.String("clusterID", st.req.GetClusterID()),
 		zap.Int64("taskID", st.req.GetTaskID()),
@@ -302,9 +309,12 @@ func (st *statsTask) sortSegment(ctx context.Context) ([]*datapb.FieldBinlog, er
 		zap.Int64("old rows", numRows),
 		zap.Int("valid rows", len(values)),
 		zap.Int("binlog batch count", flushBatchCount),
-		zap.Duration("download elapse", downloadCost),
-		zap.Duration("sort elapse", sortTimeCost),
-		zap.Duration("serWrite elapse", serWriteTimeCost),
+		zap.Duration("download elapse", downloadRecordSpan),
+		zap.Duration("sort elapse", sortRecordSpan),
+		zap.Duration("write elapse", writeRecordSpan),
+		zap.Duration("wait upload elapse", waitUploadRecordSpan),
+		zap.Duration("statslog elapse", statslogRecordSpan),
+		zap.Duration("sbm25log elapse", bm25LogRecordSpan),
 		zap.Duration("total elapse", totalElapse))
 	return insertLogs, nil
 }

diff --git a/tests/_helm/values/e2e/standalone b/tests/_helm/values/e2e/standalone
@@ -99,9 +99,9 @@ extraConfigFiles:
     indexCoord:
       scheduler:
         interval: 100
-    indexNode:
-      scheduler:
-        buildParallel: 4
+    dataCoord:
+      slot:
+        indexTaskSlotUsage: 4
 metrics:
   serviceMonitor:
     enabled: true

diff --git a/tests/_helm/values/e2e/standalone-kafka-mmap b/tests/_helm/values/e2e/standalone-kafka-mmap
@@ -107,9 +107,9 @@ extraConfigFiles:
     indexCoord:
       scheduler:
         interval: 100
-    indexNode:
-      scheduler:
-        buildParallel: 4
+    dataCoord:
+      slot:
+        indexTaskSlotUsage: 4
 metrics:
   serviceMonitor:
     enabled: true

diff --git a/tests/_helm/values/e2e/standalone-one-pod b/tests/_helm/values/e2e/standalone-one-pod
@@ -39,9 +39,9 @@ extraConfigFiles:
     indexCoord:
       scheduler:
         interval: 100
-    indexNode:
-      scheduler:
-        buildParallel: 4
+    dataCoord:
+      slot:
+        indexTaskSlotUsage: 4
 indexNode:
   disk:
     enabled: true