open-source-mirrors
/
scrutiny
mirror of https://github.com/AnalogJ/scrutiny
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

fix update, using raw flux script.

Browse Source
pull/328/head
Jason Kulatunga 3 years ago
parent 04563c0d0d
commit 0f0efac866
2 changed files with 122 additions and 100 deletions
Show Stats Download Patch File Download Diff File

63
webapp/backend/pkg/database/scrutiny_repository_tasks.go
Unescape View File

@ -11,10 +11,10 @@ import (
//////////////////////////////////////////////////////////////////////////////////////////////////////////////////// ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
func (sr *scrutinyRepository) EnsureTasks(ctx context.Context, orgID string) error { func (sr *scrutinyRepository) EnsureTasks(ctx context.Context, orgID string) error {
weeklyTaskName := "tsk-weekly-aggr" weeklyTaskName := "tsk-weekly-aggr"
weeklyTaskScript := sr.DownsampleScript("weekly") weeklyTaskScript := sr.DownsampleScript("weekly", weeklyTaskName, "0 1 * * 0")
if found, findErr := sr.influxTaskApi.FindTasks(ctx, &api.TaskFilter{Name: weeklyTaskName}); findErr == nil && len(found) == 0 { if found, findErr := sr.influxTaskApi.FindTasks(ctx, &api.TaskFilter{Name: weeklyTaskName}); findErr == nil && len(found) == 0 {
//weekly on Sunday at 1:00am //weekly on Sunday at 1:00am
_, err := sr.influxTaskApi.CreateTaskWithCron(ctx, weeklyTaskName, weeklyTaskScript, "0 1 * * 0", orgID) _, err := sr.influxTaskApi.CreateTaskByFlux(ctx, weeklyTaskScript, orgID)
if err != nil { if err != nil {
return err return err
} }
@ -32,10 +32,10 @@ func (sr *scrutinyRepository) EnsureTasks(ctx context.Context, orgID string) err
} }
monthlyTaskName := "tsk-monthly-aggr" monthlyTaskName := "tsk-monthly-aggr"
monthlyTaskScript := sr.DownsampleScript("monthly") monthlyTaskScript := sr.DownsampleScript("monthly", monthlyTaskName, "30 1 1 * *")
if found, findErr := sr.influxTaskApi.FindTasks(ctx, &api.TaskFilter{Name: monthlyTaskName}); findErr == nil && len(found) == 0 { 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 //monthly on first day of the month at 1:30am
_, err := sr.influxTaskApi.CreateTaskWithCron(ctx, monthlyTaskName, monthlyTaskScript, "30 1 1 * *", orgID) _, err := sr.influxTaskApi.CreateTaskByFlux(ctx, monthlyTaskScript, orgID)
if err != nil { if err != nil {
return err return err
} }
@ -53,10 +53,10 @@ func (sr *scrutinyRepository) EnsureTasks(ctx context.Context, orgID string) err
} }
yearlyTaskName := "tsk-yearly-aggr" yearlyTaskName := "tsk-yearly-aggr"
yearlyTaskScript := sr.DownsampleScript("yearly") yearlyTaskScript := sr.DownsampleScript("yearly", yearlyTaskName, "0 2 1 1 *")
if found, findErr := sr.influxTaskApi.FindTasks(ctx, &api.TaskFilter{Name: yearlyTaskName}); findErr == nil && len(found) == 0 { 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 //yearly on the first day of the year at 2:00am
_, err := sr.influxTaskApi.CreateTaskWithCron(ctx, yearlyTaskName, yearlyTaskScript, "0 2 1 1 *", orgID) _, err := sr.influxTaskApi.CreateTaskByFlux(ctx, yearlyTaskScript, orgID)
if err != nil { if err != nil {
return err return err
} }
@ -75,7 +75,7 @@ func (sr *scrutinyRepository) EnsureTasks(ctx context.Context, orgID string) err
return nil return nil
} }
func (sr *scrutinyRepository) DownsampleScript(aggregationType string) string { func (sr *scrutinyRepository) DownsampleScript(aggregationType string, name string, cron string) string {
var sourceBucket string // the source of the data var sourceBucket string // the source of the data
var destBucket string // the destination for the aggregated data var destBucket string // the destination for the aggregated data
var rangeStart string var rangeStart string
@ -124,30 +124,37 @@ func (sr *scrutinyRepository) DownsampleScript(aggregationType string) string {
*/ */
return fmt.Sprintf(` return fmt.Sprintf(`
sourceBucket = "%s" option task = {
rangeStart = %s name: "%s",
rangeEnd = %s cron: "%s",
aggWindow = %s }
destBucket = "%s"
destOrg = "%s" sourceBucket = "%s"
rangeStart = %s
rangeEnd = %s
aggWindow = %s
destBucket = "%s"
destOrg = "%s"
from(bucket: sourceBucket) from(bucket: sourceBucket)
|> range(start: rangeStart, stop: rangeEnd) |> range(start: rangeStart, stop: rangeEnd)
|> filter(fn: (r) => r["_measurement"] == "smart" ) |> filter(fn: (r) => r["_measurement"] == "smart" )
|> group(columns: ["device_wwn", "_field"]) |> group(columns: ["device_wwn", "_field"])
|> aggregateWindow(every: aggWindow, fn: last, createEmpty: false) |> aggregateWindow(every: aggWindow, fn: last, createEmpty: false)
|> to(bucket: destBucket, org: destOrg) |> to(bucket: destBucket, org: destOrg)
from(bucket: sourceBucket) from(bucket: sourceBucket)
|> range(start: rangeStart, stop: rangeEnd) |> range(start: rangeStart, stop: rangeEnd)
|> filter(fn: (r) => r["_measurement"] == "temp") |> filter(fn: (r) => r["_measurement"] == "temp")
|> group(columns: ["device_wwn"]) |> group(columns: ["device_wwn"])
|> toInt() |> toInt()
|> aggregateWindow(fn: mean, every: aggWindow, createEmpty: false) |> aggregateWindow(fn: mean, every: aggWindow, createEmpty: false)
|> set(key: "_measurement", value: "temp") |> set(key: "_measurement", value: "temp")
|> set(key: "_field", value: "temp") |> set(key: "_field", value: "temp")
|> to(bucket: destBucket, org: destOrg) |> to(bucket: destBucket, org: destOrg)
`, `,
name,
cron,
sourceBucket, sourceBucket,
rangeStart, rangeStart,
rangeEnd, rangeEnd,

159
webapp/backend/pkg/database/scrutiny_repository_tasks_test.go
Unescape View File

@ -24,33 +24,38 @@ func Test_DownsampleScript_Weekly(t *testing.T) {
aggregationType := "weekly" aggregationType := "weekly"
//test //test
influxDbScript := deviceRepo.DownsampleScript(aggregationType) influxDbScript := deviceRepo.DownsampleScript(aggregationType, "tsk-weekly-aggr", "0 1 * * 0")
//assert //assert
require.Equal(t, ` require.Equal(t, `
sourceBucket = "metrics" option task = {
rangeStart = -2w name: "tsk-weekly-aggr",
rangeEnd = -1w cron: "0 1 * * 0",
aggWindow = 1w }
destBucket = "metrics_weekly"
destOrg = "scrutiny" sourceBucket = "metrics"
rangeStart = -2w
from(bucket: sourceBucket) rangeEnd = -1w
|> range(start: rangeStart, stop: rangeEnd) aggWindow = 1w
|> filter(fn: (r) => r["_measurement"] == "smart" ) destBucket = "metrics_weekly"
|> group(columns: ["device_wwn", "_field"]) destOrg = "scrutiny"
|> aggregateWindow(every: aggWindow, fn: last, createEmpty: false)
|> to(bucket: destBucket, org: destOrg) from(bucket: sourceBucket)
|> range(start: rangeStart, stop: rangeEnd)
from(bucket: sourceBucket) |> filter(fn: (r) => r["_measurement"] == "smart" )
|> range(start: rangeStart, stop: rangeEnd) |> group(columns: ["device_wwn", "_field"])
|> filter(fn: (r) => r["_measurement"] == "temp") |> aggregateWindow(every: aggWindow, fn: last, createEmpty: false)
|> group(columns: ["device_wwn"]) |> to(bucket: destBucket, org: destOrg)
|> toInt()
|> aggregateWindow(fn: mean, every: aggWindow, createEmpty: false) from(bucket: sourceBucket)
|> set(key: "_measurement", value: "temp") |> range(start: rangeStart, stop: rangeEnd)
|> set(key: "_field", value: "temp") |> filter(fn: (r) => r["_measurement"] == "temp")
|> to(bucket: destBucket, org: destOrg) |> group(columns: ["device_wwn"])
|> toInt()
|> aggregateWindow(fn: mean, every: aggWindow, createEmpty: false)
|> set(key: "_measurement", value: "temp")
|> set(key: "_field", value: "temp")
|> to(bucket: destBucket, org: destOrg)
`, influxDbScript) `, influxDbScript)
} }
@ -71,33 +76,38 @@ func Test_DownsampleScript_Monthly(t *testing.T) {
aggregationType := "monthly" aggregationType := "monthly"
//test //test
influxDbScript := deviceRepo.DownsampleScript(aggregationType) influxDbScript := deviceRepo.DownsampleScript(aggregationType, "tsk-monthly-aggr", "30 1 1 * *")
//assert //assert
require.Equal(t, ` require.Equal(t, `
sourceBucket = "metrics_weekly" option task = {
rangeStart = -2mo name: "tsk-monthly-aggr",
rangeEnd = -1mo cron: "30 1 1 * *",
aggWindow = 1mo }
destBucket = "metrics_monthly"
destOrg = "scrutiny" sourceBucket = "metrics_weekly"
rangeStart = -2mo
from(bucket: sourceBucket) rangeEnd = -1mo
|> range(start: rangeStart, stop: rangeEnd) aggWindow = 1mo
|> filter(fn: (r) => r["_measurement"] == "smart" ) destBucket = "metrics_monthly"
|> group(columns: ["device_wwn", "_field"]) destOrg = "scrutiny"
|> aggregateWindow(every: aggWindow, fn: last, createEmpty: false)
|> to(bucket: destBucket, org: destOrg) from(bucket: sourceBucket)
|> range(start: rangeStart, stop: rangeEnd)
from(bucket: sourceBucket) |> filter(fn: (r) => r["_measurement"] == "smart" )
|> range(start: rangeStart, stop: rangeEnd) |> group(columns: ["device_wwn", "_field"])
|> filter(fn: (r) => r["_measurement"] == "temp") |> aggregateWindow(every: aggWindow, fn: last, createEmpty: false)
|> group(columns: ["device_wwn"]) |> to(bucket: destBucket, org: destOrg)
|> toInt()
|> aggregateWindow(fn: mean, every: aggWindow, createEmpty: false) from(bucket: sourceBucket)
|> set(key: "_measurement", value: "temp") |> range(start: rangeStart, stop: rangeEnd)
|> set(key: "_field", value: "temp") |> filter(fn: (r) => r["_measurement"] == "temp")
|> to(bucket: destBucket, org: destOrg) |> group(columns: ["device_wwn"])
|> toInt()
|> aggregateWindow(fn: mean, every: aggWindow, createEmpty: false)
|> set(key: "_measurement", value: "temp")
|> set(key: "_field", value: "temp")
|> to(bucket: destBucket, org: destOrg)
`, influxDbScript) `, influxDbScript)
} }
@ -118,32 +128,37 @@ func Test_DownsampleScript_Yearly(t *testing.T) {
aggregationType := "yearly" aggregationType := "yearly"
//test //test
influxDbScript := deviceRepo.DownsampleScript(aggregationType) influxDbScript := deviceRepo.DownsampleScript(aggregationType, "tsk-yearly-aggr", "0 2 1 1 *")
//assert //assert
require.Equal(t, ` require.Equal(t, `
sourceBucket = "metrics_monthly" option task = {
rangeStart = -2y name: "tsk-yearly-aggr",
rangeEnd = -1y cron: "0 2 1 1 *",
aggWindow = 1y }
destBucket = "metrics_yearly"
destOrg = "scrutiny" sourceBucket = "metrics_monthly"
rangeStart = -2y
from(bucket: sourceBucket) rangeEnd = -1y
|> range(start: rangeStart, stop: rangeEnd) aggWindow = 1y
|> filter(fn: (r) => r["_measurement"] == "smart" ) destBucket = "metrics_yearly"
|> group(columns: ["device_wwn", "_field"]) destOrg = "scrutiny"
|> aggregateWindow(every: aggWindow, fn: last, createEmpty: false)
|> to(bucket: destBucket, org: destOrg) from(bucket: sourceBucket)
|> range(start: rangeStart, stop: rangeEnd)
from(bucket: sourceBucket) |> filter(fn: (r) => r["_measurement"] == "smart" )
|> range(start: rangeStart, stop: rangeEnd) |> group(columns: ["device_wwn", "_field"])
|> filter(fn: (r) => r["_measurement"] == "temp") |> aggregateWindow(every: aggWindow, fn: last, createEmpty: false)
|> group(columns: ["device_wwn"]) |> to(bucket: destBucket, org: destOrg)
|> toInt()
|> aggregateWindow(fn: mean, every: aggWindow, createEmpty: false) from(bucket: sourceBucket)
|> set(key: "_measurement", value: "temp") |> range(start: rangeStart, stop: rangeEnd)
|> set(key: "_field", value: "temp") |> filter(fn: (r) => r["_measurement"] == "temp")
|> to(bucket: destBucket, org: destOrg) |> group(columns: ["device_wwn"])
|> toInt()
|> aggregateWindow(fn: mean, every: aggWindow, createEmpty: false)
|> set(key: "_measurement", value: "temp")
|> set(key: "_field", value: "temp")
|> to(bucket: destBucket, org: destOrg)
`, influxDbScript) `, influxDbScript)
} }

Write Preview
Loading…
Cancel
Save