Sync dependencies

cmd/icingadb/main.go

@@ -30,7 +30,7 @@ import (
 const (
 	ExitSuccess                = 0
 	ExitFailure                = 1
-	expectedRedisSchemaVersion = "5"
+	expectedRedisSchemaVersion = "6"
 )
 
 func main() {

pkg/contracts/contracts.go

@@ -27,3 +27,10 @@ func SafeInit(v any) {
 		initer.Init()
 	}
 }
+
+// Equaler is implemented by any entity that can be compared with another entity of the same type.
+// The Equal method should return true if the receiver is equal to the other entity.
+type Equaler interface {
+	// Equal returns whether the receiver is equal to the other entity.
+	Equal(other any) bool
+}

pkg/icingadb/delta.go

@@ -53,7 +53,7 @@ func (delta *Delta) run(ctx context.Context, actualCh, desiredCh <-chan database
 	desired := EntitiesById{} // only read from desiredCh (so far)
 
 	var update EntitiesById
-	if delta.Subject.WithChecksum() {
+	if _, ok := delta.Subject.Entity().(contracts.Equaler); ok || delta.Subject.WithChecksum() {
 		update = EntitiesById{} // read from actualCh and desiredCh with mismatching checksums
 	}
 
@@ -70,7 +70,7 @@ func (delta *Delta) run(ctx context.Context, actualCh, desiredCh <-chan database
 			id := actualValue.ID().String()
 			if desiredValue, ok := desired[id]; ok {
 				delete(desired, id)
-				if update != nil && !checksumsMatch(actualValue, desiredValue) {
+				if update != nil && !entitiesEqual(actualValue, desiredValue) {
 					update[id] = desiredValue
 				}
 			} else {
@@ -88,7 +88,7 @@ func (delta *Delta) run(ctx context.Context, actualCh, desiredCh <-chan database
 			id := desiredValue.ID().String()
 			if actualValue, ok := actual[id]; ok {
 				delete(actual, id)
-				if update != nil && !checksumsMatch(actualValue, desiredValue) {
+				if update != nil && !entitiesEqual(actualValue, desiredValue) {
 					update[id] = desiredValue
 				}
 			} else {
@@ -117,8 +117,14 @@ func (delta *Delta) run(ctx context.Context, actualCh, desiredCh <-chan database
 		zap.Int("delete", len(delta.Delete)))
 }
 
-// checksumsMatch returns whether the checksums of two entities are the same.
-// Both entities must implement contracts.Checksumer.
-func checksumsMatch(a, b database.Entity) bool {
-	return cmp.Equal(a.(contracts.Checksumer).Checksum(), b.(contracts.Checksumer).Checksum())
+// entitiesEqual returns whether the two entities are equal either based on their checksum or by comparing them.
+//
+// Both entities must either implement contracts.Checksumer or contracts.Equaler for this to work. If neither
+// interface is implemented nor if both entities don't implement the same interface, this function will panic.
+func entitiesEqual(a, b database.Entity) bool {
+	if _, ok := a.(contracts.Checksumer); ok {
+		return cmp.Equal(a.(contracts.Checksumer).Checksum(), b.(contracts.Checksumer).Checksum())
+	}
+
+	return a.(contracts.Equaler).Equal(b)
 }

pkg/icingadb/schema.go

@@ -15,8 +15,8 @@ import (
 )
 
 const (
-	expectedMysqlSchemaVersion    = 6
-	expectedPostgresSchemaVersion = 4
+	expectedMysqlSchemaVersion    = 7
+	expectedPostgresSchemaVersion = 5
 )
 
 // ErrSchemaNotExists implies that no Icinga DB schema has been imported.

pkg/icingadb/sync.go

@@ -148,9 +148,18 @@ func (s Sync) ApplyDelta(ctx context.Context, delta *Delta) error {
 		entitiesWithoutChecksum, errs := icingaredis.CreateEntities(ctx, delta.Subject.Factory(), pairs, runtime.NumCPU())
 		// Let errors from CreateEntities cancel our group.
 		com.ErrgroupReceive(g, errs)
-		entities, errs := icingaredis.SetChecksums(ctx, entitiesWithoutChecksum, delta.Update, runtime.NumCPU())
-		// Let errors from SetChecksums cancel our group.
-		com.ErrgroupReceive(g, errs)
+
+		var entities <-chan database.Entity
+		// Apply the checksums only if the sync subject supports it, i.e, it implements contracts.Checksumer.
+		// This is necessary because not only entities that implement contracts.Checksumer can be updated, but
+		// also entities that implement contracts.Equaler interface.
+		if delta.Subject.WithChecksum() {
+			entities, errs = icingaredis.SetChecksums(ctx, entitiesWithoutChecksum, delta.Update, runtime.NumCPU())
+			// Let errors from SetChecksums cancel our group.
+			com.ErrgroupReceive(g, errs)
+		} else {
+			entities = entitiesWithoutChecksum
+		}
 
 		g.Go(func() error {
 			// Using upsert here on purpose as this is the fastest way to do bulk updates.

pkg/icingadb/v1/checkable.go

@@ -15,6 +15,7 @@ type Checkable struct {
 	CheckTimeperiodName   string       `json:"check_timeperiod_name"`
 	CheckTimeperiodId     types.Binary `json:"check_timeperiod_id"`
 	CheckRetryInterval    float64      `json:"check_retry_interval"`
+	TotalChildren         types.Int    `json:"total_children"`
 	CheckTimeout          float64      `json:"check_timeout"`
 	CheckcommandName      string       `json:"checkcommand_name"`
 	CheckcommandId        types.Binary `json:"checkcommand_id"`

pkg/icingadb/v1/dependency.go

@@ -0,0 +1,100 @@
+package v1
+
+import (
+	"bytes"
+	"github.com/icinga/icinga-go-library/database"
+	"github.com/icinga/icinga-go-library/types"
+)
+
+type Redundancygroup struct {
+	EntityWithoutChecksum `json:",inline"`
+	EnvironmentMeta       `json:",inline"`
+	DisplayName           string `json:"display_name"`
+}
+
+// TableName implements [database.TableNamer].
+func (r *Redundancygroup) TableName() string {
+	return "redundancy_group"
+}
+
+type RedundancygroupState struct {
+	EntityWithoutChecksum `json:",inline"`
+	EnvironmentMeta       `json:",inline"`
+	RedundancyGroupId     types.Binary    `json:"redundancy_group_id"`
+	Failed                types.Bool      `json:"failed"`
+	IsReachable           types.Bool      `json:"is_reachable"`
+	LastStateChange       types.UnixMilli `json:"last_state_change"`
+}
+
+// TableName implements [database.TableNamer].
+func (r *RedundancygroupState) TableName() string {
+	return "redundancy_group_state"
+}
+
+// Equal implements the [contracts.Equaler] interface.
+func (r *RedundancygroupState) Equal(other any) bool {
+	if o, ok := other.(*RedundancygroupState); ok {
+		return bytes.Equal(r.Id, o.Id) &&
+			bytes.Equal(r.EnvironmentId, o.EnvironmentId) &&
+			bytes.Equal(r.RedundancyGroupId, o.RedundancyGroupId) &&
+			r.Failed == o.Failed &&
+			r.IsReachable == o.IsReachable &&
+			r.LastStateChange.Time().Equal(o.LastStateChange.Time())
+	}
+
+	return false
+}
+
+type DependencyNode struct {
+	EntityWithoutChecksum `json:",inline"`
+	EnvironmentMeta       `json:",inline"`
+	HostId                types.Binary `json:"host_id"`
+	ServiceId             types.Binary `json:"service_id"`
+	RedundancyGroupId     types.Binary `json:"redundancy_group_id"`
+}
+
+type DependencyEdgeState struct {
+	EntityWithoutChecksum `json:",inline"`
+	EnvironmentMeta       `json:",inline"`
+	Failed                types.Bool `json:"failed"`
+}
+
+// Equal implements the [contracts.Equaler] interface.
+func (es *DependencyEdgeState) Equal(other any) bool {
+	if other, ok := other.(*DependencyEdgeState); ok {
+		return bytes.Equal(es.Id, other.Id) &&
+			bytes.Equal(es.EnvironmentId, other.EnvironmentId) &&
+			es.Failed == other.Failed
+	}
+
+	return false
+}
+
+type DependencyEdge struct {
+	EntityWithoutChecksum `json:",inline"`
+	EnvironmentMeta       `json:",inline"`
+	FromNodeId            types.Binary `json:"from_node_id"`
+	ToNodeId              types.Binary `json:"to_node_id"`
+	DependencyEdgeStateId types.Binary `json:"dependency_edge_state_id"`
+	DisplayName           string       `json:"display_name"`
+}
+
+func NewRedundancygroup() database.Entity {
+	return &Redundancygroup{}
+}
+
+func NewRedundancygroupState() database.Entity {
+	return &RedundancygroupState{}
+}
+
+func NewDependencyNode() database.Entity {
+	return &DependencyNode{}
+}
+
+func NewDependencyEdgeState() database.Entity {
+	return &DependencyEdgeState{}
+}
+
+func NewDependencyEdge() database.Entity {
+	return &DependencyEdge{}
+}

pkg/icingadb/v1/state.go

@@ -17,6 +17,7 @@ type State struct {
 	ExecutionTime             float64                            `json:"execution_time"`
 	HardState                 uint8                              `json:"hard_state"`
 	InDowntime                types.Bool                         `json:"in_downtime"`
+	AffectsChildren           types.Bool                         `json:"affects_children"`
 	IsAcknowledged            icingadbTypes.AcknowledgementState `json:"is_acknowledged"`
 	IsFlapping                types.Bool                         `json:"is_flapping"`
 	IsHandled                 types.Bool                         `json:"is_handled"`

pkg/icingadb/v1/v1.go

@@ -4,7 +4,12 @@ import (
 	"github.com/icinga/icinga-go-library/database"
 )
 
-var StateFactories = []database.EntityFactoryFunc{NewHostState, NewServiceState}
+var StateFactories = []database.EntityFactoryFunc{
+	NewHostState,
+	NewServiceState,
+	NewDependencyEdgeState,
+	NewRedundancygroupState,
+}
 
 var ConfigFactories = []database.EntityFactoryFunc{
 	NewActionUrl,
@@ -51,6 +56,9 @@ var ConfigFactories = []database.EntityFactoryFunc{
 	NewUsergroupCustomvar,
 	NewUsergroupMember,
 	NewZone,
+	NewRedundancygroup,
+	NewDependencyNode,
+	NewDependencyEdge,
 }
 
 // contextKey is an unexported type for context keys defined in this package.

schema/mysql/schema.sql

@@ -188,6 +188,8 @@ CREATE TABLE host (
   check_interval int unsigned NOT NULL,
   check_retry_interval int unsigned NOT NULL,
 
+  total_children int unsigned DEFAULT NULL,
+
   active_checks_enabled enum('n', 'y') NOT NULL,
   passive_checks_enabled enum('n', 'y') NOT NULL,
   event_handler_enabled enum('n', 'y') NOT NULL,
@@ -313,6 +315,8 @@ CREATE TABLE host_state (
 
   in_downtime enum('n', 'y') NOT NULL,
 
+  affects_children enum('n', 'y') NOT NULL,
+
   execution_time int unsigned DEFAULT NULL,
   latency int unsigned DEFAULT NULL,
   check_timeout int unsigned DEFAULT NULL,
@@ -356,6 +360,8 @@ CREATE TABLE service (
   check_interval int unsigned NOT NULL,
   check_retry_interval int unsigned NOT NULL,
 
+  total_children int unsigned DEFAULT NULL,
+
   active_checks_enabled enum('n', 'y') NOT NULL,
   passive_checks_enabled enum('n', 'y') NOT NULL,
   event_handler_enabled enum('n', 'y') NOT NULL,
@@ -482,6 +488,8 @@ CREATE TABLE service_state (
 
   in_downtime enum('n', 'y') NOT NULL,
 
+  affects_children enum('n', 'y') NOT NULL,
+
   execution_time int unsigned DEFAULT NULL,
   latency int unsigned DEFAULT NULL,
   check_timeout int unsigned DEFAULT NULL,
@@ -1334,6 +1342,63 @@ CREATE TABLE sla_history_downtime (
   INDEX idx_sla_history_downtime_env_downtime_end (environment_id, downtime_end) COMMENT 'Filter for sla history retention'
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin ROW_FORMAT=DYNAMIC;
 
+CREATE TABLE redundancy_group (
+  id binary(20) NOT NULL COMMENT 'sha1(name + all(member parent_name + timeperiod.name + states + ignore_soft_states))',
+  environment_id binary(20) NOT NULL COMMENT 'environment.id',
+  display_name text NOT NULL,
+
+  CONSTRAINT pk_redundancy_group PRIMARY KEY (id)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin ROW_FORMAT=DYNAMIC;
+
+CREATE TABLE redundancy_group_state (
+  id binary(20) NOT NULL COMMENT 'redundancy_group.id',
+  environment_id binary(20) NOT NULL COMMENT 'environment.id',
+  redundancy_group_id binary(20) NOT NULL COMMENT 'redundancy_group.id',
+  failed enum('n', 'y') NOT NULL,
+  is_reachable enum('n', 'y') NOT NULL,
+  last_state_change BIGINT UNSIGNED NOT NULL,
+
+  CONSTRAINT pk_redundancy_group_state PRIMARY KEY (id),
+
+  UNIQUE INDEX idx_redundancy_group_state_redundancy_group_id (redundancy_group_id)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin ROW_FORMAT=DYNAMIC;
+
+CREATE TABLE dependency_node (
+  id binary(20) NOT NULL COMMENT 'host.id|service.id|redundancy_group.id',
+  environment_id binary(20) NOT NULL COMMENT 'environment.id',
+  host_id binary(20) DEFAULT NULL COMMENT 'host.id',
+  service_id binary(20) DEFAULT NULL COMMENT 'service.id',
+  redundancy_group_id binary(20) DEFAULT NULL COMMENT 'redundancy_group.id',
+
+  CONSTRAINT pk_dependency_node PRIMARY KEY (id),
+
+  UNIQUE INDEX idx_dependency_node_host_service_redundancygroup_id (host_id, service_id, redundancy_group_id),
+  CONSTRAINT ck_dependency_node_either_checkable_or_redundancy_group_id CHECK (
+    IF(redundancy_group_id IS NULL, host_id IS NOT NULL, host_id IS NULL AND service_id IS NULL) = 1
+  )
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin ROW_FORMAT=DYNAMIC;
+
+CREATE TABLE dependency_edge_state (
+  id binary(20) NOT NULL COMMENT 'sha1([dependency_edge.from_node_id|parent_name + timeperiod.name + states + ignore_soft_states] + dependency_edge.to_node_id)',
+  environment_id binary(20) NOT NULL COMMENT 'environment.id',
+  failed enum('n', 'y') NOT NULL,
+
+  CONSTRAINT pk_dependency_edge_state PRIMARY KEY (id)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin ROW_FORMAT=DYNAMIC;
+
+CREATE TABLE dependency_edge (
+  id binary(20) NOT NULL COMMENT 'sha1(from_node_id + to_node_id)',
+  environment_id binary(20) NOT NULL COMMENT 'environment.id',
+  from_node_id binary(20) NOT NULL COMMENT 'dependency_node.id',
+  to_node_id binary(20) NOT NULL COMMENT 'dependency_node.id',
+  dependency_edge_state_id binary(20) NOT NULL COMMENT 'dependency_edge_state.id',
+  display_name text NOT NULL,
+
+  CONSTRAINT pk_dependency_edge PRIMARY KEY (id),
+
+  UNIQUE INDEX idx_dependency_edge_from_node_to_node_id (from_node_id, to_node_id)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin ROW_FORMAT=DYNAMIC;
+
 CREATE TABLE icingadb_schema (
   id int unsigned NOT NULL AUTO_INCREMENT,
   version smallint unsigned NOT NULL,
@@ -1343,4 +1408,4 @@ CREATE TABLE icingadb_schema (
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin ROW_FORMAT=DYNAMIC;
 
 INSERT INTO icingadb_schema (version, timestamp)
-  VALUES (6, UNIX_TIMESTAMP() * 1000);
+  VALUES (7, UNIX_TIMESTAMP() * 1000);