You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
123 lines
4.2 KiB
123 lines
4.2 KiB
package database
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"github.com/influxdata/influxdb-client-go/v2/api"
|
|
)
|
|
|
|
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
|
|
// Tasks
|
|
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
|
|
func (sr *scrutinyRepository) EnsureTasks(ctx context.Context, orgID string) error {
|
|
weeklyTaskName := "tsk-weekly-aggr"
|
|
if found, findErr := sr.influxTaskApi.FindTasks(ctx, &api.TaskFilter{Name: weeklyTaskName}); findErr == nil && len(found) == 0 {
|
|
//weekly on Sunday at 1:00am
|
|
_, err := sr.influxTaskApi.CreateTaskWithCron(ctx, weeklyTaskName, sr.DownsampleScript("weekly"), "0 1 * * 0", orgID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
monthlyTaskName := "tsk-monthly-aggr"
|
|
if found, findErr := sr.influxTaskApi.FindTasks(ctx, &api.TaskFilter{Name: monthlyTaskName}); findErr == nil && len(found) == 0 {
|
|
//monthly on first day of the month at 1:30am
|
|
_, err := sr.influxTaskApi.CreateTaskWithCron(ctx, monthlyTaskName, sr.DownsampleScript("monthly"), "30 1 1 * *", orgID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
yearlyTaskName := "tsk-yearly-aggr"
|
|
if found, findErr := sr.influxTaskApi.FindTasks(ctx, &api.TaskFilter{Name: yearlyTaskName}); findErr == nil && len(found) == 0 {
|
|
//yearly on the first day of the year at 2:00am
|
|
_, err := sr.influxTaskApi.CreateTaskWithCron(ctx, yearlyTaskName, sr.DownsampleScript("yearly"), "0 2 1 1 *", orgID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (sr *scrutinyRepository) DownsampleScript(aggregationType string) string {
|
|
var sourceBucket string // the source of the data
|
|
var destBucket string // the destination for the aggregated data
|
|
var rangeStart string
|
|
var rangeEnd string
|
|
var aggWindow string
|
|
switch aggregationType {
|
|
case "weekly":
|
|
sourceBucket = sr.appConfig.GetString("web.influxdb.bucket")
|
|
destBucket = fmt.Sprintf("%s_weekly", sr.appConfig.GetString("web.influxdb.bucket"))
|
|
rangeStart = "-2w"
|
|
rangeEnd = "-1w"
|
|
aggWindow = "1w"
|
|
case "monthly":
|
|
sourceBucket = fmt.Sprintf("%s_weekly", sr.appConfig.GetString("web.influxdb.bucket"))
|
|
destBucket = fmt.Sprintf("%s_monthly", sr.appConfig.GetString("web.influxdb.bucket"))
|
|
rangeStart = "-2mo"
|
|
rangeEnd = "-1mo"
|
|
aggWindow = "1mo"
|
|
case "yearly":
|
|
sourceBucket = fmt.Sprintf("%s_monthly", sr.appConfig.GetString("web.influxdb.bucket"))
|
|
destBucket = fmt.Sprintf("%s_yearly", sr.appConfig.GetString("web.influxdb.bucket"))
|
|
rangeStart = "-2y"
|
|
rangeEnd = "-1y"
|
|
aggWindow = "1y"
|
|
}
|
|
|
|
// TODO: using "last" function for aggregation. This should eventually be replaced with a more accurate represenation
|
|
/*
|
|
import "types"
|
|
smart_data = from(bucket: sourceBucket)
|
|
|> range(start: rangeStart, stop: rangeEnd)
|
|
|> filter(fn: (r) => r["_measurement"] == "smart" )
|
|
|> group(columns: ["device_wwn", "_field"])
|
|
|
|
non_numeric_smart_data = smart_data
|
|
|> filter(fn: (r) => types.isType(v: r._value, type: "string") or types.isType(v: r._value, type: "bool"))
|
|
|> aggregateWindow(every: aggWindow, fn: last, createEmpty: false)
|
|
|
|
numeric_smart_data = smart_data
|
|
|> filter(fn: (r) => types.isType(v: r._value, type: "int") or types.isType(v: r._value, type: "float"))
|
|
|> aggregateWindow(every: aggWindow, fn: mean, createEmpty: false)
|
|
|
|
union(tables: [non_numeric_smart_data, numeric_smart_data])
|
|
|> to(bucket: destBucket, org: destOrg)
|
|
|
|
*/
|
|
|
|
return fmt.Sprintf(`
|
|
sourceBucket = "%s"
|
|
rangeStart = %s
|
|
rangeEnd = %s
|
|
aggWindow = %s
|
|
destBucket = "%s"
|
|
destOrg = "%s"
|
|
|
|
from(bucket: sourceBucket)
|
|
|> range(start: rangeStart, stop: rangeEnd)
|
|
|> filter(fn: (r) => r["_measurement"] == "smart" )
|
|
|> group(columns: ["device_wwn", "_field"])
|
|
|> aggregateWindow(every: aggWindow, fn: last, createEmpty: false)
|
|
|> to(bucket: destBucket, org: destOrg)
|
|
|
|
temp_data = from(bucket: sourceBucket)
|
|
|> range(start: rangeStart, stop: rangeEnd)
|
|
|> filter(fn: (r) => r["_measurement"] == "temp")
|
|
|> group(columns: ["device_wwn"])
|
|
|> toInt()
|
|
|
|
temp_data
|
|
|> aggregateWindow(fn: mean, every: aggWindow)
|
|
|> to(bucket: destBucket, org: destOrg)
|
|
`,
|
|
sourceBucket,
|
|
rangeStart,
|
|
rangeEnd,
|
|
aggWindow,
|
|
destBucket,
|
|
sr.appConfig.GetString("web.influxdb.org"),
|
|
)
|
|
}
|