enhance: accesslog support fetch hybrid search expr and field (#40167) (#41921)

relate: #40166
pr: #40167

Signed-off-by: aoiasd <[email protected]>

Author: aoiasd

configs/milvus.yaml

@@ -303,7 +303,10 @@ proxy:
         format: "[$time_now] [ACCESS] <$user_name: $user_addr> $method_name [status: $method_status] [code: $error_code] [sdk: $sdk_version] [msg: $error_msg] [traceID: $trace_id] [timeCost: $time_cost]"
       query:
         format: "[$time_now] [ACCESS] <$user_name: $user_addr> $method_name [status: $method_status] [code: $error_code] [sdk: $sdk_version] [msg: $error_msg] [traceID: $trace_id] [timeCost: $time_cost] [database: $database_name] [collection: $collection_name] [partitions: $partition_name] [expr: $method_expr]"
-        methods: "Query,Search,Delete"
+        methods: "Query, Delete"
+      search:
+        format: "[$time_now] [ACCESS] <$user_name: $user_addr> $method_name [status: $method_status] [code: $error_code] [sdk: $sdk_version] [msg: $error_msg] [traceID: $trace_id] [timeCost: $time_cost] [database: $database_name] [collection: $collection_name] [partitions: $partition_name] [anns_field: $anns_field] [expr: $method_expr]"
+        methods: "HybridSearch, Search"
     cacheSize: 0 # Size of log of write cache, in byte. (Close write cache if size was 0)
     cacheFlushInterval: 3 # time interval of auto flush write cache, in seconds. (Close auto flush if interval was 0)
   connectionCheckIntervalSeconds: 120 # the interval time(in seconds) for connection manager to scan inactive client info

internal/proxy/accesslog/info/grpc_info.go

@@ -23,6 +23,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/samber/lo"
 	"go.opentelemetry.io/otel/trace"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
@@ -264,6 +265,10 @@ func (i *GrpcAccessInfo) Expression() string {
 		return expr.(string)
 	}
 
+	if req, ok := i.req.(*milvuspb.HybridSearchRequest); ok {
+		return listToString(lo.Map(req.GetRequests(), func(req *milvuspb.SearchRequest, _ int) string { return req.GetDsl() }))
+	}
+
 	dsl, ok := requestutil.GetDSLFromRequest(i.req)
 	if ok {
 		return dsl.(string)
@@ -299,3 +304,15 @@ func (i *GrpcAccessInfo) ConsistencyLevel() string {
 	}
 	return Unknown
 }
+
+func (i *GrpcAccessInfo) AnnsField() string {
+	if req, ok := i.req.(*milvuspb.SearchRequest); ok {
+		return getAnnsFieldFromKvs(req.GetSearchParams())
+	}
+
+	if req, ok := i.req.(*milvuspb.HybridSearchRequest); ok {
+		fields := lo.Map(req.GetRequests(), func(req *milvuspb.SearchRequest, _ int) string { return getAnnsFieldFromKvs(req.GetSearchParams()) })
+		return listToString(fields)
+	}
+	return Unknown
+}

internal/proxy/accesslog/info/info.go

@@ -47,6 +47,7 @@ var MetricFuncMap = map[string]getMetricFunc{
 	"$sdk_version":       getSdkVersion,
 	"$cluster_prefix":    getClusterPrefix,
 	"$consistency_level": getConsistencyLevel,
+	"$anns_field":        getAnnsField,
 }
 
 type AccessInfo interface {
@@ -64,6 +65,7 @@ type AccessInfo interface {
 	ErrorMsg() string
 	ErrorType() string
 	DbName() string
+	AnnsField() string
 	CollectionName() string
 	PartitionName() string
 	Expression() string
@@ -165,6 +167,10 @@ func getConsistencyLevel(i AccessInfo) string {
 	return i.ConsistencyLevel()
 }
 
+func getAnnsField(i AccessInfo) string {
+	return i.AnnsField()
+}
+
 func getClusterPrefix(i AccessInfo) string {
 	return ClusterPrefix.Load()
 }

internal/proxy/accesslog/info/restful_info.go

@@ -23,7 +23,9 @@ import (
 	"time"
 
 	"github.com/gin-gonic/gin"
+	"github.com/samber/lo"
 
+	"github.com/milvus-io/milvus-proto/go-api/v2/milvuspb"
 	"github.com/milvus-io/milvus/pkg/v2/util/requestutil"
 )
 
@@ -187,6 +189,10 @@ func (i *RestfulInfo) Expression() string {
 		return expr.(string)
 	}
 
+	if req, ok := i.req.(*milvuspb.HybridSearchRequest); ok {
+		return listToString(lo.Map(req.GetRequests(), func(req *milvuspb.SearchRequest, _ int) string { return req.GetDsl() }))
+	}
+
 	dsl, ok := requestutil.GetDSLFromRequest(i.req)
 	if ok {
 		return dsl.(string)
@@ -209,3 +215,14 @@ func (i *RestfulInfo) ConsistencyLevel() string {
 	}
 	return Unknown
 }
+
+func (i *RestfulInfo) AnnsField() string {
+	if req, ok := i.req.(*milvuspb.SearchRequest); ok {
+		return getAnnsFieldFromKvs(req.GetSearchParams())
+	}
+
+	if req, ok := i.req.(*milvuspb.HybridSearchRequest); ok {
+		return listToString(lo.Map(req.GetRequests(), func(req *milvuspb.SearchRequest, _ int) string { return getAnnsFieldFromKvs(req.GetSearchParams()) }))
+	}
+	return Unknown
+}

internal/proxy/accesslog/info/util.go

@@ -25,8 +25,10 @@ import (
 	"go.uber.org/atomic"
 	"google.golang.org/grpc/metadata"
 
+	"github.com/milvus-io/milvus-proto/go-api/v2/commonpb"
 	"github.com/milvus-io/milvus/pkg/v2/util"
 	"github.com/milvus-io/milvus/pkg/v2/util/crypto"
+	"github.com/milvus-io/milvus/pkg/v2/util/funcutil"
 )
 
 var ClusterPrefix atomic.String
@@ -90,3 +92,22 @@ func getSdkTypeByUserAgent(userAgents []string) (string, bool) {
 		return "", false
 	}
 }
+
+func getAnnsFieldFromKvs(kvs []*commonpb.KeyValuePair) string {
+	field, err := funcutil.GetAttrByKeyFromRepeatedKV("anns_field", kvs)
+	if err != nil {
+		return "default"
+	}
+	return field
+}
+
+func listToString(strs []string) string {
+	result := "["
+	for i, str := range strs {
+		if i != 0 {
+			result += ", "
+		}
+		result += "\"" + str + "\""
+	}
+	return result + "]"
+}