external: deduplicate external MGR endpoints in EndpointSlice
When the active MGR IP already exists at a non-zero position in the externalMgrEndpoints list, the rotation logic that replaces endpoints[0] creates duplicate IPs. Also guards against mgr map failure which could overwrite endpoints[0] with an empty string. Fixes: #17761 Signed-off-by: beyondcloud-co <87993136+beyondcloud-co@users.noreply.github.com>
This commit is contained in:
beyondcloud-co committed
1 parent
f77e54dce5
commit
8134d683e3
2 files changed
+167
-6
No files matched your search
@@ -909,9 +909,8 @@ func ConfigureExternalMetricsEndpoint(ctx *clusterd.Context, monitoringSpec ceph
|
||||
return nil
|
||||
}
|
||||
|
||||
// Get active mgr
|
||||
var activeMgrAddr string
|
||||
// We use mgr dump and not stat because we want the IP address
|
||||
var activeMgrAddr string
|
||||
mgrMap, err := client.CephMgrMap(ctx, clusterInfo)
|
||||
if err != nil {
|
||||
logger.Errorf("failed to get mgr map. %v", err)
|
||||
@@ -920,12 +919,47 @@ func ConfigureExternalMetricsEndpoint(ctx *clusterd.Context, monitoringSpec ceph
|
||||
}
|
||||
logger.Debugf("active mgr addr %q", activeMgrAddr)
|
||||
|
||||
// If the active manager is different than the one in the spec we override it
|
||||
// This happens when a standby manager becomes active
|
||||
if activeMgrAddr != monitoringSpec.ExternalMgrEndpoints[0].IP {
|
||||
monitoringSpec.ExternalMgrEndpoints[0].IP = activeMgrAddr
|
||||
activeMgrIndex := -1
|
||||
if activeMgrAddr != "" {
|
||||
for i, ep := range monitoringSpec.ExternalMgrEndpoints {
|
||||
if ep.IP == activeMgrAddr {
|
||||
activeMgrIndex = i
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Build a deduplicated endpoint list with the active MGR at position [0].
|
||||
// Allocate a new slice to avoid mutating the caller's backing array.
|
||||
deduped := make([]v1.EndpointAddress, 0, len(monitoringSpec.ExternalMgrEndpoints))
|
||||
seen := sets.New[string]()
|
||||
|
||||
// activeMgrIsNew means the active MGR IP is known but not in the current list
|
||||
activeMgrIsNew := activeMgrAddr != "" && activeMgrIndex < 0
|
||||
|
||||
if activeMgrIndex >= 0 {
|
||||
deduped = append(deduped, monitoringSpec.ExternalMgrEndpoints[activeMgrIndex])
|
||||
seen.Insert(activeMgrAddr)
|
||||
} else if activeMgrIsNew {
|
||||
deduped = append(deduped, v1.EndpointAddress{IP: activeMgrAddr})
|
||||
seen.Insert(activeMgrAddr)
|
||||
}
|
||||
|
||||
// Append remaining unique endpoints, skipping the active mgr and old [0] if it was replaced.
|
||||
for i, ep := range monitoringSpec.ExternalMgrEndpoints {
|
||||
if i == activeMgrIndex {
|
||||
continue
|
||||
}
|
||||
if activeMgrIsNew && i == 0 {
|
||||
continue
|
||||
}
|
||||
if !seen.Has(ep.IP) {
|
||||
seen.Insert(ep.IP)
|
||||
deduped = append(deduped, ep)
|
||||
}
|
||||
}
|
||||
monitoringSpec.ExternalMgrEndpoints = deduped
|
||||
|
||||
// Create external monitoring Endpoints
|
||||
endpoint, err := createExternalMetricsEndpoints(clusterInfo.Namespace, monitoringSpec, ownerInfo)
|
||||
if err != nil {
|
||||
|
||||
@@ -405,6 +405,133 @@ func TestConfigureExternalMetricsEndpoint(t *testing.T) {
|
||||
assert.NoError(t, err)
|
||||
assert.Equal(t, "192.168.0.1", currentEndpoints.Endpoints[0].Addresses[0], currentEndpoints)
|
||||
})
|
||||
|
||||
t.Run("active mgr at non-zero position should deduplicate without losing other endpoints", func(t *testing.T) {
|
||||
monitoringSpec := cephv1.MonitoringSpec{
|
||||
Enabled: true,
|
||||
ExternalMgrEndpoints: []v1.EndpointAddress{
|
||||
{IP: "192.168.0.1"}, // [0] stale
|
||||
{IP: "192.168.0.2"}, // [1] active mgr
|
||||
{IP: "192.168.0.3"}, // [2] standby
|
||||
},
|
||||
}
|
||||
executor := &exectest.MockExecutor{
|
||||
MockExecuteCommandWithOutput: func(command string, args ...string) (string, error) {
|
||||
if args[1] == "dump" {
|
||||
return fmt.Sprintf(`{"active_addr":"%s"}`, "192.168.0.2:6801/2535462469"), nil
|
||||
}
|
||||
return "", errors.New("unknown command")
|
||||
},
|
||||
}
|
||||
ctx := &clusterd.Context{
|
||||
Clientset: test.New(t, 3),
|
||||
RookClientset: rookclient.NewSimpleClientset(),
|
||||
Executor: executor,
|
||||
}
|
||||
|
||||
err := ConfigureExternalMetricsEndpoint(ctx, monitoringSpec, clusterInfo, cephclient.NewMinimumOwnerInfo(t))
|
||||
assert.NoError(t, err)
|
||||
|
||||
currentEndpoints, err := ctx.Clientset.DiscoveryV1().EndpointSlices(namespace).Get(context.TODO(), "rook-ceph-mgr-external", metav1.GetOptions{})
|
||||
assert.NoError(t, err)
|
||||
// Active mgr at [0], stale [0] preserved, no duplicates
|
||||
assert.Equal(t, []string{"192.168.0.2", "192.168.0.1", "192.168.0.3"}, currentEndpoints.Endpoints[0].Addresses)
|
||||
})
|
||||
|
||||
t.Run("mgr map failure should not overwrite endpoints", func(t *testing.T) {
|
||||
monitoringSpec := cephv1.MonitoringSpec{
|
||||
Enabled: true,
|
||||
ExternalMgrEndpoints: []v1.EndpointAddress{
|
||||
{IP: "192.168.0.1"},
|
||||
{IP: "192.168.0.2"},
|
||||
},
|
||||
}
|
||||
executor := &exectest.MockExecutor{
|
||||
MockExecuteCommandWithOutput: func(command string, args ...string) (string, error) {
|
||||
if args[1] == "dump" {
|
||||
return "", errors.New("failed to get mgr map")
|
||||
}
|
||||
return "", errors.New("unknown command")
|
||||
},
|
||||
}
|
||||
ctx := &clusterd.Context{
|
||||
Clientset: test.New(t, 3),
|
||||
RookClientset: rookclient.NewSimpleClientset(),
|
||||
Executor: executor,
|
||||
}
|
||||
|
||||
err := ConfigureExternalMetricsEndpoint(ctx, monitoringSpec, clusterInfo, cephclient.NewMinimumOwnerInfo(t))
|
||||
assert.NoError(t, err)
|
||||
|
||||
currentEndpoints, err := ctx.Clientset.DiscoveryV1().EndpointSlices(namespace).Get(context.TODO(), "rook-ceph-mgr-external", metav1.GetOptions{})
|
||||
assert.NoError(t, err)
|
||||
// Endpoints unchanged when active mgr cannot be determined
|
||||
assert.Equal(t, []string{"192.168.0.1", "192.168.0.2"}, currentEndpoints.Endpoints[0].Addresses)
|
||||
})
|
||||
|
||||
t.Run("pre-existing duplicates should be collapsed", func(t *testing.T) {
|
||||
monitoringSpec := cephv1.MonitoringSpec{
|
||||
Enabled: true,
|
||||
ExternalMgrEndpoints: []v1.EndpointAddress{
|
||||
{IP: "192.168.0.1"}, // [0] active mgr
|
||||
{IP: "192.168.0.1"}, // [1] duplicate
|
||||
{IP: "192.168.0.2"}, // [2] standby
|
||||
},
|
||||
}
|
||||
executor := &exectest.MockExecutor{
|
||||
MockExecuteCommandWithOutput: func(command string, args ...string) (string, error) {
|
||||
if args[1] == "dump" {
|
||||
return fmt.Sprintf(`{"active_addr":"%s"}`, "192.168.0.1:6801/2535462469"), nil
|
||||
}
|
||||
return "", errors.New("unknown command")
|
||||
},
|
||||
}
|
||||
ctx := &clusterd.Context{
|
||||
Clientset: test.New(t, 3),
|
||||
RookClientset: rookclient.NewSimpleClientset(),
|
||||
Executor: executor,
|
||||
}
|
||||
|
||||
err := ConfigureExternalMetricsEndpoint(ctx, monitoringSpec, clusterInfo, cephclient.NewMinimumOwnerInfo(t))
|
||||
assert.NoError(t, err)
|
||||
|
||||
currentEndpoints, err := ctx.Clientset.DiscoveryV1().EndpointSlices(namespace).Get(context.TODO(), "rook-ceph-mgr-external", metav1.GetOptions{})
|
||||
assert.NoError(t, err)
|
||||
// Active mgr at [0], duplicate removed
|
||||
assert.Equal(t, []string{"192.168.0.1", "192.168.0.2"}, currentEndpoints.Endpoints[0].Addresses)
|
||||
})
|
||||
|
||||
t.Run("active mgr already at [0] with multi-element list should preserve order minus duplicates", func(t *testing.T) {
|
||||
monitoringSpec := cephv1.MonitoringSpec{
|
||||
Enabled: true,
|
||||
ExternalMgrEndpoints: []v1.EndpointAddress{
|
||||
{IP: "192.168.0.1"}, // [0] active mgr
|
||||
{IP: "192.168.0.2"}, // [1] standby 1
|
||||
{IP: "192.168.0.3"}, // [2] standby 2
|
||||
},
|
||||
}
|
||||
executor := &exectest.MockExecutor{
|
||||
MockExecuteCommandWithOutput: func(command string, args ...string) (string, error) {
|
||||
if args[1] == "dump" {
|
||||
return fmt.Sprintf(`{"active_addr":"%s"}`, "192.168.0.1:6801/2535462469"), nil
|
||||
}
|
||||
return "", errors.New("unknown command")
|
||||
},
|
||||
}
|
||||
ctx := &clusterd.Context{
|
||||
Clientset: test.New(t, 3),
|
||||
RookClientset: rookclient.NewSimpleClientset(),
|
||||
Executor: executor,
|
||||
}
|
||||
|
||||
err := ConfigureExternalMetricsEndpoint(ctx, monitoringSpec, clusterInfo, cephclient.NewMinimumOwnerInfo(t))
|
||||
assert.NoError(t, err)
|
||||
|
||||
currentEndpoints, err := ctx.Clientset.DiscoveryV1().EndpointSlices(namespace).Get(context.TODO(), "rook-ceph-mgr-external", metav1.GetOptions{})
|
||||
assert.NoError(t, err)
|
||||
// Active mgr stays at [0], remaining order preserved
|
||||
assert.Equal(t, []string{"192.168.0.1", "192.168.0.2", "192.168.0.3"}, currentEndpoints.Endpoints[0].Addresses)
|
||||
})
|
||||
}
|
||||
|
||||
func TestLogCollectorContainer(t *testing.T) {
|
||||
|
||||
Reference in new issue
Block a user