Skip to content

[scheduler] Configurable outdated thresholds, fix early resource scrubbing #154

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
cgalibern merged 2 commits into from
Sep 25, 2026
Merged

[scheduler] Configurable outdated thresholds, fix early resource scrubbing #154

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
221f09c
[scheduler] Make task outdated thresholds configurable
cgalibern Sep 25, 2026
7754f6f
[worker] Fix resmon not refreshed on daemon ping
cgalibern Sep 25, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 6 additions & 6 deletions cdb/db_actions.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -221,33 +221,33 @@ func (oDb *DB) GetBActionErrors(ctx context.Context) (lines []BActionErrorCount,
return
}

func (oDb *DB) UpdateUnfinishedActions(ctx context.Context) error {
func (oDb *DB) UpdateUnfinishedActions(ctx context.Context, maxAge time.Duration) error {
request := `UPDATE svcactions
SET
status = "err",
end = "1000-01-01 00:00:00"
WHERE
begin < DATE_SUB(NOW(), INTERVAL 120 MINUTE)
begin < DATE_SUB(NOW(), INTERVAL ? SECOND)
AND end IS NULL
AND status IS NULL
AND action NOT LIKE "%#%"`
if count, err := oDb.execCountContext(ctx, request); err != nil {
if count, err := oDb.execCountContext(ctx, request, maxAgeSeconds(maxAge)); err != nil {
return err
} else if count > 0 {
oDb.SetChange("svcactions")
}
return nil
}

func (oDb *DB) GetUnfinishedActions(ctx context.Context) (lines []SvcAction, err error) {
func (oDb *DB) GetUnfinishedActions(ctx context.Context, maxAge time.Duration) (lines []SvcAction, err error) {
query := `SELECT id, node_id, svc_id FROM svcactions
WHERE
begin < DATE_SUB(NOW(), INTERVAL 120 MINUTE)
begin < DATE_SUB(NOW(), INTERVAL ? SECOND)
AND end IS NULL
AND status IS NULL
AND action NOT LIKE "%#%"`
var rows *sql.Rows
rows, err = oDb.DB.QueryContext(ctx, query)
rows, err = oDb.DB.QueryContext(ctx, query, maxAgeSeconds(maxAge))
if err != nil {
return
}
Expand Down
7 changes: 4 additions & 3 deletions cdb/db_checks.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,11 +4,12 @@ import (
"context"
"fmt"
"log/slog"
"time"
)

func (oDb *DB) PurgeChecksOutdated(ctx context.Context) error {
request := fmt.Sprintf("DELETE FROM `checks_live` WHERE `chk_updated` < DATE_SUB(NOW(), INTERVAL 2 DAY)")
if count, err := oDb.execCountContext(ctx, request); err != nil {
func (oDb *DB) PurgeChecksOutdated(ctx context.Context, maxAge time.Duration) error {
request := fmt.Sprintf("DELETE FROM `checks_live` WHERE `chk_updated` < DATE_SUB(NOW(), INTERVAL ? SECOND)")
if count, err := oDb.execCountContext(ctx, request, maxAgeSeconds(maxAge)); err != nil {
return fmt.Errorf("delete from checks_live: %w", err)
} else if count > 0 {
// TODO: add metrics about purged count
Expand Down
31 changes: 16 additions & 15 deletions cdb/db_compliance.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@ import (
"database/sql"
"fmt"
"strings"
"time"
)

type Moduleset struct {
Expand Down Expand Up @@ -82,12 +83,12 @@ func (oDb *DB) PurgeCompModulesetsServices(ctx context.Context) error {
}

// purge entries older than 30 days
func (oDb *DB) PurgeCompStatusOutdated(ctx context.Context) error {
func (oDb *DB) PurgeCompStatusOutdated(ctx context.Context, maxAge time.Duration) error {
var query = `DELETE
FROM comp_status
WHERE
run_date < DATE_SUB(NOW(), INTERVAL 31 DAY)`
if count, err := oDb.execCountContext(ctx, query); err != nil {
run_date < DATE_SUB(NOW(), INTERVAL ? SECOND)`
if count, err := oDb.execCountContext(ctx, query, maxAgeSeconds(maxAge)); err != nil {
return err
} else if count > 0 {
oDb.SetChange("comp_status")
Expand Down Expand Up @@ -127,27 +128,27 @@ func (oDb *DB) PurgeCompStatusNodeOrphans(ctx context.Context) error {
return nil
}

// purge compliance status older than 7 days for modules in no moduleset, ie not schedulable
func (oDb *DB) PurgeCompStatusModulesetOrphans(ctx context.Context) error {
// purge compliance status older than maxAge for modules in no moduleset, ie not schedulable
func (oDb *DB) PurgeCompStatusModulesetOrphans(ctx context.Context, maxAge time.Duration) error {
var query = `DELETE FROM comp_status
WHERE
run_date < DATE_SUB(NOW(), INTERVAL 7 DAY) AND
run_date < DATE_SUB(NOW(), INTERVAL ? SECOND) AND
run_module NOT IN (
SELECT modset_mod_name FROM comp_moduleset_modules
)`
if count, err := oDb.execCountContext(ctx, query); err != nil {
if count, err := oDb.execCountContext(ctx, query, maxAgeSeconds(maxAge)); err != nil {
return err
} else if count > 0 {
oDb.SetChange("comp_status")
}
return nil
}

// purge node compliance status older than 7 days for unattached modules
func (oDb *DB) PurgeCompStatusNodeUnattached(ctx context.Context) error {
// purge node compliance status older than maxAge for unattached modules
func (oDb *DB) PurgeCompStatusNodeUnattached(ctx context.Context, maxAge time.Duration) error {
var query = `DELETE FROM comp_status
WHERE
run_date < DATE_SUB(NOW(), INTERVAL 7 DAY) AND
run_date < DATE_SUB(NOW(), INTERVAL ? SECOND) AND
svc_id = "" AND
run_module NOT IN (
SELECT modset_mod_name
Expand All @@ -157,19 +158,19 @@ func (oDb *DB) PurgeCompStatusNodeUnattached(ctx context.Context) error {
FROM comp_node_moduleset
)
)`
if count, err := oDb.execCountContext(ctx, query); err != nil {
if count, err := oDb.execCountContext(ctx, query, maxAgeSeconds(maxAge)); err != nil {
return err
} else if count > 0 {
oDb.SetChange("comp_status")
}
return nil
}

// purge svc compliance status older than 7 days for unattached modules
func (oDb *DB) PurgeCompStatusSvcUnattached(ctx context.Context) error {
// purge svc compliance status older than maxAge for unattached modules
func (oDb *DB) PurgeCompStatusSvcUnattached(ctx context.Context, maxAge time.Duration) error {
var query = `DELETE FROM comp_status
WHERE
run_date < DATE_SUB(NOW(), INTERVAL 7 DAY) AND
run_date < DATE_SUB(NOW(), INTERVAL ? SECOND) AND
svc_id = "" AND
run_module NOT IN (
SELECT modset_mod_name
Expand All @@ -179,7 +180,7 @@ func (oDb *DB) PurgeCompStatusSvcUnattached(ctx context.Context) error {
FROM comp_modulesets_services
)
)`
if count, err := oDb.execCountContext(ctx, query); err != nil {
if count, err := oDb.execCountContext(ctx, query, maxAgeSeconds(maxAge)); err != nil {
return err
} else if count > 0 {
oDb.SetChange("comp_status")
Expand Down
28 changes: 14 additions & 14 deletions cdb/db_dashboard.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -636,7 +636,7 @@ func (oDb *DB) PurgeAlertsOnDeletedServices(ctx context.Context) error {
return nil
}

func (oDb *DB) DashboardUpdateNodesNotUpdated(ctx context.Context) error {
func (oDb *DB) DashboardUpdateNodesNotUpdated(ctx context.Context, maxAge time.Duration) error {
request := `INSERT INTO dashboard
SELECT
NULL,
Expand All @@ -653,18 +653,18 @@ func (oDb *DB) DashboardUpdateNodesNotUpdated(ctx context.Context) error {
NULL,
NULL
FROM nodes
WHERE updated < date_sub(NOW(), interval 25 hour)
WHERE updated < date_sub(NOW(), INTERVAL ? SECOND)
ON DUPLICATE KEY UPDATE
dash_updated=NOW()`
if count, err := oDb.execCountContext(ctx, request); err != nil {
if count, err := oDb.execCountContext(ctx, request, maxAgeSeconds(maxAge)); err != nil {
return err
} else if count > 0 {
oDb.SetChange("dashboard")
}
return nil
}

func (oDb *DB) DashboardUpdateChecksNotUpdated(ctx context.Context) error {
func (oDb *DB) DashboardUpdateChecksNotUpdated(ctx context.Context, maxAge time.Duration) error {
request := `
DELETE FROM dashboard
WHERE
Expand Down Expand Up @@ -730,10 +730,10 @@ func (oDb *DB) DashboardUpdateChecksNotUpdated(ctx context.Context) error {
CONCAT(chk_type, ":", chk_instance)
FROM checks_live c
JOIN nodes n ON c.node_id = n.node_id
WHERE chk_updated < DATE_SUB(NOW(), INTERVAL 1 DAY)
WHERE chk_updated < DATE_SUB(NOW(), INTERVAL ? SECOND)
ON DUPLICATE KEY UPDATE dash_updated = NOW();
`
if count, err := oDb.execCountContext(ctx, request); err != nil {
if count, err := oDb.execCountContext(ctx, request, maxAgeSeconds(maxAge)); err != nil {
return err
} else if count > 0 {
oDb.SetChange("dashboard")
Expand Down Expand Up @@ -793,7 +793,7 @@ func (oDb *DB) DashboardDeleteActionErrorsWithNoError(ctx context.Context) error
return nil
}

func (oDb *DB) DashboardUpdateServiceConfigNotUpdated(ctx context.Context) error {
func (oDb *DB) DashboardUpdateServiceConfigNotUpdated(ctx context.Context, maxAge time.Duration) error {
request := `
INSERT INTO dashboard
SELECT
Expand All @@ -811,19 +811,19 @@ func (oDb *DB) DashboardUpdateServiceConfigNotUpdated(ctx context.Context) error
NULL,
NULL
FROM services
WHERE updated < DATE_SUB(NOW(), INTERVAL 25 HOUR)
WHERE updated < DATE_SUB(NOW(), INTERVAL ? SECOND)
ON DUPLICATE KEY UPDATE
dash_updated=NOW()
`
if count, err := oDb.execCountContext(ctx, request); err != nil {
if count, err := oDb.execCountContext(ctx, request, maxAgeSeconds(maxAge)); err != nil {
return err
} else if count > 0 {
oDb.SetChange("dashboard")
}
return nil
}

func (oDb *DB) DashboardUpdateInstancesNotUpdated(ctx context.Context) error {
func (oDb *DB) DashboardUpdateInstancesNotUpdated(ctx context.Context, maxAge time.Duration) error {
request := `
INSERT INTO dashboard
SELECT
Expand All @@ -841,11 +841,11 @@ func (oDb *DB) DashboardUpdateInstancesNotUpdated(ctx context.Context) error {
NULL,
NULL
FROM svcmon
WHERE mon_updated < DATE_SUB(NOW(), INTERVAL 16 MINUTE)
WHERE mon_updated < DATE_SUB(NOW(), INTERVAL ? SECOND)
ON DUPLICATE KEY UPDATE
dash_updated=NOW()
`
if count, err := oDb.execCountContext(ctx, request); err != nil {
if count, err := oDb.execCountContext(ctx, request, maxAgeSeconds(maxAge)); err != nil {
return err
} else if count > 0 {
oDb.SetChange("dashboard")
Expand All @@ -863,10 +863,10 @@ func (oDb *DB) DashboardUpdateInstancesNotUpdated(ctx context.Context) error {
dashboard.dash_type = "service status not updated" AND
dashboard.svc_id != "" AND
dashboard.node_id != "" AND
(svcmon.id IS NULL OR svcmon.mon_updated >= DATE_SUB(NOW(), INTERVAL 16 MINUTE))
(svcmon.id IS NULL OR svcmon.mon_updated >= DATE_SUB(NOW(), INTERVAL ? SECOND))
)
`
if count, err := oDb.execCountContext(ctx, request); err != nil {
if count, err := oDb.execCountContext(ctx, request, maxAgeSeconds(maxAge)); err != nil {
return err
} else if count > 0 {
oDb.SetChange("dashboard")
Expand Down
7 changes: 4 additions & 3 deletions cdb/db_diskinfo.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@ import (
"errors"
"fmt"
"strings"
"time"
)

type (
Expand Down Expand Up @@ -223,12 +224,12 @@ func (oDb *DB) DeleteDiskinfoByDiskID(ctx context.Context, diskID string) (int64
return count, nil
}

func (oDb *DB) PurgeDiskinfoOutdated(ctx context.Context) error {
func (oDb *DB) PurgeDiskinfoOutdated(ctx context.Context, maxAge time.Duration) error {
var query = `DELETE
FROM diskinfo
WHERE
disk_updated < DATE_SUB(NOW(), INTERVAL 2 DAY)`
if count, err := oDb.execCountContext(ctx, query); err != nil {
disk_updated < DATE_SUB(NOW(), INTERVAL ? SECOND)`
if count, err := oDb.execCountContext(ctx, query, maxAgeSeconds(maxAge)); err != nil {
return fmt.Errorf("purge diskinfo: %w", err)
} else if count > 0 {
oDb.SetChange("diskinfo")
Expand Down
39 changes: 21 additions & 18 deletions cdb/db_instances.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -232,9 +232,9 @@ func (oDb *DB) SvcmonRefreshTimestamp(ctx context.Context, nodeID string, object
return
}

// InstancePingFromNodeID updates match svcmon.mon_updated, svcmon_log_last.mon_end,
// resmon.updated and resmon_log_last.res_end when svcmon.mon_updated timestamp
// for node_id id older than 30s.
// InstancePingFromNodeID refreshes svcmon.mon_updated, svcmon_log_last.mon_end,
// resmon.updated and resmon_log_last.res_end of the node_id rows, when older
// than 30s.
func (oDb *DB) InstancePingFromNodeID(ctx context.Context, nodeID string) (updates bool, err error) {
defer logDuration("instancePing "+nodeID, time.Now())
const (
Expand All @@ -254,24 +254,27 @@ func (oDb *DB) InstancePingFromNodeID(ctx context.Context, nodeID string) (updat
count int64
)

// svcmon and resmon are refreshed independently: svcmon.mon_updated may
// have been refreshed by another path (daemon status) without refreshing
// all resmon rows, so skipping resmon when svcmon has no update would let
// resmon.updated age until the resources are scrubbed.
if count, err = oDb.execCountContext(ctx, qUpdateSvcmon, nodeID); err != nil {
return
} else if count == 0 {
return
} else if count > 0 {
updates = true
oDb.SetChange("svcmon")
}
updates = true
oDb.SetChange("svcmon")

if _, err = oDb.ExecContext(ctx, qUpdateSvcmonLogLast, nodeID); err != nil {
return
}

if count, err = oDb.execCountContext(ctx, qUpdateResmon, nodeID); err != nil {
return
} else if count == 0 {
return
} else if count > 0 {
updates = true
oDb.SetChange("resmon")
}
oDb.SetChange("resmon")

_, err = oDb.ExecContext(ctx, qUpdateResmonLogLast, nodeID)
return
Expand Down Expand Up @@ -750,12 +753,13 @@ func (oDb *DB) PurgeInstance(ctx context.Context, id InstanceID) error {
return err
}

func (oDb *DB) InstancesOutdated(ctx context.Context) (instanceIDs []InstanceID, err error) {
// InstancesOutdated returns the svcmon instance ids not updated since maxAge.
func (oDb *DB) InstancesOutdated(ctx context.Context, maxAge time.Duration) (instanceIDs []InstanceID, err error) {
var rows *sql.Rows
query := "SELECT `svc_id`, `node_id` " +
"FROM `svcmon` " +
"WHERE `mon_updated` < DATE_SUB(NOW(), INTERVAL 21 MINUTE)"
rows, err = oDb.DB.QueryContext(ctx, query)
"WHERE `mon_updated` < DATE_SUB(NOW(), INTERVAL ? SECOND)"
rows, err = oDb.DB.QueryContext(ctx, query, maxAgeSeconds(maxAge))
if err != nil {
return
}
Expand All @@ -771,14 +775,13 @@ func (oDb *DB) InstancesOutdated(ctx context.Context) (instanceIDs []InstanceID,
return
}

func (oDb *DB) LogInstancesNotUpdated(ctx context.Context) error {
age := 2
func (oDb *DB) LogInstancesNotUpdated(ctx context.Context, maxAge time.Duration) error {
request := fmt.Sprintf(`INSERT IGNORE
INTO log
SELECT NULL,
"service.status",
"scheduler",
"instance status not updated for more than %dh (%%(date)s)",
"instance status not updated for more than %s (%%(date)s)",
CONCAT('{"date": "', mon_updated, '"}'),
NOW(),
svc_id,
Expand All @@ -788,8 +791,8 @@ func (oDb *DB) LogInstancesNotUpdated(ctx context.Context) error {
"warning",
node_id
from svcmon
where mon_updated<DATE_SUB(NOW(), INTERVAL %d HOUR)`, age, age)
if count, err := oDb.execCountContext(ctx, request); err != nil {
where mon_updated<DATE_SUB(NOW(), INTERVAL ? SECOND)`, FormatMaxAge(maxAge))
if count, err := oDb.execCountContext(ctx, request, maxAgeSeconds(maxAge)); err != nil {
return err
} else if count > 0 {
slog.Debug(fmt.Sprintf("alert: instance outdated: %d", count))
Expand Down
6 changes: 3 additions & 3 deletions cdb/db_nodes.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -433,9 +433,9 @@ func (oDb *DB) NodeUpdateClusterIDForNodeID(ctx context.Context, nodeID, cluster
}
}

func (oDb *DB) PurgeNodeHBAsOutdated(ctx context.Context) error {
request := fmt.Sprintf("DELETE FROM `node_hba` WHERE `updated` < DATE_SUB(NOW(), INTERVAL 7 DAY)")
if count, err := oDb.execCountContext(ctx, request); err != nil {
func (oDb *DB) PurgeNodeHBAsOutdated(ctx context.Context, maxAge time.Duration) error {
request := fmt.Sprintf("DELETE FROM `node_hba` WHERE `updated` < DATE_SUB(NOW(), INTERVAL ? SECOND)")
if count, err := oDb.execCountContext(ctx, request, maxAgeSeconds(maxAge)); err != nil {
return err
} else if count > 0 {
slog.Debug(fmt.Sprintf("purged %d entries from table node_hba", count))
Expand Down
Loading
Loading