Fetchdata doesn't take internal state into consideration.

[axos88]

When fetching data that goes beyond the last datapoint, the current internal state that contains the data for the last (yet unsaved) datapoint is never returned.

I'd expect the last datapoint to be 1000 in each case below (scala code)

import java.nio.file.{Paths, Files}

import org.rrd4j.{ConsolFun, DsType}
import org.rrd4j.core.{RrdDb, RrdDef}

val rrdPath = "scrrd.rrd"

Files.exists(Paths.get(rrdPath))

if (Files.exists(Paths.get(rrdPath))) {
      Files.delete(Paths.get(rrdPath))
}

Files.exists(Paths.get(rrdPath))


val rrdDef = new RrdDef(rrdPath, 1)

rrdDef.setVersion(2)
rrdDef.addDatasource("water", DsType.GAUGE, Long.MaxValue, 0, Double.NaN)

rrdDef.addArchive(ConsolFun.MAX, 0.99999, 1, 8)
rrdDef.addArchive(ConsolFun.MAX, 0.99999, 2, 5)
rrdDef.addArchive(ConsolFun.MAX, 0.99999, 4, 5)
rrdDef.addArchive(ConsolFun.MAX, 0.99999, 8, 8)

val rrdDb = new RrdDb(rrdDef)

val ts = 1500000000

val sample = rrdDb.createSample()

sample.setTime(ts)
sample.setValue("water", 500)
sample.update()

sample.setTime(ts + 1)
sample.setValue("water", 600)
sample.update()

sample.setTime(ts + 2)
sample.setValue("water", 700)
sample.update()

sample.setTime(ts + 3)
sample.setValue("water", 800)
sample.update()

sample.setTime(ts + 4)
sample.setValue("water", 900)
sample.update()

sample.setTime(ts + 5)
sample.setValue("water", 1000)
sample.update()


def getData(min: Long, max: Long) = {
      val fr = rrdDb.createFetchRequest(ConsolFun.MAX, min, max)
      val fd =fr.fetchData()

      fd.getTimestamps.zip(fd.getValues("water"))
}

println(rrdDb.dump())

getData(ts - 2, ts + 6)
getData(ts - 5, ts + 6)
getData(ts - 7, ts + 6)
getData(ts - 37, ts + 6)

getData returns:

res28: Array[(Long, Double)] = Array((1499999998,500.0), (1499999999,500.0), (1500000000,500.0), (1500000001,600.0), (1500000002,700.0), (1500000003,800.0), (1500000004,900.0), (1500000005,1000.0), (1500000006,NaN))
res29: Array[(Long, Double)] = Array((1499999994,NaN), (1499999996,500.0), (1499999998,500.0), (1500000000,500.0), (1500000002,700.0), (1500000004,900.0), (1500000006,NaN))
res30: Array[(Long, Double)] = Array((1499999992,500.0), (1499999996,500.0), (1500000000,500.0), (1500000004,900.0), (1500000008,NaN))
res31: Array[(Long, Double)] = Array((1499999960,500.0), (1499999968,500.0), (1499999976,500.0), (1499999984,500.0), (1499999992,500.0), (1500000000,500.0), (1500000008,NaN))

Db dump is:

== HEADER ==
signature:RRD4J, version 0.2 lastUpdateTime:1500000005 step:1 dsCount:1 arcCount:4
== DATASOURCE ==
DS:water:GAUGE:9223372036854775807:0.0:NaN
lastValue:1000.0 nanSeconds:0 accumValue:0.0
== ARCHIVE ==
RRA:MAX:0.99999:1:8
interval [1499999998, 1500000005]
accumValue:NaN nanSteps:0
Robin 6/8: +5.0000000000E02 +5.0000000000E02 +5.0000000000E02 +6.0000000000E02 +7.0000000000E02 +8.0000000000E02 +9.0000000000E02 +1.0000000000E03 
== ARCHIVE ==
RRA:MAX:0.99999:2:5
interval [1499999996, 1500000004]
accumValue:1000.0 nanSteps:0
Robin 3/5: +5.0000000000E02 +5.0000000000E02 +5.0000000000E02 +7.0000000000E02 +9.0000000000E02 
== ARCHIVE ==
RRA:MAX:0.99999:4:5
interval [1499999988, 1500000004]
accumValue:1000.0 nanSteps:0
Robin 2/5: +5.0000000000E02 +5.0000000000E02 +5.0000000000E02 +5.0000000000E02 +9.0000000000E02 
== ARCHIVE ==
RRA:MAX:0.99999:8:8
interval [1499999944, 1500000000]
accumValue:1000.0 nanSteps:0
Robin 1/8: +5.0000000000E02 +5.0000000000E02 +5.0000000000E02 +5.0000000000E02 +5.0000000000E02 +5.0000000000E02 +5.0000000000E02 +5.0000000000E02 

This makes all data points be delayed by the amount of time the step equates to before showing up on the output, which makes using rrd4j for live data very hard to use.

[axos88]

Current workaround:

def getData(min: Long, max: Long) = {
      val fr = rrdDb.createFetchRequest(ConsolFun.MAX, min, max)
      val fd =fr.fetchData()

      val ts = fd.getTimestamps
      val dsIndex = fd.getDsIndex("water")
      val values = fd.getValues(dsIndex)

      if (!fd.getMatchingArchive.getArcState(dsIndex).getAccumValue.isNaN) {
        //We have leftover data
        val lastts = fd.getLastTimestamp
        val idx = ts.indexOf(lastts)
        values(idx) = fd.getMatchingArchive.getArcState(dsIndex).getAccumValue
      }

      ts.zip(values)
}