WIP: strategies and more

.dockerignore

+Dockerfile

Dockerfile

@@ -22,4 +22,4 @@ WORKDIR /coordinator
 RUN ./gradlew --no-daemon installDist
 
 ENV CONFIG_FILE /config.coordinator
-ENTRYPOINT /bin/bash -c "./build/install/coordinator/bin/coordinator $CONFIG_FILE"
+ENTRYPOINT ["/bin/bash", "./build/install/coordinator/bin/coordinator", "--", "$CONFIG_FILE"]

examples/ros3rag.coordinator

 components {
-  component robotCtrlA, docker compose = "ros_place_a",      mqtt topic = "ros-place-a"
-  component robotCtrlB, docker compose = "ros_place_b",      mqtt topic = "ros-place-b"
+  component robotCtrlA, docker compose = "ros_place_a",      mqtt topic = "ros-place-a", status "ready" after 1 sec
+  component robotCtrlB, docker compose = "ros_place_b",      mqtt topic = "ros-place-b", status "ready" after 1 sec
   component ragA,       docker compose = "rag_place_a",      mqtt topic = "rag-a"
   component ragB,       docker compose = "rag_place_b",      mqtt topic = "rag-b"
   component random,     docker compose = "cgv_random_dummy", status "ready" after 1 sec, start as up

src/main/jastadd/Coordinator.flex

@@ -46,11 +46,16 @@ Comment = "//" [^\n\r]+
 "components"         { return sym(Terminals.COMPONENTS); }
 "component"          { return sym(Terminals.COMPONENT); }
 "docker compose"     { return sym(Terminals.DOCKER_COMPOSE); }
-"mqtt topic"         { return sym(Terminals.MQTT_TOPIC); }
+"script"             { return sym(Terminals.SCRIPT); }
+"report"             { return sym(Terminals.REPORT); }
+"mqtt"               { return sym(Terminals.MQTT); }
 "status"             { return sym(Terminals.STATUS); }
 "after"              { return sym(Terminals.AFTER); }
 "sec"                { return sym(Terminals.SEC); }
-"start as up" { return sym(Terminals.START_AS_UP); }
+"in"                 { return sym(Terminals.IN); }
+"start"              { return sym(Terminals.START); }
+"using"              { return sym(Terminals.USING); }
+"after reqs"         { return sym(Terminals.AFTER_REQS); }
 
 "="                  { return sym(Terminals.EQUALS); }
 ","                  { return sym(Terminals.COMMA); }

src/main/jastadd/Coordinator.jrag

@@ -51,30 +51,91 @@ aspect Manipulation {
     for (String service : services) {
       resolveComponentByDockerComposeName(service).ifPresentOrElse(
         comp -> result.add(comp),
-        () -> System.err.println("Could not resolve component for '" + service + "'!")
+        () -> System.err.println("Could not resolve docker component for '" + service + "'!")
       );
     }
     return result;
   }
 
-  public boolean Component.callDockerCompose() throws IOException, InterruptedException {
-    String[] args = { "docker-compose", "up", "-d", getDockerComposeName() };
+  //--- start ---
+  public abstract boolean StartStrategy.start() throws IOException, InterruptedException;
+  @Override
+  public boolean DockerComposeStrategy.start() throws IOException, InterruptedException {
+    if (!getName().isBlank()) {
+      String[] args = { "docker-compose", "up", "-d", getName() };
       if (coordinator().DRY_RUN) {
         System.out.println("Would start > " + java.util.Arrays.toString(args));
         return true;
       }
+      System.out.println("Starting " + Arrays.toString(args));
       ProcessBuilder builder = new ProcessBuilder(args);
 
       Process process = builder.start();
       process.waitFor();
+
+      try (java.io.BufferedReader reader = new java.io.BufferedReader(new java.io.InputStreamReader(process.getInputStream()))) {
+        System.out.println("Out of " + java.util.Arrays.toString(args) + ":\n" + reader.lines().collect(java.util.stream.Collectors.joining("\n")));
+      } catch (IOException e) {
+        e.printStackTrace();
+      }
+
+      try (java.io.BufferedReader reader = new java.io.BufferedReader(new java.io.InputStreamReader(process.getErrorStream()))) {
+        System.out.println("Err of " + java.util.Arrays.toString(args) + ":\n" + reader.lines().collect(java.util.stream.Collectors.joining("\n")));
+      } catch (IOException e) {
+        e.printStackTrace();
+      }
+
       return process.exitValue() == 0;
     }
+    return false;
+  }
+
+  @Override
+  public boolean ScriptStrategy.start() throws IOException, InterruptedException {
+    String[] command = getCommand().split(" ");
+    if (coordinator().DRY_RUN) {
+      System.out.println("Would start script " + Arrays.toString(command));
+      return true;
+    }
+    System.out.println("Starting script " + Arrays.toString(command));
+    ProcessBuilder builder = new ProcessBuilder(command);
+    if (!getCwd().isBlank()) {
+      builder.directory(new java.io.File(getCwd()));
+    }
+
+    Process process = builder.start();
+    // Do not wait for process to be finished
+    return true;
+  }
+
+  @Override
+  public boolean ManualStartStrategy.start() throws IOException, InterruptedException {
+    // do nothing
+    System.out.println("Component " + containingComponent().getName() + " has to be started");
+    return true;
+  }
+
 }
 
 aspect Navigation {
   inh Coordinator Component.coordinator();
   inh Coordinator ParsedPrecedenceRelation.coordinator();
   eq Coordinator.getChild().coordinator() = this;
+
+  inh Component StartStrategy.containingComponent();
+  eq Component.getStartStrategy().containingComponent() = this;
+
+  syn boolean StartStrategy.isDockerComposeStartStrategy() = false;
+  eq DockerComposeStartStrategy.isDockerComposeStartStrategy() = true;
+
+  syn boolean StartStrategy.asDockerComposeStartStrategy() = null;
+  eq DockerComposeStartStrategy.asDockerComposeStartStrategy() = this;
+
+  syn boolean ReportStrategy.isMqttReportStrategy() = false;
+  eq MqttReportStrategy.isMqttReportStrategy() = true;
+
+  syn boolean ReportStrategy.asMqttReportStrategy() = null;
+  eq MqttReportStrategy.asMqttReportStrategy() = this;
 }
 
 aspect Printing {
@@ -117,6 +178,7 @@ aspect Printing {
     return "Component " + getName() + " is " + status;
   }
 
+  //--- details ---
   syn String Coordinator.details() {
     StringBuilder sb = new StringBuilder();
     getComponentList().forEach(comp -> sb.append(comp.details()).append("\n"));
@@ -128,8 +190,8 @@ aspect Printing {
       sj.add(pred.getName());
     }
     return "<Name: " + getName() +
-      ", DockerComposeName: " + getDockerComposeName() +
-      ", MqttTopicPrefix: " + getMqttTopicPrefix() +
+      ", StartStrategy: " + getStartStrategy().details() +
+      ", ReportStrategy: " + getReportStrategy().details() +
       ", Status: " + getStatus() +
       ", NextCommand: " + getNextCommand() +
       ", Deps: " + sj.toString() +
@@ -140,6 +202,12 @@ aspect Printing {
   syn String AutoSetStatus.details() {
     return "[" + getStatus() + " after " + getDelayInSeconds() + " sec]";
   }
+  syn String StartStrategy.details();
+  eq DockerComposeStrategy.details() = "docker compose (name: " + getName() + ")";
+  eq ScriptStrategy.details() = "script (cmd: '" + getCommand() + "'" + (getCwd().isBlank() ? "" : "cwd: '" + getCwd() + "'") + ")";
+  eq ManualStartStrategy.details() = "manual";
+  syn String ReportStrategy.details();
+  eq MqttReportStrategy.details() = "mqtt (topic: " + getTopicPrefix() + ")";
 
   syn String ParsedPrecedenceRelation.prettyPrint() {
     StringBuilder sb = new StringBuilder();

src/main/jastadd/Coordinator.parser

@@ -44,11 +44,21 @@ Component component =
   ;
 
 Component component_body =
-    COMMA DOCKER_COMPOSE EQUALS TEXT.dc component_body.c    {: c.setDockerComposeName(dc); return c; :}
-  | COMMA MQTT_TOPIC EQUALS TEXT.mqtt component_body.c      {: c.setMqttTopicPrefix(mqtt); return c; :}
+    COMMA START USING start_strategy.s component_body.c           {: c.setStartStrategy(s); return c; :}
+  | COMMA REPORT USING report_strategy.r component_body.c       {: c.setReportStrategy(r); return c; :}
   | COMMA STATUS TEXT.s AFTER INTEGER.i SEC component_body.c    {: c.setAutoSetStatus(new AutoSetStatus(s, Integer.parseInt(i))); return c; :}
-  | COMMA START_AS_UP component_body.c      {: c.setStartAsUp(true); return c; :}
-  |                                                         {: return new Component(); :}
+  | COMMA START AFTER_REQS component_body.c                          {: c.setStartAsUp(true); return c; :}
+  |                                                             {: Component c = new Component(); c.setStartStrategy(new ManualStartStrategy()); return c; :}
+  ;
+
+StartStrategy start_strategy =
+    DOCKER_COMPOSE TEXT.dc component_body.c    {: return new DockerComposeStrategy().setName(dc); :}
+  | SCRIPT TEXT.cmd                            {: return new ScriptStrategy().setCommand(cmd); :}
+  | SCRIPT TEXT.cmd IN TEXT.cwd                {: return new ScriptStrategy().setCommand(cmd).setCwd(cwd); :}
+  ;
+
+ReportStrategy report_strategy =
+    MQTT TEXT.topic component_body.c    {: return new ReportStrategy().setMqttTopicPrefix(topic); return c; :}
   ;
 
 StringList string_list =

src/main/jastadd/Coordinator.relast

 Coordinator ::= Component* ParsedPrecedenceRelation* ; // /NextComponentToStart:Component/ ;
-Component ::= <Name:String> <DockerComposeName:String> <MqttTopicPrefix:String> <Status:String> [AutoSetStatus] <StartAsUp:boolean> /<NextCommand:String>/ ;
+Component ::= <Name:String> [StartStrategy] [ReportStrategy] <Status:String> [AutoSetStatus] <StartAsUp:boolean> /<NextCommand:String>/ ;
 rel Component.Predecessor* <-> Component.Successor* ;
 
+abstract StartStrategy ;
+DockerComposeStartStrategy : StartStrategy ::= <Name:String> ;
+ScriptStartStrategy : StartStrategy ::= <Command> <Cwd> ;
+ManualStartStrategy : StartStrategy ;
+
+abstract ReportStrategy ;
+MqttReportStrategy : ReportStrategy ::= <TopicPrefix:String> ;
+
 AutoSetStatus ::= <Status:String> <DelayInSeconds:int> ;
 
 ParsedPrecedenceRelation ;

src/main/java/de/tudresden/inf/st/coordinator/MainCoordinator.java

@@ -85,8 +85,11 @@ public class MainCoordinator implements Callable<Integer> {
     Runtime.getRuntime().addShutdownHook(new Thread(this::close));
 
     for (Component comp : coordinator.getComponentList()) {
-      if (comp.getMqttTopicPrefix() != null && !comp.getMqttTopicPrefix().isBlank()) {
-        comp.connectStatus("mqtt://" + mqttHost + "/" + comp.getMqttTopicPrefix() + "/status");
+      if (comp.hasReportStrategy() && comp.getReportStrategy().isMqttReportStrategy()) {
+        MqttReportStrategy mqttStrategy = comp.getReportStrategy().asMqttReportStrategy();
+        if (mqttStrategy.getTopicPrefix() != null && !mqttStrategy.getTopicPrefix().isBlank()) {
+          comp.connectStatus("mqtt://" + mqttHost + "/" + mqttStrategy.getTopicPrefix() + "/status");
+        }
       }
       final String commandTopic;
       if (comp.getStartAsUp()) {
@@ -94,26 +97,35 @@ public class MainCoordinator implements Callable<Integer> {
         commandTopic = "coordinator/" + comp.getName();
         mainHandler.newConnection(commandTopic, bytes -> {
           try {
-            comp.callDockerCompose();
+            comp.getStartStrategy().start();
             comp.setStatus("ready");
           } catch (IOException | InterruptedException e) {
             comp.setStatus("failedStart");
             e.printStackTrace();
           }
         });
+      } else if (comp.hasReportStrategy() && comp.getReportStrategy().isMqttReportStrategy()) {
+        commandTopic = comp.getReportStrategy().asMqttReportStrategy().getTopicPrefix() + "/command";
       } else {
-        commandTopic = comp.getMqttTopicPrefix() + "/command";
+        commandTopic = null;
       }
+      if (commandTopic != null) {
         comp.connectNextCommand("mqtt://" + mqttHost + "/" + commandTopic, false);
       }
+    }
 
     Set<Component> alreadyRunning = coordinator.getRunningComponents();
     for (Component comp : coordinator.getComponentList()) {
+      if (alreadyRunning.contains(comp)) {
+        comp.setStatus("up");
+        if (comp.hasAutoSetStatus() && !comp.getStartAsUp()) {
+          schduleAutoSetStatus(comp);
+        }
+      }
       if (!alreadyRunning.contains(comp) && !comp.getStartAsUp()) {
-        comp.callDockerCompose();
+        comp.getStartStrategy().start();
         if (comp.hasAutoSetStatus()) {
-          executor.schedule(() -> comp.setStatus(comp.getAutoSetStatus().getStatus()),
-              comp.getAutoSetStatus().getDelayInSeconds(), TimeUnit.SECONDS);
+          schduleAutoSetStatus(comp);
         }
       }
     }
@@ -122,6 +134,13 @@ public class MainCoordinator implements Callable<Integer> {
     return 0;
   }
 
+  private void schduleAutoSetStatus(Component comp) {
+    executor.schedule(() -> {
+      System.out.println("Setting status of " + comp.getName() + " to " + comp.getAutoSetStatus().getStatus());
+      comp.setStatus(comp.getAutoSetStatus().getStatus());
+    }, comp.getAutoSetStatus().getDelayInSeconds(), TimeUnit.SECONDS);
+  }
+
   private void printStatus(String message) {
     String content = message.contains("detail") ? coordinator.details() : coordinator.prettyPrint();
     System.out.println(content);