diff --git a/dagger-core/src/main/java/com/gotocompany/dagger/core/sink/influx/InfluxDBWriter.java b/dagger-core/src/main/java/com/gotocompany/dagger/core/sink/influx/InfluxDBWriter.java index f130e71c2..ca59002fc 100644 --- a/dagger-core/src/main/java/com/gotocompany/dagger/core/sink/influx/InfluxDBWriter.java +++ b/dagger-core/src/main/java/com/gotocompany/dagger/core/sink/influx/InfluxDBWriter.java @@ -39,7 +39,9 @@ public class InfluxDBWriter implements SinkWriter { 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); diff --git a/dagger-core/src/main/java/com/gotocompany/dagger/core/sink/influx/InfluxSinkOverrides.java b/dagger-core/src/main/java/com/gotocompany/dagger/core/sink/influx/InfluxSinkOverrides.java index 7cfa6bb52..fc7e384cd 100644 --- a/dagger-core/src/main/java/com/gotocompany/dagger/core/sink/influx/InfluxSinkOverrides.java +++ b/dagger-core/src/main/java/com/gotocompany/dagger/core/sink/influx/InfluxSinkOverrides.java @@ -10,9 +10,10 @@ * * - * 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); } } } diff --git a/dagger-core/src/main/java/com/gotocompany/dagger/core/utils/Constants.java b/dagger-core/src/main/java/com/gotocompany/dagger/core/utils/Constants.java index 2e658730d..b7da36237 100644 --- a/dagger-core/src/main/java/com/gotocompany/dagger/core/utils/Constants.java +++ b/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"; diff --git a/dagger-core/src/test/java/com/gotocompany/dagger/core/sink/influx/InfluxDBWriterTest.java b/dagger-core/src/test/java/com/gotocompany/dagger/core/sink/influx/InfluxDBWriterTest.java index 3411b322e..22f6f8753 100644 --- a/dagger-core/src/test/java/com/gotocompany/dagger/core/sink/influx/InfluxDBWriterTest.java +++ b/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 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; diff --git a/docs/docs/reference/configuration.md b/docs/docs/reference/configuration.md index 8c632c2c9..4139dada4 100644 --- a/docs/docs/reference/configuration.md +++ b/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). diff --git a/version.txt b/version.txt index e01e0ddd8..43c2417ca 100644 --- a/version.txt +++ b/version.txt @@ -1 +1 @@ -0.12.4 +0.12.5