Merge pull request #1 from braydonk/gcloud_logs_exporter

exporter/googlecloudloggingexporter: did it

exporter/googlecloudloggingexporter/Makefile

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

exporter/googlecloudloggingexporter/config.go

@@ -0,0 +1,38 @@
+package googlecloudloggingexporter
+
+import (
+	"go.opentelemetry.io/collector/config"
+	"go.opentelemetry.io/collector/exporter/exporterhelper"
+	"go.uber.org/zap"
+)
+
+type Config struct {
+	config.ExporterSettings `mapstructure:",squash"`
+	ProjectID               string `mapstructure:"project"`
+	UserAgent               string `mapstructure:"user_agent"`
+	LogName                 string `mapstructure:"log_name"`
+
+	Endpoint string `mapstructure:"endpoint"`
+	// Only has effect if Endpoint is not ""
+	UseInsecure bool `mapstructure:"use_insecure"`
+
+	// Timeout for all API calls. If not set, defaults to 12 seconds.
+	exporterhelper.TimeoutSettings `mapstructure:",squash"` // squash ensures fields are correctly decoded in embedded struct.
+	exporterhelper.QueueSettings   `mapstructure:"sending_queue"`
+	exporterhelper.RetrySettings   `mapstructure:"retry_on_failure"`
+
+	logger *zap.Logger
+}
+
+func (c Config) Validate() error {
+	return nil
+}
+
+func (config *Config) enforcedQueueSettings() exporterhelper.QueueSettings {
+	return exporterhelper.QueueSettings{
+		Enabled: true,
+		// due to the sequence token, there can be only one request in flight
+		NumConsumers: 1,
+		QueueSize:    config.QueueSettings.QueueSize,
+	}
+}

exporter/googlecloudloggingexporter/exporter.go

@@ -0,0 +1,200 @@
+package googlecloudloggingexporter
+
+import (
+	"context"
+	"fmt"
+
+	"cloud.google.com/go/logging"
+	"github.com/google/uuid"
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/consumer"
+	"go.opentelemetry.io/collector/exporter/exporterhelper"
+	"go.opentelemetry.io/collector/model/pdata"
+	"go.uber.org/zap"
+)
+
+type exporter struct {
+	Config      *Config
+	logger      *zap.Logger
+	collectorID string
+	cloudLogger *logging.Logger
+}
+
+func newCloudLoggingExporter(config *Config, params component.ExporterCreateSettings) (component.LogsExporter, error) {
+	loggingExporter, err := newCloudLoggingExporter(config, params)
+	if err != nil {
+		return nil, err
+	}
+	return exporterhelper.NewLogsExporter(
+		config,
+		params,
+		loggingExporter.ConsumeLogs,
+		exporterhelper.WithQueue(config.enforcedQueueSettings()),
+		exporterhelper.WithRetry(config.RetrySettings))
+}
+
+func newCloudLoggingLogExporter(config *Config, params component.ExporterCreateSettings) (component.LogsExporter, error) {
+	// Validate the passed config.
+	if err := config.Validate(); err != nil {
+		return nil, err
+	}
+
+	// Generate a Collector ID.
+	collectorIdentifier, err := uuid.NewRandom()
+	if err != nil {
+		return nil, err
+	}
+
+	// Read project ID from Metadata if not specified by config.
+	if config.ProjectID == "" {
+		projectId, err := readProjectIdMetadata()
+		if err != nil {
+			return nil, fmt.Errorf("failed to read Google Cloud project ID: %v", err)
+		}
+		config.ProjectID = projectId
+	}
+
+	// Create Cloud Logging logger with project ID.
+	client, err := logging.NewClient(context.Background(), config.ProjectID)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create Google Cloud Logging client: %v", err)
+	}
+	defer client.Close()
+	logger := client.Logger(config.LogName)
+
+	// Create the logging exporter.
+	loggingExporter := &exporter{
+		Config:      config,
+		logger:      params.Logger,
+		collectorID: collectorIdentifier.String(),
+		cloudLogger: logger,
+	}
+	return loggingExporter, nil
+}
+
+func (e *exporter) ConsumeLogs(ctx context.Context, ld pdata.Logs) error {
+	logEntries, dropped := logsToEntries(e.Config, e.logger, ld)
+	if len(logEntries) == 0 {
+		return nil
+	}
+	if dropped > 0 {
+		e.logger.Debug("Dropped logs", zap.Any("logsDropped", dropped))
+	}
+
+	for _, logEntry := range logEntries {
+		e.logger.Debug("Adding log entry", zap.Any("entry", logEntry))
+		e.cloudLogger.Log(logEntry)
+	}
+	e.logger.Debug("Log entries successfully buffered")
+	err := e.cloudLogger.Flush()
+	if err != nil {
+		e.logger.Error("error force flushing logs. Skipping to next logPusher.", zap.Error(err))
+	}
+	return nil
+}
+
+func (e *exporter) Capabilities() consumer.Capabilities {
+	return consumer.Capabilities{MutatesData: false}
+}
+
+func (e *exporter) Shutdown(ctx context.Context) error {
+	// Flush the remaining logs before shutting down the exporter.
+	if e.cloudLogger != nil {
+		err := e.cloudLogger.Flush()
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (e *exporter) Start(ctx context.Context, host component.Host) error {
+	return nil
+}
+
+func logsToEntries(config *Config, logger *zap.Logger, ld pdata.Logs) ([]logging.Entry, int) {
+	entries := []logging.Entry{}
+	dropped := 0
+	rls := ld.ResourceLogs()
+	for i := 0; i < rls.Len(); i++ {
+		rl := rls.At(i)
+		resourceAttrs := attrsValue(rl.Resource().Attributes())
+		ills := rl.InstrumentationLibraryLogs()
+		for j := 0; j < ills.Len(); j++ {
+			ils := ills.At(j)
+			logs := ils.LogRecords()
+			for k := 0; k < logs.Len(); k++ {
+				log := logs.At(k)
+				entry, err := logToEntry(config, resourceAttrs, log)
+				if err != nil {
+					logger.Debug("Failed to convert to Cloud Logging Entry", zap.Error(err))
+					dropped++
+				} else {
+					entries = append(entries, entry)
+				}
+			}
+		}
+	}
+	return entries, dropped
+}
+
+type entryPayload struct {
+	Message string `json:"message"`
+}
+
+func logToEntry(config *Config, attributes map[string]interface{}, log pdata.LogRecord) (logging.Entry, error) {
+	payload := entryPayload{
+		Message: log.Body().AsString(),
+	}
+	return logging.Entry{
+		Payload:   payload,
+		Timestamp: log.Timestamp().AsTime(),
+		Severity:  logging.Severity(log.SeverityNumber()),
+		LogName:   config.LogName,
+		Trace:     log.TraceID().HexString(),
+		SpanID:    log.SpanID().HexString(),
+	}, nil
+}
+
+func attrsValue(attrs pdata.AttributeMap) map[string]interface{} {
+	if attrs.Len() == 0 {
+		return nil
+	}
+	out := make(map[string]interface{}, attrs.Len())
+	attrs.Range(func(k string, v pdata.AttributeValue) bool {
+		out[k] = attrValue(v)
+		return true
+	})
+	return out
+}
+
+func attrValue(value pdata.AttributeValue) interface{} {
+	switch value.Type() {
+	case pdata.AttributeValueTypeInt:
+		return value.IntVal()
+	case pdata.AttributeValueTypeBool:
+		return value.BoolVal()
+	case pdata.AttributeValueTypeDouble:
+		return value.DoubleVal()
+	case pdata.AttributeValueTypeString:
+		return value.StringVal()
+	case pdata.AttributeValueTypeMap:
+		values := map[string]interface{}{}
+		value.MapVal().Range(func(k string, v pdata.AttributeValue) bool {
+			values[k] = attrValue(v)
+			return true
+		})
+		return values
+	case pdata.AttributeValueTypeArray:
+		arrayVal := value.SliceVal()
+		values := make([]interface{}, arrayVal.Len())
+		for i := 0; i < arrayVal.Len(); i++ {
+			values[i] = attrValue(arrayVal.At(i))
+		}
+		return values
+	case pdata.AttributeValueTypeEmpty:
+		return nil
+	default:
+		return nil
+	}
+}

exporter/googlecloudloggingexporter/factory.go

@@ -0,0 +1,43 @@
+package googlecloudloggingexporter
+
+import (
+	"context"
+	"errors"
+	"time"
+
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/config"
+	"go.opentelemetry.io/collector/exporter/exporterhelper"
+)
+
+var (
+	typeStr        config.Type = "googlecloudlogging"
+	defaultTimeout             = 12 * time.Second
+)
+
+func NewFactory() component.ExporterFactory {
+	return component.NewExporterFactory(
+		typeStr,
+		createDefaultConfig,
+		component.WithLogsExporter(createLogsExporter))
+}
+
+// createDefaultConfig creates the default configuration for exporter.
+func createDefaultConfig() config.Exporter {
+	return &Config{
+		ExporterSettings: config.NewExporterSettings(config.NewComponentID(typeStr)),
+		TimeoutSettings:  exporterhelper.TimeoutSettings{Timeout: defaultTimeout},
+		RetrySettings:    exporterhelper.NewDefaultRetrySettings(),
+		QueueSettings:    exporterhelper.NewDefaultQueueSettings(),
+		UserAgent:        "opentelemetry-collector-contrib {{version}}",
+	}
+}
+
+// createLogsExporter creates the Google Cloud Logging exporter
+func createLogsExporter(_ context.Context, params component.ExporterCreateSettings, config config.Exporter) (component.LogsExporter, error) {
+	expConfig, ok := config.(*Config)
+	if !ok {
+		return nil, errors.New("invalid configuration type; can't cast to googlecloudloggingexporter.Config")
+	}
+	return newCloudLoggingExporter(expConfig, params)
+}

exporter/googlecloudloggingexporter/go.mod

@@ -0,0 +1,52 @@
+module github.com/open-telemetry/opentelemetry-collector-contrib/exporter/googlecloudloggingexporter
+
+go 1.17
+
+replace github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal => ../../internal/coreinternal
+
+require (
+	cloud.google.com/go/logging v1.4.2
+	github.com/google/uuid v1.3.0
+	go.opentelemetry.io/collector v0.45.1-0.20220223001941-c9c253193a75
+	go.opentelemetry.io/collector/model v0.45.1-0.20220222185228-27f7607ca13a
+	go.uber.org/zap v1.21.0
+)
+
+require (
+	cloud.google.com/go v0.81.0 // indirect
+	github.com/cenkalti/backoff/v4 v4.1.2 // indirect
+	github.com/fsnotify/fsnotify v1.5.1 // indirect
+	github.com/gogo/protobuf v1.3.2 // indirect
+	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
+	github.com/golang/protobuf v1.5.2 // indirect
+	github.com/google/go-cmp v0.5.7 // indirect
+	github.com/googleapis/gax-go/v2 v2.0.5 // indirect
+	github.com/jstemmer/go-junit-report v0.9.1 // indirect
+	github.com/knadh/koanf v1.4.0 // indirect
+	github.com/mitchellh/copystructure v1.2.0 // indirect
+	github.com/mitchellh/mapstructure v1.4.3 // indirect
+	github.com/mitchellh/reflectwalk v1.0.2 // indirect
+	github.com/pelletier/go-toml v1.9.4 // indirect
+	github.com/pkg/errors v0.9.1 // indirect
+	github.com/spf13/cast v1.4.1 // indirect
+	go.opencensus.io v0.23.0 // indirect
+	go.opentelemetry.io/otel v1.4.1 // indirect
+	go.opentelemetry.io/otel/metric v0.27.0 // indirect
+	go.opentelemetry.io/otel/trace v1.4.1 // indirect
+	go.uber.org/atomic v1.9.0 // indirect
+	go.uber.org/multierr v1.7.0 // indirect
+	golang.org/x/lint v0.0.0-20201208152925-83fdc39ff7b5 // indirect
+	golang.org/x/mod v0.4.2 // indirect
+	golang.org/x/net v0.0.0-20210813160813-60bc85c4be6d // indirect
+	golang.org/x/oauth2 v0.0.0-20210514164344-f6687ab2804c // indirect
+	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c // indirect
+	golang.org/x/sys v0.0.0-20220114195835-da31bd327af9 // indirect
+	golang.org/x/text v0.3.7 // indirect
+	golang.org/x/tools v0.1.5 // indirect
+	golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 // indirect
+	google.golang.org/api v0.46.0 // indirect
+	google.golang.org/appengine v1.6.7 // indirect
+	google.golang.org/genproto v0.0.0-20211208223120-3a66f561d7aa // indirect
+	google.golang.org/grpc v1.44.0 // indirect
+	google.golang.org/protobuf v1.27.1 // indirect
+)