Skip to content

Enhance job feed handling and database mappings #99

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
sghf merged 5 commits into from
Jan 30, 2026
Merged

Enhance job feed handling and database mappings #99

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
3dadb2e
[worker] Add operation `system/pushFromTableChanges`
cgalibern Jan 29, 2026
3bcf578
[worker, cdb] Fix `begin` mapping in the `hbmon_log_last` upsert query.
cgalibern Jan 29, 2026
adb55f4
[worker] Enhance node and peer-node validation in `job_feed_daemon_st…
cgalibern Jan 29, 2026
bb79751
[worker, cdb] Fix possible panic during feed node system and disk
cgalibern Jan 30, 2026
3e029a3
Refactor: Replace repetitive row count handling with `execCountContex…
cgalibern Jan 30, 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
33 changes: 23 additions & 10 deletions cdb/db.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -132,29 +132,42 @@ func (oDb *DB) DeleteBatched(ctx context.Context, table, dateCol, orderbyCol str
ctx, cancel := context.WithTimeout(ctx, time.Minute)

// Execute the DELETE statement
result, err := oDb.DB.ExecContext(ctx, query)
count, err := oDb.execCountContext(ctx, query)
cancel()
if err != nil {
return totalDeleted, batchCount, fmt.Errorf("%s: error executing batch %d: %w", table, batchCount, err)
}

// Check the number of affected rows
rowsAffected, err := result.RowsAffected()
if err != nil {
return totalDeleted, batchCount, fmt.Errorf("%s: error checking affected rows for batch %d: %w", table, batchCount, err)
}

totalDeleted += rowsAffected
if rowsAffected > 0 {
slog.Debug(fmt.Sprintf("DeleteBatched: %s: batch %d: deleted %d rows. total deleted: %d", table, batchCount, rowsAffected, totalDeleted))
totalDeleted += count
if count > 0 {
slog.Debug(fmt.Sprintf("DeleteBatched: %s: batch %d: deleted %d rows. total deleted: %d", table, batchCount, count, totalDeleted))
}

// If less than the batch size was deleted, we've reached the end of the matching rows.
if rowsAffected < batchSize {
if count < batchSize {
return totalDeleted, batchCount, nil
}

// Add a short sleep to yield CPU time, preventing resource monopolization
time.Sleep(10 * time.Millisecond)
}
}

// ExecContextAndCountRowsAffected executes the oDb.DB.ExecContext query with the provided context, returning the number of rows affected and an error.
func (oDb *DB) ExecContextAndCountRowsAffected(ctx context.Context, query string, args ...any) (int64, error) {
return oDb.execCountContext(ctx, query, args...)
}

// execCountContext executes the oDb.DB.ExecContext query with the provided context and arguments, returning the number of affected rows and an error.
func (oDb *DB) execCountContext(ctx context.Context, query string, args ...any) (int64, error) {
result, err := oDb.DB.ExecContext(ctx, query, args...)
if err != nil {
return 0, err
}
if result == nil {
// len data may be 0, so no rows affected.
return 0, nil
}
return result.RowsAffected()
}
28 changes: 10 additions & 18 deletions cdb/db_actions.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -67,13 +67,9 @@ func (oDb *DB) AutoAckActionErrors(ctx context.Context, ids []int64) error {
WHERE
id IN (%s)`, Placeholders(len(ids)))
args := argsFromIDs(ids)
result, err := oDb.DB.ExecContext(ctx, request, args...)
if err != nil {
return err
}
if rowAffected, err := result.RowsAffected(); err != nil {
if count, err := oDb.execCountContext(ctx, request, args...); err != nil {
return err
} else if rowAffected > 0 {
} else if count > 0 {
oDb.SetChange("svcactions")
}
return nil
Expand All @@ -97,13 +93,9 @@ func (oDb *DB) LogActionErrorsNotAcked(ctx context.Context, ids []int64) error {
WHERE
id IN (%s)`, Placeholders(len(ids)))
args := argsFromIDs(ids)
result, err := oDb.DB.ExecContext(ctx, request, args...)
if err != nil {
if count, err := oDb.execCountContext(ctx, request, args...); err != nil {
return err
}
if rowAffected, err := result.RowsAffected(); err != nil {
return err
} else if rowAffected > 0 {
} else if count > 0 {
oDb.SetChange("log")
}
return nil
Expand Down Expand Up @@ -175,13 +167,9 @@ func (oDb *DB) UpdateUnfinishedActions(ctx context.Context) error {
AND end IS NULL
AND status IS NULL
AND action NOT LIKE "%#%"`
result, err := oDb.DB.ExecContext(ctx, request)
if err != nil {
return err
}
if rowAffected, err := result.RowsAffected(); err != nil {
if count, err := oDb.execCountContext(ctx, request); err != nil {
return err
} else if rowAffected > 0 {
} else if count > 0 {
oDb.SetChange("svcactions")
}
return nil
Expand Down Expand Up @@ -232,6 +220,8 @@ func (oDb *DB) InsertSvcAction(ctx context.Context, svcID, nodeID uuid.UUID, act
result, err := oDb.DB.ExecContext(ctx, query, args...)
if err != nil {
return 0, err
} else if result == nil {
return 0, errors.New("insert svcactions returns unexpected nil result")
}

id, err := result.LastInsertId()
Expand All @@ -253,6 +243,8 @@ func (oDb *DB) UpdateSvcAction(ctx context.Context, svcActionID int64, end time.
result, err := oDb.DB.ExecContext(ctx, query, end, status, end, statusLog, svcActionID)
if err != nil {
return err
} else if result == nil {
return errors.New("update svcactions returns unexpected nil result")
}
if rowsAffected, err := result.RowsAffected(); err != nil {
return err
Expand Down
10 changes: 3 additions & 7 deletions cdb/db_checks.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,14 +8,10 @@ import (

func (oDb *DB) PurgeChecksOutdated(ctx context.Context) error {
request := fmt.Sprintf("DELETE FROM `checks_live` WHERE `chk_updated` < DATE_SUB(NOW(), INTERVAL 2 DAY)")
result, err := oDb.DB.ExecContext(ctx, request)
if err != nil {
if count, err := oDb.execCountContext(ctx, request); err != nil {
return fmt.Errorf("delete from checks_live: %w", err)
}
if rowAffected, err := result.RowsAffected(); err != nil {
return fmt.Errorf("count delete from checks_live: %w", err)
} else if rowAffected > 0 {
slog.Info(fmt.Sprintf("purged %d entries from table checks_live", rowAffected))
} else if count > 0 {
slog.Info(fmt.Sprintf("purged %d entries from table checks_live", count))
oDb.SetChange("checks_live")
}
return nil
Expand Down
60 changes: 20 additions & 40 deletions cdb/db_compliance.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,11 +11,9 @@ func (oDb *DB) PurgeCompModulesetsNodes(ctx context.Context) error {
node_id NOT IN (
SELECT DISTINCT node_id FROM nodes
)`
if result, err := oDb.DB.ExecContext(ctx, query); err != nil {
if count, err := oDb.execCountContext(ctx, query); err != nil {
return err
} else if affected, err := result.RowsAffected(); err != nil {
return err
} else if affected > 0 {
} else if count > 0 {
oDb.SetChange("comp_node_moduleset")
}
return nil
Expand All @@ -28,11 +26,9 @@ func (oDb *DB) PurgeCompRulesetsNodes(ctx context.Context) error {
node_id NOT IN (
SELECT DISTINCT node_id FROM nodes
)`
if result, err := oDb.DB.ExecContext(ctx, query); err != nil {
return err
} else if affected, err := result.RowsAffected(); err != nil {
if count, err := oDb.execCountContext(ctx, query); err != nil {
return err
} else if affected > 0 {
} else if count > 0 {
oDb.SetChange("comp_rulesets_nodes")
}
return nil
Expand All @@ -45,11 +41,9 @@ func (oDb *DB) PurgeCompRulesetsServices(ctx context.Context) error {
svc_id NOT IN (
SELECT DISTINCT svc_id FROM svcmon
)`
if result, err := oDb.DB.ExecContext(ctx, query); err != nil {
return err
} else if affected, err := result.RowsAffected(); err != nil {
if count, err := oDb.execCountContext(ctx, query); err != nil {
return err
} else if affected > 0 {
} else if count > 0 {
oDb.SetChange("comp_rulesets_services")
}
return nil
Expand All @@ -62,11 +56,9 @@ func (oDb *DB) PurgeCompModulesetsServices(ctx context.Context) error {
svc_id NOT IN (
SELECT DISTINCT svc_id FROM svcmon
)`
if result, err := oDb.DB.ExecContext(ctx, query); err != nil {
if count, err := oDb.execCountContext(ctx, query); err != nil {
return err
} else if affected, err := result.RowsAffected(); err != nil {
return err
} else if affected > 0 {
} else if count > 0 {
oDb.SetChange("comp_modulesets_services")
}
return nil
Expand All @@ -78,11 +70,9 @@ func (oDb *DB) PurgeCompStatusOutdated(ctx context.Context) error {
FROM comp_status
WHERE
run_date < DATE_SUB(NOW(), INTERVAL 31 DAY)`
if result, err := oDb.DB.ExecContext(ctx, query); err != nil {
return err
} else if affected, err := result.RowsAffected(); err != nil {
if count, err := oDb.execCountContext(ctx, query); err != nil {
return err
} else if affected > 0 {
} else if count > 0 {
oDb.SetChange("comp_status")
}
return nil
Expand All @@ -96,11 +86,9 @@ func (oDb *DB) PurgeCompStatusSvcOrphans(ctx context.Context) error {
svc_id NOT IN (
SELECT DISTINCT svc_id FROM svcmon
)`
if result, err := oDb.DB.ExecContext(ctx, query); err != nil {
if count, err := oDb.execCountContext(ctx, query); err != nil {
return err
} else if affected, err := result.RowsAffected(); err != nil {
return err
} else if affected > 0 {
} else if count > 0 {
oDb.SetChange("comp_status")
}
return nil
Expand All @@ -114,11 +102,9 @@ func (oDb *DB) PurgeCompStatusNodeOrphans(ctx context.Context) error {
node_id NOT IN (
SELECT DISTINCT node_id FROM nodes
)`
if result, err := oDb.DB.ExecContext(ctx, query); err != nil {
return err
} else if affected, err := result.RowsAffected(); err != nil {
if count, err := oDb.execCountContext(ctx, query); err != nil {
return err
} else if affected > 0 {
} else if count > 0 {
oDb.SetChange("comp_status")
}
return nil
Expand All @@ -132,11 +118,9 @@ func (oDb *DB) PurgeCompStatusModulesetOrphans(ctx context.Context) error {
run_module NOT IN (
SELECT modset_mod_name FROM comp_moduleset_modules
)`
if result, err := oDb.DB.ExecContext(ctx, query); err != nil {
return err
} else if affected, err := result.RowsAffected(); err != nil {
if count, err := oDb.execCountContext(ctx, query); err != nil {
return err
} else if affected > 0 {
} else if count > 0 {
oDb.SetChange("comp_status")
}
return nil
Expand All @@ -156,11 +140,9 @@ func (oDb *DB) PurgeCompStatusNodeUnattached(ctx context.Context) error {
FROM comp_node_moduleset
)
)`
if result, err := oDb.DB.ExecContext(ctx, query); err != nil {
if count, err := oDb.execCountContext(ctx, query); err != nil {
return err
} else if affected, err := result.RowsAffected(); err != nil {
return err
} else if affected > 0 {
} else if count > 0 {
oDb.SetChange("comp_status")
}
return nil
Expand All @@ -180,11 +162,9 @@ func (oDb *DB) PurgeCompStatusSvcUnattached(ctx context.Context) error {
FROM comp_modulesets_services
)
)`
if result, err := oDb.DB.ExecContext(ctx, query); err != nil {
return err
} else if affected, err := result.RowsAffected(); err != nil {
if count, err := oDb.execCountContext(ctx, query); err != nil {
return err
} else if affected > 0 {
} else if count > 0 {
oDb.SetChange("comp_status")
}
return nil
Expand Down
Loading
Loading