Merge pull request #5869 from travisn/init-mon-spec

ceph: Require the cluster spec for initialization of the mons to avoid removing a healthy mon
This commit is contained in:
Travis Nielsen
2020-07-22 10:46:42 -06:00
committed by GitHub
9 changed files with 60 additions and 70 deletions
+1 -1
View File
@@ -83,7 +83,7 @@ func newCluster(c *cephv1.CephCluster, context *clusterd.Context, csiMutex *sync
monitoringChannels: make(map[string]*clusterHealth),
stopCh: make(chan struct{}),
ownerRef: *ownerRef,
mons: mon.New(context, c.Namespace, c.Spec.DataDirHostPath, c.Spec.Network, *ownerRef, csiMutex),
mons: mon.New(context, c.Namespace, c.Spec, *ownerRef, csiMutex),
}
}
+13 -16
View File
@@ -22,7 +22,6 @@ import (
"time"
"github.com/pkg/errors"
cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
"github.com/rook/rook/pkg/daemon/ceph/client"
cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
cephutil "github.com/rook/rook/pkg/daemon/ceph/util"
@@ -41,27 +40,25 @@ var (
// HealthChecker aggregates the mon/cluster info needed to check the health of the monitors
type HealthChecker struct {
monCluster *Cluster
clusterSpec *cephv1.ClusterSpec
interval time.Duration
monCluster *Cluster
interval time.Duration
}
// NewHealthChecker creates a new HealthChecker object
func NewHealthChecker(monCluster *Cluster, clusterSpec *cephv1.ClusterSpec) *HealthChecker {
func NewHealthChecker(monCluster *Cluster) *HealthChecker {
h := &HealthChecker{
monCluster: monCluster,
clusterSpec: clusterSpec,
interval: HealthCheckInterval,
monCluster: monCluster,
interval: HealthCheckInterval,
}
monCRDTimeoutSetting := clusterSpec.HealthCheck.DaemonHealth.Monitor.Timeout
monCRDTimeoutSetting := monCluster.spec.HealthCheck.DaemonHealth.Monitor.Timeout
if monCRDTimeoutSetting != "" {
if monTimeout, err := time.ParseDuration(monCRDTimeoutSetting); err == nil {
MonOutTimeout = monTimeout
}
}
checkInterval := clusterSpec.HealthCheck.DaemonHealth.Monitor.Interval
checkInterval := monCluster.spec.HealthCheck.DaemonHealth.Monitor.Interval
// allow overriding the check interval
if checkInterval != "" {
if duration, err := time.ParseDuration(checkInterval); err == nil {
@@ -75,11 +72,6 @@ func NewHealthChecker(monCluster *Cluster, clusterSpec *cephv1.ClusterSpec) *Hea
// Check periodically checks the health of the monitors
func (hc *HealthChecker) Check(stopCh chan struct{}) {
// Populate spec with clusterSpec
if hc.clusterSpec.External.Enable {
hc.monCluster.spec = *hc.clusterSpec
}
for {
select {
case <-stopCh:
@@ -100,6 +92,11 @@ func (c *Cluster) checkHealth() error {
c.acquireOrchestrationLock()
defer c.releaseOrchestrationLock()
if c.spec.Mon.Count == 0 || !c.ClusterInfo.IsInitialized(true) {
logger.Warningf("skipping mon health check since cluster details are not initialized")
return nil
}
logger.Debugf("Checking health for mons in cluster. %s", c.ClusterInfo.Namespace)
// For an external connection we use a special function to get the status
@@ -257,7 +254,7 @@ func (c *Cluster) failoverMon(name string) error {
return errors.Wrap(err, "failed to place new mon on a node")
}
if c.Network.IsHost() {
if c.spec.Network.IsHost() {
node, ok := c.mapping.Node[m.DaemonName]
if !ok {
return errors.Errorf("mon %s doesn't exist in assignment map", m.DaemonName)
+9 -11
View File
@@ -59,7 +59,7 @@ func TestCheckHealth(t *testing.T) {
ConfigDir: configDir,
Executor: executor,
}
c := New(context, "ns", "", cephv1.NetworkSpec{}, metav1.OwnerReference{}, &sync.Mutex{})
c := New(context, "ns", cephv1.ClusterSpec{}, metav1.OwnerReference{}, &sync.Mutex{})
setCommonMonProperties(c, 1, cephv1.MonSpec{Count: 3, AllowMultiplePerNode: true}, "myversion")
logger.Infof("initial mons: %v", c.ClusterInfo.Monitors)
c.waitForStart = false
@@ -114,7 +114,7 @@ func TestCheckHealthNotFound(t *testing.T) {
ConfigDir: configDir,
Executor: executor,
}
c := New(context, "ns", "", cephv1.NetworkSpec{}, metav1.OwnerReference{}, &sync.Mutex{})
c := New(context, "ns", cephv1.ClusterSpec{}, metav1.OwnerReference{}, &sync.Mutex{})
setCommonMonProperties(c, 2, cephv1.MonSpec{Count: 3, AllowMultiplePerNode: true}, "myversion")
c.waitForStart = false
defer os.RemoveAll(c.context.ConfigDir)
@@ -170,7 +170,7 @@ func TestAddRemoveMons(t *testing.T) {
ConfigDir: configDir,
Executor: executor,
}
c := New(context, "ns", "", cephv1.NetworkSpec{}, metav1.OwnerReference{}, &sync.Mutex{})
c := New(context, "ns", cephv1.ClusterSpec{}, metav1.OwnerReference{}, &sync.Mutex{})
setCommonMonProperties(c, 0, cephv1.MonSpec{Count: 5, AllowMultiplePerNode: true}, "myversion")
c.maxMonID = 0 // "a" is max mon id
c.waitForStart = false
@@ -365,26 +365,24 @@ func createTestMonPod(t *testing.T, clientset kubernetes.Interface, namespace, n
}
func TestNewHealthChecker(t *testing.T) {
c := &Cluster{}
clusterSpec := &cephv1.ClusterSpec{HealthCheck: cephv1.CephClusterHealthCheckSpec{}}
c := &Cluster{spec: cephv1.ClusterSpec{HealthCheck: cephv1.CephClusterHealthCheckSpec{}}}
time10s, _ := time.ParseDuration("10s")
clusterSpec10s := &cephv1.ClusterSpec{HealthCheck: cephv1.CephClusterHealthCheckSpec{DaemonHealth: cephv1.DaemonHealthSpec{Monitor: cephv1.HealthCheckSpec{Interval: "10s"}}}}
c10s := &Cluster{spec: cephv1.ClusterSpec{HealthCheck: cephv1.CephClusterHealthCheckSpec{DaemonHealth: cephv1.DaemonHealthSpec{Monitor: cephv1.HealthCheckSpec{Interval: "10s"}}}}}
type args struct {
monCluster *Cluster
clusterSpec *cephv1.ClusterSpec
monCluster *Cluster
}
tests := []struct {
name string
args args
want *HealthChecker
}{
{"default-interval", args{c, clusterSpec}, &HealthChecker{c, clusterSpec, HealthCheckInterval}},
{"10s-interval", args{c, clusterSpec10s}, &HealthChecker{c, clusterSpec10s, time10s}},
{"default-interval", args{c}, &HealthChecker{c, HealthCheckInterval}},
{"10s-interval", args{c10s}, &HealthChecker{c10s, time10s}},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
if got := NewHealthChecker(tt.args.monCluster, tt.args.clusterSpec); !reflect.DeepEqual(got, tt.want) {
if got := NewHealthChecker(tt.args.monCluster); !reflect.DeepEqual(got, tt.want) {
t.Errorf("NewHealthChecker() = %v, want %v", got, tt.want)
}
})
+9 -12
View File
@@ -119,10 +119,8 @@ type Cluster struct {
rookVersion string
orchestrationMutex sync.Mutex
Port int32
Network cephv1.NetworkSpec
maxMonID int
waitForStart bool
dataDirHostPath string
monPodRetryInterval time.Duration
monPodTimeout time.Duration
monTimeoutList map[string]time.Time
@@ -166,17 +164,16 @@ type SchedulingResult struct {
}
// New creates an instance of a mon cluster
func New(context *clusterd.Context, namespace, dataDirHostPath string, network cephv1.NetworkSpec, ownerRef metav1.OwnerReference, csiConfigMutex *sync.Mutex) *Cluster {
func New(context *clusterd.Context, namespace string, spec cephv1.ClusterSpec, ownerRef metav1.OwnerReference, csiConfigMutex *sync.Mutex) *Cluster {
return &Cluster{
context: context,
dataDirHostPath: dataDirHostPath,
spec: spec,
Namespace: namespace,
maxMonID: -1,
waitForStart: true,
monPodRetryInterval: 6 * time.Second,
monPodTimeout: 5 * time.Minute,
monTimeoutList: map[string]time.Time{},
Network: network,
mapping: &Mapping{
Node: map[string]*NodeInfo{},
},
@@ -198,8 +195,8 @@ func (c *Cluster) Start(clusterInfo *cephclient.ClusterInfo, rookVersion string,
c.spec = spec
// fail if we were instructed to deploy more than one mon on the same machine with host networking
if c.Network.IsHost() && c.spec.Mon.AllowMultiplePerNode && c.spec.Mon.Count > 1 {
return nil, errors.Errorf("refusing to deploy %d monitors on the same host since hostNetwork is %+v and allowMultiplePerNode is %t. only one monitor per node is allowed", c.spec.Mon.Count, c.Network, c.spec.Mon.AllowMultiplePerNode)
if c.spec.Network.IsHost() && c.spec.Mon.AllowMultiplePerNode && c.spec.Mon.Count > 1 {
return nil, errors.Errorf("refusing to deploy %d monitors on the same host with host networking and allowMultiplePerNode is %t. only one monitor per node is allowed", c.spec.Mon.Count, c.spec.Mon.AllowMultiplePerNode)
}
// Validate pod's memory if specified
@@ -375,7 +372,7 @@ func (c *Cluster) initMonConfig(size int) (int, []*monConfig) {
DaemonName: monitor.Name,
Port: cephutil.GetPortFromEndpoint(monitor.Endpoint),
DataPathMap: config.NewStatefulDaemonDataPathMap(
c.dataDirHostPath, dataDirRelativeHostPath(monitor.Name), config.MonType, monitor.Name, c.Namespace),
c.spec.DataDirHostPath, dataDirRelativeHostPath(monitor.Name), config.MonType, monitor.Name, c.Namespace),
})
}
@@ -397,7 +394,7 @@ func (c *Cluster) newMonConfig(monID int) *monConfig {
DaemonName: daemonName,
Port: DefaultMsgr1Port,
DataPathMap: config.NewStatefulDaemonDataPathMap(
c.dataDirHostPath, dataDirRelativeHostPath(daemonName), config.MonType, daemonName, c.Namespace),
c.spec.DataDirHostPath, dataDirRelativeHostPath(daemonName), config.MonType, daemonName, c.Namespace),
}
}
@@ -544,7 +541,7 @@ func realScheduleMonitor(c *Cluster, mon *monConfig) (SchedulingResult, error) {
func (c *Cluster) initMonIPs(mons []*monConfig) error {
for _, m := range mons {
if c.Network.IsHost() {
if c.spec.Network.IsHost() {
logger.Infof("setting mon endpoints for hostnetwork mode")
node, ok := c.mapping.Node[m.DaemonName]
if !ok {
@@ -630,7 +627,7 @@ func (c *Cluster) assignMons(mons []*monConfig) error {
// placement is not being made. otherwise, the node choice will map
// directly to a node selector on the monitor pod.
var nodeInfo *NodeInfo = nil
if c.Network.IsHost() || c.spec.Mon.VolumeClaimTemplate == nil {
if c.spec.Network.IsHost() || c.spec.Mon.VolumeClaimTemplate == nil {
logger.Infof("assignmon: mon %s assigned to node %s", mon.DaemonName, nodeChoice.Name)
nodeInfo, err = getNodeInfoFromNode(*nodeChoice)
if err != nil {
@@ -897,7 +894,7 @@ func (c *Cluster) startMon(m *monConfig, node *NodeInfo) error {
// isn't using host networking and the deployment is using pvc storage,
// then the node selector can be removed. this may happen after
// upgrading the cluster with the k8s scheduling support for monitors.
if c.Network.IsHost() || !pvcExists {
if c.spec.Network.IsHost() || !pvcExists {
p.PodAffinity = nil
p.PodAntiAffinity = nil
k8sutil.SetNodeAntiAffinityForPod(&d.Spec.Template.Spec, p, requiredDuringScheduling(&c.spec), PreferredDuringScheduling,
+8 -8
View File
@@ -101,10 +101,9 @@ func newTestStartClusterWithQuorumResponse(t *testing.T, namespace string, monRe
}, nil
}
func newCluster(context *clusterd.Context, namespace string, network cephv1.NetworkSpec, allowMultiplePerNode bool, resources v1.ResourceRequirements) *Cluster {
func newCluster(context *clusterd.Context, namespace string, allowMultiplePerNode bool, resources v1.ResourceRequirements) *Cluster {
return &Cluster{
ClusterInfo: nil,
Network: network,
context: context,
Namespace: namespace,
rookVersion: "myversion",
@@ -146,7 +145,7 @@ func TestStartMonPods(t *testing.T) {
namespace := "ns"
context, err := newTestStartCluster(t, namespace)
assert.Nil(t, err)
c := newCluster(context, namespace, cephv1.NetworkSpec{}, true, v1.ResourceRequirements{})
c := newCluster(context, namespace, true, v1.ResourceRequirements{})
c.ClusterInfo = clienttest.CreateTestClusterInfo(1)
// start a basic cluster
@@ -167,7 +166,7 @@ func TestOperatorRestart(t *testing.T) {
namespace := "ns"
context, err := newTestStartCluster(t, namespace)
assert.Nil(t, err)
c := newCluster(context, namespace, cephv1.NetworkSpec{}, true, v1.ResourceRequirements{})
c := newCluster(context, namespace, true, v1.ResourceRequirements{})
c.ClusterInfo = clienttest.CreateTestClusterInfo(1)
// start a basic cluster
@@ -177,7 +176,7 @@ func TestOperatorRestart(t *testing.T) {
validateStart(t, c)
c = newCluster(context, namespace, cephv1.NetworkSpec{}, true, v1.ResourceRequirements{})
c = newCluster(context, namespace, true, v1.ResourceRequirements{})
c.ClusterInfo = clienttest.CreateTestClusterInfo(1)
// starting again should be a no-op, but will not result in an error
@@ -196,7 +195,7 @@ func TestOperatorRestartHostNetwork(t *testing.T) {
assert.Nil(t, err)
// cluster without host networking
c := newCluster(context, namespace, cephv1.NetworkSpec{}, false, v1.ResourceRequirements{})
c := newCluster(context, namespace, false, v1.ResourceRequirements{})
c.ClusterInfo = clienttest.CreateTestClusterInfo(1)
// start a basic cluster
@@ -207,7 +206,8 @@ func TestOperatorRestartHostNetwork(t *testing.T) {
validateStart(t, c)
// cluster with host networking
c = newCluster(context, namespace, cephv1.NetworkSpec{HostNetwork: true}, false, v1.ResourceRequirements{})
c = newCluster(context, namespace, false, v1.ResourceRequirements{})
c.spec.Network.HostNetwork = true
c.ClusterInfo = clienttest.CreateTestClusterInfo(1)
// starting again should be a no-op, but still results in an error
@@ -232,7 +232,7 @@ func TestSaveMonEndpoints(t *testing.T) {
clientset := test.New(t, 1)
configDir, _ := ioutil.TempDir("", "")
defer os.RemoveAll(configDir)
c := New(&clusterd.Context{Clientset: clientset, ConfigDir: configDir}, "ns", "", cephv1.NetworkSpec{}, metav1.OwnerReference{}, &sync.Mutex{})
c := New(&clusterd.Context{Clientset: clientset, ConfigDir: configDir}, "ns", cephv1.ClusterSpec{}, metav1.OwnerReference{}, &sync.Mutex{})
setCommonMonProperties(c, 1, cephv1.MonSpec{Count: 3, AllowMultiplePerNode: true}, "myversion")
// create the initial config map
+10 -9
View File
@@ -35,7 +35,7 @@ import (
func TestNodeAffinity(t *testing.T) {
clientset := test.New(t, 4)
c := New(&clusterd.Context{Clientset: clientset}, "ns", "", cephv1.NetworkSpec{}, metav1.OwnerReference{}, &sync.Mutex{})
c := New(&clusterd.Context{Clientset: clientset}, "ns", cephv1.ClusterSpec{}, metav1.OwnerReference{}, &sync.Mutex{})
setCommonMonProperties(c, 0, cephv1.MonSpec{Count: 3, AllowMultiplePerNode: true}, "myversion")
c.spec.Placement = map[rookv1.KeyType]rookv1.Placement{}
@@ -76,7 +76,8 @@ func TestHostNetworkSameNode(t *testing.T) {
context, err := newTestStartCluster(t, namespace)
assert.Nil(t, err)
// cluster host networking
c := newCluster(context, namespace, cephv1.NetworkSpec{HostNetwork: true}, true, v1.ResourceRequirements{})
c := newCluster(context, namespace, true, v1.ResourceRequirements{})
c.spec.Network.HostNetwork = true
c.ClusterInfo = clienttest.CreateTestClusterInfo(1)
// start a basic cluster
@@ -95,7 +96,7 @@ func TestPodMemory(t *testing.T) {
},
}
c := newCluster(context, namespace, cephv1.NetworkSpec{}, true, r)
c := newCluster(context, namespace, true, r)
c.ClusterInfo = clienttest.CreateTestClusterInfo(1)
// start a basic cluster
_, err = c.Start(c.ClusterInfo, c.rookVersion, cephver.Nautilus, c.spec)
@@ -111,7 +112,7 @@ func TestPodMemory(t *testing.T) {
},
}
c = newCluster(context, namespace, cephv1.NetworkSpec{}, true, r)
c = newCluster(context, namespace, true, r)
c.ClusterInfo = clienttest.CreateTestClusterInfo(1)
// start a basic cluster
_, err = c.Start(c.ClusterInfo, c.rookVersion, cephver.Nautilus, c.spec)
@@ -127,7 +128,7 @@ func TestPodMemory(t *testing.T) {
},
}
c = newCluster(context, namespace, cephv1.NetworkSpec{}, true, r)
c = newCluster(context, namespace, true, r)
c.ClusterInfo = clienttest.CreateTestClusterInfo(1)
// start a basic cluster
_, err = c.Start(c.ClusterInfo, c.rookVersion, cephver.Nautilus, c.spec)
@@ -143,7 +144,7 @@ func TestPodMemory(t *testing.T) {
},
}
c = newCluster(context, namespace, cephv1.NetworkSpec{}, true, r)
c = newCluster(context, namespace, true, r)
c.ClusterInfo = clienttest.CreateTestClusterInfo(1)
// start a basic cluster
_, err = c.Start(c.ClusterInfo, c.rookVersion, cephver.Nautilus, c.spec)
@@ -151,7 +152,7 @@ func TestPodMemory(t *testing.T) {
// Test no resources were specified on the pod
r = v1.ResourceRequirements{}
c = newCluster(context, namespace, cephv1.NetworkSpec{}, true, r)
c = newCluster(context, namespace, true, r)
c.ClusterInfo = clienttest.CreateTestClusterInfo(1)
// start a basic cluster
_, err = c.Start(c.ClusterInfo, c.rookVersion, cephver.Nautilus, c.spec)
@@ -161,10 +162,10 @@ func TestPodMemory(t *testing.T) {
func TestHostNetwork(t *testing.T) {
clientset := test.New(t, 3)
c := New(&clusterd.Context{Clientset: clientset}, "ns", "", cephv1.NetworkSpec{}, metav1.OwnerReference{}, &sync.Mutex{})
c := New(&clusterd.Context{Clientset: clientset}, "ns", cephv1.ClusterSpec{}, metav1.OwnerReference{}, &sync.Mutex{})
setCommonMonProperties(c, 0, cephv1.MonSpec{Count: 3, AllowMultiplePerNode: true}, "myversion")
c.Network.HostNetwork = true
c.spec.Network.HostNetwork = true
monConfig := testGenMonConfig("c")
pod := c.makeMonPod(monConfig, false, "")
+6 -6
View File
@@ -151,7 +151,7 @@ func (c *Cluster) makeMonPod(monConfig *monConfig, canary bool, PVCName string)
// we decide later whether to use a PVC volume or host volumes for mons, so only populate
// the base volumes at this point.
Volumes: controller.DaemonVolumesBase(monConfig.DataPathMap, keyringStoreName),
HostNetwork: c.Network.IsHost(),
HostNetwork: c.spec.Network.IsHost(),
PriorityClassName: cephv1.GetMonPriorityClassName(c.spec.PriorityClassNames),
}
@@ -170,10 +170,10 @@ func (c *Cluster) makeMonPod(monConfig *monConfig, canary bool, PVCName string)
}
cephv1.GetMonAnnotations(c.spec.Annotations).ApplyToObjectMeta(&pod.ObjectMeta)
if c.Network.IsHost() {
if c.spec.Network.IsHost() {
pod.Spec.DNSPolicy = v1.DNSClusterFirstWithHostNet
} else if c.Network.NetworkSpec.IsMultus() {
k8sutil.ApplyMultus(c.Network.NetworkSpec, &pod.ObjectMeta)
} else if c.spec.Network.NetworkSpec.IsMultus() {
k8sutil.ApplyMultus(c.spec.Network.NetworkSpec, &pod.ObjectMeta)
}
return pod
@@ -235,7 +235,7 @@ func (c *Cluster) makeMonDaemonContainer(monConfig *monConfig) v1.Container {
// Handle the non-default port for host networking. If host networking is not being used,
// the service created elsewhere will handle the non-default port redirection to the default port inside the container.
if c.Network.IsHost() && monConfig.Port != DefaultMsgr1Port {
if c.spec.Network.IsHost() && monConfig.Port != DefaultMsgr1Port {
logger.Warningf("Starting mon %s with host networking on a non-default port %d. The mon must be failed over before enabling msgr2.",
monConfig.DaemonName, monConfig.Port)
publicAddr = fmt.Sprintf("%s:%d", publicAddr, monConfig.Port)
@@ -282,7 +282,7 @@ func (c *Cluster) makeMonDaemonContainer(monConfig *monConfig) v1.Container {
container = config.ConfigureLivenessProbe(cephv1.KeyMon, container, c.spec.HealthCheck)
// If host networking is enabled, we don't need a bind addr that is different from the public addr
if !c.Network.IsHost() {
if !c.spec.Network.IsHost() {
// Opposite of the above, --public-bind-addr will *not* still advertise on the previous
// port, which makes sense because this is the pod IP, which changes with every new pod.
container.Args = append(container.Args,
+3 -6
View File
@@ -45,8 +45,7 @@ func testPodSpec(t *testing.T, monID string, pvc bool) {
c := New(
&clusterd.Context{Clientset: clientset, ConfigDir: "/var/lib/rook"},
"ns",
"/var/lib/rook",
cephv1.NetworkSpec{},
cephv1.ClusterSpec{},
metav1.OwnerReference{},
&sync.Mutex{},
)
@@ -94,8 +93,7 @@ func TestDeploymentPVCSpec(t *testing.T) {
c := New(
&clusterd.Context{Clientset: clientset, ConfigDir: "/var/lib/rook"},
"ns",
"/var/lib/rook",
cephv1.NetworkSpec{},
cephv1.ClusterSpec{},
metav1.OwnerReference{},
&sync.Mutex{},
)
@@ -155,8 +153,7 @@ func testRequiredDuringScheduling(t *testing.T, hostNetwork, allowMultiplePerNod
c := New(
&clusterd.Context{},
"ns",
"/var/lib/rook",
cephv1.NetworkSpec{},
cephv1.ClusterSpec{},
metav1.OwnerReference{},
&sync.Mutex{},
)
+1 -1
View File
@@ -112,7 +112,7 @@ func isMonitoringDisabled(daemon string, clusterSpec *cephv1.ClusterSpec) bool {
func (c *ClusterController) startMonitoringCheck(cluster *cluster, clusterInfo *cephclient.ClusterInfo, daemon string) {
switch daemon {
case "mon":
healthChecker := mon.NewHealthChecker(cluster.mons, cluster.Spec)
healthChecker := mon.NewHealthChecker(cluster.mons)
logger.Infof("enabling ceph %s monitoring goroutine for cluster %q", daemon, cluster.Namespace)
go healthChecker.Check(cluster.monitoringChannels[daemon].stopChan)