[Metricbeat][Aerospike Module] Add support for TLS

metricbeat/module/aerospike/_meta/config.yml

@@ -3,3 +3,15 @@
   #  - namespace
   period: 10s
   hosts: ["localhost:3000"]
+
+  # Username of hosts. Empty by default.
+  # username: root
+
+  # Password of hosts. Empty by default.
+  # password: secret
+
+  # Aerospike Cluster Name
+  # cluster_name: myclustername
+
+  # Read more about SSL Configurations https://www.elastic.co/guide/en/beats/metricbeat/current/configuration-ssl.html
+  # ssl.enabled: false

metricbeat/module/aerospike/aerospike.go

@@ -22,9 +22,47 @@ import (
 	"strconv"
 	"strings"
 
+	"github.com/elastic/elastic-agent-libs/transport/tlscommon"
+
 	as "github.com/aerospike/aerospike-client-go"
 )
 
+type Config struct {
+	ClusterName *string           `config:"cluster_name"`
+	User        *string           `config:"username"`
+	Password    *string           `config:"password"`
+	TLS         *tlscommon.Config `config:"ssl"`
+}
+
+// DefaultConfig return default config for the aerospike module.
+func DefaultConfig() Config {
+	return Config{}
+}
+
+func ParseClientPolicy(config Config) (*as.ClientPolicy, error) {
+	clientPolicy := as.NewClientPolicy()
+	if config.TLS.IsEnabled() {
+		tlsconfig, err := tlscommon.LoadTLSConfig(config.TLS)
+		if err != nil {
+			return nil, fmt.Errorf("could not initialize TLS configurations %w", err)
+		}
+		clientPolicy.TlsConfig = tlsconfig.ToConfig()
+	}
+	if config.User != nil && config.Password != nil {
+		clientPolicy.User = *config.User
+		clientPolicy.Password = *config.Password
+	} else if config.User != nil && config.Password == nil {
+		return nil, fmt.Errorf("if username is set, password should be set too")
+	} else if config.User == nil && config.Password != nil {
+		return nil, fmt.Errorf("if password is set, username should be set too")
+	}
+
+	if config.ClusterName != nil {
+		clientPolicy.ClusterName = *config.ClusterName
+	}
+	return clientPolicy, nil
+}
+
 func ParseHost(host string) (*as.Host, error) {
 	pieces := strings.Split(host, ":")
 	if len(pieces) != 2 {

metricbeat/module/aerospike/aerospike_test.go

@@ -19,10 +19,13 @@ package aerospike
 
 import (
 	"errors"
+	"fmt"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
 
+	"github.com/elastic/elastic-agent-libs/transport/tlscommon"
+
 	as "github.com/aerospike/aerospike-client-go"
 )
 
@@ -96,3 +99,103 @@ func TestParseInfo(t *testing.T) {
 		assert.Equal(t, test.expected, result, test.Name)
 	}
 }
+
+func getStringPointer(value string) *string {
+	return &value
+}
+
+func getBoolPointer(value bool) *bool {
+	return &value
+}
+
+func TestParseClientPolicy(t *testing.T) {
+	sampleUser := "Test"
+	samplePassword := "MySecretPassword"
+	sampleClusterName := "TestCluster"
+
+	UserPasswordClientPolicy := as.NewClientPolicy()
+	UserPasswordClientPolicy.User = sampleUser
+	UserPasswordClientPolicy.Password = samplePassword
+
+	TLSPolicy := as.NewClientPolicy()
+	tlsconfig, _ := tlscommon.LoadTLSConfig(&tlscommon.Config{Enabled: getBoolPointer(true)})
+	TLSPolicy.TlsConfig = tlsconfig.ToConfig()
+
+	ClusterNamePolicy := as.NewClientPolicy()
+	ClusterNamePolicy.ClusterName = sampleClusterName
+
+	tests := []struct {
+		Name                 string
+		Config               Config
+		expectedClientPolicy *as.ClientPolicy
+		expectedErr          error
+	}{
+		{
+			Name:                 "No configurations lead to default policy",
+			Config:               Config{},
+			expectedClientPolicy: as.NewClientPolicy(),
+			expectedErr:          nil,
+		},
+		{
+			Name: "Username and password are honored",
+			Config: Config{
+				User:     getStringPointer(sampleUser),
+				Password: getStringPointer(samplePassword),
+			},
+			expectedClientPolicy: UserPasswordClientPolicy,
+			expectedErr:          nil,
+		},
+		{
+			Name: "Username is set but Password is not",
+			Config: Config{
+				User: getStringPointer(sampleUser),
+			},
+			expectedClientPolicy: UserPasswordClientPolicy,
+			expectedErr:          fmt.Errorf("if username is set, password should be set too"),
+		},
+		{
+			Name: "Password is set but Username is not",
+			Config: Config{
+				Password: getStringPointer(samplePassword),
+			},
+			expectedClientPolicy: UserPasswordClientPolicy,
+			expectedErr:          fmt.Errorf("if password is set, username should be set too"),
+		},
+		{
+			Name: "TLS Declaration",
+			Config: Config{
+				TLS: &tlscommon.Config{
+					Enabled: getBoolPointer(true),
+				},
+			},
+			expectedClientPolicy: TLSPolicy,
+			expectedErr:          nil,
+		},
+		{
+			Name: "Cluster Name Setting",
+			Config: Config{
+				ClusterName: getStringPointer(sampleClusterName),
+			},
+			expectedClientPolicy: ClusterNamePolicy,
+			expectedErr:          nil,
+		},
+	}
+
+	for _, test := range tests {
+		result, err := ParseClientPolicy(test.Config)
+		if err != nil {
+			if test.expectedErr != nil {
+				assert.Equal(t, test.expectedErr.Error(), err.Error())
+				continue
+			}
+			t.Error(err)
+			continue
+		}
+		assert.Equal(t, test.expectedClientPolicy.User, result.User, test.Name)
+		assert.Equal(t, test.expectedClientPolicy.Password, result.Password, test.Name)
+		assert.Equal(t, test.expectedClientPolicy.ClusterName, result.ClusterName, test.Name)
+		if test.Config.TLS.IsEnabled() {
+			assert.NotNil(t, result.TlsConfig)
+		}
+	}
+}

metricbeat/module/aerospike/namespace/namespace.go

@@ -42,15 +42,16 @@ func init() {
 // multiple fetch calls.
 type MetricSet struct {
 	mb.BaseMetricSet
-	host   *as.Host
-	client *as.Client
+	host         *as.Host
+	clientPolicy *as.ClientPolicy
+	client       *as.Client
 }
 
 // New create a new instance of the MetricSet
 // Part of new is also setting up the configuration by processing additional
 // configuration entries if needed.
 func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
-	config := struct{}{}
+	config := aerospike.DefaultConfig()
 	if err := base.Module().UnpackConfig(&config); err != nil {
 		return nil, err
 	}
@@ -60,9 +61,15 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 		return nil, fmt.Errorf("Invalid host format, expected hostname:port: %w", err)
 	}
 
+	clientPolicy, err := aerospike.ParseClientPolicy(config)
+	if err != nil {
+		return nil, fmt.Errorf("could not initialize aerospike client policy: %w", err)
+	}
+
 	return &MetricSet{
 		BaseMetricSet: base,
 		host:          host,
+		clientPolicy:  clientPolicy,
 	}, nil
 }
 
@@ -105,7 +112,7 @@ func (m *MetricSet) Fetch(reporter mb.ReporterV2) error {
 // create an aerospike client if it doesn't exist yet
 func (m *MetricSet) connect() error {
 	if m.client == nil {
-		client, err := as.NewClientWithPolicyAndHost(as.NewClientPolicy(), m.host)
+		client, err := as.NewClientWithPolicyAndHost(m.clientPolicy, m.host)
 		if err != nil {
 			return err
 		}

metricbeat/modules.d/aerospike.yml.disabled

@@ -6,3 +6,15 @@
   #  - namespace
   period: 10s
   hosts: ["localhost:3000"]
+
+  # Username of hosts. Empty by default.
+  # username: root
+
+  # Password of hosts. Empty by default.
+  # password: secret
+
+  # Aerospike Cluster Name
+  # cluster_name: myclustername
+
+  # Read more about SSL Configurations https://www.elastic.co/guide/en/beats/metricbeat/current/configuration-ssl.html
+  # ssl.enabled: false