Supported versions of influxdb and future support for API changes

[amhankin]

What versions of influxdb are currently supported by the code defined in 'artiq/artiq/frontend/artiq_influxdb.py'? I had difficulty finding documentation regarding this point. Is it possible to include comments in the code and/or the documentation that mention the supported versions?

Finally, what are m-labs' plans for supporting future versions of influxdb if the API for communicating with the database changes in the future (for the stable release of influxdb)?

[sbourdeauducq]

We don't track changes to influxdb. I'd recommend using the version that was current when the ARTIQ code was committed and test it, you can then submit a documentation patch. If you need support for other influxdb versions, we can do it for you on a contract basis (we currently have no other plans for this).

[jordens]

The influx wire format is pretty stable. I'd guess it'll work fine with the latest releases. Last I checked it was still working. Please give it a try and report back.

[amhankin]

It may take a few weeks, but I will let you know what happens with the latest stable version of influxdb v1.3 after testing it.

[sbourdeauducq]

I thought you already tested it and had a problem ...

[amhankin]

I am currently upgrading from an old version of Influxdb (too outdated to work with the ARTIQ driver) and wanted to know whether or not the latest stable version is supported. Since we are making the change now and considering using it regularly, I would like to understand expectations regarding support for this feature in the future.

[sbourdeauducq]

@amhankin Does everything work as expected?

[amhankin]

Sorry, I have not had the free time to verify if this works yet.

[jbqubit]

Tested using the following configuration

• ARTIQ 4.0.dev
• InfluxDB 1.3.7
• Grafana 4.6.2

import random
import time


from artiq.experiment import *


class FloppingF(EnvExperiment):
	"""InfluxExample"""

	def run(self):
		while True:
			x = 1*random.random()
			# InfluxDB bridge processes datasets with persist=True
			self.set_dataset("rand.x1", x*1, persist=True, save=False, broadcast=False)
			self.set_dataset("rand.x2", x*2, persist=True, save=False, broadcast=False)
			self.set_dataset("rand.x3", x*3, persist=True, save=False, broadcast=False)
			time.sleep(1)

$ cat influxdb_patterns.cfg 
-*
+rand.x1
+rand.x3
$ 
$ artiq_influxdb --database a2
ERROR<2>:asyncio:Unclosed client session
client_session: <aiohttp.client.ClientSession object at 0x7fa47770d2e8>
ERROR<2>:asyncio:Unclosed client session
client_session: <aiohttp.client.ClientSession object at 0x7fa477b9a668>
ERROR<2>:asyncio:Unclosed client session
client_session: <aiohttp.client.ClientSession object at 0x7fa477b9a668>
ERROR<2>:asyncio:Unclosed client session
client_session: <aiohttp.client.ClientSession object at 0x7fa47770d198>
ERROR<2>:asyncio:Unclosed client session

Setup Grafana to plot the following in the a2 InfluxDB database.

SELECT mean("float") FROM "lab" WHERE ("dataset" = 'rand.x1') AND $timeFilter GROUP BY time($__interval) fill(null)
SELECT mean("float") FROM "lab" WHERE ("dataset" = 'rand.x2') AND $timeFilter GROUP BY time($__interval) fill(null)
SELECT mean("float") FROM "lab" WHERE ("dataset" = 'rand.x3') AND $timeFilter GROUP BY time($__interval) fill(null)

Resulting grafana plot.

Comments

• Functionality seems fine with these versions. Response to strings in influxdb_patterns.cfg is as expected.
• artiq_influxdb produces asyncio error message.

[sbourdeauducq]

artiq_influxdb produces asyncio error message.

That's aiohttp breaking backwards compatibility, aio-libs/aiohttp#1175

[sbourdeauducq]

Can you try the following patch and see if it fixes the problem? I don't have influx ready anymore. Test in particular the clean termination of artiq_influxdb.

diff --git a/artiq/frontend/artiq_influxdb.py b/artiq/frontend/artiq_influxdb.py
index 7205e0e4..ca9c3f34 100755
--- a/artiq/frontend/artiq_influxdb.py
+++ b/artiq/frontend/artiq_influxdb.py
@@ -94,26 +94,26 @@ class DBWriter(TaskObject):
                            "too many pending updates", k)
 
     async def _do(self):
-        while True:
-            k, v, t = await self._queue.get()
-            url = self.base_url + "/write"
-            params = {"u": self.user, "p": self.password, "db": self.database,
-                      "precision": "ms"}
-            data = "{},dataset={} {} {}".format(
-                self.table, k, format_influxdb(v), round(t*1e3))
-            try:
-                response = await aiohttp.request(
-                    "POST", url, params=params, data=data)
-            except:
-                logger.warning("got exception trying to update '%s'",
-                               k, exc_info=True)
-            else:
-                if response.status not in (200, 204):
-                    content = (await response.content.read()).decode().strip()
-                    logger.warning("got HTTP status %d "
-                                   "trying to update '%s': %s",
-                                   response.status, k, content)
-                response.close()
+        async with aiohttp.ClientSession() as session:
+            while True:
+                k, v, t = await self._queue.get()
+                url = self.base_url + "/write"
+                params = {"u": self.user, "p": self.password, "db": self.database,
+                          "precision": "ms"}
+                data = "{},dataset={} {} {}".format(
+                    self.table, k, format_influxdb(v), round(t*1e3))
+                try:
+                    response = await session.post(url, params=params, data=data)
+                except:
+                    logger.warning("got exception trying to update '%s'",
+                                   k, exc_info=True)
+                else:
+                    if response.status not in (200, 204):
+                        content = (await response.content.read()).decode().strip()
+                        logger.warning("got HTTP status %d "
+                                       "trying to update '%s': %s",
+                                       response.status, k, content)
+                    response.close()
 
 
 class _Mock:

[jbqubit]

Patch applied. Running same ARTIQ Python example as above. Writing to influxdb generates no errors. Thanks @sbourdeauducq.