[exporter/postgres] - initial Revamped PR for new Postgres Exporter Component

exporter/postgresexporter/Makefile

@@ -0,0 +1 @@
+include ../../Makefile.Common

exporter/postgresexporter/config.go

@@ -0,0 +1,20 @@
+package postgresexporter
+
+import (
+	"go.opentelemetry.io/collector/config/configretry"
+	"go.opentelemetry.io/collector/exporter/exporterhelper"
+)
+
+type Config struct {
+	TimeoutSettings           exporterhelper.TimeoutConfig `mapstructure:",squash"`
+	configretry.BackOffConfig `mapstructure:"retry_on_failure"`
+	QueueSettings             exporterhelper.QueueConfig `mapstructure:"sending_queue"`
+	Username                  string                     `mapstructure:"username"`
+	Password                  string                     `mapstructure:"password"`
+	Database                  string                     `mapstructure:"database"`
+	Port                      int                        `mapstructure:"port"`
+	Host                      string                     `mapstructure:"host"`
+	LogsTableName             string                     `mapstructure:"logs_table_name"`
+	TracesTableName           string                     `mapstructure:"traces_table_name"`
+	MetricsTableName          string                     `mapstructure:"metrics_table_name"`
+}

exporter/postgresexporter/doc.go

@@ -0,0 +1,7 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+//go:generate mdatagen metadata.yaml
+
+// Package fooreceiver bars.
+package postgresexporter // import "github.com/open-telemetry/opentelemetry-collector-contrib/exporter/postgresexporter"

exporter/postgresexporter/example/otel-collector-config.yml

@@ -0,0 +1,31 @@
+receivers:
+  otlp:
+    protocols:
+      grpc:
+        endpoint: 0.0.0.0:4317
+      http:
+        endpoint: 0.0.0.0:4318
+exporters:
+  postgresexporter:
+    username: "postgres"
+    password: "postgres"
+    database: "postgres"
+    logs_table_name: "otellogs"
+    traces_table_name: "oteltraces"
+    metrics_table_name: "otelmetrics"
+    port: 5432
+    host: "localhost"
+    timeout: 1000s
+    sending_queue:
+      queue_size: 100
+    retry_on_failure:
+      enabled: true
+      initial_interval: 5s
+      max_interval: 30s
+      max_elapsed_time: 300s
+service:
+  pipelines:
+    logs:
+      receivers: [ otlp ]
+      exporters: 
+        - postgresexporter

exporter/postgresexporter/exporter_logs.go

@@ -0,0 +1,61 @@
+package postgresexporter
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+
+	_ "github.com/lib/pq"
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/pdata/plog"
+	"go.uber.org/zap"
+)
+
+type logsExporter struct {
+	client    *sql.DB
+	insertSQL string
+	logger    *zap.Logger
+	cfg       *Config
+}
+
+func newLogsExporter(logger *zap.Logger, cfg *Config) (*logsExporter, error) {
+	client, err := newPostgresClient(cfg)
+	if err != nil {
+		return nil, err
+	}
+
+	return &logsExporter{
+		client:    client,
+		insertSQL: "",
+		logger:    logger,
+		cfg:       cfg,
+	}, nil
+}
+
+func (e *logsExporter) start(ctx context.Context, _ component.Host) error {
+	context.TODO()
+	return nil
+}
+
+func (e *logsExporter) shutdown(_ context.Context) error {
+	if e.client != nil {
+		return e.client.Close()
+	}
+	return nil
+}
+
+func (e *logsExporter) pushLogs(ctx context.Context, ld plog.Logs) error {
+	context.TODO()
+	return nil
+}
+
+func newPostgresClient(cfg *Config) (*sql.DB, error) {
+	psqlInfo := fmt.Sprintf("host=%s port=%d user=%s password=%s dbname=%s sslmode=disable",
+		cfg.Host, cfg.Port, cfg.Username, cfg.Password, cfg.Database)
+	db, err := sql.Open("postgres", psqlInfo)
+
+	if err != nil {
+		return nil, err
+	}
+	return db, nil
+}

exporter/postgresexporter/exporter_metrics.go

@@ -0,0 +1,48 @@
+package postgresexporter
+
+import (
+	"context"
+	"database/sql"
+
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/pdata/pmetric"
+	"go.uber.org/zap"
+)
+
+type metricsExporter struct {
+	client    *sql.DB
+	insertSQL string
+	logger    *zap.Logger
+	cfg       *Config
+}
+
+func newMetricsExporter(logger *zap.Logger, cfg *Config) (*metricsExporter, error) {
+	client, err := newPostgresClient(cfg)
+	if err != nil {
+		return nil, err
+	}
+
+	return &metricsExporter{
+		client:    client,
+		insertSQL: "",
+		logger:    logger,
+		cfg:       cfg,
+	}, nil
+}
+
+func (e *metricsExporter) start(ctx context.Context, _ component.Host) error {
+	context.TODO()
+	return nil
+}
+
+func (e *metricsExporter) shutdown(_ context.Context) error {
+	if e.client != nil {
+		return e.client.Close()
+	}
+	return nil
+}
+
+func (e *metricsExporter) pushMetricData(ctx context.Context, td pmetric.Metrics) error {
+	context.TODO()
+	return nil
+}

exporter/postgresexporter/exporter_traces.go

@@ -0,0 +1,48 @@
+package postgresexporter
+
+import (
+	"context"
+	"database/sql"
+
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/pdata/ptrace"
+	"go.uber.org/zap"
+)
+
+type tracesExporter struct {
+	client    *sql.DB
+	insertSQL string
+	logger    *zap.Logger
+	cfg       *Config
+}
+
+func newTracesExporter(logger *zap.Logger, cfg *Config) (*tracesExporter, error) {
+	client, err := newPostgresClient(cfg)
+	if err != nil {
+		return nil, err
+	}
+
+	return &tracesExporter{
+		client:    client,
+		insertSQL: "",
+		logger:    logger,
+		cfg:       cfg,
+	}, nil
+}
+
+func (e *tracesExporter) start(ctx context.Context, _ component.Host) error {
+	context.TODO()
+	return nil
+}
+
+func (e *tracesExporter) shutdown(_ context.Context) error {
+	if e.client != nil {
+		return e.client.Close()
+	}
+	return nil
+}
+
+func (e *tracesExporter) pushTraceData(ctx context.Context, td ptrace.Traces) error {
+	context.TODO()
+	return nil
+}

exporter/postgresexporter/factory.go

@@ -0,0 +1,148 @@
+package postgresexporter
+
+import (
+	"context"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/postgresexporter/internal/metadata"
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/config/configretry"
+	"go.opentelemetry.io/collector/exporter"
+	"go.opentelemetry.io/collector/exporter/exporterhelper"
+	"go.uber.org/zap"
+)
+
+func NewFactory() exporter.Factory {
+	return exporter.NewFactory(metadata.Type,
+		createDefaultConfig,
+		exporter.WithTraces(createTracesExporter, metadata.TracesStability),
+		exporter.WithLogs(createLogsExporter, metadata.LogsStability),
+		exporter.WithMetrics(createMetricsExporter, metadata.MetricsStability),
+	)
+}
+
+func createDefaultConfig() component.Config {
+	return &Config{
+		TimeoutSettings:  exporterhelper.NewDefaultTimeoutConfig(),
+		QueueSettings:    exporterhelper.NewDefaultQueueConfig(),
+		BackOffConfig:    configretry.NewDefaultBackOffConfig(),
+		Username:         "postgres",
+		Password:         "postgres",
+		Database:         "postgres",
+		Port:             5432,
+		Host:             "localhost",
+		LogsTableName:    "otel_logs",
+		TracesTableName:  "otel_traces",
+		MetricsTableName: "otel_metrics",
+	}
+}
+
+func createLogsExporter(
+	ctx context.Context,
+	set exporter.Settings,
+	config component.Config) (exporter.Logs, error) {
+
+	set.Logger.Info("createLogsExporter called")
+
+	cfg := config.(*Config)
+	s, err := newLogsExporter(set.Logger, cfg)
+	if err != nil {
+		set.Logger.Error("Failed to create new logs exporter", zap.Error(err))
+		return nil, err
+	}
+
+	set.Logger.Info("newLogsExporter created successfully")
+
+	exporter, err := exporterhelper.NewLogsExporter(
+		ctx,
+		set,
+		cfg,
+		s.pushLogs,
+		exporterhelper.WithQueue(cfg.QueueSettings),
+		exporterhelper.WithRetry(cfg.BackOffConfig),
+		exporterhelper.WithStart(s.start),
+		exporterhelper.WithShutdown(s.shutdown),
+		exporterhelper.WithTimeout(s.cfg.TimeoutSettings),
+	)
+
+	if err != nil {
+		set.Logger.Error("Failed to create logs exporter", zap.Error(err))
+		return nil, err
+	}
+
+	set.Logger.Info("Logs exporter created successfully")
+	return exporter, nil
+}
+
+func createTracesExporter(
+	ctx context.Context,
+	set exporter.Settings,
+	config component.Config) (exporter.Traces, error) {
+
+	set.Logger.Info("createTracesExporter called")
+
+	cfg := config.(*Config)
+	s, err := newTracesExporter(set.Logger, cfg)
+	if err != nil {
+		set.Logger.Error("Failed to create new traces exporter", zap.Error(err))
+		return nil, err
+	}
+
+	set.Logger.Info("newTracesExporter created successfully")
+
+	exporter, err := exporterhelper.NewTracesExporter(
+		ctx,
+		set,
+		cfg,
+		s.pushTraceData,
+		exporterhelper.WithQueue(cfg.QueueSettings),
+		exporterhelper.WithRetry(cfg.BackOffConfig),
+		exporterhelper.WithStart(s.start),
+		exporterhelper.WithShutdown(s.shutdown),
+		exporterhelper.WithTimeout(s.cfg.TimeoutSettings),
+	)
+
+	if err != nil {
+		set.Logger.Error("Failed to create traces exporter", zap.Error(err))
+		return nil, err
+	}
+
+	set.Logger.Info("Traces exporter created successfully")
+	return exporter, nil
+}
+
+func createMetricsExporter(
+	ctx context.Context,
+	set exporter.Settings,
+	config component.Config) (exporter.Metrics, error) {
+
+	set.Logger.Info("createMetricExporters called")
+
+	cfg := config.(*Config)
+	s, err := newMetricsExporter(set.Logger, cfg)
+	if err != nil {
+		set.Logger.Error("Failed to create new metrics exporter", zap.Error(err))
+		return nil, err
+	}
+
+	set.Logger.Info("newMetricExporter created successfully")
+
+	exporter, err := exporterhelper.NewMetricsExporter(
+		ctx,
+		set,
+		cfg,
+		s.pushMetricData,
+		exporterhelper.WithQueue(cfg.QueueSettings),
+		exporterhelper.WithRetry(cfg.BackOffConfig),
+		exporterhelper.WithStart(s.start),
+		exporterhelper.WithShutdown(s.shutdown),
+		exporterhelper.WithTimeout(s.cfg.TimeoutSettings),
+	)
+
+	if err != nil {
+		set.Logger.Error("Failed to create metrics exporter", zap.Error(err))
+		return nil, err
+	}
+
+	set.Logger.Info("Metrics exporter created successfully")
+	return exporter, nil
+}