enhance: [2.5] Make accesslog.$consistency_level represent actual value used (#44708)

Cherry-pick from master
pr: #44706 
Related to #44703

This PR:
- Add `SetActualConsistencyLevel` to `info.AccessInfo` interface and
  related util method processing it
- Make `$consistency_level` returning actual value if set

---------

Signed-off-by: Congqi Xia <congqi.xia@zilliz.com>

internal/proxy/accesslog/info/grpc_info.go

@@ -49,6 +49,9 @@ type GrpcAccessInfo struct {
 	grpcInfo *grpc.UnaryServerInfo
 	start    time.Time
 	end      time.Time
+
+	// runtime set info
+	actualConsistencyLevel *commonpb.ConsistencyLevel
 }
 
 func NewGrpcAccessInfo(ctx context.Context, grpcInfo *grpc.UnaryServerInfo, req interface{}) *GrpcAccessInfo {
@@ -298,6 +301,10 @@ func (i *GrpcAccessInfo) OutputFields() string {
 }
 
 func (i *GrpcAccessInfo) ConsistencyLevel() string {
+	// return actual consistency level if set
+	if i.actualConsistencyLevel != nil {
+		return i.actualConsistencyLevel.String()
+	}
 	level, ok := requestutil.GetConsistencyLevelFromRequst(i.req)
 	if ok {
 		return level.String()
@@ -345,3 +352,7 @@ func (i *GrpcAccessInfo) QueryParams() string {
 	}
 	return Unknown
 }
+
+func (i *GrpcAccessInfo) SetActualConsistencyLevel(acl commonpb.ConsistencyLevel) {
+	i.actualConsistencyLevel = &acl
+}

internal/proxy/accesslog/info/info.go

@@ -16,6 +16,8 @@
 
 package info
 
+import "github.com/milvus-io/milvus-proto/go-api/v2/commonpb"
+
 const (
 	Unknown            = "Unknown"
 	timeFormat         = "2006/01/02 15:04:05.000 -07:00"
@@ -78,6 +80,7 @@ type AccessInfo interface {
 	NQ() string
 	SearchParams() string
 	QueryParams() string
+	SetActualConsistencyLevel(commonpb.ConsistencyLevel)
 }
 
 func Get(i AccessInfo, keys ...string) []any {

internal/proxy/accesslog/info/restful_info.go

@@ -25,6 +25,7 @@ import (
 	"github.com/gin-gonic/gin"
 	"github.com/samber/lo"
 
+	"github.com/milvus-io/milvus-proto/go-api/v2/commonpb"
 	"github.com/milvus-io/milvus-proto/go-api/v2/milvuspb"
 	"github.com/milvus-io/milvus/pkg/v2/util/requestutil"
 )
@@ -41,6 +42,9 @@ type RestfulInfo struct {
 	params *gin.LogFormatterParams
 	start  time.Time
 	req    interface{}
+
+	// runtime set info
+	actualConsistencyLevel *commonpb.ConsistencyLevel
 }
 
 func NewRestfulInfo() *RestfulInfo {
@@ -209,6 +213,10 @@ func (i *RestfulInfo) OutputFields() string {
 }
 
 func (i *RestfulInfo) ConsistencyLevel() string {
+	// return actual consistency level if set
+	if i.actualConsistencyLevel != nil {
+		return i.actualConsistencyLevel.String()
+	}
 	level, ok := requestutil.GetConsistencyLevelFromRequst(i.req)
 	if ok {
 		return level.String()
@@ -255,3 +263,7 @@ func (i *RestfulInfo) QueryParams() string {
 	}
 	return Unknown
 }
+
+func (i *RestfulInfo) SetActualConsistencyLevel(acl commonpb.ConsistencyLevel) {
+	i.actualConsistencyLevel = &acl
+}

internal/proxy/accesslog/util.go

@@ -25,6 +25,7 @@ import (
 	"github.com/gin-gonic/gin"
 	"google.golang.org/grpc"
 
+	"github.com/milvus-io/milvus-proto/go-api/v2/commonpb"
 	"github.com/milvus-io/milvus/internal/proxy/accesslog/info"
 )
 
@@ -82,3 +83,13 @@ func timeFromName(filename, prefix, ext string) (time.Time, error) {
 	ts := filename[len(prefix) : len(filename)-len(ext)]
 	return time.Parse(timeNameFormat, ts)
 }
+
+func SetActualConsistencyLevel(ctx context.Context, acl commonpb.ConsistencyLevel) {
+	if ctx != nil {
+		v := ctx.Value(AccessKey{})
+		info, ok := v.(info.AccessInfo)
+		if ok && info != nil {
+			info.SetActualConsistencyLevel(acl)
+		}
+	}
+}

internal/proxy/task_query.go

@@ -16,6 +16,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/parser/planparserv2"
+	"github.com/milvus-io/milvus/internal/proxy/accesslog"
 	"github.com/milvus-io/milvus/internal/types"
 	"github.com/milvus-io/milvus/internal/util/exprutil"
 	"github.com/milvus-io/milvus/internal/util/reduce"
@@ -475,6 +476,10 @@ func (t *queryTask) PreExecute(ctx context.Context) error {
 			guaranteeTs = parseGuaranteeTsFromConsistency(guaranteeTs, t.BeginTs(), consistencyLevel)
 		}
 	}
+
+	// update actual consistency level
+	accesslog.SetActualConsistencyLevel(ctx, consistencyLevel)
+
 	t.GuaranteeTimestamp = guaranteeTs
 	// need modify mvccTs and guaranteeTs for iterator specially
 	if t.queryParams.isIterator && t.request.GetGuaranteeTimestamp() > 0 {

internal/proxy/task_search.go

@@ -19,6 +19,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/parser/planparserv2"
+	"github.com/milvus-io/milvus/internal/proxy/accesslog"
 	"github.com/milvus-io/milvus/internal/types"
 	"github.com/milvus-io/milvus/internal/util/exprutil"
 	"github.com/milvus-io/milvus/internal/util/reduce"
@@ -258,6 +259,10 @@ func (t *searchTask) PreExecute(ctx context.Context) error {
 			guaranteeTs = parseGuaranteeTsFromConsistency(guaranteeTs, t.BeginTs(), consistencyLevel)
 		}
 	}
+
+	// update actual consistency level
+	accesslog.SetActualConsistencyLevel(ctx, consistencyLevel)
+
 	t.SearchRequest.GuaranteeTimestamp = guaranteeTs
 	t.SearchRequest.ConsistencyLevel = consistencyLevel
 	if t.isIterator && t.request.GetGuaranteeTimestamp() > 0 {