Version upgrade of influxDB for 1.x to 2.x

Updates

- New InfluxClient added to support 2.x
- INIT env variables added to docker compose file to intial setup of
  influx db
- Telegraf version upgrade
- Env variables added in circleCI config
- output.conf updated with outputs.influxdb_v2

.circleci/config.yml

@@ -10,11 +10,11 @@ commands:
       - run:
           name : Take a nap
           command: |
-            sleep 15
+            sleep 120
       - run:
           name : Configure and run tests
           command: |
-            pip install -r test_requirements.txt
+            pip3 install -r test_requirements.txt
             pytest --junitxml=test_results/result.xml
             echo "Done"
       - store_test_results:
@@ -28,6 +28,15 @@ jobs:
       - nuodb_collector_command
     environment:
       TEST_SUITE: Docker
+      INFLUXDB_TOKEN: quickbrownfoxjumpsoveralazydog
+      INFLUXDB_ORG: nuodb
+      DOCKER_INFLUXDB_INIT_MODE: setup
+      DOCKER_INFLUXDB_INIT_USERNAME: nuodb
+      DOCKER_INFLUXDB_INIT_PASSWORD: helloworld
+      DOCKER_INFLUXDB_INIT_ORG: nuodb
+      DOCKER_INFLUXDB_INIT_RETENTION: 1w
+      DOCKER_INFLUXDB_INIT_BUCKET: telegraf
+      DOCKER_INFLUXDB_INIT_ADMIN_TOKEN: quickbrownfoxjumpsoveralazydog
     resource_class: medium
 
   build_n_run_bare_metal:

.gitignore

@@ -56,6 +56,7 @@ coverage.xml
 .hypothesis/
 .pytest_cache/
 pytestdebug.log
+test_results/
 
 # Translations
 *.mo

Dockerfile

@@ -7,6 +7,7 @@ COPY --chown=1000:0 requirements.txt ./requirements.txt
 RUN pip install --no-cache-dir -r requirements.txt && \
     apt-get update -y && \
     apt-get install -y gettext-base apt-transport-https curl gnupg procps && \
+    # influxdata-archive_compat.key GPG Fingerprint: 9D539D90D3328DC7D6C8D3B9D8FF8E1F7DF8B07E
     curl -s https://repos.influxdata.com/influxdata-archive_compat.key | apt-key add - && \
     . /etc/os-release && \
     echo "deb https://repos.influxdata.com/debian stretch stable" > /etc/apt/sources.list.d/influxdb.list && \

conf/outputs.conf

@@ -1,32 +1,26 @@
-# Configuration for influxdb server to send metrics to
-# this is an example,  of what can be done.   Not included in docker build.
-[[outputs.influxdb]]
+[[outputs.influxdb_v2]]
+  ## The URLs of the InfluxDB cluster nodes.
+  ##
+  ## Multiple URLs can be specified for a single cluster, only ONE of the
+  ## urls will be written to each interval.
+  ##   ex: urls = ["https://us-west-2-1.aws.cloud2.influxdata.com"]
   urls = ["${INFLUXURL}"]
-  database = "telegraf"
 
-  database_tag = "db_tag"
-  exclude_database_tag = true
+  ## API token for authentication.
+  token = "${INFLUXDB_TOKEN}"
 
-  ## Retention policy to write to. Empty string writes to the default rp.
-  retention_policy = ""
-  ## Write consistency (clusters only), can be: "any", "one", "quorum", "all"
-  write_consistency = "any"
+  ## Organization is the name of the organization you wish to write to; must exist.
+  organization = "${INFLUXDB_ORG}"
 
-  ## Write timeout (for the InfluxDB client), formatted as a string.
-  ## If not provided, will default to 5s. 0s means no timeout (not recommended).
-  timeout = "5s"
-
-# [[outputs.file]]
-#   ## Files to write to, "stdout" is a specially handled file.
-#   files = ["stdout"]
+  ## Destination bucket to write into.
+  bucket = "${INFLUXDB_BUCKET}"
 
-#   ## Data format to output.
-#   ## Each data format has its own unique set of configuration options, read
-#   ## more about them here:
-#   ## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_OUTPUT.md
-#   data_format = "json"
+  ## The value of this tag will be used to determine the bucket.  If this
+  ## tag is not set the 'bucket' option is used as the default.
+  bucket_tag = "db_tag"
 
-# [[outputs.prometheus_client]]
-#   ## Address to listen on.
-#   listen = ":9273"
+  ## If true, the bucket tag will not be added to the metric.
+  # exclude_bucket_tag = false
 
+  ## Timeout for HTTP messages.
+  timeout = "5s"

docker-compose.yml

@@ -1,14 +1,14 @@
 version: "3.3"
 services:
   nuoadmin1:
-    image: $DOCKER_IMAGE
+    image: nuodb/nuodb-ce:latest
     environment:
       NUODB_DOMAIN_ENTRYPOINT: nuoadmin1
     hostname: nuoadmin1
     restart: unless-stopped
     command: ["nuoadmin"]
   sm:
-    image: $DOCKER_IMAGE
+    image: nuodb/nuodb-ce:latest
     hostname: sm
     restart: unless-stopped
     environment:
@@ -19,7 +19,7 @@ services:
         "--dba-user", "dba", "--dba-password", "goalie",
         "--servers-ready-timeout", "60"]
   te:
-    image: $DOCKER_IMAGE
+    image: nuodb/nuodb-ce:latest
     hostname: te
     restart: unless-stopped
     environment:
@@ -30,11 +30,21 @@ services:
     command: ["nuodocker", "start", "te", "--db-name", "hockey", "--server-id", "nuoadmin1",
           "--servers-ready-timeout", "60"]
   influxdb:
-    image: influxdb:1.8
+    image: influxdb:2.7.1
     ports:
       # The API for InfluxDB is served on port 8086
       - "8086:8086"
       - "8082:8082"
+    environment:
+      - DOCKER_INFLUXDB_INIT_MODE
+      - DOCKER_INFLUXDB_INIT_USERNAME
+      - DOCKER_INFLUXDB_INIT_PASSWORD
+      - DOCKER_INFLUXDB_INIT_ORG
+      - DOCKER_INFLUXDB_INIT_RETENTION
+      - DOCKER_INFLUXDB_INIT_BUCKET
+      - DOCKER_INFLUXDB_INIT_ADMIN_TOKEN
+    volumes:
+      - ./scripts:/docker-entrypoint-initdb.d
   nuocd-sm:
     build: .
     restart: unless-stopped
@@ -46,6 +56,9 @@ services:
       INFLUXURL: http://influxdb:8086
       NUOCMD_API_SERVER: nuoadmin1:8888
       NUOCD_HOSTNAME: sm
+      INFLUXDB_TOKEN: $DOCKER_INFLUXDB_INIT_ADMIN_TOKEN
+      INFLUXDB_BUCKET: $DOCKER_INFLUXDB_INIT_BUCKET
+      INFLUXDB_ORG: $DOCKER_INFLUXDB_INIT_ORG
     pid: 'service:sm'
   nuocd-te:
     build: .
@@ -58,6 +71,9 @@ services:
       INFLUXURL: http://influxdb:8086
       NUOCMD_API_SERVER: nuoadmin1:8888
       NUOCD_HOSTNAME: te
+      INFLUXDB_TOKEN: $DOCKER_INFLUXDB_INIT_ADMIN_TOKEN
+      INFLUXDB_BUCKET: $DOCKER_INFLUXDB_INIT_BUCKET
+      INFLUXDB_ORG: $DOCKER_INFLUXDB_INIT_ORG
     pid: 'service:te'
   nuocd-admin1:
     build: .
@@ -69,4 +85,7 @@ services:
       INFLUXURL: http://influxdb:8086
       NUOCMD_API_SERVER: https://nuoadmin1:8888
       NUOCD_HOSTNAME: nuoadmin1
+      INFLUXDB_TOKEN: $DOCKER_INFLUXDB_INIT_ADMIN_TOKEN
+      INFLUXDB_BUCKET: $DOCKER_INFLUXDB_INIT_BUCKET
+      INFLUXDB_ORG: $DOCKER_INFLUXDB_INIT_ORG
     pid: 'service:nuoadmin1'

scripts/setup_influx.sh

@@ -0,0 +1,6 @@
+#! /bin/bash
+set -e
+
+influx bucket create --name nuodb_internal --retention 7d
+
+influx bucket create --name nuodb --retention 7d

test_requirements.txt

@@ -5,4 +5,4 @@ coverage>=3.7
 pytest-cov>=1.8.1
 coveralls>=0.5
 python-coveralls>=2.5
-influxdb>=5.3.0
+influxdb-client

tests/nuodb_test_class.py

@@ -1,12 +1,13 @@
+import os
 import time
 import unittest
-from influxdb import InfluxDBClient
+from influxdb_client import InfluxDBClient
 
 
 def extract_names(l):
     names = []
     for o in l:
-        names.append(o["name"])
+        names.append(o.name)
 
     return names
 
@@ -34,15 +35,36 @@ def assert_await(fxn, timeout=120, interval=10):
 
     if last_assert:
         raise last_assert
+
+def getClient():
+     influxToken = os.getenv("INFLUXDB_TOKEN", 'none')
+     org = os.getenv("INFLUXDB_ORG", "nuodb")
+     client = InfluxDBClient(url='http://localhost:8086', token=influxToken, org=org)
+     return client
+
+def getMeasurenment(client, bucket):
+    query = f"""
+    import \"influxdata/influxdb/schema\"
+
+    schema.measurements(bucket: \"{bucket}\")
+    """
+    query_api = client.query_api()
+    tables = query_api.query(query=query)
+
+    # Flatten output tables into list of measurements
+    measurements = [row.values["_value"] for table in tables for row in table]
+    return measurements
 
 
 class NuoDBTelegrafTestClass(unittest.TestCase):
     database_name = None
 
     @classmethod
     def isDatabaseRunning(cls, client, database):
-        databases = extract_names(client.get_list_database())
-
+        buckets_api = client.buckets_api()
+        buckets = buckets_api.find_buckets().buckets
+        databases = extract_names(buckets)
+
         if database not in databases:
             raise EnvironmentError("%s database not created" % database)
 
@@ -51,28 +73,26 @@ def isDatabaseRunning(cls, client, database):
     @classmethod
     def setUpClass(cls):
         # ensure that the database exists
-        client = InfluxDBClient('localhost', 8086)
+        client = getClient()
         assert_await(lambda: cls.isDatabaseRunning(client, cls.database_name), timeout=60)
-
-    def assertMeasurementPresent(self, client, measurement):
-        measurements = extract_names(client.get_list_measurements())
-
-        # verify a random set of stats
+
+    def assertMeasurementPresent(self, client, bucket, measurement):
+        measurements = getMeasurenment(client, bucket)
         self.assertIn(measurement, measurements)
 
-    def assertMeasurementCountGt0(self, client, measurement):
-        rs = client.query('select count(*) from %s' % measurement)
-        points = list(rs.get_points(measurement=measurement))
-        self.assertGreater(points[0]["count_raw"], 0)
-
-    def assertTraceInMeasurement(self, client, measurement, trace):
-        rs = client.query("select * from %s" % measurement)
-        points = list(rs.get_points(measurement=measurement))
-        traces = extract_traces(points)
-
-        self.assertIn(trace, traces)
-
-    def assertMeasurementNotEmpty(self, client, measurement):
-        rs = client.query("select * from %s" % measurement)
-        points = list(rs.get_points(measurement=measurement))
-        self.assertGreater(len(points), 0, "Measurement %s does not contain any points" % measurement)
+    def assertMeasurementCountGt0(self, client, bucket, measurement):
+        query_api = client.query_api()
+        query = f"""from(bucket: \"{bucket}\")
+            |> range(start: -5m)  
+            |> filter(fn: (r) => r[\"_measurement\"] == \"{measurement}\")"""
+        result = query_api.query(query)
+        self.assertGreater(len(result),0)
+
+    def assertTraceInMeasurement(self, client, bucket, measurement, trace):
+        query_api = client.query_api()
+        query = f"""from(bucket: \"{bucket}\")
+            |> range(start: -5m)  
+            |> filter(fn: (r) => r[\"_measurement\"] == \"{measurement}\")
+            |> filter(fn: (r) => r[\"msg_trace_metric\"] == \"{trace}\")"""
+        result = query_api.query(query)
+        self.assertGreater(len(result),0, f"Measurement {measurement} does not contain an trace {trace}")

tests/stats_test.py

@@ -1,41 +1,34 @@
-from nuodb_test_class import *
+from .nuodb_test_class import *
 
 
 class TestNuoDB(NuoDBTelegrafTestClass):
     database_name = "nuodb"
     host = 'localhost'
 
     def test_measurements_ActualVersion(self):
-        client = InfluxDBClient(self.host, 8086)
-        client.switch_database(self.database_name)
-        assert_await(lambda: self.assertMeasurementPresent(client, "ActualVersion"), timeout=120, interval=10)
+        client = getClient()
+        assert_await(lambda: self.assertMeasurementPresent(client, self.database_name, "ActualVersion"), timeout=120, interval=10)
 
     def test_measurements_ArchiveQueue(self):
-        client = InfluxDBClient(self.host, 8086)
-        client.switch_database(self.database_name)
-        assert_await(lambda: self.assertMeasurementPresent(client, "ArchiveQueue"), timeout=120, interval=10)
+        client = getClient()
+        assert_await(lambda: self.assertMeasurementPresent(client, self.database_name, "ArchiveQueue"), timeout=120, interval=10)
 
     def test_measurements_ChairmanMigration(self):
-        client = InfluxDBClient(self.host, 8086)
-        client.switch_database(self.database_name)
-        assert_await(lambda: self.assertMeasurementPresent(client, "ChairmanMigration"), timeout=120, interval=10)
+        client = getClient()
+        assert_await(lambda: self.assertMeasurementPresent(client,  self.database_name, "ChairmanMigration"), timeout=120, interval=10)
 
     def test_measurements_CurrentActiveTransactions(self):
-        client = InfluxDBClient(self.host, 8086)
-        client.switch_database(self.database_name)
-        assert_await(lambda: self.assertMeasurementPresent(client, "CurrentActiveTransactions"), timeout=120, interval=10)
+        client = getClient()
+        assert_await(lambda: self.assertMeasurementPresent(client,  self.database_name, "CurrentActiveTransactions"), timeout=120, interval=10)
 
     def test_measurements_CurrentCommittedTransactions(self):
-        client = InfluxDBClient(self.host, 8086)
-        client.switch_database(self.database_name)
-        assert_await(lambda: self.assertMeasurementPresent(client, "CurrentCommittedTransactions"), timeout=120, interval=10)
+        client = getClient()
+        assert_await(lambda: self.assertMeasurementPresent(client,  self.database_name, "CurrentCommittedTransactions"), timeout=120, interval=10)
 
     def test_measurements_NodeId(self):
-        client = InfluxDBClient(self.host, 8086)
-        client.switch_database(self.database_name)
-        assert_await(lambda: self.assertMeasurementPresent(client, "NodeId"), timeout=120, interval=10)
+        client = getClient()
+        assert_await(lambda: self.assertMeasurementPresent(client,  self.database_name, "NodeId"), timeout=120, interval=10)
 
     def test_measurements_CurrentActiveTransactions_count(self):
-        client = InfluxDBClient(self.host, 8086)
-        client.switch_database(self.database_name)
-        assert_await(lambda: self.assertMeasurementCountGt0(client, "CurrentActiveTransactions"), timeout=120, interval=10)
+        client = getClient()
+        assert_await(lambda: self.assertMeasurementCountGt0(client, self.database_name, "CurrentActiveTransactions"), timeout=120, interval=10)