Java Loggers

java/core/src/main/java/com/whylogs/api/logger/Logger.java

@@ -1,3 +1,85 @@
 package com.whylogs.api.logger;
 
-public class Logger {}
+import com.whylogs.api.logger.resultsets.ProfileResultSet;
+import com.whylogs.api.logger.resultsets.ResultSet;
+import com.whylogs.api.writer.Writer;
+import com.whylogs.api.writer.WritersRegistry;
+import com.whylogs.core.DatasetProfile;
+import com.whylogs.core.schemas.DatasetSchema;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Map;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.ToString;
+
+@NoArgsConstructor
+@Getter
+@EqualsAndHashCode
+@ToString
+public abstract class Logger implements AutoCloseable {
+  private boolean isClosed = false;
+  private DatasetSchema schema = new DatasetSchema();
+  private ArrayList<Writer> writers = new ArrayList<>();
+
+  public Logger(DatasetSchema schema) {
+    this.schema = schema;
+  }
+
+  public <T extends Writer> void checkWriter(T Writer) {
+    // Checks if a writer is configured correctly for this class
+    // Question: why is this empty but not an abstract?
+  }
+
+  public void appendWriter(String name) {
+    if (name == null || name.isEmpty()) {
+      throw new IllegalArgumentException("Writer name cannot be empty");
+    }
+
+    Writer writer = WritersRegistry.get(name);
+    if (writer == null) {
+      throw new IllegalArgumentException("Writer " + name + " is not registered");
+    }
+
+    appendWriter(writer);
+  }
+
+  public void appendWriter(Writer writer) {
+    if (writer == null) {
+      throw new IllegalArgumentException("Writer cannot be null");
+    }
+
+    checkWriter(writer);
+    writers.add(writer);
+  }
+
+  protected abstract ArrayList<DatasetProfile> getMatchingProfiles(Object data);
+
+  protected abstract <O> ArrayList<DatasetProfile> getMatchingProfiles(Map<String, O> data);
+
+  @Override
+  public void close() {
+    isClosed = true;
+  }
+
+  public ResultSet log(HashMap<String, Object> data) {
+    // What type of data is the object? Right now we don't process that in track.
+    if (isClosed) {
+      throw new IllegalStateException("Logger is closed");
+    } else if (data == null) {
+      throw new IllegalArgumentException("Data cannot be null");
+    }
+
+    // TODO: implement segment processing here
+
+    ArrayList<DatasetProfile> profiles = getMatchingProfiles(data);
+    for (DatasetProfile profile : profiles) {
+      profile.track(data);
+    }
+
+    // Question: Why does this only return the first profile? IS this
+    // getting ready for multiple profiles later on?
+    return new ProfileResultSet(profiles.get(0));
+  }
+}

java/core/src/main/java/com/whylogs/api/logger/TransientLogger.java

@@ -1,3 +1,32 @@
 package com.whylogs.api.logger;
 
-public class TransientLogger {}
+import com.whylogs.core.DatasetProfile;
+import com.whylogs.core.schemas.DatasetSchema;
+import java.util.ArrayList;
+import java.util.Map;
+import lombok.*;
+
+@NoArgsConstructor
+@Getter
+@EqualsAndHashCode(callSuper = false)
+@ToString
+public class TransientLogger extends Logger {
+  public TransientLogger(DatasetSchema schema) {
+    super(schema);
+  }
+
+  @Override
+  protected ArrayList<DatasetProfile> getMatchingProfiles(Object data) {
+    // In this case, we don't have any profiles to match against
+    ArrayList<DatasetProfile> profiles = new ArrayList<>();
+    DatasetProfile profile = new DatasetProfile(getSchema());
+    profiles.add(profile);
+    return profiles;
+  }
+
+  @Override
+  protected <O> ArrayList<DatasetProfile> getMatchingProfiles(Map<String, O> data) {
+    // In this case, we don't have any profiles to match against
+    return getMatchingProfiles((Object) data);
+  }
+}

java/core/src/main/java/com/whylogs/api/logger/resultsets/ProfileResultSet.java

@@ -12,7 +12,7 @@
 public class ProfileResultSet extends ResultSet {
   @NonNull private final DatasetProfile profile;
 
-  public ProfileResultSet(DatasetProfile profile) {
+  public ProfileResultSet(@NonNull DatasetProfile profile) {
     super();
     this.profile = profile;
   }

java/core/src/main/java/com/whylogs/api/logger/resultsets/ResultSet.java

@@ -28,7 +28,6 @@ public abstract class ResultSet {
   public abstract Optional<DatasetProfile> profile();
 
   // TODO: Come back for ModelPerformanceMetrics
-
   public void addMetric(String name, Metric<?> metric) throws Error {
     DatasetProfile profile =
         this.profile()

java/core/src/main/java/com/whylogs/api/logger/rollingLogger/Scheduler.java

@@ -0,0 +1,61 @@
+package com.whylogs.api.logger.rollingLogger;
+
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.ToString;
+
+@Getter
+@EqualsAndHashCode
+@ToString
+public class Scheduler {
+  // Multithreading schedule.
+  // Schedule a function to be called repeatedly based on a schedule
+  private ScheduledExecutorService scheduledService;
+  private float initial;
+  private boolean ranInitial = false;
+  private float interval;
+  private Runnable func;
+  private boolean isRunning = false;
+  private String[] args;
+  // TODO: figure out args an dkwards
+
+  public Scheduler(float initial, float interval, Runnable func, String[] args) {
+    this.initial = initial;
+    this.interval = interval;
+    this.func = func;
+    this.args = args;
+    this.start();
+  }
+
+  private void run() {
+    // TODO: Looking at this I think this is wrong to have lines 35 & 36
+    this.isRunning = false;
+    this.start(); // Question: why do we need to start again?
+    this.func.run(); // TODO: figure out args and kwargs
+  }
+
+  public void start() {
+    if (this.isRunning) {
+      return;
+    }
+
+    float initial = 0;
+    if (!this.ranInitial) {
+      initial = this.getInitial();
+      this.ranInitial = true;
+    }
+
+    this.scheduledService = Executors.newSingleThreadScheduledExecutor();
+    this.scheduledService.scheduleAtFixedRate(
+        this::run, (long) initial, (long) this.interval, TimeUnit.SECONDS);
+    this.isRunning = true;
+  }
+
+  public void stop() {
+    this.scheduledService.shutdown();
+    this.isRunning = false;
+  }
+}

java/core/src/main/java/com/whylogs/api/logger/rollingLogger/TimedRollingLogger.java

@@ -0,0 +1,173 @@
+package com.whylogs.api.logger.rollingLogger;
+
+import com.whylogs.api.logger.Logger;
+import com.whylogs.api.writer.Writer;
+import com.whylogs.core.DatasetProfile;
+import com.whylogs.core.schemas.DatasetSchema;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Map;
+import java.util.concurrent.Callable;
+
+public class TimedRollingLogger extends Logger implements AutoCloseable {
+  // A rolling logger that continuously rotates files based on time
+  private DatasetSchema schema;
+  private String baseName;
+  private String fileExtension;
+  private int interval;
+  private Character when = 'H'; // TODO: Make the Literals of S M H D
+  private boolean utc = false;
+  private boolean align = true;
+  private boolean skipEmpty = false;
+  private String suffix;
+
+  private DatasetProfile currentProfile;
+  private Callable<Writer> callback; // TODO: this isn't the write signature
+  private Scheduler scheduler;
+  private int currentBatchTimestamp;
+
+  // TODO: callback: Optional[Callable[[Writer, DatasetProfileView, str], None]]
+  public TimedRollingLogger(
+      DatasetSchema schema, String baseName, String fileExtension, int interval) {
+    this(schema, baseName, fileExtension, interval, 'H', false, true, false);
+  }
+
+  public TimedRollingLogger(
+      DatasetSchema schema, String baseName, String fileExtension, int interval, Character when) {
+    this(schema, baseName, fileExtension, interval, when, false, true, false);
+  }
+
+  public TimedRollingLogger(
+      DatasetSchema schema,
+      String baseName,
+      String fileExtension,
+      int interval,
+      Character when,
+      boolean utc,
+      boolean align,
+      boolean skipEmpty) {
+    super(schema);
+
+    this.schema = schema;
+    this.baseName = baseName;
+    this.fileExtension = fileExtension;
+    this.interval = interval;
+    this.when = Character.toUpperCase(when);
+    this.utc = utc;
+    this.align = align;
+    this.skipEmpty = skipEmpty;
+
+    if (this.baseName == null || this.baseName.isEmpty()) {
+      this.baseName = "profile";
+    }
+    if (this.fileExtension == null || this.fileExtension.isEmpty()) {
+      this.fileExtension = ".bin"; // TODO: should we make this .whylogs?
+    }
+
+    switch (this.when) {
+      case 'S':
+        this.interval = 1; // one second
+        this.suffix = "%Y-%m-%d_%H-%M-%S";
+        break;
+      case 'M':
+        this.interval = 60; // one minute
+        this.suffix = "%Y-%m-%d_%H-%M";
+        break;
+      case 'H':
+        this.interval = 60 * 60; // one hour
+        this.suffix = "%Y-%m-%d_%H";
+        break;
+      case 'D':
+        this.interval = 60 * 60 * 24; // one day
+        this.suffix = "%Y-%m-%d";
+        break;
+      default:
+        throw new IllegalArgumentException(
+            "Invalid value for when: " + this.when + ". Must be S, M, H, or D");
+    }
+
+    this.interval = this.interval * interval; // / multiply by units requested
+    this.utc = utc;
+
+    Instant currentTime = Instant.now();
+    this.currentBatchTimestamp = this.computeCurrentBatchTimestamp(currentTime.getEpochSecond());
+    this.currentProfile = new DatasetProfile(schema, currentTime, currentTime);
+    int initialRunAfter =
+        (this.currentBatchTimestamp + this.interval) - (int) currentTime.getEpochSecond();
+    if (initialRunAfter < 0) {
+      // TODO: Add logging error as this shouldn't happen
+      initialRunAfter = this.interval;
+    }
+
+    this.scheduler = new Scheduler(initialRunAfter, this.interval, this::doRollover, null);
+    this.scheduler.start();
+
+    // autocloseable closes at end
+  }
+
+  private int computeCurrentBatchTimestamp(long nowEpoch) {
+    int roundedNow = (int) nowEpoch; // rounds by going from an long to a int (truncates)
+    if (this.align) {
+      return (Math.floorDiv((roundedNow - 1), this.interval)) * this.interval + this.interval;
+    }
+    return roundedNow;
+  }
+
+  public void checkWriter(Writer writer) {
+    writer.check_interval(this.interval);
+  }
+
+  private ArrayList<DatasetProfile> getMatchingProfiles() {
+    ArrayList<DatasetProfile> matchingProfiles = new ArrayList<>();
+    matchingProfiles.add(this.currentProfile);
+    return matchingProfiles;
+  }
+
+  @Override
+  protected ArrayList<DatasetProfile> getMatchingProfiles(Object data) {
+    return this.getMatchingProfiles();
+  }
+
+  @Override
+  protected <O> ArrayList<DatasetProfile> getMatchingProfiles(Map<String, O> data) {
+    return this.getMatchingProfiles();
+  }
+
+  private void doRollover() {
+    if (this.isClosed()) {
+      return;
+    }
+
+    DatasetProfile oldProfile = this.currentProfile;
+    Instant currentTime = Instant.now();
+    this.currentBatchTimestamp = this.computeCurrentBatchTimestamp(currentTime.getEpochSecond());
+    this.currentProfile = new DatasetProfile(schema, currentTime, currentTime);
+
+    this.flush(oldProfile);
+  }
+
+  private void flush(DatasetProfile profile) {
+    if (profile == null) {
+      return;
+    } else if (this.skipEmpty && profile.isEmpty()) {
+      // set logger logger.debug("skip_empty is set. Skipping empty profiles")
+      return;
+    }
+
+    // get time to get name
+    String timedFileName = this.baseName + "_" + this.currentBatchTimestamp + this.fileExtension;
+
+    // Sleep while the profile is active?
+    // TODO: this is where we call the store list.write
+    // TODO: go through through the writers
+  }
+
+  public void close() {
+    // TODO log that we are closing the writer
+    if (!this.isClosed()) {
+      // Autoclose handles the isCLosed()
+      this.scheduler.stop();
+      this.flush(this.currentProfile);
+    }
+  }
+}

java/core/src/main/java/com/whylogs/api/writer/Writable.java

@@ -0,0 +1,22 @@
+package com.whylogs.api.writer;
+
+import java.io.File;
+import java.io.FileWriter;
+
+public interface Writable {
+
+  static FileWriter safeOpenWrite(String path) {
+    // Open 'path' for writing, creating any parent directories as needed
+    File file = new File(path);
+    FileWriter writer = null;
+    try {
+      writer = new FileWriter(file, true);
+    } catch (Exception e) {
+      System.out.println("Error: " + e);
+      e.printStackTrace();
+    }
+
+    // this close happens latter on
+    return writer;
+  }
+}