forked from rook/rook
Merge pull request #8330 from travisn/reconcile-mgr-services
ceph: Reconcile mgr services with every reconcile
This commit is contained in:
+10
-3
@@ -25,6 +25,7 @@ import (
|
||||
"github.com/rook/rook/pkg/operator/ceph/cluster/mgr"
|
||||
"github.com/rook/rook/pkg/operator/ceph/cluster/mon"
|
||||
opcontroller "github.com/rook/rook/pkg/operator/ceph/controller"
|
||||
cephver "github.com/rook/rook/pkg/operator/ceph/version"
|
||||
"github.com/rook/rook/pkg/operator/k8sutil"
|
||||
"github.com/rook/rook/pkg/util/flags"
|
||||
"github.com/spf13/cobra"
|
||||
@@ -40,7 +41,7 @@ var (
|
||||
updateMgrServicesInterval string
|
||||
daemonName string
|
||||
clusterSpec cephv1.ClusterSpec
|
||||
mgrStatSupported bool
|
||||
rawCephVersion string
|
||||
)
|
||||
|
||||
func init() {
|
||||
@@ -55,7 +56,7 @@ func init() {
|
||||
mgrSidecarCmd.Flags().StringVar(&ownerRefID, "cluster-id", "", "the UID of the cluster CR that owns this cluster")
|
||||
mgrSidecarCmd.Flags().StringVar(&clusterName, "cluster-name", "", "the name of the cluster CR that owns this cluster")
|
||||
mgrSidecarCmd.Flags().StringVar(&daemonName, "daemon-name", "", "the name of the local mgr daemon")
|
||||
mgrSidecarCmd.Flags().BoolVar(&mgrStatSupported, "mgr-stat-supported", false, "whether the version of ceph supports mgr stat")
|
||||
mgrSidecarCmd.Flags().StringVar(&rawCephVersion, "ceph-version", "", "the version of ceph")
|
||||
|
||||
flags.SetFlagsFromEnv(mgrCmd.Flags(), rook.RookEnvVarPrefix)
|
||||
flags.SetFlagsFromEnv(mgrSidecarCmd.Flags(), rook.RookEnvVarPrefix)
|
||||
@@ -82,9 +83,15 @@ func runMgrSidecar(cmd *cobra.Command, args []string) error {
|
||||
rook.TerminateFatal(err)
|
||||
}
|
||||
|
||||
version, err := cephver.ExtractCephVersion(rawCephVersion)
|
||||
if err != nil {
|
||||
rook.TerminateFatal(err)
|
||||
}
|
||||
clusterInfo.CephVersion = *version
|
||||
|
||||
m := mgr.New(context, &clusterInfo, clusterSpec, "")
|
||||
for {
|
||||
err := m.ReconcileMultipleServices(daemonName, mgrStatSupported)
|
||||
err := m.ReconcileActiveMgrServices(daemonName)
|
||||
if err != nil {
|
||||
logger.Errorf("failed to reconcile services. %v", err)
|
||||
} else {
|
||||
|
||||
@@ -181,10 +181,20 @@ func (c *Cluster) Start() error {
|
||||
// check if any extra mgrs need to be removed
|
||||
c.removeExtraMgrs(daemonIDs)
|
||||
|
||||
if len(daemonIDs) == 1 {
|
||||
// Only reconcile the mgr services in the operator if there is a single mgr.
|
||||
// If there are multiple mgrs they will be managed by the mgr sidecar.
|
||||
if err := c.reconcileService(daemonIDs[0]); err != nil {
|
||||
activeMgr := daemonIDs[0]
|
||||
if len(daemonIDs) > 1 {
|
||||
// When multiple mgrs are running, the mgr sidecar for the active mgr
|
||||
// will create the services. However, the sidecar will only reconcile all
|
||||
// the services when the active mgr changes. Here as part of the regular reconcile
|
||||
// we trigger reconciling all the services to ensure they are current.
|
||||
activeMgr, err = c.getActiveMgr()
|
||||
if err != nil || activeMgr == "" {
|
||||
activeMgr = ""
|
||||
logger.Infof("cannot reconcile mgr services, no active mgr found. err=%v", err)
|
||||
}
|
||||
}
|
||||
if activeMgr != "" {
|
||||
if err := c.reconcileServices(activeMgr); err != nil {
|
||||
return errors.Wrap(err, "failed to enable mgr services")
|
||||
}
|
||||
}
|
||||
@@ -223,9 +233,9 @@ func (c *Cluster) removeExtraMgrs(daemonIDs []string) {
|
||||
}
|
||||
}
|
||||
|
||||
// ReconcileMultipleServices reconciles the services if the active mgr is the one running
|
||||
// ReconcileActiveMgrServices reconciles the services if the active mgr is the one running
|
||||
// in the sidecar
|
||||
func (c *Cluster) ReconcileMultipleServices(daemonNameToUpdate string, mgrStatSupported bool) error {
|
||||
func (c *Cluster) ReconcileActiveMgrServices(daemonNameToUpdate string) error {
|
||||
// If the services are already set to this daemon, no need to attempt to update
|
||||
svc, err := c.context.Clientset.CoreV1().Services(c.clusterInfo.Namespace).Get(context.TODO(), AppName, metav1.GetOptions{})
|
||||
if err != nil {
|
||||
@@ -239,23 +249,9 @@ func (c *Cluster) ReconcileMultipleServices(daemonNameToUpdate string, mgrStatSu
|
||||
logger.Infof("mgr service currently set to %q, checking if need to update to %q", currentDaemon, daemonNameToUpdate)
|
||||
}
|
||||
|
||||
var activeName string
|
||||
// mgrStatSupported is a flag instead of calling c.clusterInfo.CephVersion.IsAtLeastPacific() directly since we're inside the sidecar
|
||||
// where the cephVersion is not directly available
|
||||
if mgrStatSupported {
|
||||
// The preferred way to query the active mgr is "ceph mgr stat", which is only available in pacific
|
||||
mgrStat, err := cephclient.CephMgrStat(c.context, c.clusterInfo)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "failed to get mgr stat for the active mgr")
|
||||
}
|
||||
activeName = mgrStat.ActiveName
|
||||
} else {
|
||||
// The legacy way to query the active mgr is with the verbose "ceph mgr dump"
|
||||
mgrMap, err := cephclient.CephMgrMap(c.context, c.clusterInfo)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "failed to get mgr map for the active mgr")
|
||||
}
|
||||
activeName = mgrMap.ActiveName
|
||||
activeName, err := c.getActiveMgr()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if activeName == "" {
|
||||
return errors.New("active mgr not found")
|
||||
@@ -265,11 +261,30 @@ func (c *Cluster) ReconcileMultipleServices(daemonNameToUpdate string, mgrStatSu
|
||||
return nil
|
||||
}
|
||||
|
||||
return c.reconcileService(activeName)
|
||||
return c.reconcileServices(activeName)
|
||||
}
|
||||
|
||||
func (c *Cluster) getActiveMgr() (string, error) {
|
||||
// The preferred way to query the active mgr is "ceph mgr stat", which is only available in pacific or newer
|
||||
if c.clusterInfo.CephVersion.IsAtLeastPacific() {
|
||||
mgrStat, err := cephclient.CephMgrStat(c.context, c.clusterInfo)
|
||||
if err != nil {
|
||||
return "", errors.Wrap(err, "failed to get mgr stat for the active mgr")
|
||||
}
|
||||
return mgrStat.ActiveName, nil
|
||||
}
|
||||
|
||||
// The legacy way to query the active mgr is with the verbose "ceph mgr dump"
|
||||
mgrMap, err := cephclient.CephMgrMap(c.context, c.clusterInfo)
|
||||
if err != nil {
|
||||
return "", errors.Wrap(err, "failed to get mgr map for the active mgr")
|
||||
}
|
||||
|
||||
return mgrMap.ActiveName, nil
|
||||
}
|
||||
|
||||
// reconcile the services, if the active mgr is not detected, use the default mgr
|
||||
func (c *Cluster) reconcileService(activeDaemon string) error {
|
||||
func (c *Cluster) reconcileServices(activeDaemon string) error {
|
||||
logger.Infof("setting services to point to mgr %q", activeDaemon)
|
||||
|
||||
if err := c.configureDashboardService(activeDaemon); err != nil {
|
||||
|
||||
@@ -46,6 +46,10 @@ func TestStartMgr(t *testing.T) {
|
||||
|
||||
executor := &exectest.MockExecutor{
|
||||
MockExecuteCommandWithOutputFile: func(command string, outFileArg string, args ...string) (string, error) {
|
||||
logger.Infof("Execute: %s %v", command, args)
|
||||
if args[0] == "mgr" && args[1] == "stat" {
|
||||
return `{"active_name": "a"}`, nil
|
||||
}
|
||||
return "{\"key\":\"mysecurekey\"}", nil
|
||||
},
|
||||
}
|
||||
@@ -63,7 +67,7 @@ func TestStartMgr(t *testing.T) {
|
||||
Clientset: clientset,
|
||||
RequestCancelOrchestration: abool.New()}
|
||||
ownerInfo := cephclient.NewMinimumOwnerInfo(t)
|
||||
clusterInfo := &cephclient.ClusterInfo{Namespace: "ns", FSID: "myfsid", OwnerInfo: ownerInfo}
|
||||
clusterInfo := &cephclient.ClusterInfo{Namespace: "ns", FSID: "myfsid", OwnerInfo: ownerInfo, CephVersion: cephver.CephVersion{Major: 16, Minor: 2, Build: 5}}
|
||||
clusterInfo.SetName("test")
|
||||
clusterSpec := cephv1.ClusterSpec{
|
||||
Annotations: map[rook.KeyType]rook.Annotations{cephv1.KeyMgr: {"my": "annotation"}},
|
||||
@@ -99,13 +103,8 @@ func TestStartMgr(t *testing.T) {
|
||||
assert.Nil(t, err)
|
||||
err = c.Start()
|
||||
assert.Nil(t, err)
|
||||
// trigger the sidecar reconcile since the operator didn't do it so we can perform the full validation
|
||||
err = c.reconcileService("a")
|
||||
assert.Nil(t, err)
|
||||
validateStart(t, c)
|
||||
|
||||
// the dashboard service is only deleted by the operator reconcile if the replicas are 1,
|
||||
// otherwise the sidecar has the responsibility
|
||||
c.spec.Mgr.Count = 1
|
||||
c.spec.Dashboard.Enabled = false
|
||||
// clean the previous deployments
|
||||
@@ -201,7 +200,8 @@ func TestMgrSidecarReconcile(t *testing.T) {
|
||||
c := &Cluster{spec: spec, context: ctx, clusterInfo: clusterInfo}
|
||||
|
||||
// Update services according to the active mgr
|
||||
err := c.ReconcileMultipleServices(activeMgr, false)
|
||||
clusterInfo.CephVersion = cephver.CephVersion{Major: 15, Minor: 2, Build: 0}
|
||||
err := c.ReconcileActiveMgrServices(activeMgr)
|
||||
assert.NoError(t, err)
|
||||
assert.False(t, calledMgrStat)
|
||||
assert.True(t, calledMgrDump)
|
||||
@@ -210,7 +210,8 @@ func TestMgrSidecarReconcile(t *testing.T) {
|
||||
|
||||
// nothing is created or updated when the requested mgr is not the active mgr
|
||||
calledMgrDump = false
|
||||
err = c.ReconcileMultipleServices("b", true)
|
||||
clusterInfo.CephVersion = cephver.CephVersion{Major: 16, Minor: 2, Build: 5}
|
||||
err = c.ReconcileActiveMgrServices("b")
|
||||
assert.NoError(t, err)
|
||||
assert.True(t, calledMgrStat)
|
||||
assert.False(t, calledMgrDump)
|
||||
@@ -219,7 +220,7 @@ func TestMgrSidecarReconcile(t *testing.T) {
|
||||
|
||||
// nothing is updated when the requested mgr is not the active mgr
|
||||
activeMgr = "b"
|
||||
err = c.ReconcileMultipleServices("b", true)
|
||||
err = c.ReconcileActiveMgrServices("b")
|
||||
assert.NoError(t, err)
|
||||
validateServices(t, c)
|
||||
validateServiceMatches(t, c, "b")
|
||||
|
||||
@@ -226,7 +226,7 @@ func (c *Cluster) makeMgrSidecarContainer(mgrConfig *mgrConfig) v1.Container {
|
||||
{Name: "ROOK_MONITORING_ENABLED", Value: strconv.FormatBool(c.spec.Monitoring.Enabled)},
|
||||
{Name: "ROOK_UPDATE_INTERVAL", Value: "15s"},
|
||||
{Name: "ROOK_DAEMON_NAME", Value: mgrConfig.DaemonID},
|
||||
{Name: "ROOK_MGR_STAT_SUPPORTED", Value: strconv.FormatBool(c.clusterInfo.CephVersion.IsAtLeastPacific())},
|
||||
{Name: "ROOK_CEPH_VERSION", Value: "ceph version " + c.clusterInfo.CephVersion.String()},
|
||||
}
|
||||
|
||||
return v1.Container{
|
||||
|
||||
@@ -44,9 +44,7 @@ func TestReleaseName(t *testing.T) {
|
||||
assert.Equal(t, unknownVersionString, ver.ReleaseName())
|
||||
}
|
||||
|
||||
func extractVersionHelper(t *testing.T, text string,
|
||||
major, minor, extra, build int,
|
||||
commitID string) {
|
||||
func extractVersionHelper(t *testing.T, text string, major, minor, extra, build int, commitID string) {
|
||||
v, err := ExtractCephVersion(text)
|
||||
if assert.NoError(t, err) {
|
||||
assert.Equal(t, *v, CephVersion{major, minor, extra, build, commitID})
|
||||
@@ -93,6 +91,13 @@ ceph version Development (no_version) nautilus (rc)
|
||||
v, err = ExtractCephVersion(v2d)
|
||||
assert.Error(t, err)
|
||||
assert.Nil(t, v)
|
||||
|
||||
// Test the round trip for serializing and deserializing the version
|
||||
v3c := "ceph version 16.2.5-1 pacific"
|
||||
v, err = ExtractCephVersion(v3c)
|
||||
assert.NoError(t, err)
|
||||
assert.NotNil(t, v)
|
||||
assert.Equal(t, "16.2.5-1 pacific", v.String())
|
||||
}
|
||||
|
||||
func TestSupported(t *testing.T) {
|
||||
|
||||
Reference in New Issue
Block a user