[Serving] Verify that functions referenced in steps are added as chil…

…dren (#1866)

mlrun/runtimes/serving.py

@@ -447,6 +447,7 @@ def _deploy_function_refs(self):
             function_object.metadata.labels[
                 "mlrun/parent-function"
             ] = self.metadata.name
+            function_object._is_child_function = True
             if not function_object.spec.graph:
                 # copy the current graph only if the child doesnt have a graph of his own
                 function_object.set_env("SERVING_CURRENT_FUNCTION", function_ref.name)
@@ -537,10 +538,20 @@ def deploy(
         if not self.spec.graph:
             raise ValueError("nothing to deploy, .spec.graph is none, use .add_model()")
 
-        if self.spec.graph.kind != StepKinds.router:
+        if self.spec.graph.kind != StepKinds.router and not getattr(
+            self, "_is_child_function", None
+        ):
             # initialize or create required streams/queues
             self.spec.graph.check_and_process_graph()
             self.spec.graph.init_queues()
+            functions_in_steps = self.spec.graph.list_child_functions()
+            child_functions = list(self._spec.function_refs.keys())
+            for function in functions_in_steps:
+                if function not in child_functions:
+                    raise mlrun.errors.MLRunInvalidArgumentError(
+                        f"function {function} is used in steps and is not defined, "
+                        "use the .add_child_function() to specify child function attributes"
+                    )
 
         # Handle secret processing before handling child functions, since secrets are transferred to them
         if self.spec.secret_sources:

mlrun/serving/states.py

@@ -1076,6 +1076,14 @@ def init_queues(self):
             if step.kind == StepKinds.queue:
                 step.init_object(self.context, None)
 
+    def list_child_functions(self):
+        """return a list of child function names referred to in the steps"""
+        functions = []
+        for step in self.get_children():
+            if step.function and step.function not in functions:
+                functions.append(step.function)
+        return functions
+
     def is_empty(self):
         """is the graph empty (no child steps)"""
         return len(self.steps) == 0

tests/serving/test_flow.py

@@ -339,3 +339,15 @@ def check_function(name, fn):
         "new_function",
         mlrun.new_function("test2", command=function_path, kind="serving"),
     )
+
+
+def test_missing_functions():
+    function = mlrun.new_function("tests", kind="serving")
+    graph = function.set_topology("flow", engine="async")
+    graph.to(name="s1", class_name="Echo").to(
+        name="s2", class_name="Echo", function="child_func"
+    )
+    with pytest.raises(
+        mlrun.errors.MLRunInvalidArgumentError, match=r"function child_func*"
+    ):
+        function.deploy()