feat/override-influx-dbName - Allow custom jobs to write data to multiple influx databases residing in same Influx server

dagger-core/src/main/java/com/gotocompany/dagger/core/sink/influx/InfluxDBWriter.java

@@ -39,7 +39,9 @@ public class InfluxDBWriter implements SinkWriter<Row, Void, Void> {
 
     public InfluxDBWriter(Configuration configuration, InfluxDB influxDB, String[] columnNames, ErrorHandler errorHandler,
                           ErrorReporter errorReporter, InfluxSinkOverrides overrides) {
-        databaseName = configuration.getString(Constants.SINK_INFLUX_DB_NAME_KEY, Constants.SINK_INFLUX_DB_NAME_DEFAULT);
+        databaseName = overrides.hasDatabaseName()
+                ? overrides.getDatabaseName()
+                : configuration.getString(Constants.SINK_INFLUX_DB_NAME_KEY, Constants.SINK_INFLUX_DB_NAME_DEFAULT);
         retentionPolicy = overrides.hasRetentionPolicy()
                 ? overrides.getRetentionPolicy()
                 : configuration.getString(Constants.SINK_INFLUX_RETENTION_POLICY_KEY, Constants.SINK_INFLUX_RETENTION_POLICY_DEFAULT);

dagger-core/src/main/java/com/gotocompany/dagger/core/sink/influx/InfluxSinkOverrides.java

@@ -10,9 +10,10 @@
  * <ul>
  *     <li>measurement name (overrides {@code SINK_INFLUX_MEASUREMENT_NAME})</li>
  *     <li>retention policy (overrides {@code SINK_INFLUX_RETENTION_POLICY})</li>
+ *     <li>database name (overrides {@code SINK_INFLUX_DB_NAME})</li>
  * </ul>
  *
- * Either field may be {@code null} or blank, which means "fall back to the
+ * Any field may be {@code null} or blank, which means "fall back to the
  * value resolved from configuration". Using this object instead of multiple
  * positional {@code String} parameters keeps sink construction readable as
  * the number of overrides grows.
@@ -22,6 +23,7 @@
  * InfluxSinkOverrides overrides = InfluxSinkOverrides.builder()
  *         .measurementName(measurements[i])
  *         .retentionPolicy(retentionPolicies[i])
+ *         .databaseName(databaseNames[i])
  *         .build();
  * Sink sink = sinkOrchestrator.getSink(configuration, columnNames,
  *         stencilClientOrchestrator, daggerStatsDReporter, influxOverrides);
@@ -31,14 +33,16 @@ public final class InfluxSinkOverrides implements Serializable {
 
     private static final long serialVersionUID = 1L;
 
-    private static final InfluxSinkOverrides NONE = new InfluxSinkOverrides(null, null);
+    private static final InfluxSinkOverrides NONE = new InfluxSinkOverrides(null, null, null);
 
     private final String measurementName;
     private final String retentionPolicy;
+    private final String databaseName;
 
-    private InfluxSinkOverrides(String measurementName, String retentionPolicy) {
+    private InfluxSinkOverrides(String measurementName, String retentionPolicy, String databaseName) {
         this.measurementName = measurementName;
         this.retentionPolicy = retentionPolicy;
+        this.databaseName = databaseName;
     }
 
     /** Returns an instance that applies no overrides (configuration values are used). */
@@ -47,15 +51,19 @@ public static InfluxSinkOverrides none() {
     }
 
     public static InfluxSinkOverrides withMeasurementName(String measurementName) {
-        return new InfluxSinkOverrides(measurementName, null);
+        return new InfluxSinkOverrides(measurementName, null, null);
     }
 
     public static InfluxSinkOverrides withRetentionPolicy(String retentionPolicy) {
-        return new InfluxSinkOverrides(null, retentionPolicy);
+        return new InfluxSinkOverrides(null, retentionPolicy, null);
     }
 
-    public static InfluxSinkOverrides of(String measurementName, String retentionPolicy) {
-        return new InfluxSinkOverrides(measurementName, retentionPolicy);
+    public static InfluxSinkOverrides withDatabaseName(String databaseName) {
+        return new InfluxSinkOverrides(null, null, databaseName);
+    }
+
+    public static InfluxSinkOverrides of(String measurementName, String retentionPolicy, String databaseName) {
+        return new InfluxSinkOverrides(measurementName, retentionPolicy, databaseName);
     }
 
     public static Builder builder() {
@@ -72,6 +80,11 @@ public String getRetentionPolicy() {
         return retentionPolicy;
     }
 
+    /** May be {@code null} or empty, meaning "use the value from configuration". */
+    public String getDatabaseName() {
+        return databaseName;
+    }
+
     public boolean hasMeasurementName() {
         return !Strings.isNullOrEmpty(measurementName);
     }
@@ -80,6 +93,10 @@ public boolean hasRetentionPolicy() {
         return !Strings.isNullOrEmpty(retentionPolicy);
     }
 
+    public boolean hasDatabaseName() {
+        return !Strings.isNullOrEmpty(databaseName);
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) {
@@ -90,24 +107,27 @@ public boolean equals(Object o) {
         }
         InfluxSinkOverrides that = (InfluxSinkOverrides) o;
         return Objects.equals(measurementName, that.measurementName)
-                && Objects.equals(retentionPolicy, that.retentionPolicy);
+                && Objects.equals(retentionPolicy, that.retentionPolicy)
+                && Objects.equals(databaseName, that.databaseName);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(measurementName, retentionPolicy);
+        return Objects.hash(measurementName, retentionPolicy, databaseName);
     }
 
     @Override
     public String toString() {
         return "InfluxSinkOverrides{measurementName='" + measurementName
-                + "', retentionPolicy='" + retentionPolicy + "'}";
+                + "', retentionPolicy='" + retentionPolicy
+                + "', databaseName='" + databaseName + "'}";
     }
 
     /** Fluent builder for {@link InfluxSinkOverrides}. */
     public static final class Builder {
         private String measurementName;
         private String retentionPolicy;
+        private String databaseName;
 
         private Builder() {
         }
@@ -122,11 +142,16 @@ public Builder retentionPolicy(String policy) {
             return this;
         }
 
+        public Builder databaseName(String name) {
+            this.databaseName = name;
+            return this;
+        }
+
         public InfluxSinkOverrides build() {
-            if (measurementName == null && retentionPolicy == null) {
+            if (measurementName == null && retentionPolicy == null && databaseName == null) {
                 return NONE;
             }
-            return new InfluxSinkOverrides(measurementName, retentionPolicy);
+            return new InfluxSinkOverrides(measurementName, retentionPolicy, databaseName);
         }
     }
 }

dagger-core/src/main/java/com/gotocompany/dagger/core/utils/Constants.java

@@ -169,6 +169,12 @@ public class Constants {
     // configuration without changing the code.
     public static final String SINK_INFLUX_RETENTION_POLICY_LIST_KEY = "SINK_INFLUX_RETENTION_POLICY_LIST";
 
+    // A custom job can use this configuration to get all Influx database names as a list
+    // (positionally aligned with SINK_INFLUX_MEASUREMENTS_LIST_KEY) and configure them in the
+    // job builder pipeline accordingly. This allows database names to be updated through
+    // configuration without changing the code.
+    public static final String SINK_INFLUX_DB_NAME_LIST_KEY = "SINK_INFLUX_DB_NAME_LIST";
+
     public static final String SINK_INFLUX_URL_KEY = "SINK_INFLUX_URL";
     public static final String SINK_INFLUX_URL_DEFAULT = "";
     public static final String SINK_INFLUX_USERNAME_KEY = "SINK_INFLUX_USERNAME";

dagger-core/src/test/java/com/gotocompany/dagger/core/sink/influx/InfluxDBWriterTest.java

@@ -177,15 +177,15 @@ public void shouldWriteToInfluxWithOverrideRetentionPolicy() throws Exception {
     }
 
     @Test
-    public void shouldWriteToInfluxWithOverrideMeasurementAndRetentionPolicy() throws Exception {
+    public void shouldWriteToInfluxWithOverrideMeasurementRetentionPolicyAndDatabase() throws Exception {
         Row row = new Row(1);
         row.setField(0, "some field");
         InfluxDBWriter influxDBWriter = new InfluxDBWriter(configuration, influxDb, new String[]{"some_field_name"},
-                errorHandler, errorReporter, InfluxSinkOverrides.of("override_measurement", "override_policy"));
+                errorHandler, errorReporter, InfluxSinkOverrides.of("override_measurement", "override_policy", "override_db"));
         influxDBWriter.write(row, context);
 
         ArgumentCaptor<Point> pointArg = ArgumentCaptor.forClass(Point.class);
-        verify(influxDb).write(eq("dagger_test"), eq("override_policy"), pointArg.capture());
+        verify(influxDb).write(eq("override_db"), eq("override_policy"), pointArg.capture());
         // measurement is in line protocol prefix
         assertTrue(pointArg.getValue().lineProtocol().startsWith("override_measurement"));
     }
@@ -201,6 +201,28 @@ public void shouldFallBackToConfigRetentionPolicyWhenOverrideIsBlank() throws Ex
         verify(influxDb).write(eq("dagger_test"), eq("two_day_policy"), any());
     }
 
+    @Test
+    public void shouldWriteToInfluxWithOverrideDatabaseName() throws Exception {
+        Row row = new Row(1);
+        row.setField(0, "some field");
+        InfluxDBWriter influxDBWriter = new InfluxDBWriter(configuration, influxDb, new String[]{"some_field_name"},
+                errorHandler, errorReporter, InfluxSinkOverrides.withDatabaseName("override_db"));
+        influxDBWriter.write(row, context);
+
+        verify(influxDb).write(eq("override_db"), eq("two_day_policy"), any());
+    }
+
+    @Test
+    public void shouldFallBackToConfigDatabaseNameWhenOverrideIsBlank() throws Exception {
+        Row row = new Row(1);
+        row.setField(0, "some field");
+        InfluxDBWriter influxDBWriter = new InfluxDBWriter(configuration, influxDb, new String[]{"some_field_name"},
+                errorHandler, errorReporter, InfluxSinkOverrides.withDatabaseName(""));
+        influxDBWriter.write(row, context);
+
+        verify(influxDb).write(eq("dagger_test"), eq("two_day_policy"), any());
+    }
+
     @Test
     public void shouldNotWriteNullColumnsInRowToInfluxAsfields() throws Exception {
         final int numberOfRows = 3;

docs/docs/reference/configuration.md

@@ -360,9 +360,16 @@ Defines the password to connect to InfluxDB.
 
 Defines the InfluxDB database name.
 
-* Example value: `DAGGER`
+* Example value: `default_db`
 * Type: `required`
 
+#### `SINK_INFLUX_DB_NAME_LIST`
+
+Defines a comma-separated list of InfluxDB database names for custom jobs that write to multiple databases. The values are positionally aligned with `SINK_INFLUX_MEASUREMENTS_LIST` and let a custom job override the database name per sink through configuration instead of code. When unset, the sink uses `SINK_INFLUX_DB_NAME`.
+
+* Example value: `default_db, ssd_db`
+* Type: `optional`
+
 #### `SINK_INFLUX_MEASUREMENT_NAME`
 
 Defines the InfluxDB measurement name. Find more details on using this config [here](https://docs.influxdata.com/influxdb/v1.8/concepts/glossary/#measurement).

version.txt

@@ -1 +1 @@
-0.12.4
+0.12.5