feat: implement OpenVINO Model Server (OVMS) adapter (#18)

#### Motivation

Adds another adapter server implementation to interface between the ModelMesh API and the OpenVINO Model Server (OVMS). From [the OVMS docs](https://docs.openvino.ai/latest/ovms_what_is_openvino_model_server.html):

> OpenVINO Model Server (OVMS) is a high-performance system for serving machine learning models. It is based on C++ for high scalability and optimized for Intel solutions, so that you can take advantage of all the power of the Intel® Xeon® processor or Intel’s AI accelerators and expose it over a network interface. OVMS uses the same architecture and API as [TensorFlow Serving](https://github.com/tensorflow/serving), while applying OpenVINO for inference execution. Inference service is provided via gRPC or REST API, making it easy to deploy new algorithms and AI experiments.

The adapter will be used to create a new OVMS `ServingRuntime` in KServe to enable use of this optimized runtime. OVMS supports running models in [OpenVINO's Intermediate Representation (IR) format](https://docs.openvino.ai/latest/openvino_docs_MO_DG_IR_and_opsets.html), as well as ONNX models.

#### Modifications

The model layout adaptations are very similar to Triton, but the multi-model management is significantly different. OVMS implements a subset of the TFS API, including multi-model management via a declarative configuration file. Reload of the configuration can be triggered with a call to OVMS's [`/v1/config/reload` REST API](https://github.com/openvinotoolkit/model_server/blob/main/docs/model_server_rest_api.md#config-reload). This is very different than the MLServer and Triton adapters. The approach used in this PR is to create a "ModelManager" that accepts a queue of load/unload requests, batches the changes to the config file, and manages the reload calls to OVMS.

- adds the `model-mesh-ovms-adapter` directory with packages implementing the first pass at an adapter server for the OVMS runtime
- small refactor to all adapters to use a single implementation of CalcMemCapcity from the utils package

NB: Processing of the model schema is not included in this PR.

Signed-off-by: Travis Johnson <[email protected]>
Signed-off-by: Joe Runde <[email protected]>
Signed-off-by: Nick Hill <[email protected]>

Co-authored-by: Joe Runde <[email protected]>
Co-authored-by: Nick Hill <[email protected]>

Author: 3 people

Dockerfile

@@ -85,6 +85,7 @@ COPY . ./
 RUN go build -o puller model-serving-puller/main.go
 RUN go build -o triton-adapter model-mesh-triton-adapter/main.go
 RUN go build -o mlserver-adapter model-mesh-mlserver-adapter/main.go
+RUN go build -o ovms-adapter model-mesh-ovms-adapter/main.go
 
 
 ###############################################################################
@@ -119,6 +120,7 @@ COPY --from=build /opt/app/puller /opt/app/
 COPY --from=build /opt/app/triton-adapter /opt/app/
 COPY --from=build /opt/app/mlserver-adapter /opt/app/
 COPY --from=build /opt/app/model-mesh-triton-adapter/scripts/tf_pb.py /opt/scripts/
+COPY --from=build /opt/app/ovms-adapter /opt/app/
 
 # Don't define an entrypoint. This is a multi-purpose image so the user should specify which binary they want to run (e.g. /opt/app/puller or /opt/app/triton-adapter)
 # ENTRYPOINT ["/opt/app/puller"]

internal/envconfig/envconfig.go

@@ -16,6 +16,7 @@ package envconfig
 import (
 	"os"
 	"strconv"
+	"time"
 
 	"github.com/go-logr/logr"
 )
@@ -71,3 +72,18 @@ func GetEnvBool(key string, defaultValue bool, log logr.Logger) bool {
 	}
 	return defaultValue
 }
+
+// Returns the duration value of environment variable "key" or a default value
+// Note if the environment variable cannot be parsed as a duration, including an
+// empty string, this will fail and exit.
+func GetEnvDuration(key string, defaultValue time.Duration, log logr.Logger) time.Duration {
+	if strVal, found := os.LookupEnv(key); found {
+		val, err := time.ParseDuration(strVal)
+		if err != nil {
+			log.Error(err, "Environment variable must be a duration", "env_var", key, "value", strVal)
+			os.Exit(1)
+		}
+		return val
+	}
+	return defaultValue
+}

internal/util/loadmodel.go

@@ -22,8 +22,9 @@ import (
 )
 
 const (
-	modelTypeJSONKey  string = "model_type"
-	schemaPathJSONKey string = "schema_path"
+	modelTypeJSONKey     string = "model_type"
+	schemaPathJSONKey    string = "schema_path"
+	diskSizeBytesJSONKey string = "disk_size_bytes"
 )
 
 // GetModelType first tries to read the type from the LoadModelRequest.ModelKey json
@@ -68,3 +69,27 @@ func GetSchemaPath(req *mmesh.LoadModelRequest) (string, error) {
 
 	return schemaPath, nil
 }
+
+func CalcMemCapacity(reqModelKey string, defaultSize int, multiplier float64, log logr.Logger) uint64 {
+	// Try to calculate the model size from the disk size passed in the LoadModelRequest.ModelKey
+	// but first set the default to fall back on if we cannot get the disk size.
+	size := uint64(defaultSize)
+	var modelKey map[string]interface{}
+	err := json.Unmarshal([]byte(reqModelKey), &modelKey)
+	if err != nil {
+		log.Info("'SizeInBytes' will be defaulted as LoadModelRequest.ModelKey value is not valid JSON", "SizeInBytes", size, "model_key", reqModelKey, "error", err)
+	} else {
+		if modelKey[diskSizeBytesJSONKey] != nil {
+			diskSize, ok := modelKey[diskSizeBytesJSONKey].(float64)
+			if ok {
+				size = uint64(diskSize * multiplier)
+				log.Info("Setting 'SizeInBytes' to a multiple of model disk size", "SizeInBytes", size, "disk_size", diskSize, "multiplier", multiplier)
+			} else {
+				log.Info("'SizeInBytes' will be defaulted as LoadModelRequest.ModelKey 'disk_size_bytes' value is not a number", "SizeInBytes", size, "model_key", modelKey)
+			}
+		} else {
+			log.Info("'SizeInBytes' will be defaulted as LoadModelRequest.ModelKey did not contain a value for 'disk_size_bytes'", "SizeInBytes", size, "model_key", modelKey)
+		}
+	}
+	return size
+}

model-mesh-mlserver-adapter/server/server.go

@@ -41,7 +41,6 @@ import (
 
 const (
 	mlserverServiceName              string = "inference.GRPCInferenceService"
-	diskSizeBytesJSONKey             string = "disk_size_bytes"
 	mlserverModelSubdir              string = "_mlserver_models"
 	mlserverRepositoryConfigFilename string = "model-settings.json"
 )
@@ -147,7 +146,7 @@ func (s *MLServerAdapterServer) LoadModel(ctx context.Context, req *mmesh.LoadMo
 		return nil, status.Errorf(status.Code(mlserverErr), "Failed to load Model due to MLServer runtime error: %s", mlserverErr)
 	}
 
-	size := calcMemCapacity(req.ModelKey, s.AdapterConfig, log)
+	size := util.CalcMemCapacity(req.ModelKey, s.AdapterConfig.DefaultModelSizeInBytes, s.AdapterConfig.ModelSizeMultiplier, log)
 
 	log.Info("MLServer model loaded")
 
@@ -426,31 +425,6 @@ func tensorMetadataToJson(tm modelschema.TensorMetadata) map[string]interface{}
 	return json
 }
 
-func calcMemCapacity(reqModelKey string, adapterConfig *AdapterConfiguration, log logr.Logger) uint64 {
-	// Try to calculate the model size from the disk size passed in the LoadModelRequest.ModelKey
-	// but first set the default to fall back on if we cannot get the disk size.
-	size := uint64(adapterConfig.DefaultModelSizeInBytes)
-	var modelKey map[string]interface{}
-	err := json.Unmarshal([]byte(reqModelKey), &modelKey)
-	if err != nil {
-		log.Info("'SizeInBytes' will be defaulted as LoadModelRequest.ModelKey value is not valid JSON", "SizeInBytes", size, "LoadModelRequest.ModelKey", reqModelKey, "Error", err)
-	} else {
-		if modelKey[diskSizeBytesJSONKey] != nil {
-			diskSize, ok := modelKey[diskSizeBytesJSONKey].(float64)
-			if ok {
-				size = uint64(diskSize * adapterConfig.ModelSizeMultiplier)
-				log.Info("Setting 'SizeInBytes' to multiples of model disk size", "SizeInBytes", size, "ModelSizeMultiplier", adapterConfig.ModelSizeMultiplier)
-			} else {
-				log.Info("'SizeInBytes' will be defaulted as LoadModelRequest.ModelKey value is not a number", "SizeInBytes", size, "disc size bytes json key", diskSizeBytesJSONKey,
-					"Model key", modelKey[diskSizeBytesJSONKey])
-			}
-		} else {
-			log.Info("'SizeInBytes' will be defaulted as LoadModelRequest.ModelKey did not contain a value", "SizeInBytes", size, "diskSizeBytesJSONKey", diskSizeBytesJSONKey)
-		}
-	}
-	return size
-}
-
 func (s *MLServerAdapterServer) UnloadModel(ctx context.Context, req *mmesh.UnloadModelRequest) (*mmesh.UnloadModelResponse, error) {
 	_, mlserverErr := s.ModelRepoClient.RepositoryModelUnload(ctx, &modelrepo.RepositoryModelUnloadRequest{
 		ModelName: req.ModelId,

model-mesh-ovms-adapter/.gitignore

@@ -0,0 +1,5 @@
+# binaries created by test
+main
+
+server/testdata/model_config_list.json
+server/testdata/generated

model-mesh-ovms-adapter/Makefile

@@ -0,0 +1,21 @@
+# Copyright 2022 IBM Corporation
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+test:
+	./scripts/run_tests.sh
+
+fmt:
+	cd .. && make fmt
+
+# Remove $(MAKECMDGOALS) if you don't intend make to just be a taskrunner
+.PHONY: $(MAKECMDGOALS)

model-mesh-ovms-adapter/README.md

@@ -0,0 +1,5 @@
+# Model Mesh OpenVINO Model Server (OVMS)
+
+This is an adapter which implements the internal model-mesh model management API for [OpenVINO Model Server](https://github.com/openvinotoolkit/model_server).
+
+This adapter is different than the other adapters because OpenVINO is modeled after TensorflowServing and does not implement the KFSv2 API. OVMS also does not have a direct model management API; its multimodel support is implemented with an on-disk config file that can be reloaded with a REST API call.

model-mesh-ovms-adapter/main.go

@@ -0,0 +1,58 @@
+// Copyright 2022 IBM Corporation
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+package main
+
+import (
+	"fmt"
+	"net"
+	"os"
+
+	"github.com/kserve/modelmesh-runtime-adapter/internal/proto/mmesh"
+	"github.com/kserve/modelmesh-runtime-adapter/model-mesh-ovms-adapter/server"
+	"google.golang.org/grpc"
+	"sigs.k8s.io/controller-runtime/pkg/log/zap"
+)
+
+func main() {
+	log := zap.New(zap.UseDevMode(true))
+	log = log.WithName("OpenVINO Adapter")
+
+	adapterConfig, err := server.GetAdapterConfigurationFromEnv(log)
+	if err != nil {
+		log.Error(err, "Error reading configuration")
+		os.Exit(1)
+	}
+	log.Info("Starting OpenVINO Adapter Server", "adapter_config", adapterConfig)
+
+	server := server.NewOvmsAdapterServer(adapterConfig.OvmsPort, adapterConfig, log)
+
+	lis, err := net.Listen("tcp", fmt.Sprintf("localhost:%d", adapterConfig.Port))
+	if err != nil {
+		log.Error(err, "*** Adapter failed to listen port")
+		os.Exit(1)
+	}
+	log.Info("Adapter will run at port", "port", adapterConfig.Port, "OpenVINO port", adapterConfig.OvmsPort)
+
+	grpcServer := grpc.NewServer()
+	mmesh.RegisterModelRuntimeServer(grpcServer, server)
+	log.Info("Adapter gRPC Server Registered...")
+
+	err = grpcServer.Serve(lis)
+
+	if err != nil {
+		log.Error(err, "*** Adapter terminated with error ")
+	} else {
+		log.Info("*** Adapter terminated")
+	}
+}

model-mesh-ovms-adapter/scripts/run_tests.sh

@@ -0,0 +1,21 @@
+#!/bin/sh
+# Copyright 2022 IBM Corporation
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.#
+
+set -e
+DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" >/dev/null 2>&1 && pwd )"
+cd "${DIR}/.."
+
+go build -o main main.go
+go test ./server -v

model-mesh-ovms-adapter/server/adaptmodellayout.go

@@ -0,0 +1,148 @@
+// Copyright 2022 IBM Corporation
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+package server
+
+import (
+	"context"
+	"fmt"
+	"io/ioutil"
+	"os"
+	"path/filepath"
+	"strconv"
+	"strings"
+
+	"github.com/go-logr/logr"
+
+	"github.com/kserve/modelmesh-runtime-adapter/internal/util"
+)
+
+func adaptModelLayoutForRuntime(ctx context.Context, rootModelDir, modelID, modelType, modelPath, schemaPath string, log logr.Logger) error {
+	// convert to lower case and remove anything after the :
+	modelType = strings.ToLower(strings.Split(modelType, ":")[0])
+
+	ovmsModelIDDir, err := util.SecureJoin(rootModelDir, ovmsModelSubdir, modelID)
+	if err != nil {
+		log.Error(err, "Unable to securely join", "rootModelDir", rootModelDir, "ovmsModelSubdir", ovmsModelSubdir, "modelID", modelID)
+		return err
+	}
+	// clean up and then create directory where the rewritten model repo will live
+	if removeErr := os.RemoveAll(ovmsModelIDDir); removeErr != nil {
+		log.Info("Ignoring error trying to remove dir", "Directory", ovmsModelIDDir, "Error", removeErr)
+	}
+	if mkdirErr := os.MkdirAll(ovmsModelIDDir, 0755); mkdirErr != nil {
+		return fmt.Errorf("Error creating directories for path %s: %w", ovmsModelIDDir, mkdirErr)
+	}
+
+	modelPathInfo, err := os.Stat(modelPath)
+	if err != nil {
+		return fmt.Errorf("Error calling stat on model file: %w", err)
+	}
+
+	if !modelPathInfo.IsDir() {
+		// simple case if ModelPath points to a file
+		err = createOvmsModelRepositoryFromPath(modelPath, "1", schemaPath, modelType, ovmsModelIDDir, log)
+	} else {
+		files, err1 := ioutil.ReadDir(modelPath)
+		if err1 != nil {
+			return fmt.Errorf("Could not read files in dir %s: %w", modelPath, err1)
+		}
+		err = createOvmsModelRepositoryFromDirectory(files, modelPath, schemaPath, modelType, ovmsModelIDDir, log)
+	}
+	if err != nil {
+		return fmt.Errorf("Error processing model/schema files for model %s: %w", modelID, err)
+	}
+
+	return nil
+}
+
+// Creates the ovms model structure /models/_ovms_models/model-id/1/<model files>
+// Within this path there will be a symlink back to the original /models/model-id directory tree.
+func createOvmsModelRepositoryFromDirectory(files []os.FileInfo, modelPath, schemaPath, modelType, ovmsModelIDDir string, log logr.Logger) error {
+	var err error
+
+	// allow the directory to contain version directories
+	// try to find the largest version directory
+	versionNumber := largestNumberDir(files)
+	if versionNumber != "" {
+		// found a version directory so step into it
+		if modelPath, err = util.SecureJoin(modelPath, versionNumber); err != nil {
+			log.Error(err, "Unable to securely join", "modelPath", modelPath, "versionNumber", versionNumber)
+			return err
+		}
+	} else {
+		versionNumber = "1"
+	}
+
+	return createOvmsModelRepositoryFromPath(modelPath, versionNumber, schemaPath, modelType, ovmsModelIDDir, log)
+}
+
+func createOvmsModelRepositoryFromPath(modelPath, versionNumber, schemaPath, modelType, ovmsModelIDDir string, log logr.Logger) error {
+	var err error
+
+	modelPathInfo, err := os.Stat(modelPath)
+	if err != nil {
+		return fmt.Errorf("Error calling stat on %s: %w", modelPath, err)
+	}
+
+	linkPathComponents := []string{ovmsModelIDDir, versionNumber}
+	if !modelPathInfo.IsDir() {
+		// special case to rename the file for an ONNX model
+		if modelType == "onnx" {
+			linkPathComponents = append(linkPathComponents, onnxModelFilename)
+		} else {
+			linkPathComponents = append(linkPathComponents, modelPathInfo.Name())
+		}
+	}
+
+	linkPath, err := util.SecureJoin(linkPathComponents...)
+	if err != nil {
+		return fmt.Errorf("Error joining link path: %w", err)
+	}
+
+	if err = os.MkdirAll(filepath.Dir(linkPath), 0755); err != nil {
+		return fmt.Errorf("Error creating directories for path %s: %w", linkPath, err)
+	}
+
+	if err = os.Symlink(modelPath, linkPath); err != nil {
+		return fmt.Errorf("Error creating symlink: %w", err)
+	}
+
+	if schemaPath == "" {
+		return nil
+	}
+
+	return nil
+}
+
+// Returns the largest positive int dir as long as all fileInfo dirs are integers (files are ignored).
+// If fileInfos is empty or contains any any non-integer dirs, this will return the empty string.
+func largestNumberDir(fileInfos []os.FileInfo) string {
+	largestInt := 0
+	largestDir := ""
+	for _, f := range fileInfos {
+		if !f.IsDir() {
+			continue
+		}
+		i, err := strconv.Atoi(f.Name())
+		if err != nil {
+			// must all be numbers
+			return ""
+		}
+		if i > largestInt {
+			largestInt = i
+			largestDir = f.Name()
+		}
+	}
+	return largestDir
+}