SC-20509

exporter/sematextexporter/README.md

@@ -1,5 +1,14 @@
 # Sematext Exporter
 <!-- status autogenerated section -->
+| Status        |           |
+| ------------- |-----------|
+| Stability     | [development]: metrics, logs   |
+| Distributions | [contrib] |
+| Issues        | [![Open issues](https://img.shields.io/github/issues-search/open-telemetry/opentelemetry-collector-contrib?query=is%3Aissue%20is%3Aopen%20label%3Aexporter%2Fsematext%20&label=open&color=orange&logo=opentelemetry)](https://github.com/open-telemetry/opentelemetry-collector-contrib/issues?q=is%3Aopen+is%3Aissue+label%3Aexporter%2Fsematext) [![Closed issues](https://img.shields.io/github/issues-search/open-telemetry/opentelemetry-collector-contrib?query=is%3Aissue%20is%3Aclosed%20label%3Aexporter%2Fsematext%20&label=closed&color=blue&logo=opentelemetry)](https://github.com/open-telemetry/opentelemetry-collector-contrib/issues?q=is%3Aclosed+is%3Aissue+label%3Aexporter%2Fsematext) |
+| [Code Owners](https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/main/CONTRIBUTING.md#becoming-a-code-owner)    | [@AkhigbeEromo](https://www.github.com/AkhigbeEromo) |
+
+[development]: https://github.com/open-telemetry/opentelemetry-collector#development
+[contrib]: https://github.com/open-telemetry/opentelemetry-collector-releases/tree/main/distributions/otelcol-contrib
 <!-- end autogenerated section -->
 
 This exporter supports sending metrics to [Sematext Cloud](https://sematext.com/) in Influx line protocol format
@@ -15,11 +24,11 @@ The following configuration options are supported:
 * `payload_max_lines` (default = 1_000) Maximum number of lines allowed per HTTP POST request
 * `payload_max_bytes` (default = 300_000) Maximum number of bytes allowed per HTTP POST request
 * `metrics_schema` (default = telegraf-prometheus-v2) The chosen metrics schema to write
-* `sending_queue` [details here](https://github.com/open-telemetry/opentelemetry-collector/blob/v0.25.0/exporter/exporterhelper/README.md#configuration)
+* `sending_queue` 
     * `enabled` (default = true)
     * `num_consumers` (default = 10) The number of consumers from the queue
     * `queue_size` (default = 1000) Maximum number of batches allowed in queue at a given time
-* `retry_on_failure` [details here](https://github.com/open-telemetry/opentelemetry-collector/blob/v0.25.0/exporter/exporterhelper/README.md#configuration)
+* `retry_on_failure` 
     * `enabled` (default = true)
     * `initial_interval` (default = 5s) Time to wait after the first failure before retrying
     * `max_interval` (default = 30s) Upper bound on backoff interval

exporter/sematextexporter/config.go

@@ -12,6 +12,15 @@ import (
 	"go.opentelemetry.io/collector/exporter/exporterhelper"
 )
 
+const (
+	euRegion          = "eu"
+	usRegion          = "us"
+	euMetricsEndpoint = "https://spm-receiver.eu.sematext.com"
+	euLogsEndpoint    = "https://logsene-receiver.eu.sematext.com"
+	usMetricsEndpoint = "https://spm-receiver.sematext.com"
+	usLogsEndpoint    = "https://logsene-receiver.sematext.com"
+)
+
 type Config struct {
 	confighttp.ClientConfig   `mapstructure:",squash"`
 	configretry.BackOffConfig `mapstructure:"retry_on_failure"`
@@ -20,37 +29,53 @@ type Config struct {
 	// - EU
 	// - US
 	Region string `mapstructure:"region"`
+	// MetricsConfig defines the configuration specific to metrics
 	MetricsConfig `mapstructure:"metrics"`
+	// LogsConfig defines the configuration specific to logs
+	LogsConfig `mapstructure:"logs"`
 }
 
 type MetricsConfig struct {
 	// App token is the token of Sematext Monitoring App to which you want to send the metrics.
-	AppToken         string                     `mapstructure:"app_token"`
+	AppToken string `mapstructure:"app_token"`
 	// MetricsEndpoint specifies the endpoint for receiving metrics in Sematext
-	MetricsEndpoint  string                     `mapstructure:"metrics_endpoint"`
-	QueueSettings    exporterhelper.QueueConfig `mapstructure:"sending_queue"`
+	MetricsEndpoint string                     `mapstructure:"metrics_endpoint"`
+	QueueSettings   exporterhelper.QueueConfig `mapstructure:"sending_queue"`
 	// MetricsSchema indicates the metrics schema to emit to line protocol.
 	// Default: telegraf-prometheus-v2
-	MetricsSchema    string                     `mapstructure:"metrics_schema"`
+	MetricsSchema string `mapstructure:"metrics_schema"`
 	// PayloadMaxLines is the maximum number of line protocol lines to POST in a single request.
-	PayloadMaxLines  int                        `mapstructure:"payload_max_lines"`
+	PayloadMaxLines int `mapstructure:"payload_max_lines"`
 	// PayloadMaxBytes is the maximum number of line protocol bytes to POST in a single request.
-	PayloadMaxBytes  int                        `mapstructure:"payload_max_bytes"`
+	PayloadMaxBytes int `mapstructure:"payload_max_bytes"`
+}
+type LogsConfig struct {
+	// App token is the token of Sematext Monitoring App to which you want to send the logs.
+	AppToken string `mapstructure:"app_token"`
+	// LogsEndpoint specifies the endpoint for receiving logs in Sematext
+	LogsEndpoint string `mapstructure:"logs_endpoint"`
+	// LogRequests determines whether request tracking is enabled
+	// LogRequests bool `mapstructure:"logs_requests"`
 }
 
 // Validate checks for invalid or missing entries in the configuration.
 func (cfg *Config) Validate() error {
-	if strings.ToLower(cfg.Region) != "eu" && strings.ToLower(cfg.Region) != "us" && strings.ToLower(cfg.Region) != "custom"{
+	if strings.ToLower(cfg.Region) != euRegion && strings.ToLower(cfg.Region) != usRegion {
 		return fmt.Errorf("invalid region: %s. please use either 'EU' or 'US'", cfg.Region)
 	}
-	if len(cfg.AppToken) != 36 {
-		return fmt.Errorf("invalid app_token: %s. app_token should be 36 characters", cfg.AppToken)
+	if len(cfg.MetricsConfig.AppToken) != 36 {
+		return fmt.Errorf("invalid metrics app_token: %s. app_token should be 36 characters", cfg.MetricsConfig.AppToken)
+	}
+	if len(cfg.LogsConfig.AppToken) != 36 {
+		return fmt.Errorf("invalid logs app_token: %s. app_token should be 36 characters", cfg.LogsConfig.AppToken)
 	}
-	if strings.ToLower(cfg.Region) == "eu" {
-		cfg.MetricsEndpoint ="https://spm-receiver.eu.sematext.com"
+	if strings.ToLower(cfg.Region) == euRegion {
+		cfg.MetricsEndpoint = euMetricsEndpoint
+		cfg.LogsEndpoint = euLogsEndpoint
 	}
-	if strings.ToLower(cfg.Region) == "us"{
-		cfg.MetricsEndpoint ="https://spm-receiver.sematext.com"
+	if strings.ToLower(cfg.Region) == usRegion {
+		cfg.MetricsEndpoint = usMetricsEndpoint
+		cfg.LogsEndpoint = usLogsEndpoint
 	}
 
 	return nil

exporter/sematextexporter/config_test.go

@@ -39,8 +39,8 @@ func TestLoadConfig(t *testing.T) {
 			id: component.NewIDWithName(metadata.Type, "override-config"),
 			expected: &Config{
 				ClientConfig: confighttp.ClientConfig{
-					Timeout:  500 * time.Millisecond,
-					Headers:  map[string]configopaque.String{"User-Agent": "OpenTelemetry -> Sematext"},
+					Timeout: 500 * time.Millisecond,
+					Headers: map[string]configopaque.String{"User-Agent": "OpenTelemetry -> Sematext"},
 				},
 				MetricsConfig: MetricsConfig{
 					MetricsEndpoint: "https://spm-receiver.sematext.com",
@@ -49,12 +49,16 @@ func TestLoadConfig(t *testing.T) {
 						NumConsumers: 3,
 						QueueSize:    10,
 					},
-					AppToken:       "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx",
+					AppToken:        "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx",
 					MetricsSchema:   "telegraf-prometheus-v2",
 					PayloadMaxLines: 72,
 					PayloadMaxBytes: 27,
 				},
-
+				LogsConfig: LogsConfig{
+					AppToken:      "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx",
+					LogsEndpoint:  "https://logsene-receiver.sematext.com",
+				},
+
 				BackOffConfig: configretry.BackOffConfig{
 					Enabled:             true,
 					InitialInterval:     1 * time.Second,
@@ -63,7 +67,7 @@ func TestLoadConfig(t *testing.T) {
 					RandomizationFactor: backoff.DefaultRandomizationFactor,
 					Multiplier:          backoff.DefaultMultiplier,
 				},
-				Region:          "US",
+				Region: "us",
 			},
 		},
 	}
@@ -82,3 +86,87 @@ func TestLoadConfig(t *testing.T) {
 		})
 	}
 }
+func TestConfigValidation(t *testing.T) {
+	tests := []struct {
+		name        string
+		config      *Config
+		expectError bool
+	}{
+		{
+			name: "Valid configuration 1",
+			config: &Config{
+				Region: "US",
+				MetricsConfig: MetricsConfig{
+					AppToken: "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx",
+				},
+				LogsConfig: LogsConfig{
+					AppToken: "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx",
+				},
+			},
+			expectError: false,
+		},
+		{
+			name: "Valid configuration 2",
+			config: &Config{
+				Region: "EU",
+				MetricsConfig: MetricsConfig{
+					AppToken: "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx",
+				},
+				LogsConfig: LogsConfig{
+					AppToken: "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx",
+				},
+			},
+			expectError: false,
+		},
+		{
+			name: "Invalid region",
+			config: &Config{
+				Region: "ASIA",
+				MetricsConfig: MetricsConfig{
+					AppToken: "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx",
+				},
+				LogsConfig: LogsConfig{
+					AppToken: "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx",
+				},
+			},
+			expectError: true,
+		},
+		{
+			name: "Invalid metrics AppToken length",
+			config: &Config{
+				Region: "US",
+				MetricsConfig: MetricsConfig{
+					AppToken: "short-token",
+				},
+				LogsConfig: LogsConfig{
+					AppToken: "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx",
+				},
+			},
+			expectError: true,
+		},
+		{
+			name: "Invalid logs AppToken length",
+			config: &Config{
+				Region: "EU",
+				MetricsConfig: MetricsConfig{
+					AppToken: "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx",
+				},
+				LogsConfig: LogsConfig{
+					AppToken: "short-token",
+				},
+			},
+			expectError: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			err := tt.config.Validate()
+			if tt.expectError {
+				assert.Error(t, err, "Expected an error for invalid configuration")
+			} else {
+				assert.NoError(t, err, "Expected no error for valid configuration")
+			}
+		})
+	}
+}

exporter/sematextexporter/es.go

@@ -0,0 +1,124 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package sematextexporter // import "github.com/open-telemetry/opentelemetry-collector-contrib/exporter/sematextexporter"
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"os"
+	"reflect"
+	"time"
+
+	"github.com/elastic/go-elasticsearch"
+	"github.com/elastic/go-elasticsearch/v8/esapi"
+	"go.uber.org/zap"
+)
+
+type group struct {
+	client *elasticsearch.Client
+	token  string
+}
+
+type client struct {
+	clients  map[string]group
+	config   *Config
+	logger   *zap.Logger
+	hostname string
+}
+
+// Client represents a minimal interface client implementation has to satisfy.
+type Client interface {
+	Bulk(body any, config *Config) error
+}
+
+// NewClient creates a new instance of ES client that internally stores a reference
+// to both event and log receivers.
+func newClient(config *Config, logger *zap.Logger) (Client, error) {
+	clients := make(map[string]group)
+
+	// Client for shipping to logsene
+	if config.LogsConfig.AppToken != "" {
+		c, err := elasticsearch.NewClient(elasticsearch.Config{
+			Addresses: []string{config.LogsEndpoint},
+		})
+		if err != nil {
+			logger.Error("Failed to create Elasticsearch client", zap.Error(err))
+			return nil, fmt.Errorf("elasticsearch client creation failed: %w", err)
+		}
+		clients[config.LogsEndpoint] = group{
+			client: c,
+			token:  config.LogsConfig.AppToken,
+		}
+	}
+
+	hostname, err := os.Hostname()
+	if err != nil {
+		logger.Warn("Could not retrieve hostname", zap.Error(err))
+	}
+
+	return &client{
+		clients:  clients,
+		config:   config,
+		logger:   logger,
+		hostname: hostname,
+	}, nil
+}
+
+// Bulk processes a batch of documents and sends them to the specified LogsEndpoint.
+func (c *client) Bulk(body any, config *Config) error {
+	grp, ok := c.clients[config.LogsEndpoint]
+	if !ok {
+		return fmt.Errorf("no client known for %s endpoint", config.LogsEndpoint)
+	}
+
+	var bulkBuffer bytes.Buffer
+
+	if reflect.TypeOf(body).Kind() == reflect.Slice {
+		v := reflect.ValueOf(body)
+		for i := 0; i < v.Len(); i++ {
+			doc := v.Index(i).Interface()
+			if docMap, ok := doc.(map[string]any); ok {
+				docMap["os.host"] = c.hostname
+			}
+
+			meta := map[string]map[string]string{
+				"index": {"_index": grp.token},
+			}
+			metaBytes, _ := json.Marshal(meta)
+			docBytes, _ := json.Marshal(doc)
+
+			bulkBuffer.Write(metaBytes)
+			bulkBuffer.WriteByte('\n')
+			bulkBuffer.Write(docBytes)
+			bulkBuffer.WriteByte('\n')
+		}
+	}
+
+	if bulkBuffer.Len() > 0 {
+		ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
+		defer cancel()
+
+		req := esapi.BulkRequest{
+			Body: bytes.NewReader(bulkBuffer.Bytes()),
+		}
+
+		res, err := req.Do(ctx, grp.client)
+		if err != nil {
+			c.logger.Error("Bulk request failed", zap.Error(err))
+			return err
+		}
+		defer res.Body.Close()
+
+		if res.IsError() {
+			c.logger.Error("Bulk request returned an error", zap.String("response", res.String()))
+			return fmt.Errorf("bulk request error: %s", res.String())
+		}
+
+		c.logger.Info("Bulk request successful", zap.String("response", res.String()))
+	}
+
+	return nil
+}