controller/rootfs/api/influxdb.py at c07f22751a729d11f1c47666a801df7fcfe14312 · drycc/controller · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
import threading
import logging
from contextlib import closing
from typing import Iterator
from django.conf import settings
from influxdb_client import InfluxDBClient
from influxdb_client.client.flux_table import FluxRecord
from influxdb_client.rest import ApiException

local = threading.local()
logger = logging.getLogger(__name__)


def _get_influxdb_client() -> InfluxDBClient:
    return InfluxDBClient(
        url=settings.DRYCC_INFLUXDB_URL,
        token=settings.DRYCC_INFLUXDB_TOKEN,
        org=settings.DRYCC_INFLUXDB_ORG
    )


def _query_stream(flux_script: str) -> Iterator[FluxRecord]:
    client = _get_influxdb_client()
    try:
        query_api = client.query_api()
        records = query_api.query_stream(flux_script)
        with closing(records):
            yield from records
    except ApiException as e:
        logger.exception(e)
        yield from []
    except Exception as e:
        logger.exception(e)
        yield from []


def query_container_count(
        namespaces: Iterator[str], start: int, stop: int) -> Iterator[FluxRecord]:
    namespace_range = ' or '.join(
        [f'r["namespace"] == \"{namespace}\"' for namespace in namespaces])
    flux_script = f'''
        from(bucket: "kubernetes")
            |> range(start: {start}, stop: {stop})
            |> filter(fn: (r) => r["_measurement"] == "kubernetes_pod_container")
            |> filter(fn: (r) => r["_field"]=="cpu_usage_core_nanoseconds")
            |> filter(fn: (r) => {namespace_range})
            |> group(columns: ["_time", "namespace", "container_name"])
            |> count()
            |> group(columns: ["namespace", "container_name"])
            |> top(n: 3)
            |> min()
    '''
    yield from _query_stream(flux_script)


def query_network_flow(
        namespaces: Iterator[str], start: int, stop: int) -> Iterator[FluxRecord]:
    namespace_range = ' or '.join(
        [f'r["namespace"] == \"{namespace}\"' for namespace in namespaces])
    flux_script = f'''
        from(bucket: "kubernetes")
            |> range(start: {start}, stop: {stop})
            |> filter(fn: (r) => r["_measurement"] == "kubernetes_pod_network")
            |> filter(fn: (r) => r["_field"] == "rx_bytes" or r["_field"] == "tx_bytes")
            |> filter(fn: (r) => {namespace_range})
            |> increase()
            |> last()
            |> pivot(
                rowKey:["_time"],
                columnKey: ["_field"],
                valueColumn: "_value"
            )
    '''
    yield from _query_stream(flux_script)


def query_cpu_usage(
        namespace, container_type, start, stop, every) -> Iterator[FluxRecord]:
    flux_script = f"""
        from(bucket: "kubernetes")
            |> range(start: {start}, stop: {stop})
            |> filter(fn: (r) => r["_measurement"] == "kubernetes_pod_container")
            |> filter(fn: (r) => r["_field"] == "cpu_usage_nanocores")
            |> filter(fn: (r) => r["namespace"] == "{namespace}")
            |> filter(fn: (r) => r["pod_name"] =~ { "/%s-%s/" % (namespace, container_type) })
            |> group(columns: ["container_name"])
            |> aggregateWindow(every: {every}, fn: max, createEmpty: false)
            |> yield(name: "max")
            |> aggregateWindow(every: {every}, fn: mean, createEmpty: false)
            |> yield(name: "mean")
    """
    yield from _query_stream(flux_script)


def query_memory_usage(
        namespace, container_type, start, stop, every) -> Iterator[FluxRecord]:
    flux_script = f"""
        from(bucket: "kubernetes")
            |> range(start: {start}, stop: {stop})
            |> filter(fn: (r) => r["_measurement"] == "kubernetes_pod_container")
            |> filter(fn: (r) => r["_field"] == "memory_usage_bytes")
            |> filter(fn: (r) => r["namespace"] == "{namespace}")
            |> filter(fn: (r) => r["pod_name"] =~ { "/%s-%s/" % (namespace, container_type) })
            |> group(columns: ["container_name"])
            |> aggregateWindow(every: {every}, fn: max, createEmpty: false)
            |> yield(name: "max")
            |> aggregateWindow(every: {every}, fn: mean, createEmpty: false)
            |> yield(name: "mean")
    """
    yield from _query_stream(flux_script)


def query_network_usage(
        namespace, container_type, start, stop, every) -> Iterator[FluxRecord]:
    flux_script = f"""
        from(bucket: "kubernetes")
            |> range(start: {start}, stop: {stop})
            |> filter(fn: (r) => r["_measurement"] == "kubernetes_pod_network")
            |> filter(fn: (r) => r["_field"] == "rx_bytes" or r["_field"] == "tx_bytes")
            |> filter(fn: (r) => r["namespace"] == "{namespace}")
            |> filter(fn: (r) => r["pod_name"] =~ { "/%s-%s/" % (namespace, container_type) })
            |> group(columns: ["container_name", "_field"])
            |> aggregateWindow(every: {every}, fn: max, createEmpty: false)
            |> difference(nonNegative: true)
            |> pivot(
                rowKey:["_time"],
                columnKey: ["_field"],
                valueColumn: "_value"
            )
            |> limit(n:3000)
    """
    yield from _query_stream(flux_script)