diff --git a/README.md b/README.md
index 6052dffd4..3789c95bf 100644
--- a/README.md
+++ b/README.md
@@ -52,6 +52,7 @@ Once installed, the Fluent Operator provides the following features:
 		- [Monitoring](#monitoring)
 		- [Custom Parser](#custom-parser)
 		- [Misc](#misc)
+        - [Custom Plugin](#custom-plugin)
 	- [Roadmap](#roadmap)
 	- [Development](#development)
 		- [Requirements](#requirements)
@@ -264,6 +265,9 @@ For more info on various use cases of Fluent Operator Fluentd CRDs, you can refe
 
 If you want to learn more about Fluent-Operator, please refer to the [misc](docs/best-practice/misc.md).
 
+### Custom Plugin
+[Here](docs/best-practice/custom-plugin.md) you can learn how to use custom plugin in Fluent Bit.
+
 ## Roadmap
 
 - [x] Support containerd log format
diff --git a/apis/fluentbit/v1alpha2/clusterfilter_types.go b/apis/fluentbit/v1alpha2/clusterfilter_types.go
index 9b0be99e5..122abad85 100644
--- a/apis/fluentbit/v1alpha2/clusterfilter_types.go
+++ b/apis/fluentbit/v1alpha2/clusterfilter_types.go
@@ -25,6 +25,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	"github.com/fluent/fluent-operator/apis/fluentbit/v1alpha2/plugins"
+	"github.com/fluent/fluent-operator/apis/fluentbit/v1alpha2/plugins/custom"
 	"github.com/fluent/fluent-operator/apis/fluentbit/v1alpha2/plugins/filter"
 )
 
@@ -66,6 +67,8 @@ type FilterItem struct {
 	AWS *filter.AWS `json:"aws,omitempty"`
 	//Multiline defines a Multiline configuration.
 	Multiline *filter.Multiline `json:"multiline,omitempty"`
+	//CustomPlugin defines a Custom plugin configuration.
+	CustomPlugin *custom.CustomPlugin `json:"customPlugin,omitempty"`
 }
 
 // +kubebuilder:object:root=true
@@ -111,7 +114,9 @@ func (list ClusterFilterList) Load(sl plugins.SecretLoader) (string, error) {
 			}
 
 			buf.WriteString("[Filter]\n")
-			buf.WriteString(fmt.Sprintf("    Name    %s\n", p.Name()))
+			if p.Name() != "" {
+				buf.WriteString(fmt.Sprintf("    Name    %s\n", p.Name()))
+			}
 			if item.Spec.Match != "" {
 				buf.WriteString(fmt.Sprintf("    Match    %s\n", item.Spec.Match))
 			}
diff --git a/apis/fluentbit/v1alpha2/clusterfluentbitconfig_types_test.go b/apis/fluentbit/v1alpha2/clusterfluentbitconfig_types_test.go
index 4c303d428..eb5efc1c7 100644
--- a/apis/fluentbit/v1alpha2/clusterfluentbitconfig_types_test.go
+++ b/apis/fluentbit/v1alpha2/clusterfluentbitconfig_types_test.go
@@ -3,12 +3,14 @@ package v1alpha2
 import (
 	"testing"
 
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
 	"github.com/fluent/fluent-operator/apis/fluentbit/v1alpha2/plugins"
+	"github.com/fluent/fluent-operator/apis/fluentbit/v1alpha2/plugins/custom"
 	"github.com/fluent/fluent-operator/apis/fluentbit/v1alpha2/plugins/filter"
 	"github.com/fluent/fluent-operator/apis/fluentbit/v1alpha2/plugins/input"
 	"github.com/fluent/fluent-operator/apis/fluentbit/v1alpha2/plugins/output"
 	. "github.com/onsi/gomega"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
 var expected = `[Service]
@@ -65,6 +67,16 @@ var expected = `[Service]
     json_date_format    iso8601
     tls    On
     tls.verify    true
+[Output]
+    Name    kafka
+    Topics    fluentbit
+    Match    *
+    Brokers    192.168.100.32:9092
+    rdkafka.debug    All
+    rdkafka.request.required.acks    1
+    rdkafka.log.connection.close    false
+    rdkafka.log_level    7
+    rdkafka.metadata.broker.list    192.168.100.32:9092
 [Output]
     Name    opensearch
     Match    *
@@ -271,8 +283,24 @@ func Test_FluentBitConfig_RenderMainConfig(t *testing.T) {
 		},
 	}
 
+	kafkaOutput := ClusterOutput{
+		TypeMeta: metav1.TypeMeta{
+			APIVersion: "fluentbit.fluent.io/v1alpha2",
+			Kind:       "ClusterOutput",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:   "kafka_output",
+			Labels: labels,
+		},
+		Spec: OutputSpec{
+			CustomPlugin: &custom.CustomPlugin{
+				Config: "    Name    kafka\n    Topics    fluentbit\n    Match    *\n    Brokers    192.168.100.32:9092\n    rdkafka.debug    All\n    rdkafka.request.required.acks    1\n    rdkafka.log.connection.close    false\n    rdkafka.log_level    7\n    rdkafka.metadata.broker.list    192.168.100.32:9092",
+			},
+		},
+	}
+
 	outputs := ClusterOutputList{
-		Items: []ClusterOutput{syslogOut, httpOutput, openSearchOutput},
+		Items: []ClusterOutput{syslogOut, httpOutput, openSearchOutput, kafkaOutput},
 	}
 
 	cfg := ClusterFluentBitConfig{
diff --git a/apis/fluentbit/v1alpha2/clusterinput_types.go b/apis/fluentbit/v1alpha2/clusterinput_types.go
index 84f4293b4..a419933ae 100644
--- a/apis/fluentbit/v1alpha2/clusterinput_types.go
+++ b/apis/fluentbit/v1alpha2/clusterinput_types.go
@@ -22,9 +22,11 @@ import (
 	"reflect"
 	"sort"
 
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
 	"github.com/fluent/fluent-operator/apis/fluentbit/v1alpha2/plugins"
+	"github.com/fluent/fluent-operator/apis/fluentbit/v1alpha2/plugins/custom"
 	"github.com/fluent/fluent-operator/apis/fluentbit/v1alpha2/plugins/input"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
 // EDIT THIS FILE!  THIS IS SCAFFOLDING FOR YOU TO OWN!
@@ -47,6 +49,8 @@ type InputSpec struct {
 	PrometheusScrapeMetrics *input.PrometheusScrapeMetrics `json:"prometheusScrapeMetrics,omitempty"`
 	// FluentBitMetrics defines Fluent Bit Metrics Input configuration.
 	FluentBitMetrics *input.FluentbitMetrics `json:"fluentBitMetrics,omitempty"`
+	// CustomPlugin defines Custom Input configuration.
+	CustomPlugin *custom.CustomPlugin `json:"customPlugin,omitempty"`
 }
 
 // +kubebuilder:object:root=true
@@ -91,7 +95,9 @@ func (list ClusterInputList) Load(sl plugins.SecretLoader) (string, error) {
 			}
 
 			buf.WriteString("[Input]\n")
-			buf.WriteString(fmt.Sprintf("    Name    %s\n", p.Name()))
+			if p.Name() != "" {
+				buf.WriteString(fmt.Sprintf("    Name    %s\n", p.Name()))
+			}
 			if item.Spec.Alias != "" {
 				buf.WriteString(fmt.Sprintf("    Alias    %s\n", item.Spec.Alias))
 			}
diff --git a/apis/fluentbit/v1alpha2/clusteroutput_types.go b/apis/fluentbit/v1alpha2/clusteroutput_types.go
index 74229df07..3dcdcfa45 100644
--- a/apis/fluentbit/v1alpha2/clusteroutput_types.go
+++ b/apis/fluentbit/v1alpha2/clusteroutput_types.go
@@ -25,6 +25,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	"github.com/fluent/fluent-operator/apis/fluentbit/v1alpha2/plugins"
+	"github.com/fluent/fluent-operator/apis/fluentbit/v1alpha2/plugins/custom"
 	"github.com/fluent/fluent-operator/apis/fluentbit/v1alpha2/plugins/output"
 )
 
@@ -75,6 +76,8 @@ type OutputSpec struct {
 	OpenTelemetry *output.OpenTelemetry `json:"opentelemetry,omitempty"`
 	// PrometheusRemoteWrite_types defines Prometheus Remote Write configuration.
 	PrometheusRemoteWrite *output.PrometheusRemoteWrite `json:"prometheusRemoteWrite,omitempty"`
+	// CustomPlugin defines Custom Output configuration.
+	CustomPlugin *custom.CustomPlugin `json:"customPlugin,omitempty"`
 }
 
 // +kubebuilder:object:root=true
@@ -119,7 +122,9 @@ func (list ClusterOutputList) Load(sl plugins.SecretLoader) (string, error) {
 			}
 
 			buf.WriteString("[Output]\n")
-			buf.WriteString(fmt.Sprintf("    Name    %s\n", p.Name()))
+			if p.Name() != "" {
+				buf.WriteString(fmt.Sprintf("    Name    %s\n", p.Name()))
+			}
 			if item.Spec.Match != "" {
 				buf.WriteString(fmt.Sprintf("    Match    %s\n", item.Spec.Match))
 			}
diff --git a/apis/fluentbit/v1alpha2/plugins/custom/custom_types.go b/apis/fluentbit/v1alpha2/plugins/custom/custom_types.go
new file mode 100644
index 000000000..01c33faa4
--- /dev/null
+++ b/apis/fluentbit/v1alpha2/plugins/custom/custom_types.go
@@ -0,0 +1,37 @@
+package custom
+
+import (
+	"errors"
+	"fmt"
+	"strings"
+
+	"github.com/fluent/fluent-operator/apis/fluentbit/v1alpha2/plugins"
+	"github.com/fluent/fluent-operator/apis/fluentbit/v1alpha2/plugins/params"
+)
+
+// +kubebuilder:object:generate:=true
+
+// CustomPlugin is used to support filter plugins that are not implemented yet
+type CustomPlugin struct {
+	Config string `json:"config,omitempty"`
+}
+
+func (c *CustomPlugin) Name() string {
+	return ""
+}
+
+func (a *CustomPlugin) Params(_ plugins.SecretLoader) (*params.KVs, error) {
+	kvs := params.NewKVs()
+	splits := strings.Split(a.Config, "\n")
+	for _, i := range splits {
+		if len(i) == 0 {
+			continue
+		}
+		fields := strings.Fields(i)
+		if len(fields) < 2 {
+			return nil, errors.New(fmt.Sprintf("invalid plugin config: %s", i))
+		}
+		kvs.Insert(fields[0], strings.Join(fields[1:], " "))
+	}
+	return kvs, nil
+}
diff --git a/apis/fluentbit/v1alpha2/plugins/custom/zz_generated.deepcopy.go b/apis/fluentbit/v1alpha2/plugins/custom/zz_generated.deepcopy.go
new file mode 100644
index 000000000..9823f1e91
--- /dev/null
+++ b/apis/fluentbit/v1alpha2/plugins/custom/zz_generated.deepcopy.go
@@ -0,0 +1,39 @@
+//go:build !ignore_autogenerated
+// +build !ignore_autogenerated
+
+/*
+Copyright 2022.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by controller-gen. DO NOT EDIT.
+
+package custom
+
+import ()
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *CustomPlugin) DeepCopyInto(out *CustomPlugin) {
+	*out = *in
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new CustomPlugin.
+func (in *CustomPlugin) DeepCopy() *CustomPlugin {
+	if in == nil {
+		return nil
+	}
+	out := new(CustomPlugin)
+	in.DeepCopyInto(out)
+	return out
+}
diff --git a/apis/fluentbit/v1alpha2/zz_generated.deepcopy.go b/apis/fluentbit/v1alpha2/zz_generated.deepcopy.go
index 6e27426bd..5919df924 100644
--- a/apis/fluentbit/v1alpha2/zz_generated.deepcopy.go
+++ b/apis/fluentbit/v1alpha2/zz_generated.deepcopy.go
@@ -22,6 +22,7 @@ limitations under the License.
 package v1alpha2
 
 import (
+	"github.com/fluent/fluent-operator/apis/fluentbit/v1alpha2/plugins/custom"
 	"github.com/fluent/fluent-operator/apis/fluentbit/v1alpha2/plugins/filter"
 	"github.com/fluent/fluent-operator/apis/fluentbit/v1alpha2/plugins/input"
 	"github.com/fluent/fluent-operator/apis/fluentbit/v1alpha2/plugins/output"
@@ -393,6 +394,11 @@ func (in *FilterItem) DeepCopyInto(out *FilterItem) {
 		*out = new(filter.Multiline)
 		(*in).DeepCopyInto(*out)
 	}
+	if in.CustomPlugin != nil {
+		in, out := &in.CustomPlugin, &out.CustomPlugin
+		*out = new(custom.CustomPlugin)
+		**out = **in
+	}
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new FilterItem.
@@ -640,6 +646,11 @@ func (in *InputSpec) DeepCopyInto(out *InputSpec) {
 		*out = new(input.FluentbitMetrics)
 		(*in).DeepCopyInto(*out)
 	}
+	if in.CustomPlugin != nil {
+		in, out := &in.CustomPlugin, &out.CustomPlugin
+		*out = new(custom.CustomPlugin)
+		**out = **in
+	}
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new InputSpec.
@@ -730,6 +741,11 @@ func (in *OutputSpec) DeepCopyInto(out *OutputSpec) {
 		*out = new(output.PrometheusRemoteWrite)
 		(*in).DeepCopyInto(*out)
 	}
+	if in.CustomPlugin != nil {
+		in, out := &in.CustomPlugin, &out.CustomPlugin
+		*out = new(custom.CustomPlugin)
+		**out = **in
+	}
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new OutputSpec.
diff --git a/charts/fluent-operator/crds/fluentbit.fluent.io_clusterfilters.yaml b/charts/fluent-operator/crds/fluentbit.fluent.io_clusterfilters.yaml
index 04bbe26e5..bed6efd54 100644
--- a/charts/fluent-operator/crds/fluentbit.fluent.io_clusterfilters.yaml
+++ b/charts/fluent-operator/crds/fluentbit.fluent.io_clusterfilters.yaml
@@ -84,6 +84,12 @@ spec:
                             is false.
                           type: boolean
                       type: object
+                    customPlugin:
+                      description: CustomPlugin defines a Custom plugin configuration.
+                      properties:
+                        config:
+                          type: string
+                      type: object
                     grep:
                       description: Grep defines Grep Filter configuration.
                       properties:
diff --git a/charts/fluent-operator/crds/fluentbit.fluent.io_clusterinputs.yaml b/charts/fluent-operator/crds/fluentbit.fluent.io_clusterinputs.yaml
index 55f6b8bed..11bd66f7b 100644
--- a/charts/fluent-operator/crds/fluentbit.fluent.io_clusterinputs.yaml
+++ b/charts/fluent-operator/crds/fluentbit.fluent.io_clusterinputs.yaml
@@ -42,6 +42,12 @@ spec:
                 description: A user friendly alias name for this input plugin. Used
                   in metrics for distinction of each configured input.
                 type: string
+              customPlugin:
+                description: CustomPlugin defines Custom Input configuration.
+                properties:
+                  config:
+                    type: string
+                type: object
               dummy:
                 description: Dummy defines Dummy Input configuration.
                 properties:
diff --git a/charts/fluent-operator/crds/fluentbit.fluent.io_clusteroutputs.yaml b/charts/fluent-operator/crds/fluentbit.fluent.io_clusteroutputs.yaml
index a9127de9b..ca420ad98 100644
--- a/charts/fluent-operator/crds/fluentbit.fluent.io_clusteroutputs.yaml
+++ b/charts/fluent-operator/crds/fluentbit.fluent.io_clusteroutputs.yaml
@@ -42,6 +42,12 @@ spec:
                 description: A user friendly alias name for this output plugin. Used
                   in metrics for distinction of each configured output.
                 type: string
+              customPlugin:
+                description: CustomPlugin defines Custom Output configuration.
+                properties:
+                  config:
+                    type: string
+                type: object
               datadog:
                 description: DataDog defines DataDog Output configuration.
                 properties:
diff --git a/config/crd/bases/fluentbit.fluent.io_clusterfilters.yaml b/config/crd/bases/fluentbit.fluent.io_clusterfilters.yaml
index 04bbe26e5..bed6efd54 100644
--- a/config/crd/bases/fluentbit.fluent.io_clusterfilters.yaml
+++ b/config/crd/bases/fluentbit.fluent.io_clusterfilters.yaml
@@ -84,6 +84,12 @@ spec:
                             is false.
                           type: boolean
                       type: object
+                    customPlugin:
+                      description: CustomPlugin defines a Custom plugin configuration.
+                      properties:
+                        config:
+                          type: string
+                      type: object
                     grep:
                       description: Grep defines Grep Filter configuration.
                       properties:
diff --git a/config/crd/bases/fluentbit.fluent.io_clusterinputs.yaml b/config/crd/bases/fluentbit.fluent.io_clusterinputs.yaml
index 55f6b8bed..11bd66f7b 100644
--- a/config/crd/bases/fluentbit.fluent.io_clusterinputs.yaml
+++ b/config/crd/bases/fluentbit.fluent.io_clusterinputs.yaml
@@ -42,6 +42,12 @@ spec:
                 description: A user friendly alias name for this input plugin. Used
                   in metrics for distinction of each configured input.
                 type: string
+              customPlugin:
+                description: CustomPlugin defines Custom Input configuration.
+                properties:
+                  config:
+                    type: string
+                type: object
               dummy:
                 description: Dummy defines Dummy Input configuration.
                 properties:
diff --git a/config/crd/bases/fluentbit.fluent.io_clusteroutputs.yaml b/config/crd/bases/fluentbit.fluent.io_clusteroutputs.yaml
index a9127de9b..ca420ad98 100644
--- a/config/crd/bases/fluentbit.fluent.io_clusteroutputs.yaml
+++ b/config/crd/bases/fluentbit.fluent.io_clusteroutputs.yaml
@@ -42,6 +42,12 @@ spec:
                 description: A user friendly alias name for this output plugin. Used
                   in metrics for distinction of each configured output.
                 type: string
+              customPlugin:
+                description: CustomPlugin defines Custom Output configuration.
+                properties:
+                  config:
+                    type: string
+                type: object
               datadog:
                 description: DataDog defines DataDog Output configuration.
                 properties:
diff --git a/docs/best-practice/custom-plugin.md b/docs/best-practice/custom-plugin.md
new file mode 100644
index 000000000..47cd54045
--- /dev/null
+++ b/docs/best-practice/custom-plugin.md
@@ -0,0 +1,43 @@
+# Custom Plugin
+
+
+You can use custom plugin configuration to support plugins that fluent-operator does not currently support. The following are some examples of how to use the custom plugin configuration.
+
+```yaml 
+apiVersion: fluentbit.fluent.io/v1alpha2
+kind: ClusterInput
+metadata:
+  namespace: fluent
+  name: cpu-input
+  labels:
+    fluentbit.fluent.io/enabled: "true"
+    fluentbit.fluent.io/mode: "fluentbit-only"
+spec:
+  customPlugin:
+    config: |
+      Name    cpu
+      Tag    my_cpu
+```
+
+```yaml
+apiVersion: fluentbit.fluent.io/v1alpha2
+kind: ClusterOutput
+metadata:
+  namespace: fluent
+  name: kafka-output
+  labels:
+    fluentbit.fluent.io/enabled: "true"
+    fluentbit.fluent.io/mode: "fluentbit-only"
+spec:
+  customPlugin:
+    config: |
+      Name    kafka
+      Topics     fluentbit
+      Match       *
+      Brokers     192.168.100.32:9092
+      rdkafka.debug All
+      rdkafka.request.required.acks 1
+      rdkafka.log.connection.close false
+      rdkafka.log_level 7
+      rdkafka.metadata.broker.list 192.168.100.32:9092
+```
\ No newline at end of file
diff --git a/docs/fluentbit.md b/docs/fluentbit.md
index b8d4705e4..3218d45c2 100644
--- a/docs/fluentbit.md
+++ b/docs/fluentbit.md
@@ -3,31 +3,27 @@ This Document documents the types introduced by the fluentbit Operator.
 > Note this document is generated from code comments. When contributing a change to this document please do so by changing the code comments.
 ## Table of Contents
 * [ClusterFilter](#clusterfilter)
+* [ClusterFilterList](#clusterfilterlist)
 * [ClusterFluentBitConfig](#clusterfluentbitconfig)
+* [ClusterFluentBitConfigList](#clusterfluentbitconfiglist)
 * [ClusterInput](#clusterinput)
+* [ClusterInputList](#clusterinputlist)
 * [ClusterOutput](#clusteroutput)
+* [ClusterOutputList](#clusteroutputlist)
 * [ClusterParser](#clusterparser)
+* [ClusterParserList](#clusterparserlist)
 * [Decorder](#decorder)
 * [FilterItem](#filteritem)
-* [ClusterFilterList](#clusterFilterList)
 * [FilterSpec](#filterspec)
 * [FluentBit](#fluentbit)
-* [ClusterFluentBitConfigList](#clusterfluentBitconfiglist)
 * [FluentBitConfigSpec](#fluentbitconfigspec)
 * [FluentBitList](#fluentbitlist)
 * [FluentBitSpec](#fluentbitspec)
-* [ClusterInputList](#clusterinputlist)
 * [InputSpec](#inputspec)
-* [ClusterOutputList](#clusteroutputlist)
 * [OutputSpec](#outputspec)
-* [ClusterParserList](#clusterparserlist)
 * [ParserSpec](#parserspec)
 * [Script](#script)
 * [Service](#service)
-* [Monitoring](#monitoring)
-* [Path Convention](#path Convention)
-* [Custom Parser](#custom Parser)
-
 # ClusterFilter
 
 ClusterFilter defines a cluster-level Filter configuration.
@@ -38,6 +34,17 @@ ClusterFilter defines a cluster-level Filter configuration.
 | metadata |  | [metav1.ObjectMeta](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.17/#objectmeta-v1-meta) |
 | spec | Specification of desired Filter configuration. | FilterSpec |
 
+[Back to TOC](#table-of-contents)
+# ClusterFilterList
+
+ClusterFilterList contains a list of ClusterFilter
+
+
+| Field | Description | Scheme |
+| ----- | ----------- | ------ |
+| metadata |  | [metav1.ListMeta](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.17/#listmeta-v1-meta) |
+| items |  | []ClusterFilter |
+
 [Back to TOC](#table-of-contents)
 # ClusterFluentBitConfig
 
@@ -49,6 +56,17 @@ ClusterFluentBitConfig is the Schema for the cluster-level fluentbitconfigs API
 | metadata |  | [metav1.ObjectMeta](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.17/#objectmeta-v1-meta) |
 | spec |  | FluentBitConfigSpec |
 
+[Back to TOC](#table-of-contents)
+# ClusterFluentBitConfigList
+
+ClusterFluentBitConfigList contains a list of ClusterFluentBitConfig
+
+
+| Field | Description | Scheme |
+| ----- | ----------- | ------ |
+| metadata |  | [metav1.ListMeta](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.17/#listmeta-v1-meta) |
+| items |  | []ClusterFluentBitConfig |
+
 [Back to TOC](#table-of-contents)
 # ClusterInput
 
@@ -60,6 +78,17 @@ ClusterInput is the Schema for the inputs API
 | metadata |  | [metav1.ObjectMeta](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.17/#objectmeta-v1-meta) |
 | spec |  | InputSpec |
 
+[Back to TOC](#table-of-contents)
+# ClusterInputList
+
+ClusterInputList contains a list of ClusterInput
+
+
+| Field | Description | Scheme |
+| ----- | ----------- | ------ |
+| metadata |  | [metav1.ListMeta](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.17/#listmeta-v1-meta) |
+| items |  | []ClusterInput |
+
 [Back to TOC](#table-of-contents)
 # ClusterOutput
 
@@ -71,6 +100,17 @@ ClusterOutput is the Schema for the cluster-level outputs API
 | metadata |  | [metav1.ObjectMeta](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.17/#objectmeta-v1-meta) |
 | spec |  | OutputSpec |
 
+[Back to TOC](#table-of-contents)
+# ClusterOutputList
+
+ClusterOutputList contains a list of ClusterOutput
+
+
+| Field | Description | Scheme |
+| ----- | ----------- | ------ |
+| metadata |  | [metav1.ListMeta](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.17/#listmeta-v1-meta) |
+| items |  | []ClusterOutput |
+
 [Back to TOC](#table-of-contents)
 # ClusterParser
 
@@ -82,6 +122,17 @@ ClusterParser is the Schema for the cluster-level parsers API
 | metadata |  | [metav1.ObjectMeta](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.17/#objectmeta-v1-meta) |
 | spec |  | ParserSpec |
 
+[Back to TOC](#table-of-contents)
+# ClusterParserList
+
+ClusterParserList contains a list of ClusterParser
+
+
+| Field | Description | Scheme |
+| ----- | ----------- | ------ |
+| metadata |  | [metav1.ListMeta](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.17/#listmeta-v1-meta) |
+| items |  | []ClusterParser |
+
 [Back to TOC](#table-of-contents)
 # Decorder
 
@@ -112,17 +163,7 @@ ClusterParser is the Schema for the cluster-level parsers API
 | rewriteTag | RewriteTag defines a RewriteTag configuration. | *[filter.RewriteTag](plugins/filter/rewritetag.md) |
 | aws | Aws defines a Aws configuration. | *[filter.AWS](plugins/filter/aws.md) |
 | multiline | Multiline defines a Multiline configuration. | *[filter.Multiline](plugins/filter/multiline.md) |
-
-[Back to TOC](#table-of-contents)
-# FilterList
-
-FilterList contains a list of ClusterFilter
-
-
-| Field | Description | Scheme |
-| ----- | ----------- | ------ |
-| metadata |  | [metav1.ListMeta](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.17/#listmeta-v1-meta) |
-| items |  | []ClusterFilter |
+| customPlugin | CustomPlugin defines a Custom plugin configuration. | *custom.CustomPlugin |
 
 [Back to TOC](#table-of-contents)
 # FilterSpec
@@ -148,17 +189,6 @@ FluentBit is the Schema for the fluentbits API
 | spec |  | FluentBitSpec |
 | status |  | FluentBitStatus |
 
-[Back to TOC](#table-of-contents)
-# ClusterFluentBitConfigList
-
-ClusterFluentBitConfigList contains a list of ClusterFluentBitConfig
-
-
-| Field | Description | Scheme |
-| ----- | ----------- | ------ |
-| metadata |  | [metav1.ListMeta](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.17/#listmeta-v1-meta) |
-| items |  | []ClusterFluentBitConfig |
-
 [Back to TOC](#table-of-contents)
 # FluentBitConfigSpec
 
@@ -209,17 +239,9 @@ FluentBitSpec defines the desired state of FluentBit
 | priorityClassName | PriorityClassName represents the pod's priority class. | string |
 | volumes | List of volumes that can be mounted by containers belonging to the pod. | []corev1.Volume |
 | volumesMounts | Pod volumes to mount into the container's filesystem. | []corev1.VolumeMount |
-
-[Back to TOC](#table-of-contents)
-# ClusterInputList
-
-ClusterInputList contains a list of Input
-
-
-| Field | Description | Scheme |
-| ----- | ----------- | ------ |
-| metadata |  | [metav1.ListMeta](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.17/#listmeta-v1-meta) |
-| items |  | []ClusterInput |
+| annotations | Annotations to add to each Fluentbit pod. | map[string]string |
+| securityContext | SecurityContext holds pod-level security attributes and common container settings. | *corev1.PodSecurityContext |
+| hostNetwork | Host networking is requested for this pod. Use the host's network namespace. If this option is set, the ports that will be used must be specified. Default to false. | bool |
 
 [Back to TOC](#table-of-contents)
 # InputSpec
@@ -233,17 +255,10 @@ InputSpec defines the desired state of ClusterInput
 | dummy | Dummy defines Dummy Input configuration. | *[input.Dummy](plugins/input/dummy.md) |
 | tail | Tail defines Tail Input configuration. | *[input.Tail](plugins/input/tail.md) |
 | systemd | Systemd defines Systemd Input configuration. | *[input.Systemd](plugins/input/systemd.md) |
-
-[Back to TOC](#table-of-contents)
-# ClusterOutputList
-
-ClusterOutputList contains a list of ClusterOutput
-
-
-| Field | Description | Scheme |
-| ----- | ----------- | ------ |
-| metadata |  | [metav1.ListMeta](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.17/#listmeta-v1-meta) |
-| items |  | []ClusterOutput |
+| nodeExporterMetrics | NodeExporterMetrics defines Node Exporter Metrics Input configuration. | *[input.NodeExporterMetrics](plugins/input/nodeexportermetrics.md) |
+| prometheusScrapeMetrics | PrometheusScrapeMetrics  defines Prometheus Scrape Metrics Input configuration. | *[input.PrometheusScrapeMetrics](plugins/input/prometheusscrapemetrics.md) |
+| fluentBitMetrics | FluentBitMetrics defines Fluent Bit Metrics Input configuration. | *[input.FluentbitMetrics](plugins/input/fluentbitmetrics.md) |
+| customPlugin | CustomPlugin defines Custom Input configuration. | *custom.CustomPlugin |
 
 [Back to TOC](#table-of-contents)
 # OutputSpec
@@ -269,17 +284,10 @@ OutputSpec defines the desired state of ClusterOutput
 | syslog | Syslog defines Syslog Output configuration. | *[output.Syslog](plugins/output/syslog.md) |
 | datadog | DataDog defines DataDog Output configuration. | *[output.DataDog](plugins/output/datadog.md) |
 | firehose | Firehose defines Firehose Output configuration. | *[output.Firehose](plugins/output/firehose.md) |
-
-[Back to TOC](#table-of-contents)
-# ClusterParserList
-
-ClusterParserList contains a list of ClusterParser
-
-
-| Field | Description | Scheme |
-| ----- | ----------- | ------ |
-| metadata |  | [metav1.ListMeta](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.17/#listmeta-v1-meta) |
-| items |  | []ClusterParser |
+| opensearch | OpenSearch defines OpenSearch Output configuration. | *[output.OpenSearch](plugins/output/opensearch.md) |
+| opentelemetry | OpenTelemetry defines OpenTelemetry Output configuration. | *[output.OpenTelemetry](plugins/output/opentelemetry.md) |
+| prometheusRemoteWrite | PrometheusRemoteWrite_types defines Prometheus Remote Write configuration. | *[output.PrometheusRemoteWrite](plugins/output/prometheusremotewrite.md) |
+| customPlugin | CustomPlugin defines Custom Output configuration. | *custom.CustomPlugin |
 
 [Back to TOC](#table-of-contents)
 # ParserSpec
diff --git a/docs/plugins/fluentbit/clusterinput/fluentbit-metrics.md b/docs/plugins/fluentbit/clusterinput/fluentbit-metrics.md
new file mode 100644
index 000000000..56cd7c2d8
--- /dev/null
+++ b/docs/plugins/fluentbit/clusterinput/fluentbit-metrics.md
@@ -0,0 +1,12 @@
+# FluentbitMetrics
+
+Fluent Bit exposes its own metrics to allow you to monitor the internals of your pipeline. 
+The collected metrics can be processed similarly to those from the Prometheus Node Exporter input plugin. 
+They can be sent to output plugins including Prometheus Exporter, Prometheus Remote Write or OpenTelemetry.
+> Important note: Metrics collected with Node Exporter Metrics flow through a separate pipeline from logs and current filters do not operate on top of metrics.
+
+| Field | Description | Scheme |
+| ----- | ----------- |--------|
+| tag | Tag name associated to all records comming from this plugin. | string |
+| scrape_interval | The rate at which metrics are collected from the host operating system. | string |
+| scrape_on_start | Scrape metrics upon start, useful to avoid waiting for 'scrape_interval' for the first round of metrics. | *bool  |
diff --git a/docs/plugins/fluentbit/index.md b/docs/plugins/fluentbit/index.md
index 808297889..ce7e94e94 100644
--- a/docs/plugins/fluentbit/index.md
+++ b/docs/plugins/fluentbit/index.md
@@ -3,6 +3,7 @@
   - [tail](clusterinput/tail.md)
   - [systemd](clusterinput/systemd.md)
   - [node exporter metrics](clusterinput/node-exporter-metrics.md)
+  - [fluentbit metrics](clusterinput/fluentbit-metrics.md)
 - [ClusterParser](../../fluentbit.md#ClusterParser)
   - [json](clusterparser/json.md)
   - [logfmt](clusterparser/logfmt.md)
diff --git a/manifests/setup/fluent-operator-crd.yaml b/manifests/setup/fluent-operator-crd.yaml
index 4104dae61..66f81001c 100644
--- a/manifests/setup/fluent-operator-crd.yaml
+++ b/manifests/setup/fluent-operator-crd.yaml
@@ -81,6 +81,12 @@ spec:
                             is false.
                           type: boolean
                       type: object
+                    customPlugin:
+                      description: CustomPlugin defines a Custom plugin configuration.
+                      properties:
+                        config:
+                          type: string
+                      type: object
                     grep:
                       description: Grep defines Grep Filter configuration.
                       properties:
@@ -1487,6 +1493,12 @@ spec:
                 description: A user friendly alias name for this input plugin. Used
                   in metrics for distinction of each configured input.
                 type: string
+              customPlugin:
+                description: CustomPlugin defines Custom Input configuration.
+                properties:
+                  config:
+                    type: string
+                type: object
               dummy:
                 description: Dummy defines Dummy Input configuration.
                 properties:
@@ -1836,6 +1848,12 @@ spec:
                 description: A user friendly alias name for this output plugin. Used
                   in metrics for distinction of each configured output.
                 type: string
+              customPlugin:
+                description: CustomPlugin defines Custom Output configuration.
+                properties:
+                  config:
+                    type: string
+                type: object
               datadog:
                 description: DataDog defines DataDog Output configuration.
                 properties:
diff --git a/manifests/setup/setup.yaml b/manifests/setup/setup.yaml
index e57ae7a74..0e01ea7cc 100644
--- a/manifests/setup/setup.yaml
+++ b/manifests/setup/setup.yaml
@@ -81,6 +81,12 @@ spec:
                             is false.
                           type: boolean
                       type: object
+                    customPlugin:
+                      description: CustomPlugin defines a Custom plugin configuration.
+                      properties:
+                        config:
+                          type: string
+                      type: object
                     grep:
                       description: Grep defines Grep Filter configuration.
                       properties:
@@ -1487,6 +1493,12 @@ spec:
                 description: A user friendly alias name for this input plugin. Used
                   in metrics for distinction of each configured input.
                 type: string
+              customPlugin:
+                description: CustomPlugin defines Custom Input configuration.
+                properties:
+                  config:
+                    type: string
+                type: object
               dummy:
                 description: Dummy defines Dummy Input configuration.
                 properties:
@@ -1836,6 +1848,12 @@ spec:
                 description: A user friendly alias name for this output plugin. Used
                   in metrics for distinction of each configured output.
                 type: string
+              customPlugin:
+                description: CustomPlugin defines Custom Output configuration.
+                properties:
+                  config:
+                    type: string
+                type: object
               datadog:
                 description: DataDog defines DataDog Output configuration.
                 properties: