Merge pull request #797 from lihonghao314/dev/borwein-log

feat(sysadvisor): remove borwein v1 model

Author: xu282934741

cmd/katalyst-agent/app/options/sysadvisor/qosaware/model/borwein/borwein.go

@@ -26,7 +26,6 @@ import (
 )
 
 type BorweinOptions struct {
-	InferenceServiceSocketAbsPath      string
 	ModelNameToInferenceSvcSockAbsPath map[string]string
 	FeatureDescriptionFilePath         string
 	NodeFeatureNames                   []string
@@ -43,8 +42,6 @@ func NewBorweinOptions() *BorweinOptions {
 
 // AddFlags adds flags to the specified FlagSet.
 func (o *BorweinOptions) AddFlags(fs *pflag.FlagSet) {
-	fs.StringVar(&o.InferenceServiceSocketAbsPath, "borwein-inference-svc-socket-path", o.InferenceServiceSocketAbsPath,
-		"socket path which borwein inference server listens at. it's deprecated, use borwein-inference-model-to-svc-socket-path intead")
 	fs.StringToStringVar(&o.ModelNameToInferenceSvcSockAbsPath, "borwein-inference-model-to-svc-socket-path", o.ModelNameToInferenceSvcSockAbsPath,
 		"model name to socket path which its borwein inference server listens at")
 	fs.StringVar(&o.FeatureDescriptionFilePath, "feature-description-filepath", o.FeatureDescriptionFilePath,
@@ -63,7 +60,6 @@ func (o *BorweinOptions) ApplyTo(c *borwein.BorweinConfiguration) error {
 		ContainerFeatureNames []string `json:"container_feature_names"`
 	}{}
 
-	c.InferenceServiceSocketAbsPath = o.InferenceServiceSocketAbsPath
 	c.ModelNameToInferenceSvcSockAbsPath = o.ModelNameToInferenceSvcSockAbsPath
 
 	if len(o.NodeFeatureNames)+len(o.ContainerFeatureNames) > 0 {

pkg/agent/sysadvisor/metacache/metacache_test.go

@@ -27,6 +27,7 @@ import (
 	"k8s.io/kubernetes/pkg/kubelet/checkpointmanager"
 
 	borweinconsts "github.com/kubewharf/katalyst-core/pkg/agent/sysadvisor/plugin/inference/models/borwein/consts"
+	borweinutils "github.com/kubewharf/katalyst-core/pkg/agent/sysadvisor/plugin/inference/models/borwein/utils"
 	"github.com/kubewharf/katalyst-core/pkg/agent/sysadvisor/types"
 	"github.com/kubewharf/katalyst-core/pkg/metrics"
 )
@@ -53,11 +54,11 @@ func TestMetaCacheImp_GetFilteredInferenceResult(t *testing.T) {
 			fields: fields{
 				emitter: metrics.DummyMetrics{},
 				modelToResult: map[string]interface{}{
-					borweinconsts.ModelNameBorwein: []int{1, 2, 3},
+					borweinutils.GetInferenceResultKey(borweinconsts.ModelNameBorweinLatencyRegression): []int{1, 2, 3},
 				},
 			},
 			args: args{
-				modelName: borweinconsts.ModelNameBorwein,
+				modelName: borweinconsts.ModelNameBorweinLatencyRegression,
 			},
 			want:    []int{1, 2, 3},
 			wantErr: false,
@@ -67,7 +68,7 @@ func TestMetaCacheImp_GetFilteredInferenceResult(t *testing.T) {
 			fields: fields{
 				emitter: metrics.DummyMetrics{},
 				modelToResult: map[string]interface{}{
-					borweinconsts.ModelNameBorwein: []int{1, 2, 3},
+					borweinutils.GetInferenceResultKey(borweinconsts.ModelNameBorweinLatencyRegression): []int{1, 2, 3},
 				},
 			},
 			args: args{
@@ -88,7 +89,7 @@ func TestMetaCacheImp_GetFilteredInferenceResult(t *testing.T) {
 
 					return filteredResult, nil
 				},
-				modelName: borweinconsts.ModelNameBorwein,
+				modelName: borweinconsts.ModelNameBorweinLatencyRegression,
 			},
 			want:    []int{1, 2},
 			wantErr: false,
@@ -98,7 +99,7 @@ func TestMetaCacheImp_GetFilteredInferenceResult(t *testing.T) {
 			fields: fields{
 				emitter: metrics.DummyMetrics{},
 				modelToResult: map[string]interface{}{
-					borweinconsts.ModelNameBorwein: []string{"1", "2", "3"},
+					borweinutils.GetInferenceResultKey(borweinconsts.ModelNameBorweinLatencyRegression): []string{"1", "2", "3"},
 				},
 			},
 			args: args{
@@ -119,7 +120,7 @@ func TestMetaCacheImp_GetFilteredInferenceResult(t *testing.T) {
 
 					return filteredResult, nil
 				},
-				modelName: borweinconsts.ModelNameBorwein,
+				modelName: borweinconsts.ModelNameBorweinLatencyRegression,
 			},
 			want:    nil,
 			wantErr: true,
@@ -133,7 +134,7 @@ func TestMetaCacheImp_GetFilteredInferenceResult(t *testing.T) {
 				emitter:       tt.fields.emitter,
 				modelToResult: tt.fields.modelToResult,
 			}
-			got, err := mc.GetFilteredInferenceResult(tt.args.filterFunc, tt.args.modelName)
+			got, err := mc.GetFilteredInferenceResult(tt.args.filterFunc, borweinutils.GetInferenceResultKey(tt.args.modelName))
 			if (err != nil) != tt.wantErr {
 				t.Errorf("MetaCacheImp.GetFilteredInferenceResult() error = %v, wantErr %v", err, tt.wantErr)
 				return
@@ -166,11 +167,11 @@ func TestMetaCacheImp_GetInferenceResult(t *testing.T) {
 			fields: fields{
 				emitter: metrics.DummyMetrics{},
 				modelToResult: map[string]interface{}{
-					borweinconsts.ModelNameBorwein: []int{1, 2, 3},
+					borweinconsts.ModelNameBorweinLatencyRegression: []int{1, 2, 3},
 				},
 			},
 			args: args{
-				modelName: borweinconsts.ModelNameBorwein,
+				modelName: borweinconsts.ModelNameBorweinLatencyRegression,
 			},
 			want:    []int{1, 2, 3},
 			wantErr: false,
@@ -221,7 +222,7 @@ func TestMetaCacheImp_SetInferenceResult(t *testing.T) {
 				modelToResult: make(map[string]interface{}),
 			},
 			args: args{
-				modelName: borweinconsts.ModelNameBorwein,
+				modelName: borweinconsts.ModelNameBorweinLatencyRegression,
 				result:    []int{1, 2, 3},
 			},
 			want:    []int{1, 2, 3},

pkg/agent/sysadvisor/plugin/inference/modelresultfetcher/borwein/borwein.go

@@ -32,7 +32,6 @@ import (
 
 	"github.com/kubewharf/katalyst-core/pkg/agent/sysadvisor/metacache"
 	"github.com/kubewharf/katalyst-core/pkg/agent/sysadvisor/plugin/inference/modelresultfetcher"
-	borweinconsts "github.com/kubewharf/katalyst-core/pkg/agent/sysadvisor/plugin/inference/models/borwein/consts"
 	borweininfsvc "github.com/kubewharf/katalyst-core/pkg/agent/sysadvisor/plugin/inference/models/borwein/inferencesvc"
 	borweintypes "github.com/kubewharf/katalyst-core/pkg/agent/sysadvisor/plugin/inference/models/borwein/types"
 	borweinutils "github.com/kubewharf/katalyst-core/pkg/agent/sysadvisor/plugin/inference/models/borwein/utils"
@@ -62,14 +61,12 @@ type BorweinModelResultFetcher struct {
 	name      string
 	qosConfig *generic.QoSConfiguration
 
-	nodeFeatureNames                   []string // handled by GetNodeFeature
-	containerFeatureNames              []string // handled by GetContainerFeature
-	inferenceServiceSocketAbsPath      string
+	nodeFeatureNames                   []string          // handled by GetNodeFeature
+	containerFeatureNames              []string          // handled by GetContainerFeature
 	modelNameToInferenceSvcSockAbsPath map[string]string // map modelName to inference server sock path
 
 	emitter metrics.MetricEmitter
 
-	infSvcClient                  borweininfsvc.InferenceServiceClient
 	modelNameToInferenceSvcClient map[string]borweininfsvc.InferenceServiceClient // map modelName to its inference client
 	clientLock                    sync.RWMutex
 }
@@ -137,7 +134,7 @@ func (bmrf *BorweinModelResultFetcher) FetchModelResult(ctx context.Context, met
 	metaWriter metacache.MetaWriter, metaServer *metaserver.MetaServer,
 ) error {
 	bmrf.clientLock.RLock()
-	if bmrf.infSvcClient == nil && len(bmrf.modelNameToInferenceSvcClient) == 0 {
+	if len(bmrf.modelNameToInferenceSvcClient) == 0 {
 		bmrf.clientLock.RUnlock()
 		return fmt.Errorf("infSvcClient isn't initialized")
 	}
@@ -175,16 +172,7 @@ func (bmrf *BorweinModelResultFetcher) FetchModelResult(ctx context.Context, met
 		return fmt.Errorf("getInferenceRequestForPods failed with error: %v", err)
 	}
 
-	bmrf.clientLock.RLock()
-	var infSvcClients map[string]borweininfsvc.InferenceServiceClient
-	if len(bmrf.modelNameToInferenceSvcClient) > 0 {
-		infSvcClients = bmrf.modelNameToInferenceSvcClient
-	} else {
-		infSvcClients = map[string]borweininfsvc.InferenceServiceClient{
-			borweinconsts.ModelNameBorwein: bmrf.infSvcClient,
-		}
-	}
-	bmrf.clientLock.RUnlock()
+	infSvcClients := bmrf.modelNameToInferenceSvcClient
 
 	errCh := make(chan error, len(infSvcClients))
 	for modelName, client := range infSvcClients {
@@ -380,57 +368,44 @@ func (bmrf *BorweinModelResultFetcher) initInferenceSvcClientConn() (bool, error
 	// todo: emit metrics when initializing client connection failed
 
 	// never success
-	if bmrf.inferenceServiceSocketAbsPath == "" && len(bmrf.modelNameToInferenceSvcSockAbsPath) == 0 {
+	if len(bmrf.modelNameToInferenceSvcSockAbsPath) == 0 {
 		return false, fmt.Errorf("empty inference service socks information")
 	}
 
-	if len(bmrf.modelNameToInferenceSvcSockAbsPath) > 0 {
-		modelNameToConn := make(map[string]*grpc.ClientConn, len(bmrf.modelNameToInferenceSvcSockAbsPath))
-
-		allSuccess := true
-		for modelName, sockAbsPath := range bmrf.modelNameToInferenceSvcSockAbsPath {
-			infSvcConn, err := process.Dial(sockAbsPath, 5*time.Second)
-			if err != nil {
-				general.Errorf("get inference svc connection with socket: %s for model: %s failed with error",
-					sockAbsPath, modelName)
-				allSuccess = false
-				break
-			}
-			general.Infof("init inference svc connection with socket: %s for model: %s success", sockAbsPath, modelName)
-
-			modelNameToConn[modelName] = infSvcConn
-		}
+	modelNameToConn := make(map[string]*grpc.ClientConn, len(bmrf.modelNameToInferenceSvcSockAbsPath))
 
-		if !allSuccess {
-			for modelName, conn := range modelNameToConn {
-				err := conn.Close()
-				if err != nil {
-					general.Errorf("close connection for model: %s failed with error: %v",
-						modelName, err)
-				}
-			}
-		} else {
-			bmrf.clientLock.Lock()
-			bmrf.modelNameToInferenceSvcClient = make(map[string]borweininfsvc.InferenceServiceClient, len(modelNameToConn))
-			for modelName, conn := range modelNameToConn {
-				bmrf.modelNameToInferenceSvcClient[modelName] = borweininfsvc.NewInferenceServiceClient(conn)
-			}
-			bmrf.clientLock.Unlock()
+	allSuccess := true
+	for modelName, sockAbsPath := range bmrf.modelNameToInferenceSvcSockAbsPath {
+		infSvcConn, err := process.Dial(sockAbsPath, 5*time.Second)
+		if err != nil {
+			general.Errorf("get inference svc connection with socket: %s for model: %s failed with error",
+				sockAbsPath, modelName)
+			allSuccess = false
+			break
 		}
+		general.Infof("init inference svc connection with socket: %s for model: %s success", sockAbsPath, modelName)
 
-		return allSuccess, nil
+		modelNameToConn[modelName] = infSvcConn
 	}
 
-	infSvcConn, err := process.Dial(bmrf.inferenceServiceSocketAbsPath, 5*time.Second)
-	if err != nil {
-		general.Errorf("get inference svc connection with socket: %s failed with error: %v", bmrf.inferenceServiceSocketAbsPath, err)
-		return false, nil
+	if !allSuccess {
+		for modelName, conn := range modelNameToConn {
+			err := conn.Close()
+			if err != nil {
+				general.Errorf("close connection for model: %s failed with error: %v",
+					modelName, err)
+			}
+		}
+	} else {
+		bmrf.clientLock.Lock()
+		bmrf.modelNameToInferenceSvcClient = make(map[string]borweininfsvc.InferenceServiceClient, len(modelNameToConn))
+		for modelName, conn := range modelNameToConn {
+			bmrf.modelNameToInferenceSvcClient[modelName] = borweininfsvc.NewInferenceServiceClient(conn)
+		}
+		bmrf.clientLock.Unlock()
 	}
 
-	bmrf.clientLock.Lock()
-	bmrf.infSvcClient = borweininfsvc.NewInferenceServiceClient(infSvcConn)
-	bmrf.clientLock.Unlock()
-	return true, nil
+	return allSuccess, nil
 }
 
 func NewBorweinModelResultFetcher(fetcherName string, conf *config.Configuration, extraConf interface{},
@@ -455,11 +430,10 @@ func NewBorweinModelResultFetcher(fetcherName string, conf *config.Configuration
 		qosConfig:                          conf.QoSConfiguration,
 		nodeFeatureNames:                   conf.BorweinConfiguration.NodeFeatureNames,
 		containerFeatureNames:              conf.BorweinConfiguration.ContainerFeatureNames,
-		inferenceServiceSocketAbsPath:      conf.BorweinConfiguration.InferenceServiceSocketAbsPath,
 		modelNameToInferenceSvcSockAbsPath: conf.BorweinConfiguration.ModelNameToInferenceSvcSockAbsPath,
 	}
 
-	// fetcher initializing doesn't block sys-adviosr main process
+	// fetcher initializing doesn't block sys-advisor main process
 	go func() {
 		err := wait.PollImmediateInfinite(5*time.Second, bmrf.initInferenceSvcClientConn)
 		if err != nil {