fix model monitor and add function uri (#484)

* fix model monitor and add function uri
* remove ser_router() (was stale code, not used)
* fix nuclio spec.resources
* add set_metric to v2 serving
Co-authored-by: yaron haviv <yaronh@iguazio.com>

mlrun/runtimes/function.py

@@ -266,6 +266,8 @@ def get_fullname(config, name, project, tag):
         handler = self.spec.function_handler
         if self.spec.readiness_timeout:
             spec.set_config("spec.readinessTimeoutSeconds", self.spec.readiness_timeout)
+        if self.spec.resources:
+            spec.set_config("spec.resources", self.spec.resources)
         if self.spec.no_cache:
             spec.set_config("spec.build.noCache", True)
         if self.spec.replicas:

mlrun/runtimes/serving.py

@@ -86,9 +86,7 @@ def __init__(
         service_account=None,
         readiness_timeout=None,
         models=None,
-        router=None,
         graph=None,
-        router_args=None,
         parameters=None,
         default_class=None,
         load_mode=None,
@@ -119,9 +117,7 @@ def __init__(
         )
 
         self.models = models or {}
-        self.router = router
         self.graph: ServingRouterState = graph
-        self.router_args = router_args
         self.parameters = parameters or {}
         self.default_class = default_class
         self.load_mode = load_mode
@@ -153,11 +149,6 @@ def set_topology(
             class_name=class_name, class_args=class_args
         )
 
-    def set_router(self, router, **router_args):
-        """set the routing class and router arguments"""
-        self.spec.router = router
-        self.spec.router_args = router_args
-
     def set_tracking(self, stream_path, batch=None, sample=None):
         """set tracking log stream parameters"""
         self.spec.parameters["log_stream"] = stream_path
@@ -236,6 +227,7 @@ def deploy(self, dashboard="", project="", tag=""):
         # we currently support a minimal topology of one router + multiple child routes/models
         # in the future we will extend the support to a full graph, the spec is already built accordingly
         serving_spec = {
+            "function_uri": self._function_uri(),
             "version": "v2",
             "parameters": self.spec.parameters,
             "graph": self.spec.graph.to_dict(),

mlrun/serving/server.py

@@ -26,9 +26,10 @@
 
 
 class _StreamContext:
-    def __init__(self, parameters):
+    def __init__(self, parameters, function_uri):
         self.hostname = socket.gethostname()
         self.output_stream = None
+        self.function_uri = function_uri
         out_stream = parameters.get("log_stream", "")
         self.stream_sample = int(parameters.get("log_stream_sample", "1"))
         self.stream_batch = int(parameters.get("log_stream_batch", "1"))
@@ -42,10 +43,17 @@ class ModelServerHost(ModelObj):
     kind = "server"
 
     def __init__(
-        self, graph=None, parameters=None, load_mode=None, verbose=False, version=None
+        self,
+        graph=None,
+        function_uri=None,
+        parameters=None,
+        load_mode=None,
+        verbose=False,
+        version=None,
     ):
         self._graph = None
         self.graph: ServingRouterState = graph
+        self.function_uri = function_uri
         self.parameters = parameters or {}
         self.verbose = verbose
         self.load_mode = load_mode or "sync"
@@ -73,7 +81,7 @@ def init(self, context, namespace):
         self.context = context
         # enrich the context with classes and methods which will be used when
         # initializing classes or handling the event
-        setattr(context, "stream", _StreamContext(self.parameters))
+        setattr(context, "stream", _StreamContext(self.parameters, self.function_uri))
         setattr(context, "merge_root_params", self.merge_root_params)
         setattr(context, "verbose", self.verbose)
 

mlrun/serving/v2_serving.py

@@ -103,6 +103,10 @@ def get_param(self, key: str, default=None):
         """get param by key (specified in the model or the function)"""
         return self._params.get(key, default)
 
+    def set_metric(self, name: str, value):
+        """set real time metric (for model monitoring)"""
+        self.metrics[name] = value
+
     def get_model(self, suffix=""):
         """get the model file(s) and metadata from model store
 
@@ -254,6 +258,7 @@ class _ModelLogPusher:
     def __init__(self, model, context, output_stream=None):
         self.model = model
         self.hostname = context.stream.hostname
+        self.function_uri = context.stream.function_uri
         self.stream_batch = context.stream.stream_batch
         self.stream_sample = context.stream.stream_sample
         self.output_stream = output_stream or context.stream.output_stream
@@ -265,10 +270,11 @@ def __init__(self, model, context, output_stream=None):
     def base_data(self):
         base_data = {
             "class": self.model.__class__.__name__,
-            "worker": self.worker,
+            "worker": self._worker,
             "model": self.model.name,
             "version": self.model.version,
             "host": self.hostname,
+            "function_uri": self.function_uri,
         }
         if getattr(self.model, "labels", None):
             base_data["labels"] = self.model.labels