diff --git a/pkg/config/config.go b/pkg/config/config.go
index d8ab14d3d..09126aa3c 100644
--- a/pkg/config/config.go
+++ b/pkg/config/config.go
@@ -904,6 +904,14 @@ type Config struct {
 	// Public: Yes
 	LoggingRetryLimit string `yaml:"logging_retry_limit" envconfig:"logging_retry_limit" public:"true"`
 
+	// LoggingAddCustomAtts adds custom_attributes to fluent-bit forwarder tags.
+	// Default: False
+	LoggingAddCustomAtts bool `yaml:"logging_add_custom_attributes" envconfig:"logging_add_custom_attributes" public:"false"`
+
+	// LoggingDropAttFbInput drops fb.input from fluent-bit forwarder tags.
+	// Default: False
+	LoggingDropAttFbInput bool `yaml:"logging_drop_attribute_fb_input" envconfig:"logging_drop_attribute_fb_input" public:"false"`
+
 	// FluentBitExePath is the location from where the agent can execute fluent-bit.
 	// Default (Linux): /opt/td-agent-bit/bin/td-agent-bit
 	// Default (Windows): C:\Program Files\New Relic\newrelic-infra\newrelic-integrations\logging\fluent-bit
@@ -1511,6 +1519,8 @@ type LogForward struct {
 	ProxyCfg         LogForwardProxy
 	RetryLimit       string
 	FluentBitVerbose bool
+	CommonAttributes CustomAttributeMap
+	DropAttFbInput   bool
 }
 
 type LogForwardProxy struct {
@@ -1523,15 +1533,20 @@ type LogForwardProxy struct {
 
 // NewLogForward creates a valid log forwarder config.
 func NewLogForward(config *Config, troubleshoot Troubleshoot) LogForward {
+
+	// Add any config specified common attributes for logs
+	var commonAttributes CustomAttributeMap
+	if config.LoggingAddCustomAtts {
+		commonAttributes = config.CustomAttributes
+	}
+
 	return LogForward{
-		Troubleshoot:     troubleshoot,
-		ConfigsDir:       config.LoggingConfigsDir,
-		HomeDir:          config.LoggingHomeDir,
-		License:          config.License,
-		IsFedramp:        config.Fedramp,
-		IsStaging:        config.Staging,
-		RetryLimit:       config.LoggingRetryLimit,
-		FluentBitVerbose: config.Log.Level == LogLevelTrace && config.Log.HasIncludeFilter(TracesFieldName, SupervisorTrace),
+		Troubleshoot: troubleshoot,
+		ConfigsDir:   config.LoggingConfigsDir,
+		HomeDir:      config.LoggingHomeDir,
+		License:      config.License,
+		IsFedramp:    config.Fedramp,
+		IsStaging:    config.Staging,
 		ProxyCfg: LogForwardProxy{
 			IgnoreSystemProxy: config.IgnoreSystemProxy,
 			Proxy:             config.Proxy,
@@ -1539,6 +1554,10 @@ func NewLogForward(config *Config, troubleshoot Troubleshoot) LogForward {
 			CABundleDir:       config.CABundleDir,
 			ValidateCerts:     config.ProxyValidateCerts,
 		},
+		RetryLimit:       config.LoggingRetryLimit,
+		FluentBitVerbose: config.Log.Level == LogLevelTrace && config.Log.HasIncludeFilter(TracesFieldName, SupervisorTrace),
+		CommonAttributes: commonAttributes,
+		DropAttFbInput:   config.LoggingDropAttFbInput,
 	}
 }
 
@@ -1827,6 +1846,8 @@ func NewConfig() *Config {
 		DebugLogSec:                   defaultDebugLogSec,
 		TruncTextValues:               defaultTruncTextValues,
 		LogFormat:                     defaultLogFormat,
+		LoggingAddCustomAtts:          defaultLoggingAddCustomAtts,
+		LoggingDropAttFbInput:         defaultLoggingDropAttFbInput,
 		LoggingRetryLimit:             defaultLoggingRetryLimit,
 		HTTPServerHost:                defaultHTTPServerHost,
 		HTTPServerPort:                defaultHTTPServerPort,
diff --git a/pkg/config/defaults.go b/pkg/config/defaults.go
index 73e63fcef..96de127bb 100644
--- a/pkg/config/defaults.go
+++ b/pkg/config/defaults.go
@@ -79,6 +79,8 @@ var (
 	defaultLogFormat                     = LogFormatText
 	defaultLogLevel                      = LogLevelInfo
 	defaultLogForward                    = false
+	defaultLoggingAddCustomAtts          = false
+	defaultLoggingDropAttFbInput         = false
 	defaultLoggingRetryLimit             = "5"         // nolint:gochecknoglobals
 	defaultMaxInventorySize              = 1000 * 1000 // Size limit from Vortex collector service (1MB)
 	defaultPayloadCompressionLevel       = 6           // default compression level used in go, higher than this does not show tangible benefits
diff --git a/pkg/integrations/v4/logs/cfg.go b/pkg/integrations/v4/logs/cfg.go
index be02c4fe0..ccb9d6c00 100644
--- a/pkg/integrations/v4/logs/cfg.go
+++ b/pkg/integrations/v4/logs/cfg.go
@@ -287,7 +287,17 @@ func NewFBConf(loggingCfgs LogsCfg, logFwdCfg *config.LogForward, entityGUID, ho
 			fb.Inputs = append(fb.Inputs, input)
 		}
 
-		fb.Filters = append(fb.Filters, filters...)
+		for _, filter := range filters {
+			_, ok := filter.Records["fb.input"]
+			if logFwdCfg.DropAttFbInput && ok {
+				// Delete fb.input if in filters
+				delete(filter.Records, "fb.input")
+			}
+			// If any records remain, add to filters
+			if len(filter.Records) > 0 {
+				fb.Filters = append(fb.Filters, filter)
+			}
+		}
 
 		if (external != FBCfgExternal{} && fb.ExternalCfg != FBCfgExternal{}) {
 			cfgLogger.Warn("External Fluent Bit configuration specified more than once. Only first one is considered, please remove any duplicates from the configuration.")
@@ -301,7 +311,7 @@ func NewFBConf(loggingCfgs LogsCfg, logFwdCfg *config.LogForward, entityGUID, ho
 	}
 
 	// This record_modifier FILTER adds common attributes for all the log records
-	fb.Filters = append(fb.Filters, FBCfgFilter{
+	commonFilter := FBCfgFilter{
 		Name:  fbFilterTypeRecordModifier,
 		Match: "*",
 		Records: map[string]string{
@@ -309,7 +319,18 @@ func NewFBConf(loggingCfgs LogsCfg, logFwdCfg *config.LogForward, entityGUID, ho
 			rAttPluginType: logRecordModifierSource,
 			rAttHostname:   hostname,
 		},
-	})
+	}
+	// Add custom_attributes if provided
+	if logFwdCfg.CommonAttributes != nil {
+		for key, value := range logFwdCfg.CommonAttributes {
+			if !isReserved(key) {
+				commonFilter.Records[key] = fmt.Sprintf("%v", value)
+			} else {
+				cfgLogger.WithField("attribute", key).Warn("attribute name is a reserved keyword and will be ignored, please use a different name")
+			}
+		}
+	}
+	fb.Filters = append(fb.Filters, commonFilter)
 
 	// Newrelic OUTPUT plugin will send all the collected logs to Vortex
 	fb.Output = newNROutput(logFwdCfg)