[beatreceivers] Add support for kerberos (#11163)

* Support loadbalance:false option

Author: khushijain21

internal/pkg/otel/translate/otelconfig.go

@@ -14,6 +14,7 @@ import (
 	"github.com/go-viper/mapstructure/v2"
 	koanfmaps "github.com/knadh/koanf/maps"
 
+	"github.com/elastic/elastic-agent-client/v7/pkg/client"
 	"github.com/elastic/elastic-agent/internal/pkg/agent/application/monitoring/monitoringhelpers"
 
 	"github.com/elastic/elastic-agent-libs/logp"
@@ -25,7 +26,7 @@ import (
 
 	"github.com/elastic/beats/v7/libbeat/outputs/elasticsearch"
 	"github.com/elastic/beats/v7/x-pack/libbeat/management"
-	"github.com/elastic/elastic-agent-client/v7/pkg/client"
+	"github.com/elastic/beats/v7/x-pack/otel/extension/beatsauthextension"
 	"github.com/elastic/elastic-agent-libs/config"
 	"github.com/elastic/elastic-agent/internal/pkg/agent/application/info"
 	"github.com/elastic/elastic-agent/internal/pkg/agent/application/paths"
@@ -649,15 +650,18 @@ func BeatDataPath(componentId string) string {
 // getBeatsAuthExtensionConfig sets http transport settings on beatsauth
 // currently this is only supported for elasticsearch output
 func getBeatsAuthExtensionConfig(outputCfg *config.C) (map[string]any, error) {
-	defaultTransportSettings := elasticsearch.ESDefaultTransportSettings()
+
+	authSettings := beatsauthextension.BeatsAuthConfig{
+		Transport: elasticsearch.ESDefaultTransportSettings(),
+	}
 
 	var resultMap map[string]any
 	if err := outputCfg.Unpack(&resultMap); err != nil {
 		return nil, err
 	}
 
 	decoder, err := mapstructure.NewDecoder(&mapstructure.DecoderConfig{
-		Result:          &defaultTransportSettings,
+		Result:          &authSettings,
 		TagName:         "config",
 		SquashTagOption: "inline",
 		DecodeHook:      cfgDecodeHookFunc(),
@@ -670,20 +674,27 @@ func getBeatsAuthExtensionConfig(outputCfg *config.C) (map[string]any, error) {
 		return nil, err
 	}
 
-	newConfig, err := config.NewConfigFrom(defaultTransportSettings)
+	newConfig, err := config.NewConfigFrom(authSettings)
 	if err != nil {
 		return nil, err
 	}
 
 	// proxy_url on newConfig is of type url.URL. Beatsauth extension expects it to be of string type instead
 	// this logic here converts url.URL to string type similar to what a user would set on filebeat config
-	if defaultTransportSettings.Proxy.URL != nil {
-		err = newConfig.SetString("proxy_url", -1, defaultTransportSettings.Proxy.URL.String())
+	if authSettings.Transport.Proxy.URL != nil {
+		err = newConfig.SetString("proxy_url", -1, authSettings.Transport.Proxy.URL.String())
 		if err != nil {
 			return nil, fmt.Errorf("error settingg proxy url:%w ", err)
 		}
 	}
 
+	if authSettings.Kerberos != nil {
+		err = newConfig.SetString("kerberos.auth_type", -1, authSettings.Kerberos.AuthType.String())
+		if err != nil {
+			return nil, fmt.Errorf("error setting kerberos auth type url:%w ", err)
+		}
+	}
+
 	var newMap map[string]any
 	err = newConfig.Unpack(&newMap)
 	if err != nil {

internal/pkg/otel/translate/otelconfig_test.go

@@ -1302,10 +1302,10 @@ func TestGetBeatsAuthExtensionConfig(t *testing.T) {
 			},
 		},
 		{
-			name: "with ssl enabled and verification_mode full",
+			name: "with ssl enabled and verification_mode certificate",
 			outputCfg: map[string]any{
 				"ssl.enabled":           true,
-				"ssl.verification_mode": "full",
+				"ssl.verification_mode": "certificate",
 			},
 			expected: map[string]any{
 				"continue_on_error":       true,
@@ -1324,11 +1324,41 @@ func TestGetBeatsAuthExtensionConfig(t *testing.T) {
 					"key_passphrase_path":     "",
 					"renegotiation":           int64(0),
 					"supported_protocols":     []interface{}{},
-					"verification_mode":       uint64(0),
+					"verification_mode":       uint64(2),
 				},
 				"timeout": "1m30s",
 			},
 		},
+		{
+			name: "with kerberos is enabled",
+			outputCfg: map[string]any{
+				"kerberos": map[string]any{
+					"enabled":     true,
+					"auth_type":   "password",
+					"config_path": "temp/krb5.conf",
+					"username":    "beats",
+					"password":    "testing",
+					"realm":       "elastic",
+				},
+			},
+			expected: map[string]any{
+				"continue_on_error":       true,
+				"idle_connection_timeout": "3s",
+				"timeout":                 "1m30s",
+				"kerberos": map[string]any{
+					"enabled":          true,
+					"auth_type":        "password",
+					"config_path":      "temp/krb5.conf",
+					"username":         "beats",
+					"password":         "testing",
+					"realm":            "elastic",
+					"enable_krb5_fast": false,
+					"service_name":     "",
+					"keytab":           "",
+				},
+				"proxy_disable": false,
+			},
+		},
 	}
 
 	for _, tt := range tests {

internal/pkg/otel/translate/output_elasticsearch.go

@@ -17,6 +17,7 @@ import (
 	"github.com/go-viper/mapstructure/v2"
 
 	"github.com/elastic/beats/v7/libbeat/common"
+	"github.com/elastic/beats/v7/libbeat/common/transport/kerberos"
 	"github.com/elastic/beats/v7/libbeat/outputs"
 	"github.com/elastic/beats/v7/libbeat/outputs/elasticsearch"
 	"github.com/elastic/beats/v7/libbeat/publisher/queue/memqueue"
@@ -196,8 +197,6 @@ func checkUnsupportedConfig(cfg *config.C) error {
 		return fmt.Errorf("ladbalance:false is currently not supported: %w", errors.ErrUnsupported)
 	} else if cfg.HasField("non_indexable_policy") {
 		return fmt.Errorf("non_indexable_policy is currently not supported: %w", errors.ErrUnsupported)
-	} else if cfg.HasField("kerberos") {
-		return fmt.Errorf("kerberos is currently not supported: %w", errors.ErrUnsupported)
 	}
 
 	return nil
@@ -273,6 +272,12 @@ func cfgDecodeHookFunc() mapstructure.DecodeHookFunc {
 				return nil, fmt.Errorf("failed parsing proxy_url: %w", err)
 			}
 			return proxyURL, nil
+		case t == reflect.TypeOf(kerberos.AuthType(0)):
+			var authType kerberos.AuthType
+			if err := authType.Unpack(data.(string)); err != nil {
+				return nil, fmt.Errorf("failed parsing kerberos.auth_type: %w", err)
+			}
+			return authType, nil
 		default:
 			return data, nil
 		}