Improve handling of Consul env variables (#953)

archetypes/streampipes-archetype-extensions-jvm/src/main/resources/archetype-resources/Dockerfile

@@ -15,9 +15,6 @@
 
 FROM adoptopenjdk/openjdk8-openj9:alpine
 
-EXPOSE 8090
-ENV CONSUL_LOCATION consul
-
 COPY ./target/${artifactId}.jar  /streampipes-extensions-service.jar
 
 ENTRYPOINT ["java", "-jar", "/streampipes-extensions-service.jar"]

archetypes/streampipes-archetype-pe-processors-flink/src/main/resources/archetype-resources/Dockerfile

@@ -15,9 +15,6 @@
 
 FROM adoptopenjdk/openjdk8-openj9:alpine
 
-EXPOSE 8090
-ENV CONSUL_LOCATION consul
-
 COPY ./target/${artifactId}.jar  /streampipes-processing-element-container.jar
 
 ENTRYPOINT ["java", "-jar", "/streampipes-processing-element-container.jar"]

archetypes/streampipes-archetype-pe-sinks-flink/src/main/resources/archetype-resources/Dockerfile

@@ -15,9 +15,6 @@
 
 FROM adoptopenjdk/openjdk8-openj9:alpine
 
-EXPOSE 8090
-ENV CONSUL_LOCATION consul
-
 COPY ./target/${artifactId}.jar  /streampipes-processing-element-container.jar
 
 ENTRYPOINT ["java", "-jar", "/streampipes-processing-element-container.jar"]

streampipes-backend/Dockerfile

@@ -15,9 +15,6 @@
 
 FROM eclipse-temurin:11-jre-focal
 
-EXPOSE 8030
-ENV CONSUL_LOCATION consul
-
 COPY target/streampipes-backend.jar  /streampipes-backend.jar
 
 ENTRYPOINT ["java", "-jar", "/streampipes-backend.jar"]

streampipes-commons/src/main/java/org/apache/streampipes/commons/constants/DefaultEnvValues.java

@@ -23,6 +23,11 @@ public class DefaultEnvValues {
   public static final String INITIAL_ADMIN_PW_DEFAULT = "admin";
   public static final String INITIAL_CLIENT_USER_DEFAULT = "sp-service-client";
   public static final String INITIAL_CLIENT_SECRET_DEFAULT = "my-apache-streampipes-secret-key-change-me";
+
+  public static final String CONSUL_HOST_DEFAULT = "consul";
+  public static final String CONSUL_HOST_LOCAL = "localhost";
+  public static final int CONSUL_PORT_DEFAULT = 8500;
+
   public static final int MAX_WAIT_TIME_AT_SHUTDOWN_DEFAULT = 10000;
   public static final boolean INSTALL_PIPELINE_ELEMENTS = true;
 

streampipes-commons/src/main/java/org/apache/streampipes/commons/constants/Envs.java

@@ -21,7 +21,12 @@ public enum Envs {
 
   SP_HOST("SP_HOST"),
   SP_PORT("SP_PORT"),
+
+  @Deprecated(since = "0.90.0", forRemoval = true)
   SP_CONSUL_LOCATION("CONSUL_LOCATION"),
+
+  SP_CONSUL_HOST("SP_CONSUL_HOST"),
+  SP_CONSUL_PORT("SP_CONSUL_PORT"),
   SP_KAFKA_RETENTION_MS("SP_KAFKA_RETENTION_MS"),
   SP_JWT_SECRET("JWT_SECRET"),
   SP_JWT_SIGNING_MODE("SP_JWT_SIGNING_MODE"),
@@ -65,6 +70,10 @@ public Boolean getValueAsBoolean() {
     return CustomEnvs.getEnvAsBoolean(this.envVariableName);
   }
 
+  public boolean getValueAsBooleanOrDefault(boolean defaultValue) {
+    return this.exists() ? this.getValueAsBoolean() : defaultValue;
+  }
+
   public String getEnvVariableName() {
     return envVariableName;
   }

streampipes-extensions/streampipes-connect-adapters-iiot/Dockerfile

@@ -15,10 +15,6 @@
 
 FROM eclipse-temurin:11-jre-focal
 
-ENV CONSUL_LOCATION consul
-
-EXPOSE 8090
-
 COPY target/streampipes-connect-adapters-iiot.jar  /streampipes-connect-adapters-iiot.jar
 
 ENTRYPOINT ["java", "-jar", "/streampipes-connect-adapters-iiot.jar"]

streampipes-extensions/streampipes-connect-adapters/Dockerfile

@@ -15,10 +15,6 @@
 
 FROM eclipse-temurin:11-jre-focal
 
-ENV CONSUL_LOCATION consul
-
-EXPOSE 8090
-
 COPY target/streampipes-connect-adapters.jar  /streampipes-connect-adapters.jar
 
 ENTRYPOINT ["java", "-jar", "/streampipes-connect-adapters.jar"]

streampipes-extensions/streampipes-extensions-all-iiot/Dockerfile

@@ -15,10 +15,6 @@
 
 FROM eclipse-temurin:11-jre-focal
 
-ENV CONSUL_LOCATION consul
-
-EXPOSE 8090
-
 # needed pkgs processors-image-processing-jvm
 RUN apt-get update && apt-get install ttf-dejavu-core && apt-get install fontconfig
 

streampipes-extensions/streampipes-extensions-all-jvm/Dockerfile

@@ -15,10 +15,6 @@
 
 FROM eclipse-temurin:11-jre-focal
 
-ENV CONSUL_LOCATION consul
-
-EXPOSE 8090
-
 # needed pkgs processors-image-processing-jvm
 RUN apt-get update && apt-get install ttf-dejavu-core && apt-get install fontconfig
 

streampipes-extensions/streampipes-pipeline-elements-all-flink/Dockerfile

@@ -15,10 +15,6 @@
 
 FROM eclipse-temurin:11-jre-focal
 
-ENV CONSUL_LOCATION consul
-
-EXPOSE 8090
-
 COPY target/streampipes-pipeline-elements-all-flink.jar  /streampipes-processing-element-container.jar
 
 ENTRYPOINT ["java", "-jar", "/streampipes-processing-element-container.jar"]

streampipes-extensions/streampipes-pipeline-elements-all-jvm/Dockerfile

@@ -15,10 +15,6 @@
 
 FROM eclipse-temurin:11-jre-focal
 
-ENV CONSUL_LOCATION consul
-
-EXPOSE 8090
-
 # needed pkgs processors-image-processing-jvm
 RUN apt-get update && apt-get install ttf-dejavu-core && apt-get install fontconfig
 

streampipes-extensions/streampipes-sinks-internal-jvm/Dockerfile

@@ -15,10 +15,6 @@
 
 FROM eclipse-temurin:11-jre-focal
 
-ENV CONSUL_LOCATION consul
-
-EXPOSE 8090
-
 COPY target/streampipes-sinks-internal-jvm.jar  /streampipes-processing-element-container.jar
 
 ENTRYPOINT ["java", "-jar", "/streampipes-processing-element-container.jar"]

streampipes-extensions/streampipes-sources-watertank-simulator/Dockerfile

@@ -15,10 +15,6 @@
 
 FROM eclipse-temurin:11-jre-focal
 
-ENV CONSUL_LOCATION consul
-
-EXPOSE 8090
-
 COPY target/streampipes-sources-watertank-simulator.jar  /streampipes-processing-element-container.jar
 
 ENTRYPOINT ["java", "-jar", "/streampipes-processing-element-container.jar"]

streampipes-service-discovery-consul/src/main/java/org/apache/streampipes/svcdiscovery/consul/ConsulProvider.java

@@ -17,6 +17,7 @@
  */
 package org.apache.streampipes.svcdiscovery.consul;
 
+import org.apache.streampipes.commons.constants.DefaultEnvValues;
 import org.apache.streampipes.commons.constants.Envs;
 
 import com.orbitz.consul.Consul;
@@ -25,61 +26,52 @@
 
 import java.io.IOException;
 import java.net.InetSocketAddress;
-import java.net.MalformedURLException;
 import java.net.Socket;
-import java.net.URL;
+import java.util.concurrent.TimeUnit;
 
 public class ConsulProvider {
 
   private static final Logger LOG = LoggerFactory.getLogger(ConsulProvider.class);
+  private static final int CHECK_INTERVAL = 1;
 
-  private static final int CONSUL_DEFAULT_PORT = 8500;
-  private static final String CONSUL_URL_REGISTER_SERVICE = "v1/agent/service/register";
+  private final String consulHost;
+  private final String consulUrlString;
+  private final int consulPort;
+
+  public ConsulProvider() {
+    this.consulHost = getConsulHost();
+    this.consulPort = getConsulPort();
+    this.consulUrlString = makeConsulUrl();
+  }
 
   public Consul consulInstance() {
-    URL consulUrl = consulURL();
     boolean connected;
 
     do {
-      LOG.info("Checking if consul is available...");
-      connected = checkConsulAvailable(consulUrl);
+      LOG.info("Checking if consul is available on host {} and port {}", consulHost, consulPort);
+      connected = checkConsulAvailable();
 
       if (!connected) {
-        LOG.info("Retrying in 1 second");
+        LOG.info("Retrying in {} second", CHECK_INTERVAL);
         try {
-          Thread.sleep(1000);
+          TimeUnit.SECONDS.sleep(CHECK_INTERVAL);
         } catch (InterruptedException e) {
           e.printStackTrace();
         }
       }
     } while (!connected);
 
     LOG.info("Successfully connected to Consul");
-    return Consul.builder().withUrl(consulURL()).build();
+    return Consul.builder().withUrl(consulUrlString).build();
   }
 
-  private URL consulURL() {
-    URL url = null;
-
-    if (Envs.SP_CONSUL_LOCATION.exists()) {
-      try {
-        url = new URL("http", Envs.SP_CONSUL_LOCATION.getValue(), CONSUL_DEFAULT_PORT, "");
-      } catch (MalformedURLException e) {
-        e.printStackTrace();
-      }
-    } else {
-      try {
-        url = new URL("http", "localhost", CONSUL_DEFAULT_PORT, "");
-      } catch (MalformedURLException e) {
-        e.printStackTrace();
-      }
-    }
-    return url;
+  public String makeConsulUrl() {
+    return "http://" + consulHost + ":" + consulPort;
   }
 
-  private boolean checkConsulAvailable(URL consulUrl) {
+  private boolean checkConsulAvailable() {
     try {
-      InetSocketAddress sa = new InetSocketAddress(consulUrl.getHost(), consulUrl.getPort());
+      InetSocketAddress sa = new InetSocketAddress(consulHost, consulPort);
       Socket ss = new Socket();
       ss.connect(sa, 1000);
       ss.close();
@@ -89,11 +81,21 @@ private boolean checkConsulAvailable(URL consulUrl) {
       LOG.info("Could not connect to Consul instance...");
       return false;
     }
+  }
 
-
+  private int getConsulPort() {
+    return Envs.SP_CONSUL_PORT.getValueAsIntOrDefault(DefaultEnvValues.CONSUL_PORT_DEFAULT);
   }
 
-  public String makeConsulEndpoint() {
-    return consulURL().toString() + "/" + CONSUL_URL_REGISTER_SERVICE;
+  private String getConsulHost() {
+    if (Envs.SP_CONSUL_LOCATION.exists()) {
+      return Envs.SP_CONSUL_LOCATION.getValue();
+    } else {
+      if (Envs.SP_DEBUG.getValueAsBooleanOrDefault(false)) {
+        return DefaultEnvValues.CONSUL_HOST_LOCAL;
+      } else {
+        return Envs.SP_CONSUL_HOST.getValueOrDefault(DefaultEnvValues.CONSUL_HOST_DEFAULT);
+      }
+    }
   }
 }