From 41b9f0a245a24d20f20d3074869ea5d70f7c07ec Mon Sep 17 00:00:00 2001 From: arindahills <293051436+arindahills@users.noreply.github.com> Date: Tue, 30 Jun 2026 03:54:22 +0530 Subject: [PATCH 1/4] [Backend] Timedata.InfluxDB: implement resend (backfill) write write(edgeId, ResendDataNotification) was an empty stub, so data replayed by an Edge after a reconnect (the rrd4j resend pipeline) was silently dropped and gaps were never filled. Persist it via the same writeData path as timestamped data. Signed-off-by: arindahills --- .../backend/timedata/influx/TimedataInfluxDb.java | 14 +++++++++++++- 1 file changed, 13 insertions(+), 1 deletion(-) diff --git a/io.openems.backend.timedata.influx/src/io/openems/backend/timedata/influx/TimedataInfluxDb.java b/io.openems.backend.timedata.influx/src/io/openems/backend/timedata/influx/TimedataInfluxDb.java index ace656ded08..63ef1d08197 100644 --- a/io.openems.backend.timedata.influx/src/io/openems/backend/timedata/influx/TimedataInfluxDb.java +++ b/io.openems.backend.timedata.influx/src/io/openems/backend/timedata/influx/TimedataInfluxDb.java @@ -159,7 +159,19 @@ public void write(String edgeId, AggregatedDataNotification notification) { @Override public void write(String edgeId, ResendDataNotification data) { - // TODO Auto-generated method stub + if (this.config.isReadOnly()) { + return; + } + + // Write resent (backfilled) data to the default location, like timestamped data, + // so gaps from a reconnect are filled instead of silently dropped. + this.writeData(// + edgeId, // + data, // + (influxEdgeId, channel) -> { + this.timestampedChannelsForEdge.put(influxEdgeId, channel); + return true; + }); } private boolean isTimestampedChannel(int edgeId, String channel) { From 92be16d9afa5feaebdc539daebeda23a0f51e696 Mon Sep 17 00:00:00 2001 From: arindahills <293051436+arindahills@users.noreply.github.com> Date: Wed, 1 Jul 2026 10:28:01 +0530 Subject: [PATCH 2/4] [Edge] Timedata.Rrd4j: resend AVERAGE channels with correct field type queryResendData silently dropped AVERAGE-consolidated channels (Voltage, Current, Frequency, ActivePower) from the backfill: the 'adjustSeconds = arcStep - 300' shift was 0 for the 300s AVERAGE archive, so the narrow (<=300s) resend window left rrd4j slots NaN or just outside the window and only cumulated (3600s) channels survived a reconnect. Read each channel at the archive matching its consolidation and native step (mirroring queryHistoricData), and coerce the rrd4j double back to the channel's OpenemsType so integer/boolean channels are written with the same field type as live data - otherwise a float-vs-integer field conflict makes the timeseries backend drop the whole resent point. Completes resend/backfill support for a plain Timedata.InfluxDB backend. Signed-off-by: arindahills <293051436+arindahills@users.noreply.github.com> --- .../edge/timedata/rrd4j/Rrd4jReadHandler.java | 67 ++++++++++--------- .../timedata/rrd4j/Rrd4jReadHandlerTest.java | 30 +++++++++ 2 files changed, 67 insertions(+), 30 deletions(-) diff --git a/io.openems.edge.timedata.rrd4j/src/io/openems/edge/timedata/rrd4j/Rrd4jReadHandler.java b/io.openems.edge.timedata.rrd4j/src/io/openems/edge/timedata/rrd4j/Rrd4jReadHandler.java index 0be558ae28e..b9fe49d7c2a 100644 --- a/io.openems.edge.timedata.rrd4j/src/io/openems/edge/timedata/rrd4j/Rrd4jReadHandler.java +++ b/io.openems.edge.timedata.rrd4j/src/io/openems/edge/timedata/rrd4j/Rrd4jReadHandler.java @@ -34,6 +34,7 @@ import io.openems.common.exceptions.OpenemsException; import io.openems.common.timedata.Resolution; import io.openems.common.types.ChannelAddress; +import io.openems.common.types.OpenemsType; import io.openems.edge.common.channel.Channel; import io.openems.edge.common.component.ComponentManager; import io.openems.edge.timedata.api.Timeranges; @@ -223,42 +224,48 @@ public SortedMap> queryResendData(/ continue; } - for (int i = 0; i < database.getArcCount(); i++) { - final var archive = database.getArchive(i); - final var arcStep = archive.getArcStep(); + // Read at the channel's native storage resolution, mirroring + // queryHistoricData. The previous per-archive 'adjustSeconds = arcStep - 300' + // logic silently dropped AVERAGE channels (e.g. Voltage, Current, Frequency): + // for a 300s AVERAGE archive the shift was 0, so the narrow (<=300s) resend + // window left rrd4j slots either NaN or just outside the window and only + // cumulated (3600s) channels survived. Selecting the archive that matches the + // channel's consolidation and reading it at its own step fixes backfill for + // instantaneous channels while keeping cumulated channels hourly. + final var chDef = Rrd4jSupplier.getDsDefForChannel(channel.channelDoc().getUnit()); - final var adjustSeconds = arcStep - Rrd4jConstants.DEFAULT_STEP_SECONDS; + long resolution = Rrd4jConstants.DEFAULT_STEP_SECONDS; + for (int i = 0; i < database.getArcCount(); i++) { + if (database.getArchive(i).getConsolFun() == chDef.consolFun()) { + resolution = database.getArchive(i).getArcStep(); + break; + } + } - final var start = Math.max(fromTime - adjustSeconds, archive.getStartTime()); - final var stop = Math.min(toTime - adjustSeconds, archive.getEndTime()); - if (start > archive.getEndTime()) { + final var fetchData = database + .createFetchRequest(chDef.consolFun(), fromTime, toTime, resolution) // + .fetchData(); + final var result = Rrd4jSupplier.postProcessData(fetchData, resolution); + final var firstTimestamp = fetchData.getFirstTimestamp(); + for (int i = 0; i < result.length; i++) { + final var value = result[i]; + if (Double.isNaN(value)) { continue; } - if (stop < archive.getStartTime()) { + final var timestamp = firstTimestamp + (i * resolution); + if (timestamp < fromTime || timestamp > toTime) { continue; } - - final var fetchData = database.createFetchRequest(archive.getConsolFun(), start, stop, arcStep) // - .fetchData(); - - final var timestamps = fetchData.getTimestamps(); - final var values = fetchData.getValues()[0]; - for (int j = 0; j < values.length; j++) { - final var value = values[j]; - if (Double.isNaN(value)) { - continue; - } - final var timestamp = timestamps[j] + adjustSeconds; - - if (timestamp < fromTime // - || timestamp > toTime) { - continue; - } - - // return timestamps in milliseconds - resultMap.computeIfAbsent(timestamp * 1000, t -> new TreeMap<>()) // - .put(channelAddress, new JsonPrimitive(value)); - } + // coerce the rrd4j double back to the channel's type so integer/boolean + // channels are written with the same InfluxDB field type as live data; + // otherwise a float-vs-integer field conflict makes InfluxDB drop the point. + final JsonPrimitive jsonValue = switch (channel.getType()) { + case BOOLEAN, SHORT, INTEGER, LONG -> new JsonPrimitive(Long.valueOf(Math.round(value))); + default -> new JsonPrimitive(Double.valueOf(value)); + }; + // return timestamps in milliseconds + resultMap.computeIfAbsent(timestamp * 1000, t -> new TreeMap<>()) // + .put(channelAddress, jsonValue); } } catch (Exception e) { diff --git a/io.openems.edge.timedata.rrd4j/test/io/openems/edge/timedata/rrd4j/Rrd4jReadHandlerTest.java b/io.openems.edge.timedata.rrd4j/test/io/openems/edge/timedata/rrd4j/Rrd4jReadHandlerTest.java index c9135ed3ab4..ff82c0dd69f 100644 --- a/io.openems.edge.timedata.rrd4j/test/io/openems/edge/timedata/rrd4j/Rrd4jReadHandlerTest.java +++ b/io.openems.edge.timedata.rrd4j/test/io/openems/edge/timedata/rrd4j/Rrd4jReadHandlerTest.java @@ -163,6 +163,36 @@ public void testQueryHistoricDataWithResolution15minutes() throws Exception { ), this.query(new Resolution(15, ChronoUnit.MINUTES))); } + @Test + public void testQueryResendDataReturnsAverageChannel() throws Exception { + // AVERAGE-consolidated channel (Unit.WATT maps to ConsolFun.AVERAGE; see + // Rrd4jSupplier.getDsDefForChannel). The previous adjustSeconds = arcStep - 300 + // logic silently dropped these from the resend. This test asserts that the + // WATT samples written by setUp (0, 100, 200, ... every 5 min) are now returned + // by queryResendData over a window that spans several of those samples. + final var channelAddress = this.dummyComponent.channel(DummyComponent.ChannelId.DUMMY_CHANNEL).address(); + + final var result = this.readHandler.queryResendData(this.rrdbId, // + START.atZone(ZoneId.of("UTC")), // + START.plus(30, ChronoUnit.MINUTES).atZone(ZoneId.of("UTC")), // + Set.of(channelAddress), // + false); + + // KEY assertion: the resend must NOT be empty (it was empty under the old code + // for AVERAGE channels). + assertTrue("queryResendData must return AVERAGE-consolidated WATT data", !result.isEmpty()); + + // Each AVERAGE 5-minute sample: 0, 100, 200, 300, 400, 500 at START + n*5min. + // Keys are timestamps in MILLISECONDS (see Rrd4jReadHandler.queryResendData). + for (int i = 0; i < 6; i++) { + final var expectedTimestampMillis = START.plus(5 * i, ChronoUnit.MINUTES).toEpochMilli(); + final var row = result.get(expectedTimestampMillis); + assertTrue("Expected a resend row at " + expectedTimestampMillis + "ms", row != null); + assertEquals("Expected WATT value " + (i * 100) + " at sample " + i, // + new JsonPrimitive((double) (i * 100)), row.get(channelAddress)); + } + } + @Test public void testStreamRanges() throws Exception { final var utc = ZoneId.of("UTC"); From d34859cea8807c1a5cc8561ac47c6c2bd1ef6a68 Mon Sep 17 00:00:00 2001 From: tushabe Date: Wed, 1 Jul 2026 16:36:53 +0300 Subject: [PATCH 3/4] Test Influx resend backfill writes --- io.openems.backend.timedata.influx/bnd.bnd | 1 + .../timedata/influx/TimedataInfluxDbTest.java | 92 +++++++++++++++++++ 2 files changed, 93 insertions(+) create mode 100644 io.openems.backend.timedata.influx/test/io/openems/backend/timedata/influx/TimedataInfluxDbTest.java diff --git a/io.openems.backend.timedata.influx/bnd.bnd b/io.openems.backend.timedata.influx/bnd.bnd index 15b8a02e46e..72c8fb02de6 100644 --- a/io.openems.backend.timedata.influx/bnd.bnd +++ b/io.openems.backend.timedata.influx/bnd.bnd @@ -10,6 +10,7 @@ Bundle-Version: 1.0.0.${tstamp} io.openems.shared.influxdb,\ io.openems.wrapper.influxdb-client-core,\ io.openems.wrapper.influxdb-client-java,\ + io.openems.wrapper.influxdb-client-utils,\ -testpath: \ ${testpath} diff --git a/io.openems.backend.timedata.influx/test/io/openems/backend/timedata/influx/TimedataInfluxDbTest.java b/io.openems.backend.timedata.influx/test/io/openems/backend/timedata/influx/TimedataInfluxDbTest.java new file mode 100644 index 00000000000..247723a159d --- /dev/null +++ b/io.openems.backend.timedata.influx/test/io/openems/backend/timedata/influx/TimedataInfluxDbTest.java @@ -0,0 +1,92 @@ +package io.openems.backend.timedata.influx; + +import static io.openems.common.utils.ReflectionUtils.setAttributeViaReflection; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; +import static org.mockito.Mockito.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import org.junit.Test; +import org.mockito.ArgumentCaptor; + +import com.google.common.collect.TreeBasedTable; +import com.google.gson.JsonElement; +import com.google.gson.JsonPrimitive; +import com.influxdb.client.write.Point; + +import io.openems.common.jsonrpc.notification.AggregatedDataNotification; +import io.openems.common.jsonrpc.notification.ResendDataNotification; +import io.openems.common.oem.DummyOpenemsBackendOem; +import io.openems.shared.influxdb.InfluxConnector; + +public class TimedataInfluxDbTest { + + private static final String EDGE_ID = "edge0"; + private static final String CHANNEL = "ess0/ActivePower"; + private static final long TIMESTAMP = 1577836800000L; + + @Test + public void testWriteResendDataNotification() throws Exception { + final var influxConnector = mock(InfluxConnector.class); + final var sut = createTimedataInfluxDb(false, influxConnector); + + sut.write(EDGE_ID, createResendDataNotification()); + sut.write(EDGE_ID, createAggregatedDataNotification()); + + final var captor = ArgumentCaptor.forClass(Point.class); + verify(influxConnector, times(2)).write(captor.capture()); + + final var resendPoint = captor.getAllValues().get(0); + assertTrue(resendPoint.hasFields()); + assertEquals("0", resendPoint.getTags().get("edge")); + assertEquals(123L, resendPoint.getFields().get(CHANNEL)); + assertFalse(captor.getAllValues().get(1).hasFields()); + } + + @Test + public void testWriteResendDataNotificationInReadOnlyMode() throws Exception { + final var influxConnector = mock(InfluxConnector.class); + final var sut = createTimedataInfluxDb(true, influxConnector); + + sut.write(EDGE_ID, createResendDataNotification()); + + verify(influxConnector, never()).write(any(Point.class)); + } + + private static TimedataInfluxDb createTimedataInfluxDb(boolean isReadOnly, InfluxConnector influxConnector) + throws Exception { + final var sut = new TimedataInfluxDb(); + setAttributeViaReflection(sut, "config", createConfig(isReadOnly)); + setAttributeViaReflection(sut, "timeFilter", TimeFilter.from("", "")); + setAttributeViaReflection(sut, "channelFilter", ChannelFilter.from(new String[0], new String[0])); + setAttributeViaReflection(sut, "influxConnector", influxConnector); + setAttributeViaReflection(sut, "oem", new DummyOpenemsBackendOem()); + return sut; + } + + private static Config createConfig(boolean isReadOnly) { + final var config = mock(Config.class); + when(config.isReadOnly()).thenReturn(isReadOnly); + when(config.measurement()).thenReturn("data"); + when(config.blacklistedChannels()).thenReturn(new String[0]); + when(config.blacklistedChannelIds()).thenReturn(new String[0]); + return config; + } + + private static ResendDataNotification createResendDataNotification() { + final var data = TreeBasedTable.create(); + data.put(TIMESTAMP, CHANNEL, new JsonPrimitive(123)); + return new ResendDataNotification(data); + } + + private static AggregatedDataNotification createAggregatedDataNotification() { + final var data = TreeBasedTable.create(); + data.put(TIMESTAMP, CHANNEL, new JsonPrimitive(456)); + return new AggregatedDataNotification(data); + } +} From ca2ea7dbab2f5f10001008979953a48621b9a3b7 Mon Sep 17 00:00:00 2001 From: arindahills <293051436+arindahills@users.noreply.github.com> Date: Thu, 2 Jul 2026 02:37:30 +0530 Subject: [PATCH 4/4] [Edge] Timedata.Rrd4j: assert integer resend values serialize without a decimal gson JsonPrimitive#equals treats 100 and 100.0 as equal, so the value check did not actually verify the OpenemsType coercion. Assert the serialized form so the regression for the float-vs-integer InfluxDB field conflict is real. Signed-off-by: arindahills <293051436+arindahills@users.noreply.github.com> --- .../openems/edge/timedata/rrd4j/Rrd4jReadHandlerTest.java | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/io.openems.edge.timedata.rrd4j/test/io/openems/edge/timedata/rrd4j/Rrd4jReadHandlerTest.java b/io.openems.edge.timedata.rrd4j/test/io/openems/edge/timedata/rrd4j/Rrd4jReadHandlerTest.java index ff82c0dd69f..9f09aeeca2e 100644 --- a/io.openems.edge.timedata.rrd4j/test/io/openems/edge/timedata/rrd4j/Rrd4jReadHandlerTest.java +++ b/io.openems.edge.timedata.rrd4j/test/io/openems/edge/timedata/rrd4j/Rrd4jReadHandlerTest.java @@ -190,6 +190,14 @@ public void testQueryResendDataReturnsAverageChannel() throws Exception { assertTrue("Expected a resend row at " + expectedTimestampMillis + "ms", row != null); assertEquals("Expected WATT value " + (i * 100) + " at sample " + i, // new JsonPrimitive((double) (i * 100)), row.get(channelAddress)); + + // DUMMY_CHANNEL is OpenemsType.INTEGER, so the resend MUST emit an integer + // JSON (no decimal) to match the field type live data writes; otherwise the + // timeseries backend drops the whole point on a float-vs-integer conflict. + // gson's JsonPrimitive#equals treats 100 and 100.0 as equal, so the check + // above passes either way - assert the serialized form explicitly. + assertEquals("INTEGER channel must resend as an integer, not a float", // + String.valueOf(i * 100), row.get(channelAddress).toString()); } }