Add LoggingBackend; Add headers; Make json payload not a string; default configuration (#92)

* Add LoggingBackend; Add headers; Make json payload not a string

Signed-off-by: Julien Le Dem <[email protected]>

* removed unecessary class

Signed-off-by: Julien Le Dem <[email protected]>

* improve tests coverage

Signed-off-by: Julien Le Dem <[email protected]>

* improve tests coverage

Signed-off-by: Julien Le Dem <[email protected]>

Co-authored-by: Willy Lulciuc <[email protected]>

Author: julienledem

build.gradle

@@ -32,6 +32,11 @@ repositories {
   mavenCentral()
 }
 
+java {
+    sourceCompatibility = JavaVersion.VERSION_11
+    targetCompatibility = JavaVersion.VERSION_11
+}
+
 ext {
   lombokVersion = '1.18.8'
   slf4jVersion = '1.7.26'

src/main/java/marquez/client/Backend.java

@@ -1,12 +1,13 @@
 package marquez.client;
 
+import java.io.Closeable;
 import javax.annotation.Nullable;
 
 /**
  * The backend contract for sending Marquez instrumentation. Information operations can be sent
  * synchronously or asynchronously over various protocols
  */
-public interface Backend {
+public interface Backend extends Closeable {
 
   void put(String path, String json);
 

src/main/java/marquez/client/Backends.java

@@ -13,6 +13,8 @@
 @Slf4j
 public class Backends {
 
+  @VisibleForTesting static final String DEFAULT_URL = "http://localhost:8080";
+
   /**
    * Will write to a file.
    *
@@ -26,17 +28,26 @@ public static Backend newFileBackend(File file) {
   /**
    * Will issue http requests.
    *
-   * @param baseURL the base url for http requests
+   * @param baseUrl the base url for http requests
    * @return the corresponding backend implementation
    */
   public static Backend newHttpBackend(URL baseUrl) {
     return new HttpBackend(baseUrl);
   }
 
+  /**
+   * Will log requests.
+   *
+   * @return the corresponding backend implementation
+   */
+  public static Backend newLoggingBackend() {
+    return new LoggingBackend();
+  }
+
   /**
    * Initializes the backend base on environment variable configuration.
    *
-   * <p>
+   * <p>configuration:
    *
    * <ul>
    *   <li>MARQUEZ_BACKEND=FILE|HTTP
@@ -52,12 +63,12 @@ public static Backend newBackendFromEnv() {
 
   @VisibleForTesting
   static Backend newBackendFromEnv(Map<String, String> env) {
-    String backendName = env.get("MARQUEZ_BACKEND");
+    String backendName = env.getOrDefault("MARQUEZ_BACKEND", "http");
     switch (backendName.toUpperCase(US)) {
       case "FILE":
         return newFileBackend(new File(env.get("MARQUEZ_FILE")));
       case "HTTP":
-        String configuredBaseUrl = env.get("MARQUEZ_URL");
+        String configuredBaseUrl = env.getOrDefault("MARQUEZ_URL", DEFAULT_URL);
         try {
           return newHttpBackend(new URL(configuredBaseUrl));
         } catch (MalformedURLException e) {
@@ -68,6 +79,8 @@ static Backend newBackendFromEnv(Map<String, String> env) {
                   + " Defaulting to doing nothing.");
           return new NullBackend();
         }
+      case "LOG":
+        return newLoggingBackend();
       default:
         log.error(
             "Could not initialize Marquez backend for "

src/main/java/marquez/client/FileBackend.java

@@ -1,30 +1,42 @@
 package marquez.client;
 
 import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.apache.http.protocol.HTTP.USER_AGENT;
 
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.JsonNode;
 import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.ImmutableMap;
 import java.io.File;
 import java.io.FileWriter;
 import java.io.IOException;
 import java.io.Writer;
 import java.util.LinkedHashMap;
 import java.util.Map;
 import lombok.extern.slf4j.Slf4j;
+import marquez.client.MarquezClient.Version;
+import marquez.client.MarquezHttp.UserAgent;
 
 /** A backend that writes events to a file in json. */
 @Slf4j
 class FileBackend implements Backend {
 
+  private static final TypeReference<JsonNode> JSONNODE = new TypeReference<JsonNode>() {};
+
   private final File file;
   private Writer writer = null;
+  private final Map<String, String> headers;
 
-  public FileBackend(File file) {
+  FileBackend(File file) {
     this(file, initWriter(file));
   }
 
+  @VisibleForTesting
   FileBackend(File file, Writer writer) {
     this.file = file;
     this.writer = writer;
+    Version version = MarquezClient.Version.get();
+    this.headers = ImmutableMap.of(USER_AGENT, UserAgent.of(version).getValue());
   }
 
   @VisibleForTesting
@@ -58,10 +70,11 @@ public void post(String path, String json) {
 
   private void write(String method, String path, String json) {
     if (writer != null) {
-      Map<String, String> call = new LinkedHashMap<String, String>(3);
+      Map<String, Object> call = new LinkedHashMap<>(3);
       call.put("method", method);
+      call.put("headers", headers);
       call.put("path", path);
-      call.put("payload", json);
+      call.put("payload", Utils.fromJson(json, JSONNODE));
       String line = Utils.toJson(call) + "\n";
       try {
         writer.append(line).flush();
@@ -77,4 +90,9 @@ private void write(String method, String path, String json) {
       }
     }
   }
+
+  @Override
+  public void close() throws IOException {
+    writer.close();
+  }
 }

src/main/java/marquez/client/HttpBackend.java

@@ -1,6 +1,7 @@
 package marquez.client;
 
 import com.google.common.annotations.VisibleForTesting;
+import java.io.IOException;
 import java.net.MalformedURLException;
 import java.net.URL;
 
@@ -13,7 +14,7 @@ class HttpBackend implements Backend {
   private final URL baseUrl;
   private final MarquezHttp http;
 
-  public HttpBackend(URL baseUrl) {
+  HttpBackend(URL baseUrl) {
     this(baseUrl, MarquezHttp.create(MarquezClient.Version.get()));
   }
 
@@ -23,6 +24,10 @@ public HttpBackend(URL baseUrl) {
     this.http = http;
   }
 
+  public URL getBaseUrl() {
+    return baseUrl;
+  }
+
   private URL url(String path) {
     try {
       return new URL(this.baseUrl.toString() + path);
@@ -40,4 +45,9 @@ public void put(String path, String json) {
   public void post(String path, String json) {
     http.post(url(path), json);
   }
+
+  @Override
+  public void close() throws IOException {
+    http.close();
+  }
 }

src/main/java/marquez/client/LoggingBackend.java

@@ -0,0 +1,36 @@
+package marquez.client;
+
+import com.google.common.annotations.VisibleForTesting;
+import java.io.IOException;
+import lombok.extern.slf4j.Slf4j;
+import org.slf4j.Logger;
+
+@Slf4j
+class LoggingBackend implements Backend {
+
+  private final Logger logger;
+
+  LoggingBackend() {
+    this(log);
+  }
+
+  @VisibleForTesting
+  LoggingBackend(Logger logger) {
+    this.logger = logger;
+  }
+
+  @Override
+  public void close() throws IOException {
+    this.logger.info("closing");
+  }
+
+  @Override
+  public void put(String path, String json) {
+    this.logger.info("PUT " + path + " " + json);
+  }
+
+  @Override
+  public void post(String path, String json) {
+    this.logger.info("POST " + path + (json == null ? "" : " " + json));
+  }
+}

src/main/java/marquez/client/MarquezHttp.java

@@ -21,6 +21,7 @@
 
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.core.type.TypeReference;
+import java.io.Closeable;
 import java.io.IOException;
 import java.net.URISyntaxException;
 import java.net.URL;
@@ -34,11 +35,12 @@
 import org.apache.http.client.methods.HttpPost;
 import org.apache.http.client.methods.HttpPut;
 import org.apache.http.entity.StringEntity;
+import org.apache.http.impl.client.CloseableHttpClient;
 import org.apache.http.impl.client.HttpClientBuilder;
 import org.apache.http.util.EntityUtils;
 
 @Slf4j
-class MarquezHttp {
+class MarquezHttp implements Closeable {
   private final HttpClient http;
 
   MarquezHttp(final HttpClient http) {
@@ -47,7 +49,8 @@ class MarquezHttp {
 
   static final MarquezHttp create(final MarquezClient.Version version) {
     final UserAgent userAgent = UserAgent.of(version);
-    final HttpClient http = HttpClientBuilder.create().setUserAgent(userAgent.getValue()).build();
+    final CloseableHttpClient http =
+        HttpClientBuilder.create().setUserAgent(userAgent.getValue()).build();
     return new MarquezHttp(http);
   }
 
@@ -122,6 +125,13 @@ private void throwOnHttpError(HttpResponse response) throws IOException {
     }
   }
 
+  @Override
+  public void close() throws IOException {
+    if (http instanceof Closeable) {
+      ((Closeable) http).close();
+    }
+  }
+
   @Value
   static class UserAgent {
     @Getter String value;

src/main/java/marquez/client/MarquezWriteOnlyClient.java

@@ -5,6 +5,7 @@
 import static marquez.client.models.RunState.FAILED;
 import static marquez.client.models.RunState.RUNNING;
 
+import java.io.Closeable;
 import java.time.Instant;
 import javax.annotation.Nullable;
 import marquez.client.models.DatasetMeta;
@@ -18,7 +19,7 @@
  * The contract of a write only client to instrument jobs actions. Can be taken synchronously or
  * asynchronously
  */
-public interface MarquezWriteOnlyClient {
+public interface MarquezWriteOnlyClient extends Closeable {
 
   public void createNamespace(String namespaceName, NamespaceMeta namespaceMeta);
 

src/main/java/marquez/client/MarquezWriteOnlyClientImpl.java

@@ -10,6 +10,7 @@
 import static marquez.client.MarquezPathV1.sourcePath;
 
 import com.google.common.collect.ImmutableMap;
+import java.io.IOException;
 import java.net.URLEncoder;
 import java.time.Instant;
 import java.util.Map;
@@ -80,4 +81,9 @@ public void markRunAs(String runId, RunState runState, Instant at) {
         at == null ? null : ImmutableMap.of("at", ISO_INSTANT.format(at));
     backend.post(path(runTransitionPath(runId, runState), queryParams));
   }
+
+  @Override
+  public void close() throws IOException {
+    backend.close();
+  }
 }

src/main/java/marquez/client/NullBackend.java

@@ -1,5 +1,7 @@
 package marquez.client;
 
+import java.io.IOException;
+
 /** A backend that does not do anything. */
 class NullBackend implements Backend {
 
@@ -8,4 +10,7 @@ public void put(String path, String json) {}
 
   @Override
   public void post(String path, String json) {}
+
+  @Override
+  public void close() throws IOException {}
 }