Influxdb with scala


I am trying to transfer this data to influxdb by pulling data from an address, but I could not transfer the data to influxdb even though I pulled the data. I am working in Scala language, service is running but data is not coming to influxdb. I want the monitoring EMC Recover Point

Could you help me with this please?

My Collector.xml code

<?xml version="1.0" encoding="UTF-8"?>
    <extractor name="rp">
    <output name="influxdb">
      <interval>10 min</interval>
      <extractor name="rp">

My Influxdb.scala code

package universalstoragecollector

import scala.util.Properties.propOrElse
import scala.util.Try
import java.util.concurrent.TimeUnit

import org.influxdb.{InfluxDB, InfluxDBFactory}
import org.influxdb.InfluxDB.ConsistencyLevel
import org.influxdb.dto.Point.Builder
import org.influxdb.dto.{BatchPoints, Point}

// Output to InfluxDB
class Influx(name: String, config: Map[String, String],
               sysConfig: Map[String, Option[String]])
  extends Output(name, config, sysConfig) with Logger {

  val loggerFile: String = propOrElse("USC_HOME", "") + "log/collector-error.log"

  // Common parameters
  val address: Option[String] =
    if (config.contains("address"))

  val port: Option[Int] =
    if (config.contains("port"))

  val dbname: Option[String] =
    if (config.contains("dbname"))

  var influxDB: InfluxDB = _

  // Validation by common parameters
  //def isValid: Boolean = config.contains("address") & config.contains("port") &

  // Connection initialization before data output
  def start(): Unit = {
    influxDB = InfluxDBFactory.connect("http://Influx-IP","username","pass")
    String databaseName = "foo"
    influxDB.query(new Query("CREATE DATABASE " + databaseName, databaseName))

  // Convert numbers in data from String type to Double type
  def parseDouble(s: String): Option[Double] = Try { s.toDouble }.toOption

  // Output data comes in 'msg' and 'data' structures
  def out(msg: Map[Int, (String, String)], timestamp: Long, data: Map[String, String]): Unit = {

    val header: Map[String, String] = (msg.values map (v => v._1 -> v._2)).toMap

    val batchPoints: BatchPoints = BatchPoints

    val point: Builder = Point.measurement(header("measurement"))
      .time(timestamp.toLong, TimeUnit.SECONDS)
      .tag("storage", sysConfig("name").get)
      .tag("class", sysConfig("class").get)

    if (sysConfig("type").isDefined) point.tag("type", sysConfig("type").get)

    (header.keySet - "measurement") foreach {k => point.tag(k, header(k))}

    data.foreach {p =>
      if (parseDouble(p._2).isDefined) point.addField(p._1, parseDouble(p._2).get)
      else point.addField(p._1, p._2)


  // Closing connection after data output
  def stop(): Unit = {

And my logs:
Wed Nov 30 09:24:36 UTC 2022 rp01, /clusters/statistics: java.util.NoSuchElementException: key not found: measurement
Wed Nov 30 09:24:37 UTC 2022 rp01, /rpas/statistics: java.util.NoSuchElementException: key not found: measurement
Wed Nov 30 09:24:37 UTC 2022 rp01, /groups: net.liftweb.json.MappingException: No usable value for innerSet
No usable value for groupCopiesInformation
No usable value for vmsInformation
Do not know how to convert JObject(List(JField(vmUID,JObject(List(JField(uuid,JString(500823dd-6738-b445-c705-7ad0d2b4e67c)), JField(virtualCenterUID,JObject(List(JField(uuid,JString(1a5e7115-0def-4796-b155-89d0c20e4ea5)))))))), JField(vmName,JString(YalcinMustafa_Management-win)), JField(vmBiosUuid,JString(4208652e-3aae-11bb-03a8-47d5ad33e1fc)))) into class java.lang.String

Thanks in advance

I am not sure the Grafana community forum is the best place to ask that question. Have you tried the InfluxDB forum?

Thanks for your reply

Is this xml from a result of an api or isnit a file on disk?

It’s file on disk. I created it

What is the source of the xml data?

You could also try using telegraf to consume this data into influxdb

  ## Commands array
  commands = [

Or just pure python, here are some samples for csv import to influxdb