[receiver/mongodbatlasreceiver] Add Log collection (#12800)

Added functionality for a logs receiver to collect logs from project clusters specified by the user.
Along with cluster logs, audit logs from those clusters are also scraped.
New Log and Audit log structures added in to the internal package logs.go
Log conversion from new custom log structures into plog.Logs

internal/components/receivers_test.go

@@ -23,7 +23,9 @@ import (
 	"errors"
 	"path/filepath"
 	"runtime"
+	"strconv"
 	"testing"
+	"time"
 
 	promconfig "github.com/prometheus/prometheus/config"
 	"github.com/stretchr/testify/assert"
@@ -38,6 +40,7 @@ import (
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/carbonreceiver"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/chronyreceiver"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/filelogreceiver"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/mongodbatlasreceiver"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/otlpjsonfilereceiver"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/prometheusreceiver"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/syslogreceiver"
@@ -198,6 +201,13 @@ func TestDefaultReceivers(t *testing.T) {
 		},
 		{
 			receiver: "mongodbatlas",
+			// MongoDB Atlas needs unique config IDs
+			getConfigFn: func() config.Receiver {
+				cfg := rcvrFactories["mongodbatlas"].CreateDefaultConfig().(*mongodbatlasreceiver.Config)
+				cfg.SetIDName(strconv.Itoa(int(time.Now().UnixNano())))
+				cfg.Logs.Enabled = true
+				return cfg
+			},
 		},
 		{
 			receiver: "mysql",

receiver/mongodbatlasreceiver/README.md

@@ -13,10 +13,14 @@ as well as alerts via a configured [webhook](https://www.mongodb.com/docs/atlas/
 ## Getting Started
 
 The MongoDB Atlas receiver takes the following parameters. `public_key` and 
-`private_key` are the only two required values to receive metrics and are obtained via the 
+`private_key` are the only two required values to receive metrics and logs and are obtained via the 
 "API Keys" tab of the MongoDB Atlas Project Access Manager. In the example
 below both values are being pulled from the environment.
 
+In order to collect logs, at least one project must be specified. By default, logs for all clusters within a project will be collected. Clusters can be limited using either the `include_clusters` or `exclude_clusters` setting.
+
+MongoDB Atlas [Documentation](https://www.mongodb.com/docs/atlas/reference/api/logs/#logs) recommends a polling interval of 5 minutes. 
+
 - `public_key` (required for metrics)
 - `private_key` (required for metrics)
 - `granularity` (default `PT1M` - See [MongoDB Atlas Documentation](https://docs.atlas.mongodb.com/reference/api/process-measurements/))
@@ -32,6 +36,14 @@ below both values are being pulled from the environment.
   - `tls`
     - `key_file`
     - `cert_file`
+- `logs`
+  - `enabled` (default false)
+  - `projects` (required if enabled)
+    - `name` (required if enabled)
+    - `collect_audit_logs` (default false)
+    - `include_clusters` (default empty)
+    - `exclude_clusters` (default empty)
+
 
 Examples:
 
@@ -53,5 +65,16 @@ receivers:
       endpoint: "0.0.0.0:7706"
 ```
 
+Receive logs:
+```yaml
+receivers:
+  mongodbatlas:
+    logs:
+      enabled: true
+      projects: 
+        - name: "project 1"
+          collect_audit_logs: true
+```
+
 [beta]:https://github.com/open-telemetry/opentelemetry-collector#beta
 [contrib]:https://github.com/open-telemetry/opentelemetry-collector-releases/tree/main/distributions/otelcol-contrib

receiver/mongodbatlasreceiver/combined_logs.go

@@ -0,0 +1,66 @@
+// Copyright  The OpenTelemetry Authors
+//
+// 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.
+
+package mongodbatlasreceiver // import "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/mongodbatlasreceiver"
+
+import (
+	"context"
+
+	"github.com/hashicorp/go-multierror"
+	"go.opentelemetry.io/collector/component"
+)
+
+// combindedLogsReceiver wraps alerts and log receivers in a single log receiver to be consumed by the factory
+type combindedLogsReceiver struct {
+	alerts *alertsReceiver
+	logs   *logsReceiver
+}
+
+// Starts up the combined MongoDB Atlas Logs and Alert Receiver
+func (c *combindedLogsReceiver) Start(ctx context.Context, host component.Host) error {
+	var errs error
+
+	if c.alerts != nil {
+		if err := c.alerts.Start(ctx, host); err != nil {
+			errs = multierror.Append(errs, err)
+		}
+	}
+
+	if c.logs != nil {
+		if err := c.logs.Start(ctx, host); err != nil {
+			errs = multierror.Append(errs, err)
+		}
+	}
+
+	return errs
+}
+
+// Shutsdown the combined MongoDB Atlas Logs and Alert Receiver
+func (c *combindedLogsReceiver) Shutdown(ctx context.Context) error {
+	var errs error
+
+	if c.alerts != nil {
+		if err := c.alerts.Shutdown(ctx); err != nil {
+			errs = multierror.Append(errs, err)
+		}
+	}
+
+	if c.logs != nil {
+		if err := c.logs.Shutdown(ctx); err != nil {
+			errs = multierror.Append(errs, err)
+		}
+	}
+
+	return errs
+}

receiver/mongodbatlasreceiver/config.go

@@ -37,6 +37,7 @@ type Config struct {
 	Granularity                             string                       `mapstructure:"granularity"`
 	Metrics                                 metadata.MetricsSettings     `mapstructure:"metrics"`
 	Alerts                                  AlertConfig                  `mapstructure:"alerts"`
+	Logs                                    LogConfig                    `mapstructure:"logs"`
 	RetrySettings                           exporterhelper.RetrySettings `mapstructure:"retry_on_failure"`
 }
 
@@ -47,18 +48,55 @@ type AlertConfig struct {
 	TLS      *configtls.TLSServerSetting `mapstructure:"tls"`
 }
 
+type LogConfig struct {
+	Enabled  bool             `mapstructure:"enabled"`
+	Projects []*ProjectConfig `mapstructure:"projects"`
+}
+
+type ProjectConfig struct {
+	Name            string   `mapstructure:"name"`
+	ExcludeClusters []string `mapstructure:"exclude_clusters"`
+	IncludeClusters []string `mapstructure:"include_clusters"`
+	EnableAuditLogs bool     `mapstructure:"collect_audit_logs"`
+}
+
 var (
+	// Alerts Receiver Errors
 	errNoEndpoint = errors.New("an endpoint must be specified")
 	errNoSecret   = errors.New("a webhook secret must be specified")
 	errNoCert     = errors.New("tls was configured, but no cert file was specified")
 	errNoKey      = errors.New("tls was configured, but no key file was specified")
+
+	// Logs Receiver Errors
+	errNoProjects    = errors.New("at least one 'project' must be specified")
+	errClusterConfig = errors.New("only one of 'include_clusters' or 'exclude_clusters' may be specified")
 )
 
 func (c *Config) Validate() error {
 	var errs error
 
 	errs = multierr.Append(errs, c.ScraperControllerSettings.Validate())
 	errs = multierr.Append(errs, c.Alerts.validate())
+	errs = multierr.Append(errs, c.Logs.validate())
+
+	return errs
+}
+
+func (l *LogConfig) validate() error {
+	if !l.Enabled {
+		return nil
+	}
+
+	var errs error
+	if len(l.Projects) == 0 {
+		errs = multierr.Append(errs, errNoProjects)
+	}
+
+	for _, project := range l.Projects {
+		if len(project.ExcludeClusters) != 0 && len(project.IncludeClusters) != 0 {
+			errs = multierr.Append(errs, errClusterConfig)
+		}
+	}
 
 	return errs
 }

receiver/mongodbatlasreceiver/config_test.go

@@ -104,6 +104,46 @@ func TestValidate(t *testing.T) {
 			},
 			expectedErr: errNoCert.Error(),
 		},
+		{
+			name: "Valid Logs Config",
+			input: Config{
+				Logs: LogConfig{
+					Enabled: true,
+					Projects: []*ProjectConfig{
+						{
+							Name:            "Project1",
+							EnableAuditLogs: false,
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "Invalid Logs Config",
+			input: Config{
+				Logs: LogConfig{
+					Enabled: true,
+				},
+			},
+			expectedErr: errNoProjects.Error(),
+		},
+		{
+			name: "Invalid ProjectConfig",
+			input: Config{
+				Logs: LogConfig{
+					Enabled: true,
+					Projects: []*ProjectConfig{
+						{
+							Name:            "Project1",
+							EnableAuditLogs: false,
+							ExcludeClusters: []string{"cluster1"},
+							IncludeClusters: []string{"cluster2"},
+						},
+					},
+				},
+			},
+			expectedErr: errClusterConfig.Error(),
+		},
 	}
 
 	for _, tc := range testCases {

receiver/mongodbatlasreceiver/factory.go

@@ -16,6 +16,7 @@ package mongodbatlasreceiver // import "github.com/open-telemetry/opentelemetry-
 
 import (
 	"context"
+	"errors"
 	"fmt"
 
 	"go.opentelemetry.io/collector/component"
@@ -32,6 +33,7 @@ const (
 	stability            = component.StabilityLevelBeta
 	defaultGranularity   = "PT1M" // 1-minute, as per https://docs.atlas.mongodb.com/reference/api/process-measurements/
 	defaultAlertsEnabled = false
+	defaultLogsEnabled   = false
 )
 
 // NewFactory creates a factory for MongoDB Atlas receiver
@@ -40,7 +42,8 @@ func NewFactory() component.ReceiverFactory {
 		typeStr,
 		createDefaultConfig,
 		component.WithMetricsReceiver(createMetricsReceiver, stability),
-		component.WithLogsReceiver(createLogsReceiver, stability))
+		component.WithLogsReceiver(createCombinedLogReceiver, stability))
+
 }
 
 func createMetricsReceiver(
@@ -50,24 +53,46 @@ func createMetricsReceiver(
 	consumer consumer.Metrics,
 ) (component.MetricsReceiver, error) {
 	cfg := rConf.(*Config)
-	ms, err := newMongoDBAtlasScraper(params, cfg)
+	recv, err := newMongoDBAtlasReceiver(params, cfg)
 	if err != nil {
 		return nil, fmt.Errorf("unable to create a MongoDB Atlas Receiver instance: %w", err)
 	}
 
+	ms, err := newMongoDBAtlasScraper(recv)
+	if err != nil {
+		return nil, fmt.Errorf("unable to create a MongoDB Atlas Scaper instance: %w", err)
+	}
+
 	return scraperhelper.NewScraperControllerReceiver(&cfg.ScraperControllerSettings, params, consumer, scraperhelper.AddScraper(ms))
 }
 
-func createLogsReceiver(
+func createCombinedLogReceiver(
 	_ context.Context,
 	params component.ReceiverCreateSettings,
 	rConf config.Receiver,
 	consumer consumer.Logs,
 ) (component.LogsReceiver, error) {
 	cfg := rConf.(*Config)
-	recv, err := newAlertsReceiver(params.Logger, cfg.Alerts, consumer)
-	if err != nil {
-		return nil, fmt.Errorf("unable to create a MongoDB Atlas Receiver instance: %w", err)
+
+	if !cfg.Alerts.Enabled && !cfg.Logs.Enabled {
+		return nil, errors.New("one of 'alerts' or 'logs' must be enabled")
+	}
+
+	var err error
+	recv := &combindedLogsReceiver{}
+
+	if cfg.Alerts.Enabled {
+		recv.alerts, err = newAlertsReceiver(params.Logger, cfg.Alerts, consumer)
+		if err != nil {
+			return nil, fmt.Errorf("unable to create a MongoDB Atlas Alerts Receiver instance: %w", err)
+		}
+	}
+
+	if cfg.Logs.Enabled {
+		recv.logs, err = newMongoDBAtlasLogsReceiver(params, cfg, consumer)
+		if err != nil {
+			return nil, fmt.Errorf("unable to create a MongoDB Atlas Logs Receiver instance: %w", err)
+		}
 	}
 
 	return recv, nil
@@ -82,5 +107,9 @@ func createDefaultConfig() config.Receiver {
 		Alerts: AlertConfig{
 			Enabled: defaultAlertsEnabled,
 		},
+		Logs: LogConfig{
+			Enabled:  defaultLogsEnabled,
+			Projects: []*ProjectConfig{},
+		},
 	}
 }