Killing Vertx Server on failure and ability to use Pipeline + InferenceConfiguration JSON as cli (start and build) input config

konduit-serving-build/src/main/java/ai/konduit/serving/build/util/ModuleUtils.java

@@ -24,8 +24,8 @@
 import ai.konduit.serving.pipeline.util.ObjectMappers;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.io.FileUtils;
-import org.nd4j.common.base.Preconditions;
 import org.nd4j.common.io.ClassPathResource;
+import org.nd4j.shade.jackson.databind.JsonNode;
 
 import java.io.File;
 import java.io.IOException;
@@ -38,19 +38,43 @@ public class ModuleUtils {
     private ModuleUtils(){ }
 
     public static Map<StepId, List<RunnerInfo>> runnersForFile(File f){
-        //First determine if JSON or YAML...
-        boolean json = true;    //TODO
         try {
-            if (json) {
-                return runnersForJson(FileUtils.readFileToString(f, StandardCharsets.UTF_8));
+            JsonNode jsonConfiguration = readConfiguration(FileUtils.readFileToString(f, StandardCharsets.UTF_8));
+            if(jsonConfiguration == null) {
+                throw new IllegalStateException("Unable to parse string into a valid pipeline configuration from file: " + f.getAbsolutePath());
             } else {
-                return runnersForYaml(FileUtils.readFileToString(f, StandardCharsets.UTF_8));
+                return runnersForJson(
+                        (jsonConfiguration.has("pipeline") ?
+                            jsonConfiguration.get("pipeline") :
+                            jsonConfiguration)
+                        .toString());
             }
         } catch (IOException e){
             throw new RuntimeException("Error reading JSON/YAML from file: " + f.getAbsolutePath(), e);
         }
     }
 
+    /**
+     * Parse the given configuration yaml/json string to {@link JsonNode}.
+     *
+     * @param configurationString given configuration string. Can be a JSON/YAML string
+     * @return Read configuration to {@link JsonNode}. Returns null on failure.
+     */
+    private static JsonNode readConfiguration(String configurationString) {
+        try {
+            return ObjectMappers.json().readTree(configurationString);
+        } catch (Exception jsonProcessingErrors) {
+            try {
+                return ObjectMappers.yaml().readTree(configurationString);
+            } catch (Exception yamlProcessingErrors) {
+                log.error("Given configuration: '{}' does not contain a valid JSON/YAML object", configurationString);
+                log.error("\n\nErrors while processing as a json string:", jsonProcessingErrors);
+                log.error("\n\nErrors while processing as a yaml string:", yamlProcessingErrors);
+                return null;
+            }
+        }
+    }
+
     public static Map<StepId, List<RunnerInfo>> runnersForJson(String json){
         //System.out.println(json);
 
@@ -110,10 +134,6 @@ public static Map<StepId, List<RunnerInfo>> runnersForJson(String json){
         return out;
     }
 
-    public static Map<StepId, List<RunnerInfo>> runnersForYaml(String yaml){
-        throw new UnsupportedOperationException("Not yet implemented");
-    }
-
     public static Module moduleForJsonType(String jsonType){
         Map<String,List<RunnerInfo>> map = jsonNameToRunnerClass();
         if(!map.containsKey(jsonType)){

konduit-serving-cli/src/main/java/ai/konduit/serving/cli/launcher/command/KonduitRunCommand.java

@@ -161,6 +161,12 @@ protected void deploy() {
             throw new CLIException(String.format("Unsupported service type %s", serviceType));
         }
 
-        deploy(mainVerticle, vertx, deploymentOptions, res -> {});
+        deploy(mainVerticle, vertx, deploymentOptions, handler -> {
+            if (handler.failed()) {
+                out.format("Unable to deploy server for configuration %n%s%n", inferenceConfiguration.toJson());
+
+                vertx.close();
+            }
+        });
     }
 }

konduit-serving-cli/src/main/java/ai/konduit/serving/cli/launcher/command/ServeCommand.java

@@ -54,6 +54,8 @@
 @Slf4j
 public class ServeCommand extends DefaultCommand {
 
+    protected String host;
+    protected int port;
     protected String id;
     protected String launcher;
     protected int instances = 1;
@@ -64,6 +66,32 @@ public class ServeCommand extends DefaultCommand {
     protected boolean redirect;
     protected String jvmOptions;
 
+    /**
+     * Sets the host name of the konduit server.
+     *
+     * @param host host name
+     */
+    @Option(shortName = "h", longName = "host", argName = "host")
+    @DefaultValue("localhost")
+    @Description("Specifies the host name of the konduit server when the configuration provided is " +
+            "just a pipeline configuration instead of a whole inference configuration. Defaults to 'localhost'.")
+    public void setHost(String host) {
+        this.host = host;
+    }
+
+    /**
+     * Sets the port of the konduit server.
+     *
+     * @param port port number
+     */
+    @Option(shortName = "p", longName = "port", argName = "port")
+    @DefaultValue("0")
+    @Description("Specifies the port number of the konduit server when the configuration provided is " +
+            "just a pipeline configuration instead of a whole inference configuration. Defaults to '0'.")
+    public void setPort(int port) {
+        this.port = port;
+    }
+
     /**
      * Sets the number of instance of the verticle to create.
      *

konduit-serving-cli/src/main/java/ai/konduit/serving/cli/launcher/command/build/extension/ServeBuildCommand.java

@@ -31,6 +31,8 @@
 import org.apache.commons.io.FileUtils;
 import org.nd4j.shade.guava.base.Strings;
 import org.nd4j.shade.jackson.databind.JsonNode;
+import org.nd4j.shade.jackson.databind.node.JsonNodeFactory;
+import org.nd4j.shade.jackson.databind.node.ObjectNode;
 import org.nd4j.shade.jackson.databind.node.TextNode;
 
 import java.io.File;
@@ -92,6 +94,15 @@ public void run() {
                 out.format("Invalid JSON/YAML configuration or invalid configuration file path defined by: %n%s", configuration);
                 System.exit(1);
             } else {
+                if(!(jsonConfiguration.has("host") || jsonConfiguration.has("port") ||
+                        jsonConfiguration.has("pipeline"))) {
+                    // Assume that it's a json for a konduit serving pipeline and not a complete inference configuration
+                    jsonConfiguration = new ObjectNode(JsonNodeFactory.instance)
+                            .put("host", this.host)
+                            .put("port", this.port)
+                            .set("pipeline", jsonConfiguration.deepCopy());
+                }
+
                 Object pipeline = jsonConfiguration.get("pipeline");
                 if(pipeline == null) {
                     out.format("Invalid JSON/YAML configuration or invalid configuration file path defined by: %n%s", configuration);
@@ -180,7 +191,7 @@ protected JsonNode getConfigurationFromFileOrString(String jsonOrYamlFileOrStrin
     /**
      * Parse the given configuration yaml/json string to {@link JsonNode}.
      *
-     * @param configurationString given configuration string. Can be a file path or a JSON string
+     * @param configurationString given configuration string. Can be a JSON/YAML string
      * @return Read configuration to {@link JsonNode}. Returns null on failure.
      */
     private JsonNode readConfiguration(String configurationString) {
@@ -190,8 +201,7 @@ private JsonNode readConfiguration(String configurationString) {
             try {
                 JsonNode jsonNode = ObjectMappers.yaml().readTree(configurationString);
                 if(jsonNode instanceof TextNode) {
-                    throw new IllegalStateException("Expected JsonNode to be ObjectNode but was TextNode while parsing as a YAML object. " +
-                            "This usually indicates that the YAML was not parsed as expected. Could be a bad configuration file path in: " + configurationString);
+                    throw new FileNotFoundException("File does not exist at path: " + configurationString);
                 } else {
                     return jsonNode;
                 }

konduit-serving-vertx-protocols/konduit-serving-grpc/src/main/java/ai/konduit/serving/vertx/protocols/grpc/verticle/InferenceVerticleGrpc.java

@@ -37,6 +37,13 @@ public class InferenceVerticleGrpc extends InferenceVerticle {
 
     @Override
     public void start(Promise<Void> startPromise) {
+        try {
+            initialize();
+        } catch (Exception exception) {
+            startPromise.fail(exception);
+            return;
+        }
+
         int port;
 
         String portEnvValue = System.getenv(EnvironmentConstants.KONDUIT_SERVING_PORT);

konduit-serving-vertx-protocols/konduit-serving-http/src/main/java/ai/konduit/serving/vertx/protocols/http/verticle/InferenceVerticleHttp.java

@@ -64,6 +64,13 @@ public class InferenceVerticleHttp extends InferenceVerticle {
 
     @Override
     public void start(Promise<Void> startPromise) {
+        try {
+            initialize();
+        } catch (Exception exception) {
+            startPromise.fail(exception);
+            return;
+        }
+
         int port;
 
         String portEnvValue = System.getenv(EnvironmentConstants.KONDUIT_SERVING_PORT);

konduit-serving-vertx/src/main/java/ai/konduit/serving/vertx/verticle/InferenceVerticle.java

@@ -43,10 +43,7 @@ public abstract class InferenceVerticle extends AbstractVerticle {
     protected Pipeline pipeline;
     protected PipelineExecutor pipelineExecutor;
 
-    @Override
-    public void init(Vertx vertx, Context context) {
-        super.init(vertx, context);
-
+    protected void initialize() throws Exception {
         inferenceConfiguration = InferenceConfiguration.fromJson(context.config().encode());
         pipeline = inferenceConfiguration.pipeline();
         pipelineExecutor = pipeline.executor();