diff --git a/dashboards/customized/dial_analytics.json b/dashboards/customized/dial_analytics.json index f1683dc..6a22771 100644 --- a/dashboards/customized/dial_analytics.json +++ b/dashboards/customized/dial_analytics.json @@ -1,16 +1,27 @@ { + "__inputs": [ + { + "name": "DS_INFLUXDB", + "label": "InfluxDB", + "description": "", + "type": "datasource", + "pluginId": "influxdb", + "pluginName": "InfluxDB" + } + ], + "__elements": {}, "__requires": [ { "type": "panel", "id": "ae3e-plotly-panel", "name": "Plotly panel", - "version": "0.3.3" + "version": "0.5.0" }, { "type": "grafana", "id": "grafana", "name": "Grafana", - "version": "10.0.3" + "version": "12.0.0" }, { "type": "datasource", @@ -22,7 +33,7 @@ "type": "panel", "id": "magnesium-wordcloud-panel", "name": "Word cloud", - "version": "1.2.4" + "version": "1.2.11" }, { "type": "panel", @@ -59,12 +70,11 @@ } ] }, - "editable": false, + "editable": true, "fiscalYearStartMonth": 0, "graphTooltip": 0, "id": null, "links": [], - "liveNow": false, "panels": [ { "datasource": { @@ -82,8 +92,7 @@ "mode": "absolute", "steps": [ { - "color": "green", - "value": null + "color": "green" }, { "color": "red", @@ -106,6 +115,7 @@ "graphMode": "area", "justifyMode": "auto", "orientation": "auto", + "percentChangeColorMode": "standard", "reduceOptions": { "calcs": [ "last" @@ -113,17 +123,40 @@ "fields": "/.*/", "values": false }, - "textMode": "auto" + "showPercentChange": false, + "textMode": "auto", + "wideLayout": true }, - "pluginVersion": "10.0.3", + "pluginVersion": "12.0.0", "targets": [ { + "dataset": "iox", "datasource": { "type": "influxdb", - "uid": "${datasource}" + "uid": "${DS_INFLUXDB}" }, + "editorMode": "code", + "format": "table", "query": "bucket = \"${INFLUX_BUCKET}\"\r\n\r\nfrom(bucket: bucket)\r\n |> range(start: v.timeRangeStart, stop:v.timeRangeStop)\r\n |> filter(fn: (r) => r._measurement == \"analytics\" and r._field == \"user_hash\")\r\n |> group()\r\n |> distinct()\r\n |> count()", - "refId": "A" + "rawQuery": true, + "rawSql": "SELECT COUNT(DISTINCT user_hash)\nFROM analytics\nWHERE time >= $__timeFrom AND time < $__timeTo", + "refId": "A", + "sql": { + "columns": [ + { + "parameters": [], + "type": "function" + } + ], + "groupBy": [ + { + "property": { + "type": "string" + }, + "type": "groupBy" + } + ] + } } ], "title": "Unique Users", @@ -164,14 +197,36 @@ "transitionDuration": 800 } }, + "pluginVersion": "1.2.11", "targets": [ { + "dataset": "iox", "datasource": { "type": "influxdb", - "uid": "${datasource}" + "uid": "${DS_INFLUXDB}" }, + "editorMode": "code", + "format": "table", "query": "bucket = \"${INFLUX_BUCKET}\"\r\n\r\nfrom(bucket: bucket)\r\n |> range(start: v.timeRangeStart, stop:v.timeRangeStop)\r\n |> filter(fn: (r) => r._measurement == \"analytics\")\r\n |> filter(fn: (r) => r._field == \"number_request_messages\")\r\n |> group(columns: [\"topic\"])\r\n |> count()\r\n |> group()", - "refId": "A" + "rawQuery": true, + "rawSql": "SELECT topic, COUNT(*) AS _value\nFROM analytics\nWHERE time >= $__timeFrom \n AND time < $__timeTo \n AND topic IS NOT NULL\n AND TRIM(topic) <> ''\nGROUP BY topic", + "refId": "A", + "sql": { + "columns": [ + { + "parameters": [], + "type": "function" + } + ], + "groupBy": [ + { + "property": { + "type": "string" + }, + "type": "groupBy" + } + ] + } } ], "title": "Popular Topics", @@ -230,15 +285,36 @@ "onclick": "//console.log(data)\n//window.updateVariables({query:{'var-project':'test'}, partial: true})", "script": "var title = data.series[0].fields.find(x => x.name === \"title\")?.values\nvar topic = data.series[0].fields.find(x => x.name === \"topic\")?.values\nvar count = data.series[0].fields.find(x => x.name === \"_value\")?.values\n\nvar trace = {\n x: title?.buffer || title || [\"No data\"],\n y: topic?.buffer || topic || [\"No data\"],\n z: count?.buffer || count || [NaN],\n type: 'heatmap',\n colorscale: \"YlOrRd\",\n};\n\nreturn {data:[trace]};" }, - "pluginVersion": "9.2.4", + "pluginVersion": "0.5.0", "targets": [ { + "dataset": "iox", "datasource": { "type": "influxdb", - "uid": "${datasource}" + "uid": "${DS_INFLUXDB}" }, + "editorMode": "code", + "format": "table", "query": "import \"influxdata/influxdb/schema\"\r\n\r\nbucket = \"${INFLUX_BUCKET}\"\r\n\r\nfrom(bucket: bucket)\r\n |> range(start: v.timeRangeStart, stop:v.timeRangeStop)\r\n |> filter(fn: (r) => r._measurement == \"analytics\")\r\n |> filter(fn: (r) => r._field == \"number_request_messages\")\r\n |> group(columns: [\"topic\", \"title\"])\r\n |> count()\r\n |> group()\r\n", - "refId": "A" + "rawQuery": true, + "rawSql": "SELECT COUNT(*) AS _value, topic, title\nFROM analytics\nWHERE time >= $__timeFrom \n AND time < $__timeTo\nGROUP BY topic, title", + "refId": "A", + "sql": { + "columns": [ + { + "parameters": [], + "type": "function" + } + ], + "groupBy": [ + { + "property": { + "type": "string" + }, + "type": "groupBy" + } + ] + } } ], "title": "Title-Topic Heatmap", @@ -267,11 +343,13 @@ "mode": "palette-classic" }, "custom": { + "axisBorderShow": false, "axisCenteredZero": false, "axisColorMode": "text", "axisLabel": "", "axisPlacement": "auto", "barAlignment": 0, + "barWidthFactor": 0.6, "drawStyle": "line", "fillOpacity": 30, "gradientMode": "hue", @@ -280,6 +358,7 @@ "tooltip": false, "viz": false }, + "insertNulls": false, "lineInterpolation": "smooth", "lineWidth": 1, "pointSize": 5, @@ -302,8 +381,7 @@ "mode": "absolute", "steps": [ { - "color": "green", - "value": null + "color": "green" }, { "color": "red", @@ -345,18 +423,41 @@ "showLegend": true }, "tooltip": { + "hideZeros": false, "mode": "single", "sort": "none" } }, + "pluginVersion": "12.0.0", "targets": [ { + "dataset": "iox", "datasource": { "type": "influxdb", - "uid": "${datasource}" + "uid": "${DS_INFLUXDB}" }, + "editorMode": "code", + "format": "table", "query": "bucket = \"${INFLUX_BUCKET}\"\r\n\r\nfrom(bucket: bucket)\r\n |> range(start: v.timeRangeStart, stop:v.timeRangeStop)\r\n |> filter(fn: (r) => r._measurement == \"analytics\")\r\n |> filter(fn: (r) => r._field == \"number_request_messages\")\r\n |> group()\r\n |> aggregateWindow(every: v.windowPeriod, fn: count)\r\n |> keep(columns: [\"_time\", \"_value\"])", - "refId": "A" + "rawQuery": true, + "rawSql": "SELECT\n $__dateBin(time) AS time,\n COUNT(*) AS count\nFROM analytics\nWHERE time >= $__timeFrom\n AND time < $__timeTo\nGROUP BY 1", + "refId": "A", + "sql": { + "columns": [ + { + "parameters": [], + "type": "function" + } + ], + "groupBy": [ + { + "property": { + "type": "string" + }, + "type": "groupBy" + } + ] + } } ], "title": "System Usage", @@ -385,8 +486,7 @@ "mode": "absolute", "steps": [ { - "color": "green", - "value": null + "color": "green" }, { "color": "red", @@ -455,16 +555,37 @@ } ] }, - "pluginVersion": "10.0.3", + "pluginVersion": "12.0.0", "targets": [ { + "dataset": "iox", "datasource": { "type": "influxdb", - "uid": "${datasource}" + "uid": "${DS_INFLUXDB}" }, + "editorMode": "code", + "format": "table", "hide": false, "query": "import \"influxdata/influxdb/schema\"\r\n\r\ngetOrDefault = (f, d) => if exists f then f else d\r\n\r\nbucket = \"${INFLUX_BUCKET}\"\r\n\r\nfrom(bucket: bucket)\r\n |> range(start: v.timeRangeStart, stop:v.timeRangeStop)\r\n |> filter(fn: (r) => r._measurement == \"analytics\")\r\n |> filter(fn: (r) => r._field == \"prompt_tokens\" or r._field == \"completion_tokens\" or r._field == \"number_request_messages\" or r._field == \"user_hash\" or r._field == \"price\")\r\n |> schema.fieldsAsCols()\r\n |> group(columns: [\"project_id\", \"deployment\", \"model\", \"user_hash\"])\r\n |> reduce(\r\n fn: (r, accumulator) => ({\r\n request_count: accumulator.request_count + 1,\r\n prompt_tokens: getOrDefault(f: r.prompt_tokens, d: 0) + accumulator.prompt_tokens,\r\n completion_tokens: getOrDefault(f: r.completion_tokens, d: 0) + accumulator.completion_tokens,\r\n money: getOrDefault(f: r.price, d: 0.0) + accumulator.money\r\n }),\r\n identity: {request_count: 0, prompt_tokens: 0, completion_tokens: 0, money: 0.0}\r\n )\r\n |> group(columns: [\"project_id\", \"deployment\", \"model\"])\r\n |> reduce(\r\n fn: (r, accumulator) => ({\r\n user_count: accumulator.user_count + 1,\r\n request_count: r.request_count + accumulator.request_count,\r\n prompt_tokens: r.prompt_tokens + accumulator.prompt_tokens,\r\n completion_tokens: r.completion_tokens + accumulator.completion_tokens,\r\n money: r.money + accumulator.money\r\n }),\r\n identity: {user_count: 0, request_count: 0, prompt_tokens: 0, completion_tokens: 0, money: 0.0}\r\n )\r\n |> group()", - "refId": "A" + "rawQuery": true, + "rawSql": "SELECT\n project_id,\n deployment,\n model,\n COUNT(DISTINCT user_hash) AS user_count,\n COUNT(*) AS request_count,\n SUM(COALESCE(prompt_tokens, 0)) AS prompt_tokens,\n SUM(COALESCE(completion_tokens, 0)) AS completion_tokens,\n SUM(COALESCE(price, 0.0)) AS money\nFROM analytics\nWHERE time >= $__timeFrom AND time < $__timeTo\nGROUP BY project_id, deployment, model", + "refId": "A", + "sql": { + "columns": [ + { + "parameters": [], + "type": "function" + } + ], + "groupBy": [ + { + "property": { + "type": "string" + }, + "type": "groupBy" + } + ] + } } ], "title": "Stats Table", @@ -536,8 +657,7 @@ "mode": "absolute", "steps": [ { - "color": "green", - "value": null + "color": "green" }, { "color": "red", @@ -601,16 +721,37 @@ } ] }, - "pluginVersion": "10.0.3", + "pluginVersion": "12.0.0", "targets": [ { + "dataset": "iox", "datasource": { "type": "influxdb", - "uid": "${datasource}" + "uid": "${DS_INFLUXDB}" }, + "editorMode": "code", + "format": "table", "hide": false, "query": "import \"influxdata/influxdb/schema\"\r\n\r\ngetOrDefault = (f, d) => if exists f then f else d\r\n\r\nbucket = \"${INFLUX_BUCKET}\"\r\n\r\nfrom(bucket: bucket)\r\n |> range(start: v.timeRangeStart, stop:v.timeRangeStop)\r\n |> filter(fn: (r) => r._measurement == \"analytics\")\r\n |> filter(fn: (r) => r._field == \"prompt_tokens\" or r._field == \"completion_tokens\" or r._field == \"number_request_messages\" or r._field == \"price\")\r\n |> schema.fieldsAsCols()\r\n |> group(columns: [\"project_id\"])\r\n |> reduce(\r\n fn: (r, accumulator) => ({\r\n request_count: accumulator.request_count + 1,\r\n prompt_tokens: getOrDefault(f: r.prompt_tokens, d: 0) + accumulator.prompt_tokens,\r\n completion_tokens: getOrDefault(f: r.completion_tokens, d: 0) + accumulator.completion_tokens,\r\n money: getOrDefault(f: r.price, d: 0.0) + accumulator.money\r\n }),\r\n identity: {request_count: 0, prompt_tokens: 0, completion_tokens: 0, money: 0.0}\r\n )\r\n |> group()", - "refId": "A" + "rawQuery": true, + "rawSql": "SELECT\n project_id,\n COUNT(*) AS request_count,\n SUM(COALESCE(prompt_tokens, 0)) AS prompt_tokens,\n SUM(COALESCE(completion_tokens, 0)) AS completion_tokens,\n SUM(COALESCE(price, 0.0)) AS money\nFROM analytics\nWHERE time >= $__timeFrom AND time < $__timeTo\nGROUP BY project_id", + "refId": "A", + "sql": { + "columns": [ + { + "parameters": [], + "type": "function" + } + ], + "groupBy": [ + { + "property": { + "type": "string" + }, + "type": "groupBy" + } + ] + } } ], "title": "Project Stats Table", @@ -677,8 +818,7 @@ "mode": "absolute", "steps": [ { - "color": "green", - "value": null + "color": "green" }, { "color": "red", @@ -742,16 +882,37 @@ } ] }, - "pluginVersion": "10.0.3", + "pluginVersion": "12.0.0", "targets": [ { + "dataset": "iox", "datasource": { "type": "influxdb", - "uid": "${datasource}" + "uid": "${DS_INFLUXDB}" }, + "editorMode": "code", + "format": "table", "hide": false, "query": "import \"influxdata/influxdb/schema\"\r\n\r\ngetOrDefault = (f, d) => if exists f then f else d\r\n\r\nbucket = \"${INFLUX_BUCKET}\"\r\n\r\nfrom(bucket: bucket)\r\n |> range(start: v.timeRangeStart, stop:v.timeRangeStop)\r\n |> filter(fn: (r) => r._measurement == \"analytics\")\r\n |> filter(fn: (r) => r._field == \"prompt_tokens\" or r._field == \"completion_tokens\" or r._field == \"number_request_messages\" or r._field == \"price\")\r\n |> schema.fieldsAsCols()\r\n |> group(columns: [\"model\", \"deployment\"])\r\n |> reduce(\r\n fn: (r, accumulator) => ({\r\n request_count: accumulator.request_count + 1,\r\n prompt_tokens: getOrDefault(f: r.prompt_tokens, d: 0) + accumulator.prompt_tokens,\r\n completion_tokens: getOrDefault(f: r.completion_tokens, d: 0) + accumulator.completion_tokens,\r\n money: getOrDefault(f: r.price, d: 0.0) + accumulator.money\r\n }),\r\n identity: {request_count: 0, prompt_tokens: 0, completion_tokens: 0, money: 0.0}\r\n )\r\n |> group()", - "refId": "A" + "rawQuery": true, + "rawSql": "SELECT\n model,\n deployment,\n COUNT(*) AS request_count,\n SUM(COALESCE(prompt_tokens, 0)) AS prompt_tokens,\n SUM(COALESCE(completion_tokens, 0)) AS completion_tokens,\n SUM(COALESCE(price, 0.0)) AS money\nFROM analytics\nWHERE time >= $__timeFrom AND time < $__timeTo\nGROUP BY model, deployment", + "refId": "A", + "sql": { + "columns": [ + { + "parameters": [], + "type": "function" + } + ], + "groupBy": [ + { + "property": { + "type": "string" + }, + "type": "groupBy" + } + ] + } } ], "title": "Deployment/Model Stats Table", @@ -799,8 +960,7 @@ } ], "refresh": "", - "schemaVersion": 38, - "style": "dark", + "schemaVersion": 41, "tags": [ "DIAL", "ai-dial-analytics-realtime" @@ -808,42 +968,35 @@ "templating": { "list": [ { - "current": { - "selected": true, - "text": "default", - "value": "default" - }, + "current": {}, "description": "InfluxDB datasource", - "hide": 0, "includeAll": false, "label": "Data source", - "multi": false, "name": "datasource", "options": [], "query": "influxdb", - "queryValue": "", "refresh": 1, "regex": "", - "skipUrlSync": false, "type": "datasource" }, { + "current": { + "selected": true, + "text": "default", + "value": "default" + }, "datasource": { "type": "influxdb", "uid": "${datasource}" }, - "definition": "buckets()", + "definition": "import \"strings\"\n\nbuckets()\n |> filter(fn: (r) => not strings.hasPrefix(v: r.name, prefix: \"_\"))", "description": "A InfluxDB bucket name", - "hide": 0, "includeAll": false, - "multi": false, "name": "INFLUX_BUCKET", "options": [], "query": "buckets()", "refresh": 1, "regex": "", - "skipUrlSync": false, - "sort": 0, "type": "query" } ] @@ -856,6 +1009,6 @@ "timezone": "", "title": "DIAL Analytics", "uid": "435858b2-4106-4d11-b6e8-9b0ff107b460", - "version": 1, + "version": 19, "weekStart": "" -} \ No newline at end of file +} diff --git a/dashboards/customized/dial_analytics_mcp.json b/dashboards/customized/dial_analytics_mcp.json index ee08f01..9eb399e 100644 --- a/dashboards/customized/dial_analytics_mcp.json +++ b/dashboards/customized/dial_analytics_mcp.json @@ -1,4 +1,47 @@ { + "__inputs": [ + { + "name": "DS_INFLUXDB", + "label": "InfluxDB", + "description": "", + "type": "datasource", + "pluginId": "influxdb", + "pluginName": "InfluxDB" + } + ], + "__elements": {}, + "__requires": [ + { + "type": "grafana", + "id": "grafana", + "name": "Grafana", + "version": "12.0.0" + }, + { + "type": "datasource", + "id": "influxdb", + "name": "InfluxDB", + "version": "1.0.0" + }, + { + "type": "panel", + "id": "stat", + "name": "Stat", + "version": "" + }, + { + "type": "panel", + "id": "table", + "name": "Table", + "version": "" + }, + { + "type": "panel", + "id": "timeseries", + "name": "Time series", + "version": "" + } + ], "annotations": { "list": [ { @@ -36,8 +79,7 @@ "mode": "absolute", "steps": [ { - "color": "green", - "value": null + "color": "green" }, { "color": "yellow", @@ -65,6 +107,7 @@ "graphMode": "area", "justifyMode": "auto", "orientation": "auto", + "percentChangeColorMode": "standard", "reduceOptions": { "calcs": [ "lastNotNull" @@ -79,12 +122,33 @@ "pluginVersion": "12.0.0", "targets": [ { + "dataset": "iox", "datasource": { "type": "influxdb", - "uid": "${datasource}" + "uid": "${DS_INFLUXDB}" }, + "editorMode": "code", + "format": "table", "query": "bucket = \"${INFLUX_BUCKET}\"\nfrom(bucket: bucket)\n |> range(start: v.timeRangeStart, stop: v.timeRangeStop)\n |> filter(fn: (r) => r._measurement == \"mcp_analytics\")\n |> filter(fn: (r) => r._field == \"upstream\")\n |> count()\n |> group()\n |> sum()", - "refId": "A" + "rawQuery": true, + "rawSql": "SELECT COUNT(*)\nFROM mcp_analytics\nWHERE time >= $__timeFrom AND time < $__timeTo", + "refId": "A", + "sql": { + "columns": [ + { + "parameters": [], + "type": "function" + } + ], + "groupBy": [ + { + "property": { + "type": "string" + }, + "type": "groupBy" + } + ] + } } ], "title": "Total MCP Calls", @@ -105,8 +169,7 @@ "mode": "absolute", "steps": [ { - "color": "green", - "value": null + "color": "green" }, { "color": "yellow", @@ -134,6 +197,7 @@ "graphMode": "area", "justifyMode": "auto", "orientation": "auto", + "percentChangeColorMode": "standard", "reduceOptions": { "calcs": [ "lastNotNull" @@ -148,12 +212,33 @@ "pluginVersion": "12.0.0", "targets": [ { + "dataset": "iox", "datasource": { "type": "influxdb", - "uid": "${datasource}" + "uid": "${DS_INFLUXDB}" }, + "editorMode": "code", + "format": "table", "query": "bucket = \"${INFLUX_BUCKET}\"\nfrom(bucket: bucket)\n |> range(start: v.timeRangeStart, stop: v.timeRangeStop)\n |> filter(fn: (r) => r._measurement == \"mcp_analytics\")\n |> filter(fn: (r) => r._field == \"user_hash\")\n |> keep(columns: [\"_value\"])\n |> distinct(column: \"_value\")\n |> count()", - "refId": "A" + "rawQuery": true, + "rawSql": "SELECT COUNT(DISTINCT user_hash)\nFROM mcp_analytics\nWHERE time >= $__timeFrom AND time < $__timeTo", + "refId": "A", + "sql": { + "columns": [ + { + "parameters": [], + "type": "function" + } + ], + "groupBy": [ + { + "property": { + "type": "string" + }, + "type": "groupBy" + } + ] + } } ], "title": "Unique Users", @@ -174,8 +259,7 @@ "mode": "absolute", "steps": [ { - "color": "green", - "value": null + "color": "green" }, { "color": "yellow", @@ -203,6 +287,7 @@ "graphMode": "area", "justifyMode": "auto", "orientation": "auto", + "percentChangeColorMode": "standard", "reduceOptions": { "calcs": [ "lastNotNull" @@ -217,12 +302,33 @@ "pluginVersion": "12.0.0", "targets": [ { + "dataset": "iox", "datasource": { "type": "influxdb", - "uid": "${datasource}" + "uid": "${DS_INFLUXDB}" }, + "editorMode": "code", + "format": "table", "query": "bucket = \"${INFLUX_BUCKET}\"\nfrom(bucket: bucket)\n |> range(start: v.timeRangeStart, stop: v.timeRangeStop)\n |> filter(fn: (r) => r._measurement == \"mcp_analytics\")\n |> filter(fn: (r) => r.mcp_method == \"tools/call\")\n |> filter(fn: (r) => r._field == \"upstream\")\n |> count()\n |> group()\n |> sum()", - "refId": "A" + "rawQuery": true, + "rawSql": "SELECT COUNT(*)\nFROM mcp_analytics\nWHERE time >= $__timeFrom AND time < $__timeTo\n AND mcp_method = 'tools/call'", + "refId": "A", + "sql": { + "columns": [ + { + "parameters": [], + "type": "function" + } + ], + "groupBy": [ + { + "property": { + "type": "string" + }, + "type": "groupBy" + } + ] + } } ], "title": "Tool Calls", @@ -245,13 +351,14 @@ "axisLabel": "", "axisPlacement": "auto", "barAlignment": 0, + "barWidthFactor": 0.6, "drawStyle": "line", "fillOpacity": 10, "gradientMode": "none", "hideFrom": { + "legend": false, "tooltip": false, - "viz": false, - "legend": false + "viz": false }, "insertNulls": false, "lineInterpolation": "linear", @@ -275,8 +382,7 @@ "mode": "absolute", "steps": [ { - "color": "green", - "value": null + "color": "green" } ] }, @@ -299,6 +405,7 @@ "showLegend": true }, "tooltip": { + "hideZeros": false, "mode": "single", "sort": "none" } @@ -306,12 +413,33 @@ "pluginVersion": "12.0.0", "targets": [ { + "dataset": "iox", "datasource": { "type": "influxdb", - "uid": "${datasource}" + "uid": "${DS_INFLUXDB}" }, + "editorMode": "code", + "format": "time_series", "query": "bucket = \"${INFLUX_BUCKET}\"\nfrom(bucket: bucket)\n |> range(start: v.timeRangeStart, stop: v.timeRangeStop)\n |> filter(fn: (r) => r._measurement == \"mcp_analytics\" and r._field == \"upstream\")\n |> map(fn: (r) => ({\n r with\n project_id: if exists r.project_id then r.project_id else \"undefined\"\n }))\n |> group(columns: [\"project_id\"])\n |> aggregateWindow(every: v.windowPeriod, fn: count)", - "refId": "A" + "rawQuery": true, + "rawSql": "SELECT\n COALESCE(project_id, 'undefined') AS project_id,\n date_bin_gapfill($__interval, time) AS time,\n COUNT(*) AS requests\nFROM mcp_analytics\nWHERE time >= $__timeFrom AND time < $__timeTo\nGROUP BY 1, 2\nORDER BY 2 ASC", + "refId": "A", + "sql": { + "columns": [ + { + "parameters": [], + "type": "function" + } + ], + "groupBy": [ + { + "property": { + "type": "string" + }, + "type": "groupBy" + } + ] + } } ], "title": "Requests per DIAL Project", @@ -334,13 +462,14 @@ "axisLabel": "", "axisPlacement": "auto", "barAlignment": 0, + "barWidthFactor": 0.6, "drawStyle": "line", "fillOpacity": 10, "gradientMode": "none", "hideFrom": { + "legend": false, "tooltip": false, - "viz": false, - "legend": false + "viz": false }, "insertNulls": false, "lineInterpolation": "linear", @@ -364,8 +493,7 @@ "mode": "absolute", "steps": [ { - "color": "green", - "value": null + "color": "green" } ] }, @@ -388,6 +516,7 @@ "showLegend": true }, "tooltip": { + "hideZeros": false, "mode": "single", "sort": "none" } @@ -395,12 +524,33 @@ "pluginVersion": "12.0.0", "targets": [ { + "dataset": "iox", "datasource": { "type": "influxdb", - "uid": "${datasource}" + "uid": "${DS_INFLUXDB}" }, + "editorMode": "code", + "format": "time_series", "query": "bucket = \"${INFLUX_BUCKET}\"\nfrom(bucket: bucket)\n |> range(start: v.timeRangeStart, stop: v.timeRangeStop)\n |> filter(fn: (r) => r._measurement == \"mcp_analytics\" and r._field == \"upstream\")\n |> group(columns: [\"mcp_method\"])\n |> aggregateWindow(every: v.windowPeriod, fn: count)", - "refId": "A" + "rawQuery": true, + "rawSql": "SELECT\n mcp_method,\n date_bin_gapfill($__interval, time) AS time,\n COUNT(*) AS method\nFROM mcp_analytics\nWHERE time >= $__timeFrom AND time < $__timeTo\nGROUP BY 1, 2\nORDER BY 2 ASC", + "refId": "A", + "sql": { + "columns": [ + { + "parameters": [], + "type": "function" + } + ], + "groupBy": [ + { + "property": { + "type": "string" + }, + "type": "groupBy" + } + ] + } } ], "title": "Requests per MCP Method", @@ -425,8 +575,7 @@ "mode": "absolute", "steps": [ { - "color": "green", - "value": null + "color": "green" } ] } @@ -437,56 +586,77 @@ "h": 8, "w": 12, "x": 0, - "y": 20 + "y": 12 }, "id": 7, "options": { - "showHeader": true, "cellHeight": "sm", "footer": { - "show": false, + "countRows": false, + "fields": "", "reducer": [ "sum" ], - "countRows": false, - "fields": "" + "show": false }, + "showHeader": true, "sortBy": [ { - "displayName": "_value", - "desc": true + "desc": true, + "displayName": "_value" } ] }, "pluginVersion": "12.0.0", "targets": [ { + "dataset": "iox", "datasource": { "type": "influxdb", - "uid": "${datasource}" + "uid": "${DS_INFLUXDB}" }, + "editorMode": "code", + "format": "table", "query": "bucket = \"${INFLUX_BUCKET}\"\nfrom(bucket: bucket)\n |> range(start: v.timeRangeStart, stop: v.timeRangeStop)\n |> filter(fn: (r) => r._measurement == \"mcp_analytics\")\n |> filter(fn: (r) => r._field == \"upstream\")\n |> group(columns: [\"deployment\"])\n |> count()\n |> group()\n |> sort(columns: [\"_value\"], desc: true)\n |> keep(columns: [\"deployment\", \"_value\"])", - "refId": "A" + "rawQuery": true, + "rawSql": "SELECT deployment, COUNT(*) AS count\nFROM mcp_analytics\nWHERE time >= $__timeFrom AND time < $__timeTo\nGROUP BY deployment", + "refId": "A", + "sql": { + "columns": [ + { + "parameters": [], + "type": "function" + } + ], + "groupBy": [ + { + "property": { + "type": "string" + }, + "type": "groupBy" + } + ] + } } ], "title": "Calls by Deployment", - "type": "table", "transformations": [ { "id": "organize", "options": { "excludeByName": {}, "indexByName": { - "deployment": 0, - "_value": 1 + "_value": 1, + "deployment": 0 }, "renameByName": { - "deployment": "Deployment", - "_value": "Calls" + "_value": "Calls", + "deployment": "Deployment" } } } - ] + ], + "type": "table" }, { "datasource": { @@ -507,8 +677,7 @@ "mode": "absolute", "steps": [ { - "color": "green", - "value": null + "color": "green" } ] } @@ -519,56 +688,77 @@ "h": 8, "w": 12, "x": 12, - "y": 20 + "y": 12 }, "id": 9, "options": { - "showHeader": true, "cellHeight": "sm", "footer": { - "show": false, + "countRows": false, + "fields": "", "reducer": [ "sum" ], - "countRows": false, - "fields": "" + "show": false }, + "showHeader": true, "sortBy": [ { - "displayName": "_value", - "desc": true + "desc": true, + "displayName": "_value" } ] }, "pluginVersion": "12.0.0", "targets": [ { + "dataset": "iox", "datasource": { "type": "influxdb", - "uid": "${datasource}" + "uid": "${DS_INFLUXDB}" }, + "editorMode": "code", + "format": "table", "query": "bucket = \"${INFLUX_BUCKET}\"\nfrom(bucket: bucket)\n |> range(start: v.timeRangeStart, stop: v.timeRangeStop)\n |> filter(fn: (r) => r._measurement == \"mcp_analytics\")\n |> filter(fn: (r) => r._field == \"mcp_tool_call_name\")\n |> filter(fn: (r) => r._value != \"undefined\" and r._value != \"\")\n |> map(fn: (r) => ({ r with tool_name: r._value }))\n |> group(columns: [\"tool_name\"])\n |> count()\n |> group()\n |> sort(columns: [\"_value\"], desc: true)", - "refId": "A" + "rawQuery": true, + "rawSql": "SELECT mcp_tool_call_name, COUNT(*) AS count\nFROM mcp_analytics\nWHERE time >= $__timeFrom AND time < $__timeTo\n AND mcp_method == 'tools/call'\nGROUP BY mcp_tool_call_name\nORDER BY 2 DESC, 1 DESC", + "refId": "A", + "sql": { + "columns": [ + { + "parameters": [], + "type": "function" + } + ], + "groupBy": [ + { + "property": { + "type": "string" + }, + "type": "groupBy" + } + ] + } } ], "title": "Most Called Tools", - "type": "table", "transformations": [ { "id": "organize", "options": { "excludeByName": {}, "indexByName": { - "tool_name": 0, - "_value": 1 + "_value": 1, + "tool_name": 0 }, "renameByName": { - "tool_name": "Tool Name", - "_value": "Calls" + "_value": "Calls", + "tool_name": "Tool Name" } } } - ] + ], + "type": "table" }, { "datasource": { @@ -589,8 +779,7 @@ "mode": "absolute", "steps": [ { - "color": "green", - "value": null + "color": "green" } ] } @@ -638,40 +827,60 @@ "h": 11, "w": 24, "x": 0, - "y": 28 + "y": 20 }, "id": 10, "options": { - "showHeader": true, "cellHeight": "sm", "footer": { - "show": false, - "reducer": [], "countRows": false, + "enablePagination": true, "fields": "", - "enablePagination": true + "reducer": [], + "show": false }, + "frameIndex": 0, + "showHeader": true, "sortBy": [ { - "displayName": "_time", - "desc": true + "desc": true, + "displayName": "_time" } - ], - "frameIndex": 0 + ] }, "pluginVersion": "12.0.0", "targets": [ { + "dataset": "iox", "datasource": { "type": "influxdb", - "uid": "${datasource}" + "uid": "${DS_INFLUXDB}" }, + "editorMode": "code", + "format": "table", "query": "bucket = \"${INFLUX_BUCKET}\"\nfrom(bucket: bucket)\n |> range(start: v.timeRangeStart, stop: v.timeRangeStop)\n |> filter(fn: (r) => r._measurement == \"mcp_analytics\")\n |> pivot(rowKey:[\"_time\"], columnKey: [\"_field\"], valueColumn: \"_value\")\n |> map(fn: (r) => ({\n _time: r._time,\n project_id: if exists r.project_id then r.project_id else \"undefined\",\n deployment: r.deployment,\n mcp_method: r.mcp_method,\n mcp_tool_call_name: if exists r.mcp_tool_call_name then r.mcp_tool_call_name else \"undefined\",\n trace_id: if exists r.trace_id then r.trace_id else \"\"\n }))\n |> group()\n |> sort(columns: [\"_time\"], desc: true)\n |> limit(n: 100)", - "refId": "A" + "rawQuery": true, + "rawSql": "SELECT\n time as _time,\n COALESCE(project_id, 'undefined') AS project_id,\n deployment,\n mcp_method,\n mcp_tool_call_name,\n trace_id\nFROM mcp_analytics\nORDER BY time DESC\nLIMIT 100", + "refId": "A", + "sql": { + "columns": [ + { + "parameters": [], + "type": "function" + } + ], + "groupBy": [ + { + "property": { + "type": "string" + }, + "type": "groupBy" + } + ] + } } ], "title": "Recent Activity Log", - "type": "table", "transformations": [ { "id": "organize", @@ -679,27 +888,28 @@ "excludeByName": {}, "indexByName": { "_time": 0, - "project_id": 1, "deployment": 2, "mcp_method": 3, "mcp_tool_call_name": 4, + "project_id": 1, "trace_id": 5 }, "renameByName": { "_time": "Time", - "project_id": "Project", "deployment": "Deployment", "mcp_method": "Method", "mcp_tool_call_name": "Tool Name", + "project_id": "Project", "trace_id": "Trace ID" } } } - ] + ], + "type": "table" } ], "refresh": "", - "schemaVersion": 39, + "schemaVersion": 41, "tags": [ "DIAL", "ai-dial-analytics-realtime", @@ -708,15 +918,9 @@ "templating": { "list": [ { - "current": { - "selected": false, - "text": "default", - "value": "default" - }, - "hide": 0, + "current": {}, "includeAll": false, "label": "Data Source", - "multi": false, "name": "datasource", "options": [], "query": "influxdb", @@ -725,15 +929,12 @@ "type": "datasource" }, { - "current": { - "text": "default", - "value": "default" - }, + "current": {}, "datasource": { "type": "influxdb", "uid": "${datasource}" }, - "definition": "buckets()", + "definition": "import \"strings\"\n\nbuckets()\n |> filter(fn: (r) => not strings.hasPrefix(v: r.name, prefix: \"_\"))", "description": "InfluxDB bucket name", "includeAll": false, "label": "Bucket", @@ -754,5 +955,6 @@ "timezone": "", "title": "DIAL Analytics MCP", "uid": "c935f785-18d2-4754-b95f-354d0cdd149f", - "version": 1 -} \ No newline at end of file + "version": 12, + "weekStart": "" +} diff --git a/dashboards/customized/dial_analytics_raw_data.json b/dashboards/customized/dial_analytics_raw_data.json index ebf5bb4..69c6217 100644 --- a/dashboards/customized/dial_analytics_raw_data.json +++ b/dashboards/customized/dial_analytics_raw_data.json @@ -1,10 +1,21 @@ { + "__inputs": [ + { + "name": "DS_INFLUXDB", + "label": "InfluxDB", + "description": "", + "type": "datasource", + "pluginId": "influxdb", + "pluginName": "InfluxDB" + } + ], + "__elements": {}, "__requires": [ { "type": "grafana", "id": "grafana", "name": "Grafana", - "version": "10.0.3" + "version": "12.0.0" }, { "type": "datasource", @@ -35,12 +46,11 @@ } ] }, - "editable": false, + "editable": true, "fiscalYearStartMonth": 0, "graphTooltip": 0, "id": null, "links": [], - "liveNow": false, "panels": [ { "datasource": { @@ -64,8 +74,7 @@ "mode": "absolute", "steps": [ { - "color": "green", - "value": null + "color": "green" }, { "color": "red", @@ -108,15 +117,36 @@ }, "showHeader": true }, - "pluginVersion": "10.0.3", + "pluginVersion": "12.0.0", "targets": [ { + "dataset": "iox", "datasource": { "type": "influxdb", - "uid": "${datasource}" + "uid": "${DS_INFLUXDB}" }, + "editorMode": "code", + "format": "table", "query": "import \"influxdata/influxdb/schema\"\r\nbucket = \"${INFLUX_BUCKET}\"\r\n\r\nfrom(bucket: bucket)\r\n |> range(start: v.timeRangeStart, stop:v.timeRangeStop)\r\n |> filter(fn: (r) => r._measurement == \"analytics\")\r\n |> filter(fn: (r) => r._field == \"number_request_messages\" or r._field == \"price\")\r\n |> keep(columns: [\"_time\", \"_field\", \"_value\", \"response_id\", \"project_id\", \"topic\", \"title\", \"deployment\", \"model\"])\r\n |> schema.fieldsAsCols()\r\n |> keep(columns: [\"_time\", \"project_id\", \"topic\", \"title\", \"deployment\", \"model\", \"price\"])\r\n |> group()\r\n |> sort(columns: [\"_time\"], desc: true)", - "refId": "A" + "rawQuery": true, + "rawSql": "SELECT\n time AS \"_time\",\n project_id,\n topic,\n title,\n deployment,\n model,\n price\nFROM analytics\nWHERE time >= $__timeFrom AND time < $__timeTo\nORDER BY time DESC", + "refId": "A", + "sql": { + "columns": [ + { + "parameters": [], + "type": "function" + } + ], + "groupBy": [ + { + "property": { + "type": "string" + }, + "type": "groupBy" + } + ] + } } ], "title": "Panel Title", @@ -150,8 +180,7 @@ } ], "refresh": "1m", - "schemaVersion": 38, - "style": "dark", + "schemaVersion": 41, "tags": [ "DIAL", "ai-dial-analytics-realtime" @@ -159,42 +188,35 @@ "templating": { "list": [ { - "current": { - "selected": true, - "text": "default", - "value": "default" - }, + "current": {}, "description": "InfluxDB datasource", - "hide": 0, "includeAll": false, "label": "Data source", - "multi": false, "name": "datasource", "options": [], "query": "influxdb", - "queryValue": "", "refresh": 1, "regex": "", - "skipUrlSync": false, "type": "datasource" }, { + "current": { + "selected": true, + "text": "default", + "value": "default" + }, "datasource": { "type": "influxdb", "uid": "${datasource}" }, - "definition": "buckets()", + "definition": "import \"strings\"\n\nbuckets()\n |> filter(fn: (r) => not strings.hasPrefix(v: r.name, prefix: \"_\"))", "description": "A InfluxDB bucket name", - "hide": 0, "includeAll": false, - "multi": false, "name": "INFLUX_BUCKET", "options": [], "query": "buckets()", "refresh": 1, "regex": "", - "skipUrlSync": false, - "sort": 0, "type": "query" } ] @@ -207,6 +229,6 @@ "timezone": "", "title": "DIAL Analytics Raw Data", "uid": "98a34b22-dd32-4e2a-8c4c-81147d2ff3df", - "version": 1, + "version": 4, "weekStart": "" -} \ No newline at end of file +} diff --git a/dashboards/customized/influx2_to_influx3_migration_guide.md b/dashboards/customized/influx2_to_influx3_migration_guide.md new file mode 100644 index 0000000..e1255f6 --- /dev/null +++ b/dashboards/customized/influx2_to_influx3_migration_guide.md @@ -0,0 +1,73 @@ +# Migrating Data from InfluxDB2 to InfluxDB3 + +## Overview + +This document describes the recommended procedure for migrating analytics data from **InfluxDB2** to **InfluxDB3**. + +Only the **`default` bucket** needs to be migrated. + +All other buckets contain **derived / aggregated data** and should **not be migrated**. Instead, they should be recreated using the **backfilling mode** of the [aggregation plugin](./influxdb_v3/README.md). + + +The migration process consists of: + +1. Exporting the `default` bucket from **InfluxDB2** as **Line Protocol** +2. Splitting the export into chunks +3. Uploading the chunks in parallel to **InfluxDB3** + +The instructions below assume that **InfluxDB2** is running using the Bitnami container image: + +```yaml +image: + registry: bitnamilegacy + repository: influxdb + tag: 2.7.11-debian-12-r20 +``` + +In this image the storage engine resides at: + +``` +/bitnami/influxdb +``` + +--- + +## Migration script + +Set the environment variables pointing to the InfluxDB 3 instance. + +```init +INFLUX_URL= +INFLUX_API_TOKEN= +``` + +Run the migration script on the InfluxDB 2 instance. +The script exports the `default` bucket into multiple files and sends them to the InfluxDB 3 instance. + +```sh +set -euo pipefail + +# Get the bucket id of the "default" bucket +BUCKET_ID=$(influx bucket list --org dial --json \ + | jq -r '.[] | select(.name == "default") | .id') + +echo "Bucket id: ${BUCKET_ID}" + +cd /tmp + +# Export and split into chunks 1Mb at most each +influxd inspect export-lp \ + --bucket-id "${BUCKET_ID}" \ + --engine-path /bitnami/influxdb \ + --output-path - \ + | split -d -C 1M - chunk_ + +# Upload chunks to InfluxDB 3 in parallel +find . -maxdepth 1 -name 'chunk_*' -print0 \ + | xargs -0 -n1 -P4 -I{} \ + curl --fail -sS \ + -X POST "${INFLUX_URL}/api/v3/write_lp?db=default&precision=nanosecond&accept_partial=false" \ + -H "Authorization: Bearer ${INFLUX_API_TOKEN}" \ + -H "Content-Type: text/plain; charset=utf-8" \ + --data-binary @{} +``` diff --git a/dashboards/customized/influxdb/tasks/tasks_template.yml b/dashboards/customized/influxdb/tasks/tasks_template.yml index 1c8221b..1875d0b 100644 --- a/dashboards/customized/influxdb/tasks/tasks_template.yml +++ b/dashboards/customized/influxdb/tasks/tasks_template.yml @@ -624,306 +624,306 @@ spec: cron: 0 0 1 * * name: monthly_agg query: |- - import "influxdata/influxdb/schema" - import "experimental" import "date" - import "influxdata/influxdb/secrets" - - - option org = secrets.get(key: "org") - - getOrDefault = (f, d) => if exists f then f else d - - - now_time = now() - - - first_day_current_month = date.truncate(t: now_time, unit: 1mo) - - - start_time = experimental.addDuration(d: -1mo, to: first_day_current_month) - - - end_time = experimental.addDuration(d: -1s, to: first_day_current_month) - - - toBucket_monthly = "default_agg_month" - - - project_api_data = - from(bucket: "default_agg_stats") - |> range(start: start_time, stop: end_time) - |> filter(fn: (r) => r._field == "request_count" or r._field == "price") - |> filter(fn: (r) => r["project_id"] != "") - |> schema.fieldsAsCols() - |> group() - - monthly_api_data = - project_api_data - |> map( - fn: (r) => - ({r with month: date.truncate(t: r._time, unit: 1mo), - // Truncate time to the month - - }), - ) - |> group(columns: ["month", "project_id"]) - // Group by month and user_hash - |> reduce( - fn: (r, accumulator) => - ({ - request_count: getOrDefault(f: r.request_count, d: 0) + accumulator.request_count, - price: getOrDefault(f: r.price, d: 0.0) + accumulator.price, - }), - identity: {request_count: 0, price: 0.0}, - ) - - monthly_model_data = - project_api_data - |> map( - fn: (r) => - ({r with month: date.truncate(t: r._time, unit: 1mo), - // Truncate time to the month - - }), - ) - |> group(columns: ["month", "model"]) - // Group by month and user_hash - |> reduce( - fn: (r, accumulator) => - ({ - request_count: getOrDefault(f: r.request_count, d: 0) + accumulator.request_count, - price: getOrDefault(f: r.price, d: 0.0) + accumulator.price, - }), - identity: {request_count: 0, price: 0.0}, - ) - - total_api_cost = - monthly_api_data - |> group(columns: ["month"]) - // Group by month only - |> sum(column: "price") - // Sum all user costs per month - |> rename(columns: {price: "Total_Cost_Per_Api"}) - - avg_api_cost = - monthly_api_data - |> group(columns: ["month"]) - // Group by month only - |> mean(column: "price") - // Compute the average cost per user - |> rename(columns: {price: "Avg_Cost_Per_Api"}) - - unique_api = - monthly_api_data - |> group(columns: ["month"]) - // Group by month only - |> count(column: "project_id") - // Compute the average cost per user - |> rename(columns: {project_id: "Active_Apis"}) - - total_api_request = - monthly_api_data - |> group(columns: ["month"]) - // Group by month only - |> sum(column: "request_count") - // Compute the average cost per user - |> rename(columns: {request_count: "Total_RC_Per_Api"}) - - avg_api_request = - monthly_api_data - |> group(columns: ["month"]) - // Group by month only - |> mean(column: "request_count") - // Compute the average cost per user - |> rename(columns: {request_count: "Avg_RC_Per_Api"}) - - total_model_cost = - monthly_model_data - |> group(columns: ["month"]) - // Group by month only - |> sum(column: "price") - // Sum all user costs per month - |> rename(columns: {price: "Total_Cost_Per_Model"}) - - avg_model_cost = - monthly_model_data - |> group(columns: ["month"]) - // Group by month only - |> mean(column: "price") - // Compute the average cost per user - |> rename(columns: {price: "Avg_Cost_Per_Model"}) - - kpi_data = - from(bucket: "default_agg_kpi") - |> range(start: start_time, stop: end_time) - |> filter(fn: (r) => r["_measurement"] == "analytics") - |> filter(fn: (r) => r["user_hash"] != "undefined") - |> schema.fieldsAsCols() - |> group() - - user_cost_per_month = - kpi_data - |> map( - fn: (r) => - ({r with month: date.truncate(t: r._time, unit: 1mo), - // Truncate time to the month - - }), - ) - |> group(columns: ["month", "user_hash"]) - // Group by month and user_hash - |> sum(column: "cost") - - total_user_cost = - user_cost_per_month - |> group(columns: ["month"]) - // Group by month only - |> sum(column: "cost") - // Sum all user costs per month - |> rename(columns: {cost: "total_user_cost"}) - - avg_per_user_per_month = - user_cost_per_month - |> group(columns: ["month"]) - // Group by month only - |> mean(column: "cost") - // Compute the average cost per user - |> rename(columns: {cost: "Avg_Cost_Per_User"}) - - unique_user_per_month = - user_cost_per_month - |> group(columns: ["month"]) - // Group by month only - |> count(column: "user_hash") - // Compute the average cost per user - |> rename(columns: {user_hash: "Unique_Users"}) - - final_transformed = - union( - tables: [ - total_user_cost - |> rename(columns: {month: "_time"}) - |> map( - fn: (r) => - ({ - _time: r._time, - _measurement: "analytics", - _field: "total_user_cost", - _value: float(v: r.total_user_cost), - }), - ) - |> keep(columns: ["_time", "_measurement", "_field", "_value"]), - avg_per_user_per_month - |> rename(columns: {month: "_time"}) - |> map( - fn: (r) => - ({ - _time: r._time, - _measurement: "analytics", - _field: "Avg_Cost_Per_User", - _value: float(v: r.Avg_Cost_Per_User), - }), - ) - |> keep(columns: ["_time", "_measurement", "_field", "_value"]), - unique_user_per_month - |> rename(columns: {month: "_time"}) - |> map( - fn: (r) => - ({ - _time: r._time, - _measurement: "analytics", - _field: "Unique_Users", - _value: float(v: r.Unique_Users), - }), - ) - |> keep(columns: ["_time", "_measurement", "_field", "_value"]), - total_api_cost - |> rename(columns: {month: "_time"}) - |> map( - fn: (r) => - ({ - _time: r._time, - _measurement: "analytics", - _field: "Total_Cost_Per_Api", - _value: float(v: r.Total_Cost_Per_Api), - }), - ) - |> keep(columns: ["_time", "_measurement", "_field", "_value"]), - avg_api_cost - |> rename(columns: {month: "_time"}) - |> map( - fn: (r) => - ({ - _time: r._time, - _measurement: "analytics", - _field: "Avg_Cost_Per_Api", - _value: float(v: r.Avg_Cost_Per_Api), - }), - ) - |> keep(columns: ["_time", "_measurement", "_field", "_value"]), - unique_api - |> rename(columns: {month: "_time"}) - |> map( - fn: (r) => - ({ - _time: r._time, - _measurement: "analytics", - _field: "Active_Apis", - _value: float(v: r.Active_Apis), - }), - ) - |> keep(columns: ["_time", "_measurement", "_field", "_value"]), - total_api_request - |> rename(columns: {month: "_time"}) - |> map( - fn: (r) => - ({ - _time: r._time, - _measurement: "analytics", - _field: "Total_RC_Per_Api", - _value: float(v: r.Total_RC_Per_Api), - }), - ) - |> keep(columns: ["_time", "_measurement", "_field", "_value"]), - avg_api_request - |> rename(columns: {month: "_time"}) - |> map( - fn: (r) => - ({ - _time: r._time, - _measurement: "analytics", - _field: "Avg_RC_Per_Api", - _value: float(v: r.Avg_RC_Per_Api), - }), - ) - |> keep(columns: ["_time", "_measurement", "_field", "_value"]), - total_model_cost - |> rename(columns: {month: "_time"}) - |> map( - fn: (r) => - ({ - _time: r._time, - _measurement: "analytics", - _field: "Total_Cost_Per_Model", - _value: float(v: r.Total_Cost_Per_Model), - }), - ) - |> keep(columns: ["_time", "_measurement", "_field", "_value"]), - avg_model_cost - |> rename(columns: {month: "_time"}) - |> map( - fn: (r) => - ({ - _time: r._time, - _measurement: "analytics", - _field: "Avg_Cost_Per_Model", - _value: float(v: r.Avg_Cost_Per_Model), - }), - ) - |> keep(columns: ["_time", "_measurement", "_field", "_value"]), - ], - ) - - final_transformed - |> group() - |> to(bucket: toBucket_monthly, org: org) + import "influxdata/influxdb/schema" + import "experimental" import "date" + import "influxdata/influxdb/secrets" + + + option org = secrets.get(key: "org") + + getOrDefault = (f, d) => if exists f then f else d + + + now_time = now() + + + first_day_current_month = date.truncate(t: now_time, unit: 1mo) + + + start_time = experimental.addDuration(d: -1mo, to: first_day_current_month) + + + end_time = experimental.addDuration(d: -1s, to: first_day_current_month) + + + toBucket_monthly = "default_agg_month" + + + project_api_data = + from(bucket: "default_agg_stats") + |> range(start: start_time, stop: end_time) + |> filter(fn: (r) => r._field == "request_count" or r._field == "price") + |> filter(fn: (r) => r["project_id"] != "") + |> schema.fieldsAsCols() + |> group() + + monthly_api_data = + project_api_data + |> map( + fn: (r) => + ({r with month: date.truncate(t: r._time, unit: 1mo), + // Truncate time to the month + + }), + ) + |> group(columns: ["month", "project_id"]) + // Group by month and user_hash + |> reduce( + fn: (r, accumulator) => + ({ + request_count: getOrDefault(f: r.request_count, d: 0) + accumulator.request_count, + price: getOrDefault(f: r.price, d: 0.0) + accumulator.price, + }), + identity: {request_count: 0, price: 0.0}, + ) + + monthly_model_data = + project_api_data + |> map( + fn: (r) => + ({r with month: date.truncate(t: r._time, unit: 1mo), + // Truncate time to the month + + }), + ) + |> group(columns: ["month", "model"]) + // Group by month and user_hash + |> reduce( + fn: (r, accumulator) => + ({ + request_count: getOrDefault(f: r.request_count, d: 0) + accumulator.request_count, + price: getOrDefault(f: r.price, d: 0.0) + accumulator.price, + }), + identity: {request_count: 0, price: 0.0}, + ) + + total_api_cost = + monthly_api_data + |> group(columns: ["month"]) + // Group by month only + |> sum(column: "price") + // Sum all user costs per month + |> rename(columns: {price: "Total_Cost_Per_Api"}) + + avg_api_cost = + monthly_api_data + |> group(columns: ["month"]) + // Group by month only + |> mean(column: "price") + // Compute the average cost per user + |> rename(columns: {price: "Avg_Cost_Per_Api"}) + + unique_api = + monthly_api_data + |> group(columns: ["month"]) + // Group by month only + |> count(column: "project_id") + // Compute the average cost per user + |> rename(columns: {project_id: "Active_Apis"}) + + total_api_request = + monthly_api_data + |> group(columns: ["month"]) + // Group by month only + |> sum(column: "request_count") + // Compute the average cost per user + |> rename(columns: {request_count: "Total_RC_Per_Api"}) + + avg_api_request = + monthly_api_data + |> group(columns: ["month"]) + // Group by month only + |> mean(column: "request_count") + // Compute the average cost per user + |> rename(columns: {request_count: "Avg_RC_Per_Api"}) + + total_model_cost = + monthly_model_data + |> group(columns: ["month"]) + // Group by month only + |> sum(column: "price") + // Sum all user costs per month + |> rename(columns: {price: "Total_Cost_Per_Model"}) + + avg_model_cost = + monthly_model_data + |> group(columns: ["month"]) + // Group by month only + |> mean(column: "price") + // Compute the average cost per user + |> rename(columns: {price: "Avg_Cost_Per_Model"}) + + kpi_data = + from(bucket: "default_agg_kpi") + |> range(start: start_time, stop: end_time) + |> filter(fn: (r) => r["_measurement"] == "analytics") + |> filter(fn: (r) => r["user_hash"] != "undefined") + |> schema.fieldsAsCols() + |> group() + + user_cost_per_month = + kpi_data + |> map( + fn: (r) => + ({r with month: date.truncate(t: r._time, unit: 1mo), + // Truncate time to the month + + }), + ) + |> group(columns: ["month", "user_hash"]) + // Group by month and user_hash + |> sum(column: "cost") + + total_user_cost = + user_cost_per_month + |> group(columns: ["month"]) + // Group by month only + |> sum(column: "cost") + // Sum all user costs per month + |> rename(columns: {cost: "total_user_cost"}) + + avg_per_user_per_month = + user_cost_per_month + |> group(columns: ["month"]) + // Group by month only + |> mean(column: "cost") + // Compute the average cost per user + |> rename(columns: {cost: "Avg_Cost_Per_User"}) + + unique_user_per_month = + user_cost_per_month + |> group(columns: ["month"]) + // Group by month only + |> count(column: "user_hash") + // Compute the average cost per user + |> rename(columns: {user_hash: "Unique_Users"}) + + final_transformed = + union( + tables: [ + total_user_cost + |> rename(columns: {month: "_time"}) + |> map( + fn: (r) => + ({ + _time: r._time, + _measurement: "analytics", + _field: "total_user_cost", + _value: float(v: r.total_user_cost), + }), + ) + |> keep(columns: ["_time", "_measurement", "_field", "_value"]), + avg_per_user_per_month + |> rename(columns: {month: "_time"}) + |> map( + fn: (r) => + ({ + _time: r._time, + _measurement: "analytics", + _field: "Avg_Cost_Per_User", + _value: float(v: r.Avg_Cost_Per_User), + }), + ) + |> keep(columns: ["_time", "_measurement", "_field", "_value"]), + unique_user_per_month + |> rename(columns: {month: "_time"}) + |> map( + fn: (r) => + ({ + _time: r._time, + _measurement: "analytics", + _field: "Unique_Users", + _value: float(v: r.Unique_Users), + }), + ) + |> keep(columns: ["_time", "_measurement", "_field", "_value"]), + total_api_cost + |> rename(columns: {month: "_time"}) + |> map( + fn: (r) => + ({ + _time: r._time, + _measurement: "analytics", + _field: "Total_Cost_Per_Api", + _value: float(v: r.Total_Cost_Per_Api), + }), + ) + |> keep(columns: ["_time", "_measurement", "_field", "_value"]), + avg_api_cost + |> rename(columns: {month: "_time"}) + |> map( + fn: (r) => + ({ + _time: r._time, + _measurement: "analytics", + _field: "Avg_Cost_Per_Api", + _value: float(v: r.Avg_Cost_Per_Api), + }), + ) + |> keep(columns: ["_time", "_measurement", "_field", "_value"]), + unique_api + |> rename(columns: {month: "_time"}) + |> map( + fn: (r) => + ({ + _time: r._time, + _measurement: "analytics", + _field: "Active_Apis", + _value: float(v: r.Active_Apis), + }), + ) + |> keep(columns: ["_time", "_measurement", "_field", "_value"]), + total_api_request + |> rename(columns: {month: "_time"}) + |> map( + fn: (r) => + ({ + _time: r._time, + _measurement: "analytics", + _field: "Total_RC_Per_Api", + _value: float(v: r.Total_RC_Per_Api), + }), + ) + |> keep(columns: ["_time", "_measurement", "_field", "_value"]), + avg_api_request + |> rename(columns: {month: "_time"}) + |> map( + fn: (r) => + ({ + _time: r._time, + _measurement: "analytics", + _field: "Avg_RC_Per_Api", + _value: float(v: r.Avg_RC_Per_Api), + }), + ) + |> keep(columns: ["_time", "_measurement", "_field", "_value"]), + total_model_cost + |> rename(columns: {month: "_time"}) + |> map( + fn: (r) => + ({ + _time: r._time, + _measurement: "analytics", + _field: "Total_Cost_Per_Model", + _value: float(v: r.Total_Cost_Per_Model), + }), + ) + |> keep(columns: ["_time", "_measurement", "_field", "_value"]), + avg_model_cost + |> rename(columns: {month: "_time"}) + |> map( + fn: (r) => + ({ + _time: r._time, + _measurement: "analytics", + _field: "Avg_Cost_Per_Model", + _value: float(v: r.Avg_Cost_Per_Model), + }), + ) + |> keep(columns: ["_time", "_measurement", "_field", "_value"]), + ], + ) + + final_transformed + |> group() + |> to(bucket: toBucket_monthly, org: org) diff --git a/dashboards/customized/influxdb_v3/.gitignore b/dashboards/customized/influxdb_v3/.gitignore new file mode 100644 index 0000000..5b6c096 --- /dev/null +++ b/dashboards/customized/influxdb_v3/.gitignore @@ -0,0 +1 @@ +config.toml diff --git a/dashboards/customized/influxdb_v3/README.md b/dashboards/customized/influxdb_v3/README.md new file mode 100644 index 0000000..7271f00 --- /dev/null +++ b/dashboards/customized/influxdb_v3/README.md @@ -0,0 +1,203 @@ +# InfluxDB 3 Aggregation Triggers for Analytics Dashboards + +This document explains how aggregated analytics tables are produced in **InfluxDB 3** using **scheduled triggers and a Python plugin**. +These aggregates are used by Grafana dashboards for AI DIAL Realtime Analytics. + +--- + +## Database Layout + +We use two databases: + +| Database | Purpose | +| --------------- | ---------------------------------------------------------- | +| `default` | Stores **raw analytics events** (source of truth). | +| `analytics_agg` | Stores **aggregated / roll-up tables** used by dashboards. | + +--- + +## Aggregation Tables (in `analytics_agg`) + +Aggregated data is written into the following **tables** (tables are created automatically on first write): + +| Table Name | Description | +| --------------------- | -------------------------------------------------------------- | +| `default_agg_stats` | 6-hour aggregates: tokens, price, request counts, unique users | +| `default_agg_topic` | Topic-level aggregates (request counts, tokens, price) | +| `default_agg_topic_2` | Additional topic-level metrics with alternative grouping | +| `default_agg_kpi` | User / project-level KPIs (cost, usage, tokens) | +| `default_agg_chatid` | Request counts grouped by `chat_id` | +| `default_agg_month` | Monthly roll-ups used by KPI dashboards | + +> [!NOTE] +> Tables are **schema-on-write** in InfluxDB 3 — no explicit creation step is required. + +--- + +## Aggregation Logic Overview + +All aggregation logic lives in a **single Python scheduled plugin**, structured as a small [package](./plugin/). + +This plugin: + +* queries raw data from the `default.analytics` table +* writes hourly and monthly roll-ups into `analytics_agg.*` tables +* supports **scheduled execution** and **manual backfills** using the same code paths + +--- + +## Scheduled Triggers + +InfluxDB 3 executes plugins via **triggers**, which can be time-based using `cron:` or `every:` specifications. + +We define **two triggers**, both referencing the same plugin package. + +--- + +### 6-Hour Aggregation Trigger + +Runs at the following UTC times every day: + +```txt +00:02, 06:02, 12:02, 18:02 (UTC) +``` + +The 2-minute offset ensures that all raw data for the previous window is ingested before aggregation runs. + +Trigger spec: + +```txt +cron:2 0 */6 * * * +``` + +What it does: + +* aggregates the previous 6-hour window of raw data +* window bins are aligned to 6-hour boundaries (00:00, 06:00, 12:00, 18:00) which makes the query **idempotent** and backfill straightforward +* writes results into: + + * `default_agg_stats` + * `default_agg_topic` + * `default_agg_topic_2` + * `default_agg_kpi` + * `default_agg_chatid` + +Example trigger creation: + +```sh +influxdb3 create trigger \ + --database default \ + --path analytics_rollups \ + --trigger-spec "cron:2 0 */6 * * *" \ + --trigger-arguments mode=hourly,raw_table=analytics,agg_database=analytics_agg,window_hours=6 \ + analytics_6h_rollups +``` + +--- + +### Monthly Aggregation Trigger + +Runs at: + +```txt +00:00:10 UTC on the 1st day of each month +``` + +The 10-minute offset ensures that the data for the last 6-hours window of the month was been processed by the [hourly trigger](#6-hour-aggregation-trigger) and was inserted into the aggregate tables. + +Trigger spec: + +```txt +cron:10 0 0 1 * * +``` + +What it does: + +* reads from 6-hour aggregate tables +* window bins are aligned to monthly boundaries which makes the query **idempotent** and backfill straightforward +* computes monthly totals, averages, and uniques +* writes results into `default_agg_month` + +Example trigger creation: + +```sh +influxdb3 create trigger \ + --database analytics_agg \ + --path analytics_rollups \ + --trigger-spec "cron:10 0 0 1 * *" \ + --trigger-arguments mode=monthly,agg_database=analytics_agg \ + analytics_monthly_rollups +``` + +--- + +## Backfill Procedure + +Scheduled triggers **do not process historical data automatically**. +If you already have existing raw data, you **must backfill aggregates explicitly**. + +### Built-in Backfill Support + +The plugin supports two optional arguments: + +* `start_time` — ISO 8601 timestamp (inclusive) +* `end_time` — ISO 8601 timestamp (exclusive) + +When provided, these **override the scheduled window**. + +Example (single 6-hour window): + +```text +mode=hourly, +window_hours=6, +raw_table=analytics, +agg_database=analytics_agg, +start_time=2026-01-01T00:00:00Z, +end_time=2026-01-01T06:00:00Z +``` + +--- + +### Recommended Backfill Strategy + +### Step 1: Disable scheduled triggers + +```sh +influxdb3 update trigger analytics_6h_rollups --disable +influxdb3 update trigger analytics_monthly_rollups --disable +``` + +### Step 2: Backfill in deterministic windows + +Run the plugin first in `hourly` mode, then in the `monthly` mode since the results of the former are used as inputs by the latter. + +Set `start_time` and `end_time` parameters spanning the desired windows that are needed to be backfilled. + +Re-running the same window is safe and idempotent, because each aggregate point is written with a deterministic timestamp and the tags uniquely identify the aggregation dimensions. + +### Step 3: Re-enable scheduled triggers + +```bash +influxdb3 update trigger analytics_6h_rollups --enable +influxdb3 update trigger analytics_monthly_rollups --enable +``` + +## Running the plugin manually + +The plugin script could be ran manually on an InfluxDB instance configured by the environment variables: + +|Variable|Description| +|---|---| +|INFLUX_URL|URL to the InfluxDB to write the analytics data| +|INFLUX_API_TOKEN|InfluxDB API Token with the write access to the target database| + +Go to script directory and run it with an optional configuration file: + +```sh +cd dashboards/customized/influxdb_v3 +export INFLUX_URL=... +export INFLUX_API_TOKEN=... +python -m plugin config.toml +``` + +Find the example configuration at `config.example.toml`. diff --git a/dashboards/customized/influxdb_v3/config.example.toml b/dashboards/customized/influxdb_v3/config.example.toml new file mode 100644 index 0000000..29aad08 --- /dev/null +++ b/dashboards/customized/influxdb_v3/config.example.toml @@ -0,0 +1,27 @@ +# Example configuration for the InfluxDB rollup plugin. +# All options are optional; default values are shown below. + +# Rollup mode: "hourly" or "monthly" +# mode = "hourly" + +# Raw source table (in the trigger database) +# raw_table = "analytics" + +# Database where aggregates will be written +# agg_database = "analytics_agg" + +# Hourly window size in hours (only used when mode="hourly") +# Must divide 24 evenly. +# window_hours = 6 + +# Backfill window (ISO 8601 timestamps) +# If both are set, the plugin processes this interval instead of the scheduler time. +# start_time = "2026-01-01T00:00:00Z" +# end_time = "2026-01-01T06:00:00Z" + +# Enable verbose logging (SQL queries and written data points) +# verbose = false + +# Local CLI option (not used by scheduled plugin) +# Prevents writes when running via __main__.py +# readonly = true diff --git a/dashboards/customized/influxdb_v3/plugin/__init__.py b/dashboards/customized/influxdb_v3/plugin/__init__.py new file mode 100644 index 0000000..2c08876 --- /dev/null +++ b/dashboards/customized/influxdb_v3/plugin/__init__.py @@ -0,0 +1,101 @@ +""" +{ + "plugin_type": [ + "scheduled" + ], + "scheduled_args_config": [ + { + "name": "mode", + "example": "hourly", + "description": "Which rollup to run: 'hourly' (default) or 'monthly'.", + "required": false + }, + { + "name": "raw_table", + "example": "analytics", + "description": "Raw source table name (in the trigger database).", + "required": false + }, + { + "name": "agg_database", + "example": "analytics_agg", + "description": "Database to write aggregates into.", + "required": false + }, + { + "name": "window_hours", + "example": "6", + "description": "For mode=hourly: window size in hours. Must be a divisor of 24. (default: 6 => windows are [00,06), [06,12), [12,18), [18,24)).", + "required": false + }, + { + "name": "start_time", + "example": "2026-01-01T00:00:00Z", + "description": "Optional backfill start (ISO 8601). If set with end_time, overrides call_time-derived window.", + "required": false + }, + { + "name": "end_time", + "example": "2026-01-01T06:00:00Z", + "description": "Optional backfill end (ISO 8601)). If set with start_time, overrides call_time-derived window.", + "required": false + } + { + "name": "verbose", + "example": "true", + "description": "Enable verbose logging including SQL queries and data points written to InfluxDB. False by default.", + "required": false + } + ] +} +""" + +import uuid +from datetime import datetime, timezone +from typing import assert_never + +from .config import Config, Mode +from .influx.client_wrapper import InfluxDBClientWrapper +from .influx.types import InfluxDBClient +from .rollup_hourly import run_hourly +from .rollup_monthly import run_monthly +from .utils.concurrency import Exec + + +def process_scheduled_call( + influxdb3_local: InfluxDBClient, + call_time: datetime, + args: dict | None = None, + exec: Exec | None = None, +): + """ + InfluxDB 3 scheduled plugin entrypoint. + """ + config = Config.parse(args) + call_time = call_time.replace(tzinfo=timezone.utc) + + client = InfluxDBClientWrapper( + client=influxdb3_local, + exec=exec, + database=config.input_database, + verbose=config.verbose, + ).add_prefix(f"[{str(uuid.uuid4())}]") + + client.info( + f"scheduled call mode={config.mode.value} call_time={call_time.isoformat()} args={args}" + ) + + try: + match config.mode: + case Mode.MONTHLY: + run_monthly(client, call_time, config) + case Mode.HOURLY: + run_hourly(client, call_time, config) + case _: + assert_never(config.mode) + + client.info("completed OK") + + except Exception as e: + client.error(f"failed: {e}") + raise diff --git a/dashboards/customized/influxdb_v3/plugin/__main__.py b/dashboards/customized/influxdb_v3/plugin/__main__.py new file mode 100644 index 0000000..2fdc65f --- /dev/null +++ b/dashboards/customized/influxdb_v3/plugin/__main__.py @@ -0,0 +1,47 @@ +from __future__ import annotations + +import os +import sys +import tomllib +from datetime import datetime, timezone +from pathlib import Path + +from . import process_scheduled_call +from .config import Config +from .influx.client_http import HTTPInfluxDBClient +from .utils.concurrency import ThreadPoolExec + + +def _get_env(name: str) -> str: + if (v := os.getenv(name)) is None: + raise ValueError(f"Env variable {name!r} is unset") + return v + + +def main() -> None: + if len(sys.argv) > 1: + config_file = Path(sys.argv[1]) + args = tomllib.loads(Path(config_file).read_text()) + else: + args = {} + + readonly = args.pop("readonly", "false").lower() == "true" + + url = _get_env("INFLUX_URL") + token = _get_env("INFLUX_API_TOKEN") + + database = Config.parse(args or {}).input_database + client = HTTPInfluxDBClient( + url=url, token=token, database=database, readonly=readonly + ) + + process_scheduled_call( + influxdb3_local=client, + call_time=datetime.now(timezone.utc), + args=args, + exec=ThreadPoolExec(), + ) + + +if __name__ == "__main__": + main() diff --git a/dashboards/customized/influxdb_v3/plugin/config.py b/dashboards/customized/influxdb_v3/plugin/config.py new file mode 100644 index 0000000..dde78f5 --- /dev/null +++ b/dashboards/customized/influxdb_v3/plugin/config.py @@ -0,0 +1,98 @@ +from dataclasses import dataclass +from datetime import datetime, timezone +from enum import Enum +from typing import Dict, List, assert_never + +from .utils.dates import parse_iso_date +from .utils.window import Window +from .utils.window_roller import HourlyRoller, MonthlyRoller, roll_windows + + +class Mode(Enum): + HOURLY = "hourly" + MONTHLY = "monthly" + + +@dataclass +class Config: + mode: Mode + agg_database: str + + raw_table: str # used only in mode=hourly + window_hours: int # used only in mode=hourly + + start_time: datetime | None + end_time: datetime | None + + verbose: bool + + @classmethod + def parse(cls, d: Dict[str, str] | None) -> "Config": + d = (d or {}).copy() + + mode_s = d.pop("mode", None) or "hourly" + try: + mode = Mode(mode_s.strip().lower()) + except Exception as e: + raise ValueError(f"Unsupported mode: {mode_s!r}") from e + + agg_database = d.pop("agg_database", None) or "analytics_agg" + raw_table = d.pop("raw_table", None) or "analytics" + + start_arg = d.pop("start_time", None) + end_arg = d.pop("end_time", None) + + start_time: datetime | None = ( + parse_iso_date("start_time", start_arg) if start_arg else None + ) + end_time: datetime | None = ( + parse_iso_date("end_time", end_arg) if end_arg else None + ) + + window_hours = int(d.pop("window_hours", None) or 6) + if 24 % window_hours: + raise ValueError( + f"window_hours must divide 24 evenly (got {window_hours})" + ) + + verbose = (d.pop("verbose", None) or "false").lower() == "true" + + if d: + raise ValueError(f"Unexpected config keys: {', '.join(d.keys())}") + + return cls( + mode=mode, + agg_database=agg_database, + raw_table=raw_table, + start_time=start_time, + end_time=end_time, + window_hours=window_hours, + verbose=verbose, + ) + + @property + def input_database(self) -> str: + match self.mode: + case Mode.HOURLY: + return "default" + case Mode.MONTHLY: + return self.agg_database + case _: + assert_never(self.mode) + + def get_windows(self, call_time: datetime) -> List[Window]: + call_time = call_time.astimezone(timezone.utc) + if self.end_time: + call_time = min(call_time, self.end_time) + + match self.mode: + case Mode.HOURLY: + roller = HourlyRoller(window_hours=self.window_hours) + case Mode.MONTHLY: + roller = MonthlyRoller() + case _: + assert_never(self.mode) + + return roll_windows( + roller, start_time=self.start_time, call_time=call_time + ) diff --git a/dashboards/customized/influxdb_v3/plugin/influx/client_http.py b/dashboards/customized/influxdb_v3/plugin/influx/client_http.py new file mode 100644 index 0000000..401e229 --- /dev/null +++ b/dashboards/customized/influxdb_v3/plugin/influx/client_http.py @@ -0,0 +1,75 @@ +import json +from typing import Any, Dict, List +from urllib.request import Request, urlopen + +from .types import InfluxDBClient, LineBuilderProtocol + + +class HTTPInfluxDBClient(InfluxDBClient): + _influxdb_url: str + _influxdb_token: str + _database: str + _readonly: bool + + def __init__(self, *, url: str, token: str, database: str, readonly: bool): + self._database = database + self._influxdb_url = url + self._influxdb_token = token + self._readonly = readonly + + def query(self, query: str) -> list[Dict[str, Any]]: + endpoint = f"{self._influxdb_url}/api/v3/query_sql" + + payload = json.dumps( + {"formats": "json", "db": self._database, "q": query} + ).encode("utf-8") + + req = Request( + endpoint, + data=payload, + method="POST", + headers={ + "Authorization": f"Token {self._influxdb_token}", + "Content-Type": "application/json", + }, + ) + + with urlopen(req) as resp: + rows = json.load(resp) + + return rows + + def write_to_db( + self, + db_name: str, + line: LineBuilderProtocol | List[LineBuilderProtocol], + ) -> None: + if self._readonly: + print("skipping writing...") + return + + endpoint = f"{self._influxdb_url}/api/v3/write_lp?db={db_name}&precision=nanosecond&accept_partial=false&no_sync=false" + + lines = line if isinstance(line, list) else [line] + data = "\n".join(line.build() for line in lines).encode() + + req = Request( + endpoint, + data=data, + method="POST", + headers={ + "Authorization": f"Token {self._influxdb_token}", + "Content-Type": "text/plain", + }, + ) + + with urlopen(req): + # InfluxDB returns 204 on success. + # We are not interested in the response. + pass + + def info(self, msg: str) -> None: + print(msg, flush=True) + + def error(self, msg: str) -> None: + print(msg, flush=True) diff --git a/dashboards/customized/influxdb_v3/plugin/influx/client_wrapper.py b/dashboards/customized/influxdb_v3/plugin/influx/client_wrapper.py new file mode 100644 index 0000000..bf86150 --- /dev/null +++ b/dashboards/customized/influxdb_v3/plugin/influx/client_wrapper.py @@ -0,0 +1,106 @@ +import json +from typing import Any, Dict, List + +from ..utils.concurrency import Exec, SequentialExec +from .client_http import HTTPInfluxDBClient +from .types import InfluxDBClient, LineBuilderProtocol + + +class InfluxDBClientWrapper(InfluxDBClient): + _log_prefix: str + _database: str + _verbose: bool + _client: InfluxDBClient + exec: Exec + + def __init__( + self, + *, + client: InfluxDBClient, + database: str, + verbose: bool, + exec: Exec | None = None, + log_prefix: str = "", + ): + self._database = database + self._log_prefix = log_prefix + self._client = client + self._verbose = verbose + self.exec = exec or SequentialExec() + + def add_prefix(self, log_prefix: str) -> "InfluxDBClientWrapper": + return InfluxDBClientWrapper( + log_prefix=self._log_prefix + log_prefix, + verbose=self._verbose, + database=self._database, + client=self._client, + ) + + def query(self, query: str) -> list[Dict[str, Any]]: + if self._verbose: + self.info(f"SQL query:\n{_prettify(query)}") + + rows = self._client.query(query) + + if self._verbose: + prefix = "\n".join(json.dumps(row) for row in rows[:3]) + self.info( + f"SQL query returned {len(rows):>3} rows. First 3 are:\n{_prettify(prefix)}" + ) + else: + self.info(f"SQL query returned {len(rows):>3} rows.") + + return rows + + def write_to_db(self, db_name: str, line: LineBuilderProtocol) -> None: + if self._verbose: + self.info(f"Writing to database {db_name}: {line.build()}") + else: + self.info(f"Writing to database {db_name}...") + + self._client.write_to_db(db_name, line) + + def write_to_db_many( + self, db_name: str, lines: List[LineBuilderProtocol] + ) -> None: + if not lines: + return + + n = len(lines) + if isinstance(self._client, HTTPInfluxDBClient): + # TODO: missing info logs about the lines being written + self._client.write_to_db(db_name, lines) + else: + for idx, line in enumerate(lines, start=1): + prefix = f"[row|{idx:>2}/{n}]" + self.add_prefix(prefix).write_to_db(db_name, line) + + self.info(f"wrote {n:>3} rows to {db_name}") + + def write_to_db_batched( + self, + db_name: str, + lines: List[LineBuilderProtocol], + *, + batch_size: int, + ) -> None: + n = (len(lines) // batch_size) + bool(len(lines) % batch_size) + for idx in range(1, n + 1): + batch = lines[:batch_size] + lines = lines[batch_size:] + prefix = f"[batch|{idx:>2}/{n}]" + self.add_prefix(prefix).write_to_db_many(db_name, batch) + + def info(self, msg: str) -> None: + self._client.info(f"{self._log_prefix}[INFO] {msg}") + + def error(self, msg: str) -> None: + self._client.error(f"{self._log_prefix}[ERROR] {msg}") + + +def _prettify(text: str) -> str: + return _add_prefix_to_lines(" | ", text.strip()) + + +def _add_prefix_to_lines(prefix: str, text: str) -> str: + return "\n".join(prefix + line for line in text.splitlines()) diff --git a/dashboards/customized/influxdb_v3/plugin/influx/line_builder.py b/dashboards/customized/influxdb_v3/plugin/influx/line_builder.py new file mode 100644 index 0000000..06ae503 --- /dev/null +++ b/dashboards/customized/influxdb_v3/plugin/influx/line_builder.py @@ -0,0 +1,160 @@ +""" +Copied as is from the actual LineBuilder: +https://github.com/influxdata/influxdb/blob/37ff7e6cd4598c312df3688026764a322969c1de/influxdb3_py_api/src/system_py.rs#L488-L613 + +The line protocol itself: +https://docs.influxdata.com/influxdb3/core/reference/line-protocol/ +""" + +from collections import OrderedDict +from typing import Optional + + +class InfluxDBError(Exception): + """Base exception for InfluxDB-related errors""" + + pass + + +class InvalidMeasurementError(InfluxDBError): + """Raised when measurement name is invalid""" + + pass + + +class InvalidKeyError(InfluxDBError): + """Raised when a tag or field key is invalid""" + + pass + + +class InvalidLineError(InfluxDBError): + """Raised when a line protocol string is invalid""" + + pass + + +class LineBuilder: + def __init__(self, measurement: str): + if " " in measurement: + raise InvalidMeasurementError( + "Measurement name cannot contain spaces" + ) + self.measurement = measurement + self.tags: OrderedDict[str, str] = OrderedDict() + self.fields: OrderedDict[str, str] = OrderedDict() + self._timestamp_ns: Optional[int] = None + + def _validate_key(self, key: str, key_type: str) -> None: + """Validate that a key does not contain spaces, commas, or equals signs.""" + if not key: + raise InvalidKeyError(f"{key_type} key cannot be empty") + if " " in key: + raise InvalidKeyError( + f"{key_type} key '{key}' cannot contain spaces" + ) + if "," in key: + raise InvalidKeyError( + f"{key_type} key '{key}' cannot contain commas" + ) + if "=" in key: + raise InvalidKeyError( + f"{key_type} key '{key}' cannot contain equals signs" + ) + + def _escape_measurement(self, value: str) -> str: + """Escape characters in measurement names according to line protocol.""" + return value.replace(",", "\\,").replace(" ", "\\ ") + + def _escape_tag_value(self, value: str) -> str: + """Escape characters in tag values according to line protocol.""" + return ( + value.replace("\\", "\\\\") + .replace(",", "\\,") + .replace("=", "\\=") + .replace(" ", "\\ ") + ) + + def _escape_field_key(self, value: str) -> str: + """Escape characters in field keys according to line protocol.""" + return ( + value.replace("\\", "\\\\") + .replace(",", "\\,") + .replace("=", "\\=") + .replace(" ", "\\ ") + ) + + def tag(self, key: str, value: str) -> "LineBuilder": + """Add a tag to the line protocol.""" + self._validate_key(key, "tag") + self.tags[key] = str(value) + return self + + def uint64_field(self, key: str, value: int) -> "LineBuilder": + """Add an unsigned integer field to the line protocol.""" + self._validate_key(key, "field") + if value < 0: + raise ValueError(f"uint64 field '{key}' cannot be negative") + self.fields[key] = f"{value}u" + return self + + def int64_field(self, key: str, value: int) -> "LineBuilder": + """Add an integer field to the line protocol.""" + self._validate_key(key, "field") + self.fields[key] = f"{value}i" + return self + + def float64_field(self, key: str, value: float) -> "LineBuilder": + """Add a float field to the line protocol.""" + self._validate_key(key, "field") + # Check if value has no decimal component + self.fields[key] = f"{int(value)}.0" if value % 1 == 0 else str(value) + return self + + def string_field(self, key: str, value: str) -> "LineBuilder": + """Add a string field to the line protocol.""" + self._validate_key(key, "field") + # Escape quotes and backslashes in string values + escaped_value = value.replace("\\", "\\\\").replace('"', '\\"') + self.fields[key] = f'"{escaped_value}"' + return self + + def bool_field(self, key: str, value: bool) -> "LineBuilder": + """Add a boolean field to the line protocol.""" + self._validate_key(key, "field") + self.fields[key] = "t" if value else "f" + return self + + def time_ns(self, timestamp_ns: int) -> "LineBuilder": + """Set the timestamp in nanoseconds.""" + self._timestamp_ns = timestamp_ns + return self + + def build(self) -> str: + """Build the line protocol string.""" + # Start with measurement name (escape commas and spaces) + line = self._escape_measurement(self.measurement) + + # Add tags if present + if self.tags: + tags_str = ",".join( + f"{key}={self._escape_tag_value(value)}" + for key, value in self.tags.items() + ) + line += f",{tags_str}" + + # Add fields (required) + if not self.fields: + raise InvalidLineError(f"At least one field is required: {line}") + + fields_str = ",".join( + f"{self._escape_field_key(key)}={value}" + for key, value in self.fields.items() + ) + line += f" {fields_str}" + + # Add timestamp if present + if self._timestamp_ns is not None: + line += f" {self._timestamp_ns}" + + return line diff --git a/dashboards/customized/influxdb_v3/plugin/influx/types.py b/dashboards/customized/influxdb_v3/plugin/influx/types.py new file mode 100644 index 0000000..750f90d --- /dev/null +++ b/dashboards/customized/influxdb_v3/plugin/influx/types.py @@ -0,0 +1,25 @@ +from typing import Any, Dict, Protocol, Self, runtime_checkable + + +# Find the actual implementation in Python: +# https://github.com/influxdata/influxdb/blob/37ff7e6cd4598c312df3688026764a322969c1de/influxdb3_py_api/src/system_py.rs#L488-L613 +@runtime_checkable +class LineBuilderProtocol(Protocol): + def __init__(self, measurement: str) -> None: ... + def time_ns(self, timestamp_ns: int) -> Self: ... + def tag(self, key: str, value: str) -> Self: ... + def string_field(self, key: str, value: str) -> Self: ... + def int64_field(self, key: str, value: int) -> Self: ... + def float64_field(self, key: str, value: float) -> Self: ... + def bool_field(self, key: str, value: bool) -> Self: ... + def build(self) -> str: ... + + +# Find the actual implementation in Rust: +# https://github.com/influxdata/influxdb/blob/37ff7e6cd4598c312df3688026764a322969c1de/influxdb3_py_api/src/system_py.rs#L203 +@runtime_checkable +class InfluxDBClient(Protocol): + def query(self, query: str) -> list[Dict[str, Any]]: ... + def write_to_db(self, db_name: str, line: LineBuilderProtocol) -> None: ... + def info(self, msg: str) -> None: ... + def error(self, msg: str) -> None: ... diff --git a/dashboards/customized/influxdb_v3/plugin/influx/write.py b/dashboards/customized/influxdb_v3/plugin/influx/write.py new file mode 100644 index 0000000..b7bcdf2 --- /dev/null +++ b/dashboards/customized/influxdb_v3/plugin/influx/write.py @@ -0,0 +1,110 @@ +import json +from datetime import datetime, timezone +from typing import Any, Dict, List, Tuple + +from ..utils.dates import parse_iso_date +from .client_http import HTTPInfluxDBClient +from .client_wrapper import InfluxDBClientWrapper +from .line_builder import LineBuilder as LineBuilderImpl +from .types import LineBuilderProtocol + + +def write_points( + client: InfluxDBClientWrapper, + *, + db: str, + table: str, + rows: List[Dict[str, Any]], + time_col: str, + tag_cols: Tuple[str, ...], + field_cols: Tuple[str, ...], +) -> None: + """ + Write each row as a point into db_name.table_name. + time_col can be ISO string, datetime, or ns int. + """ + lines = [ + _create_data_point( + _create_line_builder(client, table), + row, + time_col=time_col, + tag_cols=tag_cols, + field_cols=field_cols, + ) + for row in rows + ] + + client.write_to_db_batched(db, lines, batch_size=50) + + +def _create_data_point( + builder: LineBuilderProtocol, + row: Dict[str, Any], + *, + time_col: str, + tag_cols: Tuple[str, ...], + field_cols: Tuple[str, ...], +) -> LineBuilderProtocol: + r = row.copy() + + def r_json(): + return json.dumps(row) + + t = r.pop(time_col, None) + if isinstance(t, str): + builder.time_ns(_ns(parse_iso_date(f"{time_col!r} column", t))) + else: + raise ValueError( + f"Unexpected time value in the column '{time_col!r}': {t} of type {type(t)}" + ) + + for k in tag_cols: + if (v := r.pop(k, None)) is None: + raise ValueError( + f"Expected to find a tag column {k!r}, but it's missing from the given row: {r_json()}" + ) + if isinstance(v, str): + builder.tag(k, v) + else: + raise ValueError( + f"Unexpected tag value in the column '{k!r}': {v} of type {type(v)}, but tags are expected to always be strings." + ) + + for k in field_cols: + if (v := r.pop(k, None)) is None: + raise ValueError( + f"Expected to find a field column {k!r}, but it's missing from the given row: {r_json()}" + ) + + if isinstance(v, bool): + builder.bool_field(k, v) + elif isinstance(v, int): + builder.int64_field(k, v) + elif isinstance(v, float): + builder.float64_field(k, v) + else: + raise ValueError( + f"Unexpected field value in the column '{k!r}': {v} of type {type(v)}" + ) + + if r: + raise ValueError(f"There are unhandled fields in the row: {r_json()}") + + return builder + + +def _ns(dt: datetime) -> int: + return int(dt.astimezone(timezone.utc).timestamp() * 1_000_000_000) + + +def _create_line_builder( + client: InfluxDBClientWrapper, table_name: str +) -> LineBuilderProtocol: + if isinstance(client._client, HTTPInfluxDBClient): + return LineBuilderImpl(table_name) + else: + # LineBuilder is available in the runtime: + # https://docs.influxdata.com/influxdb3/enterprise/plugins/extend-plugin/#write-data + # The actual LineBuilder code: + # https://github.com/influxdata/influxdb/blob/37ff7e6cd4598c312df3688026764a322969c1de/influxdb3_py_api/src/system_py.rs#L508-L613 + return LineBuilder(table_name) # type: ignore # noqa: F821 diff --git a/dashboards/customized/influxdb_v3/plugin/rollup_hourly.py b/dashboards/customized/influxdb_v3/plugin/rollup_hourly.py new file mode 100644 index 0000000..882d9a4 --- /dev/null +++ b/dashboards/customized/influxdb_v3/plugin/rollup_hourly.py @@ -0,0 +1,256 @@ +from datetime import datetime +from typing import Any, Dict + +from .config import Config +from .influx.client_wrapper import InfluxDBClientWrapper +from .influx.write import write_points +from .utils.window import Window + + +def run_hourly( + client: InfluxDBClientWrapper, call_time: datetime, config: Config +) -> None: + """ + Reads: raw_table + Writes: default_agg_stats, default_agg_topic, default_agg_topic_2, default_agg_kpi, default_agg_chatid + """ + + windows = config.get_windows(call_time) + + n = len(windows) + for idx, window in enumerate(windows, start=1): + run_hourly_window(client.add_prefix(f"[win|{idx}/{n}]"), config, window) + + +def default_agg_stats( + client: InfluxDBClientWrapper, config: Config, start_s: str, in_window: str +): + stats_sql = f""" +SELECT + '{start_s}' AS time, + deployment, + model, + project_id, + parent_deployment, + language, + SUM(prompt_tokens) AS prompt_tokens, + SUM(completion_tokens) AS completion_tokens, + SUM(price) AS price, + SUM(number_request_messages) AS number_request_messages, + SUM(deployment_price) AS deployment_price, + COUNT(*) AS request_count, + COUNT(DISTINCT user_hash) AS unique_user_count +FROM {config.raw_table} +WHERE {in_window} +GROUP BY deployment, model, project_id, parent_deployment, language +""" + + stats_rows = client.query(stats_sql) + write_points( + client, + db=config.agg_database, + table="default_agg_stats", + rows=_normalize_project_id(stats_rows), + time_col="time", + tag_cols=( + "deployment", + "model", + "project_id", + "parent_deployment", + "language", + ), + field_cols=( + "prompt_tokens", + "completion_tokens", + "price", + "number_request_messages", + "deployment_price", + "request_count", + "unique_user_count", + ), + ) + + +def default_agg_topic_2( + client: InfluxDBClientWrapper, config: Config, start_s: str, in_window: str +): + topic_sql = f""" +SELECT + '{start_s}' AS time, + title, + topic, + model, + COUNT(*) AS topic_count, + SUM(number_request_messages) AS number_request_messages, + SUM(price) AS price, + SUM(prompt_tokens) AS prompt_tokens, + SUM(completion_tokens) AS completion_tokens +FROM {config.raw_table} +WHERE {in_window} +GROUP BY title, topic, model +""" + + topic_rows = client.query(topic_sql) + write_points( + client, + db=config.agg_database, + table="default_agg_topic_2", + rows=_normalize_topic(topic_rows), + time_col="time", + tag_cols=("title", "topic", "model"), + field_cols=( + "topic_count", + "number_request_messages", + "price", + "prompt_tokens", + "completion_tokens", + ), + ) + + +def default_agg_topic( + client: InfluxDBClientWrapper, config: Config, start_s: str, in_window: str +): + token_sql = f""" +SELECT + '{start_s}' AS time, + CASE + WHEN user_hash = 'undefined' + THEN 'project' + ELSE 'user' + END AS user_type, + SUM(CAST(50000 <= prompt_tokens AS INT)) AS class_1, + SUM(CAST(10000 < prompt_tokens AND prompt_tokens < 50000 AS INT)) AS class_2, + SUM(CAST( 5000 < prompt_tokens AND prompt_tokens < 10000 AS INT)) AS class_3, + SUM(CAST( 1000 < prompt_tokens AND prompt_tokens < 5000 AS INT)) AS class_4, + SUM(CAST( 100 < prompt_tokens AND prompt_tokens < 1000 AS INT)) AS class_5, + SUM(CAST( prompt_tokens <= 100 AS INT)) AS class_6 +FROM {config.raw_table} +WHERE {in_window} +GROUP BY user_type +""" + + token_rows = client.query(token_sql) + write_points( + client, + db=config.agg_database, + table="default_agg_topic", + rows=token_rows, + time_col="time", + tag_cols=("user_type",), + field_cols=( + "class_1", + "class_2", + "class_3", + "class_4", + "class_5", + "class_6", + ), + ) + + +def default_agg_kpi( + client: InfluxDBClientWrapper, config: Config, start_s: str, in_window: str +): + kpi_sql = f""" +SELECT + '{start_s}' AS time, + user_hash, + project_id, + parent_deployment, + title, + COUNT(*) AS request_count, + SUM(completion_tokens) AS completion_tokens, + SUM(prompt_tokens) AS prompt_tokens, + SUM(price) AS cost +FROM {config.raw_table} +WHERE {in_window} +GROUP BY user_hash, project_id, parent_deployment, title +""" + + kpi_rows = client.query(kpi_sql) + write_points( + client, + db=config.agg_database, + table="default_agg_kpi", + rows=_normalize_project_id(kpi_rows), + time_col="time", + tag_cols=("user_hash", "project_id", "parent_deployment", "title"), + field_cols=( + "cost", + "request_count", + "completion_tokens", + "prompt_tokens", + ), + ) + + +def default_agg_chatid( + client: InfluxDBClientWrapper, config: Config, start_s: str, in_window: str +): + chat_sql = f""" +SELECT + '{start_s}' AS time, + chat_id, + COUNT(*) AS request_count +FROM {config.raw_table} +WHERE {in_window} +GROUP BY chat_id +""" + + chat_rows = client.query(chat_sql) + write_points( + client, + db=config.agg_database, + table="default_agg_chatid", + rows=chat_rows, + time_col="time", + tag_cols=("chat_id",), + field_cols=("request_count",), + ) + + +def run_hourly_window( + client: InfluxDBClientWrapper, config: Config, window: Window +) -> None: + client.info( + f"{config.window_hours}-hours rollup window: {window.display()}" + ) + + start_s = window.start_s + in_window = window.in_window_sql() + + job_defs = [ + ("default_agg_stats", default_agg_stats), + ("default_agg_topic_2", default_agg_topic_2), + ("default_agg_topic", default_agg_topic), + ("default_agg_kpi", default_agg_kpi), + ("default_agg_chatid", default_agg_chatid), + ] + + jobs = [ + lambda client=client, name=name, func=func: func( + client.add_prefix(f"[{name:<19}]"), config, start_s, in_window + ) + for name, func in job_defs + ] + + client.exec.run(jobs) + + +def _normalize_project_id(rows: list[Dict[str, Any]]) -> list[Dict[str, Any]]: + # Analytics may produce data points with missing (NULL) project_id field. + # We are filling the gaps with a default to ease further processing of the data. + for r in rows: + if r.get("project_id") is None: + r["project_id"] = "undefined" + return rows + + +def _normalize_topic(rows: list[Dict[str, Any]]) -> list[Dict[str, Any]]: + # Analytics may produce data points with missing (NULL) topic field. + # We are filling the gaps with a default to ease further processing of the data. + for r in rows: + if r.get("topic") is None: + r["topic"] = "undefined" + return rows diff --git a/dashboards/customized/influxdb_v3/plugin/rollup_monthly.py b/dashboards/customized/influxdb_v3/plugin/rollup_monthly.py new file mode 100644 index 0000000..1e133f0 --- /dev/null +++ b/dashboards/customized/influxdb_v3/plugin/rollup_monthly.py @@ -0,0 +1,134 @@ +from datetime import datetime +from typing import Any, Dict + +from .config import Config +from .influx.client_wrapper import InfluxDBClientWrapper +from .influx.write import write_points +from .utils.window import Window + + +def run_monthly( + client: InfluxDBClientWrapper, call_time: datetime, config: Config +) -> None: + """ + Reads: default_agg_stats, default_agg_kpi + Writes: default_agg_month + """ + windows = config.get_windows(call_time) + + n = len(windows) + for idx, window in enumerate(windows, start=1): + run_monthly_window( + client.add_prefix(f"[win|{idx}/{n}]"), config, window + ) + + +def run_monthly_window( + client: InfluxDBClientWrapper, config: Config, window: Window +) -> None: + client.info(f"monthly rollup window: {window.display()}") + + start_s = window.start_s + in_window = window.in_window_sql() + + # API-level (project_id) rollups from default_agg_stats + api_sql = f""" +SELECT + '{start_s}' AS time, + COALESCE(SUM(price),0) AS total_cost_per_api, + COALESCE(AVG(price),0) AS avg_cost_per_api, + COALESCE(SUM(request_count),0) AS total_rc_per_api, + COALESCE(AVG(request_count),0) AS avg_rc_per_api, + COUNT(*) AS active_apis +FROM ({_get_stats_sub_table("project_id", in_window)}) +""" + + # Model-level rollups from default_agg_stats + model_sql = f""" +SELECT + '{start_s}' AS time, + COALESCE(SUM(price),0) AS total_cost_per_model, + COALESCE(AVG(price),0) AS avg_cost_per_model, + COUNT(*) AS active_models +FROM ({_get_stats_sub_table("model", in_window)}) +""" + + # User-level rollups from default_agg_kpi + user_sql = f""" +SELECT + '{start_s}' AS time, + COALESCE(SUM(cost),0) AS total_user_cost, + COALESCE(AVG(cost),0) AS avg_cost_per_user, + COUNT(*) AS unique_users +FROM ({_get_kpi_sub_table(in_window)}) +""" + + jobs = [ + lambda: client.add_prefix("[api ]").query(api_sql), + lambda: client.add_prefix("[model]").query(model_sql), + lambda: client.add_prefix("[user ]").query(user_sql), + ] + + results = client.exec.run(jobs) + + merged: Dict[str, Any] = {"time": start_s} + for result in results: + if result: + merged.update(result[0]) + + if ( + merged.get("unique_users", 0) + + merged.get("active_apis", 0) + + merged.get("active_models", 0) + == 0 + ): + client.info("No data. Skipping.") + return + + merged.pop("active_models", None) + + write_points( + client, + db=config.agg_database, + table="default_agg_month", + rows=[merged], + time_col="time", + tag_cols=(), + field_cols=( + "total_user_cost", + "avg_cost_per_user", + "unique_users", + "total_cost_per_api", + "avg_cost_per_api", + "active_apis", + "total_rc_per_api", + "avg_rc_per_api", + "total_cost_per_model", + "avg_cost_per_model", + ), + ) + + +def _get_stats_sub_table(group_by: str, in_window: str) -> str: + return f""" +SELECT + SUM(price) AS price, + SUM(request_count) AS request_count +FROM default_agg_stats +WHERE {in_window} + AND project_id IS NOT NULL + AND project_id <> '' +GROUP BY {group_by} +""".strip() + + +def _get_kpi_sub_table(in_window: str) -> str: + return f""" +SELECT + SUM(cost) AS cost +FROM default_agg_kpi +WHERE {in_window} + AND user_hash IS NOT NULL + AND user_hash <> 'undefined' +GROUP BY user_hash +""".strip() diff --git a/dashboards/customized/influxdb_v3/plugin/utils/concurrency.py b/dashboards/customized/influxdb_v3/plugin/utils/concurrency.py new file mode 100644 index 0000000..40a83f4 --- /dev/null +++ b/dashboards/customized/influxdb_v3/plugin/utils/concurrency.py @@ -0,0 +1,22 @@ +from concurrent.futures import ThreadPoolExecutor +from typing import Callable, Iterable, Protocol, TypeVar + +_T = TypeVar("_T") + + +class Exec(Protocol): + def run(self, jobs: Iterable[Callable[[], _T]]) -> list[_T]: ... + + +class SequentialExec: + def run(self, jobs: Iterable[Callable[[], _T]]) -> list[_T]: + return [job() for job in jobs] + + +class ThreadPoolExec: + def __init__(self, max_workers: int = 8): + self._max_workers = max_workers + + def run(self, jobs: Iterable[Callable[[], _T]]) -> list[_T]: + with ThreadPoolExecutor(max_workers=self._max_workers) as ex: + return list(ex.map(lambda job: job(), jobs)) diff --git a/dashboards/customized/influxdb_v3/plugin/utils/dates.py b/dashboards/customized/influxdb_v3/plugin/utils/dates.py new file mode 100644 index 0000000..85d21e1 --- /dev/null +++ b/dashboards/customized/influxdb_v3/plugin/utils/dates.py @@ -0,0 +1,20 @@ +from datetime import datetime, timezone + + +def parse_iso_date(name: str, value: str) -> datetime: + try: + # In Python<=3.10, "Z" was not supported by fromisoformat() + dt: datetime = datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError: + raise ValueError( + f"Invalid ISO 8601 datetime from the {name!r} column: {value!r}." + ) + if dt.tzinfo is None: + raise ValueError( + f"Date from the {name!r} column must include timezone info (e.g., '+00:00'): {value!r}" + ) + return dt.astimezone(timezone.utc) + + +def to_iso_date(dt: datetime) -> str: + return dt.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") diff --git a/dashboards/customized/influxdb_v3/plugin/utils/window.py b/dashboards/customized/influxdb_v3/plugin/utils/window.py new file mode 100644 index 0000000..cd60f52 --- /dev/null +++ b/dashboards/customized/influxdb_v3/plugin/utils/window.py @@ -0,0 +1,24 @@ +from dataclasses import dataclass +from datetime import datetime + +from .dates import to_iso_date + + +@dataclass +class Window: + start: datetime + end: datetime + + def display(self) -> str: + return f"[{self.start_s} .. {self.end_s})" + + def in_window_sql(self) -> str: + return f"'{self.start_s}' <= time AND time < '{self.end_s}'" + + @property + def start_s(self) -> str: + return to_iso_date(self.start) + + @property + def end_s(self) -> str: + return to_iso_date(self.end) diff --git a/dashboards/customized/influxdb_v3/plugin/utils/window_roller.py b/dashboards/customized/influxdb_v3/plugin/utils/window_roller.py new file mode 100644 index 0000000..afdbab0 --- /dev/null +++ b/dashboards/customized/influxdb_v3/plugin/utils/window_roller.py @@ -0,0 +1,56 @@ +from datetime import datetime, timedelta +from typing import List, Protocol + +from .window import Window + + +class WindowRoller(Protocol): + def align(self, call_time: datetime) -> datetime: ... + def prev(self, time: datetime) -> datetime: ... + + +class HourlyRoller: + def __init__(self, window_hours: int): + if 24 % window_hours: + raise ValueError("window_hours must divide 24 evenly") + self.window_hours = window_hours + + def align(self, call_time: datetime) -> datetime: + hour = call_time.hour - (call_time.hour % self.window_hours) + return call_time.replace(hour=hour, minute=0, second=0, microsecond=0) + + def prev(self, time: datetime) -> datetime: + return time - timedelta(hours=self.window_hours) + + +class MonthlyRoller: + def align(self, call_time: datetime) -> datetime: + return self._month_start(call_time) + + def prev(self, time: datetime) -> datetime: + return self._month_start(time - timedelta(minutes=1)) + + @staticmethod + def _month_start(dt: datetime) -> datetime: + return dt.replace(day=1, hour=0, minute=0, second=0, microsecond=0) + + +def roll_windows( + roller: WindowRoller, *, start_time: datetime | None, call_time: datetime +) -> List[Window]: + + end = roller.align(call_time) + start = roller.prev(end) + + if start_time is None: + return [Window(start=start, end=end)] + + ret: List[Window] = [] + while start_time <= start: + ret.append(Window(start=start, end=end)) + end = start + start = roller.prev(end) + + ret = ret[::-1] + + return ret diff --git a/dashboards/public/dial_analytics.json b/dashboards/public/dial_analytics.json index ba3ca82..92cc4a0 100644 --- a/dashboards/public/dial_analytics.json +++ b/dashboards/public/dial_analytics.json @@ -723,4 +723,4 @@ "uid": "435858b2-4106-4d11-b6e8-9b0ff107b470", "version": 1, "weekStart": "" -} +} \ No newline at end of file diff --git a/dashboards/public/dial_analytics_raw_data.json b/dashboards/public/dial_analytics_raw_data.json index e16aa85..d7467d5 100644 --- a/dashboards/public/dial_analytics_raw_data.json +++ b/dashboards/public/dial_analytics_raw_data.json @@ -1,212 +1,212 @@ { - "__requires": [ - { - "type": "grafana", - "id": "grafana", - "name": "Grafana", - "version": "10.0.3" - }, - { - "type": "datasource", - "id": "influxdb", - "name": "InfluxDB", - "version": "1.0.0" - }, - { - "type": "panel", - "id": "table", - "name": "Table", - "version": "" - } - ], - "annotations": { - "list": [ - { - "builtIn": 1, - "datasource": { - "type": "grafana", - "uid": "-- Grafana --" - }, - "enable": true, - "hide": true, - "iconColor": "rgba(0, 211, 255, 1)", - "name": "Annotations & Alerts", - "type": "dashboard" - } - ] + "__requires": [ + { + "type": "grafana", + "id": "grafana", + "name": "Grafana", + "version": "10.0.3" }, - "editable": false, - "fiscalYearStartMonth": 0, - "graphTooltip": 0, - "id": null, - "links": [], - "liveNow": false, - "panels": [ + { + "type": "datasource", + "id": "influxdb", + "name": "InfluxDB", + "version": "1.0.0" + }, + { + "type": "panel", + "id": "table", + "name": "Table", + "version": "" + } + ], + "annotations": { + "list": [ { + "builtIn": 1, "datasource": { - "type": "influxdb", - "uid": "${datasource}" + "type": "grafana", + "uid": "-- Grafana --" }, - "fieldConfig": { - "defaults": { - "color": { - "mode": "thresholds" - }, - "custom": { - "align": "auto", - "cellOptions": { - "type": "auto" - }, - "inspect": false + "enable": true, + "hide": true, + "iconColor": "rgba(0, 211, 255, 1)", + "name": "Annotations & Alerts", + "type": "dashboard" + } + ] + }, + "editable": false, + "fiscalYearStartMonth": 0, + "graphTooltip": 0, + "id": null, + "links": [], + "liveNow": false, + "panels": [ + { + "datasource": { + "type": "influxdb", + "uid": "${datasource}" + }, + "fieldConfig": { + "defaults": { + "color": { + "mode": "thresholds" + }, + "custom": { + "align": "auto", + "cellOptions": { + "type": "auto" }, - "mappings": [], - "thresholds": { - "mode": "absolute", - "steps": [ - { - "color": "green", - "value": null - }, - { - "color": "red", - "value": 80 - } - ] - } + "inspect": false }, - "overrides": [ - { - "matcher": { - "id": "byName", - "options": "price" + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null }, - "properties": [ - { - "id": "unit", - "value": "currencyUSD" - } - ] - } - ] - }, - "gridPos": { - "h": 24, - "w": 24, - "x": 0, - "y": 0 - }, - "id": 2, - "options": { - "cellHeight": "sm", - "footer": { - "countRows": false, - "fields": "", - "reducer": [ - "sum" - ], - "show": false - }, - "showHeader": true + { + "color": "red", + "value": 80 + } + ] + } }, - "pluginVersion": "10.0.3", - "targets": [ + "overrides": [ { - "datasource": { - "type": "influxdb", - "uid": "${datasource}" + "matcher": { + "id": "byName", + "options": "price" }, - "query": "import \"influxdata/influxdb/schema\"\r\nbucket = \"${INFLUX_BUCKET}\"\r\n\r\nfrom(bucket: bucket)\r\n |> range(start: v.timeRangeStart, stop:v.timeRangeStop)\r\n |> filter(fn: (r) => r._measurement == \"analytics\")\r\n |> filter(fn: (r) => r._field == \"number_request_messages\" or r._field == \"price\")\r\n |> keep(columns: [\"_time\", \"_field\", \"_value\", \"response_id\", \"project_id\", \"deployment\", \"model\"])\r\n |> schema.fieldsAsCols()\r\n |> keep(columns: [\"_time\", \"project_id\", \"deployment\", \"model\", \"price\"])\r\n |> group()\r\n |> sort(columns: [\"_time\"], desc: true)", - "refId": "A" - } - ], - "title": "Panel Title", - "transformations": [ - { - "id": "organize", - "options": { - "excludeByName": {}, - "indexByName": { - "_time": 0, - "deployment": 4, - "model": 5, - "price": 6, - "project_id": 1, - "title": 3, - "topic": 2 - }, - "renameByName": { - "_time": "Time (UTC)", - "deployment": "Deployment", - "model": "Model", - "price": "Price", - "project_id": "Project", - "title": "Title", - "topic": "Topic" + "properties": [ + { + "id": "unit", + "value": "currencyUSD" } - } + ] } - ], - "type": "table" - } - ], - "refresh": "1m", - "schemaVersion": 38, - "style": "dark", - "tags": [ - "DIAL", - "ai-dial-analytics-realtime" - ], - "templating": { - "list": [ - { - "current": { - "selected": true, - "text": "default", - "value": "default" - }, - "description": "InfluxDB datasource", - "hide": 0, - "includeAll": false, - "label": "Data source", - "multi": false, - "name": "datasource", - "options": [], - "query": "influxdb", - "queryValue": "", - "refresh": 1, - "regex": "", - "skipUrlSync": false, - "type": "datasource" + ] + }, + "gridPos": { + "h": 24, + "w": 24, + "x": 0, + "y": 0 + }, + "id": 2, + "options": { + "cellHeight": "sm", + "footer": { + "countRows": false, + "fields": "", + "reducer": [ + "sum" + ], + "show": false }, + "showHeader": true + }, + "pluginVersion": "10.0.3", + "targets": [ { "datasource": { "type": "influxdb", "uid": "${datasource}" }, - "definition": "buckets()", - "description": "A InfluxDB bucket name", - "hide": 0, - "includeAll": false, - "multi": false, - "name": "INFLUX_BUCKET", - "options": [], - "query": "buckets()", - "refresh": 1, - "regex": "", - "skipUrlSync": false, - "sort": 0, - "type": "query" + "query": "import \"influxdata/influxdb/schema\"\r\nbucket = \"${INFLUX_BUCKET}\"\r\n\r\nfrom(bucket: bucket)\r\n |> range(start: v.timeRangeStart, stop:v.timeRangeStop)\r\n |> filter(fn: (r) => r._measurement == \"analytics\")\r\n |> filter(fn: (r) => r._field == \"number_request_messages\" or r._field == \"price\")\r\n |> keep(columns: [\"_time\", \"_field\", \"_value\", \"response_id\", \"project_id\", \"deployment\", \"model\"])\r\n |> schema.fieldsAsCols()\r\n |> keep(columns: [\"_time\", \"project_id\", \"deployment\", \"model\", \"price\"])\r\n |> group()\r\n |> sort(columns: [\"_time\"], desc: true)", + "refId": "A" } - ] - }, - "time": { - "from": "now-2d", - "to": "now" - }, - "timepicker": {}, - "timezone": "", - "title": "DIAL Analytics Raw Data", - "uid": "98a34b22-dd32-4e2a-8c4c-81147d2ff3ef", - "version": 1, - "weekStart": "" - } + ], + "title": "Panel Title", + "transformations": [ + { + "id": "organize", + "options": { + "excludeByName": {}, + "indexByName": { + "_time": 0, + "deployment": 4, + "model": 5, + "price": 6, + "project_id": 1, + "title": 3, + "topic": 2 + }, + "renameByName": { + "_time": "Time (UTC)", + "deployment": "Deployment", + "model": "Model", + "price": "Price", + "project_id": "Project", + "title": "Title", + "topic": "Topic" + } + } + } + ], + "type": "table" + } + ], + "refresh": "1m", + "schemaVersion": 38, + "style": "dark", + "tags": [ + "DIAL", + "ai-dial-analytics-realtime" + ], + "templating": { + "list": [ + { + "current": { + "selected": true, + "text": "default", + "value": "default" + }, + "description": "InfluxDB datasource", + "hide": 0, + "includeAll": false, + "label": "Data source", + "multi": false, + "name": "datasource", + "options": [], + "query": "influxdb", + "queryValue": "", + "refresh": 1, + "regex": "", + "skipUrlSync": false, + "type": "datasource" + }, + { + "datasource": { + "type": "influxdb", + "uid": "${datasource}" + }, + "definition": "buckets()", + "description": "A InfluxDB bucket name", + "hide": 0, + "includeAll": false, + "multi": false, + "name": "INFLUX_BUCKET", + "options": [], + "query": "buckets()", + "refresh": 1, + "regex": "", + "skipUrlSync": false, + "sort": 0, + "type": "query" + } + ] + }, + "time": { + "from": "now-2d", + "to": "now" + }, + "timepicker": {}, + "timezone": "", + "title": "DIAL Analytics Raw Data", + "uid": "98a34b22-dd32-4e2a-8c4c-81147d2ff3ef", + "version": 1, + "weekStart": "" +} \ No newline at end of file