Skip to content

Commit

Permalink
fix: link with install path's libblob-chunk-manager (milvus-io#29496)
Browse files Browse the repository at this point in the history 
issue: milvus-io#29494

1. link with install path's libblob-chunk-manager
2. performance of `ShouldBindWith` is better than `ShouldBindBodyWith`
3. the middleware shouldn't read the unrefreshed parameter repeatly

Signed-off-by: PowderLi <[email protected]>
  • Loading branch information
@PowderLi
PowderLi authored Dec 31, 2023
1 parent 3f46c6d commit 5f00bad
Show file tree
Hide file tree
Showing 8 changed files with 104 additions and 84 deletions.
1 change: 1 addition & 0 deletions internal/core/src/storage/CMakeLists.txt
View file
Original file line number Diff line number Diff line change
Expand Up @@ -63,6 +63,7 @@ add_library(milvus_storage SHARED ${STORAGE_FILES})
if (DEFINED AZURE_BUILD_DIR)
target_link_libraries(milvus_storage PUBLIC
"-L${AZURE_BUILD_DIR} -lblob-chunk-manager"
blob-chunk-manager
milvus_common
milvus-storage
pthread
Expand Down
2 changes: 1 addition & 1 deletion internal/core/src/storage/azure-blob-storage/test/CMakeLists.txt
View file
Original file line number Diff line number Diff line change
Expand Up @@ -16,4 +16,4 @@ project(azure-blob-test)

add_executable(azure-blob-test test_azure_blob_chunk_manager.cpp ../AzureBlobChunkManager.cpp)
find_package(GTest CONFIG REQUIRED)
target_link_libraries(azure-blob-test PRIVATE Azure::azure-identity Azure::azure-storage-blobs GTest::gtest)
target_link_libraries(azure-blob-test PRIVATE Azure::azure-identity Azure::azure-storage-blobs GTest::gtest blob-chunk-manager)
6 changes: 1 addition & 5 deletions internal/distributed/proxy/httpserver/handler.go
View file
Original file line number Diff line number Diff line change
@@ -1,7 +1,6 @@
package httpserver

import (
"context"
"fmt"

"github.com/gin-gonic/gin"
Expand All @@ -11,12 +10,9 @@ import (
"github.com/milvus-io/milvus/internal/types"
)

type RestRequestInterceptor func(ctx context.Context, ginCtx *gin.Context, req any, handler func(reqCtx context.Context, req any) (any, error)) (any, error)

// Handlers handles http requests
type Handlers struct {
proxy types.ProxyComponent
interceptors []RestRequestInterceptor
proxy types.ProxyComponent
}

// NewHandlers creates a new Handlers
Expand Down
140 changes: 78 additions & 62 deletions internal/distributed/proxy/httpserver/handler_v1.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@ import (
"github.com/milvus-io/milvus-proto/go-api/v2/milvuspb"
"github.com/milvus-io/milvus-proto/go-api/v2/schemapb"
"github.com/milvus-io/milvus/internal/proxy"
"github.com/milvus-io/milvus/internal/types"
"github.com/milvus-io/milvus/pkg/common"
"github.com/milvus-io/milvus/pkg/log"
"github.com/milvus-io/milvus/pkg/util/merr"
Expand All @@ -27,22 +28,69 @@ import (
var RestRequestInterceptorErr = errors.New("interceptor error placeholder")

func checkAuthorization(ctx context.Context, c *gin.Context, req interface{}) error {
if proxy.Params.CommonCfg.AuthorizationEnabled.GetAsBool() {
username, ok := c.Get(ContextUsername)
if !ok || username.(string) == "" {
c.JSON(http.StatusUnauthorized, gin.H{HTTPReturnCode: merr.Code(merr.ErrNeedAuthenticate), HTTPReturnMessage: merr.ErrNeedAuthenticate.Error()})
return RestRequestInterceptorErr
}
_, authErr := proxy.PrivilegeInterceptor(ctx, req)
if authErr != nil {
c.JSON(http.StatusForbidden, gin.H{HTTPReturnCode: merr.Code(authErr), HTTPReturnMessage: authErr.Error()})
return RestRequestInterceptorErr
}
username, ok := c.Get(ContextUsername)
if !ok || username.(string) == "" {
c.JSON(http.StatusUnauthorized, gin.H{HTTPReturnCode: merr.Code(merr.ErrNeedAuthenticate), HTTPReturnMessage: merr.ErrNeedAuthenticate.Error()})
return RestRequestInterceptorErr
}
_, authErr := proxy.PrivilegeInterceptor(ctx, req)
if authErr != nil {
c.JSON(http.StatusForbidden, gin.H{HTTPReturnCode: merr.Code(authErr), HTTPReturnMessage: authErr.Error()})
return RestRequestInterceptorErr
}

return nil
}

func (h *Handlers) checkDatabase(ctx context.Context, c *gin.Context, dbName string) error {
type RestRequestInterceptor func(ctx context.Context, ginCtx *gin.Context, req any, handler func(reqCtx context.Context, req any) (any, error)) (any, error)

// HandlersV1 handles http requests
type HandlersV1 struct {
proxy types.ProxyComponent
interceptors []RestRequestInterceptor
}

// NewHandlers creates a new HandlersV1
func NewHandlersV1(proxyComponent types.ProxyComponent) *HandlersV1 {
h := &HandlersV1{
proxy: proxyComponent,
interceptors: []RestRequestInterceptor{},
}
if proxy.Params.CommonCfg.AuthorizationEnabled.GetAsBool() {
h.interceptors = append(h.interceptors,
// authorization
func(ctx context.Context, ginCtx *gin.Context, req any, handler func(reqCtx context.Context, req any) (any, error)) (any, error) {
err := checkAuthorization(ctx, ginCtx, req)
if err != nil {
return nil, err
}
return handler(ctx, req)
})
}
h.interceptors = append(h.interceptors,
// check database
func(ctx context.Context, ginCtx *gin.Context, req any, handler func(reqCtx context.Context, req any) (any, error)) (any, error) {
value, ok := requestutil.GetDbNameFromRequest(req)
if !ok {
return handler(ctx, req)
}
err := h.checkDatabase(ctx, ginCtx, value.(string))
if err != nil {
return nil, err
}
return handler(ctx, req)
})
h.interceptors = append(h.interceptors,
// trace request
func(ctx context.Context, ginCtx *gin.Context, req any, handler func(reqCtx context.Context, req any) (any, error)) (any, error) {
return proxy.TraceLogInterceptor(ctx, req, &grpc.UnaryServerInfo{
FullMethod: ginCtx.Request.URL.Path,
}, handler)
})
return h
}

func (h *HandlersV1) checkDatabase(ctx context.Context, c *gin.Context, dbName string) error {
if dbName == DefaultDbName {
return nil
}
Expand All @@ -69,7 +117,7 @@ func (h *Handlers) checkDatabase(ctx context.Context, c *gin.Context, dbName str
return RestRequestInterceptorErr
}

func (h *Handlers) describeCollection(ctx context.Context, c *gin.Context, dbName string, collectionName string) (*schemapb.CollectionSchema, error) {
func (h *HandlersV1) describeCollection(ctx context.Context, c *gin.Context, dbName string, collectionName string) (*schemapb.CollectionSchema, error) {
collSchema, err := proxy.GetCachedCollectionSchema(ctx, dbName, collectionName)
if err == nil {
return collSchema, nil
Expand All @@ -94,7 +142,7 @@ func (h *Handlers) describeCollection(ctx context.Context, c *gin.Context, dbNam
return response.Schema, nil
}

func (h *Handlers) hasCollection(ctx context.Context, c *gin.Context, dbName string, collectionName string) (bool, error) {
func (h *HandlersV1) hasCollection(ctx context.Context, c *gin.Context, dbName string, collectionName string) (bool, error) {
req := milvuspb.HasCollectionRequest{
DbName: dbName,
CollectionName: collectionName,
Expand All @@ -110,8 +158,7 @@ func (h *Handlers) hasCollection(ctx context.Context, c *gin.Context, dbName str
return response.Value, nil
}

func (h *Handlers) RegisterRoutesToV1(router gin.IRouter) {
h.registerRestRequestInterceptor()
func (h *HandlersV1) RegisterRoutesToV1(router gin.IRouter) {
router.GET(VectorCollectionsPath, h.listCollections)
router.POST(VectorCollectionsCreatePath, h.createCollection)
router.GET(VectorCollectionsDescribePath, h.getCollectionDetails)
Expand All @@ -124,38 +171,7 @@ func (h *Handlers) RegisterRoutesToV1(router gin.IRouter) {
router.POST(VectorSearchPath, h.search)
}

func (h *Handlers) registerRestRequestInterceptor() {
h.interceptors = []RestRequestInterceptor{
// authorization
func(ctx context.Context, ginCtx *gin.Context, req any, handler func(reqCtx context.Context, req any) (any, error)) (any, error) {
err := checkAuthorization(ctx, ginCtx, req)
if err != nil {
return nil, err
}
return handler(ctx, req)
},
// check database
func(ctx context.Context, ginCtx *gin.Context, req any, handler func(reqCtx context.Context, req any) (any, error)) (any, error) {
value, ok := requestutil.GetDbNameFromRequest(req)
if !ok {
return handler(ctx, req)
}
err := h.checkDatabase(ctx, ginCtx, value.(string))
if err != nil {
return nil, err
}
return handler(ctx, req)
},
// trace request
func(ctx context.Context, ginCtx *gin.Context, req any, handler func(reqCtx context.Context, req any) (any, error)) (any, error) {
return proxy.TraceLogInterceptor(ctx, req, &grpc.UnaryServerInfo{
FullMethod: ginCtx.Request.URL.Path,
}, handler)
},
}
}

func (h *Handlers) executeRestRequestInterceptor(ctx context.Context,
func (h *HandlersV1) executeRestRequestInterceptor(ctx context.Context,
ginCtx *gin.Context,
req any, handler func(reqCtx context.Context, req any) (any, error),
) (any, error) {
Expand All @@ -170,7 +186,7 @@ func (h *Handlers) executeRestRequestInterceptor(ctx context.Context,
return f(ctx, req)
}

func (h *Handlers) listCollections(c *gin.Context) {
func (h *HandlersV1) listCollections(c *gin.Context) {
dbName := c.DefaultQuery(HTTPDbName, DefaultDbName)
req := &milvuspb.ShowCollectionsRequest{
DbName: dbName,
Expand Down Expand Up @@ -201,14 +217,14 @@ func (h *Handlers) listCollections(c *gin.Context) {
c.JSON(http.StatusOK, gin.H{HTTPReturnCode: http.StatusOK, HTTPReturnData: collections})
}

func (h *Handlers) createCollection(c *gin.Context) {
func (h *HandlersV1) createCollection(c *gin.Context) {
httpReq := CreateCollectionReq{
DbName: DefaultDbName,
MetricType: DefaultMetricType,
PrimaryField: DefaultPrimaryFieldName,
VectorField: DefaultVectorFieldName,
}
if err := c.ShouldBindBodyWith(&httpReq, binding.JSON); err != nil {
if err := c.ShouldBindWith(&httpReq, binding.JSON); err != nil {
log.Warn("high level restful api, the parameter of create collection is incorrect", zap.Any("request", httpReq), zap.Error(err))
c.AbortWithStatusJSON(http.StatusOK, gin.H{
HTTPReturnCode: merr.Code(merr.ErrIncorrectParameterFormat),
Expand Down Expand Up @@ -310,7 +326,7 @@ func (h *Handlers) createCollection(c *gin.Context) {
c.JSON(http.StatusOK, gin.H{HTTPReturnCode: http.StatusOK, HTTPReturnData: gin.H{}})
}

func (h *Handlers) getCollectionDetails(c *gin.Context) {
func (h *HandlersV1) getCollectionDetails(c *gin.Context) {
collectionName := c.Query(HTTPCollectionName)
if collectionName == "" {
log.Warn("high level restful api, desc collection require parameter: [collectionName], but miss")
Expand Down Expand Up @@ -400,11 +416,11 @@ func (h *Handlers) getCollectionDetails(c *gin.Context) {
}})
}

func (h *Handlers) dropCollection(c *gin.Context) {
func (h *HandlersV1) dropCollection(c *gin.Context) {
httpReq := DropCollectionReq{
DbName: DefaultDbName,
}
if err := c.ShouldBindBodyWith(&httpReq, binding.JSON); err != nil {
if err := c.ShouldBindWith(&httpReq, binding.JSON); err != nil {
log.Warn("high level restful api, the parameter of drop collection is incorrect", zap.Any("request", httpReq), zap.Error(err))
c.AbortWithStatusJSON(http.StatusOK, gin.H{
HTTPReturnCode: merr.Code(merr.ErrIncorrectParameterFormat),
Expand Down Expand Up @@ -453,13 +469,13 @@ func (h *Handlers) dropCollection(c *gin.Context) {
}
}

func (h *Handlers) query(c *gin.Context) {
func (h *HandlersV1) query(c *gin.Context) {
httpReq := QueryReq{
DbName: DefaultDbName,
Limit: 100,
OutputFields: []string{DefaultOutputFields},
}
if err := c.ShouldBindBodyWith(&httpReq, binding.JSON); err != nil {
if err := c.ShouldBindWith(&httpReq, binding.JSON); err != nil {
log.Warn("high level restful api, the parameter of query is incorrect", zap.Any("request", httpReq), zap.Error(err))
c.AbortWithStatusJSON(http.StatusOK, gin.H{
HTTPReturnCode: merr.Code(merr.ErrIncorrectParameterFormat),
Expand Down Expand Up @@ -518,7 +534,7 @@ func (h *Handlers) query(c *gin.Context) {
}
}

func (h *Handlers) get(c *gin.Context) {
func (h *HandlersV1) get(c *gin.Context) {
httpReq := GetReq{
DbName: DefaultDbName,
OutputFields: []string{DefaultOutputFields},
Expand Down Expand Up @@ -589,7 +605,7 @@ func (h *Handlers) get(c *gin.Context) {
}
}

func (h *Handlers) delete(c *gin.Context) {
func (h *HandlersV1) delete(c *gin.Context) {
httpReq := DeleteReq{
DbName: DefaultDbName,
}
Expand Down Expand Up @@ -649,7 +665,7 @@ func (h *Handlers) delete(c *gin.Context) {
}
}

func (h *Handlers) insert(c *gin.Context) {
func (h *HandlersV1) insert(c *gin.Context) {
httpReq := InsertReq{
DbName: DefaultDbName,
}
Expand Down Expand Up @@ -741,7 +757,7 @@ func (h *Handlers) insert(c *gin.Context) {
}
}

func (h *Handlers) upsert(c *gin.Context) {
func (h *HandlersV1) upsert(c *gin.Context) {
httpReq := UpsertReq{
DbName: DefaultDbName,
}
Expand Down Expand Up @@ -838,12 +854,12 @@ func (h *Handlers) upsert(c *gin.Context) {
}
}

func (h *Handlers) search(c *gin.Context) {
func (h *HandlersV1) search(c *gin.Context) {
httpReq := SearchReq{
DbName: DefaultDbName,
Limit: 100,
}
if err := c.ShouldBindBodyWith(&httpReq, binding.JSON); err != nil {
if err := c.ShouldBindWith(&httpReq, binding.JSON); err != nil {
log.Warn("high level restful api, the parameter of search is incorrect", zap.Any("request", httpReq), zap.Error(err))
c.AbortWithStatusJSON(http.StatusOK, gin.H{
HTTPReturnCode: merr.Code(merr.ErrIncorrectParameterFormat),
Expand Down
6 changes: 3 additions & 3 deletions internal/distributed/proxy/httpserver/handler_v1_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -86,7 +86,7 @@ func versional(path string) string {
}

func initHTTPServer(proxy types.ProxyComponent, needAuth bool) *gin.Engine {
h := NewHandlers(proxy)
h := NewHandlersV1(proxy)
ginHandler := gin.Default()
ginHandler.Use(func(c *gin.Context) {
_, err := strconv.ParseBool(c.Request.Header.Get(HTTPHeaderAllowInt64))
Expand All @@ -101,7 +101,7 @@ func initHTTPServer(proxy types.ProxyComponent, needAuth bool) *gin.Engine {
c.Next()
})
app := ginHandler.Group(URIPrefixV1, genAuthMiddleWare(needAuth))
NewHandlers(h.proxy).RegisterRoutesToV1(app)
NewHandlersV1(h.proxy).RegisterRoutesToV1(app)
return ginHandler
}

Expand Down Expand Up @@ -1763,7 +1763,7 @@ func wrapWithDescribeIndex(t *testing.T, mp *mocks.MockProxy, returnType int, ti
}

func TestInterceptor(t *testing.T) {
h := Handlers{}
h := HandlersV1{}
v := atomic.NewInt32(0)
h.interceptors = []RestRequestInterceptor{
func(ctx context.Context, ginCtx *gin.Context, req any, handler func(reqCtx context.Context, req any) (any, error)) (any, error) {
Expand Down
26 changes: 14 additions & 12 deletions internal/distributed/proxy/service.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -125,10 +125,6 @@ func NewServer(ctx context.Context, factory dependency.Factory) (*Server, error)
}

func authenticate(c *gin.Context) {
c.Set(httpserver.ContextUsername, "")
if !proxy.Params.CommonCfg.AuthorizationEnabled.GetAsBool() {
return
}
username, password, ok := httpserver.ParseUsernamePassword(c)
if ok {
if proxy.PasswordVerify(c, username, password) {
Expand Down Expand Up @@ -178,15 +174,15 @@ func (s *Server) startHTTPServer(errChan chan error) {
SkipPaths: proxy.Params.ProxyCfg.GinLogSkipPaths.GetAsStrings(),
})
ginHandler.Use(ginLogger, gin.Recovery())
httpHeaderAllowInt64 := "false"
httpParams := &paramtable.Get().HTTPCfg
if httpParams.AcceptTypeAllowInt64.GetAsBool() {
httpHeaderAllowInt64 = "true"
}
ginHandler.Use(func(c *gin.Context) {
_, err := strconv.ParseBool(c.Request.Header.Get(httpserver.HTTPHeaderAllowInt64))
if err != nil {
httpParams := &paramtable.Get().HTTPCfg
if httpParams.AcceptTypeAllowInt64.GetAsBool() {
c.Request.Header.Set(httpserver.HTTPHeaderAllowInt64, "true")
} else {
c.Request.Header.Set(httpserver.HTTPHeaderAllowInt64, "false")
}
c.Request.Header.Set(httpserver.HTTPHeaderAllowInt64, httpHeaderAllowInt64)
}
c.Writer.Header().Set("Access-Control-Allow-Origin", "*")
c.Writer.Header().Set("Access-Control-Allow-Credentials", "true")
Expand All @@ -197,9 +193,15 @@ func (s *Server) startHTTPServer(errChan chan error) {
return
}
c.Next()
}, authenticate)
})
ginHandler.Use(func(c *gin.Context) {
c.Set(httpserver.ContextUsername, "")
})
if proxy.Params.CommonCfg.AuthorizationEnabled.GetAsBool() {
ginHandler.Use(authenticate)
}
app := ginHandler.Group("/v1")
httpserver.NewHandlers(s.proxy).RegisterRoutesToV1(app)
httpserver.NewHandlersV1(s.proxy).RegisterRoutesToV1(app)
s.httpServer = &http.Server{Handler: ginHandler, ReadHeaderTimeout: time.Second}
errChan <- nil
if err := s.httpServer.Serve(s.httpListener); err != nil && err != cmux.ErrServerClosed {
Expand Down
2 changes: 1 addition & 1 deletion scripts/azure_build.sh
View file
Original file line number Diff line number Diff line change
Expand Up @@ -50,4 +50,4 @@ fi
echo ${AZURE_CMAKE_CMD}
${AZURE_CMAKE_CMD}

make & make install
make install
Loading

0 comments on commit 5f00bad

Please sign in to comment.