From 64091f8b10225940c56516ac67f5728a1b07b00e Mon Sep 17 00:00:00 2001 From: rajuGT Date: Wed, 15 Jul 2026 20:57:22 +0530 Subject: [PATCH] feat/override-influx-dbName - Allow custom jobs to write data to multiple influx databases residing in same Influx Server - This allows to seggregate critical metrics vs non-critical, or - or in one of our usecase, the _series values are dropped/leaked due to the nature of the data and been recreated every day causing increasing db memory. So the new approach is to keep this high cardinality series data into new database. With this maybe the new database can be re-indexed every night or recreate the db / based on one's usecase they can take the call. --- .../core/sink/influx/InfluxDBWriter.java | 4 +- .../core/sink/influx/InfluxSinkOverrides.java | 49 ++++++++++++++----- .../dagger/core/utils/Constants.java | 6 +++ .../core/sink/influx/InfluxDBWriterTest.java | 28 +++++++++-- docs/docs/reference/configuration.md | 9 +++- version.txt | 2 +- 6 files changed, 80 insertions(+), 18 deletions(-) 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