feat: isolate process management code into ProcessWrapper

This allows for different implementations to be tested.

Author: metacosm

backend/src/main/java/net/laprun/sustainability/power/Security.java

@@ -70,15 +70,31 @@ public Process sudo(String... cmd) throws IOException {
             throw new IllegalArgumentException("No command specified to run with sudo");
         }
 
+        final Process exec;
         if (!isRunningAsAdministrator()) {
             final var args = new String[cmd.length + 2];
             args[0] = "sudo";
             args[1] = "-S";
             System.arraycopy(cmd, 0, args, 2, cmd.length);
             final var runWithSudo = new ProcessBuilder(args);
-            return ProcessBuilder.startPipeline(List.of(new ProcessBuilder("echo", pwd), runWithSudo)).getLast();
+            exec = ProcessBuilder.startPipeline(List.of(new ProcessBuilder("echo", pwd), runWithSudo)).getLast();
         } else {
-            return new ProcessBuilder().command(cmd).start();
+            exec = new ProcessBuilder().command(cmd).start();
         }
+
+        // if the process is already dead, get the error
+        if (!exec.isAlive()) {
+            final var exitValue = exec.exitValue();
+            if (exitValue != 0) {
+                final String errorMsg;
+                try (final var error = exec.errorReader()) {
+                    errorMsg = error.readLine();
+                }
+                throw new RuntimeException(
+                        "Couldn't execute powermetrics. Error code: " + exitValue + ", message: " + errorMsg);
+            }
+        }
+
+        return exec;
     }
 }

backend/src/main/java/net/laprun/sustainability/power/sensors/PowerSensorProducer.java

@@ -1,28 +1,23 @@
 package net.laprun.sustainability.power.sensors;
 
 import jakarta.enterprise.inject.Produces;
-import jakarta.inject.Inject;
 import jakarta.inject.Singleton;
 
-import net.laprun.sustainability.power.Security;
 import net.laprun.sustainability.power.sensors.linux.rapl.IntelRAPLSensor;
 import net.laprun.sustainability.power.sensors.macos.powermetrics.ProcessMacOSPowermetricsSensor;
 
 @Singleton
 public class PowerSensorProducer {
     private static final String OS_NAME = System.getProperty("os.name").toLowerCase();
 
-    @Inject
-    Security security;
-
     @Produces
     public PowerSensor sensor() {
-        return determinePowerSensor(security);
+        return determinePowerSensor();
     }
 
-    public static PowerSensor determinePowerSensor(Security security) {
+    public static PowerSensor determinePowerSensor() {
         if (OS_NAME.contains("mac os x")) {
-            return new ProcessMacOSPowermetricsSensor(security);
+            return new ProcessMacOSPowermetricsSensor();
         }
 
         if (!OS_NAME.contains("linux")) {

backend/src/main/java/net/laprun/sustainability/power/sensors/macos/powermetrics/FileMacOSPowermetricsSensor.java

@@ -39,6 +39,7 @@ public void start(long samplingFrequencyInMillis) {
     @Override
     public void stop() {
         started = false;
+        // need to defer reading metadata until we know the file has been populated
         initMetadata(getInputStream());
     }
 }

backend/src/main/java/net/laprun/sustainability/power/sensors/macos/powermetrics/JavaProcessWrapper.java

@@ -0,0 +1,83 @@
+package net.laprun.sustainability.power.sensors.macos.powermetrics;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+import io.quarkus.logging.Log;
+
+public class JavaProcessWrapper implements ProcessWrapper {
+    private Process powermetrics;
+
+    @Override
+    public InputStream streamForMetadata() {
+        final Process exec;
+        try {
+            exec = execPowermetrics("cpu_power", "-i", "10", "-n", "1");
+            exec.waitFor();
+        } catch (IOException | InterruptedException e) {
+            throw new RuntimeException(e);
+        }
+
+        return exec.getInputStream();
+    }
+
+    @Override
+    public void start(long periodInMilliSeconds) {
+        if (!isRunning()) {
+            // it takes some time for the external process in addition to the sampling time so adjust the sampling frequency to account for this so that at most one measure occurs during the sampling time window
+            periodInMilliSeconds = periodInMilliSeconds > 100 ? periodInMilliSeconds - 50 : periodInMilliSeconds;
+            final var freq = Long.toString(periodInMilliSeconds);
+            try {
+                powermetrics = execPowermetrics("cpu_power,tasks", "--show-process-samp-norm", "--show-process-gpu", "-i",
+                        freq);
+            } catch (IOException e) {
+                throw new RuntimeException(e);
+            }
+        }
+
+    }
+
+    @Override
+    public void stop() {
+        powermetrics.destroy();
+    }
+
+    @Override
+    public boolean isRunning() {
+        return powermetrics != null && powermetrics.isAlive();
+    }
+
+    @Override
+    public InputStream streamForMeasure() {
+        return powermetrics.getInputStream();
+    }
+
+    public Process execPowermetrics(String... options) throws IOException {
+        if (options == null || options.length == 0) {
+            throw new IllegalArgumentException("No powermetrics options specified");
+        }
+        final var additionalArgsCardinality = 3;
+        final var args = new String[options.length + additionalArgsCardinality];
+        args[0] = "sudo";
+        args[1] = "powermetrics";
+        args[2] = "--samplers";
+        System.arraycopy(options, 0, args, additionalArgsCardinality, options.length);
+        final var start = System.currentTimeMillis();
+        final Process exec = new ProcessBuilder().command(args).start();
+        Log.info("Starting process took " + (System.currentTimeMillis() - start) + "ms");
+        // if the process is already dead, get the error
+        if (!exec.isAlive()) {
+            final var exitValue = exec.exitValue();
+            if (exitValue != 0) {
+                final String errorMsg;
+                try (final var error = exec.errorReader()) {
+                    errorMsg = error.readLine();
+                }
+                throw new RuntimeException(
+                        "Couldn't execute powermetrics. Error code: " + exitValue + ", message: " + errorMsg);
+            }
+        }
+        return exec;
+    }
+
+}

backend/src/main/java/net/laprun/sustainability/power/sensors/macos/powermetrics/MacOSPowermetricsSensor.java

@@ -158,6 +158,7 @@ Measures extractPowerMeasure(InputStream powerMeasureInput, Long tick) {
             final var hasGPU = totalSampledGPU != 0;
             double finalTotalSampledGPU = totalSampledGPU;
             double finalTotalSampledCPU = totalSampledCPU;
+            final var endMs = System.currentTimeMillis();
             pidMeasures.forEach((pid, record) -> {
                 final var cpuShare = record.cpu / finalTotalSampledCPU;
                 final var measure = new double[metadata.componentCardinality()];
@@ -178,8 +179,7 @@ Measures extractPowerMeasure(InputStream powerMeasureInput, Long tick) {
                         measure[index] = value;
                     }
                 });
-
-                measures.record(pid, new SensorMeasure(measure, start, System.currentTimeMillis()));
+                measures.record(pid, new SensorMeasure(measure, start, endMs));
             });
         } catch (Exception exception) {
             throw new RuntimeException(exception);

backend/src/main/java/net/laprun/sustainability/power/sensors/macos/powermetrics/PowermetricsProcessHandler.java

@@ -0,0 +1,88 @@
+package net.laprun.sustainability.power.sensors.macos.powermetrics;
+
+import java.io.ByteArrayInputStream;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.concurrent.TimeUnit;
+
+import com.zaxxer.nuprocess.NuAbstractProcessHandler;
+import com.zaxxer.nuprocess.NuProcess;
+
+public class PowermetricsProcessHandler extends NuAbstractProcessHandler {
+    private String errorMsg;
+    private NuProcess process;
+    private final String[] command;
+    private ByteArrayInputStream bais;
+
+    public PowermetricsProcessHandler(String... command) {
+        if (command == null || command.length == 0) {
+            throw new IllegalArgumentException("No powermetrics options specified");
+        }
+        final var additionalArgsCardinality = 3;
+        final var args = new String[command.length + additionalArgsCardinality];
+        args[0] = "sudo";
+        args[1] = "powermetrics";
+        args[2] = "--samplers";
+        System.arraycopy(command, 0, args, additionalArgsCardinality, command.length);
+        this.command = args;
+    }
+
+    public String[] comand() {
+        return command;
+    }
+
+    @Override
+    public void onPreStart(NuProcess nuProcess) {
+        this.process = nuProcess;
+    }
+
+    @Override
+    public void onExit(int statusCode) {
+        if (Integer.MIN_VALUE == statusCode) {
+            throw new IllegalArgumentException("Unknown command " + Arrays.toString(command));
+        }
+        if (statusCode != 0) {
+            throw new RuntimeException("Couldn't execute command " + Arrays.toString(command)
+                    + ". Error code: " + statusCode + ", message: " + errorMsg);
+        }
+    }
+
+    public void stop() {
+        if (process.isRunning()) {
+            process.destroy(false);
+            try {
+                process.waitFor(5, TimeUnit.SECONDS);
+            } catch (InterruptedException e) {
+                throw new RuntimeException(e);
+            } finally {
+                process.destroy(true);
+            }
+        }
+    }
+
+    @Override
+    public void onStdout(ByteBuffer buffer, boolean closed) {
+        if (!closed) {
+            bais = new ByteArrayInputStream(buffer.array());
+        }
+    }
+
+    @Override
+    public void onStderr(ByteBuffer buffer, boolean closed) {
+        if (!closed) {
+            byte[] bytes = new byte[buffer.remaining()];
+            buffer.get(bytes);
+            errorMsg = new String(bytes);
+        }
+        super.onStderr(buffer, closed);
+    }
+
+    public InputStream getInputStream() {
+        return bais;
+    }
+
+    public boolean isRunning() {
+        return process != null && process.isRunning();
+    }
+}

backend/src/main/java/net/laprun/sustainability/power/sensors/macos/powermetrics/ProcessMacOSPowermetricsSensor.java

@@ -2,60 +2,34 @@
 
 import java.io.InputStream;
 
-import net.laprun.sustainability.power.Security;
-
 public class ProcessMacOSPowermetricsSensor extends MacOSPowermetricsSensor {
-    private final Security security;
-    private Process powermetrics;
-
-    public ProcessMacOSPowermetricsSensor(Security security) {
-        this.security = security;
+    private final ProcessWrapper processWrapper = new JavaProcessWrapper();
 
+    public ProcessMacOSPowermetricsSensor() {
         // extract metadata
         try {
-            final var exec = security.execPowermetrics("cpu_power", "-i", "10", "-n", "1");
-
-            // if the process is already dead, get the error
-            if (!exec.isAlive()) {
-                final var exitValue = exec.exitValue();
-                if (exitValue != 0) {
-                    final String errorMsg;
-                    try (final var error = exec.errorReader()) {
-                        errorMsg = error.readLine();
-                    }
-                    throw new RuntimeException(
-                            "Couldn't execute powermetrics. Error code: " + exitValue + ", message: " + errorMsg);
-                }
-            }
-
-            initMetadata(exec.getInputStream());
+            initMetadata(processWrapper.streamForMetadata());
         } catch (Exception e) {
             throw new RuntimeException("Couldn't extract sensor metadata", e);
         }
     }
 
     public void start(long frequency) throws Exception {
-        if (!isStarted()) {
-            // it takes some time for the external process in addition to the sampling time so adjust the sampling frequency to account for this so that at most one measure occurs during the sampling time window
-            frequency = Math.min(0, frequency - 50);
-            final var freq = Long.toString(frequency);
-            powermetrics = security.execPowermetrics("cpu_power,tasks", "--show-process-samp-norm", "--show-process-gpu", "-i",
-                    freq);
-        }
+        processWrapper.start(frequency);
     }
 
     @Override
     public boolean isStarted() {
-        return powermetrics != null && powermetrics.isAlive();
+        return processWrapper.isRunning();
     }
 
     @Override
     protected InputStream getInputStream() {
-        return powermetrics.getInputStream();
+        return processWrapper.streamForMeasure();
     }
 
     @Override
     public void stop() {
-        powermetrics.destroy();
+        processWrapper.stop();
     }
 }

backend/src/main/java/net/laprun/sustainability/power/sensors/macos/powermetrics/ProcessWrapper.java

@@ -0,0 +1,15 @@
+package net.laprun.sustainability.power.sensors.macos.powermetrics;
+
+import java.io.InputStream;
+
+public interface ProcessWrapper {
+    InputStream streamForMetadata();
+
+    void start(long periodInMilliSeconds);
+
+    void stop();
+
+    boolean isRunning();
+
+    InputStream streamForMeasure();
+}