chore: Implement supported telemetry type pattern

generator/apache/apache.go

@@ -12,6 +12,7 @@ import (
 	"github.com/cenkalti/backoff/v4"
 	"github.com/observiq/blitz/internal/generator/security"
 	"github.com/observiq/blitz/output"
+	"github.com/observiq/blitz/telemetry"
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/metric"
@@ -96,6 +97,11 @@ func New(logger *zap.Logger, workers int, rate time.Duration) (*ApacheLogGenerat
 
 // Start starts the Apache log generator and writes data using the
 // provided generator writer.
+// SupportedTelemetry returns the telemetry types this generator supports.
+func (g *ApacheLogGenerator) SupportedTelemetry() []telemetry.Type {
+	return []telemetry.Type{telemetry.Logs}
+}
+
 func (g *ApacheLogGenerator) Start(writer output.Writer) error {
 	g.logger.Info("Starting Apache log generator",
 		zap.Int("workers", g.workers),

generator/apache_combined/apache_combined.go

@@ -11,6 +11,7 @@ import (
 	"github.com/cenkalti/backoff/v4"
 	"github.com/observiq/blitz/internal/useragent"
 	"github.com/observiq/blitz/output"
+	"github.com/observiq/blitz/telemetry"
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/metric"
@@ -97,6 +98,11 @@ func New(logger *zap.Logger, workers int, rate time.Duration) (*ApacheCombinedLo
 
 // Start starts the Apache Combined log generator and writes data using the
 // provided generator writer.
+// SupportedTelemetry returns the telemetry types this generator supports.
+func (g *ApacheCombinedLogGenerator) SupportedTelemetry() []telemetry.Type {
+	return []telemetry.Type{telemetry.Logs}
+}
+
 func (g *ApacheCombinedLogGenerator) Start(writer output.Writer) error {
 	g.logger.Info("Starting Apache Combined log generator",
 		zap.Int("workers", g.workers),

generator/apache_error/apache_error.go

@@ -11,6 +11,7 @@ import (
 
 	"github.com/cenkalti/backoff/v4"
 	"github.com/observiq/blitz/output"
+	"github.com/observiq/blitz/telemetry"
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/metric"
@@ -94,6 +95,11 @@ func New(logger *zap.Logger, workers int, rate time.Duration) (*ApacheErrorLogGe
 
 // Start starts the Apache Error log generator and writes data using the
 // provided generator writer.
+// SupportedTelemetry returns the telemetry types this generator supports.
+func (g *ApacheErrorLogGenerator) SupportedTelemetry() []telemetry.Type {
+	return []telemetry.Type{telemetry.Logs}
+}
+
 func (g *ApacheErrorLogGenerator) Start(writer output.Writer) error {
 	g.logger.Info("Starting Apache Error log generator",
 		zap.Int("workers", g.workers),

generator/filegen/filegen.go

@@ -16,6 +16,7 @@ import (
 	"github.com/hashicorp/golang-lru/v2/expirable"
 	"github.com/observiq/blitz/internal/generator/ctime"
 	"github.com/observiq/blitz/output"
+	"github.com/observiq/blitz/telemetry"
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/metric"
@@ -151,6 +152,11 @@ func New(logger *zap.Logger, workers int, rate time.Duration, source string, cac
 }
 
 // Start starts the File log generator
+// SupportedTelemetry returns the telemetry types this generator supports.
+func (g *FileLogGenerator) SupportedTelemetry() []telemetry.Type {
+	return []telemetry.Type{telemetry.Logs}
+}
+
 func (g *FileLogGenerator) Start(writer output.Writer) error {
 	g.logger.Info("Starting File log generator",
 		zap.String("source", g.source),

generator/generator.go

@@ -4,10 +4,14 @@ import (
 	"context"
 
 	"github.com/observiq/blitz/output"
+	"github.com/observiq/blitz/telemetry"
 )
 
 // Generator is the interface for generating data.
 type Generator interface {
+	// SupportedTelemetry returns the telemetry types this generator can produce.
+	SupportedTelemetry() []telemetry.Type
+
 	// Start starts the generator and writes data using the
 	// provided generator writer.
 	Start(writer output.Writer) error

generator/json/json.go

@@ -11,6 +11,7 @@ import (
 	jsonlib "github.com/goccy/go-json"
 	"github.com/observiq/blitz/internal/generator/logtypes"
 	"github.com/observiq/blitz/output"
+	"github.com/observiq/blitz/telemetry"
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/metric"
@@ -40,6 +41,11 @@ type JSONLogGenerator struct {
 	jsonWriteErrors   metric.Int64Counter
 }
 
+// SupportedTelemetry returns the telemetry types this generator supports.
+func (g *JSONLogGenerator) SupportedTelemetry() []telemetry.Type {
+	return []telemetry.Type{telemetry.Logs}
+}
+
 // New creates a new JSON log generator
 func New(logger *zap.Logger, workers int, rate time.Duration, logType string) (*JSONLogGenerator, error) {
 	if logger == nil {

generator/kubernetes/kubernetes.go

@@ -11,6 +11,7 @@ import (
 
 	"github.com/cenkalti/backoff/v4"
 	"github.com/observiq/blitz/output"
+	"github.com/observiq/blitz/telemetry"
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/metric"
@@ -124,6 +125,11 @@ func New(logger *zap.Logger, workers int, rate time.Duration, format string) (*G
 
 // Start starts the Kubernetes container log generator and writes data using the
 // provided generator writer.
+// SupportedTelemetry returns the telemetry types this generator supports.
+func (g *Generator) SupportedTelemetry() []telemetry.Type {
+	return []telemetry.Type{telemetry.Logs}
+}
+
 func (g *Generator) Start(writer output.Writer) error {
 	g.logger.Info("Starting Kubernetes container log generator",
 		zap.Int("workers", g.workers),

generator/nginx/nginx.go

@@ -12,6 +12,7 @@ import (
 	"github.com/observiq/blitz/internal/generator/security"
 	"github.com/observiq/blitz/internal/useragent"
 	"github.com/observiq/blitz/output"
+	"github.com/observiq/blitz/telemetry"
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/metric"
@@ -184,6 +185,11 @@ func New(logger *zap.Logger, workers int, rate time.Duration) (*Generator, error
 
 // Start starts the NGINX log generator and writes data using the
 // provided generator writer.
+// SupportedTelemetry returns the telemetry types this generator supports.
+func (g *Generator) SupportedTelemetry() []telemetry.Type {
+	return []telemetry.Type{telemetry.Logs}
+}
+
 func (g *Generator) Start(writer output.Writer) error {
 	g.logger.Info("Starting NGINX log generator",
 		zap.Int("workers", g.workers),

generator/nop/nop.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 
 	"github.com/observiq/blitz/output"
+	"github.com/observiq/blitz/telemetry"
 	"go.uber.org/zap"
 )
 
@@ -24,6 +25,11 @@ func New(logger *zap.Logger) (*NopGenerator, error) {
 	}, nil
 }
 
+// SupportedTelemetry returns the telemetry types this generator supports.
+func (g *NopGenerator) SupportedTelemetry() []telemetry.Type {
+	return []telemetry.Type{telemetry.Logs}
+}
+
 // Start starts the nop generator (performs no work)
 func (g *NopGenerator) Start(writer output.Writer) error {
 	g.logger.Info("Starting NOP generator (no work performed)")

generator/okta/okta.go

@@ -10,6 +10,7 @@ import (
 	"github.com/cenkalti/backoff/v4"
 	jsonlib "github.com/goccy/go-json"
 	"github.com/observiq/blitz/output"
+	"github.com/observiq/blitz/telemetry"
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/metric"
@@ -239,6 +240,11 @@ func New(logger *zap.Logger, workers int, rate time.Duration) (*Generator, error
 }
 
 // Start starts the Okta log generator
+// SupportedTelemetry returns the telemetry types this generator supports.
+func (g *Generator) SupportedTelemetry() []telemetry.Type {
+	return []telemetry.Type{telemetry.Logs}
+}
+
 func (g *Generator) Start(writer output.Writer) error {
 	g.logger.Info("Starting Okta log generator",
 		zap.Int("workers", g.workers),

generator/paloalto/paloalto.go

@@ -13,6 +13,7 @@ import (
 
 	"github.com/cenkalti/backoff/v4"
 	"github.com/observiq/blitz/output"
+	"github.com/observiq/blitz/telemetry"
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/metric"
@@ -35,6 +36,11 @@ type Generator struct {
 	writeErrors   metric.Int64Counter
 }
 
+// SupportedTelemetry returns the telemetry types this generator supports.
+func (g *Generator) SupportedTelemetry() []telemetry.Type {
+	return []telemetry.Type{telemetry.Logs}
+}
+
 // New creates a new Palo Alto generator.
 func New(logger *zap.Logger, workers int, rate time.Duration) (*Generator, error) {
 	if logger == nil {

generator/postgres/postgres.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/cenkalti/backoff/v4"
 	"github.com/observiq/blitz/output"
+	"github.com/observiq/blitz/telemetry"
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/metric"
@@ -257,6 +258,11 @@ func New(logger *zap.Logger, workers int, rate time.Duration) (*Generator, error
 
 // Start starts the PostgreSQL log generator and writes data using the
 // provided generator writer.
+// SupportedTelemetry returns the telemetry types this generator supports.
+func (g *Generator) SupportedTelemetry() []telemetry.Type {
+	return []telemetry.Type{telemetry.Logs}
+}
+
 func (g *Generator) Start(writer output.Writer) error {
 	g.logger.Info("Starting PostgreSQL log generator",
 		zap.Int("workers", g.workers),

generator/winevt/winevt.go

@@ -9,6 +9,7 @@ import (
 	"github.com/cenkalti/backoff/v4"
 	"github.com/observiq/blitz/internal/generators/winevt/templates"
 	"github.com/observiq/blitz/output"
+	"github.com/observiq/blitz/telemetry"
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/metric"
@@ -31,6 +32,11 @@ type WinevtGenerator struct {
 	winevtWriteErrors   metric.Int64Counter
 }
 
+// SupportedTelemetry returns the telemetry types this generator supports.
+func (g *WinevtGenerator) SupportedTelemetry() []telemetry.Type {
+	return []telemetry.Type{telemetry.Logs}
+}
+
 // New creates a new Windows Event generator.
 func New(logger *zap.Logger, workers int, rate time.Duration) (*WinevtGenerator, error) {
 	if logger == nil {

internal/service/service.go

@@ -7,6 +7,7 @@ import (
 
 	"github.com/observiq/blitz/generator"
 	"github.com/observiq/blitz/output"
+	"github.com/observiq/blitz/telemetry"
 	"go.uber.org/zap"
 )
 
@@ -27,6 +28,12 @@ func New(logger *zap.Logger, generator generator.Generator, output output.Output
 		return nil, fmt.Errorf("output cannot be nil")
 	}
 
+	telemetryType, err := telemetry.Compatible(generator.SupportedTelemetry(), output.SupportedTelemetry())
+	if err != nil {
+		return nil, fmt.Errorf("incompatible generator and output: %w", err)
+	}
+	logger.Info("telemetry type", zap.String("type", string(telemetryType)))
+
 	return &Service{
 		Logger:    logger,
 		Generator: generator,

output/file/file.go

@@ -7,6 +7,7 @@ import (
 
 	"github.com/observiq/blitz/internal/workermanager"
 	"github.com/observiq/blitz/output"
+	"github.com/observiq/blitz/telemetry"
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/metric"
@@ -200,6 +201,11 @@ func (f *File) Write(ctx context.Context, data output.LogRecord) error {
 }
 
 // Stop gracefully stops workers and closes the writer
+// SupportedTelemetry returns the telemetry types this output supports.
+func (f *File) SupportedTelemetry() []telemetry.Type {
+	return []telemetry.Type{telemetry.Logs}
+}
+
 func (f *File) Stop(ctx context.Context) error {
 	f.logger.Info("Stopping File output")
 

output/nop/nop.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 
 	"github.com/observiq/blitz/output"
+	"github.com/observiq/blitz/telemetry"
 	"go.uber.org/zap"
 )
 
@@ -24,6 +25,11 @@ func New(logger *zap.Logger) (*NopOutput, error) {
 	}, nil
 }
 
+// SupportedTelemetry returns the telemetry types this output supports.
+func (o *NopOutput) SupportedTelemetry() []telemetry.Type {
+	return []telemetry.Type{telemetry.Logs}
+}
+
 // Write performs no work (data is discarded)
 func (o *NopOutput) Write(ctx context.Context, data output.LogRecord) error {
 	// No-op: data is discarded

output/otlp_grpc/otlp_grpc.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/observiq/blitz/internal/workermanager"
 	"github.com/observiq/blitz/output"
+	"github.com/observiq/blitz/telemetry"
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/metric"
@@ -406,6 +407,11 @@ func (o *OTLPGrpc) Write(ctx context.Context, data output.LogRecord) error {
 // Stop shall not be called more than once.
 // If the provided context is done, Stop will return immediately
 // even if workers are still shutting down.
+// SupportedTelemetry returns the telemetry types this output supports.
+func (o *OTLPGrpc) SupportedTelemetry() []telemetry.Type {
+	return []telemetry.Type{telemetry.Logs}
+}
+
 func (o *OTLPGrpc) Stop(ctx context.Context) error {
 	o.logger.Info("Stopping OTLP gRPC output")
 

output/output.go

@@ -3,6 +3,8 @@ package output
 import (
 	"context"
 	"time"
+
+	"github.com/observiq/blitz/telemetry"
 )
 
 type LogRecord struct {
@@ -34,6 +36,9 @@ type Writer interface {
 type Output interface {
 	Writer
 
+	// SupportedTelemetry returns the telemetry types this output can send.
+	SupportedTelemetry() []telemetry.Type
+
 	// Stop stops the output.
 	Stop(ctx context.Context) error
 }

output/stdout/stdout.go

@@ -6,6 +6,7 @@ import (
 	"os"
 
 	"github.com/observiq/blitz/output"
+	"github.com/observiq/blitz/telemetry"
 	"go.uber.org/zap"
 )
 
@@ -25,6 +26,11 @@ func New(logger *zap.Logger) (*StdoutOutput, error) {
 	}, nil
 }
 
+// SupportedTelemetry returns the telemetry types this output supports.
+func (o *StdoutOutput) SupportedTelemetry() []telemetry.Type {
+	return []telemetry.Type{telemetry.Logs}
+}
+
 // Write writes the log record to stdout
 func (o *StdoutOutput) Write(ctx context.Context, data output.LogRecord) error {
 	_, err := fmt.Fprintln(os.Stdout, data.Message)

output/syslog/syslog.go

@@ -10,6 +10,7 @@ import (
 	"time"
 
 	"github.com/observiq/blitz/output"
+	"github.com/observiq/blitz/telemetry"
 	"github.com/observiq/blitz/output/syslog/ident"
 	"github.com/observiq/blitz/output/tcp"
 	"github.com/observiq/blitz/output/udp"
@@ -158,6 +159,11 @@ func (s *Syslog) Write(ctx context.Context, rec output.LogRecord) error {
 	return s.transport.Write(ctx, output.LogRecord{Message: formatted})
 }
 
+// SupportedTelemetry returns the telemetry types this output supports.
+func (s *Syslog) SupportedTelemetry() []telemetry.Type {
+	return []telemetry.Type{telemetry.Logs}
+}
+
 // Stop delegates to the underlying transport.
 func (s *Syslog) Stop(ctx context.Context) error {
 	return s.transport.Stop(ctx)