mirror of
https://github.com/openziti/zrok.git
synced 2024-11-07 08:44:14 +01:00
118 lines
3.1 KiB
Go
118 lines
3.1 KiB
Go
package controller
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"github.com/openziti/zrok/controller/store"
|
|
"github.com/sirupsen/logrus"
|
|
"strconv"
|
|
)
|
|
|
|
func sparkDataForEnvironments(envs []*store.Environment) (rx, tx map[int][]int64, err error) {
|
|
rx = make(map[int][]int64)
|
|
tx = make(map[int][]int64)
|
|
if len(envs) > 0 {
|
|
qapi := idb.QueryAPI(cfg.Metrics.Influx.Org)
|
|
|
|
envFilter := "|> filter(fn: (r) =>"
|
|
for i, env := range envs {
|
|
if i > 0 {
|
|
envFilter += " or"
|
|
}
|
|
envFilter += fmt.Sprintf(" r[\"envId\"] == \"%d\"", env.Id)
|
|
}
|
|
envFilter += ")"
|
|
query := fmt.Sprintf("from(bucket: \"%v\")\n", cfg.Metrics.Influx.Bucket) +
|
|
"|> range(start: -5m)\n" +
|
|
"|> filter(fn: (r) => r[\"_measurement\"] == \"xfer\")\n" +
|
|
"|> filter(fn: (r) => r[\"_field\"] == \"rx\" or r[\"_field\"] == \"tx\")\n" +
|
|
"|> filter(fn: (r) => r[\"namespace\"] == \"backend\")\n" +
|
|
envFilter +
|
|
"|> drop(columns: [\"share\", \"acctId\"])\n" +
|
|
"|> aggregateWindow(every: 10s, fn: sum, createEmpty: true)\n"
|
|
|
|
result, err := qapi.Query(context.Background(), query)
|
|
if err != nil {
|
|
return nil, nil, err
|
|
}
|
|
|
|
for result.Next() {
|
|
envIdS := result.Record().ValueByKey("envId").(string)
|
|
envId, err := strconv.ParseInt(envIdS, 10, 32)
|
|
if err != nil {
|
|
logrus.Errorf("error parsing '%v': %v", envIdS, err)
|
|
continue
|
|
}
|
|
switch result.Record().Field() {
|
|
case "rx":
|
|
rxV := int64(0)
|
|
if v, ok := result.Record().Value().(int64); ok {
|
|
rxV = v
|
|
}
|
|
rxData := append(rx[int(envId)], rxV)
|
|
rx[int(envId)] = rxData
|
|
|
|
case "tx":
|
|
txV := int64(0)
|
|
if v, ok := result.Record().Value().(int64); ok {
|
|
txV = v
|
|
}
|
|
txData := append(tx[int(envId)], txV)
|
|
tx[int(envId)] = txData
|
|
}
|
|
}
|
|
}
|
|
return rx, tx, nil
|
|
}
|
|
|
|
func sparkDataForShares(shrs []*store.Share) (rx, tx map[string][]int64, err error) {
|
|
rx = make(map[string][]int64)
|
|
tx = make(map[string][]int64)
|
|
if len(shrs) > 0 {
|
|
qapi := idb.QueryAPI(cfg.Metrics.Influx.Org)
|
|
|
|
shrFilter := "|> filter(fn: (r) =>"
|
|
for i, shr := range shrs {
|
|
if i > 0 {
|
|
shrFilter += " or"
|
|
}
|
|
shrFilter += fmt.Sprintf(" r[\"share\"] == \"%v\"", shr.Token)
|
|
}
|
|
shrFilter += ")"
|
|
query := fmt.Sprintf("from(bucket: \"%v\")\n", cfg.Metrics.Influx.Bucket) +
|
|
"|> range(start: -5m)\n" +
|
|
"|> filter(fn: (r) => r[\"_measurement\"] == \"xfer\")\n" +
|
|
"|> filter(fn: (r) => r[\"_field\"] == \"rx\" or r[\"_field\"] == \"tx\")\n" +
|
|
"|> filter(fn: (r) => r[\"namespace\"] == \"backend\")\n" +
|
|
shrFilter +
|
|
"|> aggregateWindow(every: 10s, fn: sum, createEmpty: true)\n"
|
|
|
|
result, err := qapi.Query(context.Background(), query)
|
|
if err != nil {
|
|
return nil, nil, err
|
|
}
|
|
|
|
for result.Next() {
|
|
shrToken := result.Record().ValueByKey("share").(string)
|
|
switch result.Record().Field() {
|
|
case "rx":
|
|
rxV := int64(0)
|
|
if v, ok := result.Record().Value().(int64); ok {
|
|
rxV = v
|
|
}
|
|
rxData := append(rx[shrToken], rxV)
|
|
rx[shrToken] = rxData
|
|
|
|
case "tx":
|
|
txV := int64(0)
|
|
if v, ok := result.Record().Value().(int64); ok {
|
|
txV = v
|
|
}
|
|
txData := append(tx[shrToken], txV)
|
|
tx[shrToken] = txData
|
|
}
|
|
}
|
|
}
|
|
return rx, tx, nil
|
|
}
|