Time-Series Databases

# Is it ready yet?
curl -s http://localhost:8086/health

# Reach the CLI inside the container. `influx ping` needs no credentials;
# anything touching data needs the admin token.
docker exec influx-demo influx ping

docker exec influx-demo influx bucket list \
  --org my-org --token my-super-secret-token

import psycopg
from datetime import datetime, timedelta, timezone
import random

# Connect to TimescaleDB (PostgreSQL with extension)
conn = psycopg.connect("host=localhost dbname=demo user=demo password=demo")
cursor = conn.cursor()

# Simulate sensor data
random.seed(42)   # so the aggregates in the next example are reproducible

sensors = ['sensor_1', 'sensor_2', 'sensor_3']
locations = ['building_a', 'building_b', 'building_c']

print("Inserting 10,000 sensor readings...")
# datetime.now() is NAIVE: it carries no offset, so PostgreSQL
# interprets it in the session's TimeZone. Write that into a
# TIMESTAMPTZ column from a machine set to anything but UTC and
# every reading silently lands hours away from where you meant.
base_time = datetime.now(timezone.utc) - timedelta(days=7)

for i in range(10000):
    timestamp = base_time + timedelta(minutes=i)
    sensor_id = random.choice(sensors)
    location = random.choice(locations)
    temperature = round(random.uniform(18.0, 28.0), 2)
    humidity = round(random.uniform(30.0, 70.0), 2)

    cursor.execute("""
        INSERT INTO sensor_data (time, sensor_id, temperature, humidity, location)
        VALUES (%s, %s, %s, %s, %s)
    """, (timestamp, sensor_id, temperature, humidity, location))

    if (i + 1) % 1000 == 0:
        print(f"  Inserted {i + 1} records...")
        conn.commit()

conn.commit()
print("✓ Data insertion complete!")

cursor.execute("""
    SELECT count(*) FROM timescaledb_information.chunks
    WHERE hypertable_name = 'sensor_data'
""")
print(f"chunks created automatically: {cursor.fetchone()[0]}")

cursor.close()
conn.close()

import psycopg
from datetime import datetime, timedelta, timezone

conn = psycopg.connect("host=localhost dbname=demo user=demo password=demo")
cursor = conn.cursor()

# Query 1: Recent data (last 24 hours)
cursor.execute("""
    SELECT
        sensor_id,
        AVG(temperature) as avg_temp,
        MAX(temperature) as max_temp,
        MIN(temperature) as min_temp,
        COUNT(*) as reading_count
    FROM sensor_data
    WHERE time >= NOW() - INTERVAL '24 hours'
    GROUP BY sensor_id
    ORDER BY sensor_id
""")

print("Sensor statistics (last 24 hours):")
for sensor, avg_t, max_t, min_t, count in cursor.fetchall():
    print(f"  {sensor}: Avg={avg_t:.1f}°C, Max={max_t:.1f}°C, Min={min_t:.1f}°C ({count} readings)")

# Query 2: Time-bucket aggregation (hourly averages)
cursor.execute("""
    SELECT
        time_bucket('1 hour', time) AS hour,
        sensor_id,
        AVG(temperature) as avg_temp,
        AVG(humidity) as avg_humidity
    FROM sensor_data
    WHERE time >= NOW() - INTERVAL '7 days'
    GROUP BY hour, sensor_id
    ORDER BY hour DESC, sensor_id
    LIMIT 10
""")

print("\nHourly averages (last 7 days, most recent first):")
for hour, sensor, temp, humidity in cursor.fetchall():
    print(f"  {hour} - {sensor}: {temp:.1f}°C, {humidity:.1f}%")

cursor.close()
conn.close()

import psycopg

conn = psycopg.connect("host=localhost dbname=demo user=demo password=demo")
cursor = conn.cursor()

# Check compression statistics
cursor.execute("""
    SELECT
        pg_size_pretty(before_compression_total_bytes) as uncompressed,
        pg_size_pretty(after_compression_total_bytes) as compressed,
        ROUND(100 - (after_compression_total_bytes::numeric /
              before_compression_total_bytes::numeric * 100), 2) as compression_ratio
    FROM hypertable_compression_stats('sensor_data')
""")

result = cursor.fetchone()
# A row comes back even when nothing has been compressed, with NULLs in
# it, so test the values rather than the row.
if result and result[0] is not None:
    uncompressed, compressed, ratio = result
    print("Compression Statistics:")
    print(f"  Uncompressed size: {uncompressed}")
    print(f"  Compressed size: {compressed}")
    print(f"  Space saved: {ratio}%")
else:
    print("No compression data available yet")

cursor.close()
conn.close()

from influxdb_client import InfluxDBClient, Point
from influxdb_client.client.write_api import SYNCHRONOUS
from datetime import datetime, timedelta, timezone
import random

# Connect to InfluxDB
client = InfluxDBClient(
    url="http://localhost:8086",
    token="my-super-secret-token",   # from the docker run above
    org="my-org"
)

write_api = client.write_api(write_options=SYNCHRONOUS)
bucket = "sensors"

# Write sensor readings
print("Writing sensor data to InfluxDB...")
random.seed(42)   # so the averages in the next example are reproducible

sensors = ['sensor_1', 'sensor_2', 'sensor_3']
locations = ['building_a', 'building_b']

base_time = datetime.now(timezone.utc) - timedelta(hours=24)

for i in range(1000):
    timestamp = base_time + timedelta(minutes=i)

    for sensor in sensors:
        for location in locations:
            point = Point("sensor_readings") \
                .tag("sensor_id", sensor) \
                .tag("location", location) \
                .field("temperature", round(random.uniform(18.0, 28.0), 2)) \
                .field("humidity", round(random.uniform(30.0, 70.0), 2)) \
                .time(timestamp)

            write_api.write(bucket=bucket, record=point)

    if (i + 1) % 100 == 0:
        print(f"  Written {(i + 1) * len(sensors) * len(locations)} points...")

print("✓ Data written successfully!")
client.close()

from influxdb_client import InfluxDBClient

client = InfluxDBClient(
    url="http://localhost:8086",
    token="my-super-secret-token",   # from the docker run above
    org="my-org"
)

query_api = client.query_api()

# Query 1: Average temperature by sensor (last 24 hours)
query = '''
from(bucket: "sensors")
  |> range(start: -24h)
  |> filter(fn: (r) => r._measurement == "sensor_readings")
  |> filter(fn: (r) => r._field == "temperature")
  |> group(columns: ["sensor_id"])
  |> mean()
'''

tables = query_api.query(query)

print("Average temperature by sensor (last 24h):")
for table in tables:
    for record in table.records:
        sensor = record.values.get("sensor_id")
        avg_temp = record.get_value()
        print(f"  {sensor}: {avg_temp:.2f}°C")

# Query 2: Downsampled data (hourly averages)
query = '''
from(bucket: "sensors")
  |> range(start: -7d)
  |> filter(fn: (r) => r._measurement == "sensor_readings")
  |> filter(fn: (r) => r.sensor_id == "sensor_1")
  |> filter(fn: (r) => r._field == "temperature")
  |> aggregateWindow(every: 1h, fn: mean, createEmpty: false)
  |> limit(n: 10)
'''

tables = query_api.query(query)

# sensor_1 was written under two locations, so Flux returns TWO tables,
# one per series, and limit(n: 10) applied to each of them separately.
# Print the location or the rows look like inexplicable duplicates.
print(f"\nHourly averages for sensor_1 ({len(tables)} tables returned):")
for table in tables:
    location = table.records[0].values.get("location")
    print(f"  location={location}")
    for record in table.records:
        time = record.get_time()
        temp = record.get_value()
        print(f"    {time}: {temp:.2f}°C")

client.close()

# Create downsampling task (runs every hour)
# Reduces high-resolution data to hourly averages

from influxdb_client import InfluxDBClient

client = InfluxDBClient(url="http://localhost:8086",
                        token="my-super-secret-token", org="my-org")

# create_task_every() wants an Organization OBJECT, not the name string,
# so look it up first. Passing "my-org" here fails with a 404.
org = client.organizations_api().find_organizations(org="my-org")[0]

# The destination bucket has to exist before the task can write to it.
buckets_api = client.buckets_api()
if buckets_api.find_bucket_by_name("sensor_data_hourly") is None:
    buckets_api.create_bucket(bucket_name="sensor_data_hourly", org_id=org.id)

# NOTE: no 'option task = {...}' line here. create_task_every() injects
# one for you from the name/every arguments below. Supply both and
# InfluxDB rejects the request:
#   400 Bad Request: invalid options: multiple task options defined
# Either write the option yourself and call create_task(), or leave it
# out and call create_task_every(). Not both.
task_flux = '''
from(bucket: "sensors")
  |> range(start: -1h)
  |> filter(fn: (r) => r._measurement == "sensor_readings")
  |> aggregateWindow(every: 1h, fn: mean, createEmpty: false)
  |> to(bucket: "sensor_data_hourly")
'''

task = client.tasks_api().create_task_every(
    name="downsample_hourly",
    flux=task_flux,
    every="1h",
    organization=org
)

print(f"Created downsampling task: {task.name}")
print(f"  Runs every: {task.every}")
print(f"  Status: {task.status}")
print(f"  Task ID: {task.id}")

client.close()

from influxdb_client import InfluxDBClient, BucketRetentionRules

client = InfluxDBClient(url="http://localhost:8086",
                        token="my-super-secret-token", org="my-org")
buckets_api = client.buckets_api()

DAY = 24 * 60 * 60


def ensure_bucket(name, days):
    """Create a bucket with a retention period, or leave it alone if it exists."""
    existing = buckets_api.find_bucket_by_name(name)
    if existing is not None:
        return existing
    return buckets_api.create_bucket(
        bucket_name=name,
        # The class is BucketRetentionRules, exported from the package root.
        # There is no RetentionRules.
        retention_rules=BucketRetentionRules(type="expire",
                                             every_seconds=days * DAY),
        org="my-org",
    )


raw = ensure_bucket("sensor_data_raw", 30)
hourly = ensure_bucket("sensor_data_hourly", 365)

print(f"Created bucket: {raw.name}")
print(f"  retention: {raw.retention_rules[0].every_seconds} seconds (30 days)")

# find_buckets() does not promise an order, so sort before printing.
print("\nBuckets now:")
for b in sorted(buckets_api.find_buckets().buckets, key=lambda b: b.name):
    if b.retention_rules:
        secs = b.retention_rules[0].every_seconds
        retention = f"{secs // DAY} days" if secs else "infinite"
    else:
        retention = "infinite"
    print(f"  {b.name + ':':<20} retention {retention}")

client.close()

from prometheus_client import start_http_server, Counter, Gauge, Histogram
import random
import time

# Define metrics
request_counter = Counter(
    'http_requests_total',
    'Total HTTP requests',
    ['method', 'endpoint', 'status']
)

active_users = Gauge(
    'active_users',
    'Number of active users'
)

request_duration = Histogram(
    'request_duration_seconds',
    'Request duration in seconds',
    ['endpoint']
)

# Simulate application metrics
def simulate_traffic():
    """Simulate web application traffic"""
    endpoints = ['/api/users', '/api/products', '/api/orders']
    methods = ['GET', 'POST']
    statuses = ['200', '404', '500']

    while True:
        # Increment request counter
        endpoint = random.choice(endpoints)
        method = random.choice(methods)
        status = random.choice(statuses)
        request_counter.labels(method=method, endpoint=endpoint, status=status).inc()

        # Update active users gauge
        active_users.set(random.randint(100, 500))

        # Record request duration
        duration = random.uniform(0.01, 2.0)
        request_duration.labels(endpoint=endpoint).observe(duration)

        time.sleep(1)  # Wait 1 second

if __name__ == '__main__':
    # Start metrics server on port 8000
    start_http_server(8000)
    # flush=True matters in Docker: stdout is a pipe rather than a
    # terminal, so Python block-buffers it and these lines would not
    # reach "docker logs" until the buffer filled, making a perfectly
    # healthy container look silent.
    print("Metrics server started on port 8000", flush=True)
    print("Prometheus can scrape: http://localhost:8000/metrics", flush=True)

    # Simulate traffic
    simulate_traffic()

# Querying needs a different package from the exporter above. The exporter
# used prometheus_client; reading data back uses prometheus-api-client.
# Note the names differ: hyphens to install, underscores to import.
#
#   python3 -m venv .venv && source .venv/bin/activate
#   pip install prometheus-api-client
#
from prometheus_api_client import PrometheusConnect

# Connect to Prometheus server
prom = PrometheusConnect(url="http://localhost:9090", disable_ssl=True)

# Query 1: Current request rate (requests per second)
query = 'rate(http_requests_total[5m])'
result = prom.custom_query(query)

print("Current request rate (requests/sec):")
for item in result:
    metric = item['metric']
    value = float(item['value'][1])
    # Include status: the series is keyed on all three labels, so
    # printing only two makes distinct series look like duplicates.
    print(f"  {metric['endpoint']} ({metric['method']}, "
          f"{metric['status']}): {value:.2f} req/s")

# Query 2: 95th percentile request duration
query = 'histogram_quantile(0.95, rate(request_duration_seconds_bucket[5m]))'
result = prom.custom_query(query)

print("\n95th percentile request duration:")
for item in result:
    endpoint = item['metric'].get('endpoint', 'unknown')
    p95 = float(item['value'][1])
    print(f"  {endpoint}: {p95:.3f}s")

# Query 3: Error rate (percentage)
query = '''
  sum(rate(http_requests_total{status=~"5.."}[5m])) /
  sum(rate(http_requests_total[5m])) * 100
'''
result = prom.custom_query(query)

if result:
    error_rate = float(result[0]['value'][1])
    print(f"\nCurrent error rate: {error_rate:.2f}%")

-- Create continuous aggregate for hourly data
CREATE MATERIALIZED VIEW sensor_data_hourly
WITH (timescaledb.continuous) AS
SELECT
    time_bucket('1 hour', time) AS hour,
    sensor_id,
    location,
    AVG(temperature) as avg_temperature,
    MAX(temperature) as max_temperature,
    MIN(temperature) as min_temperature,
    AVG(humidity) as avg_humidity,
    COUNT(*) as reading_count
FROM sensor_data
GROUP BY hour, sensor_id, location;

-- Create continuous aggregate for daily data
CREATE MATERIALIZED VIEW sensor_data_daily
WITH (timescaledb.continuous) AS
SELECT
    time_bucket('1 day', time) AS day,
    sensor_id,
    location,
    AVG(temperature) as avg_temperature,
    MAX(temperature) as max_temperature,
    MIN(temperature) as min_temperature,
    AVG(humidity) as avg_humidity,
    COUNT(*) as reading_count
FROM sensor_data
GROUP BY day, sensor_id, location;

-- Retention policies. Same story as compression below: the retention
-- section earlier in this lesson already put a 90-day policy on
-- sensor_data, and a second one fails with "retention policy already
-- exists for hypertable". if_not_exists makes this block re-runnable,
-- but note it also means the 7-day interval here is NOT applied if the
-- older policy is still in place. Drop the old one first if you mean to
-- change it: SELECT remove_retention_policy('sensor_data');
SELECT add_retention_policy('sensor_data', INTERVAL '7 days',
                            if_not_exists => true);                    -- Raw: 7 days
SELECT add_retention_policy('sensor_data_hourly', INTERVAL '90 days'); -- Hourly: 90 days
SELECT add_retention_policy('sensor_data_daily', INTERVAL '2 years');  -- Daily: 2 years

-- Compression policies.
-- A policy only SCHEDULES compression; it does not turn the feature on.
-- Call add_compression_policy on a table that has never had compression
-- enabled and TimescaleDB 2.18+ rejects it outright:
--   ERROR:  columnstore not enabled on hypertable "sensor_data"
--   HINT:   Enable columnstore before adding a columnstore policy.
-- So enable it first, on the hypertable and on each continuous aggregate:
ALTER TABLE sensor_data SET (
    timescaledb.compress,
    timescaledb.compress_segmentby = 'sensor_id, location',
    timescaledb.compress_orderby   = 'time DESC'
);
ALTER MATERIALIZED VIEW sensor_data_hourly SET (timescaledb.compress = true);

-- if_not_exists matters here: the compression section earlier in this
-- lesson already put a policy on sensor_data, and adding a second one
-- fails with "columnstore policy already exists for hypertable".
SELECT add_compression_policy('sensor_data', INTERVAL '1 day',
                              if_not_exists => true);
SELECT add_compression_policy('sensor_data_hourly', INTERVAL '7 days');

-- A continuous aggregate does NOT refresh itself just because it was
-- created. Without a refresh policy it holds only the data that existed
-- at CREATE time and silently goes stale. This is the step most often
-- missed:
SELECT add_continuous_aggregate_policy('sensor_data_hourly',
    start_offset => INTERVAL '3 days',
    end_offset   => INTERVAL '1 hour',
    schedule_interval => INTERVAL '1 hour');

SELECT add_continuous_aggregate_policy('sensor_data_daily',
    start_offset => INTERVAL '30 days',
    end_offset   => INTERVAL '1 day',
    schedule_interval => INTERVAL '1 day');

import psycopg
from datetime import datetime, timedelta, timezone

conn = psycopg.connect("host=localhost dbname=demo user=demo password=demo")
cursor = conn.cursor()

# Query recent data (uses raw table - high resolution)
cursor.execute("""
    SELECT
        time_bucket('5 minutes', time) as interval,
        AVG(temperature) as avg_temp
    FROM sensor_data
    WHERE time >= NOW() - INTERVAL '6 hours'
        AND sensor_id = 'sensor_1'
    GROUP BY interval
    ORDER BY interval DESC
    LIMIT 10
""")

print("Recent data (5-minute intervals, last 6 hours):")
for interval, temp in cursor.fetchall():
    print(f"  {interval}: {temp:.2f}°C")

# Query older data (uses hourly aggregate - pre-computed)
cursor.execute("""
    -- The aggregate's group key is (hour, sensor_id, location), so
    -- filtering on sensor_id alone still returns one row per location
    -- and the timestamps look duplicated. Select location too.
    SELECT
        hour,
        location,
        avg_temperature
    FROM sensor_data_hourly
    WHERE hour >= NOW() - INTERVAL '30 days'
        AND hour < NOW() - INTERVAL '7 days'
        AND sensor_id = 'sensor_1'
    ORDER BY hour DESC, location
    LIMIT 10
""")

print("\nOlder data (hourly averages, 7-30 days ago):")
for hour, location, temp in cursor.fetchall():
    print(f"  {hour} {location}: {temp:.2f}°C")

# Query historical data (uses daily aggregate)
cursor.execute("""
    SELECT
        day,
        location,
        avg_temperature,
        max_temperature,
        min_temperature
    FROM sensor_data_daily
    WHERE day >= NOW() - INTERVAL '1 year'
        AND sensor_id = 'sensor_1'
    ORDER BY day DESC, location
    LIMIT 10
""")

print("\nHistorical data (daily averages, last year):")
for day, location, avg_t, max_t, min_t in cursor.fetchall():
    print(f"  {day} {location}: Avg={avg_t:.1f}°C, Max={max_t:.1f}°C, "
          f"Min={min_t:.1f}°C")

cursor.close()
conn.close()