Implement distributed observability and parallel media pipeline

src/com/videoprocessor/Main.java

@@ -9,12 +9,15 @@
 import com.videoprocessor.storage.VideoStorageScanner;
 import com.videoprocessor.processor.FFmpegExecutor;
 import com.videoprocessor.resource.FFmpegResourceManager;
+// Metrics
+import com.videoprocessor.telemetry.MetricsPublisher;
 
 public class Main {
 
     public static void main(String[] args)
             throws Exception {
 
+
         JobQueue jobQueue = new JobQueue();
 
         MetricsTracker metricsTracker =
@@ -32,6 +35,15 @@ public static void main(String[] args)
         FFmpegExecutor ffmpegExecutor =
                 new FFmpegExecutor(resourceManager);
 
+        // Metrics Section
+        MetricsPublisher metricsPublisher =
+                new MetricsPublisher(
+                        metricsTracker,
+                        jobQueue,
+                        resourceManager
+                );
+        // Metrics end
+
         WorkerManager manager =
                 new WorkerManager(
                         metricsTracker,
@@ -40,6 +52,8 @@ public static void main(String[] args)
                         ffmpegExecutor
                 );
 
+        // Initializing metrics
+        metricsPublisher.start();
         manager.start(jobQueue);
 
         int totalJobs = 50000;
@@ -66,7 +80,22 @@ public static void main(String[] args)
             jobQueue.submitJob(job);
         }
 
-        Thread.sleep(5000);
+        while (true) {
+
+            boolean queueEmpty =
+                    jobQueue.size() == 0;
+
+            boolean noActiveJobs =
+                    metricsTracker
+                            .getActiveJobs() == 0;
+
+            if (queueEmpty && noActiveJobs) {
+
+                break;
+            }
+
+            Thread.sleep(1000);
+        }
 
         manager.shutdown();
 

src/com/videoprocessor/MonitorApplication.java

@@ -0,0 +1,15 @@
+package com.videoprocessor;
+
+import com.videoprocessor.monitor
+        .MetricsServer;
+
+public class MonitorApplication {
+
+    public static void main(String[] args) {
+
+        MetricsServer server =
+                new MetricsServer();
+
+        server.start();
+    }
+}

src/com/videoprocessor/metrics/MetricsTracker.java

@@ -7,6 +7,12 @@ public class MetricsTracker {
     private final AtomicInteger processedJobs =
             new AtomicInteger(0);
 
+    private final AtomicInteger failedJobs =
+            new AtomicInteger(0);
+
+    private final AtomicInteger activeJobs =
+            new AtomicInteger(0);
+
     public void incrementProcessedJobs() {
 
         int updatedValue =
@@ -18,7 +24,39 @@ public void incrementProcessedJobs() {
         );
     }
 
+    public void incrementFailedJobs() {
+
+        int updatedValue =
+                failedJobs.incrementAndGet();
+
+        System.out.println(
+                "[METRICS] Failed Jobs: "
+                        + updatedValue
+        );
+    }
+
     public int getProcessedJobs() {
+
         return processedJobs.get();
     }
+
+    public int getFailedJobs() {
+
+        return failedJobs.get();
+    }
+
+    public void incrementActiveJobs() {
+
+        activeJobs.incrementAndGet();
+    }
+
+    public void decrementActiveJobs() {
+
+        activeJobs.decrementAndGet();
+    }
+
+    public int getActiveJobs() {
+
+        return activeJobs.get();
+    }
 }

src/com/videoprocessor/monitor/DashboardRenderer.java

@@ -0,0 +1,4 @@
+package com.videoprocessor.monitor;
+
+public class DashboardRenderer {
+}

src/com/videoprocessor/monitor/MetricsServer.java

@@ -0,0 +1,63 @@
+package com.videoprocessor.monitor;
+
+import java.io.ObjectInputStream;
+import java.net.ServerSocket;
+import java.net.Socket;
+
+public class MetricsServer {
+
+    private static final int PORT = 8081;
+
+    public void start() {
+
+        System.out.println(
+                "[MONITOR] Starting metrics server..."
+        );
+
+        try (ServerSocket serverSocket =
+                     new ServerSocket(PORT)) {
+
+            while (true) {
+
+                Socket socket =
+                        serverSocket.accept();
+
+                Thread.ofVirtual().start(() -> {
+
+                    handleClient(socket);
+                });
+            }
+
+        } catch (Exception e) {
+
+            System.out.println(
+                    "[MONITOR ERROR] "
+                            + e.getMessage()
+            );
+        }
+    }
+
+    private void handleClient(
+            Socket socket
+    ) {
+
+        try (ObjectInputStream input =
+                     new ObjectInputStream(
+                             socket.getInputStream()
+                     )) {
+
+            MetricsSnapshot snapshot =
+                    (MetricsSnapshot)
+                            input.readObject();
+
+            System.out.println(snapshot);
+
+        } catch (Exception e) {
+
+            System.out.println(
+                    "[CLIENT ERROR] "
+                            + e.getMessage()
+            );
+        }
+    }
+}

src/com/videoprocessor/monitor/MetricsSnapshot.java

@@ -0,0 +1,72 @@
+package com.videoprocessor.monitor;
+
+import java.io.Serializable;
+
+public class MetricsSnapshot
+        implements Serializable {
+
+    private final int processedJobs;
+
+    private final int failedJobs;
+
+    private final int queueSize;
+
+    private final int activeFFmpegProcesses;
+
+    public MetricsSnapshot(
+            int processedJobs,
+            int failedJobs,
+            int queueSize,
+            int activeFFmpegProcesses
+    ) {
+
+        this.processedJobs =
+                processedJobs;
+
+        this.failedJobs =
+                failedJobs;
+
+        this.queueSize =
+                queueSize;
+
+        this.activeFFmpegProcesses =
+                activeFFmpegProcesses;
+    }
+
+    public int getProcessedJobs() {
+        return processedJobs;
+    }
+
+    public int getFailedJobs() {
+        return failedJobs;
+    }
+
+    public int getQueueSize() {
+        return queueSize;
+    }
+
+    public int getActiveFFmpegProcesses() {
+        return activeFFmpegProcesses;
+    }
+
+    @Override
+    public String toString() {
+
+        return """
+                ==============================
+                METRICS SNAPSHOT
+                ==============================
+                Processed Jobs      : %d
+                Failed Jobs         : %d
+                Queue Size          : %d
+                Active FFmpeg       : %d
+                ==============================
+                """
+                .formatted(
+                        processedJobs,
+                        failedJobs,
+                        queueSize,
+                        activeFFmpegProcesses
+                );
+    }
+}

src/com/videoprocessor/processor/FFmpegExecutor.java

@@ -1,6 +1,6 @@
 //package com.videoprocessor.processor;
 //
-//import com.videoprocessor.resource
+//import com.videoprocessor.com.videoprocessor.resource
 //        .FFmpegResourceManager;
 //
 //import java.io.BufferedReader;

src/com/videoprocessor/queue/JobQueue.java

@@ -13,6 +13,11 @@ public JobQueue() {
         this.queue = new LinkedBlockingQueue<>();
     }
 
+    public int size() {
+
+        return queue.size();
+    }
+
     public void submitJob(Job job) throws InterruptedException {
         queue.put(job);
         System.out.println(

src/resource/FFmpegResourceManager.java → src/com/videoprocessor/resource/FFmpegResourceManager.java

@@ -1,12 +1,15 @@
 package com.videoprocessor.resource;
 
 import java.util.concurrent.Semaphore;
+import java.util.concurrent.atomic.AtomicInteger;
 
 public class FFmpegResourceManager {
 
-    // max concurrent FFmpeg processes
     private final Semaphore semaphore;
 
+    private final AtomicInteger activeProcesses =
+            new AtomicInteger(0);
+
     public FFmpegResourceManager(
             int maxConcurrentProcesses
     ) {
@@ -22,19 +25,30 @@ public void acquire()
 
         semaphore.acquire();
 
+        int active =
+                activeProcesses.incrementAndGet();
+
         System.out.println(
-                "[RESOURCE] Permit acquired | Available: "
-                        + semaphore.availablePermits()
+                "[RESOURCE] Permit acquired | Active FFmpeg: "
+                        + active
         );
     }
 
     public void release() {
 
         semaphore.release();
 
+        int active =
+                activeProcesses.decrementAndGet();
+
         System.out.println(
-                "[RESOURCE] Permit released | Available: "
-                        + semaphore.availablePermits()
+                "[RESOURCE] Permit released | Active FFmpeg: "
+                        + active
         );
     }
+
+    public int getActiveProcesses() {
+
+        return activeProcesses.get();
+    }
 }

src/com/videoprocessor/telemetry/MetricsClient.java

@@ -0,0 +1,44 @@
+package com.videoprocessor.telemetry;
+
+import com.videoprocessor.monitor
+        .MetricsSnapshot;
+
+import java.io.ObjectOutputStream;
+import java.net.Socket;
+
+public class MetricsClient {
+
+    private static final String HOST =
+            "localhost";
+
+    private static final int PORT = 8081;
+
+    public void send(
+            MetricsSnapshot snapshot
+    ) {
+
+        try (
+
+                Socket socket =
+                        new Socket(HOST, PORT);
+
+                ObjectOutputStream output =
+                        new ObjectOutputStream(
+                                socket.getOutputStream()
+                        )
+
+        ) {
+
+            output.writeObject(snapshot);
+
+            output.flush();
+
+        } catch (Exception e) {
+
+            System.out.println(
+                    "[TELEMETRY ERROR] "
+                            + e.getMessage()
+            );
+        }
+    }
+}