improve: service console output reading (#1605)

### Motivation
Currently service process output streams are read once per second on a
single shared thread. This causes issues with delayed logs, especially
when the JVM crashes at startup as 99% of the times the JVM crash reason
wasn't read from the process error stream. This leaves users with nearly
undiagnosable startup issues.

Example log output with an invalid JVM flag before the change:
```
4.0.0-RC11-SNAPSHOT-f66323ca => create by Proxy 1 --start
[26.08 09:26:50.417] INFO : Starting to create 1 services for Proxy
[26.08 09:26:50.641] INFO : CloudService [uniqueId=04789662-d056-48ba-975a-f69993ed56dd task=Proxy name=Proxy-1] has been started
[26.08 09:26:50.642] INFO : The services were created based on the task. They can be managed with the service command
[26.08 09:26:50.703] INFO : CloudService [uniqueId=04789662-d056-48ba-975a-f69993ed56dd task=Proxy name=Proxy-1] has been stopped
```

### Modification
Now the log cache uses a scaling task executor paired with a custom
non-blocking input stream reader to read the console output produced by
services. Due to the better buffering and faster read times, more log
output should be catched more quickly which also includes jvm crashes.
The result is not perfect, but way better than it was before.

Additionally, empty log lines are now ignored in the parent log cache
class and there is no need anymore for log caches to implement the
handling themselfes. Also a small issue with caching was fixed that
caused the cache to still contain 1 item even if the log cache entry
limit was set to 0.

### Result
Log output of services is not read immediately once data is available
from the input or output stream, making the output more responsive (as
it is nearly immediately printed) and also catches everything that is
being logged.

Example log output with an invalid JVM flag & this change applied:
```
4.0.0-RC11-SNAPSHOT-b99d7fbc => create by Proxy 1 --start
[26.08 09:21:59.245] INFO : Starting to create 1 services for Proxy
[26.08 09:21:59.289] INFO : CloudService [uniqueId=3a718134-d248-41d4-9f40-75751e1c1d64 task=Proxy name=Proxy-1] has been started
[26.08 09:21:59.289] INFO : The services were created based on the task. They can be managed with the service command
[26.08 09:21:59.295] WARN : [Proxy-1/WARN]: Unrecognized option: -XIReallyKnowWhatIAmDoingISwear
[26.08 09:21:59.296] WARN : [Proxy-1/WARN]: Error: Could not create the Java Virtual Machine.
[26.08 09:21:59.296] WARN : [Proxy-1/WARN]: Error: A fatal exception has occurred. Program will exit.
[26.08 09:22:00.303] INFO : CloudService [uniqueId=3a718134-d248-41d4-9f40-75751e1c1d64 task=Proxy name=Proxy-1] has been stopped
```

Author: derklaro

.idea/codeStyles/Project.xml

@@ -20,7 +20,7 @@
   "https://checkstyle.org/dtds/configuration_1_3.dtd">
 <module name="Checker">
   <module name="BeforeExecutionExclusionFileFilter">
-    <property name="fileNamePattern" value="(module\-info\.java$)|(.*[\\|\/]fabric[\\|\/].*$)"/>
+    <property name="fileNamePattern" value="(module\-info\.java$)|(.*[\\|\/]fabric[\\|\/].*|(JVM|Dockerized)Service\.java$)"/>
   </module>
   <module name="SuppressionFilter">
     <property default="checkstyle-suppressions.xml" name="file"

checkstyle.xml

@@ -89,7 +89,6 @@ public class DockerizedService extends JVMService {
 
   protected final DockerClient dockerClient;
   protected final DockerConfiguration configuration;
-  protected final DockerizedServiceLogCache logCache;
 
   protected volatile String containerId;
 
@@ -109,21 +108,20 @@ protected DockerizedService(
     @NonNull DockerClient dockerClient,
     @NonNull DockerConfiguration dockerConfiguration
   ) {
+    var logCache = new DockerizedServiceLogCache(nodeConfig, configuration.serviceId());
     super(
       i18n,
       tickLoop,
       nodeConfig,
       configuration,
       manager,
       eventManager,
+      logCache,
       versionProvider,
       serviceConfigurationPreparer);
 
     this.dockerClient = dockerClient;
     this.configuration = dockerConfiguration;
-
-    super.logCache = this.logCache = new DockerizedServiceLogCache(nodeConfig, this);
-    this.initLogHandler();
   }
 
   @Override
@@ -379,8 +377,8 @@ protected boolean needsImagePull(@NonNull DockerImage image) {
   public final class ServiceLogCacheAdapter extends ResultCallback.Adapter<Frame> {
 
     @Override
-    public void onNext(Frame object) {
-      DockerizedService.this.logCache.handle(object);
+    public void onNext(@NonNull Frame object) {
+      ((DockerizedServiceLogCache) DockerizedService.this.logCache).handle(object);
     }
   }
 }

modules/dockerized-services/impl/src/main/java/eu/cloudnetservice/modules/docker/impl/DockerizedService.java

@@ -17,22 +17,16 @@
 package eu.cloudnetservice.modules.docker.impl;
 
 import com.github.dockerjava.api.model.Frame;
+import eu.cloudnetservice.driver.service.ServiceId;
 import eu.cloudnetservice.node.config.Configuration;
 import eu.cloudnetservice.node.impl.service.defaults.log.AbstractServiceLogCache;
-import eu.cloudnetservice.node.service.CloudService;
-import eu.cloudnetservice.node.service.ServiceConsoleLogCache;
 import java.nio.charset.StandardCharsets;
 import lombok.NonNull;
 
 public class DockerizedServiceLogCache extends AbstractServiceLogCache {
 
-  public DockerizedServiceLogCache(@NonNull Configuration configuration, @NonNull CloudService service) {
-    super(configuration, service);
-  }
-
-  @Override
-  public @NonNull ServiceConsoleLogCache update() {
-    return this;
+  public DockerizedServiceLogCache(@NonNull Configuration configuration, @NonNull ServiceId associatedServiceId) {
+    super(configuration, associatedServiceId);
   }
 
   public void handle(@NonNull Frame frame) {
@@ -49,9 +43,7 @@ protected void handleItem(@NonNull String content, boolean comesFromErrorStream)
     if (content.contains("\n") || content.contains("\r")) {
       for (var input : content.split("\r")) {
         for (var text : input.split("\n")) {
-          if (!text.trim().isEmpty()) {
-            super.handleItem(text, comesFromErrorStream);
-          }
+          super.handleItem(text, comesFromErrorStream);
         }
       }
     }

modules/dockerized-services/impl/src/main/java/eu/cloudnetservice/modules/docker/impl/DockerizedServiceLogCache.java

@@ -16,18 +16,19 @@
 
 package eu.cloudnetservice.node.service;
 
+import eu.cloudnetservice.driver.service.ServiceId;
 import java.util.Collection;
 import java.util.Queue;
 import lombok.NonNull;
 import org.jetbrains.annotations.UnmodifiableView;
 
 public interface ServiceConsoleLogCache {
 
-  @NonNull Queue<String> cachedLogMessages();
-
-  @NonNull ServiceConsoleLogCache update();
+  @NonNull
+  Queue<String> cachedLogMessages();
 
-  @NonNull CloudService service();
+  @NonNull
+  ServiceId associatedServiceId();
 
   int logCacheSize();
 
@@ -42,5 +43,6 @@ public interface ServiceConsoleLogCache {
   void removeHandler(@NonNull ServiceConsoleLineHandler handler);
 
   @NonNull
-  @UnmodifiableView Collection<ServiceConsoleLineHandler> handlers();
+  @UnmodifiableView
+  Collection<ServiceConsoleLineHandler> handlers();
 }

node/api/src/main/java/eu/cloudnetservice/node/service/ServiceConsoleLogCache.java

@@ -24,13 +24,13 @@
 import eu.cloudnetservice.driver.channel.ChannelMessageTarget;
 import eu.cloudnetservice.driver.document.Document;
 import eu.cloudnetservice.driver.event.EventManager;
+import eu.cloudnetservice.driver.event.events.service.CloudServiceLogEntryEvent;
 import eu.cloudnetservice.driver.impl.network.NetworkConstants;
 import eu.cloudnetservice.driver.language.I18n;
 import eu.cloudnetservice.driver.network.HostAndPort;
 import eu.cloudnetservice.driver.network.NetworkChannel;
 import eu.cloudnetservice.driver.network.buffer.DataBuf;
 import eu.cloudnetservice.driver.network.ssl.SSLConfiguration;
-import eu.cloudnetservice.driver.registry.Service;
 import eu.cloudnetservice.driver.service.ProcessSnapshot;
 import eu.cloudnetservice.driver.service.ServiceConfiguration;
 import eu.cloudnetservice.driver.service.ServiceDeployment;
@@ -105,6 +105,7 @@ public abstract class AbstractService implements InternalCloudService {
   protected final DefaultTickLoop mainThread;
   protected final EventManager eventManager;
   protected final Configuration configuration;
+  protected final ServiceConsoleLogCache logCache;
   protected final InternalCloudServiceManager cloudServiceManager;
   protected final ServiceConfiguration serviceConfiguration;
   protected final ServiceVersionProvider serviceVersionProvider;
@@ -121,25 +122,25 @@ public abstract class AbstractService implements InternalCloudService {
   protected final Collection<ServiceRemoteInclusion> installedInclusions = ConcurrentHashMap.newKeySet();
   protected final Collection<ServiceDeployment> installedDeployments = ConcurrentHashMap.newKeySet();
 
-  protected ServiceConsoleLogCache logCache;
-
   protected volatile NetworkChannel networkChannel;
   protected volatile long connectionTimestamp = -1;
 
   protected volatile ServiceInfoSnapshot lastServiceInfo;
   protected volatile ServiceInfoSnapshot currentServiceInfo;
 
   protected AbstractService(
-    @NonNull @Service I18n i18n,
+    @NonNull I18n i18n,
     @NonNull DefaultTickLoop tickLoop,
     @NonNull Configuration nodeConfig,
     @NonNull ServiceConfiguration configuration,
     @NonNull InternalCloudServiceManager manager,
     @NonNull EventManager eventManager,
+    @NonNull ServiceConsoleLogCache logCache,
     @NonNull ServiceVersionProvider versionProvider,
     @NonNull ServiceConfigurationPreparer serviceConfigurationPreparer
   ) {
     this.i18n = i18n;
+    this.logCache = logCache;
     this.mainThread = tickLoop;
     this.configuration = nodeConfig;
     this.eventManager = eventManager;
@@ -162,6 +163,7 @@ protected AbstractService(
       ServiceLifeCycle.PREPARED,
       configuration.propertyHolder().immutableCopy());
     this.pushServiceInfoSnapshotUpdate(ServiceLifeCycle.PREPARED, false);
+    this.initStandardServiceLogHandler();
 
     // register the service locally for now
     manager.registerUnacceptedService(this);
@@ -884,6 +886,32 @@ protected void downloadInclusionFile(@NonNull ServiceRemoteInclusion inclusion,
       this.serviceId().nodeUniqueId()};
   }
 
+  protected void initStandardServiceLogHandler() {
+    this.logCache.addHandler((_, line, stderr) -> {
+      for (var logTarget : this.logTargets) {
+        if (logTarget._1().equals(ChannelMessageSender.self().toTarget())) {
+          // the current target is the node this service is running on, print it directly here
+          this.eventManager.callEvent(logTarget._2(), new CloudServiceLogEntryEvent(this.currentServiceInfo,
+            line,
+            stderr ? CloudServiceLogEntryEvent.StreamType.STDERR : CloudServiceLogEntryEvent.StreamType.STDOUT));
+        } else {
+          // the listener is listening remotely, send the line to the network component
+          ChannelMessage.builder()
+            .target(logTarget._1())
+            .channel(NetworkConstants.INTERNAL_MSG_CHANNEL)
+            .message("screen_new_line")
+            .buffer(DataBuf.empty()
+              .writeObject(this.currentServiceInfo)
+              .writeString(logTarget._2())
+              .writeString(line)
+              .writeBoolean(stderr))
+            .build()
+            .send();
+        }
+      }
+    });
+  }
+
   protected abstract void startProcess();
 
   protected abstract void stopProcess();

node/impl/src/main/java/eu/cloudnetservice/node/impl/service/defaults/AbstractService.java

@@ -170,20 +170,14 @@ public DefaultCloudServiceManager(
         })
         .currentGetter(group -> this.serviceProviderByName(group.name()).serviceInfo())
         .build());
-    // schedule the updating of the local service log cache
+
+    // schedule the service watchdog to run once per second
     mainThread.scheduleTask(() -> {
       for (var service : this.localCloudServices()) {
-        // we only need to look at running services
-        if (service.lifeCycle() == ServiceLifeCycle.RUNNING) {
-          // detect dead services and stop them
-          if (service.alive()) {
-            service.serviceConsoleLogCache().update();
-            LOGGER.trace("Updated service log cache of {}", service.serviceId().name());
-          } else {
-            eventManager.callEvent(new CloudServicePreForceStopEvent(service));
-            service.stop();
-            LOGGER.trace("Stopped dead service {}", service.serviceId().name());
-          }
+        if (service.lifeCycle() == ServiceLifeCycle.RUNNING && !service.alive()) {
+          eventManager.callEvent(new CloudServicePreForceStopEvent(service));
+          service.stop();
+          LOGGER.debug("Stopped dead service {}", service.serviceId().name());
         }
       }
       return null;