diff --git a/Documentation/ceph-cluster-crd.md b/Documentation/ceph-cluster-crd.md
index d88bfcf62..c7d8ad6e1 100755
--- a/Documentation/ceph-cluster-crd.md
+++ b/Documentation/ceph-cluster-crd.md
@@ -199,6 +199,9 @@ For more details on the mons and when to choose a number other than `3`, see the
 * `crashCollector`: The settings for crash collector daemon(s).
   * `disable`: is set to `true`, the crash collector will not run on any node where a Ceph daemon runs
   * `daysToRetain`: specifies the number of days to keep crash entries in the Ceph cluster. By default the entries are kept indefinitely.
+* `logCollector`: The settings for log collector daemon.
+  * `enabled`: if set to `true`, the log collector will run as a side-car next to each Ceph daemon. The Ceph configuration option `log_to_file` will be turned on, meaning Ceph daemons will log on files in addition to still logging to container's stdout. These logs will be rotated. (default: false)
+  * `periodicity`: how often to rotate daemon's log. (default: 24h). Specified with a time suffix which may be 'h' for hours or 'd' for days. **Rotating too often will slightly impact the daemon's performance since the signal briefly interrupts the program.**
 * `annotations`: [annotations configuration settings](#annotations-and-labels)
 * `labels`: [labels configuration settings](#annotations-and-labels)
 * `placement`: [placement configuration settings](#placement-configuration-settings)
@@ -544,6 +547,7 @@ You can set resource requests/limits for Rook components through the [Resource R
 * `crashcollector`: Set resource requests/limits for crash. This pod runs wherever there is a Ceph pod running.
 It scrapes for Ceph daemon core dumps and sends them to the Ceph manager crash module so that core dumps are centralized and can be easily listed/accessed.
 You can read more about the [Ceph Crash module](https://docs.ceph.com/docs/master/mgr/crash/).
+* `logcollector`: Set resource requests/limits for the log collector. When enabled, this container runs as side-car to each Ceph daemons.
 * `cleanup`: Set resource requests/limits for cleanup job, responsible for wiping cluster's data after uninstall
 
 In order to provide the best possible experience running Ceph in containers, Rook internally recommends minimum memory limits if resource limits are passed.
diff --git a/Documentation/ceph-common-issues.md b/Documentation/ceph-common-issues.md
index 5bea484cd..9af9191a3 100644
--- a/Documentation/ceph-common-issues.md
+++ b/Documentation/ceph-common-issues.md
@@ -724,7 +724,7 @@ 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:
 
 ```console
-ceph config daemon mon.a log_to_file true
+ceph config set 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`.
diff --git a/cluster/charts/rook-ceph/templates/resources.yaml b/cluster/charts/rook-ceph/templates/resources.yaml
index 95aa36dfd..a6dfa5a34 100644
--- a/cluster/charts/rook-ceph/templates/resources.yaml
+++ b/cluster/charts/rook-ceph/templates/resources.yaml
@@ -321,6 +321,13 @@ spec:
                           x-kubernetes-preserve-unknown-fields: true
                         tokenSecretName:
                           type: string
+                logCollector:
+                  type: object
+                  properties:
+                    enabled:
+                      type: boolean
+                    periodicity:
+                      type: string
                 annotations:
                   type: object
                   nullable: true
diff --git a/cluster/examples/kubernetes/ceph/cluster.yaml b/cluster/examples/kubernetes/ceph/cluster.yaml
index 0207d1a61..a2c95d8d8 100644
--- a/cluster/examples/kubernetes/ceph/cluster.yaml
+++ b/cluster/examples/kubernetes/ceph/cluster.yaml
@@ -91,8 +91,12 @@ spec:
     # Uncomment daysToRetain to prune ceph crash entries older than the
     # specified number of days.
     #daysToRetain: 30
+  # enable log collector, daemons will log on files and rotate
+  # logCollector:
+  #   enabled: true
+  #   periodicity: 24h # SUFFIX may be 'h' for hours or 'd' for days.
   # automate [data cleanup process](https://github.com/rook/rook/blob/master/Documentation/ceph-teardown.md#delete-the-data-on-hosts) in cluster destruction.
-  cleanupPolicy:
+  # cleanupPolicy:
     # Since cluster cleanup is destructive to data, confirmation is required.
     # To destroy all Rook data on hosts during uninstall, confirmation must be set to "yes-really-destroy-data".
     # This value should only be set when the cluster is about to be deleted. After the confirmation is set,
@@ -172,6 +176,7 @@ spec:
 #    osd:
 #    prepareosd:
 #    crashcollector:
+#    logcollector:
 #    cleanup:
   # The option to automatically remove OSDs that are out and are safe to destroy.
   removeOSDsIfOutAndSafeToRemove: false
diff --git a/cluster/examples/kubernetes/ceph/crds.yaml b/cluster/examples/kubernetes/ceph/crds.yaml
index 2b387551d..7db87ddd8 100644
--- a/cluster/examples/kubernetes/ceph/crds.yaml
+++ b/cluster/examples/kubernetes/ceph/crds.yaml
@@ -323,6 +323,13 @@ spec:
                           x-kubernetes-preserve-unknown-fields: true
                         tokenSecretName:
                           type: string
+                logCollector:
+                  type: object
+                  properties:
+                    enabled:
+                      type: boolean
+                    periodicity:
+                      type: string
                 annotations:
                   type: object
                   nullable: true
diff --git a/cluster/examples/kubernetes/ceph/pre-k8s-1.16/crds.yaml b/cluster/examples/kubernetes/ceph/pre-k8s-1.16/crds.yaml
index f3047edb4..f929135f2 100644
--- a/cluster/examples/kubernetes/ceph/pre-k8s-1.16/crds.yaml
+++ b/cluster/examples/kubernetes/ceph/pre-k8s-1.16/crds.yaml
@@ -206,6 +206,7 @@ spec:
                       type: integer
                       format: int32
             security: {}
+            logCollector: {}
             placement: {}
             resources: {}
             healthCheck: {}
diff --git a/pkg/apis/ceph.rook.io/v1/resources.go b/pkg/apis/ceph.rook.io/v1/resources.go
index f735b75d6..e5acdf5c5 100644
--- a/pkg/apis/ceph.rook.io/v1/resources.go
+++ b/pkg/apis/ceph.rook.io/v1/resources.go
@@ -34,6 +34,8 @@ const (
 	ResourcesKeyMDS = "mds"
 	// ResourcesKeyCrashCollector represents the name of resource in the CR for the crash
 	ResourcesKeyCrashCollector = "crashcollector"
+	// ResourcesKeyLogCollector represents the name of resource in the CR for the log
+	ResourcesKeyLogCollector = "logcollector"
 	// ResourcesKeyRBDMirror represents the name of resource in the CR for the rbd mirror
 	ResourcesKeyRBDMirror = "rbdmirror"
 	// ResourcesKeyCleanup represents the name of resource in the CR for the cleanup
@@ -65,6 +67,11 @@ func GetCrashCollectorResources(p rook.ResourceSpec) v1.ResourceRequirements {
 	return p[ResourcesKeyCrashCollector]
 }
 
+// GetLogCollectorResources returns the placement for the crash daemon
+func GetLogCollectorResources(p rook.ResourceSpec) v1.ResourceRequirements {
+	return p[ResourcesKeyCrashCollector]
+}
+
 // GetCleanupResources returns the placement for the cleanup job
 func GetCleanupResources(p rook.ResourceSpec) v1.ResourceRequirements {
 	return p[ResourcesKeyCleanup]
diff --git a/pkg/apis/ceph.rook.io/v1/types.go b/pkg/apis/ceph.rook.io/v1/types.go
index c598ee5ec..f06a03900 100755
--- a/pkg/apis/ceph.rook.io/v1/types.go
+++ b/pkg/apis/ceph.rook.io/v1/types.go
@@ -132,6 +132,15 @@ type ClusterSpec struct {
 
 	// Security represents security settings
 	Security SecuritySpec `json:"security,omitempty"`
+
+	// Logging represents loggings settings
+	LogCollector LogCollectorSpec `json:"logCollector,omitempty"`
+}
+
+// LogCollectorSpec is the logging spec
+type LogCollectorSpec struct {
+	Enabled     bool   `json:"enabled,omitempty"`
+	Periodicity string `json:"periodicity,omitempty"`
 }
 
 // SecuritySpec is security spec to include various security items such as kms
diff --git a/pkg/apis/ceph.rook.io/v1/zz_generated.deepcopy.go b/pkg/apis/ceph.rook.io/v1/zz_generated.deepcopy.go
index 4c241b787..a5c54fd27 100644
--- a/pkg/apis/ceph.rook.io/v1/zz_generated.deepcopy.go
+++ b/pkg/apis/ceph.rook.io/v1/zz_generated.deepcopy.go
@@ -1051,6 +1051,7 @@ func (in *ClusterSpec) DeepCopyInto(out *ClusterSpec) {
 	out.CleanupPolicy = in.CleanupPolicy
 	in.HealthCheck.DeepCopyInto(&out.HealthCheck)
 	in.Security.DeepCopyInto(&out.Security)
+	out.LogCollector = in.LogCollector
 	return
 }
 
@@ -1451,6 +1452,22 @@ func (in *KeyManagementServiceSpec) DeepCopy() *KeyManagementServiceSpec {
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *LogCollectorSpec) DeepCopyInto(out *LogCollectorSpec) {
+	*out = *in
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new LogCollectorSpec.
+func (in *LogCollectorSpec) DeepCopy() *LogCollectorSpec {
+	if in == nil {
+		return nil
+	}
+	out := new(LogCollectorSpec)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *MetadataServerSpec) DeepCopyInto(out *MetadataServerSpec) {
 	*out = *in
diff --git a/pkg/operator/ceph/cluster/crash/crash.go b/pkg/operator/ceph/cluster/crash/crash.go
index db9d54276..9fff47fac 100644
--- a/pkg/operator/ceph/cluster/crash/crash.go
+++ b/pkg/operator/ceph/cluster/crash/crash.go
@@ -24,7 +24,6 @@ import (
 
 	"github.com/pkg/errors"
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
-	"github.com/rook/rook/pkg/operator/ceph/cluster/mon"
 	"github.com/rook/rook/pkg/operator/ceph/config"
 	"github.com/rook/rook/pkg/operator/ceph/config/keyring"
 	"github.com/rook/rook/pkg/operator/ceph/controller"
@@ -193,7 +192,7 @@ func getCrashDirInitContainer(cephCluster cephv1.CephCluster) corev1.Container {
 			crashPostedDir,
 		},
 		Image:           cephCluster.Spec.CephVersion.Image,
-		SecurityContext: mon.PodSecurityContext(),
+		SecurityContext: controller.PodSecurityContext(),
 		Resources:       cephv1.GetCrashCollectorResources(cephCluster.Spec.Resources),
 		VolumeMounts:    controller.DaemonVolumeMounts(dataPathMap, ""),
 	}
@@ -208,7 +207,7 @@ func getCrashChownInitContainer(cephCluster cephv1.CephCluster) corev1.Container
 		cephCluster.Spec.CephVersion.Image,
 		controller.DaemonVolumeMounts(dataPathMap, ""),
 		cephv1.GetCrashCollectorResources(cephCluster.Spec.Resources),
-		mon.PodSecurityContext(),
+		controller.PodSecurityContext(),
 	)
 }
 
@@ -229,7 +228,7 @@ func getCrashDaemonContainer(cephCluster cephv1.CephCluster, cephVersion version
 		Env:             envVars,
 		VolumeMounts:    volumeMounts,
 		Resources:       cephv1.GetCrashCollectorResources(cephCluster.Spec.Resources),
-		SecurityContext: mon.PodSecurityContext(),
+		SecurityContext: controller.PodSecurityContext(),
 	}
 
 	return container
@@ -258,7 +257,7 @@ func getCrashPruneContainer(cephCluster cephv1.CephCluster, cephVersion version.
 		Env:             envVars,
 		VolumeMounts:    volumeMounts,
 		Resources:       cephv1.GetCrashCollectorResources(cephCluster.Spec.Resources),
-		SecurityContext: mon.PodSecurityContext(),
+		SecurityContext: controller.PodSecurityContext(),
 	}
 
 	return container
diff --git a/pkg/operator/ceph/cluster/mgr/spec.go b/pkg/operator/ceph/cluster/mgr/spec.go
index 61307ab5b..c755b7d27 100644
--- a/pkg/operator/ceph/cluster/mgr/spec.go
+++ b/pkg/operator/ceph/cluster/mgr/spec.go
@@ -27,7 +27,6 @@ import (
 	rookcephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
 	rookv1 "github.com/rook/rook/pkg/apis/rook.io/v1"
 	"github.com/rook/rook/pkg/daemon/ceph/client"
-	"github.com/rook/rook/pkg/operator/ceph/cluster/mon"
 	"github.com/rook/rook/pkg/operator/ceph/config"
 	"github.com/rook/rook/pkg/operator/ceph/config/keyring"
 	"github.com/rook/rook/pkg/operator/ceph/controller"
@@ -67,6 +66,13 @@ func (c *Cluster) makeDeployment(mgrConfig *mgrConfig) (*apps.Deployment, error)
 		},
 	}
 
+	// If the log collector is enabled we add the side-car container
+	if c.spec.LogCollector.Enabled {
+		shareProcessNamespace := true
+		podSpec.Spec.ShareProcessNamespace = &shareProcessNamespace
+		podSpec.Spec.Containers = append(podSpec.Spec.Containers, *controller.LogCollectorContainer(fmt.Sprintf("ceph-mgr.%s", mgrConfig.DaemonID), c.clusterInfo.Namespace, c.spec))
+	}
+
 	// Replace default unreachable node toleration
 	k8sutil.AddUnreachableNodeToleration(&podSpec.Spec)
 
@@ -165,7 +171,7 @@ func (c *Cluster) makeChownInitContainer(mgrConfig *mgrConfig) v1.Container {
 		c.spec.CephVersion.Image,
 		controller.DaemonVolumeMounts(mgrConfig.DataPathMap, mgrConfig.ResourceName),
 		cephv1.GetMgrResources(c.spec.Resources),
-		mon.PodSecurityContext(),
+		controller.PodSecurityContext(),
 	)
 }
 
@@ -246,7 +252,7 @@ func (c *Cluster) makeMgrDaemonContainer(mgrConfig *mgrConfig) v1.Container {
 			c.cephMgrOrchestratorModuleEnvs()...,
 		),
 		Resources:       cephv1.GetMgrResources(c.spec.Resources),
-		SecurityContext: mon.PodSecurityContext(),
+		SecurityContext: controller.PodSecurityContext(),
 		LivenessProbe:   getDefaultMgrLivenessProbe(),
 	}
 
diff --git a/pkg/operator/ceph/cluster/mon/mon.go b/pkg/operator/ceph/cluster/mon/mon.go
index 606744dd7..7a7fd6d6f 100644
--- a/pkg/operator/ceph/cluster/mon/mon.go
+++ b/pkg/operator/ceph/cluster/mon/mon.go
@@ -236,7 +236,7 @@ func (c *Cluster) startMons(targetCount int) error {
 	// only once and do it as early as possible in the mon orchestration.
 	setConfigsNeedsRetry := false
 	if existingCount > 0 {
-		err := config.SetDefaultConfigs(c.context, c.ClusterInfo, c.spec.Network)
+		err := config.SetDefaultConfigs(c.context, c.ClusterInfo, c.spec)
 		if err != nil {
 			// If we fail here, it could be because the mons are not healthy, and this might be
 			// fixed by updating the mon deployments. Instead of returning error here, log a
@@ -262,7 +262,7 @@ func (c *Cluster) startMons(targetCount int) error {
 			// values in the config database. Do this only when the existing count is zero so that
 			// this is only done once when the cluster is created.
 			if existingCount == 0 {
-				err := config.SetDefaultConfigs(c.context, c.ClusterInfo, c.spec.Network)
+				err := config.SetDefaultConfigs(c.context, c.ClusterInfo, c.spec)
 				if err != nil {
 					return errors.Wrap(err, "failed to set Rook and/or user-defined Ceph config options after creating the first mon")
 				}
@@ -270,7 +270,7 @@ func (c *Cluster) startMons(targetCount int) error {
 				// Or if we need to retry, only do this when we are on the first iteration of the
 				// loop. This could be in the same if statement as above, but separate it to get a
 				// different error message.
-				err := config.SetDefaultConfigs(c.context, c.ClusterInfo, c.spec.Network)
+				err := config.SetDefaultConfigs(c.context, c.ClusterInfo, c.spec)
 				if err != nil {
 					return errors.Wrap(err, "failed to set Rook and/or user-defined Ceph config options after updating the existing mons")
 				}
@@ -284,7 +284,7 @@ func (c *Cluster) startMons(targetCount int) error {
 		}
 
 		if setConfigsNeedsRetry {
-			err := config.SetDefaultConfigs(c.context, c.ClusterInfo, c.spec.Network)
+			err := config.SetDefaultConfigs(c.context, c.ClusterInfo, c.spec)
 			if err != nil {
 				return errors.Wrap(err, "failed to set Rook and/or user-defined Ceph config options after forcefully updating the existing mons")
 			}
diff --git a/pkg/operator/ceph/cluster/mon/spec.go b/pkg/operator/ceph/cluster/mon/spec.go
index fa1f34aa7..e748321ac 100644
--- a/pkg/operator/ceph/cluster/mon/spec.go
+++ b/pkg/operator/ceph/cluster/mon/spec.go
@@ -18,7 +18,6 @@ package mon
 
 import (
 	"fmt"
-	"os"
 	"path"
 	"strings"
 
@@ -179,6 +178,13 @@ func (c *Cluster) makeMonPod(monConfig *monConfig, canary bool) (*v1.Pod, error)
 		PriorityClassName: cephv1.GetMonPriorityClassName(c.spec.PriorityClassNames),
 	}
 
+	// If the log collector is enabled we add the side-car container
+	if c.spec.LogCollector.Enabled {
+		shareProcessNamespace := true
+		podSpec.ShareProcessNamespace = &shareProcessNamespace
+		podSpec.Containers = append(podSpec.Containers, *controller.LogCollectorContainer(fmt.Sprintf("%s.%s", cephMonCommand, monConfig.DaemonName), c.ClusterInfo.Namespace, c.spec))
+	}
+
 	// Replace default unreachable node toleration
 	if c.monVolumeClaimTemplate(monConfig) != nil {
 		k8sutil.AddUnreachableNodeToleration(&podSpec)
@@ -220,25 +226,13 @@ func (c *Cluster) makeMonPod(monConfig *monConfig, canary bool) (*v1.Pod, error)
 
 // Init and daemon containers require the same context, so we call it 'pod' context
 
-// PodSecurityContext detects if the pod needs privileges to run
-func PodSecurityContext() *v1.SecurityContext {
-	privileged := false
-	if os.Getenv("ROOK_HOSTPATH_REQUIRES_PRIVILEGED") == "true" {
-		privileged = true
-	}
-
-	return &v1.SecurityContext{
-		Privileged: &privileged,
-	}
-}
-
 func (c *Cluster) makeChownInitContainer(monConfig *monConfig) v1.Container {
 	return controller.ChownCephDataDirsInitContainer(
 		*monConfig.DataPathMap,
 		c.spec.CephVersion.Image,
 		controller.DaemonVolumeMounts(monConfig.DataPathMap, keyringStoreName),
 		cephv1.GetMonResources(c.spec.Resources),
-		PodSecurityContext(),
+		controller.PodSecurityContext(),
 	)
 }
 
@@ -257,7 +251,7 @@ func (c *Cluster) makeMonFSInitContainer(monConfig *monConfig) v1.Container {
 		),
 		Image:           c.spec.CephVersion.Image,
 		VolumeMounts:    controller.DaemonVolumeMounts(monConfig.DataPathMap, keyringStoreName),
-		SecurityContext: PodSecurityContext(),
+		SecurityContext: controller.PodSecurityContext(),
 		// filesystem creation does not require ports to be exposed
 		Env:       controller.DaemonEnvVars(c.spec.CephVersion.Image),
 		Resources: cephv1.GetMonResources(c.spec.Resources),
@@ -297,7 +291,7 @@ func (c *Cluster) makeMonDaemonContainer(monConfig *monConfig) v1.Container {
 		),
 		Image:           c.spec.CephVersion.Image,
 		VolumeMounts:    controller.DaemonVolumeMounts(monConfig.DataPathMap, keyringStoreName),
-		SecurityContext: PodSecurityContext(),
+		SecurityContext: controller.PodSecurityContext(),
 		Ports: []v1.ContainerPort{
 			{
 				Name:          "tcp-msgr1",
diff --git a/pkg/operator/ceph/cluster/osd/spec.go b/pkg/operator/ceph/cluster/osd/spec.go
index 4437408a0..ee380cfa3 100644
--- a/pkg/operator/ceph/cluster/osd/spec.go
+++ b/pkg/operator/ceph/cluster/osd/spec.go
@@ -26,7 +26,6 @@ import (
 	"github.com/pkg/errors"
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
 	kms "github.com/rook/rook/pkg/daemon/ceph/osd/kms"
-	opmon "github.com/rook/rook/pkg/operator/ceph/cluster/mon"
 	"github.com/rook/rook/pkg/operator/ceph/cluster/osd/config"
 	opconfig "github.com/rook/rook/pkg/operator/ceph/config"
 	"github.com/rook/rook/pkg/operator/ceph/controller"
@@ -546,6 +545,13 @@ func (c *Cluster) makeDeployment(osdProps osdProperties, osd OSDInfo, provisionC
 		},
 	}
 
+	// If the log collector is enabled we add the side-car container
+	if c.spec.LogCollector.Enabled {
+		shareProcessNamespace := true
+		podTemplateSpec.Spec.ShareProcessNamespace = &shareProcessNamespace
+		podTemplateSpec.Spec.Containers = append(podTemplateSpec.Spec.Containers, *controller.LogCollectorContainer(fmt.Sprintf("ceph-osd.%s", osdID), c.clusterInfo.Namespace, c.spec))
+	}
+
 	// If the liveness probe is enabled
 	podTemplateSpec.Spec.Containers[0] = opconfig.ConfigureLivenessProbe(cephv1.KeyOSD, podTemplateSpec.Spec.Containers[0], c.spec.HealthCheck)
 
@@ -713,7 +719,7 @@ func (c *Cluster) getPVCInitContainer(osdProps osdProperties) v1.Container {
 			},
 		},
 		VolumeMounts:    []v1.VolumeMount{getPvcOSDBridgeMount(osdProps.pvc.ClaimName)},
-		SecurityContext: opmon.PodSecurityContext(),
+		SecurityContext: controller.PodSecurityContext(),
 		Resources:       osdProps.resources,
 	}
 }
@@ -744,7 +750,7 @@ func (c *Cluster) getPVCInitContainerActivate(mountPath string, osdProps osdProp
 			},
 		},
 		VolumeMounts:    []v1.VolumeMount{getPvcOSDBridgeMountActivate(mountPath, osdProps.pvc.ClaimName)},
-		SecurityContext: opmon.PodSecurityContext(),
+		SecurityContext: controller.PodSecurityContext(),
 		Resources:       osdProps.resources,
 	}
 }
@@ -843,7 +849,7 @@ func (c *Cluster) generateEncryptionCopyBlockContainer(resources v1.ResourceRequ
 		// volumeMountPVCName is crucial, especially when the block we copy is the metadata block
 		// its value must be the name of the block PV so that all init containers use the same bridge (the emptyDir shared by all the init containers)
 		VolumeMounts:    []v1.VolumeMount{getPvcOSDBridgeMountActivate(mountPath, volumeMountPVCName), getDeviceMapperMount()},
-		SecurityContext: opmon.PodSecurityContext(),
+		SecurityContext: controller.PodSecurityContext(),
 		Resources:       resources,
 	}
 }
@@ -889,7 +895,7 @@ func (c *Cluster) getPVCMetadataInitContainer(mountPath string, osdProps osdProp
 				Name:      fmt.Sprintf("%s-bridge", osdProps.metadataPVC.ClaimName),
 			},
 		},
-		SecurityContext: opmon.PodSecurityContext(),
+		SecurityContext: controller.PodSecurityContext(),
 		Resources:       osdProps.resources,
 	}
 }
@@ -922,7 +928,7 @@ func (c *Cluster) getPVCMetadataInitContainerActivate(mountPath string, osdProps
 		// We need to call getPvcOSDBridgeMountActivate() so that we can copy the metadata block into the "main" empty dir
 		// This empty dir is passed along every init container
 		VolumeMounts:    []v1.VolumeMount{getPvcOSDBridgeMountActivate(mountPath, osdProps.pvc.ClaimName)},
-		SecurityContext: opmon.PodSecurityContext(),
+		SecurityContext: controller.PodSecurityContext(),
 		Resources:       osdProps.resources,
 	}
 }
@@ -947,7 +953,7 @@ func (c *Cluster) getPVCWalInitContainer(mountPath string, osdProps osdPropertie
 				Name:      fmt.Sprintf("%s-bridge", osdProps.walPVC.ClaimName),
 			},
 		},
-		SecurityContext: opmon.PodSecurityContext(),
+		SecurityContext: controller.PodSecurityContext(),
 		Resources:       osdProps.resources,
 	}
 }
@@ -980,7 +986,7 @@ func (c *Cluster) getPVCWalInitContainerActivate(mountPath string, osdProps osdP
 		// We need to call getPvcOSDBridgeMountActivate() so that we can copy the wal block into the "main" empty dir
 		// This empty dir is passed along every init container
 		VolumeMounts:    []v1.VolumeMount{getPvcOSDBridgeMountActivate(mountPath, osdProps.pvc.ClaimName)},
-		SecurityContext: opmon.PodSecurityContext(),
+		SecurityContext: controller.PodSecurityContext(),
 		Resources:       osdProps.resources,
 	}
 }
diff --git a/pkg/operator/ceph/cluster/rbd/spec.go b/pkg/operator/ceph/cluster/rbd/spec.go
index 6e2621925..68fd4ce7c 100644
--- a/pkg/operator/ceph/cluster/rbd/spec.go
+++ b/pkg/operator/ceph/cluster/rbd/spec.go
@@ -20,7 +20,6 @@ import (
 	"fmt"
 
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
-	"github.com/rook/rook/pkg/operator/ceph/cluster/mon"
 	"github.com/rook/rook/pkg/operator/ceph/config"
 	"github.com/rook/rook/pkg/operator/ceph/controller"
 	"github.com/rook/rook/pkg/operator/k8sutil"
@@ -48,6 +47,14 @@ func (r *ReconcileCephRBDMirror) makeDeployment(daemonConfig *daemonConfig, rbdM
 			PriorityClassName: rbdMirror.Spec.PriorityClassName,
 		},
 	}
+
+	// If the log collector is enabled we add the side-car container
+	if r.cephClusterSpec.LogCollector.Enabled {
+		shareProcessNamespace := true
+		podSpec.Spec.ShareProcessNamespace = &shareProcessNamespace
+		podSpec.Spec.Containers = append(podSpec.Spec.Containers, *controller.LogCollectorContainer(fmt.Sprintf("ceph-client.rbd-mirror.%s", daemonConfig.DaemonID), r.clusterInfo.Namespace, *r.cephClusterSpec))
+	}
+
 	// Replace default unreachable node toleration
 	k8sutil.AddUnreachableNodeToleration(&podSpec.Spec)
 	rbdMirror.Spec.Annotations.ApplyToObjectMeta(&podSpec.ObjectMeta)
@@ -102,7 +109,7 @@ func (r *ReconcileCephRBDMirror) makeChownInitContainer(daemonConfig *daemonConf
 		r.cephClusterSpec.CephVersion.Image,
 		controller.DaemonVolumeMounts(daemonConfig.DataPathMap, daemonConfig.ResourceName),
 		rbdMirror.Spec.Resources,
-		mon.PodSecurityContext(),
+		controller.PodSecurityContext(),
 	)
 }
 
@@ -121,7 +128,7 @@ func (r *ReconcileCephRBDMirror) makeMirroringDaemonContainer(daemonConfig *daem
 		VolumeMounts:    controller.DaemonVolumeMounts(daemonConfig.DataPathMap, daemonConfig.ResourceName),
 		Env:             controller.DaemonEnvVars(r.cephClusterSpec.CephVersion.Image),
 		Resources:       rbdMirror.Spec.Resources,
-		SecurityContext: mon.PodSecurityContext(),
+		SecurityContext: controller.PodSecurityContext(),
 		// TODO:
 		// Not implemented at this point since the socket name is '/run/ceph/ceph-client.rbd-mirror.a.1.94362516231272.asok'
 		// Also the command to run will be:
diff --git a/pkg/operator/ceph/config/config.go b/pkg/operator/ceph/config/config.go
index 6ee776a3b..4e4f1a26d 100644
--- a/pkg/operator/ceph/config/config.go
+++ b/pkg/operator/ceph/config/config.go
@@ -103,7 +103,7 @@ func NewFlag(key, value string) string {
 func SetDefaultConfigs(
 	context *clusterd.Context,
 	clusterInfo *cephclient.ClusterInfo,
-	networkSpec cephv1.NetworkSpec,
+	clusterSpec cephv1.ClusterSpec,
 ) error {
 	// ceph.conf is never used. All configurations are made in the centralized mon config database,
 	// or they are specified on the commandline when daemons are called.
@@ -113,14 +113,37 @@ func SetDefaultConfigs(
 		return errors.Wrapf(err, "failed to apply default Ceph configurations")
 	}
 
+	// When enabled the collector will logrotate logs from files
+	if clusterSpec.LogCollector.Enabled {
+		// Override "log file" for existing clusters since it is empty
+		logOptions := []Option{
+			configOverride("global", "log file", "/var/log/ceph/$cluster-$name.log"),
+			configOverride("global", "log to file", "true"),
+		}
+
+		if err := monStore.SetAll(logOptions...); err != nil {
+			return errors.Wrapf(err, "failed to apply logging configuration for log collector")
+		}
+		// If the log collector is disabled we do not log to file since we collect nothing
+	} else {
+		logOptions := []Option{
+			configOverride("global", "log file", ""),
+			configOverride("global", "log to file", "false"),
+		}
+
+		if err := monStore.SetAll(logOptions...); err != nil {
+			return errors.Wrapf(err, "failed to apply logging configuration")
+		}
+	}
+
 	if err := monStore.SetAll(DefaultLegacyConfigs()...); err != nil {
 		return errors.Wrapf(err, "failed to apply legacy config overrides")
 	}
 
 	// Apply Multus if needed
-	if networkSpec.IsMultus() {
+	if clusterSpec.Network.IsMultus() {
 		logger.Info("configuring ceph network(s) with multus")
-		cephNetworks, err := generateNetworkSettings(context, clusterInfo.Namespace, networkSpec.Selectors)
+		cephNetworks, err := generateNetworkSettings(context, clusterInfo.Namespace, clusterSpec.Network.Selectors)
 		if err != nil {
 			return errors.Wrap(err, "failed to generate network settings")
 		}
diff --git a/pkg/operator/ceph/config/defaults.go b/pkg/operator/ceph/config/defaults.go
index 71510e3cb..f03f381bd 100644
--- a/pkg/operator/ceph/config/defaults.go
+++ b/pkg/operator/ceph/config/defaults.go
@@ -60,7 +60,6 @@ func LoggingFlags() []string {
 func DefaultCentralizedConfigs(cephVersion version.CephVersion) []Option {
 	overrides := []Option{
 		configOverride("global", "mon allow pool delete", "true"),
-		configOverride("global", "log file", ""),
 		configOverride("global", "mon cluster log file", ""),
 	}
 
diff --git a/pkg/operator/ceph/controller/spec.go b/pkg/operator/ceph/controller/spec.go
index 26fc64602..1cf7b6622 100644
--- a/pkg/operator/ceph/controller/spec.go
+++ b/pkg/operator/ceph/controller/spec.go
@@ -19,7 +19,9 @@ package controller
 
 import (
 	"fmt"
+	"os"
 	"path"
+	"strings"
 
 	"github.com/coreos/pkg/capnslog"
 	"github.com/pkg/errors"
@@ -42,6 +44,7 @@ const (
 	daemonSocketDir                       = "/run/ceph"
 	initialDelaySecondsNonOSDDaemon int32 = 10
 	initialDelaySecondsOSDDaemon    int32 = 45
+	logCollector                          = "log-collector"
 )
 
 type daemonConfig struct {
@@ -51,6 +54,32 @@ type daemonConfig struct {
 
 var logger = capnslog.NewPackageLogger("github.com/rook/rook", "ceph-spec")
 
+var (
+	cronLogRotate = `
+set -xe
+
+CEPH_CLIENT_ID=%s
+PERIODICITY=%s
+LOG_ROTATE_CEPH_FILE=/etc/logrotate.d/ceph
+
+if [ -z "$PERIODICITY" ]; then
+	PERIODICITY=24h
+fi
+
+# edit the logrotate file to only rotate a specific daemon log
+# otherwise we will logrotate log files without reloading certain daemons
+# this might happen when multiple daemons run on the same machine
+sed -i "s|*.log|$CEPH_CLIENT_ID.log|" "$LOG_ROTATE_CEPH_FILE"
+
+while true; do
+	sleep "$PERIODICITY"
+	echo "starting log rotation"
+	logrotate --verbose --force "$LOG_ROTATE_CEPH_FILE"
+	echo "I am going to sleep now, see you in $PERIODICITY"
+done
+`
+)
+
 // return the volume and matching volume mount for mounting the config override ConfigMap into
 // containers as "/etc/ceph/ceph.conf".
 func configOverrideConfigMapVolumeAndMount() (v1.Volume, v1.VolumeMount) {
@@ -538,3 +567,35 @@ func (c *daemonConfig) buildAdminSocketCommand() string {
 
 	return command
 }
+
+// PodSecurityContext detects if the pod needs privileges to run
+func PodSecurityContext() *v1.SecurityContext {
+	privileged := false
+	if os.Getenv("ROOK_HOSTPATH_REQUIRES_PRIVILEGED") == "true" {
+		privileged = true
+	}
+
+	return &v1.SecurityContext{
+		Privileged: &privileged,
+	}
+}
+
+// LogCollectorContainer runs a cron job to rotate logs
+func LogCollectorContainer(daemonID, ns string, c cephv1.ClusterSpec) *v1.Container {
+	return &v1.Container{
+		Name: logCollectorContainerName(daemonID),
+		Command: []string{
+			"/bin/bash",
+			"-c",
+			fmt.Sprintf(cronLogRotate, daemonID, c.LogCollector.Periodicity),
+		},
+		Image:           c.CephVersion.Image,
+		VolumeMounts:    DaemonVolumeMounts(config.NewDatalessDaemonDataPathMap(ns, c.DataDirHostPath), ""),
+		SecurityContext: PodSecurityContext(),
+		Resources:       cephv1.GetLogCollectorResources(c.Resources),
+	}
+}
+
+func logCollectorContainerName(daemon string) string {
+	return fmt.Sprintf("%s-%s", strings.Replace(daemon, ".", "-", -1), logCollector)
+}
diff --git a/pkg/operator/ceph/file/mds/spec.go b/pkg/operator/ceph/file/mds/spec.go
index aa74ed108..f7df4be13 100644
--- a/pkg/operator/ceph/file/mds/spec.go
+++ b/pkg/operator/ceph/file/mds/spec.go
@@ -22,7 +22,6 @@ import (
 
 	"github.com/pkg/errors"
 	"github.com/rook/rook/pkg/clusterd"
-	"github.com/rook/rook/pkg/operator/ceph/cluster/mon"
 	"github.com/rook/rook/pkg/operator/ceph/config"
 	"github.com/rook/rook/pkg/operator/ceph/controller"
 	"github.com/rook/rook/pkg/operator/k8sutil"
@@ -60,6 +59,13 @@ func (c *Cluster) makeDeployment(mdsConfig *mdsConfig) (*apps.Deployment, error)
 	// Replace default unreachable node toleration
 	k8sutil.AddUnreachableNodeToleration(&podSpec.Spec)
 
+	// If the log collector is enabled we add the side-car container
+	if c.clusterSpec.LogCollector.Enabled {
+		shareProcessNamespace := true
+		podSpec.Spec.ShareProcessNamespace = &shareProcessNamespace
+		podSpec.Spec.Containers = append(podSpec.Spec.Containers, *controller.LogCollectorContainer(fmt.Sprintf("ceph-mds.%s", mdsConfig.DaemonID), c.clusterInfo.Namespace, *c.clusterSpec))
+	}
+
 	c.fs.Spec.MetadataServer.Annotations.ApplyToObjectMeta(&podSpec.ObjectMeta)
 	c.fs.Spec.MetadataServer.Labels.ApplyToObjectMeta(&podSpec.ObjectMeta)
 	c.fs.Spec.MetadataServer.Placement.ApplyToPodSpec(&podSpec.Spec)
@@ -105,7 +111,7 @@ func (c *Cluster) makeChownInitContainer(mdsConfig *mdsConfig) v1.Container {
 		c.clusterSpec.CephVersion.Image,
 		controller.DaemonVolumeMounts(mdsConfig.DataPathMap, mdsConfig.ResourceName),
 		c.fs.Spec.MetadataServer.Resources,
-		mon.PodSecurityContext(),
+		controller.PodSecurityContext(),
 	)
 }
 
@@ -125,7 +131,7 @@ func (c *Cluster) makeMdsDaemonContainer(mdsConfig *mdsConfig) v1.Container {
 		VolumeMounts:    controller.DaemonVolumeMounts(mdsConfig.DataPathMap, mdsConfig.ResourceName),
 		Env:             controller.DaemonEnvVars(c.clusterSpec.CephVersion.Image),
 		Resources:       c.fs.Spec.MetadataServer.Resources,
-		SecurityContext: mon.PodSecurityContext(),
+		SecurityContext: controller.PodSecurityContext(),
 		LivenessProbe:   controller.GenerateLivenessProbeExecDaemon(config.MdsType, mdsConfig.DaemonID),
 	}
 
diff --git a/pkg/operator/ceph/nfs/spec.go b/pkg/operator/ceph/nfs/spec.go
index eb82fdf8e..24193708c 100644
--- a/pkg/operator/ceph/nfs/spec.go
+++ b/pkg/operator/ceph/nfs/spec.go
@@ -22,7 +22,6 @@ import (
 	"github.com/pkg/errors"
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
 	cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
-	"github.com/rook/rook/pkg/operator/ceph/cluster/mon"
 	"github.com/rook/rook/pkg/operator/ceph/config/keyring"
 	"github.com/rook/rook/pkg/operator/ceph/controller"
 	"github.com/rook/rook/pkg/operator/k8sutil"
@@ -185,7 +184,7 @@ func (r *ReconcileCephNFS) connectionConfigInitContainer(nfs *cephv1.CephNFS, na
 			keyring.VolumeMount().Resource(instanceName(nfs, name)),
 		},
 		nfs.Spec.Server.Resources,
-		mon.PodSecurityContext(),
+		controller.PodSecurityContext(),
 	)
 }
 
@@ -218,7 +217,7 @@ func (r *ReconcileCephNFS) daemonContainer(nfs *cephv1.CephNFS, cfg daemonConfig
 		},
 		Env:             controller.DaemonEnvVars(r.cephClusterSpec.CephVersion.Image),
 		Resources:       nfs.Spec.Server.Resources,
-		SecurityContext: mon.PodSecurityContext(),
+		SecurityContext: controller.PodSecurityContext(),
 	}
 }
 
diff --git a/pkg/operator/ceph/object/spec.go b/pkg/operator/ceph/object/spec.go
index 213756e82..615c50b46 100644
--- a/pkg/operator/ceph/object/spec.go
+++ b/pkg/operator/ceph/object/spec.go
@@ -22,7 +22,6 @@ import (
 
 	"github.com/pkg/errors"
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
-	"github.com/rook/rook/pkg/operator/ceph/cluster/mon"
 	cephconfig "github.com/rook/rook/pkg/operator/ceph/config"
 	"github.com/rook/rook/pkg/operator/ceph/controller"
 	"github.com/rook/rook/pkg/operator/k8sutil"
@@ -73,9 +72,7 @@ func (c *clusterConfig) makeRGWPodSpec(rgwConfig *rgwConfig) (v1.PodTemplateSpec
 		InitContainers: []v1.Container{
 			c.makeChownInitContainer(rgwConfig),
 		},
-		Containers: []v1.Container{
-			c.makeDaemonContainer(rgwConfig),
-		},
+		Containers:    []v1.Container{c.makeDaemonContainer(rgwConfig)},
 		RestartPolicy: v1.RestartPolicyAlways,
 		Volumes: append(
 			controller.DaemonVolumes(c.DataPathMap, rgwConfig.ResourceName),
@@ -84,6 +81,14 @@ func (c *clusterConfig) makeRGWPodSpec(rgwConfig *rgwConfig) (v1.PodTemplateSpec
 		HostNetwork:       c.clusterSpec.Network.IsHost(),
 		PriorityClassName: c.store.Spec.Gateway.PriorityClassName,
 	}
+
+	// If the log collector is enabled we add the side-car container
+	if c.clusterSpec.LogCollector.Enabled {
+		shareProcessNamespace := true
+		podSpec.ShareProcessNamespace = &shareProcessNamespace
+		podSpec.Containers = append(podSpec.Containers, *controller.LogCollectorContainer(strings.TrimPrefix(generateCephXUser(fmt.Sprintf("ceph-client.%s", rgwConfig.ResourceName)), "client."), c.clusterInfo.Namespace, *c.clusterSpec))
+	}
+
 	// Replace default unreachable node toleration
 	k8sutil.AddUnreachableNodeToleration(&podSpec)
 
@@ -136,7 +141,7 @@ func (c *clusterConfig) makeChownInitContainer(rgwConfig *rgwConfig) v1.Containe
 		c.clusterSpec.CephVersion.Image,
 		controller.DaemonVolumeMounts(c.DataPathMap, rgwConfig.ResourceName),
 		c.store.Spec.Gateway.Resources,
-		mon.PodSecurityContext(),
+		controller.PodSecurityContext(),
 	)
 }
 
@@ -166,7 +171,7 @@ func (c *clusterConfig) makeDaemonContainer(rgwConfig *rgwConfig) v1.Container {
 		Env:             controller.DaemonEnvVars(c.clusterSpec.CephVersion.Image),
 		Resources:       c.store.Spec.Gateway.Resources,
 		LivenessProbe:   c.generateLiveProbe(),
-		SecurityContext: mon.PodSecurityContext(),
+		SecurityContext: controller.PodSecurityContext(),
 	}
 
 	// If the liveness probe is enabled
diff --git a/tests/framework/installer/ceph_manifests.go b/tests/framework/installer/ceph_manifests.go
index d18a950be..bf1c58551 100644
--- a/tests/framework/installer/ceph_manifests.go
+++ b/tests/framework/installer/ceph_manifests.go
@@ -404,6 +404,13 @@ spec:
                         iteration:
                           type: integer
                           format: int32
+                logCollector:
+                  type: object
+                  properties:
+                    enabled:
+                      type: boolean
+                    periodicity:
+                      type: string
                 annotations:
                   type: object
                   nullable: true