milvus-io
/
milvus
mirror of https://github.com/milvus-io/milvus.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Implements remaining rest APIs (#16387) 

Signed-off-by: shaoyue.chen <shaoyue.chen@zilliz.com>
pull/16747/head
shaoyue 2022-04-29 15:15:47 +08:00 committed by GitHub
parent 415a0e5978
commit 2f49db9d3c
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 618 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
configs/milvus.yaml
View File

@ -111,6 +111,7 @@ proxy:
http:
enabled: true # Whether to enable the http server
debug_mode: false # Whether to enable http server debug mode
port: 8080 # Whether to enable the http server
readTimeout: 30000 # 30000 ms http read timeout
writeTimeout: 30000 # 30000 ms http write timeout

340
internal/distributed/proxy/httpserver/handler.go
View File

@ -23,7 +23,34 @@ func NewHandlers(proxy types.ProxyComponent) *Handlers {
// RegisterRouters registers routes to given router
func (h *Handlers) RegisterRoutesTo(router gin.IRouter) {
router.GET("/health", wrapHandler(h.handleGetHealth))
router.POST("/dummy", wrapHandler(h.handlePostDummy))
router.POST("/dummy", wrapHandler(h.handleDummy))
router.POST("/collection", wrapHandler(h.handleCreateCollection))
router.DELETE("/collection", wrapHandler(h.handleDropCollection))
router.GET("/collection/existence", wrapHandler(h.handleHasCollection))
router.GET("/collection", wrapHandler(h.handleDescribeCollection))
router.POST("/collection/load", wrapHandler(h.handleLoadCollection))
router.DELETE("/collection/load", wrapHandler(h.handleReleaseCollection))
router.GET("/collection/statistics", wrapHandler(h.handleGetCollectionStatistics))
router.GET("/collections", wrapHandler(h.handleShowCollections))
router.POST("/partition", wrapHandler(h.handleCreatePartition))
router.DELETE("/partition", wrapHandler(h.handleDropPartition))
router.GET("/partition/existence", wrapHandler(h.handleHasPartition))
router.POST("/partitions/load", wrapHandler(h.handleLoadPartitions))
router.DELETE("/partitions/load", wrapHandler(h.handleReleasePartitions))
router.GET("/partition/statistics", wrapHandler(h.handleGetPartitionStatistics))
router.GET("/partitions", wrapHandler(h.handleShowPartitions))
router.POST("/alias", wrapHandler(h.handleCreateAlias))
router.DELETE("/alias", wrapHandler(h.handleDropAlias))
router.PATCH("/alias", wrapHandler(h.handleAlterAlias))
router.POST("/index", wrapHandler(h.handleCreateIndex))
router.GET("/index", wrapHandler(h.handleDescribeIndex))
router.GET("/index/state", wrapHandler(h.handleGetIndexState))
router.GET("/index/progress", wrapHandler(h.handleGetIndexBuildProgress))
router.DELETE("/index", wrapHandler(h.handleDropIndex))
router.POST("/entities", wrapHandler(h.handleInsert))
router.DELETE("/entities", wrapHandler(h.handleDelete))
@ -32,13 +59,25 @@ func (h *Handlers) RegisterRoutesTo(router gin.IRouter) {
router.POST("/persist", wrapHandler(h.handleFlush))
router.GET("/distance", wrapHandler(h.handleCalcDistance))
router.GET("/persist/state", wrapHandler(h.handleGetFlushState))
router.GET("/persist/segment-info", wrapHandler(h.handleGetPersistentSegmentInfo))
router.GET("/query-segment-info", wrapHandler(h.handleGetQuerySegmentInfo))
router.GET("/metrics", wrapHandler(h.handleGetMetrics))
router.POST("/load-balance", wrapHandler(h.handleLoadBalance))
router.GET("/compaction/state", wrapHandler(h.handleGetCompactionState))
router.GET("/compaction/plans", wrapHandler(h.handleGetCompactionStateWithPlans))
router.POST("/compaction", wrapHandler(h.handleManualCompaction))
router.POST("/import", wrapHandler(h.handleImport))
router.GET("/import/state", wrapHandler(h.handleGetImportState))
}
func (h *Handlers) handleGetHealth(c *gin.Context) (interface{}, error) {
return gin.H{"status": "ok"}, nil
}
func (h *Handlers) handlePostDummy(c *gin.Context) (interface{}, error) {
func (h *Handlers) handleDummy(c *gin.Context) (interface{}, error) {
req := milvuspb.DummyRequest{}
// use ShouldBind to supports binding JSON, XML, YAML, and protobuf.
err := shouldBind(c, &req)
@ -48,6 +87,213 @@ func (h *Handlers) handlePostDummy(c *gin.Context) (interface{}, error) {
return h.proxy.Dummy(c, &req)
}
func (h *Handlers) handleCreateCollection(c *gin.Context) (interface{}, error) {
req := milvuspb.CreateCollectionRequest{}
err := shouldBind(c, &req)
if err != nil {
return nil, fmt.Errorf("%w: parse body failed: %v", errBadRequest, err)
}
return h.proxy.CreateCollection(c, &req)
}
func (h *Handlers) handleDropCollection(c *gin.Context) (interface{}, error) {
req := milvuspb.DropCollectionRequest{}
err := shouldBind(c, &req)
if err != nil {
return nil, fmt.Errorf("%w: parse body failed: %v", errBadRequest, err)
}
return h.proxy.DropCollection(c, &req)
}
func (h *Handlers) handleHasCollection(c *gin.Context) (interface{}, error) {
req := milvuspb.HasCollectionRequest{}
err := shouldBind(c, &req)
if err != nil {
return nil, fmt.Errorf("%w: parse body failed: %v", errBadRequest, err)
}
return h.proxy.HasCollection(c, &req)
}
func (h *Handlers) handleDescribeCollection(c *gin.Context) (interface{}, error) {
req := milvuspb.DescribeCollectionRequest{}
err := shouldBind(c, &req)
if err != nil {
return nil, fmt.Errorf("%w: parse body failed: %v", errBadRequest, err)
}
return h.proxy.DescribeCollection(c, &req)
}
func (h *Handlers) handleLoadCollection(c *gin.Context) (interface{}, error) {
req := milvuspb.LoadCollectionRequest{}
err := shouldBind(c, &req)
if err != nil {
return nil, fmt.Errorf("%w: parse body failed: %v", errBadRequest, err)
}
return h.proxy.LoadCollection(c, &req)
}
func (h *Handlers) handleReleaseCollection(c *gin.Context) (interface{}, error) {
req := milvuspb.ReleaseCollectionRequest{}
err := shouldBind(c, &req)
if err != nil {
return nil, fmt.Errorf("%w: parse body failed: %v", errBadRequest, err)
}
return h.proxy.ReleaseCollection(c, &req)
}
func (h *Handlers) handleGetCollectionStatistics(c *gin.Context) (interface{}, error) {
req := milvuspb.GetCollectionStatisticsRequest{}
err := shouldBind(c, &req)
if err != nil {
return nil, fmt.Errorf("%w: parse body failed: %v", errBadRequest, err)
}
return h.proxy.GetCollectionStatistics(c, &req)
}
func (h *Handlers) handleShowCollections(c *gin.Context) (interface{}, error) {
req := milvuspb.ShowCollectionsRequest{}
err := shouldBind(c, &req)
if err != nil {
return nil, fmt.Errorf("%w: parse body failed: %v", errBadRequest, err)
}
return h.proxy.ShowCollections(c, &req)
}
func (h *Handlers) handleCreatePartition(c *gin.Context) (interface{}, error) {
req := milvuspb.CreatePartitionRequest{}
err := shouldBind(c, &req)
if err != nil {
return nil, fmt.Errorf("%w: parse body failed: %v", errBadRequest, err)
}
return h.proxy.CreatePartition(c, &req)
}
func (h *Handlers) handleDropPartition(c *gin.Context) (interface{}, error) {
req := milvuspb.DropPartitionRequest{}
err := shouldBind(c, &req)
if err != nil {
return nil, fmt.Errorf("%w: parse body failed: %v", errBadRequest, err)
}
return h.proxy.DropPartition(c, &req)
}
func (h *Handlers) handleHasPartition(c *gin.Context) (interface{}, error) {
req := milvuspb.HasPartitionRequest{}
err := shouldBind(c, &req)
if err != nil {
return nil, fmt.Errorf("%w: parse body failed: %v", errBadRequest, err)
}
return h.proxy.HasPartition(c, &req)
}
func (h *Handlers) handleLoadPartitions(c *gin.Context) (interface{}, error) {
req := milvuspb.LoadPartitionsRequest{}
err := shouldBind(c, &req)
if err != nil {
return nil, fmt.Errorf("%w: parse body failed: %v", errBadRequest, err)
}
return h.proxy.LoadPartitions(c, &req)
}
func (h *Handlers) handleReleasePartitions(c *gin.Context) (interface{}, error) {
req := milvuspb.ReleasePartitionsRequest{}
err := shouldBind(c, &req)
if err != nil {
return nil, fmt.Errorf("%w: parse body failed: %v", errBadRequest, err)
}
return h.proxy.ReleasePartitions(c, &req)
}
func (h *Handlers) handleGetPartitionStatistics(c *gin.Context) (interface{}, error) {
req := milvuspb.GetPartitionStatisticsRequest{}
err := shouldBind(c, &req)
if err != nil {
return nil, fmt.Errorf("%w: parse body failed: %v", errBadRequest, err)
}
return h.proxy.GetPartitionStatistics(c, &req)
}
func (h *Handlers) handleShowPartitions(c *gin.Context) (interface{}, error) {
req := milvuspb.ShowPartitionsRequest{}
err := shouldBind(c, &req)
if err != nil {
return nil, fmt.Errorf("%w: parse body failed: %v", errBadRequest, err)
}
return h.proxy.ShowPartitions(c, &req)
}
func (h *Handlers) handleCreateAlias(c *gin.Context) (interface{}, error) {
req := milvuspb.CreateAliasRequest{}
err := shouldBind(c, &req)
if err != nil {
return nil, fmt.Errorf("%w: parse body failed: %v", errBadRequest, err)
}
return h.proxy.CreateAlias(c, &req)
}
func (h *Handlers) handleDropAlias(c *gin.Context) (interface{}, error) {
req := milvuspb.DropAliasRequest{}
err := shouldBind(c, &req)
if err != nil {
return nil, fmt.Errorf("%w: parse body failed: %v", errBadRequest, err)
}
return h.proxy.DropAlias(c, &req)
}
func (h *Handlers) handleAlterAlias(c *gin.Context) (interface{}, error) {
req := milvuspb.AlterAliasRequest{}
err := shouldBind(c, &req)
if err != nil {
return nil, fmt.Errorf("%w: parse body failed: %v", errBadRequest, err)
}
return h.proxy.AlterAlias(c, &req)
}
func (h *Handlers) handleCreateIndex(c *gin.Context) (interface{}, error) {
req := milvuspb.CreateIndexRequest{}
err := shouldBind(c, &req)
if err != nil {
return nil, fmt.Errorf("%w: parse body failed: %v", errBadRequest, err)
}
return h.proxy.CreateIndex(c, &req)
}
func (h *Handlers) handleDescribeIndex(c *gin.Context) (interface{}, error) {
req := milvuspb.DescribeIndexRequest{}
err := shouldBind(c, &req)
if err != nil {
return nil, fmt.Errorf("%w: parse body failed: %v", errBadRequest, err)
}
return h.proxy.DescribeIndex(c, &req)
}
func (h *Handlers) handleGetIndexState(c *gin.Context) (interface{}, error) {
req := milvuspb.GetIndexStateRequest{}
err := shouldBind(c, &req)
if err != nil {
return nil, fmt.Errorf("%w: parse body failed: %v", errBadRequest, err)
}
return h.proxy.GetIndexState(c, &req)
}
func (h *Handlers) handleGetIndexBuildProgress(c *gin.Context) (interface{}, error) {
req := milvuspb.GetIndexBuildProgressRequest{}
err := shouldBind(c, &req)
if err != nil {
return nil, fmt.Errorf("%w: parse body failed: %v", errBadRequest, err)
}
return h.proxy.GetIndexBuildProgress(c, &req)
}
func (h *Handlers) handleDropIndex(c *gin.Context) (interface{}, error) {
req := milvuspb.DropIndexRequest{}
err := shouldBind(c, &req)
if err != nil {
return nil, fmt.Errorf("%w: parse body failed: %v", errBadRequest, err)
}
return h.proxy.DropIndex(c, &req)
}
func (h *Handlers) handleInsert(c *gin.Context) (interface{}, error) {
req := milvuspb.InsertRequest{}
err := shouldBind(c, &req)
@ -101,3 +347,93 @@ func (h *Handlers) handleCalcDistance(c *gin.Context) (interface{}, error) {
}
return h.proxy.CalcDistance(c, &req)
}
func (h *Handlers) handleGetFlushState(c *gin.Context) (interface{}, error) {
req := milvuspb.GetFlushStateRequest{}
err := shouldBind(c, &req)
if err != nil {
return nil, fmt.Errorf("%w: parse body failed: %v", errBadRequest, err)
}
return h.proxy.GetFlushState(c, &req)
}
func (h *Handlers) handleGetPersistentSegmentInfo(c *gin.Context) (interface{}, error) {
req := milvuspb.GetPersistentSegmentInfoRequest{}
err := shouldBind(c, &req)
if err != nil {
return nil, fmt.Errorf("%w: parse body failed: %v", errBadRequest, err)
}
return h.proxy.GetPersistentSegmentInfo(c, &req)
}
func (h *Handlers) handleGetQuerySegmentInfo(c *gin.Context) (interface{}, error) {
req := milvuspb.GetQuerySegmentInfoRequest{}
err := shouldBind(c, &req)
if err != nil {
return nil, fmt.Errorf("%w: parse body failed: %v", errBadRequest, err)
}
return h.proxy.GetQuerySegmentInfo(c, &req)
}
func (h *Handlers) handleGetMetrics(c *gin.Context) (interface{}, error) {
req := milvuspb.GetMetricsRequest{}
err := shouldBind(c, &req)
if err != nil {
return nil, fmt.Errorf("%w: parse body failed: %v", errBadRequest, err)
}
return h.proxy.GetMetrics(c, &req)
}
func (h *Handlers) handleLoadBalance(c *gin.Context) (interface{}, error) {
req := milvuspb.LoadBalanceRequest{}
err := shouldBind(c, &req)
if err != nil {
return nil, fmt.Errorf("%w: parse body failed: %v", errBadRequest, err)
}
return h.proxy.LoadBalance(c, &req)
}
func (h *Handlers) handleGetCompactionState(c *gin.Context) (interface{}, error) {
req := milvuspb.GetCompactionStateRequest{}
err := shouldBind(c, &req)
if err != nil {
return nil, fmt.Errorf("%w: parse body failed: %v", errBadRequest, err)
}
return h.proxy.GetCompactionState(c, &req)
}
func (h *Handlers) handleGetCompactionStateWithPlans(c *gin.Context) (interface{}, error) {
req := milvuspb.GetCompactionPlansRequest{}
err := shouldBind(c, &req)
if err != nil {
return nil, fmt.Errorf("%w: parse body failed: %v", errBadRequest, err)
}
return h.proxy.GetCompactionStateWithPlans(c, &req)
}
func (h *Handlers) handleManualCompaction(c *gin.Context) (interface{}, error) {
req := milvuspb.ManualCompactionRequest{}
err := shouldBind(c, &req)
if err != nil {
return nil, fmt.Errorf("%w: parse body failed: %v", errBadRequest, err)
}
return h.proxy.ManualCompaction(c, &req)
}
func (h *Handlers) handleImport(c *gin.Context) (interface{}, error) {
req := milvuspb.ImportRequest{}
err := shouldBind(c, &req)
if err != nil {
return nil, fmt.Errorf("%w: parse body failed: %v", errBadRequest, err)
}
return h.proxy.Import(c, &req)
}
func (h *Handlers) handleGetImportState(c *gin.Context) (interface{}, error) {
req := milvuspb.GetImportStateRequest{}
err := shouldBind(c, &req)
if err != nil {
return nil, fmt.Errorf("%w: parse body failed: %v", errBadRequest, err)
}
return h.proxy.GetImportState(c, &req)
}

292
internal/distributed/proxy/httpserver/handler_test.go
View File

@ -30,6 +30,99 @@ func (mockProxyComponent) Dummy(ctx context.Context, request *milvuspb.DummyRequ
return nil, nil
}
var emptyBody = &gin.H{}
var testStatus = &commonpb.Status{Reason: "ok"}
func (mockProxyComponent) CreateCollection(ctx context.Context, request *milvuspb.CreateCollectionRequest) (*commonpb.Status, error) {
return testStatus, nil
}
func (mockProxyComponent) DropCollection(ctx context.Context, request *milvuspb.DropCollectionRequest) (*commonpb.Status, error) {
return testStatus, nil
}
func (mockProxyComponent) HasCollection(ctx context.Context, request *milvuspb.HasCollectionRequest) (*milvuspb.BoolResponse, error) {
return &milvuspb.BoolResponse{Status: testStatus}, nil
}
func (mockProxyComponent) LoadCollection(ctx context.Context, request *milvuspb.LoadCollectionRequest) (*commonpb.Status, error) {
return testStatus, nil
}
func (mockProxyComponent) ReleaseCollection(ctx context.Context, request *milvuspb.ReleaseCollectionRequest) (*commonpb.Status, error) {
return testStatus, nil
}
func (mockProxyComponent) DescribeCollection(ctx context.Context, request *milvuspb.DescribeCollectionRequest) (*milvuspb.DescribeCollectionResponse, error) {
return &milvuspb.DescribeCollectionResponse{Status: testStatus}, nil
}
func (mockProxyComponent) GetCollectionStatistics(ctx context.Context, request *milvuspb.GetCollectionStatisticsRequest) (*milvuspb.GetCollectionStatisticsResponse, error) {
return &milvuspb.GetCollectionStatisticsResponse{Status: testStatus}, nil
}
func (mockProxyComponent) ShowCollections(ctx context.Context, request *milvuspb.ShowCollectionsRequest) (*milvuspb.ShowCollectionsResponse, error) {
return &milvuspb.ShowCollectionsResponse{Status: testStatus}, nil
}
func (mockProxyComponent) CreatePartition(ctx context.Context, request *milvuspb.CreatePartitionRequest) (*commonpb.Status, error) {
return testStatus, nil
}
func (mockProxyComponent) DropPartition(ctx context.Context, request *milvuspb.DropPartitionRequest) (*commonpb.Status, error) {
return testStatus, nil
}
func (mockProxyComponent) HasPartition(ctx context.Context, request *milvuspb.HasPartitionRequest) (*milvuspb.BoolResponse, error) {
return &milvuspb.BoolResponse{Status: testStatus}, nil
}
func (mockProxyComponent) LoadPartitions(ctx context.Context, request *milvuspb.LoadPartitionsRequest) (*commonpb.Status, error) {
return testStatus, nil
}
func (mockProxyComponent) ReleasePartitions(ctx context.Context, request *milvuspb.ReleasePartitionsRequest) (*commonpb.Status, error) {
return testStatus, nil
}
func (mockProxyComponent) GetPartitionStatistics(ctx context.Context, request *milvuspb.GetPartitionStatisticsRequest) (*milvuspb.GetPartitionStatisticsResponse, error) {
return &milvuspb.GetPartitionStatisticsResponse{Status: testStatus}, nil
}
func (mockProxyComponent) ShowPartitions(ctx context.Context, request *milvuspb.ShowPartitionsRequest) (*milvuspb.ShowPartitionsResponse, error) {
return &milvuspb.ShowPartitionsResponse{Status: testStatus}, nil
}
func (mockProxyComponent) CreateAlias(ctx context.Context, request *milvuspb.CreateAliasRequest) (*commonpb.Status, error) {
return testStatus, nil
}
func (mockProxyComponent) DropAlias(ctx context.Context, request *milvuspb.DropAliasRequest) (*commonpb.Status, error) {
return testStatus, nil
}
func (mockProxyComponent) AlterAlias(ctx context.Context, request *milvuspb.AlterAliasRequest) (*commonpb.Status, error) {
return testStatus, nil
}
func (mockProxyComponent) CreateIndex(ctx context.Context, request *milvuspb.CreateIndexRequest) (*commonpb.Status, error) {
return testStatus, nil
}
func (mockProxyComponent) DescribeIndex(ctx context.Context, request *milvuspb.DescribeIndexRequest) (*milvuspb.DescribeIndexResponse, error) {
return &milvuspb.DescribeIndexResponse{Status: testStatus}, nil
}
func (mockProxyComponent) GetIndexState(ctx context.Context, request *milvuspb.GetIndexStateRequest) (*milvuspb.GetIndexStateResponse, error) {
return &milvuspb.GetIndexStateResponse{Status: testStatus}, nil
}
func (mockProxyComponent) GetIndexBuildProgress(ctx context.Context, request *milvuspb.GetIndexBuildProgressRequest) (*milvuspb.GetIndexBuildProgressResponse, error) {
return &milvuspb.GetIndexBuildProgressResponse{Status: testStatus}, nil
}
func (mockProxyComponent) DropIndex(ctx context.Context, request *milvuspb.DropIndexRequest) (*commonpb.Status, error) {
return testStatus, nil
}
func (mockProxyComponent) Insert(ctx context.Context, request *milvuspb.InsertRequest) (*milvuspb.MutationResult, error) {
if request.CollectionName == "" {
return nil, errors.New("body parse err")
@ -94,6 +187,46 @@ func (mockProxyComponent) CalcDistance(ctx context.Context, request *milvuspb.Ca
return &calcDistanceResult, nil
}
func (mockProxyComponent) GetFlushState(ctx context.Context, request *milvuspb.GetFlushStateRequest) (*milvuspb.GetFlushStateResponse, error) {
return &milvuspb.GetFlushStateResponse{Status: testStatus}, nil
}
func (mockProxyComponent) GetPersistentSegmentInfo(ctx context.Context, request *milvuspb.GetPersistentSegmentInfoRequest) (*milvuspb.GetPersistentSegmentInfoResponse, error) {
return &milvuspb.GetPersistentSegmentInfoResponse{Status: testStatus}, nil
}
func (mockProxyComponent) GetQuerySegmentInfo(ctx context.Context, request *milvuspb.GetQuerySegmentInfoRequest) (*milvuspb.GetQuerySegmentInfoResponse, error) {
return &milvuspb.GetQuerySegmentInfoResponse{Status: testStatus}, nil
}
func (mockProxyComponent) GetMetrics(ctx context.Context, request *milvuspb.GetMetricsRequest) (*milvuspb.GetMetricsResponse, error) {
return &milvuspb.GetMetricsResponse{Status: testStatus}, nil
}
func (mockProxyComponent) LoadBalance(ctx context.Context, request *milvuspb.LoadBalanceRequest) (*commonpb.Status, error) {
return testStatus, nil
}
func (mockProxyComponent) GetCompactionState(ctx context.Context, request *milvuspb.GetCompactionStateRequest) (*milvuspb.GetCompactionStateResponse, error) {
return &milvuspb.GetCompactionStateResponse{Status: testStatus}, nil
}
func (mockProxyComponent) GetCompactionStateWithPlans(ctx context.Context, request *milvuspb.GetCompactionPlansRequest) (*milvuspb.GetCompactionPlansResponse, error) {
return &milvuspb.GetCompactionPlansResponse{Status: testStatus}, nil
}
func (mockProxyComponent) ManualCompaction(ctx context.Context, request *milvuspb.ManualCompactionRequest) (*milvuspb.ManualCompactionResponse, error) {
return &milvuspb.ManualCompactionResponse{Status: testStatus}, nil
}
func (mockProxyComponent) Import(ctx context.Context, request *milvuspb.ImportRequest) (*milvuspb.ImportResponse, error) {
return &milvuspb.ImportResponse{Status: testStatus}, nil
}
func (mockProxyComponent) GetImportState(ctx context.Context, request *milvuspb.GetImportStateRequest) (*milvuspb.GetImportStateResponse, error) {
return &milvuspb.GetImportStateResponse{Status: testStatus}, nil
}
func TestHandlers(t *testing.T) {
mockProxy := &mockProxyComponent{}
h := NewHandlers(mockProxy)
@ -164,47 +297,118 @@ func TestHandlers(t *testing.T) {
expectedBody interface{}
}
testCases := []testCase{
{
http.MethodPost, "/collection", emptyBody,
http.StatusOK, testStatus,
},
{
http.MethodDelete, "/collection", emptyBody,
http.StatusOK, &testStatus,
},
{
http.MethodGet, "/collection/existence", emptyBody,
http.StatusOK, &milvuspb.BoolResponse{Status: testStatus},
},
{
http.MethodGet, "/collection", emptyBody,
http.StatusOK, &milvuspb.DescribeCollectionResponse{Status: testStatus},
},
{
http.MethodPost, "/collection/load", emptyBody,
http.StatusOK, &testStatus,
},
{
http.MethodDelete, "/collection/load", emptyBody,
http.StatusOK, &testStatus,
},
{
http.MethodGet, "/collection/statistics", emptyBody,
http.StatusOK, &milvuspb.GetCollectionStatisticsResponse{Status: testStatus},
},
{
http.MethodGet, "/collections", emptyBody,
http.StatusOK, &milvuspb.ShowCollectionsResponse{Status: testStatus},
},
{
http.MethodPost, "/partition", emptyBody,
http.StatusOK, testStatus,
},
{
http.MethodDelete, "/partition", emptyBody,
http.StatusOK, testStatus,
},
{
http.MethodGet, "/partition/existence", emptyBody,
http.StatusOK, &milvuspb.BoolResponse{Status: testStatus},
},
{
http.MethodPost, "/partitions/load", emptyBody,
http.StatusOK, &testStatus,
},
{
http.MethodDelete, "/partitions/load", emptyBody,
http.StatusOK, &testStatus,
},
{
http.MethodGet, "/partition/statistics", emptyBody,
http.StatusOK, milvuspb.GetPartitionStatisticsResponse{Status: testStatus},
},
{
http.MethodGet, "/partitions", emptyBody,
http.StatusOK, &milvuspb.ShowPartitionsResponse{Status: testStatus},
},
{
http.MethodPost, "/alias", emptyBody,
http.StatusOK, testStatus,
},
{
http.MethodDelete, "/alias", emptyBody,
http.StatusOK, testStatus,
},
{
http.MethodPatch, "/alias", emptyBody,
http.StatusOK, testStatus,
},
{
http.MethodPost, "/index", emptyBody,
http.StatusOK, testStatus,
},
{
http.MethodGet, "/index", emptyBody,
http.StatusOK, &milvuspb.DescribeIndexResponse{Status: testStatus},
},
{
http.MethodGet, "/index/state", emptyBody,
http.StatusOK, &milvuspb.GetIndexStateResponse{Status: testStatus},
},
{
http.MethodGet, "/index/progress", emptyBody,
http.StatusOK, &milvuspb.GetIndexBuildProgressResponse{Status: testStatus},
},
{
http.MethodDelete, "/index", emptyBody,
http.StatusOK, testStatus,
},
{
http.MethodPost, "/entities", &milvuspb.InsertRequest{CollectionName: "c1"},
http.StatusOK, &milvuspb.MutationResult{Acknowledged: true},
},
{
http.MethodPost, "/entities", []byte("bad request"),
http.StatusBadRequest, nil,
},
{
http.MethodDelete, "/entities", milvuspb.DeleteRequest{Expr: "some expr"},
http.StatusOK, &milvuspb.MutationResult{Acknowledged: true},
},
{
http.MethodDelete, "/entities", []byte("bad request"),
http.StatusBadRequest, nil,
},
{
http.MethodPost, "/search", milvuspb.SearchRequest{Dsl: "some dsl"},
http.StatusOK, &searchResult,
},
{
http.MethodPost, "/search", []byte("bad request"),
http.StatusBadRequest, nil,
},
{
http.MethodPost, "/query", milvuspb.QueryRequest{Expr: "some expr"},
http.StatusOK, &queryResult,
},
{
http.MethodPost, "/query", []byte("bad request"),
http.StatusBadRequest, nil,
},
{
http.MethodPost, "/persist", milvuspb.FlushRequest{CollectionNames: []string{"c1"}},
http.StatusOK, flushResult,
},
{
http.MethodPost, "/persist", []byte("bad request"),
http.StatusBadRequest, nil,
},
{
http.MethodGet, "/distance", milvuspb.CalcDistanceRequest{
Params: []*commonpb.KeyValuePair{
@ -213,8 +417,44 @@ func TestHandlers(t *testing.T) {
http.StatusOK, calcDistanceResult,
},
{
http.MethodGet, "/distance", []byte("bad request"),
http.StatusBadRequest, nil,
http.MethodGet, "/persist/state", emptyBody,
http.StatusOK, &milvuspb.GetFlushStateResponse{Status: testStatus},
},
{
http.MethodGet, "/persist/segment-info", emptyBody,
http.StatusOK, &milvuspb.GetPersistentSegmentInfoResponse{Status: testStatus},
},
{
http.MethodGet, "/query-segment-info", emptyBody,
http.StatusOK, &milvuspb.GetQuerySegmentInfoResponse{Status: testStatus},
},
{
http.MethodGet, "/metrics", emptyBody,
http.StatusOK, &milvuspb.GetMetricsResponse{Status: testStatus},
},
{
http.MethodPost, "/load-balance", emptyBody,
http.StatusOK, testStatus,
},
{
http.MethodGet, "/compaction/state", emptyBody,
http.StatusOK, &milvuspb.GetCompactionStateResponse{Status: testStatus},
},
{
http.MethodGet, "/compaction/plans", emptyBody,
http.StatusOK, &milvuspb.GetCompactionPlansResponse{Status: testStatus},
},
{
http.MethodPost, "/compaction", emptyBody,
http.StatusOK, &milvuspb.ManualCompactionResponse{Status: testStatus},
},
{
http.MethodPost, "/import", emptyBody,
http.StatusOK, &milvuspb.ImportResponse{Status: testStatus},
},
{
http.MethodGet, "/import/state", emptyBody,
http.StatusOK, &milvuspb.GetImportStateResponse{Status: testStatus},
},
}
for _, tt := range testCases {
@ -226,12 +466,18 @@ func TestHandlers(t *testing.T) {
req := httptest.NewRequest(tt.httpMethod, tt.path, bytes.NewReader(body))
w := httptest.NewRecorder()
testEngine.ServeHTTP(w, req)
assert.Equal(t, tt.expectedStatus, w.Code)
if tt.expectedBody != nil {
bodyBytes, err := json.Marshal(tt.expectedBody)
assert.NoError(t, err)
assert.Equal(t, bodyBytes, w.Body.Bytes())
}
// test marshal failed
req = httptest.NewRequest(tt.httpMethod, tt.path, bytes.NewReader([]byte("bad request")))
w = httptest.NewRecorder()
testEngine.ServeHTTP(w, req)
assert.Equal(t, http.StatusBadRequest, w.Code)
})
}
}

3
internal/distributed/proxy/service.go
View File

@ -118,6 +118,9 @@ func (s *Server) startHTTPServer(port int) {
gin.DefaultWriter = io.Discard
gin.DefaultErrorWriter = io.Discard
}
if !HTTPParams.DebugMode {
gin.SetMode(gin.ReleaseMode)
}
ginHandler := gin.Default()
apiv1 := ginHandler.Group("/api/v1")
httpserver.NewHandlers(s.proxy).RegisterRoutesTo(apiv1)

6
internal/util/paramtable/http_param.go
View File

@ -10,6 +10,7 @@ type HTTPConfig struct {
once sync.Once
Enabled bool
DebugMode bool
Port int
ReadTimeout time.Duration
WriteTimeout time.Duration
@ -26,6 +27,7 @@ func (p *HTTPConfig) init() {
p.BaseTable.Init()
p.initHTTPEnabled()
p.initHTTPDebugMode()
p.initHTTPPort()
p.initHTTPReadTimeout()
p.initHTTPWriteTimeout()
@ -35,6 +37,10 @@ func (p *HTTPConfig) initHTTPEnabled() {
p.Enabled = p.ParseBool("proxy.http.enabled", true)
}
func (p *HTTPConfig) initHTTPDebugMode() {
p.DebugMode = p.ParseBool("proxy.http.debug_mode", false)
}
func (p *HTTPConfig) initHTTPPort() {
p.Port = p.ParseIntWithDefault("proxy.http.port", 8080)
}

1
internal/util/paramtable/http_param_test.go
View File

@ -11,6 +11,7 @@ func TestHTTPConfig_Init(t *testing.T) {
cf := new(HTTPConfig)
cf.InitOnce()
assert.Equal(t, cf.Enabled, true)
assert.Equal(t, cf.DebugMode, false)
assert.Equal(t, cf.Port, 8080)
assert.Equal(t, cf.ReadTimeout, time.Second*30)
assert.Equal(t, cf.WriteTimeout, time.Second*30)