Queue Status Metric

frontend/server/src/main/java/org/pytorch/serve/metrics/MetricCollector.java

@@ -12,8 +12,10 @@
 import org.apache.commons.io.IOUtils;
 import org.pytorch.serve.util.ConfigManager;
 import org.pytorch.serve.util.messages.EnvironmentUtils;
+import org.pytorch.serve.wlm.ModelVersionedRefs;
 import org.pytorch.serve.wlm.ModelManager;
 import org.pytorch.serve.wlm.WorkerThread;
+import org.pytorch.serve.wlm.Model;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -31,6 +33,26 @@ public MetricCollector(ConfigManager configManager) {
     @Override
     public void run() {
         try {
+
+            // Collect Model level Metrics
+            ModelManager modelManager = ModelManager.getInstance();
+            for (Map.Entry<String, ModelVersionedRefs> mEntry : modelManager.getAllModels()) {
+                String modelName = mEntry.getKey();
+                for (Map.Entry<String, Model> vEntry : mEntry.getValue().getAllVersions()) {
+                    String versionName = vEntry.getKey();
+                    Model model = vEntry.getValue();
+                    String mvStatusString = model.getQueueStatusString();
+                    loggerMetrics.info(
+                            "{}",
+                            new Metric(
+                                    "QueueStatus",
+                                    mvStatusString,
+                                    "Requests",
+                                    modelName + "/" + versionName,
+                                        new Dimension("Level", "Host")));
+                }
+            }
+
             // Collect System level Metrics
             String[] args = new String[4];
             args[0] = configManager.getPythonExecutable();
@@ -40,9 +62,8 @@ public void run() {
             File workingDir = new File(configManager.getModelServerHome());
 
             String[] envp = EnvironmentUtils.getEnvString(workingDir.getAbsolutePath(), null, null);
-
             final Process p = Runtime.getRuntime().exec(args, envp, workingDir); // NOPMD
-            ModelManager modelManager = ModelManager.getInstance();
+
             Map<Integer, WorkerThread> workerMap = modelManager.getWorkers();
             try (OutputStream os = p.getOutputStream()) {
                 writeWorkerPids(workerMap, os);

frontend/server/src/main/java/org/pytorch/serve/util/ApiUtils.java

@@ -38,8 +38,14 @@
 import org.pytorch.serve.wlm.WorkerState;
 import org.pytorch.serve.wlm.WorkerThread;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 public final class ApiUtils {
 
+
+    private static final Logger logger = LoggerFactory.getLogger(ApiUtils.class);
+
     private ApiUtils() {}
 
     public static ListModelsResponse getModelList(int limit, int pageToken) {
@@ -396,20 +402,21 @@ public static RestJob addRESTInferenceJob(
         if (!ModelManager.getInstance().addJob(job)) {
             String priority = job.getPriority().toString();
             String responseMessage = getInferenceErrorResponseMessage(modelName, version, priority);
+            logger.warn("503 Service Unavailable: {}", responseMessage);
             throw new ServiceUnavailableException(responseMessage);
         }
         return job;
     }
 
     @SuppressWarnings("PMD")
     public static String getInferenceErrorResponseMessage(String modelName, String modelVersion, String jobPriority) {
-        String responseMessage = "Model: " + modelName + "\n";
+        String responseMessage = "Model: " + modelName + ", ";
 
         if (modelVersion != null) {
-            responseMessage += "Version: " + modelVersion + "\n";
+            responseMessage += "Version: " + modelVersion + ", ";
         }
 
-        responseMessage += "Priority: " + jobPriority + "\n";
+        responseMessage += "Priority: " + jobPriority + ", ";
 
         responseMessage +=
                 "Reason: queue full";

frontend/server/src/main/java/org/pytorch/serve/util/PriorityLinkedBlockingDeque.java

@@ -20,11 +20,13 @@ public class PriorityLinkedBlockingDeque<T extends Prioritisable> {
     final ReentrantLock lock = new ReentrantLock();
     private final Condition notEmpty = lock.newCondition();
 
+    private final int queueSize;
     private final float highPrioProb;
     private final ConcurrentHashMap<Priority, LinkedBlockingDeque<T>> priorityDeques;
 
     public PriorityLinkedBlockingDeque(int queueSize, float highPrioProb) {
 
+        this.queueSize = queueSize;
         this.highPrioProb = highPrioProb;
         this.priorityDeques = new ConcurrentHashMap<Priority, LinkedBlockingDeque<T>>();
 
@@ -129,4 +131,15 @@ public T poll(long timeout, TimeUnit unit) throws InterruptedException {
             lock.unlock();
         }
     }
+
+    public String getQueueStatusString() {
+        String response = "";
+        for (Priority priority : Priority.values()) {
+            int currentQueueStatus = this.priorityDeques.get(priority).size();
+            response = response + priority.toString() + "=" + String.valueOf(currentQueueStatus) + ",";
+        }
+        response = response + "queueSize=" + String.valueOf(this.queueSize);
+        return response;
+
+    }
 }

frontend/server/src/main/java/org/pytorch/serve/wlm/Model.java

@@ -252,4 +252,8 @@ public int getResponseTimeout() {
     public void setResponseTimeout(int responseTimeout) {
         this.responseTimeout = responseTimeout;
     }
+
+    public String getQueueStatusString() {
+        return jobsDb.get(DEFAULT_DATA_QUEUE).getQueueStatusString();
+    }
 }