Add list import tasks api

internal/datacoord/mock_test.go

@@ -460,6 +460,11 @@ func (m *mockRootCoordService) GetImportState(ctx context.Context, req *milvuspb
 	panic("not implemented") // TODO: Implement
 }
 
+// Returns id array of all import tasks
+func (m *mockRootCoordService) ListImportTasks(ctx context.Context, in *milvuspb.ListImportTasksRequest) (*milvuspb.ListImportTasksResponse, error) {
+	panic("not implemented") // TODO: Implement
+}
+
 // Report impot task state to rootcoord
 func (m *mockRootCoordService) ReportImport(ctx context.Context, req *rootcoordpb.ImportResult) (*commonpb.Status, error) {
 	panic("not implemented") // TODO: Implement

internal/datanode/data_node.go

@@ -61,6 +61,7 @@ import (
 	"github.com/milvus-io/milvus/internal/util/paramtable"
 	"github.com/milvus-io/milvus/internal/util/retry"
 	"github.com/milvus-io/milvus/internal/util/sessionutil"
+	"github.com/milvus-io/milvus/internal/util/timerecord"
 	"github.com/milvus-io/milvus/internal/util/typeutil"
 )
 
@@ -878,9 +879,11 @@ func importFlushReqFunc(node *DataNode, req *datapb.ImportTaskRequest, res *root
 			)
 			return fmt.Errorf("syncSegmentID Failed: invalid shard number %d", shardNum)
 		}
-		log.Info("import task flush segment",
-			zap.Any("channel names", req.ImportTask.ChannelNames),
-			zap.Int("shard num", shardNum))
+
+		tr := timerecord.NewTimeRecorder("import callback function")
+		defer tr.Elapse("finished")
+
+		log.Info("import task flush segment", zap.Any("ChannelNames", req.ImportTask.ChannelNames), zap.Int("shardNum", shardNum))
 		segReqs := []*datapb.SegmentIDRequest{
 			{
 				ChannelName:  req.ImportTask.ChannelNames[shardNum],

internal/distributed/proxy/service.go

@@ -669,6 +669,10 @@ func (s *Server) GetImportState(ctx context.Context, req *milvuspb.GetImportStat
 	return s.proxy.GetImportState(ctx, req)
 }
 
+func (s *Server) ListImportTasks(ctx context.Context, req *milvuspb.ListImportTasksRequest) (*milvuspb.ListImportTasksResponse, error) {
+	return s.proxy.ListImportTasks(ctx, req)
+}
+
 func (s *Server) GetReplicas(ctx context.Context, req *milvuspb.GetReplicasRequest) (*milvuspb.GetReplicasResponse, error) {
 	return s.proxy.GetReplicas(ctx, req)
 }

internal/distributed/proxy/service_test.go

@@ -210,6 +210,10 @@ func (m *MockRootCoord) GetImportState(ctx context.Context, req *milvuspb.GetImp
 	return nil, nil
 }
 
+func (m *MockRootCoord) ListImportTasks(ctx context.Context, in *milvuspb.ListImportTasksRequest) (*milvuspb.ListImportTasksResponse, error) {
+	return nil, nil
+}
+
 func (m *MockRootCoord) ReportImport(ctx context.Context, req *rootcoordpb.ImportResult) (*commonpb.Status, error) {
 	return nil, nil
 }
@@ -736,6 +740,10 @@ func (m *MockProxy) GetImportState(ctx context.Context, req *milvuspb.GetImportS
 	return nil, nil
 }
 
+func (m *MockProxy) ListImportTasks(ctx context.Context, in *milvuspb.ListImportTasksRequest) (*milvuspb.ListImportTasksResponse, error) {
+	return nil, nil
+}
+
 func (m *MockProxy) GetReplicas(ctx context.Context, req *milvuspb.GetReplicasRequest) (*milvuspb.GetReplicasResponse, error) {
 	return nil, nil
 }

internal/distributed/rootcoord/client/client.go

@@ -517,6 +517,20 @@ func (c *Client) GetImportState(ctx context.Context, req *milvuspb.GetImportStat
 	return ret.(*milvuspb.GetImportStateResponse), err
 }
 
+// List id array of all import tasks
+func (c *Client) ListImportTasks(ctx context.Context, req *milvuspb.ListImportTasksRequest) (*milvuspb.ListImportTasksResponse, error) {
+	ret, err := c.grpcClient.ReCall(ctx, func(client interface{}) (interface{}, error) {
+		if !funcutil.CheckCtxValid(ctx) {
+			return nil, ctx.Err()
+		}
+		return client.(rootcoordpb.RootCoordClient).ListImportTasks(ctx, req)
+	})
+	if err != nil || ret == nil {
+		return nil, err
+	}
+	return ret.(*milvuspb.ListImportTasksResponse), err
+}
+
 // Report impot task state to rootcoord
 func (c *Client) ReportImport(ctx context.Context, req *rootcoordpb.ImportResult) (*commonpb.Status, error) {
 	ret, err := c.grpcClient.ReCall(ctx, func(client interface{}) (interface{}, error) {

internal/distributed/rootcoord/client/client_test.go

@@ -310,6 +310,9 @@ func Test_NewClient(t *testing.T) {
 	r34Timeout, err := client.ListCredUsers(shortCtx, nil)
 	retCheck(r34Timeout, err)
 
+	r35Timeout, err := client.ListImportTasks(shortCtx, nil)
+	retCheck(r35Timeout, err)
+
 	// clean up
 	err = client.Stop()
 	assert.Nil(t, err)

internal/distributed/rootcoord/service.go

@@ -449,6 +449,11 @@ func (s *Server) GetImportState(ctx context.Context, in *milvuspb.GetImportState
 	return s.rootCoord.GetImportState(ctx, in)
 }
 
+// Returns id array of all import tasks
+func (s *Server) ListImportTasks(ctx context.Context, in *milvuspb.ListImportTasksRequest) (*milvuspb.ListImportTasksResponse, error) {
+	return s.rootCoord.ListImportTasks(ctx, in)
+}
+
 // Report impot task state to datacoord
 func (s *Server) ReportImport(ctx context.Context, in *rootcoordpb.ImportResult) (*commonpb.Status, error) {
 	return s.rootCoord.ReportImport(ctx, in)

internal/proto/milvus.proto

@@ -66,6 +66,7 @@ service MilvusService {
   // https://wiki.lfaidata.foundation/display/MIL/MEP+24+--+Support+bulk+load
   rpc Import(ImportRequest) returns (ImportResponse) {}
   rpc GetImportState(GetImportStateRequest) returns (GetImportStateResponse) {}
+  rpc ListImportTasks(ListImportTasksRequest) returns (ListImportTasksResponse) {}
 
   // https://wiki.lfaidata.foundation/display/MIL/MEP+27+--+Support+Basic+Authentication
   rpc CreateCredential(CreateCredentialRequest) returns (common.Status) {}
@@ -832,6 +833,14 @@ message GetImportStateResponse {
   repeated common.KeyValuePair infos = 5;  // more informations about the task, progress percent, file path, failed reason, etc.
 }
 
+message ListImportTasksRequest {
+}
+
+message ListImportTasksResponse {
+  common.Status status = 1;
+  repeated int64 tasks = 2;  // id list of all import tasks
+}
+
 message GetReplicasRequest {
   common.MsgBase base = 1;
   int64 collectionID = 2;