diff --git a/cmd/icinga-notifications/main.go b/cmd/icinga-notifications/main.go index 1f7f5658..15cd9297 100644 --- a/cmd/icinga-notifications/main.go +++ b/cmd/icinga-notifications/main.go @@ -2,6 +2,11 @@ package main import ( "context" + "errors" + "os/signal" + "syscall" + "time" + "github.com/icinga/icinga-go-library/database" "github.com/icinga/icinga-go-library/logging" "github.com/icinga/icinga-go-library/utils" @@ -11,10 +16,8 @@ import ( "github.com/icinga/icinga-notifications/internal/daemon" "github.com/icinga/icinga-notifications/internal/incident" "github.com/icinga/icinga-notifications/internal/listener" + "github.com/icinga/icinga-notifications/internal/retention" "github.com/okzk/sdnotify" - "os/signal" - "syscall" - "time" ) func main() { @@ -62,6 +65,13 @@ func main() { logger.Fatalf("Cannot load incidents from database: %+v", err) } + ret := retention.New(db, logs.GetChildLogger("retention")) + go func() { + if err := ret.Run(ctx); err != nil && !errors.Is(err, context.Canceled) { + logger.Fatalf("Retention has finished with an error: %+v", err) + } + }() + // When Icinga Notifications is started by systemd, we've to notify systemd that we're ready. _ = sdnotify.Ready() diff --git a/config.example.yml b/config.example.yml index 00555d3e..f751568a 100644 --- a/config.example.yml +++ b/config.example.yml @@ -60,6 +60,35 @@ database: password: CHANGEME # password_file: /run/secerets/icinga_notifications_database_password +# The retention policy used by Icinga Notifications to determine how long no longer relevant historical data +# is kept in the database before it is eligible for cleanup. Currently, this retention policy only applies to +# closed incidents and related tables, but in the future it may be extended to other components as well. +retention: + # Retention period for historical data defined as duration string. This is the duration after which historical data + # is considered old and eligible for cleanup. By default, Icinga Notifications does not automatically clean up any + # data, i.e. the retention period is set to 0s. Setting this to a non-zero value enables the retention policy and + # allows Icinga Notifications to automatically clean up historical data that is older than the specified duration. + # + # Valid units are "ms", "s", "m", "h". Defaults to "0s", which means no automatic cleanup. +# period: 0s + + # Interval for periodic cleanup of historical data defined as duration string. This is the interval at which + # Icinga Notifications will run the cleanup process to prune old historical data as defined by the retention period. + # Valid units are "ms", "s", "m", "h". Defaults to "1h". +# interval: 1h + + # Batch size for deleting historical data. This is the maximum number of records that will be deleted in a single + # batch during the cleanup process. This is used to avoid long-running transactions and high load on the database + # when deleting large amounts of historical data. Defaults to 5000. +# batch_size: 5000 + + # Map of component-retention period pairs to define a different retention period than the default value for each + # component. The retention period for each component defines how long historical data related to that component + # is kept before it is eligible for cleanup. If a component is not listed here, the default retention period defined + # above is used. +# options: +# incident: 480h + # Icinga Notifications logs its activities at various severity levels and any errors that occur either # on the console or in systemd's journal. The latter is used automatically when running under systemd. # In any case, the default log level is 'info'. @@ -84,4 +113,5 @@ database: #database: #incident: #listener: + #retention: #runtime-updates: diff --git a/doc/03-Configuration.md b/doc/03-Configuration.md index 1d39031d..6d8bfe41 100644 --- a/doc/03-Configuration.md +++ b/doc/03-Configuration.md @@ -103,6 +103,57 @@ For environment variables, each option is prefixed with `ICINGA_NOTIFICATIONS_DA | max_rows_per_transaction | **Optional.** Maximum number of rows Icinga Notifications is allowed to `SELECT`,`DELETE`,`UPDATE` or `INSERT` in a single transaction. Defaults to `8192`. | | wsrep_sync_wait | **Optional.** Enforce [Galera cluster](#galera-cluster) nodes to perform strict cluster-wide causality checks. Defaults to `7`. | +## Retention Configuration + +You can configure the data retention and cleanup behavior of Icinga Notifications to automatically remove old data +from the database after a certain period. This helps to manage the size of the database and ensure that it does not +grow indefinitely. + +For YAML configuration, the options are part of the `retention` dictionary. +For environment variables, each option is prefixed with `ICINGA_NOTIFICATIONS_RETENTION_`. + +| Option | Description | +|------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| +| period | **Optional.** Retention period defined as [duration string](#duration-string). Defaults to `0s` (disabled). Data older than this period will be automatically deleted from the database. | +| interval | **Optional.** Interval for periodic retention checks defined as [duration string](#duration-string). Defaults to `1h`. | +| batch_size | **Optional.** Maximum number of rows to delete in a single transaction during retention cleanup. Defaults to `5000`. | +| options | **Optional.** Map of component name to retention period in order to set a different retention period for each component instead of the default one. See [retention components](#retention-components) for details. | + +!!! info + + By default, Icinga Notifications does not delete any data from the database, since the default retention period is + `0s`, which means that all data is retained indefinitely. You can set the `period` option to a non-zero value to + enable automatic data cleanup after the specified period has passed. + +If you don't want to enable automatic data cleanup for all components, you can let the `period` option be `0s` and set +non-zero retention periods for individual components in the `options` as [described below](#retention-components). +This goes also for the other way around, if you want to enable automatic data cleanup for all components, you can set +the `period` option to a non-zero value and set `0s` retention periods for a specific component to disable automatic +data cleanup for that component. + +### Retention Components + +In addition to the default retention period, you can set a different retention period for each component of +Icinga Notifications. This allows you to keep certain types of data for a longer or shorter period than others, +depending on their importance or relevance. + +For YAML configuration, the options are part of the `retention.options` dictionary. +For environment variables, each option is prefixed with `ICINGA_NOTIFICATIONS_RETENTION_OPTIONS_` and expects +a [duration string](#duration-string) as value. + +Currently, the following components are available: + +| Component | Description | +|-----------|-----------------------------------------------------------------------| +| incident | Incidents and all related data, such as their history, contacts, etc. | + +!!! info + + Incidents that are still open are not eligible for retention cleanup, even if they are open for months or years. + The retention policy only applies to closed incidents, and the retention period is computed from the time they were + closed, not from the time they were opened. If you want to clean up some incidents that are still open too, you need + to close them first, and then wait for the retention period to pass. + ## Logging Configuration Configuration of the logging component used by Icinga Notifications. @@ -151,6 +202,7 @@ ICINGA_NOTIFICATIONS_LOGGING_OPTIONS=database:error,listener:debug | database | Database connection status and queries. | | incident | Incident management and changes. | | listener | HTTP listener for event submission and debugging. | +| retention | Data retention and cleanup. | | runtime-updates | Configuration changes through Icinga Notifications Web from the database. | ## Appendix diff --git a/internal/daemon/config.go b/internal/daemon/config.go index 19f114d4..914da918 100644 --- a/internal/daemon/config.go +++ b/internal/daemon/config.go @@ -6,6 +6,7 @@ import ( "fmt" "net/url" "os" + "time" "github.com/creasty/defaults" "github.com/icinga/icinga-go-library/config" @@ -57,6 +58,7 @@ type ConfigFile struct { Icingaweb2URL string `yaml:"icingaweb2_url" env:"ICINGAWEB2_URL"` Listener Listener `yaml:"listener" envPrefix:"LISTENER_"` Database database.Config `yaml:"database" envPrefix:"DATABASE_"` + Retention Retention `yaml:"retention" envPrefix:"RETENTION_"` Logging logging.Config `yaml:"logging" envPrefix:"LOGGING_"` // IcingaWeb2UrlParsed holds the parsed Icinga Web 2 URL after validation of the config file. @@ -85,6 +87,9 @@ func (c *ConfigFile) Validate() error { if err := c.Logging.Validate(); err != nil { return err } + if err := c.Retention.Validate(); err != nil { + return err + } if c.Icingaweb2URL == "" { return errors.New("icingaweb2_url must be set") @@ -106,10 +111,51 @@ func (c *ConfigFile) Validate() error { return nil } +// RetentionOpts defines additional overrides for retention periods of specific components. +// +// Currently, we only have a single component (incidents), but this leaves room for future expansion +// without breaking the config structure. The fields here must be pointers to distinguish between +// "not set" and "set to zero" (i.e. no retention) when overriding the default retention period. +type RetentionOpts struct { + Incident *time.Duration `yaml:"incident" env:"INCIDENT"` +} + +// Validate implements the [config.Validator] interface. +func (r *RetentionOpts) Validate() error { + if r.Incident != nil && *r.Incident < 0 { + return errors.New("invalid retention period for incidents") + } + return nil +} + +// Retention defines the retention policy for Icinga Notifications history cleanups. +type Retention struct { + Period time.Duration `yaml:"period" env:"PERIOD"` + Interval time.Duration `yaml:"interval" env:"INTERVAL" default:"1h"` + BatchSize uint64 `yaml:"batch_size" env:"BATCH_SIZE" default:"5000"` + Options RetentionOpts `yaml:"options" envPrefix:"OPTIONS_"` +} + +// Validate implements the [config.Validator] interface. +func (r *Retention) Validate() error { + if r.Period < 0 { + return errors.New("invalid retention period") + } + if r.Interval <= 0 { + return errors.New("interval must be greater than zero") + } + if r.BatchSize == 0 { + return errors.New("'batch_size' must be greater than zero") + } + return r.Options.Validate() +} + // Assert interface compliance. var ( _ defaults.Setter = (*ConfigFile)(nil) _ config.Validator = (*ConfigFile)(nil) + _ config.Validator = (*Listener)(nil) + _ config.Validator = (*Retention)(nil) ) // Flags defines the CLI flags supported by Icinga Notifications. diff --git a/internal/retention/prune.go b/internal/retention/prune.go new file mode 100644 index 00000000..0f261a5b --- /dev/null +++ b/internal/retention/prune.go @@ -0,0 +1,169 @@ +package retention + +import ( + "context" + "fmt" + + "github.com/icinga/icinga-go-library/backoff" + "github.com/icinga/icinga-go-library/com" + "github.com/icinga/icinga-go-library/database" + "github.com/icinga/icinga-go-library/retry" + "github.com/icinga/icinga-go-library/types" + "github.com/jmoiron/sqlx" +) + +// TimeBoundPruner defines the configuration for pruning rows from a table based on a time column. +// +// This struct is designed to be flexible and reusable for different tables with varying time columns and +// primary keys. It also supports maintaining referential integrity by allowing the definition of related +// tables that reference the primary keys of the main table, ensuring that any related rows are also pruned +// accordingly. +type TimeBoundPruner struct { + Table string + PK string + TimeColumn string + + Referrers []ReferencingRowPruner +} + +// Exec prunes rows from the specified table that are older than the given time threshold. +// +// If Referrers are defined, it first retrieves the primary keys of the rows to be deleted and then executes the +// corresponding DELETE statements for each [ReferencingRowPruner] before finally deleting the rows from the main +// table. If no Referrers are defined, it directly deletes rows based on the time column. +// +// The method returns only the total number of deleted rows from the main table. +func (tbp *TimeBoundPruner) Exec(ctx context.Context, db *database.DB, olderThan types.UnixMilli, limit uint64) (uint64, error) { + deleteStmt := tbp.assembleDelete(db.DriverName(), limit, len(tbp.Referrers) > 0) + if len(tbp.Referrers) == 0 { + return exec(ctx, db, deleteStmt, limit, olderThan) + } + + selectStmt := tbp.assembleSelect(limit) + var total uint64 + for { + var ids []any + err := retry.WithBackoff( + ctx, + func(ctx context.Context) error { + if err := db.SelectContext(ctx, &ids, db.Rebind(selectStmt), olderThan); err != nil { + return database.CantPerformQuery(err, selectStmt) + } + return nil + }, + retry.Retryable, + backoff.DefaultBackoff, + db.GetDefaultRetrySettings(), + ) + if err != nil { + return 0, err + } + + if len(ids) == 0 { + // No rows to delete, so we can skip executing the referrers and the final delete statement. + return total, nil + } + + for _, referrer := range tbp.Referrers { + if _, err := referrer.Exec(ctx, db, ids...); err != nil { + return 0, err + } + } + + if affected, err := exec(ctx, db, deleteStmt, 0, ids...); err != nil { + return 0, err + } else { + total += affected + } + } +} + +// assembleSelect constructs a select stmt to retrieve primary keys of this pruner based on the time column and limit. +func (tbp *TimeBoundPruner) assembleSelect(limit uint64) string { + return fmt.Sprintf(`SELECT %s FROM %s WHERE %[3]s IS NOT NULL AND %[3]s < ? LIMIT %d`, tbp.PK, tbp.Table, tbp.TimeColumn, limit) +} + +// assembleDelete constructs a delete stmt for this pruner based on the database driver and whether we are +// deleting by primary keys or by time column. +func (tbp *TimeBoundPruner) assembleDelete(driverName string, limit uint64, byPKs bool) string { + if byPKs { + // The limit doesn't apply when deleting by PKs, as the number of PKs is already limited by the provided arguments. + return fmt.Sprintf(`DELETE FROM %s WHERE %s IN (?)`, tbp.Table, tbp.PK) + } + + switch driverName { + case database.MySQL: + return fmt.Sprintf(`DELETE FROM %s WHERE %[2]s IS NOT NULL AND %[2]s < ? LIMIT %d`, tbp.Table, tbp.TimeColumn, limit) + case database.PostgreSQL: + return fmt.Sprintf(` +WITH rows AS (SELECT %[1]s FROM %[2]s WHERE %[3]s IS NOT NULL AND %[3]s < ? LIMIT %[4]d) +DELETE FROM %[2]s WHERE %[1]s IN (SELECT * FROM rows)`, + tbp.PK, tbp.Table, tbp.TimeColumn, limit) + default: + panic(fmt.Sprintf("invalid database type %s", driverName)) + } +} + +// ReferencingRowPruner defines the configuration for pruning rows that reference the primary keys of another table. +// +// This struct is used to maintain referential integrity when pruning rows from a main table by ensuring that +// any related rows in other tables are also pruned accordingly. In other words, it allows to explicitly define +// cascading deletes for related tables without relying on database-level foreign key constraints. +type ReferencingRowPruner struct { + Table string + FK string +} + +// Exec performs the pruning operation by deleting rows from the specified table that reference the given primary keys. +func (rrp *ReferencingRowPruner) Exec(ctx context.Context, db *database.DB, pks ...any) (uint64, error) { + return exec(ctx, db, rrp.assembleDelete(), 0, pks...) +} + +// assembleDelete constructs the delete statement for the ReferencingRowPruner based on the foreign key. +func (rrp *ReferencingRowPruner) assembleDelete() string { + return fmt.Sprintf(`DELETE FROM %s WHERE %s IN (?)`, rrp.Table, rrp.FK) +} + +// exec executes the given delete statement with the provided arguments and limit, handling retries and counting affected rows. +func exec(ctx context.Context, db *database.DB, query string, limit uint64, args ...any) (uint64, error) { + stmt, expandedArgs, err := sqlx.In(query, args) + if err != nil { + return 0, err + } + + var counter com.Counter + defer db.Log(ctx, query, &counter).Stop() + + for { + var affected uint64 + err = retry.WithBackoff( + ctx, + func(ctx context.Context) error { + res, err := db.ExecContext(ctx, db.Rebind(stmt), expandedArgs...) + if err != nil { + return database.CantPerformQuery(err, query) + } + + n, err := res.RowsAffected() + if err == nil && n > 0 { + affected = uint64(n) + } + return err + }, + retry.Retryable, + backoff.DefaultBackoff, + db.GetDefaultRetrySettings(), + ) + if err != nil { + return 0, err + } + + counter.Add(affected) + // If limit is set to 0, it means we are deleting matching rows by primary keys, so the limit check can be skipped. + if limit == 0 || affected < limit { + break + } + } + + return counter.Total(), nil +} diff --git a/internal/retention/retention.go b/internal/retention/retention.go new file mode 100644 index 00000000..26350717 --- /dev/null +++ b/internal/retention/retention.go @@ -0,0 +1,106 @@ +package retention + +import ( + "context" + + "github.com/icinga/icinga-go-library/database" + "github.com/icinga/icinga-go-library/logging" + "github.com/icinga/icinga-go-library/periodic" + "github.com/icinga/icinga-go-library/types" + "github.com/icinga/icinga-notifications/internal/daemon" + "go.uber.org/zap" +) + +// Retention is responsible for periodically pruning old data from the database. +// +// It uses a list of [TimeBoundPruner] configurations to determine which tables to prune and how to maintain +// referential integrity with related tables. The retention process runs in a separate goroutine for each pruner, +// allowing for concurrent pruning of multiple tables without blocking one another. The retention intervals and +// limits are configurable, and the process can be gracefully stopped using the provided context. +type Retention struct { + db *database.DB + logger *logging.Logger +} + +// New creates a new instance of the Retention struct with the provided database connection and logger. +func New(db *database.DB, logger *logging.Logger) *Retention { + return &Retention{db: db, logger: logger} +} + +// Run starts the retention process, which periodically prunes old data from the database. +// +// For each configured pruner, it sets up a periodic task that executes the pruning logic at the specified intervals. +func (r *Retention) Run(ctx context.Context) error { + conf := daemon.Config().Retention + + var errs chan error + for _, pruner := range dbPruners { + period := conf.Period + if pruner.Table == "incident" && conf.Options.Incident != nil { + period = *conf.Options.Incident + } + + if period == 0 { + r.logger.Debugf("Skipping retention for table %s because retention period is set to 0", pruner.Table) + continue + } + + if errs == nil { + errs = make(chan error, 1) + } + + r.logger.Debugw("Starting retention", + zap.String("table", pruner.Table), + zap.Duration("interval", conf.Interval), + zap.Duration("period", period)) + + periodic.Start(ctx, conf.Interval, func(tick periodic.Tick) { + olderThan := tick.Time.Add(-period) + + r.logger.Debugf("Pruning data from table %s older than %s", pruner.Table, olderThan) + + deleted, err := pruner.Exec(ctx, r.db, types.UnixMilli(olderThan), conf.BatchSize) + if err != nil { + select { + case errs <- err: + case <-ctx.Done(): + } + } + + level := zap.DebugLevel + if deleted > 0 { + level = zap.InfoLevel + } + r.logger.Logf(level, "Removed %d old items from table %s", deleted, pruner.Table) + }, periodic.Immediate()) + } + + if errs != nil { + select { + case err := <-errs: + return err + case <-ctx.Done(): + return ctx.Err() + } + } + return nil +} + +// dbPruners defines the list of tables and their corresponding time columns to be pruned, along with any related +// tables that need to be pruned in a cascading manner to maintain referential integrity. Each entry specifies the +// main table to prune, its primary key, the time column used for determining which rows are old enough to delete, +// and any referrer tables that have foreign key relationships with the main table. +var dbPruners = []TimeBoundPruner{ + { + Table: "incident", + PK: "id", + TimeColumn: "recovered_at", + Referrers: []ReferencingRowPruner{ + {Table: "incident_contact", FK: "incident_id"}, + {Table: "incident_rule", FK: "incident_id"}, + // Incident history references `incident_rule_escalation_state` too, so must appear before it in the cascade. + {Table: "incident_history", FK: "incident_id"}, + {Table: "incident_rule_escalation_state", FK: "incident_id"}, + }, + }, +} diff --git a/schema/mysql/schema.sql b/schema/mysql/schema.sql index e9e96995..dbd2e686 100644 --- a/schema/mysql/schema.sql +++ b/schema/mysql/schema.sql @@ -360,6 +360,8 @@ CREATE TABLE incident ( CONSTRAINT fk_incident_object FOREIGN KEY (object_id) REFERENCES object(id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin; +CREATE INDEX idx_incident_recovered_at ON incident(recovered_at); + CREATE TABLE incident_contact ( incident_id bigint NOT NULL, contact_id bigint, diff --git a/schema/mysql/upgrades/incident-recovery_at-idx.sql b/schema/mysql/upgrades/incident-recovery_at-idx.sql new file mode 100644 index 00000000..7100ba99 --- /dev/null +++ b/schema/mysql/upgrades/incident-recovery_at-idx.sql @@ -0,0 +1 @@ +CREATE INDEX idx_incident_recovered_at ON incident(recovered_at); diff --git a/schema/pgsql/schema.sql b/schema/pgsql/schema.sql index a5a3ed10..e31fc151 100644 --- a/schema/pgsql/schema.sql +++ b/schema/pgsql/schema.sql @@ -387,6 +387,8 @@ CREATE TABLE incident ( CONSTRAINT fk_incident_object FOREIGN KEY (object_id) REFERENCES object(id) ); +CREATE INDEX idx_incident_recovered_at ON incident(recovered_at); + CREATE TYPE incident_contact_role AS ENUM ('recipient', 'subscriber', 'manager'); CREATE TABLE incident_contact ( diff --git a/schema/pgsql/upgrades/incident-recovery_at-idx.sql b/schema/pgsql/upgrades/incident-recovery_at-idx.sql new file mode 100644 index 00000000..7100ba99 --- /dev/null +++ b/schema/pgsql/upgrades/incident-recovery_at-idx.sql @@ -0,0 +1 @@ +CREATE INDEX idx_incident_recovered_at ON incident(recovered_at);