diff --git a/Documentation/common-issues.md b/Documentation/common-issues.md
index 297b936eb..c882a1699 100644
--- a/Documentation/common-issues.md
+++ b/Documentation/common-issues.md
@@ -21,6 +21,7 @@ If after trying the suggestions found on this page and the problem is not resolv
 - [Rook Agent modprobe exec format error](#rook-agent-modprobe-exec-format-error)
 - [Rook Agent rbd module missing error](#rook-agent-rbd-module-missing-error)
 - [Using multiple shared filesystem (CephFS) is attempted on a kernel version older than 4.7](#using-multiple-shared-filesystem-cephfs-is-attempted-on-a-kernel-version-older-than-47)
+- [Activate log to file for a particular Ceph daemon](#activate-ceph-log-on-file)
 
 # Troubleshooting Techniques
 There are two main categories of information you will need to investigate issues in the cluster:
@@ -518,3 +519,27 @@ The only solution to this problem is to upgrade your kernel to `4.7` or higher.
 This is due to a mount flag added in the kernel version `4.7` which allows to chose the filesystem by name.
 
 For additional info on the kernel version requirement for multiple shared filesystems (CephFS), see [Filesystem - Kernel version requirement](ceph-filesystem.md#kernel-version-requirement).
+
+# Activate log to file for a particular Ceph daemon
+
+They are cases where looking at Kubernetes logs is not enough for diverse reasons, but just to name a few:
+
+* not everyone is familiar for Kubernetes logging and expects to find logs in traditional directories
+* logs get eaten (buffer limit from the log engine) and thus not requestable from Kubernetes
+
+So for each daemon, `dataDirHostPath` is used to store logs, if logging is activated.
+Rook will bindmount `dataDirHostPath` for every pod.
+As of Ceph Nautilus 14.2.1, it is possible to enable logging for a particular daemon on the fly.
+Let's say you want to enable logging for `mon.a`, but only for this daemon.
+Using the toolbox or from inside the operator run:
+
+```
+ceph config daemon mon.a log_to_file true
+```
+
+This will activate logging on the filesystem, you will be able to find logs in `dataDirHostPath/$NAMESPACE/log`, so typically this would mean `/var/lib/rook/rook-ceph/log`.
+You don't need to restart the pod, the effect will be immediate.
+
+To disable the logging on file, simply set `log_to_file` to `false`.
+
+For Ceph Luminous/Mimic releases, `mon_cluster_log_file` and `cluster_log_file` can be set to `/var/log/ceph/XXXX` in the config override ConfigMap to enable logging. See the (Advanced Documentation)[Documentation/advanced-configuration.md#kubernetes] for information about how to use the config override ConfigMap.
\ No newline at end of file
diff --git a/PendingReleaseNotes.md b/PendingReleaseNotes.md
index 04beb69bc..69af89bb3 100644
--- a/PendingReleaseNotes.md
+++ b/PendingReleaseNotes.md
@@ -22,6 +22,7 @@
 - Added Ceph CSI driver deployments on Kubernetes 1.13 and above.
 - The number of mons can be increased automatically when new nodes come online. See the [preferredCount](https://rook.io/docs/rook/master/ceph-cluster-crd.html#mon-settings) setting in the cluster CRD documentation.
 - New Kubernetes nodes or nodes which are not tainted `NoSchedule` anymore get added automatically to the existing rook cluster if `useAllNodes` is set. [Issue #2208](https://github.com/rook/rook/issues/2208)
+- Pod's logs can be written on the filesystem as of Ceph Nautilus 14.2.1 on demand (see [common issues](https://rook.io/docs/rook/master/common-issues.html#activate-ceph-log-on-file))
 
 ## Breaking Changes
 
diff --git a/pkg/daemon/ceph/config/config.go b/pkg/daemon/ceph/config/config.go
index 6909a54e0..ddf719899 100644
--- a/pkg/daemon/ceph/config/config.go
+++ b/pkg/daemon/ceph/config/config.go
@@ -30,6 +30,7 @@ import (
 	"github.com/rook/rook/pkg/clusterd"
 	"github.com/rook/rook/pkg/daemon/ceph/client"
 	cephutil "github.com/rook/rook/pkg/daemon/ceph/util"
+	cephver "github.com/rook/rook/pkg/operator/ceph/version"
 )
 
 var logger = capnslog.NewPackageLogger("github.com/rook/rook", "cephconfig")
@@ -182,7 +183,17 @@ func getQualifiedUser(user string) string {
 }
 
 // CreateDefaultCephConfig creates a default ceph config file.
-func CreateDefaultCephConfig(context *clusterd.Context, cluster *ClusterInfo, runDir string) *CephConfig {
+func CreateDefaultCephConfig(context *clusterd.Context, cluster *ClusterInfo, runDir string) (*CephConfig, error) {
+
+	cephVersionEnv := os.Getenv("ROOK_CEPH_VERSION")
+	if cephVersionEnv != "" {
+		v, err := cephver.ExtractCephVersion(cephVersionEnv)
+		if err != nil {
+			return nil, fmt.Errorf("failed to extract ceph version. %+v", err)
+		}
+		cluster.CephVersion = *v
+	}
+
 	// extract a list of just the monitor names, which will populate the "mon initial members"
 	// global config field
 	monMembers := make([]string, len(cluster.Monitors))
@@ -211,14 +222,12 @@ func CreateDefaultCephConfig(context *clusterd.Context, cluster *ClusterInfo, ru
 
 	cephLogLevel := logLevelToCephLogLevel(context.LogLevel)
 
-	return &CephConfig{
+	conf := &CephConfig{
 		GlobalConfig: &GlobalConfig{
 			FSID:                   cluster.FSID,
 			RunDir:                 runDir,
 			MonMembers:             strings.Join(monMembers, " "),
 			MonHost:                strings.Join(monHosts, ","),
-			LogFile:                "/dev/stderr",
-			MonClusterLogFile:      "/dev/stderr",
 			PublicAddr:             context.NetworkInfo.PublicAddr,
 			PublicNetwork:          context.NetworkInfo.PublicNetwork,
 			ClusterAddr:            context.NetworkInfo.ClusterAddr,
@@ -245,6 +254,17 @@ func CreateDefaultCephConfig(context *clusterd.Context, cluster *ClusterInfo, ru
 			FatalSignalHandlers:    "false",
 		},
 	}
+
+	// Everything before 14.2.1
+	// These new flags control Ceph's daemon logging behaviour to files
+	// By default we set them to False so no logs get written on file
+	// However they can be actived at any time via the centralized config store
+	if !cluster.CephVersion.IsAtLeast(cephver.CephVersion{Major: 14, Minor: 2, Extra: 1}) {
+		conf.LogFile = "/dev/stderr"
+		conf.MonClusterLogFile = "/dev/stderr"
+	}
+
+	return conf, nil
 }
 
 // create a config file with global settings configured, and return an ini file
@@ -256,7 +276,11 @@ func createGlobalConfigFileSection(context *clusterd.Context, cluster *ClusterIn
 		// use the user config since it was provided
 		ceph = userConfig
 	} else {
-		ceph = CreateDefaultCephConfig(context, cluster, runDir)
+		var err error
+		ceph, err = CreateDefaultCephConfig(context, cluster, runDir)
+		if err != nil {
+			return nil, fmt.Errorf("failed to create default ceph config. %+v", err)
+		}
 	}
 
 	configFile := ini.Empty()
diff --git a/pkg/daemon/ceph/config/config_test.go b/pkg/daemon/ceph/config/config_test.go
index ea12d526f..083972d68 100644
--- a/pkg/daemon/ceph/config/config_test.go
+++ b/pkg/daemon/ceph/config/config_test.go
@@ -54,13 +54,19 @@ func TestCreateDefaultCephConfig(t *testing.T) {
 		},
 	}
 
-	cephConfig := CreateDefaultCephConfig(context, clusterInfo, "/var/lib/rook1")
+	cephConfig, err := CreateDefaultCephConfig(context, clusterInfo, "/var/lib/rook1")
+	if err != nil {
+		t.Fatalf("failed to create default ceph config. %+v", err)
+	}
 	verifyConfig(t, cephConfig, clusterInfo, 0)
 
 	// now use DEBUG level logging
 	context.LogLevel = capnslog.DEBUG
 
-	cephConfig = CreateDefaultCephConfig(context, clusterInfo, "/var/lib/rook1")
+	cephConfig, err = CreateDefaultCephConfig(context, clusterInfo, "/var/lib/rook1")
+	if err != nil {
+		t.Fatalf("failed to create default ceph config. %+v", err)
+	}
 	verifyConfig(t, cephConfig, clusterInfo, 10)
 
 	// verify the network info config
diff --git a/pkg/daemon/ceph/osd/daemon.go b/pkg/daemon/ceph/osd/daemon.go
index b7cf2d959..45ce54814 100644
--- a/pkg/daemon/ceph/osd/daemon.go
+++ b/pkg/daemon/ceph/osd/daemon.go
@@ -88,7 +88,10 @@ func Provision(context *clusterd.Context, agent *OsdAgent) error {
 	}
 
 	// set the crush location in the osd config file
-	cephConfig := cephconfig.CreateDefaultCephConfig(context, agent.cluster, path.Join(context.ConfigDir, agent.cluster.Name))
+	cephConfig, err := cephconfig.CreateDefaultCephConfig(context, agent.cluster, path.Join(context.ConfigDir, agent.cluster.Name))
+	if err != nil {
+		return fmt.Errorf("failed to create default ceph config. %+v", err)
+	}
 	cephConfig.GlobalConfig.CrushLocation = agent.location
 
 	// write the latest config to the config dir
diff --git a/pkg/daemon/ceph/osd/device.go b/pkg/daemon/ceph/osd/device.go
index ac9a758f5..d0a6ec699 100644
--- a/pkg/daemon/ceph/osd/device.go
+++ b/pkg/daemon/ceph/osd/device.go
@@ -501,7 +501,10 @@ func WriteConfigFile(context *clusterd.Context, cluster *cephconfig.ClusterInfo,
 }
 
 func writeConfigFile(cfg *osdConfig, context *clusterd.Context, cluster *cephconfig.ClusterInfo, location string) error {
-	cephConfig := cephconfig.CreateDefaultCephConfig(context, cluster, cfg.rootPath)
+	cephConfig, err := cephconfig.CreateDefaultCephConfig(context, cluster, cfg.rootPath)
+	if err != nil {
+		return fmt.Errorf("failed to create default ceph config. %+v", err)
+	}
 	if isBluestore(cfg) {
 		cephConfig.GlobalConfig.OsdObjectStore = config.Bluestore
 	} else {
diff --git a/pkg/operator/ceph/cluster/cluster.go b/pkg/operator/ceph/cluster/cluster.go
index 7f46a6147..e3a18c437 100755
--- a/pkg/operator/ceph/cluster/cluster.go
+++ b/pkg/operator/ceph/cluster/cluster.go
@@ -196,14 +196,14 @@ func (c *cluster) doOrchestration(rookImage string, cephVersion cephver.CephVers
 
 	mgrs := mgr.New(c.Info, c.context, c.Namespace, rookImage,
 		spec.CephVersion, cephv1.GetMgrPlacement(spec.Placement), spec.Network.HostNetwork,
-		spec.Dashboard, cephv1.GetMgrResources(spec.Resources), c.ownerRef)
+		spec.Dashboard, cephv1.GetMgrResources(spec.Resources), c.ownerRef, c.Spec.DataDirHostPath)
 	err = mgrs.Start()
 	if err != nil {
 		return fmt.Errorf("failed to start the ceph mgr. %+v", err)
 	}
 
 	// Start the OSDs
-	osds := osd.New(c.context, c.Namespace, rookImage, spec.CephVersion, spec.Storage, spec.DataDirHostPath,
+	osds := osd.New(c.Info, c.context, c.Namespace, rookImage, spec.CephVersion, spec.Storage, spec.DataDirHostPath,
 		cephv1.GetOSDPlacement(spec.Placement), spec.Network.HostNetwork, cephv1.GetOSDResources(spec.Resources), c.ownerRef)
 	err = osds.Start()
 	if err != nil {
@@ -212,7 +212,7 @@ func (c *cluster) doOrchestration(rookImage string, cephVersion cephver.CephVers
 
 	// Start the rbd mirroring daemon(s)
 	rbdmirror := rbd.New(c.Info, c.context, c.Namespace, rookImage, spec.CephVersion, cephv1.GetRBDMirrorPlacement(spec.Placement),
-		spec.Network.HostNetwork, spec.RBDMirroring, cephv1.GetRBDMirrorResources(spec.Resources), c.ownerRef)
+		spec.Network.HostNetwork, spec.RBDMirroring, cephv1.GetRBDMirrorResources(spec.Resources), c.ownerRef, c.Spec.DataDirHostPath)
 	err = rbdmirror.Start()
 	if err != nil {
 		return fmt.Errorf("failed to start the rbd mirrors. %+v", err)
diff --git a/pkg/operator/ceph/cluster/controller.go b/pkg/operator/ceph/cluster/controller.go
index 28dbd0bc8..4400fdb7d 100644
--- a/pkg/operator/ceph/cluster/controller.go
+++ b/pkg/operator/ceph/cluster/controller.go
@@ -286,7 +286,7 @@ func (c *ClusterController) onAdd(obj interface{}) {
 	poolController.StartWatch(cluster.Namespace, cluster.stopCh)
 
 	// Start object store CRD watcher
-	objectStoreController := object.NewObjectStoreController(cluster.Info, c.context, c.rookImage, cluster.Spec.CephVersion, cluster.Spec.Network.HostNetwork, cluster.ownerRef)
+	objectStoreController := object.NewObjectStoreController(cluster.Info, c.context, c.rookImage, cluster.Spec.CephVersion, cluster.Spec.Network.HostNetwork, cluster.ownerRef, cluster.Spec.DataDirHostPath)
 	objectStoreController.StartWatch(cluster.Namespace, cluster.stopCh)
 
 	// Start object store user CRD watcher
@@ -294,7 +294,7 @@ func (c *ClusterController) onAdd(obj interface{}) {
 	objectStoreUserController.StartWatch(cluster.Namespace, cluster.stopCh)
 
 	// Start file system CRD watcher
-	fileController := file.NewFilesystemController(cluster.Info, c.context, c.rookImage, cluster.Spec.CephVersion, cluster.Spec.Network.HostNetwork, cluster.ownerRef)
+	fileController := file.NewFilesystemController(cluster.Info, c.context, c.rookImage, cluster.Spec.CephVersion, cluster.Spec.Network.HostNetwork, cluster.ownerRef, cluster.Spec.DataDirHostPath)
 	fileController.StartWatch(cluster.Namespace, cluster.stopCh)
 
 	// Start nfs ganesha CRD watcher
diff --git a/pkg/operator/ceph/cluster/mgr/mgr.go b/pkg/operator/ceph/cluster/mgr/mgr.go
index 57b29b57c..e12f64980 100644
--- a/pkg/operator/ceph/cluster/mgr/mgr.go
+++ b/pkg/operator/ceph/cluster/mgr/mgr.go
@@ -47,19 +47,20 @@ const (
 
 // Cluster represents the Rook and environment configuration settings needed to set up Ceph mgrs.
 type Cluster struct {
-	clusterInfo *cephconfig.ClusterInfo
-	Namespace   string
-	Replicas    int
-	placement   rookalpha.Placement
-	context     *clusterd.Context
-	dataDir     string
-	HostNetwork bool
-	resources   v1.ResourceRequirements
-	ownerRef    metav1.OwnerReference
-	dashboard   cephv1.DashboardSpec
-	cephVersion cephv1.CephVersionSpec
-	rookVersion string
-	exitCode    func(err error) (int, bool)
+	clusterInfo     *cephconfig.ClusterInfo
+	Namespace       string
+	Replicas        int
+	placement       rookalpha.Placement
+	context         *clusterd.Context
+	dataDir         string
+	HostNetwork     bool
+	resources       v1.ResourceRequirements
+	ownerRef        metav1.OwnerReference
+	dashboard       cephv1.DashboardSpec
+	cephVersion     cephv1.CephVersionSpec
+	rookVersion     string
+	exitCode        func(err error) (int, bool)
+	dataDirHostPath string
 }
 
 // New creates an instance of the mgr
@@ -73,21 +74,23 @@ func New(
 	dashboard cephv1.DashboardSpec,
 	resources v1.ResourceRequirements,
 	ownerRef metav1.OwnerReference,
+	dataDirHostPath string,
 ) *Cluster {
 	return &Cluster{
-		clusterInfo: clusterInfo,
-		context:     context,
-		Namespace:   namespace,
-		placement:   placement,
-		rookVersion: rookVersion,
-		cephVersion: cephVersion,
-		Replicas:    1,
-		dataDir:     k8sutil.DataDir,
-		dashboard:   dashboard,
-		HostNetwork: hostNetwork,
-		resources:   resources,
-		ownerRef:    ownerRef,
-		exitCode:    getExitCode,
+		clusterInfo:     clusterInfo,
+		context:         context,
+		Namespace:       namespace,
+		placement:       placement,
+		rookVersion:     rookVersion,
+		cephVersion:     cephVersion,
+		Replicas:        1,
+		dataDir:         k8sutil.DataDir,
+		dashboard:       dashboard,
+		HostNetwork:     hostNetwork,
+		resources:       resources,
+		ownerRef:        ownerRef,
+		exitCode:        getExitCode,
+		dataDirHostPath: dataDirHostPath,
 	}
 }
 
@@ -115,7 +118,7 @@ func (c *Cluster) Start() error {
 			DaemonID:      daemonID,
 			ResourceName:  resourceName,
 			DashboardPort: c.dashboardPort(),
-			DataPathMap:   config.NewStatelessDaemonDataPathMap(config.MgrType, daemonID),
+			DataPathMap:   config.NewStatelessDaemonDataPathMap(config.MgrType, daemonID, c.Namespace, c.dataDirHostPath),
 		}
 
 		// generate keyring specific to this mgr daemon saved to k8s secret
diff --git a/pkg/operator/ceph/cluster/mgr/mgr_test.go b/pkg/operator/ceph/cluster/mgr/mgr_test.go
index c5eadc30f..1bd7366dd 100755
--- a/pkg/operator/ceph/cluster/mgr/mgr_test.go
+++ b/pkg/operator/ceph/cluster/mgr/mgr_test.go
@@ -31,7 +31,7 @@ import (
 	exectest "github.com/rook/rook/pkg/util/exec/test"
 	"github.com/stretchr/testify/assert"
 	apps "k8s.io/api/apps/v1"
-	"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"
 )
@@ -64,6 +64,7 @@ func TestStartMGR(t *testing.T) {
 		cephv1.DashboardSpec{Enabled: true},
 		v1.ResourceRequirements{},
 		metav1.OwnerReference{},
+		"/var/lib/rook/",
 	)
 	defer os.RemoveAll(c.dataDir)
 
diff --git a/pkg/operator/ceph/cluster/mgr/spec_test.go b/pkg/operator/ceph/cluster/mgr/spec_test.go
index 91af00a9f..759871850 100644
--- a/pkg/operator/ceph/cluster/mgr/spec_test.go
+++ b/pkg/operator/ceph/cluster/mgr/spec_test.go
@@ -54,13 +54,14 @@ func TestPodSpec(t *testing.T) {
 			},
 		},
 		metav1.OwnerReference{},
+		"/var/lib/rook/",
 	)
 
 	mgrTestConfig := mgrConfig{
 		DaemonID:      "a",
 		ResourceName:  "rook-ceph-mgr-a",
 		DashboardPort: 1234,
-		DataPathMap:   config.NewStatelessDaemonDataPathMap(config.MgrType, "a"),
+		DataPathMap:   config.NewStatelessDaemonDataPathMap(config.MgrType, "a", "rook-ceph", "/var/lib/rook/"),
 	}
 
 	d := c.makeDeployment(&mgrTestConfig)
@@ -91,6 +92,7 @@ func TestServiceSpec(t *testing.T) {
 		cephv1.DashboardSpec{},
 		v1.ResourceRequirements{},
 		metav1.OwnerReference{},
+		"/var/lib/rook/",
 	)
 
 	s := c.makeMetricsService("rook-mgr")
@@ -112,13 +114,14 @@ func TestHostNetwork(t *testing.T) {
 		cephv1.DashboardSpec{},
 		v1.ResourceRequirements{},
 		metav1.OwnerReference{},
+		"/var/lib/rook/",
 	)
 
 	mgrTestConfig := mgrConfig{
 		DaemonID:      "a",
 		ResourceName:  "mgr-a",
 		DashboardPort: 1234,
-		DataPathMap:   config.NewStatelessDaemonDataPathMap(config.MgrType, "a"),
+		DataPathMap:   config.NewStatelessDaemonDataPathMap(config.MgrType, "a", "rook-ceph", "/var/lib/rook/"),
 	}
 
 	d := c.makeDeployment(&mgrTestConfig)
diff --git a/pkg/operator/ceph/cluster/mon/mon.go b/pkg/operator/ceph/cluster/mon/mon.go
index 7671cfbc0..de77788a7 100644
--- a/pkg/operator/ceph/cluster/mon/mon.go
+++ b/pkg/operator/ceph/cluster/mon/mon.go
@@ -301,7 +301,7 @@ func (c *Cluster) initMonConfig(size int) (int, []*monConfig) {
 			DaemonName:   monitor.Name,
 			Port:         cephutil.GetPortFromEndpoint(monitor.Endpoint),
 			DataPathMap: config.NewStatefulDaemonDataPathMap(
-				c.dataDirHostPath, dataDirRelativeHostPath(monitor.Name), config.MonType, monitor.Name),
+				c.dataDirHostPath, dataDirRelativeHostPath(monitor.Name), config.MonType, monitor.Name, c.Namespace),
 		})
 	}
 
@@ -323,7 +323,7 @@ func (c *Cluster) newMonConfig(monID int) *monConfig {
 		DaemonName:   daemonName,
 		Port:         DefaultMsgr1Port,
 		DataPathMap: config.NewStatefulDaemonDataPathMap(
-			c.dataDirHostPath, dataDirRelativeHostPath(daemonName), config.MonType, daemonName),
+			c.dataDirHostPath, dataDirRelativeHostPath(daemonName), config.MonType, daemonName, c.Namespace),
 	}
 }
 
diff --git a/pkg/operator/ceph/cluster/mon/mon_test.go b/pkg/operator/ceph/cluster/mon/mon_test.go
index 425c615a8..8307b4b92 100644
--- a/pkg/operator/ceph/cluster/mon/mon_test.go
+++ b/pkg/operator/ceph/cluster/mon/mon_test.go
@@ -66,7 +66,7 @@ func testGenMonConfig(monID string) *monConfig {
 		PublicIP:     fmt.Sprintf("2.4.6.%d", index+1),
 		// dataDirHostPath assumed to be /var/lib/rook
 		DataPathMap: config.NewStatefulDaemonDataPathMap(
-			"/var/lib/rook", dataDirRelativeHostPath(monID), config.MonType, monID),
+			"/var/lib/rook", dataDirRelativeHostPath(monID), config.MonType, monID, "rook-ceph"),
 	}
 }
 
diff --git a/pkg/operator/ceph/cluster/osd/osd.go b/pkg/operator/ceph/cluster/osd/osd.go
index 17fe182d0..6d07fc8f2 100644
--- a/pkg/operator/ceph/cluster/osd/osd.go
+++ b/pkg/operator/ceph/cluster/osd/osd.go
@@ -28,6 +28,7 @@ import (
 	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"
 	osdconfig "github.com/rook/rook/pkg/operator/ceph/cluster/osd/config"
 	opspec "github.com/rook/rook/pkg/operator/ceph/spec"
 	"github.com/rook/rook/pkg/operator/discover"
@@ -58,6 +59,7 @@ const (
 
 // Cluster keeps track of the OSDs
 type Cluster struct {
+	clusterInfo     *cephconfig.ClusterInfo
 	context         *clusterd.Context
 	Namespace       string
 	placement       rookalpha.Placement
@@ -75,6 +77,7 @@ type Cluster struct {
 
 // New creates an instance of the OSD manager
 func New(
+	clusterInfo *cephconfig.ClusterInfo,
 	context *clusterd.Context,
 	namespace string,
 	rookVersion string,
@@ -88,6 +91,7 @@ func New(
 ) *Cluster {
 
 	return &Cluster{
+		clusterInfo:     clusterInfo,
 		context:         context,
 		Namespace:       namespace,
 		placement:       placement,
diff --git a/pkg/operator/ceph/cluster/osd/osd_test.go b/pkg/operator/ceph/cluster/osd/osd_test.go
index a22ab0f35..b3e6faf4a 100644
--- a/pkg/operator/ceph/cluster/osd/osd_test.go
+++ b/pkg/operator/ceph/cluster/osd/osd_test.go
@@ -23,15 +23,17 @@ import (
 	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/daemon/ceph/config"
 	discoverDaemon "github.com/rook/rook/pkg/daemon/discover"
 	"github.com/rook/rook/pkg/operator/ceph/cluster/osd/config"
+	cephver "github.com/rook/rook/pkg/operator/ceph/version"
 	"github.com/rook/rook/pkg/operator/k8sutil"
 	exectest "github.com/rook/rook/pkg/util/exec/test"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	apps "k8s.io/api/apps/v1"
-	"k8s.io/api/core/v1"
 	corev1 "k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/watch"
@@ -41,7 +43,10 @@ import (
 
 func TestStart(t *testing.T) {
 	clientset := fake.NewSimpleClientset()
-	c := New(&clusterd.Context{Clientset: clientset, ConfigDir: "/var/lib/rook", Executor: &exectest.MockExecutor{}}, "ns", "myversion", cephv1.CephVersionSpec{},
+	clusterInfo := &cephconfig.ClusterInfo{
+		CephVersion: cephver.Nautilus,
+	}
+	c := New(clusterInfo, &clusterd.Context{Clientset: clientset, ConfigDir: "/var/lib/rook", Executor: &exectest.MockExecutor{}}, "ns", "myversion", cephv1.CephVersionSpec{},
 		rookalpha.StorageScopeSpec{}, "", rookalpha.Placement{}, false, v1.ResourceRequirements{}, metav1.OwnerReference{})
 
 	// Start the first time
@@ -116,7 +121,10 @@ func TestAddRemoveNode(t *testing.T) {
 	statusMapWatcher := watch.NewFake()
 	clientset.PrependWatchReactor("configmaps", k8stesting.DefaultWatchReactor(statusMapWatcher, nil))
 
-	c := New(&clusterd.Context{Clientset: clientset, ConfigDir: "/var/lib/rook", Executor: &exectest.MockExecutor{}}, "ns-add-remove", "myversion", cephv1.CephVersionSpec{},
+	clusterInfo := &cephconfig.ClusterInfo{
+		CephVersion: cephver.Nautilus,
+	}
+	c := New(clusterInfo, &clusterd.Context{Clientset: clientset, ConfigDir: "/var/lib/rook", Executor: &exectest.MockExecutor{}}, "ns-add-remove", "myversion", cephv1.CephVersionSpec{},
 		storageSpec, "/foo", rookalpha.Placement{}, false, v1.ResourceRequirements{}, metav1.OwnerReference{})
 
 	// kick off the start of the orchestration in a goroutine
@@ -201,7 +209,7 @@ func TestAddRemoveNode(t *testing.T) {
 
 	// modify the storage spec to remove the node from the cluster
 	storageSpec.Nodes = []rookalpha.Node{}
-	c = New(&clusterd.Context{Clientset: clientset, ConfigDir: "/var/lib/rook", Executor: mockExec}, "ns-add-remove", "myversion", cephv1.CephVersionSpec{},
+	c = New(clusterInfo, &clusterd.Context{Clientset: clientset, ConfigDir: "/var/lib/rook", Executor: mockExec}, "ns-add-remove", "myversion", cephv1.CephVersionSpec{},
 		storageSpec, "", rookalpha.Placement{}, false, v1.ResourceRequirements{}, metav1.OwnerReference{})
 
 	// reset the orchestration status watcher
@@ -240,7 +248,10 @@ func TestGetIDFromDeployment(t *testing.T) {
 }
 
 func TestDiscoverOSDs(t *testing.T) {
-	c := New(&clusterd.Context{}, "ns", "myversion", cephv1.CephVersionSpec{},
+	clusterInfo := &cephconfig.ClusterInfo{
+		CephVersion: cephver.Nautilus,
+	}
+	c := New(clusterInfo, &clusterd.Context{}, "ns", "myversion", cephv1.CephVersionSpec{},
 		rookalpha.StorageScopeSpec{}, "", rookalpha.Placement{}, false, v1.ResourceRequirements{}, metav1.OwnerReference{})
 	node1 := "n1"
 	node2 := "n2"
@@ -308,7 +319,10 @@ func TestAddNodeFailure(t *testing.T) {
 	cmErr := createDiscoverConfigmap(nodeName, "rook-system", clientset)
 	assert.Nil(t, cmErr)
 
-	c := New(&clusterd.Context{Clientset: clientset, ConfigDir: "/var/lib/rook", Executor: &exectest.MockExecutor{}}, "ns-add-remove", "myversion", cephv1.CephVersionSpec{},
+	clusterInfo := &cephconfig.ClusterInfo{
+		CephVersion: cephver.Nautilus,
+	}
+	c := New(clusterInfo, &clusterd.Context{Clientset: clientset, ConfigDir: "/var/lib/rook", Executor: &exectest.MockExecutor{}}, "ns-add-remove", "myversion", cephv1.CephVersionSpec{},
 		storageSpec, "/foo", rookalpha.Placement{}, false, v1.ResourceRequirements{}, metav1.OwnerReference{})
 
 	// kick off the start of the orchestration in a goroutine
diff --git a/pkg/operator/ceph/cluster/osd/spec.go b/pkg/operator/ceph/cluster/osd/spec.go
index 660ec07be..078047d72 100644
--- a/pkg/operator/ceph/cluster/osd/spec.go
+++ b/pkg/operator/ceph/cluster/osd/spec.go
@@ -29,6 +29,7 @@ import (
 	opmon "github.com/rook/rook/pkg/operator/ceph/cluster/mon"
 	"github.com/rook/rook/pkg/operator/ceph/cluster/osd/config"
 	opspec "github.com/rook/rook/pkg/operator/ceph/spec"
+	"github.com/rook/rook/pkg/operator/ceph/version"
 	"github.com/rook/rook/pkg/operator/k8sutil"
 	apps "k8s.io/api/apps/v1"
 	batch "k8s.io/api/batch/v1"
@@ -83,7 +84,7 @@ func (c *Cluster) makeDeployment(nodeName string, selection rookalpha.Selection,
 	replicaCount := int32(1)
 	volumeMounts := opspec.CephVolumeMounts()
 	configVolumeMounts := opspec.RookVolumeMounts()
-	volumes := opspec.PodVolumes(c.dataDirHostPath)
+	volumes := opspec.PodVolumes(c.dataDirHostPath, c.Namespace)
 
 	var dataDir string
 	if osd.IsDirectory {
@@ -136,6 +137,7 @@ func (c *Cluster) makeDeployment(nodeName string, selection rookalpha.Selection,
 	configEnvVars := append(c.getConfigEnvVars(storeConfig, dataDir, nodeName, location), []v1.EnvVar{
 		tiniEnvVar,
 		{Name: "ROOK_OSD_ID", Value: osdID},
+		{Name: "ROOK_CEPH_VERSION", Value: c.clusterInfo.CephVersion.CephVersionFormatted()},
 	}...)
 
 	if !osd.IsDirectory {
@@ -163,6 +165,10 @@ func (c *Cluster) makeDeployment(nodeName string, selection rookalpha.Selection,
 		commonArgs = append(commonArgs, fmt.Sprintf("--osd-journal=%s", osd.Journal))
 	}
 
+	if c.clusterInfo.CephVersion.IsAtLeast(version.CephVersion{Major: 14, Minor: 2, Extra: 1}) {
+		commonArgs = append(commonArgs, "--default-log-to-file", "false")
+	}
+
 	// Add the volume to the spec and the mount to the daemon container
 	copyBinariesVolume, copyBinariesContainer := c.getCopyBinariesContainer()
 	volumes = append(volumes, copyBinariesVolume)
@@ -207,6 +213,10 @@ func (c *Cluster) makeDeployment(nodeName string, selection rookalpha.Selection,
 			}
 		}
 
+		if c.clusterInfo.CephVersion.IsAtLeast(version.CephVersion{Major: 14, Minor: 2, Extra: 1}) {
+			args = append(args, "--default-log-to-file", "false")
+		}
+
 	} else {
 		// other osds can launch the osd daemon directly
 		command = []string{"ceph-osd"}
@@ -324,7 +334,7 @@ func (c *Cluster) provisionPodTemplateSpec(devices []rookalpha.Device, selection
 
 	copyBinariesVolume, copyBinariesContainer := c.getCopyBinariesContainer()
 
-	volumes := append(opspec.PodVolumes(c.dataDirHostPath), copyBinariesVolume)
+	volumes := append(opspec.PodVolumes(c.dataDirHostPath, c.Namespace), copyBinariesVolume)
 
 	// by default, don't define any volume config unless it is required
 	if len(devices) > 0 || selection.DeviceFilter != "" || selection.GetUseAllDevices() || metadataDevice != "" {
diff --git a/pkg/operator/ceph/cluster/osd/spec_test.go b/pkg/operator/ceph/cluster/osd/spec_test.go
index 6f63970dd..5ff2054f0 100644
--- a/pkg/operator/ceph/cluster/osd/spec_test.go
+++ b/pkg/operator/ceph/cluster/osd/spec_test.go
@@ -23,7 +23,9 @@ import (
 	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/daemon/ceph/config"
 	"github.com/rook/rook/pkg/operator/ceph/cluster/osd/config"
+	cephver "github.com/rook/rook/pkg/operator/ceph/version"
 	"github.com/rook/rook/pkg/operator/k8sutil"
 	exectest "github.com/rook/rook/pkg/util/exec/test"
 	"github.com/stretchr/testify/assert"
@@ -73,7 +75,10 @@ func testPodDevices(t *testing.T, dataDir, deviceName string, allDevices bool) {
 
 	clientset := fake.NewSimpleClientset()
 	cephVersion := cephv1.CephVersionSpec{Image: "ceph/ceph:v12.2.8"}
-	c := New(&clusterd.Context{Clientset: clientset, ConfigDir: "/var/lib/rook", Executor: &exectest.MockExecutor{}}, "ns", "rook/rook:myversion", cephVersion,
+	clusterInfo := &cephconfig.ClusterInfo{
+		CephVersion: cephver.Nautilus,
+	}
+	c := New(clusterInfo, &clusterd.Context{Clientset: clientset, ConfigDir: "/var/lib/rook", Executor: &exectest.MockExecutor{}}, "ns", "rook/rook:myversion", cephVersion,
 		storageSpec, dataDir, rookalpha.Placement{}, false, v1.ResourceRequirements{}, metav1.OwnerReference{})
 
 	devMountNeeded := deviceName != "" || allDevices
@@ -96,10 +101,10 @@ func testPodDevices(t *testing.T, dataDir, deviceName string, allDevices bool) {
 	assert.Equal(t, "node1", deployment.Spec.Template.Spec.NodeSelector[apis.LabelHostname])
 	assert.Equal(t, v1.RestartPolicyAlways, deployment.Spec.Template.Spec.RestartPolicy)
 	if devMountNeeded && len(dataDir) > 0 {
-		assert.Equal(t, 5, len(deployment.Spec.Template.Spec.Volumes))
+		assert.Equal(t, 6, len(deployment.Spec.Template.Spec.Volumes))
 	}
 	if devMountNeeded && len(dataDir) == 0 {
-		assert.Equal(t, 5, len(deployment.Spec.Template.Spec.Volumes))
+		assert.Equal(t, 6, len(deployment.Spec.Template.Spec.Volumes))
 	}
 	if !devMountNeeded && len(dataDir) > 0 {
 		assert.Equal(t, 2, len(deployment.Spec.Template.Spec.Volumes))
@@ -117,12 +122,12 @@ func testPodDevices(t *testing.T, dataDir, deviceName string, allDevices bool) {
 	initCont := deployment.Spec.Template.Spec.InitContainers[0]
 	assert.Equal(t, "rook/rook:myversion", initCont.Image)
 	assert.Equal(t, "config-init", initCont.Name)
-	assert.Equal(t, 3, len(initCont.VolumeMounts))
+	assert.Equal(t, 4, len(initCont.VolumeMounts))
 
 	assert.Equal(t, 1, len(deployment.Spec.Template.Spec.Containers))
 	cont := deployment.Spec.Template.Spec.Containers[0]
 	assert.Equal(t, cephVersion.Image, cont.Image)
-	assert.Equal(t, 4, len(cont.VolumeMounts))
+	assert.Equal(t, 5, len(cont.VolumeMounts))
 	assert.Equal(t, "ceph-osd", cont.Command[0])
 }
 
@@ -156,7 +161,10 @@ func TestStorageSpecDevicesAndDirectories(t *testing.T) {
 	}
 
 	clientset := fake.NewSimpleClientset()
-	c := New(&clusterd.Context{Clientset: clientset, ConfigDir: "/var/lib/rook", Executor: &exectest.MockExecutor{}}, "ns", "rook/rook:myversion", cephv1.CephVersionSpec{},
+	clusterInfo := &cephconfig.ClusterInfo{
+		CephVersion: cephver.Nautilus,
+	}
+	c := New(clusterInfo, &clusterd.Context{Clientset: clientset, ConfigDir: "/var/lib/rook", Executor: &exectest.MockExecutor{}}, "ns", "rook/rook:myversion", cephv1.CephVersionSpec{},
 		storageSpec, "/var/lib/rook", rookalpha.Placement{}, false, v1.ResourceRequirements{}, metav1.OwnerReference{})
 
 	n := c.DesiredStorage.ResolveNode(storageSpec.Nodes[0].Name)
@@ -170,21 +178,23 @@ func TestStorageSpecDevicesAndDirectories(t *testing.T) {
 	assert.Nil(t, err)
 	// pod spec should have a volume for the given dir in the main container and the init container
 	podSpec := deployment.Spec.Template.Spec
-	assert.Equal(t, 5, len(podSpec.Volumes))
+	assert.Equal(t, 6, len(podSpec.Volumes))
 	require.Equal(t, 1, len(podSpec.Containers))
 	cont := podSpec.Containers[0]
-	assert.Equal(t, 4, len(cont.VolumeMounts))
+	assert.Equal(t, 5, len(cont.VolumeMounts))
 	assert.Equal(t, "/var/lib/rook", cont.VolumeMounts[0].MountPath)
 	assert.Equal(t, "/etc/ceph", cont.VolumeMounts[1].MountPath)
-	assert.Equal(t, "/my/root/path", cont.VolumeMounts[2].MountPath)
+	assert.Equal(t, "/var/log/ceph", cont.VolumeMounts[2].MountPath)
+	assert.Equal(t, "/my/root/path", cont.VolumeMounts[3].MountPath)
 
 	require.Equal(t, 2, len(podSpec.InitContainers))
 	initCont := podSpec.InitContainers[0]
-	assert.Equal(t, 4, len(initCont.VolumeMounts))
+	assert.Equal(t, 5, len(initCont.VolumeMounts))
 	assert.Equal(t, "/var/lib/rook", initCont.VolumeMounts[0].MountPath)
 	assert.Equal(t, "/etc/ceph", initCont.VolumeMounts[1].MountPath)
-	assert.Equal(t, "/etc/rook/config", initCont.VolumeMounts[2].MountPath)
-	assert.Equal(t, "/my/root/path", initCont.VolumeMounts[3].MountPath)
+	assert.Equal(t, "/var/log/ceph", initCont.VolumeMounts[2].MountPath)
+	assert.Equal(t, "/etc/rook/config", initCont.VolumeMounts[3].MountPath)
+	assert.Equal(t, "/my/root/path", initCont.VolumeMounts[4].MountPath)
 
 	// the default osd created on a node will be under /var/lib/rook, which won't need an extra mount
 	osd = OSDInfo{
@@ -197,10 +207,10 @@ func TestStorageSpecDevicesAndDirectories(t *testing.T) {
 	assert.Nil(t, err)
 	// pod spec should have a volume for the given dir in the main container and the init container
 	podSpec = deployment.Spec.Template.Spec
-	assert.Equal(t, 4, len(podSpec.Volumes))
+	assert.Equal(t, 5, len(podSpec.Volumes))
 	require.Equal(t, 1, len(podSpec.Containers))
 	cont = podSpec.Containers[0]
-	require.Equal(t, 3, len(cont.VolumeMounts))
+	require.Equal(t, 4, len(cont.VolumeMounts))
 	assert.Equal(t, "/var/lib/rook", cont.VolumeMounts[0].MountPath)
 	assert.Equal(t, "/etc/ceph", cont.VolumeMounts[1].MountPath)
 
@@ -208,10 +218,11 @@ func TestStorageSpecDevicesAndDirectories(t *testing.T) {
 
 	require.Equal(t, 2, len(podSpec.InitContainers))
 	initCont = podSpec.InitContainers[0]
-	require.Equal(t, 3, len(initCont.VolumeMounts))
+	require.Equal(t, 4, len(initCont.VolumeMounts))
 	assert.Equal(t, "/var/lib/rook", initCont.VolumeMounts[0].MountPath)
 	assert.Equal(t, "/etc/ceph", initCont.VolumeMounts[1].MountPath)
-	assert.Equal(t, "/etc/rook/config", initCont.VolumeMounts[2].MountPath)
+	assert.Equal(t, "/var/log/ceph", initCont.VolumeMounts[2].MountPath)
+	assert.Equal(t, "/etc/rook/config", initCont.VolumeMounts[3].MountPath)
 }
 
 func TestStorageSpecConfig(t *testing.T) {
@@ -245,7 +256,10 @@ func TestStorageSpecConfig(t *testing.T) {
 	}
 
 	clientset := fake.NewSimpleClientset()
-	c := New(&clusterd.Context{Clientset: clientset, ConfigDir: "/var/lib/rook", Executor: &exectest.MockExecutor{}}, "ns", "rook/rook:myversion", cephv1.CephVersionSpec{},
+	clusterInfo := &cephconfig.ClusterInfo{
+		CephVersion: cephver.Nautilus,
+	}
+	c := New(clusterInfo, &clusterd.Context{Clientset: clientset, ConfigDir: "/var/lib/rook", Executor: &exectest.MockExecutor{}}, "ns", "rook/rook:myversion", cephv1.CephVersionSpec{},
 		storageSpec, "", rookalpha.Placement{}, false, v1.ResourceRequirements{}, metav1.OwnerReference{})
 
 	n := c.DesiredStorage.ResolveNode(storageSpec.Nodes[0].Name)
@@ -297,7 +311,10 @@ func TestHostNetwork(t *testing.T) {
 	}
 
 	clientset := fake.NewSimpleClientset()
-	c := New(&clusterd.Context{Clientset: clientset, ConfigDir: "/var/lib/rook", Executor: &exectest.MockExecutor{}}, "ns", "myversion", cephv1.CephVersionSpec{},
+	clusterInfo := &cephconfig.ClusterInfo{
+		CephVersion: cephver.Nautilus,
+	}
+	c := New(clusterInfo, &clusterd.Context{Clientset: clientset, ConfigDir: "/var/lib/rook", Executor: &exectest.MockExecutor{}}, "ns", "myversion", cephv1.CephVersionSpec{},
 		storageSpec, "", rookalpha.Placement{}, true, v1.ResourceRequirements{}, metav1.OwnerReference{})
 
 	n := c.DesiredStorage.ResolveNode(storageSpec.Nodes[0].Name)
diff --git a/pkg/operator/ceph/cluster/osd/status_test.go b/pkg/operator/ceph/cluster/osd/status_test.go
index 5091cc1d1..43ea75ca1 100644
--- a/pkg/operator/ceph/cluster/osd/status_test.go
+++ b/pkg/operator/ceph/cluster/osd/status_test.go
@@ -24,10 +24,12 @@ import (
 	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/daemon/ceph/config"
+	cephver "github.com/rook/rook/pkg/operator/ceph/version"
 	"github.com/rook/rook/pkg/operator/k8sutil"
 	exectest "github.com/rook/rook/pkg/util/exec/test"
 	"github.com/stretchr/testify/assert"
-	"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/apimachinery/pkg/watch"
@@ -36,7 +38,10 @@ import (
 
 func TestOrchestrationStatus(t *testing.T) {
 	clientset := fake.NewSimpleClientset()
-	c := New(&clusterd.Context{Clientset: clientset, ConfigDir: "/var/lib/rook", Executor: &exectest.MockExecutor{}}, "ns", "myversion", cephv1.CephVersionSpec{},
+	clusterInfo := &cephconfig.ClusterInfo{
+		CephVersion: cephver.Nautilus,
+	}
+	c := New(clusterInfo, &clusterd.Context{Clientset: clientset, ConfigDir: "/var/lib/rook", Executor: &exectest.MockExecutor{}}, "ns", "myversion", cephv1.CephVersionSpec{},
 		rookalpha.StorageScopeSpec{}, "", rookalpha.Placement{}, false, v1.ResourceRequirements{}, metav1.OwnerReference{})
 	kv := k8sutil.NewConfigMapKVStore(c.Namespace, clientset, metav1.OwnerReference{})
 	nodeName := "mynode"
diff --git a/pkg/operator/ceph/cluster/rbd/mirror.go b/pkg/operator/ceph/cluster/rbd/mirror.go
index 271432f61..85e95c89b 100644
--- a/pkg/operator/ceph/cluster/rbd/mirror.go
+++ b/pkg/operator/ceph/cluster/rbd/mirror.go
@@ -27,7 +27,7 @@ import (
 	cephconfig "github.com/rook/rook/pkg/daemon/ceph/config"
 	"github.com/rook/rook/pkg/operator/ceph/config"
 	"github.com/rook/rook/pkg/operator/k8sutil"
-	"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"
 )
@@ -40,16 +40,17 @@ const (
 
 // Mirroring represents the Rook and environment configuration settings needed to set up rbd mirroring.
 type Mirroring struct {
-	ClusterInfo *cephconfig.ClusterInfo
-	Namespace   string
-	placement   rookalpha.Placement
-	context     *clusterd.Context
-	resources   v1.ResourceRequirements
-	ownerRef    metav1.OwnerReference
-	spec        cephv1.RBDMirroringSpec
-	cephVersion cephv1.CephVersionSpec
-	rookVersion string
-	hostNetwork bool
+	ClusterInfo     *cephconfig.ClusterInfo
+	Namespace       string
+	placement       rookalpha.Placement
+	context         *clusterd.Context
+	resources       v1.ResourceRequirements
+	ownerRef        metav1.OwnerReference
+	spec            cephv1.RBDMirroringSpec
+	cephVersion     cephv1.CephVersionSpec
+	rookVersion     string
+	hostNetwork     bool
+	dataDirHostPath string
 }
 
 // New creates an instance of the rbd mirroring
@@ -63,18 +64,20 @@ func New(
 	spec cephv1.RBDMirroringSpec,
 	resources v1.ResourceRequirements,
 	ownerRef metav1.OwnerReference,
+	dataDirHostPath string,
 ) *Mirroring {
 	return &Mirroring{
-		ClusterInfo: cluster,
-		context:     context,
-		Namespace:   namespace,
-		placement:   placement,
-		rookVersion: rookVersion,
-		cephVersion: cephVersion,
-		spec:        spec,
-		hostNetwork: hostNetwork,
-		resources:   resources,
-		ownerRef:    ownerRef,
+		ClusterInfo:     cluster,
+		context:         context,
+		Namespace:       namespace,
+		placement:       placement,
+		rookVersion:     rookVersion,
+		cephVersion:     cephVersion,
+		spec:            spec,
+		hostNetwork:     hostNetwork,
+		resources:       resources,
+		ownerRef:        ownerRef,
+		dataDirHostPath: dataDirHostPath,
 	}
 }
 
@@ -90,7 +93,7 @@ func (m *Mirroring) Start() error {
 		daemonConf := &daemonConfig{
 			DaemonID:     daemonID,
 			ResourceName: resourceName,
-			DataPathMap:  config.NewDatalessDaemonDataPathMap(),
+			DataPathMap:  config.NewDatalessDaemonDataPathMap(m.Namespace, m.dataDirHostPath),
 		}
 
 		if err := m.generateKeyring(daemonConf); err != nil {
diff --git a/pkg/operator/ceph/cluster/rbd/mirror_test.go b/pkg/operator/ceph/cluster/rbd/mirror_test.go
index 4055609c0..1dc3de8e8 100644
--- a/pkg/operator/ceph/cluster/rbd/mirror_test.go
+++ b/pkg/operator/ceph/cluster/rbd/mirror_test.go
@@ -26,7 +26,7 @@ import (
 	testop "github.com/rook/rook/pkg/operator/test"
 	exectest "github.com/rook/rook/pkg/util/exec/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"
 )
 
@@ -55,6 +55,7 @@ func TestRBDMirror(t *testing.T) {
 		cephv1.RBDMirroringSpec{Workers: 2},
 		v1.ResourceRequirements{},
 		metav1.OwnerReference{},
+		"/var/lib/rook/",
 	)
 
 	err := c.Start()
diff --git a/pkg/operator/ceph/cluster/rbd/spec_test.go b/pkg/operator/ceph/cluster/rbd/spec_test.go
index e75c4f100..345cc7fa5 100644
--- a/pkg/operator/ceph/cluster/rbd/spec_test.go
+++ b/pkg/operator/ceph/cluster/rbd/spec_test.go
@@ -53,11 +53,12 @@ func TestPodSpec(t *testing.T) {
 			},
 		},
 		metav1.OwnerReference{},
+		"/var/lib/rook/",
 	)
 	daemonConf := daemonConfig{
 		DaemonID:     "a",
 		ResourceName: "rook-ceph-rbd-mirror-a",
-		DataPathMap:  config.NewDatalessDaemonDataPathMap(),
+		DataPathMap:  config.NewDatalessDaemonDataPathMap("rook-ceph", "/var/lib/rook"),
 	}
 
 	d := c.makeDeployment(&daemonConf)
diff --git a/pkg/operator/ceph/config/config.go b/pkg/operator/ceph/config/config.go
index f5568918f..c2c0bafe7 100644
--- a/pkg/operator/ceph/config/config.go
+++ b/pkg/operator/ceph/config/config.go
@@ -50,16 +50,23 @@ const (
 	// RgwType defines the rgw DaemonType
 	RgwType DaemonType = "rgw"
 
+	// RbdMirrorType defines the rbd-mirror DaemonType
 	RbdMirrorType DaemonType = "rbd-mirror"
 )
 
-// VarLibCephDir is simply "/var/lib/ceph". It is made overwriteable only for unit tests where it
-// may be needed to send data intended for /var/lib/ceph to a temporary test dir.
-var VarLibCephDir = "/var/lib/ceph"
+var (
+	// VarLibCephDir is simply "/var/lib/ceph". It is made overwriteable only for unit tests where it
+	// may be needed to send data intended for /var/lib/ceph to a temporary test dir.
+	VarLibCephDir = "/var/lib/ceph"
 
-// EtcCephDir is simply "/etc/ceph". It is made overwriteable only for unit tests where it
-// may be needed to send data intended for /etc/ceph to a temporary test dir.
-var EtcCephDir = "/etc/ceph"
+	// EtcCephDir is simply "/etc/ceph". It is made overwriteable only for unit tests where it
+	// may be needed to send data intended for /etc/ceph to a temporary test dir.
+	EtcCephDir = "/etc/ceph"
+
+	// VarLogCephDir defines Ceph logging directory. It is made overwriteable only for unit tests where it
+	// may be needed to send data intended for /var/log/ceph to a temporary test dir.
+	VarLogCephDir = "/var/log/ceph"
+)
 
 // Config is a Ceph config struct containing zero or more Ceph config sections.
 type Config struct {
diff --git a/pkg/operator/ceph/config/datapath.go b/pkg/operator/ceph/config/datapath.go
index 3a0aa5116..4b5fc7030 100644
--- a/pkg/operator/ceph/config/datapath.go
+++ b/pkg/operator/ceph/config/datapath.go
@@ -38,6 +38,12 @@ type DataPathMap struct {
 	// ContainerDataDir should be set to the path in the container where the specific daemon's data
 	// is stored.
 	ContainerDataDir string
+
+	// HostLogDir represents Ceph's logging directory on the host
+	HostLogDir string
+
+	// ContainerLogDir represents Ceph's logging directory
+	ContainerLogDir string
 }
 
 // NewStatefulDaemonDataPathMap returns a new DataPathMap for a daemon which requires a persistent
@@ -47,35 +53,42 @@ type DataPathMap struct {
 // which may include data from other daemons.
 func NewStatefulDaemonDataPathMap(
 	dataDirHostPath, daemonDataDirHostRelativePath string,
-	daemonType DaemonType, daemonID string,
+	daemonType DaemonType, daemonID, namespace string,
 ) *DataPathMap {
 	return &DataPathMap{
 		PersistData:      true,
 		HostDataDir:      path.Join(dataDirHostPath, daemonDataDirHostRelativePath),
 		NoData:           false,
 		ContainerDataDir: cephDataDir(daemonType, daemonID),
+		HostLogDir:       path.Join(dataDirHostPath, namespace, "log"),
+		ContainerLogDir:  VarLogCephDir,
 	}
 }
 
 // NewStatelessDaemonDataPathMap returns a new DataPathMap for a daemon which does not persist data
 // to the host (mgrs, mdses, rgws)
 func NewStatelessDaemonDataPathMap(
-	daemonType DaemonType, daemonID string,
+	daemonType DaemonType, daemonID, namespace, dataDirHostPath string,
 ) *DataPathMap {
 	return &DataPathMap{
 		PersistData:      false,
 		HostDataDir:      "",
 		NoData:           false,
 		ContainerDataDir: cephDataDir(daemonType, daemonID),
+		HostLogDir:       path.Join(dataDirHostPath, namespace, "log"),
+		ContainerLogDir:  VarLogCephDir,
 	}
 }
 
-func NewDatalessDaemonDataPathMap() *DataPathMap {
+// NewDatalessDaemonDataPathMap returns a new DataPathMap for a daemon which does not utilize a data dir in the container as the mon, mgr, osd, mds, and rgw daemons do
+func NewDatalessDaemonDataPathMap(namespace, dataDirHostPath string) *DataPathMap {
 	return &DataPathMap{
 		PersistData:      false,
 		HostDataDir:      "",
 		NoData:           true,
 		ContainerDataDir: "",
+		HostLogDir:       path.Join(dataDirHostPath, namespace, "log"),
+		ContainerLogDir:  VarLogCephDir,
 	}
 }
 
diff --git a/pkg/operator/ceph/config/datapath_test.go b/pkg/operator/ceph/config/datapath_test.go
index 232e05b1a..8ec776f8f 100644
--- a/pkg/operator/ceph/config/datapath_test.go
+++ b/pkg/operator/ceph/config/datapath_test.go
@@ -24,44 +24,54 @@ import (
 
 func TestNewStatefulDaemonDataPathMap(t *testing.T) {
 	// mon
-	d := NewStatefulDaemonDataPathMap("/var/lib/rook", "/mon-a/data", MonType, "a")
+	d := NewStatefulDaemonDataPathMap("/var/lib/rook", "/mon-a/data", MonType, "a", "rook-ceph")
 	assert.Equal(t, &DataPathMap{
 		PersistData:      true,
 		HostDataDir:      "/var/lib/rook/mon-a/data",
 		ContainerDataDir: "/var/lib/ceph/mon/ceph-a",
+		ContainerLogDir:  "/var/log/ceph",
+		HostLogDir:       "/var/lib/rook/rook-ceph/log",
 	}, d)
 
 	// osd
-	d = NewStatefulDaemonDataPathMap("/var/lib/rook/", "osd0/", OsdType, "0")
+	d = NewStatefulDaemonDataPathMap("/var/lib/rook/", "osd0/", OsdType, "0", "rook-ceph")
 	assert.Equal(t, &DataPathMap{
 		PersistData:      true,
 		HostDataDir:      "/var/lib/rook/osd0",
 		ContainerDataDir: "/var/lib/ceph/osd/ceph-0",
+		ContainerLogDir:  "/var/log/ceph",
+		HostLogDir:       "/var/lib/rook/rook-ceph/log",
 	}, d)
 }
 
 func TestNewStatelessDaemonDataPathMap(t *testing.T) {
 	// mgr
-	d := NewStatelessDaemonDataPathMap(MgrType, "a")
+	d := NewStatelessDaemonDataPathMap(MgrType, "a", "rook-ceph", "/var/lib/rook")
 	assert.Equal(t, &DataPathMap{
 		PersistData:      false,
 		HostDataDir:      "",
 		ContainerDataDir: "/var/lib/ceph/mgr/ceph-a",
+		ContainerLogDir:  "/var/log/ceph",
+		HostLogDir:       "/var/lib/rook/rook-ceph/log",
 	}, d)
 
 	// mds
-	d = NewStatelessDaemonDataPathMap(MdsType, "myfs.a")
+	d = NewStatelessDaemonDataPathMap(MdsType, "myfs.a", "rook-ceph", "/var/lib/rook")
 	assert.Equal(t, &DataPathMap{
 		PersistData:      false,
 		HostDataDir:      "",
 		ContainerDataDir: "/var/lib/ceph/mds/ceph-myfs.a",
+		ContainerLogDir:  "/var/log/ceph",
+		HostLogDir:       "/var/lib/rook/rook-ceph/log",
 	}, d)
 
 	// rgw
-	d = NewStatelessDaemonDataPathMap(RgwType, "objstore")
+	d = NewStatelessDaemonDataPathMap(RgwType, "objstore", "rook-ceph", "/var/lib/rook")
 	assert.Equal(t, &DataPathMap{
 		PersistData:      false,
 		HostDataDir:      "",
 		ContainerDataDir: "/var/lib/ceph/rgw/ceph-objstore",
+		ContainerLogDir:  "/var/log/ceph",
+		HostLogDir:       "/var/lib/rook/rook-ceph/log",
 	}, d)
 }
diff --git a/pkg/operator/ceph/config/defaults.go b/pkg/operator/ceph/config/defaults.go
index 09a644e00..5c8d2818b 100644
--- a/pkg/operator/ceph/config/defaults.go
+++ b/pkg/operator/ceph/config/defaults.go
@@ -17,36 +17,18 @@ limitations under the License.
 // Package config provides default configurations which Rook will set in Ceph clusters.
 package config
 
-import (
-	"bytes"
-	"fmt"
-)
+import "github.com/rook/rook/pkg/operator/ceph/version"
 
 // DefaultFlags returns the default configuration flags Rook will set on the command line for all
 // calls to Ceph daemons and tools. Values specified here will not be able to be overridden using
 // the mon's central KV store, and that is (and should be) by intent.
-func DefaultFlags(fsid, mountedKeyringPath string) []string {
-	return defaultFlagConfigs(fsid, mountedKeyringPath).GlobalFlags()
+func DefaultFlags(fsid, mountedKeyringPath string, cephVersion version.CephVersion) []string {
+	return defaultFlagConfigs(fsid, mountedKeyringPath, cephVersion).GlobalFlags()
 }
 
-// DefaultFlagsAsConfigFile is the same as DefaultFlags except that it returns the ceph.conf file
-// text which corresponds to the flags. In some versions of Ceph and for some Daemons, the default
-// flags are not accepted in flag form.
-func DefaultFlagsAsConfigFile(fsid, mountedKeyringPath string) (string, error) {
-	c := defaultFlagConfigs(fsid, mountedKeyringPath)
-	f, err := c.IniFile()
-	if err != nil {
-		return "", fmt.Errorf("failed to generate text for default flags compatible with ceph.conf. %v", err)
-	}
-	b := new(bytes.Buffer)
-	if _, err := f.WriteTo(b); err != nil {
-		return "", fmt.Errorf("failed to generate text for default flags compatible with ceph.conf. %v", err)
-	}
-	return b.String(), nil
-}
-
-func defaultFlagConfigs(fsid, mountedKeyringPath string) *Config {
+func defaultFlagConfigs(fsid, mountedKeyringPath string, cephVersion version.CephVersion) *Config {
 	c := NewConfig()
+
 	c.Section("global").
 		// fsid unnecessary but is a safety to make sure daemons can only connect to their cluster
 		Set("fsid", fsid).
@@ -57,6 +39,17 @@ func defaultFlagConfigs(fsid, mountedKeyringPath string) *Config {
 		Set("mon-cluster-log-to-stderr", "true").
 		Set("log-stderr-prefix", "debug ")
 		// ^ differentiate debug text from audit text, and the space after 'debug' is critical
+
+	// As of Nautilus 14.2.1 at least
+	// These new flags control Ceph's daemon logging behaviour to files
+	// By default we set them to False so no logs get written on file
+	// However they can be actived at any time via the centralized config store
+	if cephVersion.IsAtLeast(version.CephVersion{Major: 14, Minor: 2, Extra: 1}) {
+		c.Section("global").
+			Set("default-log-to-file", "false").
+			Set("default-mon-cluster-log-to-file", "false")
+	}
+
 	m := StoredMonHostEnvVarReferences()
 	c.Merge(m)
 	return c
@@ -65,15 +58,24 @@ func defaultFlagConfigs(fsid, mountedKeyringPath string) *Config {
 // DefaultCentralizedConfigs returns the default configuration options Rook will set in Ceph's
 // centralized config store. If the version of Ceph does not support the centralized config store,
 // these will be set in a shared config file instead.
-func DefaultCentralizedConfigs() *Config {
+func DefaultCentralizedConfigs(cephVersion version.CephVersion) *Config {
 	c := NewConfig()
 	c.Section("global").
+		Set("mon allow pool delete", "true")
+
+	// Everything before Nautilus 14.2.1
+	// Prior to Nautilus 14.2.1 certain log flags were not present
+	// so in order to not log anything on files we must set the following flags to null
+	// Since Nautilus 14.2.1 introduced both 'default-log-to-file' and 'default-mon-cluster-log-to-file' (see above defaultFlagConfigs)
+	// these are not needed
+	if !cephVersion.IsAtLeast(version.CephVersion{Major: 14, Minor: 2, Extra: 1}) {
 		// Set the default log files to empty so they don't bloat containers. Can be changed in
 		// Mimic+ by users if needed.
-		Set("log file", "").
-		Set("mon cluster log file", "").
-		//
-		Set("mon allow pool delete", "true")
+		c.Section("global").
+			Set("log file", "").
+			Set("mon cluster log file", "")
+	}
+
 	return c
 }
 
diff --git a/pkg/operator/ceph/config/store.go b/pkg/operator/ceph/config/store.go
index 68c038250..c0deaf8ce 100644
--- a/pkg/operator/ceph/config/store.go
+++ b/pkg/operator/ceph/config/store.go
@@ -69,7 +69,7 @@ func GetStore(context *clusterd.Context, namespace string, ownerRef *metav1.Owne
 
 // CreateOrUpdate creates or updates the stored Ceph config based on the cluster info.
 func (s *Store) CreateOrUpdate(clusterInfo *cephconfig.ClusterInfo) error {
-	c := DefaultCentralizedConfigs()
+	c := DefaultCentralizedConfigs(clusterInfo.CephVersion)
 
 	// DefaultLegacyConfigs need to be added to the Ceph config file until the integration tests can be
 	// made to override these options for the Ceph clusters it creates.
diff --git a/pkg/operator/ceph/file/controller.go b/pkg/operator/ceph/file/controller.go
index 37089e856..491fa835a 100644
--- a/pkg/operator/ceph/file/controller.go
+++ b/pkg/operator/ceph/file/controller.go
@@ -57,12 +57,13 @@ var filesystemResourceRookLegacy = opkit.CustomResource{
 
 // FilesystemController represents a controller for filesystem custom resources
 type FilesystemController struct {
-	clusterInfo *cephconfig.ClusterInfo
-	context     *clusterd.Context
-	rookVersion string
-	cephVersion cephv1.CephVersionSpec
-	hostNetwork bool
-	ownerRef    metav1.OwnerReference
+	clusterInfo     *cephconfig.ClusterInfo
+	context         *clusterd.Context
+	rookVersion     string
+	cephVersion     cephv1.CephVersionSpec
+	hostNetwork     bool
+	ownerRef        metav1.OwnerReference
+	dataDirHostPath string
 }
 
 // NewFilesystemController create controller for watching filesystem custom resources created
@@ -73,14 +74,16 @@ func NewFilesystemController(
 	cephVersion cephv1.CephVersionSpec,
 	hostNetwork bool,
 	ownerRef metav1.OwnerReference,
+	dataDirHostPath string,
 ) *FilesystemController {
 	return &FilesystemController{
-		clusterInfo: clusterInfo,
-		context:     context,
-		rookVersion: rookVersion,
-		cephVersion: cephVersion,
-		hostNetwork: hostNetwork,
-		ownerRef:    ownerRef,
+		clusterInfo:     clusterInfo,
+		context:         context,
+		rookVersion:     rookVersion,
+		cephVersion:     cephVersion,
+		hostNetwork:     hostNetwork,
+		ownerRef:        ownerRef,
+		dataDirHostPath: dataDirHostPath,
 	}
 }
 
@@ -117,7 +120,7 @@ func (c *FilesystemController) onAdd(obj interface{}) {
 		return
 	}
 
-	err = createFilesystem(c.clusterInfo, c.context, *filesystem, c.rookVersion, c.cephVersion, c.hostNetwork, c.filesystemOwners(filesystem))
+	err = createFilesystem(c.clusterInfo, c.context, *filesystem, c.rookVersion, c.cephVersion, c.hostNetwork, c.filesystemOwners(filesystem), c.dataDirHostPath)
 	if err != nil {
 		logger.Errorf("failed to create filesystem %s: %+v", filesystem.Name, err)
 	}
@@ -149,7 +152,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))
+	err = createFilesystem(c.clusterInfo, c.context, *newFS, c.rookVersion, c.cephVersion, c.hostNetwork, c.filesystemOwners(newFS), c.dataDirHostPath)
 	if err != nil {
 		logger.Errorf("failed to create (modify) filesystem %s: %+v", newFS.Name, err)
 	}
diff --git a/pkg/operator/ceph/file/controller_test.go b/pkg/operator/ceph/file/controller_test.go
index 96de56478..b1510839d 100644
--- a/pkg/operator/ceph/file/controller_test.go
+++ b/pkg/operator/ceph/file/controller_test.go
@@ -29,7 +29,7 @@ import (
 
 	testop "github.com/rook/rook/pkg/operator/test"
 	"github.com/stretchr/testify/assert"
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -87,7 +87,7 @@ func TestMigrateFilesystemObject(t *testing.T) {
 	}
 	clusterInfo := &cephconfig.ClusterInfo{FSID: "myfsid"}
 
-	controller := NewFilesystemController(clusterInfo, context, "", cephv1.CephVersionSpec{}, false, metav1.OwnerReference{})
+	controller := NewFilesystemController(clusterInfo, context, "", cephv1.CephVersionSpec{}, false, metav1.OwnerReference{}, "/var/lib/rook/")
 
 	// convert the legacy filesystem object in memory and assert that a migration is needed
 	convertedFilesystem, migrationNeeded, err := getFilesystemObject(legacyFilesystem)
diff --git a/pkg/operator/ceph/file/filesystem.go b/pkg/operator/ceph/file/filesystem.go
index 04f64d7ca..fb0a1ab83 100644
--- a/pkg/operator/ceph/file/filesystem.go
+++ b/pkg/operator/ceph/file/filesystem.go
@@ -53,6 +53,7 @@ func createFilesystem(
 	cephVersion cephv1.CephVersionSpec,
 	hostNetwork bool,
 	ownerRefs []metav1.OwnerReference,
+	dataDirHostPath string,
 ) error {
 	if err := validateFilesystem(context, fs); err != nil {
 		return err
@@ -91,7 +92,7 @@ func createFilesystem(
 	}
 
 	logger.Infof("start running mdses for filesystem %s", fs.Name)
-	c := mds.NewCluster(clusterInfo, context, rookVersion, cephVersion, hostNetwork, fs, filesystem, ownerRefs)
+	c := mds.NewCluster(clusterInfo, context, rookVersion, cephVersion, hostNetwork, 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 ad46720e2..37c2bec33 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{})
+	err := createFilesystem(clusterInfo, context, fs, "v0.1", cephv1.CephVersionSpec{}, false, []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{})
+	err = createFilesystem(clusterInfo, context, fs, "v0.1", cephv1.CephVersionSpec{}, false, []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{})
+	err = createFilesystem(clusterInfo, context, fs, "v0.1", cephv1.CephVersionSpec{}, false, []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{})
+	err := createFilesystem(clusterInfo, context, fs, "v0.1", cephv1.CephVersionSpec{}, false, []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{})
+	err = createFilesystem(clusterInfo, context, fs, "v0.1", cephv1.CephVersionSpec{}, false, []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 e40c3bd63..2090a78b7 100644
--- a/pkg/operator/ceph/file/mds/mds.go
+++ b/pkg/operator/ceph/file/mds/mds.go
@@ -51,14 +51,15 @@ const (
 
 // Cluster represents a Ceph mds cluster.
 type Cluster struct {
-	clusterInfo *cephconfig.ClusterInfo
-	context     *clusterd.Context
-	rookVersion string
-	cephVersion cephv1.CephVersionSpec
-	HostNetwork bool
-	fs          cephv1.CephFilesystem
-	fsID        string
-	ownerRefs   []metav1.OwnerReference
+	clusterInfo     *cephconfig.ClusterInfo
+	context         *clusterd.Context
+	rookVersion     string
+	cephVersion     cephv1.CephVersionSpec
+	HostNetwork     bool
+	fs              cephv1.CephFilesystem
+	fsID            string
+	ownerRefs       []metav1.OwnerReference
+	dataDirHostPath string
 }
 
 type mdsConfig struct {
@@ -77,16 +78,18 @@ func NewCluster(
 	fs cephv1.CephFilesystem,
 	fsdetails *client.CephFilesystemDetails,
 	ownerRefs []metav1.OwnerReference,
+	dataDirHostPath string,
 ) *Cluster {
 	return &Cluster{
-		clusterInfo: clusterInfo,
-		context:     context,
-		rookVersion: rookVersion,
-		cephVersion: cephVersion,
-		HostNetwork: hostNetwork,
-		fs:          fs,
-		fsID:        strconv.Itoa(fsdetails.ID),
-		ownerRefs:   ownerRefs,
+		clusterInfo:     clusterInfo,
+		context:         context,
+		rookVersion:     rookVersion,
+		cephVersion:     cephVersion,
+		HostNetwork:     hostNetwork,
+		fs:              fs,
+		fsID:            strconv.Itoa(fsdetails.ID),
+		ownerRefs:       ownerRefs,
+		dataDirHostPath: dataDirHostPath,
 	}
 }
 
@@ -128,7 +131,7 @@ func (c *Cluster) Start() error {
 		mdsConfig := &mdsConfig{
 			ResourceName: resourceName,
 			DaemonID:     daemonName,
-			DataPathMap:  config.NewStatelessDaemonDataPathMap(config.MdsType, daemonName),
+			DataPathMap:  config.NewStatelessDaemonDataPathMap(config.MdsType, daemonName, c.fs.Namespace, c.dataDirHostPath),
 		}
 
 		// start the deployment
diff --git a/pkg/operator/ceph/file/mds/spec_test.go b/pkg/operator/ceph/file/mds/spec_test.go
index 97b7966e6..3614866a0 100644
--- a/pkg/operator/ceph/file/mds/spec_test.go
+++ b/pkg/operator/ceph/file/mds/spec_test.go
@@ -70,11 +70,12 @@ func testDeploymentObject(hostNetwork bool) *apps.Deployment {
 		fs,
 		&client.CephFilesystemDetails{ID: 15},
 		[]metav1.OwnerReference{{}},
+		"/var/lib/rook/",
 	)
 	mdsTestConfig := &mdsConfig{
 		DaemonID:     "myfs-a",
 		ResourceName: "rook-ceph-mds-myfs-a",
-		DataPathMap:  config.NewStatelessDaemonDataPathMap(config.MdsType, "myfs-a"),
+		DataPathMap:  config.NewStatelessDaemonDataPathMap(config.MdsType, "myfs-a", "rook-ceph", "/var/lib/rook/"),
 	}
 	return c.makeDeployment(mdsTestConfig)
 }
diff --git a/pkg/operator/ceph/nfs/nfs.go b/pkg/operator/ceph/nfs/nfs.go
index e71d486ed..41eef31da 100644
--- a/pkg/operator/ceph/nfs/nfs.go
+++ b/pkg/operator/ceph/nfs/nfs.go
@@ -27,7 +27,7 @@ import (
 	opspec "github.com/rook/rook/pkg/operator/ceph/spec"
 	"github.com/rook/rook/pkg/operator/k8sutil"
 	batch "k8s.io/api/batch/v1"
-	"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"
 )
@@ -163,7 +163,7 @@ func (c *CephNFSController) runGaneshaRadosGraceJob(n cephv1.CephNFS, name, acti
 							VolumeMounts: opspec.RookVolumeMounts(),
 						},
 					},
-					Volumes:       opspec.PodVolumes(""),
+					Volumes:       opspec.PodVolumes("", ""),
 					RestartPolicy: v1.RestartPolicyOnFailure,
 				},
 			},
diff --git a/pkg/operator/ceph/nfs/spec.go b/pkg/operator/ceph/nfs/spec.go
index 9ca441ff0..27f56eae4 100644
--- a/pkg/operator/ceph/nfs/spec.go
+++ b/pkg/operator/ceph/nfs/spec.go
@@ -97,7 +97,7 @@ func (c *CephNFSController) makeDeployment(n cephv1.CephNFS, name, configName st
 		Containers:     []v1.Container{c.daemonContainer(n, name, binariesMount)},
 		RestartPolicy:  v1.RestartPolicyAlways,
 		Volumes: append(
-			opspec.PodVolumes(""),
+			opspec.PodVolumes("", ""),
 			configVolume,
 			binariesVolume,
 		),
diff --git a/pkg/operator/ceph/object/controller.go b/pkg/operator/ceph/object/controller.go
index 4ba616709..0a7224218 100644
--- a/pkg/operator/ceph/object/controller.go
+++ b/pkg/operator/ceph/object/controller.go
@@ -57,12 +57,13 @@ var ObjectStoreResourceRookLegacy = opkit.CustomResource{
 
 // ObjectStoreController represents a controller object for object store custom resources
 type ObjectStoreController struct {
-	clusterInfo *daemonconfig.ClusterInfo
-	context     *clusterd.Context
-	rookImage   string
-	cephVersion cephv1.CephVersionSpec
-	hostNetwork bool
-	ownerRef    metav1.OwnerReference
+	clusterInfo     *daemonconfig.ClusterInfo
+	context         *clusterd.Context
+	rookImage       string
+	cephVersion     cephv1.CephVersionSpec
+	hostNetwork     bool
+	ownerRef        metav1.OwnerReference
+	dataDirHostPath string
 }
 
 // NewObjectStoreController create controller for watching object store custom resources created
@@ -73,14 +74,16 @@ func NewObjectStoreController(
 	cephVersion cephv1.CephVersionSpec,
 	hostNetwork bool,
 	ownerRef metav1.OwnerReference,
+	dataDirHostPath string,
 ) *ObjectStoreController {
 	return &ObjectStoreController{
-		clusterInfo: clusterInfo,
-		context:     context,
-		rookImage:   rookImage,
-		cephVersion: cephVersion,
-		hostNetwork: hostNetwork,
-		ownerRef:    ownerRef,
+		clusterInfo:     clusterInfo,
+		context:         context,
+		rookImage:       rookImage,
+		cephVersion:     cephVersion,
+		hostNetwork:     hostNetwork,
+		ownerRef:        ownerRef,
+		dataDirHostPath: dataDirHostPath,
 	}
 }
 
@@ -125,7 +128,7 @@ func (c *ObjectStoreController) onAdd(obj interface{}) {
 		cephVersion: c.cephVersion,
 		hostNetwork: c.hostNetwork,
 		ownerRefs:   c.storeOwners(objectstore),
-		DataPathMap: cephconfig.NewStatelessDaemonDataPathMap(cephconfig.RgwType, objectstore.Name),
+		DataPathMap: cephconfig.NewStatelessDaemonDataPathMap(cephconfig.RgwType, objectstore.Name, c.clusterInfo.Name, c.dataDirHostPath),
 	}
 	if err = cfg.createStore(); err != nil {
 		logger.Errorf("failed to create object store %s. %+v", objectstore.Name, err)
@@ -166,7 +169,7 @@ func (c *ObjectStoreController) onUpdate(oldObj, newObj interface{}) {
 		c.cephVersion,
 		c.hostNetwork,
 		c.storeOwners(newStore),
-		cephconfig.NewStatelessDaemonDataPathMap(cephconfig.RgwType, newStore.Name),
+		cephconfig.NewStatelessDaemonDataPathMap(cephconfig.RgwType, newStore.Name, c.clusterInfo.Name, c.dataDirHostPath),
 	}
 	if err = cfg.updateStore(); err != nil {
 		logger.Errorf("failed to create (modify) object store %s. %+v", newStore.Name, err)
diff --git a/pkg/operator/ceph/object/controller_test.go b/pkg/operator/ceph/object/controller_test.go
index bdd0ba2fd..985a7705d 100644
--- a/pkg/operator/ceph/object/controller_test.go
+++ b/pkg/operator/ceph/object/controller_test.go
@@ -27,7 +27,7 @@ import (
 	"github.com/rook/rook/pkg/clusterd"
 	testop "github.com/rook/rook/pkg/operator/test"
 	"github.com/stretchr/testify/assert"
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -92,7 +92,7 @@ func TestMigrateObjectStoreObject(t *testing.T) {
 		RookClientset: rookfake.NewSimpleClientset(legacyObjectStore),
 	}
 	info := testop.CreateConfigDir(1)
-	controller := NewObjectStoreController(info, context, "", cephv1.CephVersionSpec{}, false, metav1.OwnerReference{})
+	controller := NewObjectStoreController(info, context, "", cephv1.CephVersionSpec{}, false, metav1.OwnerReference{}, "/var/lib/rook/")
 
 	// convert the legacy objectstore object in memory and assert that a migration is needed
 	convertedObjectStore, migrationNeeded, err := getObjectStoreObject(legacyObjectStore)
diff --git a/pkg/operator/ceph/object/rgw_test.go b/pkg/operator/ceph/object/rgw_test.go
index 3a7098868..4acc2f03b 100644
--- a/pkg/operator/ceph/object/rgw_test.go
+++ b/pkg/operator/ceph/object/rgw_test.go
@@ -49,7 +49,7 @@ func TestStartRGW(t *testing.T) {
 	context := &clusterd.Context{Clientset: clientset, Executor: executor, ConfigDir: configDir}
 	store := simpleStore()
 	version := "v1.1.0"
-	data := cephconfig.NewStatelessDaemonDataPathMap(cephconfig.RgwType, "my-fs")
+	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}
@@ -113,7 +113,7 @@ func TestCreateObjectStore(t *testing.T) {
 	clientset := testop.New(3)
 	context := &clusterd.Context{Executor: executor, Clientset: clientset}
 	info := testop.CreateConfigDir(1)
-	data := cephconfig.NewStatelessDaemonDataPathMap(cephconfig.RgwType, "my-fs")
+	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}
diff --git a/pkg/operator/ceph/object/spec_test.go b/pkg/operator/ceph/object/spec_test.go
index 412df4266..85c26f350 100644
--- a/pkg/operator/ceph/object/spec_test.go
+++ b/pkg/operator/ceph/object/spec_test.go
@@ -45,7 +45,7 @@ func TestPodSpecs(t *testing.T) {
 	}
 	info := testop.CreateConfigDir(1)
 	info.CephVersion = cephver.Mimic
-	data := cephconfig.NewStatelessDaemonDataPathMap(cephconfig.RgwType, "default")
+	data := cephconfig.NewStatelessDaemonDataPathMap(cephconfig.RgwType, "default", "rook-ceph", "/var/lib/rook/")
 
 	c := &clusterConfig{
 		clusterInfo: info,
@@ -77,7 +77,7 @@ func TestSSLPodSpec(t *testing.T) {
 	}
 	info := testop.CreateConfigDir(1)
 	info.CephVersion = cephver.Mimic
-	data := cephconfig.NewStatelessDaemonDataPathMap(cephconfig.RgwType, "default")
+	data := cephconfig.NewStatelessDaemonDataPathMap(cephconfig.RgwType, "default", "rook-ceph", "/var/lib/rook/")
 	store.Spec.Gateway.SSLCertificateRef = "mycert"
 	store.Spec.Gateway.SecurePort = 443
 
diff --git a/pkg/operator/ceph/spec/spec.go b/pkg/operator/ceph/spec/spec.go
index f4bd18410..ce930e302 100644
--- a/pkg/operator/ceph/spec/spec.go
+++ b/pkg/operator/ceph/spec/spec.go
@@ -19,6 +19,7 @@ package spec
 
 import (
 	"fmt"
+	"path"
 
 	"github.com/coreos/pkg/capnslog"
 	cephconfig "github.com/rook/rook/pkg/daemon/ceph/config"
@@ -33,12 +34,13 @@ const (
 	// ConfigInitContainerName is the name which is given to the config initialization container
 	// in all Ceph pods.
 	ConfigInitContainerName = "config-init"
+	logVolumeName           = "rook-ceph-log"
 )
 
 var logger = capnslog.NewPackageLogger("github.com/rook/rook", "ceph-spec")
 
 // PodVolumes fills in the volumes parameter with the common list of Kubernetes volumes for use in Ceph pods.
-func PodVolumes(dataDirHostPath string) []v1.Volume {
+func PodVolumes(dataDirHostPath, namespace string) []v1.Volume {
 	dataDirSource := v1.VolumeSource{EmptyDir: &v1.EmptyDirVolumeSource{}}
 	if dataDirHostPath != "" {
 		dataDirSource = v1.VolumeSource{HostPath: &v1.HostPathVolumeSource{Path: dataDirHostPath}}
@@ -47,6 +49,7 @@ func PodVolumes(dataDirHostPath string) []v1.Volume {
 		{Name: k8sutil.DataDirVolume, VolumeSource: dataDirSource},
 		cephconfig.DefaultConfigVolume(),
 		k8sutil.ConfigOverrideVolume(),
+		StoredLogVolume(path.Join(dataDirHostPath, "log", namespace)),
 	}
 }
 
@@ -55,6 +58,7 @@ func CephVolumeMounts() []v1.VolumeMount {
 	return []v1.VolumeMount{
 		{Name: k8sutil.DataDirVolume, MountPath: k8sutil.DataDir},
 		cephconfig.DefaultConfigMount(),
+		StoredLogVolumeMount(),
 		// Rook doesn't run in ceph containers, so it doesn't need the config override mounted
 	}
 }
@@ -72,6 +76,7 @@ func DaemonVolumes(dataPaths *config.DataPathMap, keyringResourceName string) []
 	vols := []v1.Volume{
 		config.StoredFileVolume(),
 		keyring.Volume().Resource(keyringResourceName),
+		StoredLogVolume(dataPaths.HostLogDir),
 	}
 	if dataPaths.NoData {
 		return vols
@@ -89,6 +94,7 @@ func DaemonVolumeMounts(dataPaths *config.DataPathMap, keyringResourceName strin
 	mounts := []v1.VolumeMount{
 		config.StoredFileVolumeMount(),
 		keyring.VolumeMount().Resource(keyringResourceName),
+		StoredLogVolumeMount(),
 	}
 	if dataPaths.NoData {
 		return mounts
@@ -101,7 +107,7 @@ func DaemonVolumeMounts(dataPaths *config.DataPathMap, keyringResourceName strin
 // DaemonFlags returns the command line flags used by all Ceph daemons.
 func DaemonFlags(cluster *cephconfig.ClusterInfo, daemonID string) []string {
 	return append(
-		config.DefaultFlags(cluster.FSID, keyring.VolumeMount().KeyringFilePath()),
+		config.DefaultFlags(cluster.FSID, keyring.VolumeMount().KeyringFilePath(), cluster.CephVersion),
 		config.NewFlag("id", daemonID),
 	)
 
@@ -109,7 +115,7 @@ func DaemonFlags(cluster *cephconfig.ClusterInfo, daemonID string) []string {
 
 // AdminFlags returns the command line flags used for Ceph commands requiring admin authentication.
 func AdminFlags(cluster *cephconfig.ClusterInfo) []string {
-	return config.DefaultFlags(cluster.FSID, keyring.VolumeMount().AdminKeyringFilePath())
+	return config.DefaultFlags(cluster.FSID, keyring.VolumeMount().AdminKeyringFilePath(), cluster.CephVersion)
 }
 
 // ContainerEnvVarReference returns a reference to a Kubernetes container env var of the given name
@@ -177,3 +183,22 @@ func CheckPodMemory(resources v1.ResourceRequirements, cephPodMinimumMemory uint
 
 	return nil
 }
+
+// StoredLogVolume returns a pod volume sourced from the stored log files.
+func StoredLogVolume(HostLogDir string) v1.Volume {
+	return v1.Volume{
+		Name: logVolumeName,
+		VolumeSource: v1.VolumeSource{
+			HostPath: &v1.HostPathVolumeSource{Path: HostLogDir},
+		},
+	}
+}
+
+// StoredLogVolumeMount returns a pod volume sourced from the stored log files.
+func StoredLogVolumeMount() v1.VolumeMount {
+	return v1.VolumeMount{
+		Name:      logVolumeName,
+		ReadOnly:  false,
+		MountPath: config.VarLogCephDir,
+	}
+}
diff --git a/pkg/operator/ceph/spec/spec_test.go b/pkg/operator/ceph/spec/spec_test.go
index 0cf3f0569..bf575541c 100644
--- a/pkg/operator/ceph/spec/spec_test.go
+++ b/pkg/operator/ceph/spec/spec_test.go
@@ -24,10 +24,10 @@ import (
 )
 
 func TestPodVolumes(t *testing.T) {
-	if err := test.VolumeIsEmptyDir(k8sutil.DataDirVolume, PodVolumes("")); err != nil {
+	if err := test.VolumeIsEmptyDir(k8sutil.DataDirVolume, PodVolumes("", "")); err != nil {
 		t.Errorf("PodVolumes(\"\") - data dir source is not EmptyDir: %s", err.Error())
 	}
-	if err := test.VolumeIsHostPath(k8sutil.DataDirVolume, "/dev/sdb", PodVolumes("/dev/sdb")); err != nil {
+	if err := test.VolumeIsHostPath(k8sutil.DataDirVolume, "/dev/sdb", PodVolumes("/dev/sdb", "rook-ceph")); err != nil {
 		t.Errorf("PodVolumes(\"/dev/sdb\") - data dir source is not HostPath: %s", err.Error())
 	}
 }
@@ -35,7 +35,7 @@ func TestPodVolumes(t *testing.T) {
 func TestMountsMatchVolumes(t *testing.T) {
 	volsMountsTestDef := test.VolumesAndMountsTestDefinition{
 		VolumesSpec: &test.VolumesSpec{
-			Moniker: "PodVolumes(\"/dev/sdc\")", Volumes: PodVolumes("/dev/sdc")},
+			Moniker: "PodVolumes(\"/dev/sdc\")", Volumes: PodVolumes("/dev/sdc", "rook-ceph")},
 		MountsSpecItems: []*test.MountsSpec{
 			{Moniker: "CephVolumeMounts()", Mounts: CephVolumeMounts()},
 			{Moniker: "RookVolumeMounts()", Mounts: RookVolumeMounts()}},
diff --git a/pkg/operator/ceph/version/version.go b/pkg/operator/ceph/version/version.go
index 6da2f2448..bc89bfa45 100644
--- a/pkg/operator/ceph/version/version.go
+++ b/pkg/operator/ceph/version/version.go
@@ -6,6 +6,7 @@ import (
 	"strconv"
 )
 
+// CephVersion represents the Ceph version format
 type CephVersion struct {
 	Major int
 	Minor int
@@ -41,6 +42,12 @@ func (v *CephVersion) String() string {
 		v.Major, v.Minor, v.Extra, v.ReleaseName())
 }
 
+// CephVersionFormatted returns the Ceph version in a human readable format
+func (v *CephVersion) CephVersionFormatted() string {
+	return fmt.Sprintf("ceph version %d.%d.%d %s",
+		v.Major, v.Minor, v.Extra, v.ReleaseName())
+}
+
 // ReleaseName is the name of the Ceph release
 func (v *CephVersion) ReleaseName() string {
 	switch v.Major {
@@ -57,6 +64,7 @@ func (v *CephVersion) ReleaseName() string {
 	}
 }
 
+// ExtractCephVersion extracts the major, minor and extra digit of a Ceph release
 func ExtractCephVersion(src string) (*CephVersion, error) {
 	m := versionPattern.FindStringSubmatch(src)
 	if m == nil {
@@ -81,6 +89,7 @@ func ExtractCephVersion(src string) (*CephVersion, error) {
 	return &CephVersion{major, minor, extra}, nil
 }
 
+// Supported checks if a given release is supported
 func (v *CephVersion) Supported() bool {
 	for _, sv := range supportedVersions {
 		if v.isRelease(sv) {
@@ -94,26 +103,31 @@ func (v *CephVersion) isRelease(other CephVersion) bool {
 	return v.Major == other.Major
 }
 
+// IsLuminous checks if the Ceph version if Luminous
 func (v *CephVersion) IsLuminous() bool {
 	return v.isRelease(Luminous)
 }
 
+// IsAtLeast checks a given Ceph version is at least a given one
 func (v *CephVersion) IsAtLeast(other CephVersion) bool {
 	if v.Major > other.Major {
 		return true
 	} else if v.Major < other.Major {
 		return false
 	}
+	// If we arrive here then v.Major == other.Major
 	if v.Minor > other.Minor {
 		return true
 	} else if v.Minor < other.Minor {
 		return false
 	}
+	// If we arrive here then v.Minor == other.Minor
 	if v.Extra > other.Extra {
 		return true
 	} else if v.Extra < other.Extra {
 		return false
 	}
+	// If we arrive here then both versions are identical
 	return true
 }
 
@@ -122,10 +136,12 @@ func (v *CephVersion) IsAtLeastOctopus() bool {
 	return v.IsAtLeast(Octopus)
 }
 
+// IsAtLeastNautilus check that the Ceph version is at least Nautilus
 func (v *CephVersion) IsAtLeastNautilus() bool {
 	return v.IsAtLeast(Nautilus)
 }
 
+// IsAtLeastMimic check that the Ceph version is at least Mimic
 func (v *CephVersion) IsAtLeastMimic() bool {
 	return v.IsAtLeast(Mimic)
 }
diff --git a/pkg/operator/ceph/version/version_test.go b/pkg/operator/ceph/version/version_test.go
index 28a80f0f1..55323d516 100644
--- a/pkg/operator/ceph/version/version_test.go
+++ b/pkg/operator/ceph/version/version_test.go
@@ -16,6 +16,12 @@ func TestToString(t *testing.T) {
 	assert.Equal(t, expected, fmt.Sprintf("%s", &CephVersion{-1, 0, 0}))
 }
 
+func TestCephVersionFormatted(t *testing.T) {
+	assert.Equal(t, "ceph version 14.0.0 nautilus", Nautilus.CephVersionFormatted())
+	assert.Equal(t, "ceph version 13.0.0 mimic", Mimic.CephVersionFormatted())
+	assert.Equal(t, "ceph version 12.0.0 luminous", Luminous.CephVersionFormatted())
+}
+
 func TestReleaseName(t *testing.T) {
 	assert.Equal(t, "nautilus", Nautilus.ReleaseName())
 	assert.Equal(t, "mimic", Mimic.ReleaseName())