feat: Add Chronicle exporter

exporter/chronicleexporter/README.md

@@ -0,0 +1,47 @@
+# Chronicle Exporter
+
+This exporter facilitates the sending of logs to Chronicle, which is a security analytics platform provided by Google. It is designed to integrate with OpenTelemetry collectors to export telemetry data such as logs to a Chronicle account.
+
+## Minimum Collector Versions
+
+- Introduced: [v1.39.0](https://github.com/observIQ/bindplane-agent/releases/tag/v1.39.0)
+
+## Supported Pipelines
+
+- Logs
+
+## How It Works
+
+1. The exporter uses the configured credentials to authenticate with the Google Cloud services.
+2. It marshals logs into the format expected by Chronicle.
+3. It sends the logs to the appropriate regional Chronicle endpoint.
+
+## Configuration
+
+The exporter can be configured using the following fields:
+
+| Field              | Type   | Default | Required | Description                                                                                                                                                           |
+| ------------------ | ------ | ------- | -------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
+| `region`           | string |         | `false`  | The region where the data will be sent, it must be one of the predefined regions. if no region is specfied defaults to `https://malachiteingestion-pa.googleapis.com` |
+| `creds_file_path`  | string |         | `true`   | The file path to the Google credentials JSON file.                                                                                                                    |
+| `creds`            | string |         | `true`   | The Google credentials JSON.                                                                                                                                          |
+| `log_type`         | string |         | `true`   | The type of log that will be sent.                                                                                                                                    |
+| `raw_log_field`    | string |         | `false`  | The field name for raw logs.                                                                                                                                          |
+| `customer_id`      | string |         | `false`  | The customer ID used for sending logs.                                                                                                                                |
+| `sending_queue`    | struct |         | `false`  | Configuration for the sending queue.                                                                                                                                  |
+| `retry_on_failure` | struct |         | `false`  | Configuration for retry logic on failure.                                                                                                                             |
+| `timeout_settings` | struct |         | `false`  | Configuration for timeout settings.                                                                                                                                   |
+
+### Regions
+
+Predefined regions include multiple global locations such as `Europe Multi-Region`, `Frankfurt`, `London`, `Singapore`, `Sydney`, `Tel Aviv`, `United States Multi-Region`, and `Zurich`. Each region has a specific endpoint URL.
+
+## Example Configuration
+
+```yaml
+chronicle:
+  region: "Europe Multi-Region"
+  creds_file_path: "/path/to/google/creds.json"
+  log_type: "threat_detection"
+  customer_id: "customer-123"
+```

exporter/chronicleexporter/config.go

@@ -0,0 +1,91 @@
+// Copyright observIQ, Inc.
+//
+// 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 chronicleexporter
+
+import (
+	"errors"
+	"fmt"
+
+	"github.com/observiq/bindplane-agent/expr"
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/exporter/exporterhelper"
+	"go.uber.org/zap"
+)
+
+// Alternative regional endpoints for Chronicle.
+// https://cloud.google.com/chronicle/docs/reference/search-api#regional_endpoints
+var regions = map[string]string{
+	"Europe Multi-Region":        "https://europe-backstory.googleapis.com",
+	"Frankfurt":                  "https://europe-west3-backstory.googleapis.com",
+	"London":                     "http://europe-west2-backstory.googleapis.com",
+	"Singapore":                  "https://asia-southeast1-backstory.googleapis.com",
+	"Sydney":                     "https://australia-southeast1-backstory.googleapis.com",
+	"Tel Aviv":                   "https://me-west1-backstory.googleapis.com",
+	"United States Multi-Region": "https://united-states-backstory.googleapis.com",
+	"Zurich":                     "https://europe-west6-backstory.googleapis.com",
+}
+
+// Config defines configuration for the Chronicle exporter.
+type Config struct {
+	exporterhelper.TimeoutSettings `mapstructure:",squash"` // squash ensures fields are correctly decoded in embedded struct.
+	exporterhelper.QueueSettings   `mapstructure:"sending_queue"`
+	exporterhelper.RetrySettings   `mapstructure:"retry_on_failure"`
+
+	// Endpoint is the URL where Chronicle data will be sent.
+	Region string `mapstructure:"region"`
+
+	// CredsFilePath is the file path to the Google credentials JSON file.
+	CredsFilePath string `mapstructure:"creds_file_path"`
+
+	// Creds are the Google credentials JSON file.
+	Creds string `mapstructure:"creds"`
+
+	// LogType is the type of log that will be sent to Chronicle.
+	LogType string `mapstructure:"log_type"`
+
+	// RawLogField is the field name that will be used to send raw logs to Chronicle.
+	RawLogField string `mapstructure:"raw_log_field"`
+
+	// CustomerID is the customer ID that will be used to send logs to Chronicle.
+	CustomerID string `mapstructure:"customer_id"`
+}
+
+// Validate checks if the configuration is valid.
+func (cfg *Config) Validate() error {
+	if cfg.CredsFilePath != "" && cfg.Creds != "" {
+		return errors.New("can only specify creds_file_path or creds")
+	}
+
+	if cfg.LogType == "" {
+		return errors.New("log_type is required")
+	}
+
+	if cfg.Region != "" {
+		if _, ok := regions[cfg.Region]; !ok {
+			return errors.New("region is invalid")
+		}
+	}
+
+	if cfg.RawLogField != "" {
+		_, err := expr.NewOTTLLogRecordExpression(cfg.RawLogField, component.TelemetrySettings{
+			Logger: zap.NewNop(),
+		})
+		if err != nil {
+			return fmt.Errorf("raw_log_field is invalid: %s", err)
+		}
+	}
+
+	return nil
+}

exporter/chronicleexporter/config_test.go

@@ -0,0 +1,97 @@
+// Copyright observIQ, Inc.
+//
+// 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 chronicleexporter
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestConfigValidate(t *testing.T) {
+	testCases := []struct {
+		desc        string
+		config      *Config
+		expectedErr string
+	}{
+		{
+			desc: "Both creds_file_path and creds are set",
+			config: &Config{
+				CredsFilePath: "/path/to/creds_file",
+				Creds:         "creds_example",
+				Region:        "United States Multi-Region",
+				LogType:       "log_type_example",
+			},
+			expectedErr: "can only specify creds_file_path or creds",
+		},
+		{
+			desc: "LogType is empty",
+			config: &Config{
+				Region: "United States Multi-Region",
+				Creds:  "creds_example",
+			},
+			expectedErr: "log_type is required",
+		},
+		{
+			desc: "Region is invalid",
+			config: &Config{
+				Region:  "Invalid Region",
+				Creds:   "creds_example",
+				LogType: "log_type_example",
+			},
+			expectedErr: "region is invalid",
+		},
+		{
+			desc: "Valid config with creds",
+			config: &Config{
+				Region:  "United States Multi-Region",
+				Creds:   "creds_example",
+				LogType: "log_type_example",
+			},
+			expectedErr: "",
+		},
+		{
+			desc: "Valid config with creds_file_path",
+			config: &Config{
+				Region:        "United States Multi-Region",
+				CredsFilePath: "/path/to/creds_file",
+				LogType:       "log_type_example",
+			},
+			expectedErr: "",
+		},
+		{
+			desc: "Valid config with raw log field",
+			config: &Config{
+				Region:        "United States Multi-Region",
+				CredsFilePath: "/path/to/creds_file",
+				LogType:       "log_type_example",
+				RawLogField:   `body["field"]`,
+			},
+			expectedErr: "",
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.desc, func(t *testing.T) {
+			err := tc.config.Validate()
+			if tc.expectedErr == "" {
+				require.NoError(t, err)
+			} else {
+				require.Error(t, err)
+				require.Contains(t, err.Error(), tc.expectedErr)
+			}
+		})
+	}
+}

exporter/chronicleexporter/doc.go

@@ -0,0 +1,18 @@
+// Copyright observIQ, Inc.
+//
+// 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.
+
+//go:generate mdatagen metadata.yaml
+
+// Package chronicleexporter exports OpenTelemetry data to Chronicle.
+package chronicleexporter // import "github.com/observiq/bindplane-agent/exporter/azureblobexporter"

exporter/chronicleexporter/exporter.go

@@ -0,0 +1,133 @@
+// Copyright observIQ, Inc.
+//
+// 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 chronicleexporter
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"io"
+	"net/http"
+	"os"
+
+	"go.opentelemetry.io/collector/consumer"
+	"go.opentelemetry.io/collector/exporter"
+	"go.opentelemetry.io/collector/pdata/plog"
+	"go.uber.org/zap"
+	"golang.org/x/oauth2"
+	"golang.org/x/oauth2/google"
+)
+
+const scope = "https://www.googleapis.com/auth/malachite-ingestion"
+
+const baseEndpoint = "https://malachiteingestion-pa.googleapis.com"
+
+const apiTarget = "/v2/unstructuredlogentries:batchCreate"
+
+type chronicleExporter struct {
+	cfg        *Config
+	logger     *zap.Logger
+	httpClient *http.Client
+	marshaler  logMarshaler
+	endpoint   string
+}
+
+func newExporter(cfg *Config, params exporter.CreateSettings) (*chronicleExporter, error) {
+	var creds *google.Credentials
+	var err error
+
+	switch {
+	case cfg.Creds != "":
+		creds, err = google.CredentialsFromJSON(context.Background(), []byte(cfg.CredsFilePath), scope)
+		if err != nil {
+			return nil, fmt.Errorf("obtain credentials from JSON: %w", err)
+		}
+	case cfg.CredsFilePath != "":
+		credsData, err := os.ReadFile(cfg.CredsFilePath)
+		if err != nil {
+			return nil, fmt.Errorf("read credentials file: %w", err)
+		}
+
+		creds, err = google.CredentialsFromJSON(context.Background(), credsData, scope)
+		if err != nil {
+			return nil, fmt.Errorf("obtain credentials from JSON: %w", err)
+		}
+	default:
+		creds, err = google.FindDefaultCredentials(context.Background(), scope)
+		if err != nil {
+			return nil, fmt.Errorf("find default credentials: %w", err)
+		}
+	}
+
+	// Use the credentials to create an HTTP client
+	httpClient := oauth2.NewClient(context.Background(), creds.TokenSource)
+
+	return &chronicleExporter{
+		endpoint:   buildEndpoint(cfg),
+		cfg:        cfg,
+		logger:     params.Logger,
+		httpClient: httpClient,
+		marshaler:  newMarshaler(*cfg, params.TelemetrySettings),
+	}, nil
+}
+
+// buildEndpoint builds the endpoint to send logs to based on the region. there is a default endpoint `https://malachiteingestion-pa.googleapis.com`
+// but there are also regional endpoints that can be used instead. the regional endpoints are listed here: https://cloud.google.com/chronicle/docs/reference/search-api#regional_endpoints
+func buildEndpoint(cfg *Config) string {
+	if cfg.Region != "" && regions[cfg.Region] != "" {
+		return fmt.Sprintf("%s%s", regions[cfg.Region], apiTarget)
+	}
+	return fmt.Sprintf("%s%s", baseEndpoint, apiTarget)
+}
+
+func (ce *chronicleExporter) Capabilities() consumer.Capabilities {
+	return consumer.Capabilities{MutatesData: false}
+}
+
+func (ce *chronicleExporter) logsDataPusher(ctx context.Context, ld plog.Logs) error {
+	udmData, err := ce.marshaler.MarshalRawLogs(ctx, ld)
+	if err != nil {
+		return fmt.Errorf("marshal logs: %w", err)
+	}
+
+	return ce.uploadToChronicle(ctx, udmData)
+}
+
+func (ce *chronicleExporter) uploadToChronicle(ctx context.Context, data []byte) error {
+	request, err := http.NewRequestWithContext(ctx, "POST", ce.endpoint, bytes.NewBuffer(data))
+	if err != nil {
+		return fmt.Errorf("create request: %w", err)
+	}
+
+	request.Header.Set("Content-Type", "application/json")
+
+	resp, err := ce.httpClient.Do(request)
+	if err != nil {
+		return fmt.Errorf("send request to Chronicle: %w", err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		respBody, err := io.ReadAll(resp.Body)
+		if err != nil {
+			ce.logger.Warn("Failed to read response body", zap.Error(err))
+		} else {
+			ce.logger.Warn("Received non-OK response from Chronicle", zap.String("status", resp.Status), zap.ByteString("body", respBody))
+		}
+		return fmt.Errorf("received non-OK response from Chronicle: %s", resp.Status)
+	}
+
+	return nil
+}