feat(oteldbexporter): implement oteldb as otelcol exporter

internal/otelreceiver/oteldbexporter/config.go

@@ -0,0 +1,42 @@
+package oteldbexporter
+
+import (
+	"context"
+
+	"github.com/go-faster/errors"
+	"go.opentelemetry.io/collector/exporter"
+
+	"github.com/go-faster/oteldb/internal/chstorage"
+)
+
+// Config defines [Exporter] config.
+type Config struct {
+	DSL string `mapstructure:"dsl"`
+}
+
+func (c *Config) connect(ctx context.Context, settings exporter.CreateSettings) (*chstorage.Inserter, error) {
+	pool, err := chstorage.Dial(ctx, c.DSL, chstorage.DialOptions{
+		MeterProvider:  settings.MeterProvider,
+		TracerProvider: settings.TracerProvider,
+		Logger:         settings.Logger,
+	})
+	if err != nil {
+		return nil, errors.Wrap(err, "dial clickhouse")
+	}
+
+	tables := chstorage.DefaultTables()
+	if err := tables.Create(ctx, pool); err != nil {
+		return nil, errors.Wrap(err, "create tables")
+	}
+
+	inserter, err := chstorage.NewInserter(pool, chstorage.InserterOptions{
+		Tables:         tables,
+		MeterProvider:  settings.MeterProvider,
+		TracerProvider: settings.TracerProvider,
+	})
+	if err != nil {
+		return nil, errors.Wrap(err, "create inserter")
+	}
+
+	return inserter, nil
+}

internal/otelreceiver/oteldbexporter/oteldbexporter.go

@@ -0,0 +1,74 @@
+// Package oteldbexporter contains oteldb exporter factory.
+package oteldbexporter
+
+import (
+	"context"
+
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/exporter"
+	"go.opentelemetry.io/collector/exporter/exporterhelper"
+
+	"github.com/go-faster/oteldb/internal/logstorage"
+	"github.com/go-faster/oteldb/internal/tracestorage"
+)
+
+const (
+	typeStr   = "oteldbexporter"
+	stability = component.StabilityLevelDevelopment
+)
+
+// NewFactory creates new factory of [Exporter].
+func NewFactory() exporter.Factory {
+	return exporter.NewFactory(
+		typeStr,
+		createDefaultConfig,
+		exporter.WithTraces(createTracesExporter, stability),
+		exporter.WithMetrics(createMetricsExporter, stability),
+		exporter.WithLogs(createLogsExporter, stability),
+	)
+}
+
+func createDefaultConfig() component.Config {
+	return &Config{
+		DSL: "clickhouse://localhost:9000",
+	}
+}
+
+func createTracesExporter(
+	ctx context.Context,
+	settings exporter.CreateSettings,
+	cfg component.Config,
+) (exporter.Traces, error) {
+	ecfg := cfg.(*Config)
+	inserter, err := ecfg.connect(ctx, settings)
+	if err != nil {
+		return nil, err
+	}
+	return exporterhelper.NewTracesExporter(ctx, settings, cfg, tracestorage.NewConsumer(inserter).ConsumeTraces)
+}
+
+func createMetricsExporter(
+	ctx context.Context,
+	settings exporter.CreateSettings,
+	cfg component.Config,
+) (exporter.Metrics, error) {
+	ecfg := cfg.(*Config)
+	inserter, err := ecfg.connect(ctx, settings)
+	if err != nil {
+		return nil, err
+	}
+	return exporterhelper.NewMetricsExporter(ctx, settings, cfg, inserter.ConsumeMetrics)
+}
+
+func createLogsExporter(
+	ctx context.Context,
+	settings exporter.CreateSettings,
+	cfg component.Config,
+) (exporter.Logs, error) {
+	ecfg := cfg.(*Config)
+	inserter, err := ecfg.connect(ctx, settings)
+	if err != nil {
+		return nil, err
+	}
+	return exporterhelper.NewLogsExporter(ctx, settings, cfg, logstorage.NewConsumer(inserter).ConsumeLogs)
+}