diff --git a/Documentation/ceph-cluster-crd.md b/Documentation/ceph-cluster-crd.md index bf376761a..ab230a01b 100755 --- a/Documentation/ceph-cluster-crd.md +++ b/Documentation/ceph-cluster-crd.md @@ -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`. diff --git a/PendingReleaseNotes.md b/PendingReleaseNotes.md index 79334a657..444ecd860 100644 --- a/PendingReleaseNotes.md +++ b/PendingReleaseNotes.md @@ -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 diff --git a/cluster/examples/kubernetes/ceph/cluster-external.yaml b/cluster/examples/kubernetes/ceph/cluster-external.yaml new file mode 100644 index 000000000..72df038f9 --- /dev/null +++ b/cluster/examples/kubernetes/ceph/cluster-external.yaml @@ -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 diff --git a/cluster/examples/kubernetes/ceph/common-external.yaml b/cluster/examples/kubernetes/ceph/common-external.yaml new file mode 100644 index 000000000..8667aa40e --- /dev/null +++ b/cluster/examples/kubernetes/ceph/common-external.yaml @@ -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 \ No newline at end of file diff --git a/cluster/examples/kubernetes/ceph/common.yaml b/cluster/examples/kubernetes/ceph/common.yaml index 3727bd328..e6fb4ee94 100644 --- a/cluster/examples/kubernetes/ceph/common.yaml +++ b/cluster/examples/kubernetes/ceph/common.yaml @@ -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: diff --git a/cluster/examples/kubernetes/ceph/import-external-cluster.sh b/cluster/examples/kubernetes/ceph/import-external-cluster.sh new file mode 100644 index 000000000..cc0a6049b --- /dev/null +++ b/cluster/examples/kubernetes/ceph/import-external-cluster.sh @@ -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 \ No newline at end of file diff --git a/cluster/examples/kubernetes/ceph/operator.yaml b/cluster/examples/kubernetes/ceph/operator.yaml index fafb913bf..85b73bef0 100644 --- a/cluster/examples/kubernetes/ceph/operator.yaml +++ b/cluster/examples/kubernetes/ceph/operator.yaml @@ -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" diff --git a/pkg/apis/ceph.rook.io/v1/types.go b/pkg/apis/ceph.rook.io/v1/types.go index 7c9c953c3..f993aa7fa 100755 --- a/pkg/apis/ceph.rook.io/v1/types.go +++ b/pkg/apis/ceph.rook.io/v1/types.go @@ -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"` } diff --git a/pkg/daemon/ceph/client/mon.go b/pkg/daemon/ceph/client/mon.go index 01b9145b5..373f8c9eb 100644 --- a/pkg/daemon/ceph/client/mon.go +++ b/pkg/daemon/ceph/client/mon.go @@ -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 } diff --git a/pkg/daemon/ceph/config/config.go b/pkg/daemon/ceph/config/config.go index 0200bee73..047bb0a07 100644 --- a/pkg/daemon/ceph/config/config.go +++ b/pkg/daemon/ceph/config/config.go @@ -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 + +} diff --git a/pkg/daemon/ceph/config/config_test.go b/pkg/daemon/ceph/config/config_test.go index 083972d68..0288b6200 100644 --- a/pkg/daemon/ceph/config/config_test.go +++ b/pkg/daemon/ceph/config/config_test.go @@ -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) +} diff --git a/pkg/operator/ceph/cluster/cluster.go b/pkg/operator/ceph/cluster/cluster.go index f1fc7e35f..d161e92a0 100755 --- a/pkg/operator/ceph/cluster/cluster.go +++ b/pkg/operator/ceph/cluster/cluster.go @@ -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 { diff --git a/pkg/operator/ceph/cluster/controller.go b/pkg/operator/ceph/cluster/controller.go index 590ac8ad7..8f9d1bb01 100644 --- a/pkg/operator/ceph/cluster/controller.go +++ b/pkg/operator/ceph/cluster/controller.go @@ -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 +} diff --git a/pkg/operator/ceph/cluster/controller_test.go b/pkg/operator/ceph/cluster/controller_test.go index b7a19d5ef..dbab1ba33 100644 --- a/pkg/operator/ceph/cluster/controller_test.go +++ b/pkg/operator/ceph/cluster/controller_test.go @@ -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) +} diff --git a/pkg/operator/ceph/cluster/mon/config.go b/pkg/operator/ceph/cluster/mon/config.go index 657b16451..f62a411b9 100644 --- a/pkg/operator/ceph/cluster/mon/config.go +++ b/pkg/operator/ceph/cluster/mon/config.go @@ -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, diff --git a/pkg/operator/ceph/cluster/mon/env.go b/pkg/operator/ceph/cluster/mon/env.go index 41f18b239..4ca440cec 100644 --- a/pkg/operator/ceph/cluster/mon/env.go +++ b/pkg/operator/ceph/cluster/mon/env.go @@ -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}} } diff --git a/pkg/operator/ceph/cluster/mon/health.go b/pkg/operator/ceph/cluster/mon/health.go index bb5de6409..f75614734 100644 --- a/pkg/operator/ceph/cluster/mon/health.go +++ b/pkg/operator/ceph/cluster/mon/health.go @@ -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 +} diff --git a/pkg/operator/ceph/cluster/mon/health_test.go b/pkg/operator/ceph/cluster/mon/health_test.go index d753cd11b..e1315ca5e 100644 --- a/pkg/operator/ceph/cluster/mon/health_test.go +++ b/pkg/operator/ceph/cluster/mon/health_test.go @@ -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) +} diff --git a/pkg/operator/ceph/cluster/mon/mon.go b/pkg/operator/ceph/cluster/mon/mon.go index 5e68d733b..5db4f4378 100644 --- a/pkg/operator/ceph/cluster/mon/mon.go +++ b/pkg/operator/ceph/cluster/mon/mon.go @@ -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 diff --git a/pkg/operator/ceph/cluster/mon/mon_test.go b/pkg/operator/ceph/cluster/mon/mon_test.go index 21b7bf128..94b8f5542 100644 --- a/pkg/operator/ceph/cluster/mon/mon_test.go +++ b/pkg/operator/ceph/cluster/mon/mon_test.go @@ -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", diff --git a/pkg/operator/ceph/cluster/mon/node.go b/pkg/operator/ceph/cluster/mon/node.go index 624df5d27..0ef1744b1 100644 --- a/pkg/operator/ceph/cluster/mon/node.go +++ b/pkg/operator/ceph/cluster/mon/node.go @@ -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 diff --git a/pkg/operator/ceph/cluster/mon/node_test.go b/pkg/operator/ceph/cluster/mon/node_test.go index a66374e09..29e07d022 100644 --- a/pkg/operator/ceph/cluster/mon/node_test.go +++ b/pkg/operator/ceph/cluster/mon/node_test.go @@ -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) } diff --git a/pkg/operator/ceph/cluster/mon/service.go b/pkg/operator/ceph/cluster/mon/service.go index 7675b93ab..604453580 100644 --- a/pkg/operator/ceph/cluster/mon/service.go +++ b/pkg/operator/ceph/cluster/mon/service.go @@ -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) diff --git a/pkg/operator/ceph/cluster/mon/spec.go b/pkg/operator/ceph/cluster/mon/spec.go index 160b9d5a1..3651bce10 100644 --- a/pkg/operator/ceph/cluster/mon/spec.go +++ b/pkg/operator/ceph/cluster/mon/spec.go @@ -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: " 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) } diff --git a/pkg/operator/ceph/cluster/mon/spec_test.go b/pkg/operator/ceph/cluster/mon/spec_test.go index 9575b3c78..2ca0e3e0e 100644 --- a/pkg/operator/ceph/cluster/mon/spec_test.go +++ b/pkg/operator/ceph/cluster/mon/spec_test.go @@ -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 */) } diff --git a/pkg/operator/ceph/cluster/mon/util.go b/pkg/operator/ceph/cluster/mon/util.go index 0c0fbfc11..7c9b9188d 100644 --- a/pkg/operator/ceph/cluster/mon/util.go +++ b/pkg/operator/ceph/cluster/mon/util.go @@ -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):]) diff --git a/pkg/operator/ceph/config/store.go b/pkg/operator/ceph/config/store.go index 5a890bd99..acc190cf9 100644 --- a/pkg/operator/ceph/config/store.go +++ b/pkg/operator/ceph/config/store.go @@ -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}}}, } } diff --git a/pkg/operator/ceph/config/store_test.go b/pkg/operator/ceph/config/store_test.go index c02edf693..3a3086b08 100644 --- a/pkg/operator/ceph/config/store_test.go +++ b/pkg/operator/ceph/config/store_test.go @@ -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) } diff --git a/pkg/operator/ceph/csi/cluster_config.go b/pkg/operator/ceph/csi/cluster_config.go index 434846631..f81b7eb4b 100644 --- a/pkg/operator/ceph/csi/cluster_config.go +++ b/pkg/operator/ceph/csi/cluster_config.go @@ -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( diff --git a/pkg/operator/ceph/file/controller.go b/pkg/operator/ceph/file/controller.go index 91a476d48..85c62aa78 100644 --- a/pkg/operator/ceph/file/controller.go +++ b/pkg/operator/ceph/file/controller.go @@ -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) } } } diff --git a/pkg/operator/ceph/file/filesystem.go b/pkg/operator/ceph/file/filesystem.go index fb0a1ab83..4eed083d8 100644 --- a/pkg/operator/ceph/file/filesystem.go +++ b/pkg/operator/ceph/file/filesystem.go @@ -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 } diff --git a/pkg/operator/ceph/file/filesystem_test.go b/pkg/operator/ceph/file/filesystem_test.go index 37c2bec33..d0570542d 100644 --- a/pkg/operator/ceph/file/filesystem_test.go +++ b/pkg/operator/ceph/file/filesystem_test.go @@ -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) diff --git a/pkg/operator/ceph/file/mds/mds.go b/pkg/operator/ceph/file/mds/mds.go index 011f75c57..f39c0ce6d 100644 --- a/pkg/operator/ceph/file/mds/mds.go +++ b/pkg/operator/ceph/file/mds/mds.go @@ -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, diff --git a/pkg/operator/ceph/file/mds/spec.go b/pkg/operator/ceph/file/mds/spec.go index cedd33e4a..04969a6f9 100644 --- a/pkg/operator/ceph/file/mds/spec.go +++ b/pkg/operator/ceph/file/mds/spec.go @@ -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(""), diff --git a/pkg/operator/ceph/file/mds/spec_test.go b/pkg/operator/ceph/file/mds/spec_test.go index 3614866a0..06875ffe3 100644 --- a/pkg/operator/ceph/file/mds/spec_test.go +++ b/pkg/operator/ceph/file/mds/spec_test.go @@ -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{{}}, diff --git a/pkg/operator/ceph/nfs/controller.go b/pkg/operator/ceph/nfs/controller.go index db891bbac..e67e93186 100644 --- a/pkg/operator/ceph/nfs/controller.go +++ b/pkg/operator/ceph/nfs/controller.go @@ -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) } } } diff --git a/pkg/operator/ceph/nfs/spec.go b/pkg/operator/ceph/nfs/spec.go index 82c21aa7a..fc129975e 100644 --- a/pkg/operator/ceph/nfs/spec.go +++ b/pkg/operator/ceph/nfs/spec.go @@ -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, } diff --git a/pkg/operator/ceph/nfs/spec_test.go b/pkg/operator/ceph/nfs/spec_test.go index 2cc6456d4..998663917 100644 --- a/pkg/operator/ceph/nfs/spec_test.go +++ b/pkg/operator/ceph/nfs/spec_test.go @@ -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{}, ) diff --git a/pkg/operator/ceph/object/controller.go b/pkg/operator/ceph/object/controller.go index a7247482b..06ab44dc2 100644 --- a/pkg/operator/ceph/object/controller.go +++ b/pkg/operator/ceph/object/controller.go @@ -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) } } } diff --git a/pkg/operator/ceph/object/rgw.go b/pkg/operator/ceph/object/rgw.go index 46f968328..a6f42a68f 100644 --- a/pkg/operator/ceph/object/rgw.go +++ b/pkg/operator/ceph/object/rgw.go @@ -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 } diff --git a/pkg/operator/ceph/object/rgw_test.go b/pkg/operator/ceph/object/rgw_test.go index 7efc0c229..bd0b19cc7 100644 --- a/pkg/operator/ceph/object/rgw_test.go +++ b/pkg/operator/ceph/object/rgw_test.go @@ -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) } diff --git a/pkg/operator/ceph/object/spec.go b/pkg/operator/ceph/object/spec.go index bdf0d0528..caafbaff9 100644 --- a/pkg/operator/ceph/object/spec.go +++ b/pkg/operator/ceph/object/spec.go @@ -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 } diff --git a/pkg/operator/ceph/object/spec_test.go b/pkg/operator/ceph/object/spec_test.go index 4e3cc9045..3875d1601 100644 --- a/pkg/operator/ceph/object/spec_test.go +++ b/pkg/operator/ceph/object/spec_test.go @@ -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{ diff --git a/pkg/operator/ceph/object/user/controller.go b/pkg/operator/ceph/object/user/controller.go index 467520a73..c9e3b8bf0 100644 --- a/pkg/operator/ceph/object/user/controller.go +++ b/pkg/operator/ceph/object/user/controller.go @@ -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, } } diff --git a/pkg/operator/ceph/operator.go b/pkg/operator/ceph/operator.go index 789cfb635..df3e44997 100644 --- a/pkg/operator/ceph/operator.go +++ b/pkg/operator/ceph/operator.go @@ -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) diff --git a/pkg/operator/ceph/pool/controller.go b/pkg/operator/ceph/pool/controller.go index 21dfd5ab9..3446c5c6f 100644 --- a/pkg/operator/ceph/pool/controller.go +++ b/pkg/operator/ceph/pool/controller.go @@ -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 } diff --git a/pkg/operator/k8sutil/service.go b/pkg/operator/k8sutil/service.go index 70dd62dc4..e15e1cf12 100644 --- a/pkg/operator/k8sutil/service.go +++ b/pkg/operator/k8sutil/service.go @@ -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)