Merge pull request #3346 from leseb/external-ceph

External Ceph cluster
This commit is contained in:
Sébastien Han authored and GitHub committed 2019-08-22 20:22:40 +02:00
commit 3eff79fd94
47 files changed
+1213 -295

No files matched your search

+62
View File
@@ -87,6 +87,8 @@ Settings can be specified at the global level to apply to the cluster as a whole
### Cluster Settings
- `external`:
- `enable`: if `true`, the cluster will not be managed by Rook but via an external entity. This mode is intended to connect to an existing cluster. In this case, Rook will only consume the external cluster. However, Rook will be able to deploy various daemons in Kubernetes such as object gateways, mds and nfs. If this setting is enabled **all** the other options will be ignored except `cephVersion.image` and `dataDirHostPath`. See [external cluster configuration](#external-cluster).
- `cephVersion`: The version information for launching the ceph daemons.
- `image`: The image used for running the ceph daemons. For example, `ceph/ceph:v13.2.6-20190604` or `ceph/ceph:v14.2.2-20190722`.
For the latest ceph images, see the [Ceph DockerHub](https://hub.docker.com/r/ceph/ceph/tags/).
@@ -646,3 +648,63 @@ spec:
accessModes:
- ReadWriteOnce
```
### External cluster
#### Pre-requisites
In order to configure an external Ceph cluster with Rook, we need to inject some information in order to connect to that cluster.
You can use the `cluster/examples/kubernetes/ceph/import-external-cluster.sh` script to achieve that.
The script will look for the following populated environment variables:
* `NAMESPACE`: the namespace where the configmap and secrets should be injected
* `ROOK_EXTERNAL_FSID`: the fsid of the external Ceph cluster, it can be retrieved via the `ceph fsid` command
* `ROOK_EXTERNAL_ADMIN_SECRET`: the external Ceph cluster admin secret key, it can be retrieved via the `ceph auth get-key client.admin` command
* `ROOK_EXTERNAL_CEPH_MON_DATA`: is a common-separated list of running monitors IP address along with their ports, e.g: `a=172.17.0.4:6789,b=172.17.0.5:6789,c=172.17.0.6:6789`. You don't need to specify all the monitors, you can simply pass one and the Operator will discover the rest. The name of the monitor is the name that appears in the `ceph status` ouput.
Example:
```bash
export NAMESPACE=rook-ceph-external
export ROOK_EXTERNAL_FSID=3240b4aa-ddbc-42ee-98ba-4ea7b2a61514
export ROOK_EXTERNAL_ADMIN_SECRET=AQC6Ylxdja+NDBAAB7qy9MEAr4VLLq4dCIvxtg==
export ROOK_EXTERNAL_CEPH_MON_DATA=a=172.17.0.4:6789
```
Then you can simply execute the script like this:
```bash
bash cluster/examples/kubernetes/ceph/import-external-cluster.sh
```
#### CR example
Assuming the above section has successfully completed, here is a CR example:
```
apiVersion: ceph.rook.io/v1
kind: CephCluster
metadata:
name: rook-ceph-external
namespace: rook-ceph-external
spec:
external:
enable: true
dataDirHostPath: /var/lib/rook
cephVersion:
image: ceph/ceph:v14.2.2-20190722 # the image version **must** match the version of the external Ceph cluster
```
Choose the namespace carefully, if you have an existing cluster managed by Rook, you have likely already injected `common.yaml`.
Additionnally, you now need to inject `common-external.yaml` too.
You can now create it like this:
```bash
kubectl create -f cluster/examples/kubernetes/ceph/cluster-external.yaml
```
If the previous section has not been completed, the Rook Operator will still acknowledge the CR creation but will wait forever to receive connection information.
**WARNING**
If no cluster is managed by the current Rook Operator, you need to inject `common.yaml`, then modify `cluster-external.yaml` and specify `rook-ceph` as `namespace`.
+1
View File
@@ -43,6 +43,7 @@ an example usage
- Ceph monitors have initial support for running on PVC storage. See docs on
[monitor settings for more detail](Documentation/ceph-cluster-crd.md#mon-settings).
- Ceph OSDs can be created by using StorageClassDeviceSet. See docs on [Storage Class Device Sets](Documentation/ceph-cluster-crd.md#storage-class-device-sets).
- Rook can now connect to an external cluster, for more info about external cluster [read the design](https://github.com/rook/rook/blob/master/design/ceph-external-cluster.md) as well as the documentation [Ceph External cluster](Documentation/ceph-cluster-crd.md#external-cluster)
## Breaking Changes
@@ -0,0 +1,23 @@
#################################################################################################################
# Define the settings for the rook-ceph-external cluster with common settings for a production cluster.
# For example, if Rook is not managing any existing cluster in the 'rook-ceph' namespace do:
# kubectl create -f common.yaml
# kubectl create -f operator.yaml
# kubectl create -f cluster-external.yaml
# If there is already a cluster managed by Rook in 'rook-ceph' then do:
# kubectl create -f common-external.yaml
# kubectl create -f cluster-external.yaml
#################################################################################################################
apiVersion: ceph.rook.io/v1
kind: CephCluster
metadata:
name: rook-ceph-external
namespace: rook-ceph-external
spec:
external:
enable: true
dataDirHostPath: /var/lib/rook
cephVersion:
image: ceph/ceph:v14.2.2-20190722 # MUST match external cluster version
@@ -0,0 +1,79 @@
###################################################################################################################
# Create the common resources that are necessary to start start an external Ceph cluster in a different namespace
# These resources can be created after an operator that is already running but assumes common.yaml has been injected
# The samples all assume that your existing operator running "rook-ceph" namespace will also watch and have permissions
# to interact with an external cluster configured in "rook-ceph-external" cluster.
#
# First common.yaml
# Then operator.yaml
# Then common-external.yaml
#
# If there is no cluster managed by the current Rook Operator
# you can simply replace all occurence of rook-ceph-external with rook-ceph
#
# And remove the following code:
#
# apiVersion: v1
# kind: Namespace
# metadata:
# name: rook-ceph-external
#
# Then kubectl create -f cluster-external.yaml
###################################################################################################################
apiVersion: v1
kind: Namespace
metadata:
name: rook-ceph-external
---
kind: RoleBinding
apiVersion: rbac.authorization.k8s.io/v1beta1
metadata:
name: rook-ceph-cluster-mgmt
namespace: rook-ceph-external
roleRef:
apiGroup: rbac.authorization.k8s.io
kind: ClusterRole
name: rook-ceph-cluster-mgmt
subjects:
- kind: ServiceAccount
name: rook-ceph-system
namespace: rook-ceph
---
kind: RoleBinding
apiVersion: rbac.authorization.k8s.io/v1beta1
metadata:
name: rook-ceph-cmd-reporter
namespace: rook-ceph-external
roleRef:
apiGroup: rbac.authorization.k8s.io
kind: Role
name: rook-ceph-cmd-reporter
subjects:
- kind: ServiceAccount
name: rook-ceph-cmd-reporter
namespace: rook-ceph-external
---
apiVersion: v1
kind: ServiceAccount
metadata:
name: rook-ceph-cmd-reporter
namespace: rook-ceph-external
---
kind: Role
apiVersion: rbac.authorization.k8s.io/v1beta1
metadata:
name: rook-ceph-cmd-reporter
namespace: rook-ceph-external
rules:
- apiGroups:
- ""
resources:
- pods
- configmaps
verbs:
- get
- list
- watch
- create
- update
- delete
+4 -6
View File
@@ -64,14 +64,12 @@ spec:
type: boolean
count:
maximum: 9
minimum: 1
minimum: 0
type: integer
preferredCount:
maximum: 9
minimum: 0
type: integer
required:
- count
network:
properties:
hostNetwork:
@@ -103,7 +101,7 @@ spec:
encryptedDevice:
type: string
pattern: ^(true|false)$
useAllDevices:
useAllDevices:
type: boolean
deviceFilter: {}
directories:
@@ -122,7 +120,7 @@ spec:
location: {}
resources: {}
type: array
useAllDevices:
useAllDevices:
type: boolean
deviceFilter: {}
location: {}
@@ -133,7 +131,7 @@ spec:
path:
type: string
config: {}
topologyAware:
topologyAware:
type: boolean
monitoring:
properties:
@@ -0,0 +1,68 @@
#!/bin/bash
set -e
##############
# VARIABLES #
#############
MON_SECRET_NAME=rook-ceph-mon
MON_SECRET_CLUSTER_NAME_KEYNAME=cluster-name
MON_SECRET_FSID_KEYNAME=fsid
MON_SECRET_ADMIN_KEYRING_KEYNAME=admin-secret
MON_SECRET_MON_KEYRING_KEYNAME=mon-secret
MON_ENDPOINT_CONFIGMAP_NAME=rook-ceph-mon-endpoints
ROOK_EXTERNAL_CLUSTER_NAME=$NAMESPACE
ROOK_EXTERNAL_MAX_MON_ID=2
ROOK_EXTERNAL_MAPPING={}
ROOK_EXTERNAL_MONITOR_SECRET=mon-secret
#############
# FUNCTIONS #
#############
function checkEnvVars() {
if [ -z "$NAMESPACE" ]; then
echo "Please populate the environment variable NAMESPACE"
exit 1
fi
if [ -z "$ROOK_EXTERNAL_FSID" ]; then
echo "Please populate the environment variable ROOK_EXTERNAL_FSID"
exit 1
fi
if [ -z "$ROOK_EXTERNAL_ADMIN_SECRET" ]; then
echo "Please populate the environment variable ROOK_EXTERNAL_ADMIN_SECRET"
exit 1
fi
if [ -z "$ROOK_EXTERNAL_CEPH_MON_DATA" ]; then
echo "Please populate the environment variable ROOK_EXTERNAL_CEPH_MON_DATA"
exit 1
fi
}
function importSecret() {
kubectl -n "$NAMESPACE" \
create \
secret \
generic \
"$MON_SECRET_NAME" \
--from-literal="$MON_SECRET_CLUSTER_NAME_KEYNAME"="$ROOK_EXTERNAL_CLUSTER_NAME" \
--from-literal="$MON_SECRET_FSID_KEYNAME"="$ROOK_EXTERNAL_FSID" \
--from-literal="$MON_SECRET_ADMIN_KEYRING_KEYNAME"="$ROOK_EXTERNAL_ADMIN_SECRET" \
--from-literal="$MON_SECRET_MON_KEYRING_KEYNAME"="$ROOK_EXTERNAL_MONITOR_SECRET"
}
function importConfigMap() {
kubectl -n "$NAMESPACE" \
create \
configmap \
"$MON_ENDPOINT_CONFIGMAP_NAME" \
--from-literal=data="$ROOK_EXTERNAL_CEPH_MON_DATA" \
--from-literal=mapping="$ROOK_EXTERNAL_MAPPING" \
--from-literal=maxMonId="$ROOK_EXTERNAL_MAX_MON_ID"
}
########
# MAIN #
########
checkEnvVars
importSecret
importConfigMap
@@ -42,7 +42,7 @@ spec:
# If the operator should only watch for cluster CRDs in the same namespace, set this to "true".
# If this is not set to true, the operator will watch for cluster CRDs in all namespaces.
- name: ROOK_CURRENT_NAMESPACE_ONLY
value: "true"
value: "false"
# To disable RBAC, uncomment the following:
# - name: RBAC_ENABLED
# value: "false"
+19 -8
View File
@@ -70,7 +70,7 @@ type ClusterSpec struct {
DataDirHostPath string `json:"dataDirHostPath,omitempty"`
// A spec for mon related options
Mon MonSpec `json:"mon"`
Mon MonSpec `json:"mon,omitempty"`
// A spec for rbd mirroring
RBDMirroring RBDMirroringSpec `json:"rbdMirroring"`
@@ -80,6 +80,10 @@ type ClusterSpec struct {
// Prometheus based Monitoring settings
Monitoring MonitoringSpec `json:"monitoring,omitempty"`
// Whether the Ceph Cluster is running external to this Kubernetes cluster
// mon, mgr, osd, mds, and discover daemons will not be created for external clusters.
External ExternalSpec `json:"external"`
}
// VersionSpec represents the settings for the Ceph version that Rook is orchestrating.
@@ -136,19 +140,26 @@ type CephHealthMessage struct {
type ClusterState string
const (
ClusterStateCreating ClusterState = "Creating"
ClusterStateCreated ClusterState = "Created"
ClusterStateUpdating ClusterState = "Updating"
ClusterStateError ClusterState = "Error"
ClusterStateCreating ClusterState = "Creating"
ClusterStateCreated ClusterState = "Created"
ClusterStateUpdating ClusterState = "Updating"
ClusterStateConnecting ClusterState = "Connecting"
ClusterStateConnected ClusterState = "Connected"
ClusterStateError ClusterState = "Error"
)
type MonSpec struct {
Count int `json:"count"`
PreferredCount int `json:"preferredCount"`
AllowMultiplePerNode bool `json:"allowMultiplePerNode"`
Count int `json:"count,omitempty"`
PreferredCount int `json:"preferredCount,omitempty"`
AllowMultiplePerNode bool `json:"allowMultiplePerNode,omitempty"`
VolumeClaimTemplate *v1.PersistentVolumeClaim `json:"volumeClaimTemplate,omitempty"`
}
// ExternalSpec represents the options supported by an external cluster
type ExternalSpec struct {
Enable bool `json:"enable"`
}
type RBDMirroringSpec struct {
Workers int `json:"workers"`
}
+14 -5
View File
@@ -42,11 +42,21 @@ type MonStatusRequest struct {
Args []string `json:"args"`
}
// represents an entry in the monitor map
// MonMapEntry represents an entry in the monitor map
type MonMapEntry struct {
Name string `json:"name"`
Rank int `json:"rank"`
Address string `json:"addr"`
Name string `json:"name"`
Rank int `json:"rank"`
Address string `json:"addr"`
PublicAddrs struct {
Addrvec []AddrvecEntry `json:"addrvec"`
} `json:"public_addrs"`
}
// AddrvecEntry represents an entry type for a given messenger version
type AddrvecEntry struct {
Type string `json:"type"`
Addr string `json:"addr"`
Nonce int `json:"nonce"`
}
// GetMonStatus calls mon_status mon_command
@@ -65,7 +75,6 @@ func GetMonStatus(context *clusterd.Context, clusterName string, debug bool) (Mo
return MonStatusResponse{}, fmt.Errorf("unmarshal failed: %+v. raw buffer response: %s", err, buf)
}
logger.Debugf("MON STATUS: %+v", resp)
return resp, nil
}
+15 -2
View File
@@ -43,7 +43,9 @@ const (
// DefaultKeyringFile is the default name of the file where Ceph stores its keyring info
DefaultKeyringFile = "keyring"
// Msgr2port is the listening port of the messenger v2 protocol
Msgr2port = 3300
Msgr2port = 3300
msgr1Prefix = "v1:"
msgr2Prefix = "v2:"
)
// GlobalConfig represents the [global] sections of Ceph's config file.
@@ -225,7 +227,8 @@ func CreateDefaultCephConfig(context *clusterd.Context, cluster *ClusterInfo, ru
} else if podName != "" && strings.Contains(podName, "operator") {
// This is an operator and its version is always based on Nautilus
// so it knows how to parse both msgr1 and msgr2 syntax
monHosts[i] = "v1:" + msgr1Endpoint
prefix := msgrPrefix(currentMonPort)
monHosts[i] = prefix + msgr1Endpoint
} else {
// This is not the operator, it's an OSD and its Ceph version is before Nautilus
monHosts[i] = msgr1Endpoint
@@ -339,3 +342,13 @@ func logLevelToCephLogLevel(logLevel capnslog.LogLevel) int {
return 0
}
func msgrPrefix(currentMonPort int32) string {
// Some installation might only be listening on v2, so let's set the prefix accordingly
if currentMonPort == Msgr2port {
return msgr2Prefix
}
return msgr1Prefix
}
+11
View File
@@ -178,3 +178,14 @@ func verifyConfigValue(t *testing.T, actualConf *ini.File, section, key, expecte
actualVal := k.Value()
assert.Equal(t, expectedVal, actualVal)
}
func TestOperatorEndpoint(t *testing.T) {
var fakeCurrentMonPort int32
fakeCurrentMonPort = 3300
prefix := msgrPrefix(fakeCurrentMonPort)
assert.Equal(t, "v2:", prefix)
fakeCurrentMonPort = 6789
prefix = msgrPrefix(fakeCurrentMonPort)
assert.Equal(t, "v1:", prefix)
}
+3
View File
@@ -53,6 +53,7 @@ type cluster struct {
context *clusterd.Context
Namespace string
Spec *cephv1.ClusterSpec
crdName string
mons *mon.Cluster
initCompleted bool
stopCh chan struct{}
@@ -79,6 +80,7 @@ func newCluster(c *cephv1.CephCluster, context *clusterd.Context, csiMutex *sync
Namespace: c.Namespace,
Spec: &c.Spec,
context: context,
crdName: c.Name,
stopCh: make(chan struct{}),
ownerRef: ownerRef,
mons: mon.New(context, c.Namespace, c.Spec.DataDirHostPath, c.Spec.Network.HostNetwork, ownerRef, csiMutex),
@@ -227,6 +229,7 @@ func (c *cluster) doOrchestration(rookImage string, cephVersion cephver.CephVers
return fmt.Errorf("failed to create override configmap %s. %+v", c.Namespace, err)
}
// This gets triggered on CR update so let's not run that (mon/mgr/osd daemons)
// Start the mon pods
clusterInfo, err := c.mons.Start(c.Info, rookImage, cephVersion, *c.Spec)
if err != nil {
+224 -41
View File
@@ -33,6 +33,7 @@ import (
discoverDaemon "github.com/rook/rook/pkg/daemon/discover"
"github.com/rook/rook/pkg/operator/ceph/cluster/mon"
"github.com/rook/rook/pkg/operator/ceph/cluster/osd"
"github.com/rook/rook/pkg/operator/ceph/config"
"github.com/rook/rook/pkg/operator/ceph/file"
"github.com/rook/rook/pkg/operator/ceph/nfs"
"github.com/rook/rook/pkg/operator/ceph/object"
@@ -42,11 +43,13 @@ import (
"github.com/rook/rook/pkg/operator/k8sutil"
v1 "k8s.io/api/core/v1"
apiextensionsv1beta1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1beta1"
"k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/types"
"k8s.io/apimachinery/pkg/util/wait"
"k8s.io/apimachinery/pkg/watch"
"k8s.io/client-go/kubernetes"
"k8s.io/client-go/tools/cache"
)
@@ -89,12 +92,12 @@ type ClusterController struct {
devicesInUse bool
rookImage string
clusterMap map[string]*cluster
addClusterCallback func() error
addClusterCallback func(bool) error
csiConfigMutex *sync.Mutex
}
// NewClusterController create controller for watching cluster custom resources created
func NewClusterController(context *clusterd.Context, rookImage string, volumeAttachment attachment.Attachment, addClusterCallback func() error) *ClusterController {
func NewClusterController(context *clusterd.Context, rookImage string, volumeAttachment attachment.Attachment, addClusterCallback func(bool) error) *ClusterController {
return &ClusterController{
context: context,
volumeAttachment: volumeAttachment,
@@ -217,47 +220,146 @@ func (c *ClusterController) onK8sNodeAdd(obj interface{}) {
}
func (c *ClusterController) onAdd(obj interface{}) {
// notify the callback that a cluster crd is being added
if c.addClusterCallback != nil {
if err := c.addClusterCallback(); err != nil {
logger.Errorf("%+v", err)
}
}
clusterObj, err := getClusterObject(obj)
if err != nil {
logger.Errorf("failed to get cluster object: %+v", err)
return
}
if existing, ok := c.clusterMap[clusterObj.Namespace]; ok {
logger.Errorf("Failed to add cluster cr %s in namespace %s. Cluster cr %s already exists in this namespace. Only one cluster cr per namespace is supported.",
clusterObj.Name, clusterObj.Namespace, existing.crdName)
return
}
cluster := newCluster(clusterObj, c.context, c.csiConfigMutex)
c.clusterMap[cluster.Namespace] = cluster
logger.Infof("starting cluster in namespace %s", cluster.Namespace)
// notify the callback that a cluster crd is being added
if c.addClusterCallback != nil {
if err := c.addClusterCallback(cluster.Spec.External.Enable); err != nil {
logger.Errorf("%+v", err)
}
}
c.initializeCluster(cluster, clusterObj)
}
func (c *ClusterController) configureExternalCephCluster(namespace, name string, cluster *cluster) error {
// Make sure the spec contains all the information we need
err := validateExternalClusterSpec(cluster)
if err != nil {
return fmt.Errorf("failed to validate external cluster specs. %+v", err)
}
c.updateClusterStatus(namespace, name, cephv1.ClusterStateConnecting, "")
// loop until we find the secret necessary to connect to the external cluster
for {
var err error
cluster.Info, _, _, err = mon.LoadClusterInfo(c.context, namespace)
if err != nil {
logger.Warningf("waiting for the connection info of the external cluster. %+v", err)
time.Sleep(10 * time.Second)
continue
} else {
break
}
}
logger.Infof("found the cluster info to connect to the external cluster. mons=%+v", cluster.Info.Monitors)
// Write the rook-ceph-config configmap (used by various daemons to apply config overrides)
// If we don't do this, daemons will never start, waiting forever for this configmap to be present
err = config.GetStore(cluster.context, namespace, &cluster.ownerRef).CreateOrUpdate(cluster.Info)
if err != nil {
return err
}
// Let's write connection info (ceph config file and keyring) to the operator for health checks
err = mon.WriteConnectionConfig(cluster.context, cluster.Info)
if err != nil {
return fmt.Errorf("failed to write connection info %+v", err)
}
// Get Ceph monitors version on the external cluster
cephMonVersion, err := client.GetCephMonVersion(c.context, namespace)
if err != nil {
return fmt.Errorf("failed to get ceph mon version. %+v", err)
}
// We only support Nautilus or newer
if !cephMonVersion.IsAtLeastNautilus() {
return fmt.Errorf("unsupported ceph version %s, need at least nautilus", cephMonVersion.String())
}
logger.Infof("detecting the image version provided for the external cluster...")
specCephVersionImage, err := cluster.detectCephVersion(c.rookImage, cluster.Spec.CephVersion.Image, detectCephVersionTimeout)
if err != nil {
return fmt.Errorf("unknown ceph major version. %+v", err)
}
specCephVersion := *specCephVersionImage
// Populate clusterInfo with the external cluster version
cluster.Info.CephVersion = *cephMonVersion
// Make sure the external cluster version and the ceph version in the provided image are identical
if !cephver.IsIdentical(specCephVersion, cluster.Info.CephVersion) {
return fmt.Errorf("wrong ceph version %s, external cluster version is %s, they must match", specCephVersion.String(), cephMonVersion.String())
}
// The cluster Identity must be established at this point
if !cluster.Info.IsInitialized() {
return fmt.Errorf("the cluster identity was not established: %+v", cluster.Info)
}
logger.Info("cluster identity established.")
// Everything went well so let's update the CR's status to "connected"
c.updateClusterStatus(namespace, name, cephv1.ClusterStateConnected, "")
// Mark initialization has done
cluster.initCompleted = true
return nil
}
func (c *ClusterController) configureLocalCephCluster(namespace, name string, cluster *cluster, clusterObj *cephv1.CephCluster) error {
if cluster.Spec.Mon.Count == 0 {
logger.Warningf("mon count should be at least 1, will use default value of %d", mon.DefaultMonCount)
cluster.Spec.Mon.Count = mon.DefaultMonCount
}
if cluster.Spec.Mon.Count%2 == 0 {
logger.Warningf("mon count is even (given: %d), should be uneven, continuing", cluster.Spec.Mon.Count)
}
if c.devicesInUse && cluster.Spec.Storage.AnyUseAllDevices() {
c.updateClusterStatus(clusterObj.Namespace, clusterObj.Name, cephv1.ClusterStateError, "using all devices in more than one namespace is not supported")
return
return fmt.Errorf("using all devices in more than one namespace is not supported")
}
if cluster.Spec.Storage.AnyUseAllDevices() {
c.devicesInUse = true
}
c.initializeCluster(cluster, clusterObj)
}
func (c *ClusterController) initializeCluster(cluster *cluster, clusterObj *cephv1.CephCluster) {
cluster.Spec = &clusterObj.Spec
if cluster.Spec.Mon.Count%2 == 0 {
logger.Warningf("mon count is even (given: %d), should be uneven, continuing", cluster.Spec.Mon.Count)
}
// Start the Rook cluster components. Retry several times in case of failure.
failedMessage := ""
state := cephv1.ClusterStateError
err := wait.Poll(clusterCreateInterval, clusterCreateTimeout,
// Try to load clusterInfo early so we can compare the running version with the one from the spec image
var err error
cluster.Info, _, _, err = mon.LoadClusterInfo(c.context, cluster.Namespace)
if err == nil {
// Let's write connection info (ceph config file and keyring) to the operator for health checks
err = mon.WriteConnectionConfig(cluster.context, cluster.Info)
if err != nil {
return err
}
}
err = wait.Poll(clusterCreateInterval, clusterCreateTimeout,
func() (bool, error) {
cephVersion, canRetry, err := c.detectAndValidateCephVersion(cluster, cluster.Spec.CephVersion.Image)
if err != nil {
@@ -288,47 +390,68 @@ func (c *ClusterController) initializeCluster(cluster *cluster, clusterObj *ceph
if state == cephv1.ClusterStateError {
// the cluster could not be initialized
return
return fmt.Errorf("failed to initialized the cluster")
}
return nil
}
func (c *ClusterController) initializeCluster(cluster *cluster, clusterObj *cephv1.CephCluster) {
cluster.Spec = &clusterObj.Spec
if !cluster.Spec.External.Enable {
if err := c.configureLocalCephCluster(clusterObj.Namespace, clusterObj.Name, cluster, clusterObj); err != nil {
logger.Errorf("failed to configure local ceph cluster. %+v", err)
return
}
} else {
if err := c.configureExternalCephCluster(clusterObj.Namespace, clusterObj.Name, cluster); err != nil {
logger.Errorf("failed to configure external ceph cluster. %+v", err)
return
}
}
// Start pool CRD watcher
poolController := pool.NewPoolController(c.context, cluster.Namespace)
poolController.StartWatch(cluster.stopCh)
poolController := pool.NewPoolController(c.context, cluster.Spec)
poolController.StartWatch(cluster.Namespace, cluster.stopCh)
// Start object store CRD watcher
objectStoreController := object.NewObjectStoreController(cluster.Info, c.context, cluster.Namespace, c.rookImage, cluster.Spec.CephVersion, cluster.Spec.Network.HostNetwork, cluster.ownerRef, cluster.Spec.DataDirHostPath)
objectStoreController.StartWatch(cluster.stopCh)
objectStoreController := object.NewObjectStoreController(cluster.Info, c.context, cluster.Namespace, c.rookImage, cluster.Spec, cluster.ownerRef, cluster.Spec.DataDirHostPath)
objectStoreController.StartWatch(cluster.Namespace, cluster.stopCh)
// Start object store user CRD watcher
objectStoreUserController := objectuser.NewObjectStoreUserController(c.context, cluster.Namespace, cluster.ownerRef)
objectStoreUserController := objectuser.NewObjectStoreUserController(c.context, cluster.Spec, cluster.Namespace, cluster.ownerRef)
objectStoreUserController.StartWatch(cluster.stopCh)
// Start file system CRD watcher
fileController := file.NewFilesystemController(cluster.Info, c.context, cluster.Namespace, c.rookImage, cluster.Spec.CephVersion, cluster.Spec.Network.HostNetwork, cluster.ownerRef, cluster.Spec.DataDirHostPath)
fileController.StartWatch(cluster.stopCh)
fileController := file.NewFilesystemController(cluster.Info, c.context, cluster.Namespace, c.rookImage, cluster.Spec, cluster.ownerRef, cluster.Spec.DataDirHostPath)
fileController.StartWatch(cluster.Namespace, cluster.stopCh)
// Start nfs ganesha CRD watcher
ganeshaController := nfs.NewCephNFSController(cluster.Info, c.context, cluster.Spec.DataDirHostPath, cluster.Namespace, c.rookImage, cluster.Spec.CephVersion, cluster.Spec.Network.HostNetwork, cluster.ownerRef)
ganeshaController.StartWatch(cluster.stopCh)
ganeshaController := nfs.NewCephNFSController(cluster.Info, c.context, cluster.Spec.DataDirHostPath, cluster.Namespace, c.rookImage, cluster.Spec, cluster.ownerRef)
ganeshaController.StartWatch(cluster.Namespace, cluster.stopCh)
cluster.childControllers = []childController{
poolController, objectStoreController, objectStoreUserController, fileController, ganeshaController,
// Populate ClusterInfo
if cluster.Spec.External.Enable {
cluster.mons.ClusterInfo = cluster.Info
}
// Start mon health checker
healthChecker := mon.NewHealthChecker(cluster.mons)
healthChecker := mon.NewHealthChecker(cluster.mons, cluster.Spec)
go healthChecker.Check(cluster.stopCh)
// Start the osd health checker
osdChecker := osd.NewMonitor(c.context, cluster.Namespace)
go osdChecker.Start(cluster.stopCh)
if !cluster.Spec.External.Enable {
// Start the osd health checker only if running OSDs in the local ceph cluster
osdChecker := osd.NewMonitor(c.context, cluster.Namespace)
go osdChecker.Start(cluster.stopCh)
}
// Start the ceph status checker
cephChecker := newCephStatusChecker(c.context, cluster.Namespace, clusterObj.Name)
go cephChecker.checkCephStatus(cluster.stopCh)
// add the finalizer to the crd
err = c.addFinalizer(clusterObj)
err := c.addFinalizer(clusterObj.Namespace, clusterObj.Name)
if err != nil {
logger.Errorf("failed to add finalizer to cluster crd. %+v", err)
}
@@ -404,6 +527,12 @@ func (c *ClusterController) onUpdate(oldObj, newObj interface{}) {
logger.Debugf("update event for cluster %s", newClust.Namespace)
if existing, ok := c.clusterMap[newClust.Namespace]; ok && existing.crdName != newClust.Name {
logger.Errorf("Skipping update of cluster cr %s in namespace %s. Cluster cr %s already exists in this namespace. Only one cluster cr per namespace is supported.",
newClust.Name, newClust.Namespace, existing.crdName)
return
}
// Check if the cluster is being deleted. This code path is called when a finalizer is specified in the crd.
// When a cluster is requested for deletion, K8s will only set the deletion timestamp if there are any finalizers in the list.
// K8s will only delete the crd and child resources when the finalizers have been removed from the crd.
@@ -595,6 +724,12 @@ func (c *ClusterController) onDelete(obj interface{}) {
return
}
if existing, ok := c.clusterMap[clust.Namespace]; ok && existing.crdName != clust.Name {
logger.Errorf("Skipping deletion of cluster cr %s in namespace %s. Cluster cr %s already exists in this namespace. Only one cluster cr per namespace is supported.",
clust.Name, clust.Namespace, existing.crdName)
return
}
logger.Infof("delete event for cluster %s in namespace %s", clust.Name, clust.Namespace)
err = c.handleDelete(clust, time.Duration(clusterDeleteRetryInterval)*time.Second)
@@ -605,8 +740,17 @@ func (c *ClusterController) onDelete(obj interface{}) {
close(cluster.stopCh)
delete(c.clusterMap, clust.Namespace)
}
if clust.Spec.Storage.AnyUseAllDevices() {
c.devicesInUse = false
// Only valid when the cluster is not external
if !clust.Spec.External.Enable {
if clust.Spec.Storage.AnyUseAllDevices() {
c.devicesInUse = false
}
} else {
err := purgeExternalCluster(c.context.Clientset, clust.Namespace)
if err != nil {
logger.Errorf("failed to purge external cluster ressources. %+v", err)
}
return
}
}
@@ -656,13 +800,41 @@ func (c *ClusterController) handleDelete(cluster *cephv1.CephCluster, retryInter
return nil
}
func purgeExternalCluster(clientset kubernetes.Interface, namespace string) error {
// purge the mon endpoint config map
err := clientset.CoreV1().ConfigMaps(namespace).Delete(mon.EndpointConfigMapName, &metav1.DeleteOptions{})
if err != nil && !errors.IsNotFound(err) {
return err
}
// purge the config flag overrides
err = clientset.CoreV1().ConfigMaps(namespace).Delete(config.StoreName, &metav1.DeleteOptions{})
if err != nil && !errors.IsNotFound(err) {
return err
}
// purge config override configmap?
err = clientset.CoreV1().ConfigMaps(namespace).Delete(k8sutil.ConfigOverrideName, &metav1.DeleteOptions{})
if err != nil && !errors.IsNotFound(err) {
return err
}
// Now delete secret
err = clientset.CoreV1().Secrets(namespace).Delete(mon.AppName, &metav1.DeleteOptions{})
if err != nil {
return fmt.Errorf("failed to delete secret %+v. %+v", mon.AppName, err)
}
return nil
}
// ************************************************************************************************
// Finalizer functions
// ************************************************************************************************
func (c *ClusterController) addFinalizer(clust *cephv1.CephCluster) error {
func (c *ClusterController) addFinalizer(namespace, name string) error {
// get the latest cluster object since we probably updated it before we got to this point (e.g. by updating its status)
clust, err := c.context.RookClientset.CephV1().CephClusters(clust.Namespace).Get(clust.Name, metav1.GetOptions{})
clust, err := c.context.RookClientset.CephV1().CephClusters(namespace).Get(name, metav1.GetOptions{})
if err != nil {
return err
}
@@ -788,3 +960,14 @@ func printOverallCephVersion(context *clusterd.Context, namespace string) {
logger.Warningf("upgrade orchestration completed but somehow we still have more than one Ceph version running. %+v:", versions.Overall)
}
}
func validateExternalClusterSpec(cluster *cluster) error {
if cluster.Spec.DataDirHostPath == "" {
return fmt.Errorf("dataDirHostPath must be specified")
}
if cluster.Spec.CephVersion.Image == "" {
return fmt.Errorf("spec image must be specified")
}
return nil
}
+16 -2
View File
@@ -78,7 +78,7 @@ func TestClusterDelete(t *testing.T) {
},
}
callback := func() error {
callback := func(external bool) error {
logger.Infof("test success callback")
return nil
}
@@ -156,7 +156,7 @@ func TestRemoveFinalizer(t *testing.T) {
Clientset: clientset,
RookClientset: rookfake.NewSimpleClientset(),
}
callback := func() error {
callback := func(external bool) error {
return fmt.Errorf("test failed callback")
}
controller := NewClusterController(context, "", &attachment.MockAttachment{}, callback)
@@ -186,3 +186,17 @@ func TestRemoveFinalizer(t *testing.T) {
assert.NotNil(t, cluster)
assert.Len(t, cluster.Finalizers, 0)
}
func TestValidateExternalClusterSpec(t *testing.T) {
c := &cluster{Spec: &cephv1.ClusterSpec{}, mons: &mon.Cluster{}}
err := validateExternalClusterSpec(c)
assert.Error(t, err)
c.Spec.DataDirHostPath = "path"
err = validateExternalClusterSpec(c)
assert.Error(t, err)
c.Spec.CephVersion.Image = "ceph/ceph:v14.2.2-20190722"
err = validateExternalClusterSpec(c)
assert.NoError(t, err)
}
+4 -4
View File
@@ -54,8 +54,8 @@ const (
func (c *Cluster) genMonSharedKeyring() string {
return fmt.Sprintf(
keyringTemplate,
c.clusterInfo.MonitorSecret,
cephconfig.AdminKeyring(c.clusterInfo),
c.ClusterInfo.MonitorSecret,
cephconfig.AdminKeyring(c.ClusterInfo),
)
}
@@ -84,7 +84,7 @@ func CreateOrLoadClusterInfo(context *clusterd.Context, namespace string, ownerR
Port: map[string]int32{},
}
secrets, err := context.Clientset.CoreV1().Secrets(namespace).Get(appName, metav1.GetOptions{})
secrets, err := context.Clientset.CoreV1().Secrets(namespace).Get(AppName, metav1.GetOptions{})
if err != nil {
if !errors.IsNotFound(err) {
return nil, maxMonID, monMapping, fmt.Errorf("failed to get mon secrets. %+v", err)
@@ -216,7 +216,7 @@ func createClusterAccessSecret(clientset kubernetes.Interface, namespace string,
}
secret := &v1.Secret{
ObjectMeta: metav1.ObjectMeta{
Name: appName,
Name: AppName,
Namespace: namespace,
},
Data: secrets,
+2 -2
View File
@@ -31,12 +31,12 @@ func EndpointEnvVar() v1.EnvVar {
// SecretEnvVar is the mon secret environment var
func SecretEnvVar() v1.EnvVar {
ref := &v1.SecretKeySelector{LocalObjectReference: v1.LocalObjectReference{Name: appName}, Key: monSecretName}
ref := &v1.SecretKeySelector{LocalObjectReference: v1.LocalObjectReference{Name: AppName}, Key: monSecretName}
return v1.EnvVar{Name: "ROOK_MON_SECRET", ValueFrom: &v1.EnvVarSource{SecretKeyRef: ref}}
}
// AdminSecretEnvVar is the admin secret environment var
func AdminSecretEnvVar() v1.EnvVar {
ref := &v1.SecretKeySelector{LocalObjectReference: v1.LocalObjectReference{Name: appName}, Key: adminSecretName}
ref := &v1.SecretKeySelector{LocalObjectReference: v1.LocalObjectReference{Name: AppName}, Key: adminSecretName}
return v1.EnvVar{Name: "ROOK_ADMIN_SECRET", ValueFrom: &v1.EnvVarSource{SecretKeyRef: ref}}
}
+211 -24
View File
@@ -24,6 +24,7 @@ import (
"github.com/rook/rook/pkg/clusterd"
"github.com/rook/rook/pkg/daemon/ceph/client"
cephconfig "github.com/rook/rook/pkg/daemon/ceph/config"
cephutil "github.com/rook/rook/pkg/daemon/ceph/util"
"github.com/rook/rook/pkg/operator/k8sutil"
v1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/api/errors"
@@ -40,18 +41,25 @@ var (
// HealthChecker aggregates the mon/cluster info needed to check the health of the monitors
type HealthChecker struct {
monCluster *Cluster
monCluster *Cluster
clusterSpec *cephv1.ClusterSpec
}
// NewHealthChecker creates a new HealthChecker object
func NewHealthChecker(monCluster *Cluster) *HealthChecker {
func NewHealthChecker(monCluster *Cluster, clusterSpec *cephv1.ClusterSpec) *HealthChecker {
return &HealthChecker{
monCluster: monCluster,
monCluster: monCluster,
clusterSpec: clusterSpec,
}
}
// 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:
@@ -62,7 +70,7 @@ func (hc *HealthChecker) Check(stopCh chan struct{}) {
logger.Debugf("checking health of mons")
err := hc.monCluster.checkHealth()
if err != nil {
logger.Infof("failed to check mon health. %+v", err)
logger.Warningf("failed to check mon health. %+v", err)
}
}
}
@@ -72,7 +80,18 @@ func (c *Cluster) checkHealth() error {
c.acquireOrchestrationLock()
defer c.releaseOrchestrationLock()
logger.Debugf("Checking health for mons in cluster. %s", c.clusterInfo.Name)
logger.Debugf("Checking health for mons in cluster. %s", c.ClusterInfo.Name)
// connect to the mons
// get the status and check for quorum
status, err := client.GetMonStatus(c.context, c.ClusterInfo.Name, true)
if err != nil {
return fmt.Errorf("failed to get mon status. %+v", err)
}
logger.Debugf("Mon status: %+v", status)
if c.spec.External.Enable {
return c.handleExternalMonStatus(status)
}
// Use a local mon count in case the user updates the crd in another goroutine.
// We need to complete a health check with a consistent value.
@@ -82,17 +101,9 @@ func (c *Cluster) checkHealth() error {
}
logger.Debugf(msg)
// connect to the mons
// get the status and check for quorum
status, err := client.GetMonStatus(c.context, c.clusterInfo.Name, true)
if err != nil {
return fmt.Errorf("failed to get mon status. %+v", err)
}
logger.Debugf("Mon status: %+v", status)
// Source of truth of which mons should exist is our *clusterInfo*
monsNotFound := map[string]interface{}{}
for _, mon := range c.clusterInfo.Monitors {
for _, mon := range c.ClusterInfo.Monitors {
monsNotFound[mon.Name] = struct{}{}
}
@@ -156,7 +167,7 @@ func (c *Cluster) checkHealth() error {
// handle all mons that haven't been in the Ceph mon map
for mon := range monsNotFound {
logger.Warningf("mon %s NOT found in ceph mon map, failover", mon)
c.failMon(len(c.clusterInfo.Monitors), desiredMonCount, mon)
c.failMon(len(c.ClusterInfo.Monitors), desiredMonCount, mon)
// only deal with one "not found in ceph mon map" mon per health check
return nil
}
@@ -187,6 +198,54 @@ func (c *Cluster) checkHealth() error {
return nil
}
func (c *Cluster) checkMonsOnSameNode(desiredMonCount int) (bool, error) {
nodesUsed := map[string]struct{}{}
for name, node := range c.mapping.Node {
// when the node is already in the list we have more than one mon on that node
if _, ok := nodesUsed[node.Name]; ok {
// get list of available nodes for mons
availableNodes, _, err := c.getAvailableMonNodes()
if err != nil {
return true, fmt.Errorf("failed to get available mon nodes. %+v", err)
}
// if there are enough nodes for one mon "that is too much" to be failovered,
// fail it over to an other node
if len(availableNodes) > 0 {
logger.Infof("rebalance: enough nodes available %d to failover mon %s", len(availableNodes), name)
c.failMon(len(c.ClusterInfo.Monitors), desiredMonCount, name)
} else {
logger.Debugf("rebalance: not enough nodes available to failover mon %s", name)
}
// deal with one mon too much on a node at a time
return true, nil
}
nodesUsed[node.Name] = struct{}{}
}
return false, nil
}
func (c *Cluster) checkMonsOnValidNodes() (bool, error) {
for mon, nInfo := range c.mapping.Node {
// get node to use for validNode() func
node, err := c.context.Clientset.CoreV1().Nodes().Get(nInfo.Name, metav1.GetOptions{})
if err != nil {
return true, err
}
// check if node the mon is on is still valid
valid, err := k8sutil.ValidNode(*node, cephv1.GetMonPlacement(c.spec.Placement))
if err != nil {
logger.Warning("failed to validate node %s %v", node.Name, err)
} else if !valid {
logger.Warningf("node %s isn't valid anymore, failover mon %s", nInfo.Name, mon)
c.failoverMon(mon)
return true, nil
}
logger.Debugf("node %s with mon %s is still valid", nInfo.Name, mon)
}
return false, nil
}
// failMon compares the monCount against desiredMonCount
func (c *Cluster) failMon(monCount, desiredMonCount int, name string) {
if monCount > desiredMonCount {
@@ -231,7 +290,7 @@ func (c *Cluster) failoverMon(name string) error {
} else {
m.PublicIP = serviceIP
}
c.clusterInfo.Monitors[m.DaemonName] = cephconfig.NewMonInfo(m.DaemonName, m.PublicIP, m.Port)
c.ClusterInfo.Monitors[m.DaemonName] = cephconfig.NewMonInfo(m.DaemonName, m.PublicIP, m.Port)
// Start the deployment
if err = c.startDeployments(mConf, true); err != nil {
@@ -262,10 +321,10 @@ func (c *Cluster) removeMon(daemonName string) error {
}
// Remove the bad monitor from quorum
if err := removeMonitorFromQuorum(c.context, c.clusterInfo.Name, daemonName); err != nil {
if err := removeMonitorFromQuorum(c.context, c.ClusterInfo.Name, daemonName); err != nil {
return fmt.Errorf("failed to remove mon %s from quorum. %+v", daemonName, err)
}
delete(c.clusterInfo.Monitors, daemonName)
delete(c.ClusterInfo.Monitors, daemonName)
// check if a mapping exists for the mon
if _, ok := c.mapping.Node[daemonName]; ok {
nodeName := c.mapping.Node[daemonName].Name
@@ -296,11 +355,6 @@ func (c *Cluster) removeMon(daemonName string) error {
return fmt.Errorf("failed to save mon config after failing over mon %s. %+v", daemonName, err)
}
// make sure to rewrite the config so NO new connections are made to the removed mon
if err := WriteConnectionConfig(c.context, c.clusterInfo); err != nil {
return fmt.Errorf("failed to write connection config after failing over mon %s. %+v", daemonName, err)
}
return nil
}
@@ -329,7 +383,7 @@ func (c *Cluster) resolveInvalidMonitorPlacement(desiredMonCount int) (bool, err
for name, nodeInfo := range c.mapping.Node {
if nodeInfo.Name == nodeChoice.Node.Name {
logger.Infof("rebalance: rescheduling mon %s from node %s", name, nodeInfo.Name)
c.failMon(len(c.clusterInfo.Monitors), desiredMonCount, name)
c.failMon(len(c.ClusterInfo.Monitors), desiredMonCount, name)
return true, nil
}
}
@@ -459,3 +513,136 @@ func (c *Cluster) findInvalidMonitorPlacement(desiredMonCount int) (*NodeUsage,
return nil, nil
}
func (c *Cluster) handleExternalMonStatus(status client.MonStatusResponse) error {
changed, err := c.addOrRemoveExternalMonitor(status)
if err != nil {
return fmt.Errorf("failed to add or remove external mon. %+v", err)
}
// let's save the monitor's config if anything happened
if changed {
if err := c.saveMonConfig(); err != nil {
return fmt.Errorf("failed to save mon config after adding/removing external mon. %+v", err)
}
}
return nil
}
func (c *Cluster) addOrRemoveExternalMonitor(status client.MonStatusResponse) (bool, error) {
// func addOrRemoveExternalMonitor(status client.MonStatusResponse) (bool, error) {
var changed bool
// Populate a map with the current monitor from ClusterInfo
monsNotFound := map[string]interface{}{}
for _, mon := range c.ClusterInfo.Monitors {
monsNotFound[mon.Name] = struct{}{}
}
// Iterate over the mons first and compare it with ClusterInfo
for _, mon := range status.MonMap.Mons {
inQuorum := monInQuorum(mon, status.Quorum)
// if the mon is not in clusterInfo
if _, ok := monsNotFound[mon.Name]; !ok {
// If the mon is part of the quorum
if inQuorum {
// let's add it to ClusterInfo
// Always pick the v1 endpoint, that's the easier thing to do since some people might not have activated msgr2
// We must run on at least Nautilus so we are 100% certain that will have a field mon.PublicAddrs.Addrvec with 'v1' in it
var endpoint string
for i := range mon.PublicAddrs.Addrvec {
if mon.PublicAddrs.Addrvec[i].Type == "v1" {
endpoint = mon.PublicAddrs.Addrvec[i].Addr
}
}
// find IP and Port of that Mon
monIP := cephutil.GetIPFromEndpoint(endpoint)
monPort := cephutil.GetPortFromEndpoint(endpoint)
logger.Infof("new external mon %s found: %s, adding it", mon.Name, endpoint)
c.ClusterInfo.Monitors[mon.Name] = cephconfig.NewMonInfo(mon.Name, monIP, monPort)
changed = true
}
logger.Debugf("mon %s is not in quorum and not in ClusterInfo", mon.Name)
} else {
// mon is in ClusterInfo
logger.Debugf("mon %s is in ClusterInfo, let's test if it's in quorum", mon.Name)
if !inQuorum {
// this mon was in clusterInfo but not part of the quorum anymore
// let's remove it from ClusterInfo
logger.Infof("monitor %s is not part of the external cluster monitor quorum, removing it", mon.Name)
delete(c.ClusterInfo.Monitors, mon.Name)
changed = true
}
}
logger.Debugf("everything is fine mon %s in the clusterInfo and its quorum status is %v", mon.Name, inQuorum)
}
// Let's do a new iteration, over ClusterInfo this time to catch mon that might have disappeared
for _, mon := range c.ClusterInfo.Monitors {
// if the mon does not exist in the monmap
monInMonMap := isMonInMonMap(mon.Name, status.MonMap.Mons)
if !monInMonMap {
// mon is in clusterInfo but not part of the quorum removing it
logger.Infof("monitor %s disappeared from the external cluster monitor map, removing it", mon.Name)
delete(c.ClusterInfo.Monitors, mon.Name)
changed = true
} else {
// it's in the mon map but is it part of the quorum?
logger.Debug("need to check is the mon not in ClusterInfo is part of a quorum or not")
monInQuorum := isMonInQuorum(mon.Name, status.MonMap.Mons, status.Quorum)
if !monInQuorum {
logger.Infof("external mon %s in the mon map but not in quorum, removing it", mon.Name)
delete(c.ClusterInfo.Monitors, mon.Name)
changed = true
}
}
logger.Debugf("mon %s endpoint is %s", mon.Name, mon.Endpoint)
}
logger.Debugf("ClusterInfo.Monitors is %+v", c.ClusterInfo.Monitors)
return changed, nil
}
func isMonInMonMap(monToTest string, Mons []client.MonMapEntry) bool {
for _, m := range Mons {
if m.Name == monToTest {
logger.Debugf("mon %s is in the monmap!", monToTest)
return true
}
}
return false
}
func isMonInQuorum(monToTest string, Mons []client.MonMapEntry, quorum []int) bool {
monRank := getMonRank(monToTest, Mons)
if monRank == -1 {
// this is an error, we can't find the rank
// so we just return false and the mon will get deleted
logger.Debugf("mon %s rank is %d", monToTest, monRank)
return false
}
for _, rank := range quorum {
logger.Debugf("rank is %d, monRank is %d", rank, monRank)
if rank == monRank {
logger.Debugf("mon %s is part of the quorum", monToTest)
return true
}
}
logger.Debugf("could not find the rank %d for %s", monRank, monToTest)
return false
}
func getMonRank(monToTest string, Mons []client.MonMapEntry) int {
for _, m := range Mons {
if m.Name == monToTest {
logger.Debugf("mon rank is %d", m.Rank)
return m.Rank
}
}
return -1
}
+179 -13
View File
@@ -28,6 +28,7 @@ import (
"github.com/rook/rook/pkg/clusterd"
"github.com/rook/rook/pkg/daemon/ceph/client"
clienttest "github.com/rook/rook/pkg/daemon/ceph/client/test"
cephconfig "github.com/rook/rook/pkg/daemon/ceph/config"
testopk8s "github.com/rook/rook/pkg/operator/k8sutil/test"
"github.com/rook/rook/pkg/operator/test"
exectest "github.com/rook/rook/pkg/util/exec/test"
@@ -57,7 +58,7 @@ func TestCheckHealth(t *testing.T) {
}
c := New(context, "ns", "", false, metav1.OwnerReference{}, &sync.Mutex{})
setCommonMonProperties(c, 1, cephv1.MonSpec{Count: 3, AllowMultiplePerNode: true}, "myversion")
logger.Infof("initial mons: %v", c.clusterInfo.Monitors)
logger.Infof("initial mons: %v", c.ClusterInfo.Monitors)
c.waitForStart = false
defer os.RemoveAll(c.context.ConfigDir)
@@ -70,7 +71,7 @@ func TestCheckHealth(t *testing.T) {
err := c.checkHealth()
assert.Nil(t, err)
logger.Infof("mons after checkHealth: %v", c.clusterInfo.Monitors)
logger.Infof("mons after checkHealth: %v", c.ClusterInfo.Monitors)
assert.ElementsMatch(t, []string{"rook-ceph-mon-a", "rook-ceph-mon-f"}, testopk8s.DeploymentNamesUpdated(deploymentsUpdated))
testopk8s.ClearDeploymentsUpdated(deploymentsUpdated)
@@ -83,8 +84,8 @@ func TestCheckHealth(t *testing.T) {
"g",
}
for _, monName := range newMons {
_, ok := c.clusterInfo.Monitors[monName]
assert.True(t, ok, fmt.Sprintf("mon %s not found in monitor list. %v", monName, c.clusterInfo.Monitors))
_, ok := c.ClusterInfo.Monitors[monName]
assert.True(t, ok, fmt.Sprintf("mon %s not found in monitor list. %v", monName, c.ClusterInfo.Monitors))
}
}
@@ -469,7 +470,7 @@ func TestAddRemoveMons(t *testing.T) {
// checking the health will increase the mons as desired all in one go
err := c.checkHealth()
assert.Nil(t, err)
assert.Equal(t, 5, len(c.clusterInfo.Monitors), fmt.Sprintf("mons: %v", c.clusterInfo.Monitors))
assert.Equal(t, 5, len(c.ClusterInfo.Monitors), fmt.Sprintf("mons: %v", c.ClusterInfo.Monitors))
assert.ElementsMatch(t, []string{
// b is created first, no updates
"rook-ceph-mon-b", // b updated when c created
@@ -480,40 +481,205 @@ func TestAddRemoveMons(t *testing.T) {
testopk8s.ClearDeploymentsUpdated(deploymentsUpdated)
// reducing the mon count to 3 will reduce the mon count once each time we call checkHealth
monQuorumResponse = clienttest.MonInQuorumResponseFromMons(c.clusterInfo.Monitors)
monQuorumResponse = clienttest.MonInQuorumResponseFromMons(c.ClusterInfo.Monitors)
c.spec.Mon.Count = 3
err = c.checkHealth()
assert.Nil(t, err)
assert.Equal(t, 4, len(c.clusterInfo.Monitors))
assert.Equal(t, 4, len(c.ClusterInfo.Monitors))
// No updates in unit tests w/ workaround
assert.ElementsMatch(t, []string{}, testopk8s.DeploymentNamesUpdated(deploymentsUpdated))
testopk8s.ClearDeploymentsUpdated(deploymentsUpdated)
// after the second call we will be down to the expected count of 3
monQuorumResponse = clienttest.MonInQuorumResponseFromMons(c.clusterInfo.Monitors)
monQuorumResponse = clienttest.MonInQuorumResponseFromMons(c.ClusterInfo.Monitors)
err = c.checkHealth()
assert.Nil(t, err)
assert.Equal(t, 3, len(c.clusterInfo.Monitors))
assert.Equal(t, 3, len(c.ClusterInfo.Monitors))
// No updates in unit tests w/ workaround
assert.ElementsMatch(t, []string{}, testopk8s.DeploymentNamesUpdated(deploymentsUpdated))
testopk8s.ClearDeploymentsUpdated(deploymentsUpdated)
// now attempt to reduce the mons down to quorum size 1
monQuorumResponse = clienttest.MonInQuorumResponseFromMons(c.clusterInfo.Monitors)
monQuorumResponse = clienttest.MonInQuorumResponseFromMons(c.ClusterInfo.Monitors)
c.spec.Mon.Count = 1
err = c.checkHealth()
assert.Nil(t, err)
assert.Equal(t, 2, len(c.clusterInfo.Monitors))
assert.Equal(t, 2, len(c.ClusterInfo.Monitors))
// No updates in unit tests w/ workaround
assert.ElementsMatch(t, []string{}, testopk8s.DeploymentNamesUpdated(deploymentsUpdated))
testopk8s.ClearDeploymentsUpdated(deploymentsUpdated)
// cannot reduce from quorum size of 2 to 1
monQuorumResponse = clienttest.MonInQuorumResponseFromMons(c.clusterInfo.Monitors)
monQuorumResponse = clienttest.MonInQuorumResponseFromMons(c.ClusterInfo.Monitors)
err = c.checkHealth()
assert.Nil(t, err)
assert.Equal(t, 2, len(c.clusterInfo.Monitors))
assert.Equal(t, 2, len(c.ClusterInfo.Monitors))
// No updates in unit tests w/ workaround
assert.ElementsMatch(t, []string{}, testopk8s.DeploymentNamesUpdated(deploymentsUpdated))
testopk8s.ClearDeploymentsUpdated(deploymentsUpdated)
}
func TestAddOrRemoveExternalMonitor(t *testing.T) {
var changed bool
var err error
// populate fake monmap
fakeResp := client.MonStatusResponse{Quorum: []int{0}}
fakeAddrvecEntry := []client.AddrvecEntry{
{
Addr: "172.17.0.4:3300",
},
}
fakeResp.MonMap.Mons = []client.MonMapEntry{
{
Name: "a",
},
}
fakeResp.MonMap.Mons[0].PublicAddrs.Addrvec = fakeAddrvecEntry
// populate fake ClusterInfo
c := &Cluster{ClusterInfo: &cephconfig.ClusterInfo{}}
c.ClusterInfo = test.CreateConfigDir(1)
//
// TEST 1
//
// both clusterInfo and mon map are identical so nil is expected
changed, err = c.addOrRemoveExternalMonitor(fakeResp)
assert.NoError(t, err)
assert.False(t, changed)
assert.Equal(t, 1, len(c.ClusterInfo.Monitors))
//
// TEST 2
//
// Now let's test the case where mon disappeared from the external cluster
// ClusterInfo still has them but they are gone from the monmap.
// Thus they should be removed from ClusterInfo
c.ClusterInfo = test.CreateConfigDir(3)
changed, err = c.addOrRemoveExternalMonitor(fakeResp)
assert.NoError(t, err)
assert.True(t, changed)
// ClusterInfo should shrink to 1
assert.Equal(t, 1, len(c.ClusterInfo.Monitors))
//
// TEST 3
//
// Now let's add a new mon in the external cluster
// ClusterInfo should be updated with this new monitor
fakeResp.MonMap.Mons = []client.MonMapEntry{
{
Name: "a",
},
{
Name: "b",
},
}
fakeAddrvecEntry2 := []client.AddrvecEntry{
{
Addr: "172.17.0.5:3300",
},
}
fakeResp.MonMap.Mons[1].PublicAddrs.Addrvec = fakeAddrvecEntry2
c.ClusterInfo = test.CreateConfigDir(1)
changed, err = c.addOrRemoveExternalMonitor(fakeResp)
assert.NoError(t, err)
assert.True(t, changed)
// ClusterInfo should now have 2 monitors
assert.Equal(t, 2, len(c.ClusterInfo.Monitors))
//
// TEST 4
//
// Now let's test the case where the mon is in clusterInfo, part of the monmap but not in quorum!
c.ClusterInfo = test.CreateConfigDir(1)
fakeResp2 := client.MonStatusResponse{Quorum: []int{1}} // quorum is owned by the mon with the rank 1 and our mon rank is 0
fakeAddrvecEntry3 := []client.AddrvecEntry{
{
Addr: "172.17.0.4:3300",
},
}
fakeResp2.MonMap.Mons = []client.MonMapEntry{
{
Name: "a",
Rank: 0,
},
}
fakeResp2.MonMap.Mons[0].PublicAddrs.Addrvec = fakeAddrvecEntry3
changed, err = c.addOrRemoveExternalMonitor(fakeResp2)
assert.NoError(t, err)
assert.True(t, changed)
assert.Equal(t, 0, len(c.ClusterInfo.Monitors))
}
func TestIsMonInMonMapt(t *testing.T) {
fakeResp := client.MonStatusResponse{}
fakeResp.MonMap.Mons = []client.MonMapEntry{
{
Name: "a",
},
{
Name: "b",
},
{
Name: "c",
},
}
isIT := isMonInMonMap("a", fakeResp.MonMap.Mons)
assert.True(t, isIT)
isIT = isMonInMonMap("z", fakeResp.MonMap.Mons)
assert.False(t, isIT)
}
func TestGetMonRankt(t *testing.T) {
fakeResp := client.MonStatusResponse{}
fakeResp.MonMap.Mons = []client.MonMapEntry{
{
Name: "a",
Rank: 0,
},
{
Name: "b",
Rank: 1,
},
{
Name: "c",
Rank: 2,
},
}
isIT := getMonRank("a", fakeResp.MonMap.Mons)
assert.Equal(t, 0, isIT)
isIT = getMonRank("b", fakeResp.MonMap.Mons)
assert.Equal(t, 1, isIT)
isIT = getMonRank("z", fakeResp.MonMap.Mons)
assert.Equal(t, -1, isIT)
}
func TestIsMonInQuorum(t *testing.T) {
fakeResp := client.MonStatusResponse{Quorum: []int{0, 1}}
fakeResp.MonMap.Mons = []client.MonMapEntry{
{
Name: "a",
Rank: 0,
},
{
Name: "b",
Rank: 1,
},
{
Name: "c",
Rank: 2,
},
}
isIT := isMonInQuorum("a", fakeResp.MonMap.Mons, fakeResp.Quorum)
assert.True(t, isIT)
isIT = isMonInQuorum("c", fakeResp.MonMap.Mons, fakeResp.Quorum)
assert.False(t, isIT)
isIT = isMonInQuorum("z", fakeResp.MonMap.Mons, fakeResp.Quorum)
assert.False(t, isIT)
}
+31 -29
View File
@@ -55,7 +55,8 @@ const (
// MappingKey is the name of the mapping for the mon->node and node->port
MappingKey = "mapping"
appName = "rook-ceph-mon"
// AppName is the name of the secret storing cluster mon.admin key, fsid and name
AppName = "rook-ceph-mon"
monNodeAttr = "mon_node"
monClusterAttr = "mon_cluster"
tprName = "mon.rook.io"
@@ -90,7 +91,7 @@ var (
// Cluster represents the Rook and environment configuration settings needed to set up Ceph mons.
type Cluster struct {
clusterInfo *cephconfig.ClusterInfo
ClusterInfo *cephconfig.ClusterInfo
context *clusterd.Context
spec cephv1.ClusterSpec
Namespace string
@@ -166,7 +167,7 @@ func (c *Cluster) Start(clusterInfo *cephconfig.ClusterInfo, rookVersion string,
c.acquireOrchestrationLock()
defer c.releaseOrchestrationLock()
c.clusterInfo = clusterInfo
c.ClusterInfo = clusterInfo
c.rookVersion = rookVersion
c.spec = spec
@@ -195,7 +196,7 @@ func (c *Cluster) Start(clusterInfo *cephconfig.ClusterInfo, rookVersion string,
logger.Infof(msg)
// create the mons for a new cluster or ensure mons are running in an existing cluster
return c.clusterInfo, c.startMons(targetCount)
return c.ClusterInfo, c.startMons(targetCount)
}
func (c *Cluster) startMons(targetCount int) error {
@@ -223,13 +224,13 @@ func (c *Cluster) startMons(targetCount int) error {
}
// Enable Ceph messenger 2 protocol on Nautilus
if c.clusterInfo.CephVersion.IsAtLeastNautilus() {
v, err := client.GetCephMonVersion(c.context, c.clusterInfo.Name)
if c.ClusterInfo.CephVersion.IsAtLeastNautilus() {
v, err := client.GetCephMonVersion(c.context, c.ClusterInfo.Name)
if err != nil {
return fmt.Errorf("failed to get ceph mon version. %+v", err)
}
if v.IsAtLeastNautilus() {
versions, err := client.GetAllCephDaemonVersions(c.context, c.clusterInfo.Name)
versions, err := client.GetAllCephDaemonVersions(c.context, c.ClusterInfo.Name)
if err != nil {
return fmt.Errorf("failed to get ceph daemons versions. %+v", err)
}
@@ -242,7 +243,7 @@ func (c *Cluster) startMons(targetCount int) error {
}
}
logger.Debugf("mon endpoints used are: %s", FlattenMonEndpoints(c.clusterInfo.Monitors))
logger.Debugf("mon endpoints used are: %s", FlattenMonEndpoints(c.ClusterInfo.Monitors))
return nil
}
@@ -261,7 +262,7 @@ func (c *Cluster) ensureMonsRunning(mons []*monConfig, i, targetCount int, requi
// Calculate how many mons we expected to exist after this method is completed.
// If we are adding a new mon, we expect one more than currently exist.
// If we haven't created all the desired mons already, we will be adding a new one with this iteration
expectedMonCount := len(c.clusterInfo.Monitors)
expectedMonCount := len(c.ClusterInfo.Monitors)
if expectedMonCount < targetCount {
expectedMonCount++
}
@@ -277,7 +278,7 @@ func (c *Cluster) ensureMonsRunning(mons []*monConfig, i, targetCount int, requi
}
// make sure we have the connection info generated so connections can happen
if err := WriteConnectionConfig(c.context, c.clusterInfo); err != nil {
if err := WriteConnectionConfig(c.context, c.ClusterInfo); err != nil {
return err
}
@@ -295,11 +296,12 @@ func (c *Cluster) initClusterInfo(cephVersion cephver.CephVersion) error {
var err error
// get the cluster info from secret
c.clusterInfo, c.maxMonID, c.mapping, err = CreateOrLoadClusterInfo(c.context, c.Namespace, &c.ownerRef)
c.ClusterInfo, c.maxMonID, c.mapping, err = CreateOrLoadClusterInfo(c.context, c.Namespace, &c.ownerRef)
c.ClusterInfo.CephVersion = cephVersion
if err != nil {
return fmt.Errorf("failed to get cluster info. %+v", err)
}
c.clusterInfo.CephVersion = cephVersion
// save cluster monitor config
if err = c.saveMonConfig(); err != nil {
@@ -312,7 +314,7 @@ func (c *Cluster) initClusterInfo(cephVersion cephver.CephVersion) error {
return fmt.Errorf("failed to save mon keyring secret. %+v", err)
}
// also store the admin keyring for other daemons that might need it during init
if err := k.Admin().CreateOrUpdate(c.clusterInfo); err != nil {
if err := k.Admin().CreateOrUpdate(c.ClusterInfo); err != nil {
return fmt.Errorf("failed to save admin keyring secret. %+v", err)
}
@@ -323,7 +325,7 @@ func (c *Cluster) initMonConfig(size int) (int, []*monConfig) {
mons := []*monConfig{}
// initialize the mon pod info for mons that have been previously created
for _, monitor := range c.clusterInfo.Monitors {
for _, monitor := range c.ClusterInfo.Monitors {
mons = append(mons, &monConfig{
ResourceName: resourceName(monitor.Name),
DaemonName: monitor.Name,
@@ -334,8 +336,8 @@ func (c *Cluster) initMonConfig(size int) (int, []*monConfig) {
}
// initialize mon info if we don't have enough mons (at first startup)
existingCount := len(c.clusterInfo.Monitors)
for i := len(c.clusterInfo.Monitors); i < size; i++ {
existingCount := len(c.ClusterInfo.Monitors)
for i := len(c.ClusterInfo.Monitors); i < size; i++ {
c.maxMonID++
mons = append(mons, c.newMonConfig(c.maxMonID))
}
@@ -357,10 +359,10 @@ func (c *Cluster) newMonConfig(monID int) *monConfig {
// resourceName ensures the mon name has the rook-ceph-mon prefix
func resourceName(name string) string {
if strings.HasPrefix(name, appName) {
if strings.HasPrefix(name, AppName) {
return name
}
return fmt.Sprintf("%s-%s", appName, name)
return fmt.Sprintf("%s-%s", AppName, name)
}
func (c *Cluster) initMonIPs(mons []*monConfig) error {
@@ -379,7 +381,7 @@ func (c *Cluster) initMonIPs(mons []*monConfig) error {
}
m.PublicIP = serviceIP
}
c.clusterInfo.Monitors[m.DaemonName] = cephconfig.NewMonInfo(m.DaemonName, m.PublicIP, m.Port)
c.ClusterInfo.Monitors[m.DaemonName] = cephconfig.NewMonInfo(m.DaemonName, m.PublicIP, m.Port)
}
return nil
@@ -537,7 +539,7 @@ func (c *Cluster) startDeployments(mons []*monConfig, requireAllInQuorum bool) e
// Final verification that **all** mons are in quorum
// Do not proceed if one monitor is still syncing
// Only do this when monitors versions are different so we don't block the orchestration if a mon is down.
versions, err := client.GetAllCephDaemonVersions(c.context, c.clusterInfo.Name)
versions, err := client.GetAllCephDaemonVersions(c.context, c.ClusterInfo.Name)
if err != nil {
logger.Warningf("failed to get ceph daemons versions; this likely means there is no cluster yet. %+v", err)
} else {
@@ -560,7 +562,7 @@ func (c *Cluster) waitForMonsToJoin(mons []*monConfig, requireAllInQuorum bool)
// wait for the monitors to join quorum
sleepTime := 5
err := waitForQuorumWithMons(c.context, c.clusterInfo.Name, starting, sleepTime, requireAllInQuorum)
err := waitForQuorumWithMons(c.context, c.ClusterInfo.Name, starting, sleepTime, requireAllInQuorum)
if err != nil {
return fmt.Errorf("failed to wait for mon quorum. %+v", err)
}
@@ -583,13 +585,13 @@ func (c *Cluster) saveMonConfig() error {
}
csiConfigValue, err := csi.FormatCsiClusterConfig(
c.Namespace, c.clusterInfo.Monitors)
c.Namespace, c.ClusterInfo.Monitors)
if err != nil {
return fmt.Errorf("failed to format csi config: %+v", err)
}
configMap.Data = map[string]string{
EndpointDataKey: FlattenMonEndpoints(c.clusterInfo.Monitors),
EndpointDataKey: FlattenMonEndpoints(c.ClusterInfo.Monitors),
MaxMonIDKey: strconv.Itoa(c.maxMonID),
MappingKey: string(monMapping),
csi.ConfigKey: csiConfigValue,
@@ -610,14 +612,14 @@ func (c *Cluster) saveMonConfig() error {
// Every time the mon config is updated, must also update the global config so that all daemons
// have the most updated version if they restart.
config.GetStore(c.context, c.Namespace, &c.ownerRef).CreateOrUpdate(c.clusterInfo)
config.GetStore(c.context, c.Namespace, &c.ownerRef).CreateOrUpdate(c.ClusterInfo)
// write the latest config to the config dir
if err := WriteConnectionConfig(c.context, c.clusterInfo); err != nil {
if err := WriteConnectionConfig(c.context, c.ClusterInfo); err != nil {
return fmt.Errorf("failed to write connection config for new mons. %+v", err)
}
if err := csi.SaveClusterConfig(c.context.Clientset, c.Namespace, c.clusterInfo, c.csiConfigMutex); err != nil {
if err := csi.SaveClusterConfig(c.context.Clientset, c.Namespace, c.ClusterInfo, c.csiConfigMutex); err != nil {
return fmt.Errorf("failed to update csi cluster config: %+v", err)
}
@@ -632,11 +634,11 @@ func (c *Cluster) updateMon(m *monConfig, d *apps.Deployment) error {
daemonType := string(config.MonType)
var cephVersionToUse cephver.CephVersion
currentCephVersion, err := client.LeastUptodateDaemonVersion(c.context, c.clusterInfo.Name, daemonType)
currentCephVersion, err := client.LeastUptodateDaemonVersion(c.context, c.ClusterInfo.Name, daemonType)
if err != nil {
logger.Warningf("failed to retrieve current ceph %s version. %+v", daemonType, err)
logger.Debug("could not detect ceph version during update, this is likely an initial bootstrap, proceeding with c.clusterInfo.CephVersion")
cephVersionToUse = c.clusterInfo.CephVersion
cephVersionToUse = c.ClusterInfo.CephVersion
} else {
logger.Debugf("current cluster version for monitors before upgrading is: %+v", currentCephVersion)
cephVersionToUse = currentCephVersion
@@ -730,7 +732,7 @@ func waitForQuorumWithMons(context *clusterd.Context, clusterName string, mons [
allPodsRunning := true
var runningMonNames []string
for _, m := range mons {
running, err := k8sutil.PodsRunningWithLabel(context.Clientset, clusterName, fmt.Sprintf("app=%s,mon=%s", appName, m))
running, err := k8sutil.PodsRunningWithLabel(context.Clientset, clusterName, fmt.Sprintf("app=%s,mon=%s", AppName, m))
if err != nil {
logger.Infof("failed to query mon pod status, trying again. %+v", err)
continue
+12 -12
View File
@@ -103,7 +103,7 @@ func newTestStartClusterWithQuorumResponse(namespace string, monResponse func()
func newCluster(context *clusterd.Context, namespace string, hostNetwork bool, allowMultiplePerNode bool, resources v1.ResourceRequirements) *Cluster {
return &Cluster{
clusterInfo: nil,
ClusterInfo: nil,
HostNetwork: hostNetwork,
context: context,
Namespace: namespace,
@@ -130,7 +130,7 @@ func newCluster(context *clusterd.Context, namespace string, hostNetwork bool, a
// setCommonMonProperties is a convenience helper for setting common test properties
func setCommonMonProperties(c *Cluster, currentMons int, mon cephv1.MonSpec, rookVersion string) {
c.clusterInfo = test.CreateConfigDir(currentMons)
c.ClusterInfo = test.CreateConfigDir(currentMons)
c.spec.Mon.Count = mon.Count
c.spec.Mon.AllowMultiplePerNode = mon.AllowMultiplePerNode
c.rookVersion = rookVersion
@@ -149,13 +149,13 @@ func TestStartMonPods(t *testing.T) {
c := newCluster(context, namespace, false, true, v1.ResourceRequirements{})
// start a basic cluster
_, err := c.Start(c.clusterInfo, c.rookVersion, cephver.Mimic, c.spec)
_, err := c.Start(c.ClusterInfo, c.rookVersion, cephver.Mimic, c.spec)
assert.Nil(t, err)
validateStart(t, c)
// starting again should be a no-op, but still results in an error
_, err = c.Start(c.clusterInfo, c.rookVersion, cephver.Mimic, c.spec)
_, err = c.Start(c.ClusterInfo, c.rookVersion, cephver.Mimic, c.spec)
assert.Nil(t, err)
validateStart(t, c)
@@ -166,10 +166,10 @@ func TestOperatorRestart(t *testing.T) {
namespace := "ns"
context := newTestStartCluster(namespace)
c := newCluster(context, namespace, false, true, v1.ResourceRequirements{})
c.clusterInfo = test.CreateConfigDir(1)
c.ClusterInfo = test.CreateConfigDir(1)
// start a basic cluster
info, err := c.Start(c.clusterInfo, c.rookVersion, cephver.Mimic, c.spec)
info, err := c.Start(c.ClusterInfo, c.rookVersion, cephver.Mimic, c.spec)
assert.Nil(t, err)
assert.True(t, info.IsInitialized())
@@ -178,7 +178,7 @@ func TestOperatorRestart(t *testing.T) {
c = newCluster(context, namespace, false, true, v1.ResourceRequirements{})
// starting again should be a no-op, but will not result in an error
info, err = c.Start(c.clusterInfo, c.rookVersion, cephver.Mimic, c.spec)
info, err = c.Start(c.ClusterInfo, c.rookVersion, cephver.Mimic, c.spec)
assert.Nil(t, err)
assert.True(t, info.IsInitialized())
@@ -193,10 +193,10 @@ func TestOperatorRestartHostNetwork(t *testing.T) {
// cluster without host networking
c := newCluster(context, namespace, false, false, v1.ResourceRequirements{})
c.clusterInfo = test.CreateConfigDir(1)
c.ClusterInfo = test.CreateConfigDir(1)
// start a basic cluster
info, err := c.Start(c.clusterInfo, c.rookVersion, cephver.Mimic, c.spec)
info, err := c.Start(c.ClusterInfo, c.rookVersion, cephver.Mimic, c.spec)
assert.Nil(t, err)
assert.True(t, info.IsInitialized())
@@ -206,7 +206,7 @@ func TestOperatorRestartHostNetwork(t *testing.T) {
c = newCluster(context, namespace, true, false, v1.ResourceRequirements{})
// starting again should be a no-op, but still results in an error
info, err = c.Start(c.clusterInfo, c.rookVersion, cephver.Mimic, c.spec)
info, err = c.Start(c.ClusterInfo, c.rookVersion, cephver.Mimic, c.spec)
assert.Nil(t, err)
assert.True(t, info.IsInitialized(), info)
@@ -214,7 +214,7 @@ func TestOperatorRestartHostNetwork(t *testing.T) {
}
func validateStart(t *testing.T, c *Cluster) {
s, err := c.context.Clientset.CoreV1().Secrets(c.Namespace).Get(appName, metav1.GetOptions{})
s, err := c.context.Clientset.CoreV1().Secrets(c.Namespace).Get(AppName, metav1.GetOptions{})
assert.NoError(t, err) // there shouldn't be an error due the secret existing
assert.Equal(t, 4, len(s.Data))
@@ -245,7 +245,7 @@ func TestSaveMonEndpoints(t *testing.T) {
assert.Equal(t, "-1", cm.Data[MaxMonIDKey])
// update the config map
c.clusterInfo.Monitors["a"].Endpoint = "2.3.4.5:6789"
c.ClusterInfo.Monitors["a"].Endpoint = "2.3.4.5:6789"
c.maxMonID = 2
c.mapping.Node["a"] = &NodeInfo{
Name: "node0",
+79 -2
View File
@@ -21,7 +21,8 @@ import (
cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
"github.com/rook/rook/pkg/operator/k8sutil"
"k8s.io/api/core/v1"
"github.com/rook/rook/pkg/util"
v1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)
@@ -36,6 +37,82 @@ type NodeUsage struct {
MonValid bool
}
// getMonNodes detects the nodes that are available for new mons to start.
func (c *Cluster) getMonNodes() ([]v1.Node, error) {
availableNodes, nodes, err := c.getAvailableMonNodes()
if err != nil {
return nil, err
}
logger.Infof("Found %d running nodes without mons", len(availableNodes))
// if all nodes already have mons and the user has given the mon.count, add all nodes to be available
if c.spec.Mon.AllowMultiplePerNode && len(availableNodes) == 0 {
logger.Infof("All nodes are running mons. Adding all %d nodes to the availability.", len(nodes.Items))
for _, node := range nodes.Items {
valid, err := k8sutil.ValidNode(node, cephv1.GetMonPlacement(c.spec.Placement))
if err != nil {
logger.Warning("failed to validate node %s %v", node.Name, err)
} else if valid {
availableNodes = append(availableNodes, node)
}
}
}
return availableNodes, nil
}
func (c *Cluster) getAvailableMonNodes() ([]v1.Node, *v1.NodeList, error) {
nodeOptions := metav1.ListOptions{}
nodeOptions.TypeMeta.Kind = "Node"
nodes, err := c.context.Clientset.CoreV1().Nodes().List(nodeOptions)
if err != nil {
return nil, nil, err
}
logger.Debugf("there are %d nodes available for %d mons", len(nodes.Items), len(c.ClusterInfo.Monitors))
// get the nodes that have mons assigned
nodesInUse, err := c.getNodesWithMons(nodes)
if err != nil {
logger.Warningf("could not get nodes with mons. %+v", err)
nodesInUse = util.NewSet()
}
// choose nodes for the new mons that don't have mons currently
availableNodes := []v1.Node{}
for _, node := range nodes.Items {
if !nodesInUse.Contains(node.Name) {
valid, err := k8sutil.ValidNode(node, cephv1.GetMonPlacement(c.spec.Placement))
if err != nil {
logger.Warning("failed to validate node %s %v", node.Name, err)
} else if valid {
availableNodes = append(availableNodes, node)
}
}
}
return availableNodes, nodes, nil
}
func (c *Cluster) getNodesWithMons(nodes *v1.NodeList) (*util.Set, error) {
// get the mon pods and their node affinity
options := metav1.ListOptions{LabelSelector: fmt.Sprintf("app=%s", AppName)}
pods, err := c.context.Clientset.CoreV1().Pods(c.Namespace).List(options)
if err != nil {
return nil, err
}
nodesInUse := util.NewSet()
for _, pod := range pods.Items {
hostname := pod.Spec.NodeSelector[v1.LabelHostname]
logger.Debugf("mon pod on node %s", hostname)
name, ok := getNodeNameFromHostname(nodes, hostname)
if !ok {
logger.Errorf("mon %s on hostname %s not found in node list", pod.Name, hostname)
}
nodesInUse.Add(name)
}
return nodesInUse, nil
}
// Get the number of mons that the operator should be starting
func (c *Cluster) getTargetMonCount() (int, string, error) {
@@ -140,7 +217,7 @@ func (c *Cluster) getNodeMonUsage() ([][]NodeUsage, error) {
}
// get all pod objects labeled as a monitor
podOptions := metav1.ListOptions{LabelSelector: fmt.Sprintf("app=%s", appName)}
podOptions := metav1.ListOptions{LabelSelector: fmt.Sprintf("app=%s", AppName)}
pods, err := c.context.Clientset.CoreV1().Pods(c.Namespace).List(podOptions)
if err != nil {
return nil, err
+12 -12
View File
@@ -193,10 +193,10 @@ func TestHostNetworkSameNode(t *testing.T) {
// cluster host networking
c := newCluster(context, namespace, true, true, v1.ResourceRequirements{})
c.clusterInfo = test.CreateConfigDir(1)
c.ClusterInfo = test.CreateConfigDir(1)
// start a basic cluster
_, err := c.Start(c.clusterInfo, c.rookVersion, cephver.Mimic, c.spec)
_, err := c.Start(c.ClusterInfo, c.rookVersion, cephver.Mimic, c.spec)
assert.Error(t, err)
}
@@ -212,9 +212,9 @@ func TestPodMemory(t *testing.T) {
}
c := newCluster(context, namespace, false, true, r)
c.clusterInfo = test.CreateConfigDir(1)
c.ClusterInfo = test.CreateConfigDir(1)
// start a basic cluster
_, err := c.Start(c.clusterInfo, c.rookVersion, cephver.Mimic, c.spec)
_, err := c.Start(c.ClusterInfo, c.rookVersion, cephver.Mimic, c.spec)
assert.Error(t, err)
// Test REQUEST == LIMIT
@@ -228,9 +228,9 @@ func TestPodMemory(t *testing.T) {
}
c = newCluster(context, namespace, false, true, r)
c.clusterInfo = test.CreateConfigDir(1)
c.ClusterInfo = test.CreateConfigDir(1)
// start a basic cluster
_, err = c.Start(c.clusterInfo, c.rookVersion, cephver.Mimic, c.spec)
_, err = c.Start(c.ClusterInfo, c.rookVersion, cephver.Mimic, c.spec)
assert.Error(t, err)
// Test LIMIT != REQUEST but obviously LIMIT > REQUEST
@@ -244,9 +244,9 @@ func TestPodMemory(t *testing.T) {
}
c = newCluster(context, namespace, false, true, r)
c.clusterInfo = test.CreateConfigDir(1)
c.ClusterInfo = test.CreateConfigDir(1)
// start a basic cluster
_, err = c.Start(c.clusterInfo, c.rookVersion, cephver.Mimic, c.spec)
_, err = c.Start(c.ClusterInfo, c.rookVersion, cephver.Mimic, c.spec)
assert.Error(t, err)
// Test valid case where pod resource is set approprietly
@@ -260,17 +260,17 @@ func TestPodMemory(t *testing.T) {
}
c = newCluster(context, namespace, false, true, r)
c.clusterInfo = test.CreateConfigDir(1)
c.ClusterInfo = test.CreateConfigDir(1)
// start a basic cluster
_, err = c.Start(c.clusterInfo, c.rookVersion, cephver.Mimic, c.spec)
_, err = c.Start(c.ClusterInfo, c.rookVersion, cephver.Mimic, c.spec)
assert.Nil(t, err)
// Test no resources were specified on the pod
r = v1.ResourceRequirements{}
c = newCluster(context, namespace, false, true, r)
c.clusterInfo = test.CreateConfigDir(1)
c.ClusterInfo = test.CreateConfigDir(1)
// start a basic cluster
_, err = c.Start(c.clusterInfo, c.rookVersion, cephver.Mimic, c.spec)
_, err = c.Start(c.ClusterInfo, c.rookVersion, cephver.Mimic, c.spec)
assert.Nil(t, err)
}
+2 -2
View File
@@ -53,7 +53,7 @@ func (c *Cluster) createService(mon *monConfig) (string, error) {
}
// If deploying Nautilus or newer we need a new port for the monitor service
if c.clusterInfo.CephVersion.IsAtLeastNautilus() {
if c.ClusterInfo.CephVersion.IsAtLeastNautilus() {
addServicePort(svcDef, "msgr2", DefaultMsgr2Port)
}
@@ -70,7 +70,7 @@ func (c *Cluster) createService(mon *monConfig) (string, error) {
// mon endpoint are not actually like, they remain with the mgrs1 format
// however it's interesting to show that monitors can be addressed via 2 different ports
// in the end the service has msgr1 and msgr2 ports configured so it's not entirely wrong
if c.clusterInfo.CephVersion.IsAtLeastNautilus() {
if c.ClusterInfo.CephVersion.IsAtLeastNautilus() {
logger.Infof("mon %s endpoint are [v2:%s:%s,v1:%s:%d]", mon.DaemonName, s.Spec.ClusterIP, strconv.Itoa(int(DefaultMsgr2Port)), s.Spec.ClusterIP, mon.Port)
} else {
logger.Infof("mon %s endpoint is %s:%d", mon.DaemonName, s.Spec.ClusterIP, mon.Port)
+6 -6
View File
@@ -46,7 +46,7 @@ const (
func (c *Cluster) getLabels(daemonName string) map[string]string {
// Mons have a service for each mon, so the additional pod data is relevant for its services
// Use pod labels to keep "mon: id" for legacy
labels := opspec.PodLabels(appName, c.Namespace, "mon", daemonName)
labels := opspec.PodLabels(AppName, c.Namespace, "mon", daemonName)
// Add "mon_cluster: <namespace>" for legacy
labels[monClusterAttr] = c.Namespace
return labels
@@ -62,7 +62,7 @@ func (c *Cluster) makeDeployment(monConfig *monConfig, hostname string) *apps.De
}
k8sutil.AddRookVersionLabelToDeployment(d)
cephv1.GetMonAnnotations(c.spec.Annotations).ApplyToObjectMeta(&d.ObjectMeta)
opspec.AddCephVersionLabelToDeployment(c.clusterInfo.CephVersion, d)
opspec.AddCephVersionLabelToDeployment(c.ClusterInfo.CephVersion, d)
k8sutil.SetOwnerRef(&d.ObjectMeta, &c.ownerRef)
pod := c.makeMonPod(monConfig, hostname)
@@ -104,7 +104,7 @@ func (c *Cluster) makeDeploymentPVC(m *monConfig) (*v1.PersistentVolumeClaim, er
}
k8sutil.AddRookVersionLabelToObjectMeta(&pvc.ObjectMeta)
cephv1.GetMonAnnotations(c.spec.Annotations).ApplyToObjectMeta(&pvc.ObjectMeta)
opspec.AddCephVersionLabelToObjectMeta(c.clusterInfo.CephVersion, &pvc.ObjectMeta)
opspec.AddCephVersionLabelToObjectMeta(c.ClusterInfo.CephVersion, &pvc.ObjectMeta)
k8sutil.SetOwnerRef(&pvc.ObjectMeta, &c.ownerRef)
// k8s uses limit as the resource request fallback
@@ -224,7 +224,7 @@ func (c *Cluster) makeMonFSInitContainer(monConfig *monConfig) v1.Container {
cephMonCommand,
},
Args: append(
opspec.DaemonFlags(c.clusterInfo, monConfig.DaemonName),
opspec.DaemonFlags(c.ClusterInfo, monConfig.DaemonName),
// needed so we can generate an initial monmap
// otherwise the mkfs will say: "0  no local addrs match monmap"
config.NewFlag("public-addr", monConfig.PublicIP),
@@ -257,7 +257,7 @@ func (c *Cluster) makeMonDaemonContainer(monConfig *monConfig) v1.Container {
cephMonCommand,
},
Args: append(
opspec.DaemonFlags(c.clusterInfo, monConfig.DaemonName),
opspec.DaemonFlags(c.ClusterInfo, monConfig.DaemonName),
"--foreground",
// If the mon is already in the monmap, when the port is left off of --public-addr,
// it will still advertise on the previous port b/c monmap is saved to mon database.
@@ -299,7 +299,7 @@ func (c *Cluster) makeMonDaemonContainer(monConfig *monConfig) v1.Container {
}
// If deploying Nautilus and newer we need a new port of the monitor container
if c.clusterInfo.CephVersion.IsAtLeastNautilus() {
if c.ClusterInfo.CephVersion.IsAtLeastNautilus() {
addContainerPort(container, "msgr2", 3300)
}
+2 -2
View File
@@ -77,10 +77,10 @@ func testPodSpec(t *testing.T, monID string, pvc bool) {
// Deployment should have Ceph labels
cephtest.AssertLabelsContainCephRequirements(t, d.ObjectMeta.Labels,
config.MonType, monID, appName, "ns")
config.MonType, monID, AppName, "ns")
podTemplate := cephtest.NewPodTemplateSpecTester(t, &d.Spec.Template)
podTemplate.RunFullSuite(config.MonType, monID, appName, "ns", "ceph/ceph:myceph",
podTemplate.RunFullSuite(config.MonType, monID, AppName, "ns", "ceph/ceph:myceph",
"200", "100", "1337", "500" /* resources */)
}
+3 -3
View File
@@ -42,14 +42,14 @@ func monInQuorum(monitor client.MonMapEntry, quorum []int) bool {
// convert the mon name to the numeric mon ID
func fullNameToIndex(name string) (int, error) {
prefix := appName + "-"
prefix := AppName + "-"
if strings.Index(name, prefix) != -1 && len(prefix) < len(name) {
return k8sutil.NameToIndex(name[len(prefix)+1:])
}
// attempt to parse the legacy mon name
legacyPrefix := appName
if strings.Index(name, legacyPrefix) == -1 || len(name) < len(appName) {
legacyPrefix := AppName
if strings.Index(name, legacyPrefix) == -1 || len(name) < len(AppName) {
return -1, fmt.Errorf("unexpected mon name")
}
id, err := strconv.Atoi(name[len(legacyPrefix):])
+11 -9
View File
@@ -36,7 +36,8 @@ import (
)
const (
storeName = "rook-ceph-config"
// StoreName is the name of the configmap containing ceph configuration options
StoreName = "rook-ceph-config"
configVolumeName = "rook-ceph-config"
@@ -97,7 +98,7 @@ func (s *Store) CreateOrUpdate(clusterInfo *cephconfig.ClusterInfo) error {
txt := b.String()
// Store the config in a configmap
if err := s.configMapStore.SetValue(storeName, confFileName, txt); err != nil {
if err := s.configMapStore.SetValue(StoreName, confFileName, txt); err != nil {
return fmt.Errorf("failed to store the Ceph config file. failed to store config to configmap. %+v", err)
}
logger.Debugf("Generated and stored config file:\n%s", txt)
@@ -183,7 +184,7 @@ func (s *Store) createOrUpdateMonHostSecrets(clusterInfo *cephconfig.ClusterInfo
secret := &v1.Secret{
ObjectMeta: metav1.ObjectMeta{
// the config's secret store has the same name as the configmap store for consistency
Name: storeName,
Name: StoreName,
Namespace: s.namespace,
},
StringData: map[string]string{
@@ -195,15 +196,16 @@ func (s *Store) createOrUpdateMonHostSecrets(clusterInfo *cephconfig.ClusterInfo
clientset := s.context.Clientset
k8sutil.SetOwnerRef(&secret.ObjectMeta, s.ownerRef)
_, err := clientset.CoreV1().Secrets(s.namespace).Get(storeName, metav1.GetOptions{})
_, err := clientset.CoreV1().Secrets(s.namespace).Get(StoreName, metav1.GetOptions{})
if err != nil {
if errors.IsNotFound(err) {
logger.Debugf("creating config secret %+v", secret)
if _, err := clientset.CoreV1().Secrets(s.namespace).Create(secret); err != nil {
return fmt.Errorf("failed to create config secret %+v. %+v", secret, err)
}
} else {
return fmt.Errorf("failed to get config secret %s. %+v", StoreName, err)
}
return fmt.Errorf("failed to get config secret %s. %+v", storeName, err)
}
logger.Debugf("updating config secret %+v", secret)
@@ -231,7 +233,7 @@ func StoredFileVolume() v1.Volume {
Name: configVolumeName,
VolumeSource: v1.VolumeSource{
ConfigMap: &v1.ConfigMapVolumeSource{LocalObjectReference: v1.LocalObjectReference{
Name: storeName},
Name: StoreName},
Items: []v1.KeyToPath{
{Key: confFileName, Path: confFileName, Mode: &mode},
}}}}
@@ -242,7 +244,7 @@ func StoredFileVolume() v1.Volume {
func StoredFileVolumeMount() v1.VolumeMount {
// configmap's "ceph.conf" to "/etc/ceph/ceph.conf"
return v1.VolumeMount{
Name: storeName,
Name: StoreName,
ReadOnly: true, // should be no reason to write to the config in pods, so enforce this
MountPath: "/etc/ceph",
}
@@ -255,12 +257,12 @@ func StoredMonHostEnvVars() []v1.EnvVar {
{Name: "ROOK_CEPH_MON_HOST",
ValueFrom: &v1.EnvVarSource{
SecretKeyRef: &v1.SecretKeySelector{LocalObjectReference: v1.LocalObjectReference{
Name: storeName},
Name: StoreName},
Key: monHostKey}}},
{Name: "ROOK_CEPH_MON_INITIAL_MEMBERS",
ValueFrom: &v1.EnvVarSource{
SecretKeyRef: &v1.SecretKeySelector{LocalObjectReference: v1.LocalObjectReference{
Name: storeName},
Name: StoreName},
Key: monInitialMembersKey}}},
}
}
+6 -6
View File
@@ -29,7 +29,7 @@ import (
"github.com/rook/rook/pkg/operator/k8sutil"
testop "github.com/rook/rook/pkg/operator/test"
"github.com/stretchr/testify/assert"
"k8s.io/api/core/v1"
v1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)
@@ -46,14 +46,14 @@ func TestStore(t *testing.T) {
var previousConfigText = ""
var recentConfigText = ""
assertConfigStore := func(ci *cephconfig.ClusterInfo) {
c, e := clientset.CoreV1().ConfigMaps(ns).Get(storeName, metav1.GetOptions{})
c, e := clientset.CoreV1().ConfigMaps(ns).Get(StoreName, metav1.GetOptions{})
assert.NoError(t, e)
previousConfigText = recentConfigText
recentConfigText = c.Data[confFileName]
assert.NotEqual(t, "", recentConfigText)
sec, e := clientset.CoreV1().Secrets(ns).Get(storeName, metav1.GetOptions{})
sec, e := clientset.CoreV1().Secrets(ns).Get(StoreName, metav1.GetOptions{})
assert.NoError(t, e)
mh := strings.Split(sec.StringData["mon_host"], ",") // list of mon ip:port pairs in cluster
assert.Equal(t, len(ci.Monitors), len(mh))
@@ -188,7 +188,7 @@ func TestFileVolumeAndMount(t *testing.T) {
// Test that the configmapped file will make it into containers with the appropriate filename at
// the location where it is expected.
assert.Equal(t, v.Name, m.Name)
assert.Equal(t, storeName, v.VolumeSource.ConfigMap.LocalObjectReference.Name)
assert.Equal(t, StoreName, v.VolumeSource.ConfigMap.LocalObjectReference.Name)
assert.Equal(t, "/etc/ceph/ceph.conf", path.Join(m.MountPath, confFileName))
}
@@ -213,8 +213,8 @@ func TestEnvVarsAndFlags(t *testing.T) {
assert.Contains(t, f, fmt.Sprintf("--mon-initial-members=$(%s)", mim))
// make sure the env vars are sourced from the right place
assert.Equal(t, storeName, v[0].ValueFrom.SecretKeyRef.LocalObjectReference.Name)
assert.Equal(t, StoreName, v[0].ValueFrom.SecretKeyRef.LocalObjectReference.Name)
assert.Equal(t, "mon_host", v[0].ValueFrom.SecretKeyRef.Key)
assert.Equal(t, storeName, v[1].ValueFrom.SecretKeyRef.LocalObjectReference.Name)
assert.Equal(t, StoreName, v[1].ValueFrom.SecretKeyRef.LocalObjectReference.Name)
assert.Equal(t, "mon_initial_members", v[1].ValueFrom.SecretKeyRef.Key)
}
+1 -1
View File
@@ -163,7 +163,7 @@ func SaveClusterConfig(
return fmt.Errorf("namespace value missing (for %+v)",
k8sutil.PodNamespaceEnvVar)
}
logger.Debug("Using %+v for CSI ConfigMap Namespace", csiNamespace)
logger.Debugf("Using %+v for CSI ConfigMap Namespace", csiNamespace)
// fetch current ConfigMap contents
configMap, err := clientset.CoreV1().ConfigMaps(csiNamespace).Get(
+12 -15
View File
@@ -50,8 +50,7 @@ type FilesystemController struct {
context *clusterd.Context
namespace string
rookVersion string
cephVersion cephv1.CephVersionSpec
hostNetwork bool
clusterSpec *cephv1.ClusterSpec
ownerRef metav1.OwnerReference
dataDirHostPath string
orchestrationMutex sync.Mutex
@@ -63,8 +62,7 @@ func NewFilesystemController(
context *clusterd.Context,
namespace string,
rookVersion string,
cephVersion cephv1.CephVersionSpec,
hostNetwork bool,
clusterSpec *cephv1.ClusterSpec,
ownerRef metav1.OwnerReference,
dataDirHostPath string,
) *FilesystemController {
@@ -73,15 +71,14 @@ func NewFilesystemController(
context: context,
namespace: namespace,
rookVersion: rookVersion,
cephVersion: cephVersion,
hostNetwork: hostNetwork,
clusterSpec: clusterSpec,
ownerRef: ownerRef,
dataDirHostPath: dataDirHostPath,
}
}
// StartWatch watches for instances of Filesystem custom resources and acts on them
func (c *FilesystemController) StartWatch(stopCh chan struct{}) error {
func (c *FilesystemController) StartWatch(namespace string, stopCh chan struct{}) error {
resourceHandlerFuncs := cache.ResourceEventHandlerFuncs{
AddFunc: c.onAdd,
@@ -90,7 +87,7 @@ func (c *FilesystemController) StartWatch(stopCh chan struct{}) error {
}
logger.Infof("start watching filesystem resource in namespace %s", c.namespace)
watcher := opkit.NewWatcher(FilesystemResource, c.namespace, resourceHandlerFuncs, c.context.RookClientset.CephV1().RESTClient())
watcher := opkit.NewWatcher(FilesystemResource, namespace, resourceHandlerFuncs, c.context.RookClientset.CephV1().RESTClient())
go watcher.Watch(&cephv1.CephFilesystem{}, stopCh)
return nil
}
@@ -105,7 +102,7 @@ func (c *FilesystemController) onAdd(obj interface{}) {
c.acquireOrchestrationLock()
defer c.releaseOrchestrationLock()
err = createFilesystem(c.clusterInfo, c.context, *filesystem, c.rookVersion, c.cephVersion, c.hostNetwork, c.filesystemOwners(filesystem), c.dataDirHostPath)
err = createFilesystem(c.clusterInfo, c.context, *filesystem, c.rookVersion, c.clusterSpec, c.filesystemOwners(filesystem), c.clusterSpec.DataDirHostPath)
if err != nil {
logger.Errorf("failed to create filesystem %s: %+v", filesystem.Name, err)
}
@@ -133,7 +130,7 @@ func (c *FilesystemController) onUpdate(oldObj, newObj interface{}) {
// if the filesystem is modified, allow the filesystem to be created if it wasn't already
logger.Infof("updating filesystem %s", newFS.Name)
err = createFilesystem(c.clusterInfo, c.context, *newFS, c.rookVersion, c.cephVersion, c.hostNetwork, c.filesystemOwners(newFS), c.dataDirHostPath)
err = createFilesystem(c.clusterInfo, c.context, *newFS, c.rookVersion, c.clusterSpec, c.filesystemOwners(newFS), c.clusterSpec.DataDirHostPath)
if err != nil {
logger.Errorf("failed to create (modify) filesystem %s: %+v", newFS.Name, err)
}
@@ -141,7 +138,7 @@ func (c *FilesystemController) onUpdate(oldObj, newObj interface{}) {
func (c *FilesystemController) ParentClusterChanged(cluster cephv1.ClusterSpec, clusterInfo *cephconfig.ClusterInfo) {
c.clusterInfo = clusterInfo
if cluster.CephVersion.Image == c.cephVersion.Image {
if cluster.CephVersion.Image == c.clusterSpec.CephVersion.Image {
logger.Debugf("No need to update the file system after the parent cluster changed")
return
}
@@ -149,19 +146,19 @@ func (c *FilesystemController) ParentClusterChanged(cluster cephv1.ClusterSpec,
c.acquireOrchestrationLock()
defer c.releaseOrchestrationLock()
c.cephVersion = cluster.CephVersion
c.clusterSpec.CephVersion = cluster.CephVersion
filesystems, err := c.context.RookClientset.CephV1().CephFilesystems(c.namespace).List(metav1.ListOptions{})
if err != nil {
logger.Errorf("failed to retrieve filesystems to update the ceph version. %+v", err)
return
}
for _, fs := range filesystems.Items {
logger.Infof("updating the ceph version for filesystem %s to %s", fs.Name, c.cephVersion.Image)
err = createFilesystem(c.clusterInfo, c.context, fs, c.rookVersion, c.cephVersion, c.hostNetwork, c.filesystemOwners(&fs), c.dataDirHostPath)
logger.Infof("updating the ceph version for filesystem %s to %s", fs.Name, c.clusterSpec.CephVersion.Image)
err = createFilesystem(c.clusterInfo, c.context, fs, c.rookVersion, c.clusterSpec, c.filesystemOwners(&fs), c.clusterSpec.DataDirHostPath)
if err != nil {
logger.Errorf("failed to update filesystem %s. %+v", fs.Name, err)
} else {
logger.Infof("updated filesystem %s to ceph version %s", fs.Name, c.cephVersion.Image)
logger.Infof("updated filesystem %s to ceph version %s", fs.Name, c.clusterSpec.CephVersion.Image)
}
}
}
+2 -3
View File
@@ -50,8 +50,7 @@ func createFilesystem(
context *clusterd.Context,
fs cephv1.CephFilesystem,
rookVersion string,
cephVersion cephv1.CephVersionSpec,
hostNetwork bool,
clusterSpec *cephv1.ClusterSpec,
ownerRefs []metav1.OwnerReference,
dataDirHostPath string,
) error {
@@ -92,7 +91,7 @@ func createFilesystem(
}
logger.Infof("start running mdses for filesystem %s", fs.Name)
c := mds.NewCluster(clusterInfo, context, rookVersion, cephVersion, hostNetwork, fs, filesystem, ownerRefs, dataDirHostPath)
c := mds.NewCluster(clusterInfo, context, rookVersion, clusterSpec, fs, filesystem, ownerRefs, dataDirHostPath)
if err := c.Start(); err != nil {
return err
}
+5 -5
View File
@@ -114,14 +114,14 @@ func TestCreateFilesystem(t *testing.T) {
clusterInfo := &cephconfig.ClusterInfo{FSID: "myfsid"}
// start a basic cluster
err := createFilesystem(clusterInfo, context, fs, "v0.1", cephv1.CephVersionSpec{}, false, []metav1.OwnerReference{}, "/var/lib/rook/")
err := createFilesystem(clusterInfo, context, fs, "v0.1", &cephv1.ClusterSpec{}, []metav1.OwnerReference{}, "/var/lib/rook/")
assert.Nil(t, err)
validateStart(t, context, fs)
assert.ElementsMatch(t, []string{}, testopk8s.DeploymentNamesUpdated(deploymentsUpdated))
testopk8s.ClearDeploymentsUpdated(deploymentsUpdated)
// starting again should be a no-op
err = createFilesystem(clusterInfo, context, fs, "v0.1", cephv1.CephVersionSpec{}, false, []metav1.OwnerReference{}, "/var/lib/rook/")
err = createFilesystem(clusterInfo, context, fs, "v0.1", &cephv1.ClusterSpec{}, []metav1.OwnerReference{}, "/var/lib/rook/")
assert.Nil(t, err)
validateStart(t, context, fs)
assert.ElementsMatch(t, []string{"rook-ceph-mds-myfs-a", "rook-ceph-mds-myfs-b"}, testopk8s.DeploymentNamesUpdated(deploymentsUpdated))
@@ -142,7 +142,7 @@ func TestCreateFilesystem(t *testing.T) {
Clientset: testop.New(3)}
//Create another filesystem which should fail
err = createFilesystem(clusterInfo, context, fs, "v0.1", cephv1.CephVersionSpec{}, false, []metav1.OwnerReference{}, "/var/lib/rook/")
err = createFilesystem(clusterInfo, context, fs, "v0.1", &cephv1.ClusterSpec{}, []metav1.OwnerReference{}, "/var/lib/rook/")
assert.Equal(t, "failed to create filesystem myfs: Cannot create multiple filesystems. Enable ROOK_ALLOW_MULTIPLE_FILESYSTEMS env variable to create more than one", err.Error())
}
@@ -179,12 +179,12 @@ func TestCreateNopoolFilesystem(t *testing.T) {
clusterInfo := &cephconfig.ClusterInfo{FSID: "myfsid"}
// start a basic cluster
err := createFilesystem(clusterInfo, context, fs, "v0.1", cephv1.CephVersionSpec{}, false, []metav1.OwnerReference{}, "/var/lib/rook/")
err := createFilesystem(clusterInfo, context, fs, "v0.1", &cephv1.ClusterSpec{}, []metav1.OwnerReference{}, "/var/lib/rook/")
assert.Nil(t, err)
validateStart(t, context, fs)
// starting again should be a no-op
err = createFilesystem(clusterInfo, context, fs, "v0.1", cephv1.CephVersionSpec{}, false, []metav1.OwnerReference{}, "/var/lib/rook/")
err = createFilesystem(clusterInfo, context, fs, "v0.1", &cephv1.ClusterSpec{}, []metav1.OwnerReference{}, "/var/lib/rook/")
assert.Nil(t, err)
validateStart(t, context, fs)
+3 -6
View File
@@ -55,8 +55,7 @@ type Cluster struct {
clusterInfo *cephconfig.ClusterInfo
context *clusterd.Context
rookVersion string
cephVersion cephv1.CephVersionSpec
HostNetwork bool
clusterSpec *cephv1.ClusterSpec
fs cephv1.CephFilesystem
fsID string
ownerRefs []metav1.OwnerReference
@@ -74,8 +73,7 @@ func NewCluster(
clusterInfo *cephconfig.ClusterInfo,
context *clusterd.Context,
rookVersion string,
cephVersion cephv1.CephVersionSpec,
hostNetwork bool,
clusterSpec *cephv1.ClusterSpec,
fs cephv1.CephFilesystem,
fsdetails *client.CephFilesystemDetails,
ownerRefs []metav1.OwnerReference,
@@ -85,8 +83,7 @@ func NewCluster(
clusterInfo: clusterInfo,
context: context,
rookVersion: rookVersion,
cephVersion: cephVersion,
HostNetwork: hostNetwork,
clusterSpec: clusterSpec,
fs: fs,
fsID: strconv.Itoa(fsdetails.ID),
ownerRefs: ownerRefs,
+4 -4
View File
@@ -51,10 +51,10 @@ func (c *Cluster) makeDeployment(mdsConfig *mdsConfig) *apps.Deployment {
},
RestartPolicy: v1.RestartPolicyAlways,
Volumes: opspec.DaemonVolumes(mdsConfig.DataPathMap, mdsConfig.ResourceName),
HostNetwork: c.HostNetwork,
HostNetwork: c.clusterSpec.Network.HostNetwork,
},
}
if c.HostNetwork {
if c.clusterSpec.Network.HostNetwork {
podSpec.Spec.DNSPolicy = v1.DNSClusterFirstWithHostNet
}
c.fs.Spec.MetadataServer.Annotations.ApplyToObjectMeta(&podSpec.ObjectMeta)
@@ -111,10 +111,10 @@ func (c *Cluster) makeMdsDaemonContainer(mdsConfig *mdsConfig) v1.Container {
"ceph-mds",
},
Args: args,
Image: c.cephVersion.Image,
Image: c.clusterSpec.CephVersion.Image,
VolumeMounts: opspec.DaemonVolumeMounts(mdsConfig.DataPathMap, mdsConfig.ResourceName),
Env: append(
opspec.DaemonEnvVars(c.cephVersion.Image),
opspec.DaemonEnvVars(c.clusterSpec.CephVersion.Image),
),
Resources: c.fs.Spec.MetadataServer.Resources,
Lifecycle: opspec.PodLifeCycle(""),
+7 -2
View File
@@ -22,6 +22,7 @@ import (
"github.com/rook/rook/pkg/operator/ceph/config"
cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
rookalpha "github.com/rook/rook/pkg/apis/rook.io/v1alpha2"
"github.com/rook/rook/pkg/clusterd"
"github.com/rook/rook/pkg/daemon/ceph/client"
cephconfig "github.com/rook/rook/pkg/daemon/ceph/config"
@@ -65,8 +66,12 @@ func testDeploymentObject(hostNetwork bool) *apps.Deployment {
clusterInfo,
&clusterd.Context{Clientset: testop.New(1)},
"rook/rook:myversion",
cephv1.CephVersionSpec{Image: "ceph/ceph:testversion"},
hostNetwork,
&cephv1.ClusterSpec{
CephVersion: cephv1.CephVersionSpec{Image: "ceph/ceph:testversion"},
Network: rookalpha.NetworkSpec{
HostNetwork: hostNetwork,
},
},
fs,
&client.CephFilesystemDetails{ID: 15},
[]metav1.OwnerReference{{}},
+10 -12
View File
@@ -49,28 +49,26 @@ type CephNFSController struct {
dataDirHostPath string
namespace string
rookImage string
cephVersion cephv1.CephVersionSpec
hostNetwork bool
clusterSpec *cephv1.ClusterSpec
ownerRef metav1.OwnerReference
orchestrationMutex sync.Mutex
}
// NewCephNFSController create controller for watching NFS custom resources created
func NewCephNFSController(clusterInfo *cephconfig.ClusterInfo, context *clusterd.Context, dataDirHostPath, namespace, rookImage string, cephVersion cephv1.CephVersionSpec, hostNetwork bool, ownerRef metav1.OwnerReference) *CephNFSController {
func NewCephNFSController(clusterInfo *cephconfig.ClusterInfo, context *clusterd.Context, dataDirHostPath, namespace, rookImage string, clusterSpec *cephv1.ClusterSpec, ownerRef metav1.OwnerReference) *CephNFSController {
return &CephNFSController{
clusterInfo: clusterInfo,
context: context,
dataDirHostPath: dataDirHostPath,
namespace: namespace,
rookImage: rookImage,
cephVersion: cephVersion,
hostNetwork: hostNetwork,
clusterSpec: clusterSpec,
ownerRef: ownerRef,
}
}
// StartWatch watches for instances of CephNFS custom resources and acts on them
func (c *CephNFSController) StartWatch(stopCh chan struct{}) error {
func (c *CephNFSController) StartWatch(namespace string, stopCh chan struct{}) error {
resourceHandlerFuncs := cache.ResourceEventHandlerFuncs{
AddFunc: c.onAdd,
@@ -78,8 +76,8 @@ func (c *CephNFSController) StartWatch(stopCh chan struct{}) error {
DeleteFunc: c.onDelete,
}
logger.Infof("start watching ceph nfs resource in namespace %s", c.namespace)
watcher := opkit.NewWatcher(CephNFSResource, c.namespace, resourceHandlerFuncs, c.context.RookClientset.CephV1().RESTClient())
logger.Infof("start watching ceph nfs resource in namespace %s", namespace)
watcher := opkit.NewWatcher(CephNFSResource, namespace, resourceHandlerFuncs, c.context.RookClientset.CephV1().RESTClient())
go watcher.Watch(&cephv1.CephNFS{}, stopCh)
return nil
@@ -152,7 +150,7 @@ func (c *CephNFSController) onDelete(obj interface{}) {
// cluster has changed.
func (c *CephNFSController) ParentClusterChanged(cluster cephv1.ClusterSpec, clusterInfo *cephconfig.ClusterInfo) {
c.clusterInfo = clusterInfo
if cluster.CephVersion.Image == c.cephVersion.Image || !c.clusterInfo.CephVersion.IsAtLeastNautilus() {
if cluster.CephVersion.Image == c.clusterSpec.CephVersion.Image || !c.clusterInfo.CephVersion.IsAtLeastNautilus() {
logger.Debugf("No need to update the nfs daemons after the parent cluster changed")
return
}
@@ -160,19 +158,19 @@ func (c *CephNFSController) ParentClusterChanged(cluster cephv1.ClusterSpec, clu
c.acquireOrchestrationLock()
defer c.releaseOrchestrationLock()
c.cephVersion = cluster.CephVersion
c.clusterSpec.CephVersion = cluster.CephVersion
nfses, err := c.context.RookClientset.CephV1().CephNFSes(c.namespace).List(metav1.ListOptions{})
if err != nil {
logger.Errorf("failed to retrieve NFSes to update the ceph version. %+v", err)
return
}
for _, nfs := range nfses.Items {
logger.Infof("updating the ceph version for nfs %s to %s", nfs.Name, c.cephVersion.Image)
logger.Infof("updating the ceph version for nfs %s to %s", nfs.Name, c.clusterSpec.CephVersion.Image)
err := c.upCephNFS(nfs, 0)
if err != nil {
logger.Errorf("failed to update nfs %s. %+v", nfs.Name, err)
} else {
logger.Infof("updated nfs %s to ceph version %s", nfs.Name, c.cephVersion.Image)
logger.Infof("updated nfs %s to ceph version %s", nfs.Name, c.clusterSpec.CephVersion.Image)
}
}
}
+8 -8
View File
@@ -58,7 +58,7 @@ func (c *CephNFSController) createCephNFSService(nfs cephv1.CephNFS, cfg daemonC
},
}
k8sutil.SetOwnerRef(&svc.ObjectMeta, &c.ownerRef)
if c.hostNetwork {
if c.clusterSpec.Network.HostNetwork {
svc.Spec.ClusterIP = v1.ClusterIPNone
}
@@ -110,9 +110,9 @@ func (c *CephNFSController) makeDeployment(nfs cephv1.CephNFS, cfg daemonConfig)
nfsConfigVol,
dbusVol,
},
HostNetwork: c.hostNetwork,
HostNetwork: c.clusterSpec.Network.HostNetwork,
}
if c.hostNetwork {
if c.clusterSpec.Network.HostNetwork {
podSpec.DNSPolicy = v1.DNSClusterFirstWithHostNet
}
nfs.Spec.Server.Placement.ApplyToPodSpec(&podSpec)
@@ -155,7 +155,7 @@ func (c *CephNFSController) connectionConfigInitContainer(nfs cephv1.CephNFS) v1
keyring.VolumeMount().Admin(),
},
Env: append(
opspec.DaemonEnvVars(c.cephVersion.Image),
opspec.DaemonEnvVars(c.clusterSpec.CephVersion.Image),
),
Resources: nfs.Spec.Server.Resources,
}
@@ -175,7 +175,7 @@ func (c *CephNFSController) daemonContainer(nfs cephv1.CephNFS, cfg daemonConfig
"-F", // foreground
"-L", "STDERR", // log to stderr
},
Image: c.cephVersion.Image,
Image: c.clusterSpec.CephVersion.Image,
VolumeMounts: []v1.VolumeMount{
cephConfigMount,
keyring.VolumeMount().Admin(),
@@ -183,7 +183,7 @@ func (c *CephNFSController) daemonContainer(nfs cephv1.CephNFS, cfg daemonConfig
dbusMount,
},
Env: append(
opspec.DaemonEnvVars(c.cephVersion.Image),
opspec.DaemonEnvVars(c.clusterSpec.CephVersion.Image),
),
Resources: nfs.Spec.Server.Resources,
}
@@ -202,13 +202,13 @@ func (c *CephNFSController) dbusContainer(nfs cephv1.CephNFS) v1.Container {
"--nopidfile", // don't write a pid file
// some dbus-daemon versions have flag --nosyslog to send logs to sterr; not ceph upstream image
},
Image: c.cephVersion.Image,
Image: c.clusterSpec.CephVersion.Image,
VolumeMounts: []v1.VolumeMount{
dbusMount,
},
Env: append(
// do not need access to Ceph env vars b/c not a Ceph daemon
k8sutil.ClusterDaemonEnvVars(c.cephVersion.Image),
k8sutil.ClusterDaemonEnvVars(c.clusterSpec.CephVersion.Image),
),
Resources: nfs.Spec.Server.Resources,
}
+1 -2
View File
@@ -63,8 +63,7 @@ func TestDeploymentSpec(t *testing.T) {
"/var/lib/rook",
"rook-ceph-test-ns",
"rook/rook:testimage",
cephv1.CephVersionSpec{Image: "ceph/ceph:myceph"},
false,
&cephv1.ClusterSpec{CephVersion: cephv1.CephVersionSpec{Image: "ceph/ceph:testversion"}},
metav1.OwnerReference{},
)
+9 -13
View File
@@ -47,11 +47,10 @@ var ObjectStoreResource = opkit.CustomResource{
// ObjectStoreController represents a controller object for object store custom resources
type ObjectStoreController struct {
clusterInfo *daemonconfig.ClusterInfo
clusterSpec *cephv1.ClusterSpec
context *clusterd.Context
namespace string
rookImage string
cephVersion cephv1.CephVersionSpec
hostNetwork bool
ownerRef metav1.OwnerReference
dataDirHostPath string
orchestrationMutex sync.Mutex
@@ -63,25 +62,23 @@ func NewObjectStoreController(
context *clusterd.Context,
namespace string,
rookImage string,
cephVersion cephv1.CephVersionSpec,
hostNetwork bool,
clusterSpec *cephv1.ClusterSpec,
ownerRef metav1.OwnerReference,
dataDirHostPath string,
) *ObjectStoreController {
return &ObjectStoreController{
clusterInfo: clusterInfo,
clusterSpec: clusterSpec,
context: context,
namespace: namespace,
rookImage: rookImage,
cephVersion: cephVersion,
hostNetwork: hostNetwork,
ownerRef: ownerRef,
dataDirHostPath: dataDirHostPath,
}
}
// StartWatch watches for instances of ObjectStore custom resources and acts on them
func (c *ObjectStoreController) StartWatch(stopCh chan struct{}) error {
func (c *ObjectStoreController) StartWatch(namespace string, stopCh chan struct{}) error {
resourceHandlerFuncs := cache.ResourceEventHandlerFuncs{
AddFunc: c.onAdd,
UpdateFunc: c.onUpdate,
@@ -138,8 +135,7 @@ func (c *ObjectStoreController) createOrUpdateStore(objectstore *cephv1.CephObje
context: c.context,
store: *objectstore,
rookVersion: c.rookImage,
cephVersion: c.cephVersion,
hostNetwork: c.hostNetwork,
clusterSpec: c.clusterSpec,
ownerRefs: c.storeOwners(objectstore),
DataPathMap: cephconfig.NewStatelessDaemonDataPathMap(cephconfig.RgwType, objectstore.Name, c.clusterInfo.Name, c.dataDirHostPath),
}
@@ -166,7 +162,7 @@ func (c *ObjectStoreController) onDelete(obj interface{}) {
func (c *ObjectStoreController) ParentClusterChanged(cluster cephv1.ClusterSpec, clusterInfo *daemonconfig.ClusterInfo) {
c.clusterInfo = clusterInfo
if cluster.CephVersion.Image == c.cephVersion.Image {
if cluster.CephVersion.Image == c.clusterSpec.CephVersion.Image {
logger.Debugf("No need to update the object store after the parent cluster changed")
return
}
@@ -174,19 +170,19 @@ func (c *ObjectStoreController) ParentClusterChanged(cluster cephv1.ClusterSpec,
c.acquireOrchestrationLock()
defer c.releaseOrchestrationLock()
c.cephVersion = cluster.CephVersion
c.clusterSpec.CephVersion = cluster.CephVersion
objectStores, err := c.context.RookClientset.CephV1().CephObjectStores(c.namespace).List(metav1.ListOptions{})
if err != nil {
logger.Errorf("failed to retrieve object stores to update the ceph version. %+v", err)
return
}
for _, store := range objectStores.Items {
logger.Infof("updating the ceph version for object store %s to %s", store.Name, c.cephVersion.Image)
logger.Infof("updating the ceph version for object store %s to %s", store.Name, c.clusterSpec.CephVersion.Image)
c.createOrUpdateStore(&store)
if err != nil {
logger.Errorf("failed to update object store %s. %+v", store.Name, err)
} else {
logger.Infof("updated object store %s to ceph version %s", store.Name, c.cephVersion.Image)
logger.Infof("updated object store %s to ceph version %s", store.Name, c.clusterSpec.CephVersion.Image)
}
}
}
+1 -2
View File
@@ -39,8 +39,7 @@ type clusterConfig struct {
context *clusterd.Context
store cephv1.CephObjectStore
rookVersion string
cephVersion cephv1.CephVersionSpec
hostNetwork bool
clusterSpec *cephv1.ClusterSpec
ownerRefs []metav1.OwnerReference
DataPathMap *config.DataPathMap
}
+2 -2
View File
@@ -52,7 +52,7 @@ func TestStartRGW(t *testing.T) {
data := cephconfig.NewStatelessDaemonDataPathMap(cephconfig.RgwType, "my-fs", "rook-ceph", "/var/lib/rook/")
// start a basic cluster
c := &clusterConfig{info, context, store, version, cephv1.CephVersionSpec{}, false, []metav1.OwnerReference{}, data}
c := &clusterConfig{info, context, store, version, &cephv1.ClusterSpec{}, []metav1.OwnerReference{}, data}
err := c.startRGWPods()
assert.Nil(t, err)
@@ -94,7 +94,7 @@ func TestCreateObjectStore(t *testing.T) {
data := cephconfig.NewStatelessDaemonDataPathMap(cephconfig.RgwType, "my-fs", "rook-ceph", "/var/lib/rook/")
// create the pools
c := &clusterConfig{info, context, store, "1.2.3.4", cephv1.CephVersionSpec{}, false, []metav1.OwnerReference{}, data}
c := &clusterConfig{info, context, store, "1.2.3.4", &cephv1.ClusterSpec{}, []metav1.OwnerReference{}, data}
err := c.createOrUpdate()
assert.Nil(t, err)
}
+5 -5
View File
@@ -68,9 +68,9 @@ func (c *clusterConfig) makeRGWPodSpec(rgwConfig *rgwConfig) v1.PodTemplateSpec
opspec.DaemonVolumes(c.DataPathMap, rgwConfig.ResourceName),
c.mimeTypesVolume(),
),
HostNetwork: c.hostNetwork,
HostNetwork: c.clusterSpec.Network.HostNetwork,
}
if c.hostNetwork {
if c.clusterSpec.Network.HostNetwork {
podSpec.DNSPolicy = v1.DNSClusterFirstWithHostNet
}
@@ -106,7 +106,7 @@ func (c *clusterConfig) makeDaemonContainer(rgwConfig *rgwConfig) v1.Container {
// start the rgw daemon in the foreground
container := v1.Container{
Name: "rgw",
Image: c.cephVersion.Image,
Image: c.clusterSpec.CephVersion.Image,
Command: []string{
"radosgw",
},
@@ -123,7 +123,7 @@ func (c *clusterConfig) makeDaemonContainer(rgwConfig *rgwConfig) v1.Container {
opspec.DaemonVolumeMounts(c.DataPathMap, rgwConfig.ResourceName),
c.mimeTypesVolumeMount(),
),
Env: opspec.DaemonEnvVars(c.cephVersion.Image),
Env: opspec.DaemonEnvVars(c.clusterSpec.CephVersion.Image),
Resources: c.store.Spec.Gateway.Resources,
LivenessProbe: &v1.Probe{
Handler: v1.Handler{
@@ -160,7 +160,7 @@ func (c *clusterConfig) startService() (string, error) {
},
}
k8sutil.SetOwnerRefs(&svc.ObjectMeta, c.ownerRefs)
if c.hostNetwork {
if c.clusterSpec.Network.HostNetwork {
svc.Spec.ClusterIP = v1.ClusterIPNone
}
+13 -5
View File
@@ -21,6 +21,7 @@ import (
"testing"
cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
rookalpha "github.com/rook/rook/pkg/apis/rook.io/v1alpha2"
"github.com/rook/rook/pkg/clusterd"
cephconfig "github.com/rook/rook/pkg/operator/ceph/config"
cephtest "github.com/rook/rook/pkg/operator/ceph/test"
@@ -52,8 +53,12 @@ func TestPodSpecs(t *testing.T) {
clusterInfo: info,
store: store,
rookVersion: "rook/rook:myversion",
cephVersion: cephv1.CephVersionSpec{Image: "ceph/ceph:v13.2.1"},
hostNetwork: true,
clusterSpec: &cephv1.ClusterSpec{
CephVersion: cephv1.CephVersionSpec{Image: "ceph/ceph:v13.2.1"},
Network: rookalpha.NetworkSpec{
HostNetwork: true,
},
},
DataPathMap: data,
}
@@ -91,11 +96,14 @@ func TestSSLPodSpec(t *testing.T) {
clusterInfo: info,
store: store,
rookVersion: "rook/rook:myversion",
cephVersion: cephv1.CephVersionSpec{Image: "ceph/ceph:v13.2.1"},
hostNetwork: true,
clusterSpec: &cephv1.ClusterSpec{
CephVersion: cephv1.CephVersionSpec{Image: "ceph/ceph:v13.2.1"},
Network: rookalpha.NetworkSpec{
HostNetwork: true,
},
},
DataPathMap: data,
}
c.hostNetwork = true
resourceName := fmt.Sprintf("%s-%s", AppName, c.store.Name)
rgwConfig := &rgwConfig{
+10 -8
View File
@@ -28,7 +28,7 @@ import (
cephconfig "github.com/rook/rook/pkg/daemon/ceph/config"
"github.com/rook/rook/pkg/operator/ceph/object"
"github.com/rook/rook/pkg/operator/k8sutil"
"k8s.io/api/core/v1"
v1 "k8s.io/api/core/v1"
apiextensionsv1beta1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1beta1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/client-go/tools/cache"
@@ -52,17 +52,19 @@ var ObjectStoreUserResource = opkit.CustomResource{
// ObjectStoreUserController represents a controller object for object store user custom resources
type ObjectStoreUserController struct {
context *clusterd.Context
ownerRef metav1.OwnerReference
namespace string
context *clusterd.Context
ownerRef metav1.OwnerReference
clusterSpec *cephv1.ClusterSpec
namespace string
}
// NewObjectStoreUserController create controller for watching object store user custom resources created
func NewObjectStoreUserController(context *clusterd.Context, namespace string, ownerRef metav1.OwnerReference) *ObjectStoreUserController {
func NewObjectStoreUserController(context *clusterd.Context, clusterSpec *cephv1.ClusterSpec, namespace string, ownerRef metav1.OwnerReference) *ObjectStoreUserController {
return &ObjectStoreUserController{
context: context,
ownerRef: ownerRef,
namespace: namespace,
context: context,
ownerRef: ownerRef,
clusterSpec: clusterSpec,
namespace: namespace,
}
}
+10 -1
View File
@@ -149,7 +149,7 @@ func (o *Operator) Run() error {
}
}
func (o *Operator) startSystemDaemons() error {
func (o *Operator) startSystemDaemons(externalCeph bool) error {
if o.delayedDaemonsStarted {
return nil
}
@@ -166,6 +166,15 @@ func (o *Operator) startSystemDaemons() error {
}
}
// The discover daemon is only needed for local clusters where OSDs will be created.
// External clusters do not create OSDs locally.
if !externalCeph {
rookDiscover := discover.New(o.context.Clientset)
if err := rookDiscover.Start(namespace, o.rookImage, o.securityAccount); err != nil {
return fmt.Errorf("Error starting device discovery daemonset: %v", err)
}
}
serverVersion, err := o.context.Clientset.Discovery().ServerVersion()
if err != nil {
return fmt.Errorf("error getting server version: %v", err)
+8 -8
View File
@@ -52,20 +52,20 @@ var PoolResource = opkit.CustomResource{
// PoolController represents a controller object for pool custom resources
type PoolController struct {
context *clusterd.Context
namespace string
context *clusterd.Context
clusterSpec *cephv1.ClusterSpec
}
// NewPoolController create controller for watching pool custom resources created
func NewPoolController(context *clusterd.Context, namespace string) *PoolController {
func NewPoolController(context *clusterd.Context, clusterSpec *cephv1.ClusterSpec) *PoolController {
return &PoolController{
context: context,
namespace: namespace,
context: context,
clusterSpec: clusterSpec,
}
}
// Watch watches for instances of Pool custom resources and acts on them
func (c *PoolController) StartWatch(stopCh chan struct{}) error {
func (c *PoolController) StartWatch(namespace string, stopCh chan struct{}) error {
resourceHandlerFuncs := cache.ResourceEventHandlerFuncs{
AddFunc: c.onAdd,
@@ -73,8 +73,8 @@ func (c *PoolController) StartWatch(stopCh chan struct{}) error {
DeleteFunc: c.onDelete,
}
logger.Infof("start watching pool resources in namespace %s", c.namespace)
watcher := opkit.NewWatcher(PoolResource, c.namespace, resourceHandlerFuncs, c.context.RookClientset.CephV1().RESTClient())
logger.Infof("start watching clusters in all namespaces")
watcher := opkit.NewWatcher(PoolResource, namespace, resourceHandlerFuncs, c.context.RookClientset.CephV1().RESTClient())
go watcher.Watch(&cephv1.CephBlockPool{}, stopCh)
return nil
}
+2 -2
View File
@@ -19,7 +19,7 @@ package k8sutil
import (
"fmt"
"k8s.io/api/core/v1"
v1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/client-go/kubernetes"
@@ -50,7 +50,7 @@ func UpdateService(
clientset kubernetes.Interface, namespace string, serviceDefinition *v1.Service,
) (*v1.Service, error) {
name := serviceDefinition.Name
logger.Debug("updating service %s")
logger.Debugf("updating service %s", name)
existing, err := clientset.CoreV1().Services(namespace).Get(name, metav1.GetOptions{})
if err != nil {
return nil, fmt.Errorf("could not get existing service %s in order to update. %+v", name, err)