HYPERFLEET-774 - feat: update aggregation logic

Author: rh-amarin

cmd/hyperfleet-api/migrate/cmd.go

@@ -58,11 +58,11 @@ func runMigrateWithError(ctx context.Context, dbConfig *config.DatabaseConfig) e
 		}
 	}()
 
-	// Use MigrateWithLock to prevent concurrent migrations from multiple pods
-	if err := db.MigrateWithLock(ctx, connection); err != nil {
+	if err := db.Migrate(connection.New(ctx)); err != nil {
 		logger.WithError(ctx, err).Error("Migration failed")
 		return err
 	}
 
+	logger.Info(ctx, "Migration completed successfully")
 	return nil
 }

docs/api-resources.md

@@ -249,6 +249,100 @@ The status uses Kubernetes-style conditions instead of a single phase field:
   - One adapter reports `Available=False` for `observed_generation=1` `Available` transitions to `False`
   - One adapter reports `Available=False` for `observed_generation=2` `Available` keeps its `True` status
 
+### Aggregation logic
+
+Description of the aggregation logic for the resource status conditions
+
+- An API that stores resources entities (clusters, nodepools)
+- A sentinel that polls the API for changes and triggers messages
+- Instances of "adapters":
+  - Read the messages
+  - Reconcile the state with the world
+  - Report back to the API, using statuses "conditions"
+
+Resources keep track of its status, which is affected by the reports from adapters
+
+- Each resource keeps a `generation` property that gets increased on every change
+- Adapters associated with a resource, report their state as an array of adapter conditions
+  - Three of these conditions are always mandatory : `Available`, `Applied`, `Health`
+  - If one of the mandatory conditions is missing, the report is discarded
+  - A `observed_generation` field indicating the generation associated with the report
+  - `observed_time` for when the adapter work was done
+  - If the reported `observed_generation` is lower than the already stored `observed_generation` for that adapter, the report is discarded
+- Each resource has a list of associated "adapters" used to compute the aggregated status.conditions
+- Each resource "status.conditions" is array property composed of:
+  - The `Available` condition of each adapter, named as `<adapter-name>Successful`
+  - 2 aggregated conditions: `Ready` and `Available` computed from the array of `Available` resource statuses conditions
+    - Only `Available` condition from adapters is used to compute aggregated conditions
+
+The whole API spec is at: <https://raw.githubusercontent.com/openshift-hyperfleet/hyperfleet-api/refs/heads/main/openapi/openapi.yaml>
+
+The aggregation logic for a resource (cluster/nodepool) works as follows.
+
+**Notation:**
+
+- `X` = report's `observed_generation`
+- `G` = resource's current `generation`
+- `statuses[]` = all stored adapter condition reports
+- `lut` = `last_update_time`
+- `ltt` = `last_transition_time`
+- `obs_gen` = `observed_generation`
+- `obs_time` = report's `observed_time`
+- `—` = no change
+
+---
+
+#### Discard / Reject Rules
+
+Checked before any aggregation. A discarded or rejected report causes no state change.
+
+| Rule | Condition | Outcome |
+|---|---|---|
+| `obs_gen` too high | report `observed_generation` > resource `generation` | Discarded |
+| Stale adapter report | report `observed_generation` < adapter's stored `observed_generation` | Discarded |
+| Missing mandatory conditions | Missing any of `Available`, `Applied`, `Health`, or value not in `{True, False, Unknown}` | Discarded |
+| Available=Unknown | Report is valid but `Available=Unknown` | Discarded |
+
+---
+
+#### Lifecycle Events
+
+| Event | Condition | Target | → status | → obs_gen | → lut | → ltt |
+|---|---|---|---|---|---|---|
+| Creation | — | `Ready` | `False` | `1` | `now` | `now` |
+| Creation | — | `Available` | `False` | `1` | `now` | `now` |
+| Change (→G) | Was `Ready=True` | `Ready` | `False` | `G` | `now` | `now` |
+| Change (→G) | Was `Ready=False` | `Ready` | `False` | `G` | `now` | `—` |
+| Change (→G) | — | `Available` | unchanged | unchanged | `—` | `—` |
+
+---
+
+#### Adapter Report Aggregation Matrix
+
+The **Ready** check and **Available** check are independent — both can apply to the same incoming report.
+
+##### Report `Available=True` (obs_gen = X)
+
+| Target | Current State | Required Condition | → status | → lut | → ltt | → obs_gen |
+|---|---|---|---|---|---|---|
+| `Ready` | `Ready=True` | `X==G` AND all `statuses[].obs_gen==G` AND all `statuses[].status==True` | unchanged | `min(statuses[].lut)` | `—` | `—` |
+| `Ready` | `Ready=False` | `X==G` AND all `statuses[].obs_gen==G` AND all `statuses[].status==True` | **`True`** | `min(statuses[].lut)` | `obs_time` | `—` |
+| `Ready` | any | Conditions above not met | `—` | `—` | `—` | `—` |
+| `Available` | `Available=False` | all `statuses[].obs_gen==X` | **`True`** | `min(statuses[].lut)` | `obs_time` | `X` |
+| `Available` | `Available=True` | all `statuses[].obs_gen==X` | unchanged | `min(statuses[].lut)` | `—` | `X` |
+| `Available` | any | Conditions above not met | `—` | `—` | `—` | `—` |
+
+##### Report `Available=False` (obs_gen = X)
+
+| Target | Current State | Required Condition | → status | → lut | → ltt | → obs_gen |
+|---|---|---|---|---|---|---|
+| `Ready` | `Ready=False` | `X==G` | unchanged | `min(statuses[].lut)` | `—` | `—` |
+| `Ready` | `Ready=True` | `X==G` | **`False`** | `obs_time` | `obs_time` | `—` |
+| `Ready` | any | Conditions above not met | `—` | `—` | `—` | `—` |
+| `Available` | `Available=False` | all `statuses[].obs_gen==X` | unchanged | `min(statuses[].lut)` | `—` | `X` |
+| `Available` | `Available=True` | all `statuses[].obs_gen==X` | **`False`** | `obs_time` | `obs_time` | `X` |
+| `Available` | any | Conditions above not met | `—` | `—` | `—` | `—` |
+
 ## NodePool Management
 
 ### Endpoints
@@ -456,7 +550,7 @@ The status object contains synthesized conditions computed from adapter reports:
 - All above fields plus:
 - `observed_generation` - Generation this condition reflects
 - `created_time` - When condition was first created (API-managed)
-- `last_updated_time` - When adapter last reported (API-managed, from AdapterStatus.last_report_time)
+- `last_updated_time` - When this condition was last refreshed (API-managed). For **Available**, always the evaluation time. For **Ready**: when Ready=True, the minimum of `last_report_time` across all required adapters that report Available=True at the current generation; when Ready=False, the evaluation time (so consumers can detect staleness).
 - `last_transition_time` - When status last changed (API-managed)
 
 ## Parameter Restrictions

pkg/config/db.go

@@ -44,14 +44,12 @@ type SSLConfig struct {
 // Includes fields from HYPERFLEET-694 for connection lifecycle management
 type PoolConfig struct {
 	MaxConnections int `mapstructure:"max_connections" json:"max_connections" validate:"required,min=1,max=200"`
-	MaxIdleConnections  int           `mapstructure:"max_idle_connections" json:"max_idle_connections" validate:"min=0"`
-	ConnMaxLifetime     time.Duration `mapstructure:"conn_max_lifetime" json:"conn_max_lifetime"`
-	ConnMaxIdleTime     time.Duration `mapstructure:"conn_max_idle_time" json:"conn_max_idle_time"`
-	RequestTimeout      time.Duration `mapstructure:"request_timeout" json:"request_timeout"`
-	ConnRetryAttempts   int           `mapstructure:"conn_retry_attempts" json:"conn_retry_attempts" validate:"min=1"`
-	ConnRetryInterval   time.Duration `mapstructure:"conn_retry_interval" json:"conn_retry_interval"`
-	// HYPERFLEET-618: prevents indefinite blocking during migrations
-	AdvisoryLockTimeout time.Duration `mapstructure:"advisory_lock_timeout" json:"advisory_lock_timeout"`
+	MaxIdleConnections int `mapstructure:"max_idle_connections" json:"max_idle_connections" validate:"min=0"`
+	ConnMaxLifetime time.Duration `mapstructure:"conn_max_lifetime" json:"conn_max_lifetime"`
+	ConnMaxIdleTime time.Duration `mapstructure:"conn_max_idle_time" json:"conn_max_idle_time"`
+	RequestTimeout time.Duration `mapstructure:"request_timeout" json:"request_timeout"`
+	ConnRetryAttempts int `mapstructure:"conn_retry_attempts" json:"conn_retry_attempts" validate:"min=1"`
+	ConnRetryInterval time.Duration `mapstructure:"conn_retry_interval" json:"conn_retry_interval"`
 }
 
 // MarshalJSON implements custom JSON marshaling to redact sensitive fields
@@ -93,14 +91,13 @@ func NewDatabaseConfig() *DatabaseConfig {
 			RootCertFile: "",
 		},
 		Pool: PoolConfig{
-			MaxConnections:      50,
-			MaxIdleConnections:  10,
-			ConnMaxLifetime:     5 * time.Minute,
-			ConnMaxIdleTime:     1 * time.Minute,
-			RequestTimeout:      30 * time.Second,
-			ConnRetryAttempts:   10,
-			ConnRetryInterval:   3 * time.Second,
-			AdvisoryLockTimeout: 5 * time.Minute, // HYPERFLEET-618: prevents indefinite blocking during migrations
+			MaxConnections:     50,
+			MaxIdleConnections: 10,
+			ConnMaxLifetime:    5 * time.Minute,
+			ConnMaxIdleTime:    1 * time.Minute,
+			RequestTimeout:     30 * time.Second,
+			ConnRetryAttempts:  10,
+			ConnRetryInterval:  3 * time.Second,
 		},
 	}
 }

pkg/config/loader.go

@@ -227,41 +227,43 @@ func (l *ConfigLoader) validateConfig(config *ApplicationConfig) error {
 }
 
 // handleJSONArrayEnvVars processes environment variables containing JSON arrays
-// Viper doesn't automatically parse JSON from env vars, so we handle this explicitly
-// Used for: HYPERFLEET_ADAPTERS_CLUSTER_ADAPTERS and HYPERFLEET_ADAPTERS_NODEPOOL_ADAPTERS
+// Viper doesn't automatically parse JSON from env vars, so we handle this explicitly.
+// Each viper key is filled from the first non-empty env var in the list (canonical name first, then aliases).
 func (l *ConfigLoader) handleJSONArrayEnvVars(ctx context.Context) error {
-	// Map of env var name -> viper key
-	jsonArrayMappings := map[string]string{
-		EnvPrefix + "_ADAPTERS_REQUIRED_CLUSTER":  "adapters.required.cluster",
-		EnvPrefix + "_ADAPTERS_REQUIRED_NODEPOOL": "adapters.required.nodepool",
+	// viper key -> ordered list of env var names (first one set wins)
+	clusterEnvVars := []string{
+		EnvPrefix + "_ADAPTERS_REQUIRED_CLUSTER",
+		EnvPrefix + "_CLUSTER_ADAPTERS", // alias for user convenience
+	}
+	nodepoolEnvVars := []string{
+		EnvPrefix + "_ADAPTERS_REQUIRED_NODEPOOL",
+		EnvPrefix + "_NODEPOOL_ADAPTERS", // alias for user convenience
 	}
 
-	for envVar, viperKey := range jsonArrayMappings {
-		jsonValue := os.Getenv(envVar)
-		if jsonValue == "" {
-			continue
-		}
-
-		// Parse JSON array
-		var arrayValue []string
-		if err := json.Unmarshal([]byte(jsonValue), &arrayValue); err != nil {
-			return fmt.Errorf("failed to parse %s as JSON array: %w (value: %s)", envVar, err, jsonValue)
+	setFromEnvVars := func(viperKey string, envVars []string) error {
+		for _, envVar := range envVars {
+			jsonValue := os.Getenv(envVar)
+			if jsonValue == "" {
+				continue
+			}
+			var arrayValue []string
+			if err := json.Unmarshal([]byte(jsonValue), &arrayValue); err != nil {
+				return fmt.Errorf("failed to parse %s as JSON array: %w (value: %s)", envVar, err, jsonValue)
+			}
+			// Set() overrides Viper's auto-env CSV parsing so JSON arrays are correct.
+			l.viper.Set(viperKey, arrayValue)
+			logger.With(ctx, "env_var", envVar, "count", len(arrayValue)).Debug("Parsed JSON array from environment")
+			return nil
 		}
-
-		// Always set the parsed JSON array value to override Viper's auto-env CSV parsing.
-		// Viper's AutomaticEnv treats comma-separated strings as arrays, incorrectly parsing
-		// JSON arrays like '["a","b"]' as ["[\"a\"", "\"b\"]"] instead of ["a", "b"].
-		//
-		// We use Set() to ensure proper JSON parsing overrides Viper's CSV parsing.
-		// This maintains ENV > Config > Default precedence for adapters.
-		//
-		// NOTE: Adapters currently have no CLI flags (see bindFlags line 494).
-		// If CLI flags are added in the future, this code needs updating to check
-		// if the value came from a flag before calling Set().
-		l.viper.Set(viperKey, arrayValue)
-		logger.With(ctx, "env_var", envVar, "count", len(arrayValue)).Debug("Parsed JSON array from environment")
+		return nil
 	}
 
+	if err := setFromEnvVars("adapters.required.cluster", clusterEnvVars); err != nil {
+		return err
+	}
+	if err := setFromEnvVars("adapters.required.nodepool", nodepoolEnvVars); err != nil {
+		return err
+	}
 	return nil
 }