diff --git a/apis/fluentbit/v1alpha2/clusterinput_types.go b/apis/fluentbit/v1alpha2/clusterinput_types.go
index 29c5d79ac..dd106ada0 100644
--- a/apis/fluentbit/v1alpha2/clusterinput_types.go
+++ b/apis/fluentbit/v1alpha2/clusterinput_types.go
@@ -65,6 +65,8 @@ type InputSpec struct {
 	Collectd *input.Collectd `json:"collectd,omitempty"`
 	// Nginx defines the Nginx input plugin configuration
 	Nginx *input.Nginx `json:"nginx,omitempty"`
+	// StatsD defines the StatsD input plugin configuration
+	StatsD *input.StatsD `json:"statsd,omitempty"`
 }
 
 // +kubebuilder:object:root=true
diff --git a/apis/fluentbit/v1alpha2/fluentbit_types.go b/apis/fluentbit/v1alpha2/fluentbit_types.go
index ca41614e4..b00ea70fb 100644
--- a/apis/fluentbit/v1alpha2/fluentbit_types.go
+++ b/apis/fluentbit/v1alpha2/fluentbit_types.go
@@ -70,6 +70,8 @@ type FluentBitSpec struct {
 	Volumes []corev1.Volume `json:"volumes,omitempty"`
 	// Pod volumes to mount into the container's filesystem.
 	VolumesMounts []corev1.VolumeMount `json:"volumesMounts,omitempty"`
+	// DisableLogVolumes removes the hostPath mounts for varlibcontainers, varlogs and systemd.
+	DisableLogVolumes bool `json:"disableLogVolumes,omitempty"`
 	// Annotations to add to each Fluentbit pod.
 	Annotations map[string]string `json:"annotations,omitempty"`
 	// Annotations to add to the Fluentbit service account
diff --git a/apis/fluentbit/v1alpha2/plugins/input/statsd.go b/apis/fluentbit/v1alpha2/plugins/input/statsd.go
new file mode 100644
index 000000000..c81592353
--- /dev/null
+++ b/apis/fluentbit/v1alpha2/plugins/input/statsd.go
@@ -0,0 +1,37 @@
+package input
+
+import (
+	"fmt"
+
+	"github.com/fluent/fluent-operator/v2/apis/fluentbit/v1alpha2/plugins"
+	"github.com/fluent/fluent-operator/v2/apis/fluentbit/v1alpha2/plugins/params"
+)
+
+// +kubebuilder:object:generate:=true
+
+// The StatsD input plugin allows you to receive metrics via StatsD protocol.<br />
+// **For full documentation, refer to https://docs.fluentbit.io/manual/pipeline/inputs/statsd**
+type StatsD struct {
+	// Listener network interface, default: 0.0.0.0
+	Listen string `json:"listen,omitempty"`
+	// UDP port where listening for connections, default: 8125
+	// +kubebuilder:validation:Minimum:=1
+	// +kubebuilder:validation:Maximum:=65535
+	Port *int32 `json:"port,omitempty"`
+}
+
+func (_ *StatsD) Name() string {
+	return "statsd"
+}
+
+// implement Section() method
+func (s *StatsD) Params(_ plugins.SecretLoader) (*params.KVs, error) {
+	kvs := params.NewKVs()
+	if s.Listen != "" {
+		kvs.Insert("Listen", s.Listen)
+	}
+	if s.Port != nil {
+		kvs.Insert("Port", fmt.Sprint(*s.Port))
+	}
+	return kvs, nil
+}
diff --git a/apis/fluentbit/v1alpha2/plugins/input/zz_generated.deepcopy.go b/apis/fluentbit/v1alpha2/plugins/input/zz_generated.deepcopy.go
index 382eb35a3..f44227c8f 100644
--- a/apis/fluentbit/v1alpha2/plugins/input/zz_generated.deepcopy.go
+++ b/apis/fluentbit/v1alpha2/plugins/input/zz_generated.deepcopy.go
@@ -255,6 +255,26 @@ func (in *PrometheusScrapeMetrics) DeepCopy() *PrometheusScrapeMetrics {
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *StatsD) DeepCopyInto(out *StatsD) {
+	*out = *in
+	if in.Port != nil {
+		in, out := &in.Port, &out.Port
+		*out = new(int32)
+		**out = **in
+	}
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new StatsD.
+func (in *StatsD) DeepCopy() *StatsD {
+	if in == nil {
+		return nil
+	}
+	out := new(StatsD)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *Systemd) DeepCopyInto(out *Systemd) {
 	*out = *in
diff --git a/apis/fluentbit/v1alpha2/zz_generated.deepcopy.go b/apis/fluentbit/v1alpha2/zz_generated.deepcopy.go
index d761b7d8f..9fc1f5451 100644
--- a/apis/fluentbit/v1alpha2/zz_generated.deepcopy.go
+++ b/apis/fluentbit/v1alpha2/zz_generated.deepcopy.go
@@ -1101,6 +1101,9 @@ func (in *InputSpec) DeepCopyInto(out *InputSpec) {
 	if in.Nginx != nil {
 		in, out := &in.Nginx, &out.Nginx
 		*out = new(input.Nginx)
+	if in.StatsD != nil {
+		in, out := &in.StatsD, &out.StatsD
+		*out = new(input.StatsD)
 		(*in).DeepCopyInto(*out)
 	}
 }
diff --git a/charts/fluent-operator/charts/fluent-bit-crds/crds/fluentbit.fluent.io_clusterinputs.yaml b/charts/fluent-operator/charts/fluent-bit-crds/crds/fluentbit.fluent.io_clusterinputs.yaml
index d9c73669c..307e47687 100644
--- a/charts/fluent-operator/charts/fluent-bit-crds/crds/fluentbit.fluent.io_clusterinputs.yaml
+++ b/charts/fluent-operator/charts/fluent-bit-crds/crds/fluentbit.fluent.io_clusterinputs.yaml
@@ -375,6 +375,20 @@ spec:
                       plugin
                     type: string
                 type: object
+              statsd:
+                description: StatsD defines the StatsD input plugin configuration
+                properties:
+                  listen:
+                    description: 'Listener network interface, default: 0.0.0.0'
+                    type: string
+                  port:
+                    description: 'UDP port where listening for connections, default:
+                      8125'
+                    format: int32
+                    maximum: 65535
+                    minimum: 1
+                    type: integer
+                type: object
               systemd:
                 description: Systemd defines Systemd Input configuration.
                 properties:
diff --git a/charts/fluent-operator/charts/fluent-bit-crds/crds/fluentbit.fluent.io_fluentbits.yaml b/charts/fluent-operator/charts/fluent-bit-crds/crds/fluentbit.fluent.io_fluentbits.yaml
index 820dee8d2..c75136569 100644
--- a/charts/fluent-operator/charts/fluent-bit-crds/crds/fluentbit.fluent.io_fluentbits.yaml
+++ b/charts/fluent-operator/charts/fluent-bit-crds/crds/fluentbit.fluent.io_fluentbits.yaml
@@ -1043,6 +1043,10 @@ spec:
                         type: string
                     type: object
                 type: object
+              disableLogVolumes:
+                description: DisableLogVolumes removes the hostPath mounts for varlibcontainers,
+                  varlogs and systemd.
+                type: boolean
               disableService:
                 description: DisableService tells if the fluentbit service should
                   be deployed.
diff --git a/config/crd/bases/fluentbit.fluent.io_clusterinputs.yaml b/config/crd/bases/fluentbit.fluent.io_clusterinputs.yaml
index d9c73669c..307e47687 100644
--- a/config/crd/bases/fluentbit.fluent.io_clusterinputs.yaml
+++ b/config/crd/bases/fluentbit.fluent.io_clusterinputs.yaml
@@ -375,6 +375,20 @@ spec:
                       plugin
                     type: string
                 type: object
+              statsd:
+                description: StatsD defines the StatsD input plugin configuration
+                properties:
+                  listen:
+                    description: 'Listener network interface, default: 0.0.0.0'
+                    type: string
+                  port:
+                    description: 'UDP port where listening for connections, default:
+                      8125'
+                    format: int32
+                    maximum: 65535
+                    minimum: 1
+                    type: integer
+                type: object
               systemd:
                 description: Systemd defines Systemd Input configuration.
                 properties:
diff --git a/config/crd/bases/fluentbit.fluent.io_fluentbits.yaml b/config/crd/bases/fluentbit.fluent.io_fluentbits.yaml
index 820dee8d2..c75136569 100644
--- a/config/crd/bases/fluentbit.fluent.io_fluentbits.yaml
+++ b/config/crd/bases/fluentbit.fluent.io_fluentbits.yaml
@@ -1043,6 +1043,10 @@ spec:
                         type: string
                     type: object
                 type: object
+              disableLogVolumes:
+                description: DisableLogVolumes removes the hostPath mounts for varlibcontainers,
+                  varlogs and systemd.
+                type: boolean
               disableService:
                 description: DisableService tells if the fluentbit service should
                   be deployed.
diff --git a/config/samples/fluentbit_v1alpha2_fluentbit.yaml b/config/samples/fluentbit_v1alpha2_fluentbit.yaml
index 72fa033e8..49cf41746 100644
--- a/config/samples/fluentbit_v1alpha2_fluentbit.yaml
+++ b/config/samples/fluentbit_v1alpha2_fluentbit.yaml
@@ -11,3 +11,4 @@ spec:
     hostPath:
       path: /var/lib/fluent-bit/
   fluentBitConfigName: fluentbitconfig-sample
+  disableLogVolumes: false
diff --git a/docs/fluentbit.md b/docs/fluentbit.md
index c724b6c09..354997512 100644
--- a/docs/fluentbit.md
+++ b/docs/fluentbit.md
@@ -400,6 +400,7 @@ FluentBitSpec defines the desired state of FluentBit
 | metricsPort | MetricsPort is the port used by the metrics server. If this option is set, HttpPort from ClusterFluentBitConfig needs to match this value. Default is 2020. | int32 |
 | service | Service represents configurations on the fluent-bit service. | FluentBitService |
 | schedulerName | SchedulerName represents the desired scheduler for fluent-bit pods. | string |
+| disableLogVolumes | DisableLogVolumes removes the hostPath mounts for varlibcontainers, varlogs and systemd. | bool |
 
 [Back to TOC](#table-of-contents)
 # InputSpec
@@ -424,6 +425,7 @@ InputSpec defines the desired state of ClusterInput
 | mqtt | MQTT defines the MQTT input plugin configuration | *[input.MQTT](plugins/input/mqtt.md) |
 | collectd | Collectd defines the Collectd input plugin configuration | *[input.Collectd](plugins/input/collectd.md) |
 | nginx | Nginx defines the Nginx input plugin configuration | *[input.Nginx](plugins/input/nginx.md) |
+| statsd | StatsD defines the StatsD input plugin configuration | *[input.StatsD](plugins/input/statsd.md) |
 
 [Back to TOC](#table-of-contents)
 # NamespacedFluentBitCfgSpec
diff --git a/docs/plugins/fluentbit/input/statsd.md b/docs/plugins/fluentbit/input/statsd.md
new file mode 100644
index 000000000..d0215e60d
--- /dev/null
+++ b/docs/plugins/fluentbit/input/statsd.md
@@ -0,0 +1,9 @@
+# StatsD
+
+The StatsD input plugin allows you to receive metrics via StatsD protocol.<br /> **For full documentation, refer to https://docs.fluentbit.io/manual/pipeline/inputs/statsd**
+
+
+| Field | Description | Scheme |
+| ----- | ----------- | ------ |
+| listen | Listener network interface, default: 0.0.0.0 | string |
+| port | UDP port where listening for connections, default: 8125 | *int32 |
diff --git a/manifests/setup/fluent-operator-crd.yaml b/manifests/setup/fluent-operator-crd.yaml
index 398af8ba8..69a2f3bee 100644
--- a/manifests/setup/fluent-operator-crd.yaml
+++ b/manifests/setup/fluent-operator-crd.yaml
@@ -2124,6 +2124,20 @@ spec:
                       plugin
                     type: string
                 type: object
+              statsd:
+                description: StatsD defines the StatsD input plugin configuration
+                properties:
+                  listen:
+                    description: 'Listener network interface, default: 0.0.0.0'
+                    type: string
+                  port:
+                    description: 'UDP port where listening for connections, default:
+                      8125'
+                    format: int32
+                    maximum: 65535
+                    minimum: 1
+                    type: integer
+                type: object
               systemd:
                 description: Systemd defines Systemd Input configuration.
                 properties:
@@ -13161,6 +13175,10 @@ spec:
                         type: string
                     type: object
                 type: object
+              disableLogVolumes:
+                description: DisableLogVolumes removes the hostPath mounts for varlibcontainers,
+                  varlogs and systemd.
+                type: boolean
               disableService:
                 description: DisableService tells if the fluentbit service should
                   be deployed.
diff --git a/manifests/setup/setup.yaml b/manifests/setup/setup.yaml
index 86d0f39ff..4a08c1171 100644
--- a/manifests/setup/setup.yaml
+++ b/manifests/setup/setup.yaml
@@ -2124,6 +2124,20 @@ spec:
                       plugin
                     type: string
                 type: object
+              statsd:
+                description: StatsD defines the StatsD input plugin configuration
+                properties:
+                  listen:
+                    description: 'Listener network interface, default: 0.0.0.0'
+                    type: string
+                  port:
+                    description: 'UDP port where listening for connections, default:
+                      8125'
+                    format: int32
+                    maximum: 65535
+                    minimum: 1
+                    type: integer
+                type: object
               systemd:
                 description: Systemd defines Systemd Input configuration.
                 properties:
@@ -13161,6 +13175,10 @@ spec:
                         type: string
                     type: object
                 type: object
+              disableLogVolumes:
+                description: DisableLogVolumes removes the hostPath mounts for varlibcontainers,
+                  varlogs and systemd.
+                type: boolean
               disableService:
                 description: DisableService tells if the fluentbit service should
                   be deployed.
diff --git a/pkg/operator/daemonset.go b/pkg/operator/daemonset.go
index f19161655..71ad2df15 100644
--- a/pkg/operator/daemonset.go
+++ b/pkg/operator/daemonset.go
@@ -25,10 +25,8 @@ func MakeDaemonSet(fb fluentbitv1alpha2.FluentBit, logPath string) *appsv1.Daemo
 		metricsPort = 2020
 	}
 
-	internalMountPropagation := corev1.MountPropagationNone
-	if fb.Spec.InternalMountPropagation != nil {
-		internalMountPropagation = *fb.Spec.InternalMountPropagation
-	}
+	fbVolumeMounts := makeVolumeMounts(fb, logPath)
+	fbVolumes := makeVolumes(fb, logPath)
 
 	ds := appsv1.DaemonSet{
 		ObjectMeta: metav1.ObjectMeta{
@@ -51,41 +49,8 @@ func MakeDaemonSet(fb fluentbitv1alpha2.FluentBit, logPath string) *appsv1.Daemo
 				Spec: corev1.PodSpec{
 					ServiceAccountName: fb.Name,
 					ImagePullSecrets:   fb.Spec.ImagePullSecrets,
-					Volumes: []corev1.Volume{
-						{
-							Name: "varlibcontainers",
-							VolumeSource: corev1.VolumeSource{
-								HostPath: &corev1.HostPathVolumeSource{
-									Path: logPath,
-								},
-							},
-						},
-						{
-							Name: "config",
-							VolumeSource: corev1.VolumeSource{
-								Secret: &corev1.SecretVolumeSource{
-									SecretName: fb.Spec.FluentBitConfigName,
-								},
-							},
-						},
-						{
-							Name: "varlogs",
-							VolumeSource: corev1.VolumeSource{
-								HostPath: &corev1.HostPathVolumeSource{
-									Path: "/var/log",
-								},
-							},
-						},
-						{
-							Name: "systemd",
-							VolumeSource: corev1.VolumeSource{
-								HostPath: &corev1.HostPathVolumeSource{
-									Path: "/var/log/journal",
-								},
-							},
-						},
-					},
-					InitContainers: fb.Spec.InitContainers,
+					InitContainers:     fb.Spec.InitContainers,
+					Volumes:            fbVolumes,
 					Containers: []corev1.Container{
 						{
 							Name:            "fluent-bit",
@@ -118,31 +83,7 @@ func MakeDaemonSet(fb fluentbitv1alpha2.FluentBit, logPath string) *appsv1.Daemo
 									},
 								},
 							},
-							VolumeMounts: []corev1.VolumeMount{
-								{
-									Name:             "varlibcontainers",
-									ReadOnly:         true,
-									MountPath:        logPath,
-									MountPropagation: &internalMountPropagation,
-								},
-								{
-									Name:      "config",
-									ReadOnly:  true,
-									MountPath: "/fluent-bit/config",
-								},
-								{
-									Name:             "varlogs",
-									ReadOnly:         true,
-									MountPath:        "/var/log/",
-									MountPropagation: &internalMountPropagation,
-								},
-								{
-									Name:             "systemd",
-									ReadOnly:         true,
-									MountPath:        "/var/log/journal",
-									MountPropagation: &internalMountPropagation,
-								},
-							},
+							VolumeMounts:    fbVolumeMounts,
 							Resources:       fb.Spec.Resources,
 							SecurityContext: fb.Spec.ContainerSecurityContext,
 						},
@@ -189,13 +130,6 @@ func MakeDaemonSet(fb fluentbitv1alpha2.FluentBit, logPath string) *appsv1.Daemo
 		ds.Spec.Template.Spec.SchedulerName = fb.Spec.SchedulerName
 	}
 
-	if fb.Spec.Volumes != nil {
-		ds.Spec.Template.Spec.Volumes = append(ds.Spec.Template.Spec.Volumes, fb.Spec.Volumes...)
-	}
-	if fb.Spec.VolumesMounts != nil {
-		ds.Spec.Template.Spec.Containers[0].VolumeMounts = append(ds.Spec.Template.Spec.Containers[0].VolumeMounts, fb.Spec.VolumesMounts...)
-	}
-
 	// Mount Position DB
 	if fb.Spec.PositionDB != (corev1.VolumeSource{}) {
 		ds.Spec.Template.Spec.Volumes = append(ds.Spec.Template.Spec.Volumes, corev1.Volume{
@@ -227,3 +161,98 @@ func MakeDaemonSet(fb fluentbitv1alpha2.FluentBit, logPath string) *appsv1.Daemo
 
 	return &ds
 }
+
+func makeVolumeMounts(fb fluentbitv1alpha2.FluentBit, logPath string) []corev1.VolumeMount {
+	internalMountPropagation := corev1.MountPropagationNone
+	if fb.Spec.InternalMountPropagation != nil {
+		internalMountPropagation = *fb.Spec.InternalMountPropagation
+	}
+
+	volumeMounts := []corev1.VolumeMount{
+		{
+			Name:      "config",
+			ReadOnly:  true,
+			MountPath: "/fluent-bit/config",
+		},
+	}
+
+	if !fb.Spec.DisableLogVolumes {
+		logVolumes := []corev1.VolumeMount{
+			{
+				Name:             "varlibcontainers",
+				ReadOnly:         true,
+				MountPath:        logPath,
+				MountPropagation: &internalMountPropagation,
+			},
+
+			{
+				Name:             "varlogs",
+				ReadOnly:         true,
+				MountPath:        "/var/log/",
+				MountPropagation: &internalMountPropagation,
+			},
+			{
+				Name:             "systemd",
+				ReadOnly:         true,
+				MountPath:        "/var/log/journal",
+				MountPropagation: &internalMountPropagation,
+			},
+		}
+		volumeMounts = append(volumeMounts, logVolumes...)
+	}
+
+	if fb.Spec.VolumesMounts != nil {
+		volumeMounts = append(volumeMounts, fb.Spec.VolumesMounts...)
+	}
+
+	return volumeMounts
+}
+
+func makeVolumes(fb fluentbitv1alpha2.FluentBit, logPath string) []corev1.Volume {
+
+	volumes := []corev1.Volume{
+		{
+			Name: "config",
+			VolumeSource: corev1.VolumeSource{
+				Secret: &corev1.SecretVolumeSource{
+					SecretName: fb.Spec.FluentBitConfigName,
+				},
+			},
+		},
+	}
+
+	if !fb.Spec.DisableLogVolumes {
+		logVolumes := []corev1.Volume{
+			{
+				Name: "varlibcontainers",
+				VolumeSource: corev1.VolumeSource{
+					HostPath: &corev1.HostPathVolumeSource{
+						Path: logPath,
+					},
+				},
+			},
+			{
+				Name: "varlogs",
+				VolumeSource: corev1.VolumeSource{
+					HostPath: &corev1.HostPathVolumeSource{
+						Path: "/var/log",
+					},
+				},
+			},
+			{
+				Name: "systemd",
+				VolumeSource: corev1.VolumeSource{
+					HostPath: &corev1.HostPathVolumeSource{
+						Path: "/var/log/journal",
+					},
+				},
+			},
+		}
+		volumes = append(volumes, logVolumes...)
+	}
+
+	if fb.Spec.Volumes != nil {
+		volumes = append(volumes, fb.Spec.Volumes...)
+	}
+	return volumes
+}