influx_overwrite_bug.py

#!/usr/bin/env python
# coding: utf-8

from influxdb import InfluxDBClient
import time

DBNAME = 'test_db'
SERIES_NAME = 'test_series'
POINTS_NUMBER = 5000000
THRESHOLD = 500000
BASE_TS = 1462060800000
INFLUXDB_CLIENT = InfluxDBClient(
    host='127.0.0.1',
    port=8086,
    username=None,
    password=None,
)


def insert_points(limit=POINTS_NUMBER):

    print '==INSERTING SERIES=='
    print '* inserting {} points: from {}ms - to {}ms'.format(
        limit, BASE_TS, BASE_TS+limit)

    headers = INFLUXDB_CLIENT._headers
    headers['Content-type'] = 'application/octet-stream'

    points = ""
    for i in xrange(0, limit):
        points += '{},tag=tag{} value={} {}\n'.format(
            SERIES_NAME, i % 3, i % 10 + 1, BASE_TS+i)

        if (i % THRESHOLD == 0) or (i == limit-1):
            INFLUXDB_CLIENT.request('write',
                                    method='POST',
                                    data=points,
                                    headers=headers,
                                    params=dict(db=DBNAME, precision='ms'),
                                    expected_response_code=204,
                                    )
            points = ""


def do_query():
    print('==QUERY==')
    start = time.time()
    q = "select * from {} LIMIT 1".format(SERIES_NAME)
    res = INFLUXDB_CLIENT.query(q, epoch='ms')
    duration = time.time() - start
    print("* query: {}".format(q))
    print("* result: {}".format(list(res.get_points())[0]))
    print("* duration: {:.4f}s".format(duration))


def main():

    print('==CREATING DATABASE==')
    INFLUXDB_CLIENT.create_database(DBNAME)
    INFLUXDB_CLIENT.switch_database(DBNAME)

    insert_points()
    do_query()

    insert_points(limit=POINTS_NUMBER)
    do_query()

    print('==SLEEP 10s==')
    time.sleep(10)
    do_query()

    print('==REMOVING DATABASE==')
    INFLUXDB_CLIENT.drop_database(DBNAME)


if __name__ == "__main__":
    main()