fix(memorystore): prevent infinite reconciliation drift loop by aligning connections list length (#11547)

### Why is this change necessary?
During E2E testing of network connectivity topologies, we observed that
`MemorystoreInstance` controllers can fall into an infinite
reconciliation loop due to false drift detection on the
`endpoints[i].connections` list.

Specifically:
- In the KRM spec (`desired`), a user defines the target VPC network(s)
under `spec.endpoints[i].connections` (typically 1 connection item
representing the intended PSC network target).
- Under the hood, GCP automatically allocates and returns **multiple**
server-generated connection entries per target network (e.g., one
connection per shard or node attachment).
- Because the GCP API returns an `actual` connections list that is
longer than the user's `desired` list, `tags.DiffForTopLevelFields`
flags the length mismatch as a drift. KCC repeatedly sends update
requests attempting to shrink the connections list, which GCP ignores,
resulting in an infinite reconciliation loop.

### What does this change do?
1. **Normalizes `Connections` list length before diffing
(`pkg/controller/direct/memorystore/memorystoreinstance_controller.go`)**:
Truncates the server-generated extra `Connections` in `maskedActual` to
match `len(desiredEndpoint.Connections)` before comparison
(`actualEndpoint.Connections =
actualEndpoint.Connections[:len(desiredEndpoint.Connections)]`).
- **Why truncating is correct and does not cause data loss**: Truncation
only applies to the temporary `maskedActual` copy used for drift
comparison against the user's `spec`. It does not discard any
user-configured target connections, nor does it affect the full observed
state reported in `status.endpoints`. It ensures we compare exactly the
user-declared intent (`desired`) against the corresponding actual
connection items without failing on extra server-created replica
entries.
2. **Prevents partial status overwrites
(`pkg/controller/direct/directbase/operations.go`)**:
Updates `directbase/operations.go` to prevent partial status updates
from overwriting existing status fields (such as `conditions` or
`externalRef`) with `nil`.
3. **Sorts HTTP logs in E2E tests**:
Aligns/sorts HTTP log entries for the `MemorystoreInstance` test
fixtures to prevent test flakiness.

Author: acpana

pkg/controller/direct/memorystore/memorystoreinstance_controller.go

@@ -396,13 +396,89 @@ func compareInstance(ctx context.Context, actual, desired *memorystorepb.Instanc
 		desired.CrossInstanceReplicationConfig = maskedActual.CrossInstanceReplicationConfig
 	}
 
+	// Align actual endpoints/connections with desired when desired is a subset of actual.
+	alignEndpointsWithDesired(maskedActual, desired)
+
 	diffs, _, err := tags.DiffForTopLevelFields(ctx, desired.ProtoReflect(), maskedActual.ProtoReflect())
 	if err != nil {
 		return nil, err
 	}
 	return diffs, nil
 }
 
+// alignEndpointsWithDesired aligns actual endpoints with desired when desired is a subset of actual.
+// It matches each desired endpoint to a corresponding actual endpoint and retains only matched endpoints
+// in actual so that unmanaged server-generated extra endpoints do not trigger false drift.
+func alignEndpointsWithDesired(actual, desired *memorystorepb.Instance) {
+	if len(desired.Endpoints) == 0 {
+		actual.Endpoints = nil
+		return
+	}
+
+	alignedEndpoints := make([]*memorystorepb.Instance_InstanceEndpoint, 0, len(desired.Endpoints))
+
+	for _, desiredEndpoint := range desired.Endpoints {
+		for _, actualEndpoint := range actual.Endpoints {
+			if isConnectionSubset(desiredEndpoint.GetConnections(), actualEndpoint.GetConnections()) {
+				matchedEndpoint := &memorystorepb.Instance_InstanceEndpoint{
+					Connections: desiredEndpoint.Connections,
+				}
+				alignedEndpoints = append(alignedEndpoints, matchedEndpoint)
+				break
+			}
+		}
+	}
+
+	if len(alignedEndpoints) == len(desired.Endpoints) {
+		actual.Endpoints = alignedEndpoints
+	}
+}
+
+// isConnectionSubset checks if every desired connection is matched by at least one actual connection.
+func isConnectionSubset(desiredConnections, actualConnections []*memorystorepb.Instance_ConnectionDetail) bool {
+	for _, desiredConn := range desiredConnections {
+		matched := false
+
+		for _, actualConn := range actualConnections {
+			if connectionsMatch(desiredConn, actualConn) {
+				matched = true
+				break
+			}
+		}
+
+		if !matched {
+			return false
+		}
+	}
+
+	return true
+}
+
+// connectionsMatch checks if all specified fields in a desired connection match an actual connection.
+func connectionsMatch(desired, actual *memorystorepb.Instance_ConnectionDetail) bool {
+	if desired == nil || actual == nil {
+		return desired == actual
+	}
+
+	desiredPsc := desired.GetPscAutoConnection()
+	actualPsc := actual.GetPscAutoConnection()
+	if desiredPsc == nil || actualPsc == nil {
+		return desiredPsc == actualPsc
+	}
+
+	// Only match against project and network since CRD implementation uses project and network refs.
+	// So no PSC fields will be present in the desired spec.
+	if desiredPsc.GetProjectId() != "" && desiredPsc.GetProjectId() != actualPsc.GetProjectId() {
+		return false
+	}
+
+	if desiredPsc.GetNetwork() != "" && desiredPsc.GetNetwork() != actualPsc.GetNetwork() {
+		return false
+	}
+
+	return true
+}
+
 // Export maps the GCP object to a Config Connector resource `spec`.
 func (a *InstanceAdapter) Export(ctx context.Context) (*unstructured.Unstructured, error) {
 	if a.actual == nil {