Organize chronicle exporter by http and grpc

exporter/chronicleexporter/factory_test.go

@@ -38,6 +38,6 @@ func Test_createDefaultConfig(t *testing.T) {
 		BatchRequestSizeLimitHTTP: defaultBatchRequestSizeLimitHTTP,
 	}
 
-	actual := createDefaultConfig()
+	actual := NewFactory().CreateDefaultConfig()
 	require.Equal(t, expectedCfg, actual)
 }

exporter/chronicleexporter/grpc_exporter.go

@@ -19,6 +19,7 @@ import (
 	"fmt"
 	"net/http"
 
+	"github.com/observiq/bindplane-otel-collector/exporter/chronicleexporter/internal/marshal"
 	"github.com/observiq/bindplane-otel-collector/exporter/chronicleexporter/protos/api"
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/consumer"
@@ -40,15 +41,24 @@ type grpcExporter struct {
 	cfg        *Config
 	set        component.TelemetrySettings
 	exporterID string
-	marshaler  *protoMarshaler
+	marshaler  *marshal.GRPC
 
 	client  api.IngestionServiceV2Client
 	conn    *grpc.ClientConn
 	metrics *hostMetricsReporter
 }
 
 func newGRPCExporter(cfg *Config, params exporter.Settings) (*grpcExporter, error) {
-	marshaler, err := newProtoMarshaler(*cfg, params.TelemetrySettings)
+	marshaler, err := marshal.NewGRPC(marshal.Config{
+		CustomerID:            cfg.CustomerID,
+		Namespace:             cfg.Namespace,
+		LogType:               cfg.LogType,
+		RawLogField:           cfg.RawLogField,
+		OverrideLogType:       cfg.OverrideLogType,
+		IngestionLabels:       cfg.IngestionLabels,
+		BatchRequestSizeLimit: cfg.BatchRequestSizeLimitGRPC,
+		BatchLogCountLimit:    cfg.BatchLogCountLimitGRPC,
+	}, params.TelemetrySettings)
 	if err != nil {
 		return nil, fmt.Errorf("create proto marshaler: %w", err)
 	}
@@ -107,7 +117,7 @@ func (exp *grpcExporter) Shutdown(context.Context) error {
 }
 
 func (exp *grpcExporter) ConsumeLogs(ctx context.Context, ld plog.Logs) error {
-	payloads, err := exp.marshaler.MarshalRawLogs(ctx, ld)
+	payloads, err := exp.marshaler.MarshalLogs(ctx, ld)
 	if err != nil {
 		return fmt.Errorf("marshal logs: %w", err)
 	}

exporter/chronicleexporter/hostmetrics.go

@@ -22,6 +22,7 @@ import (
 	"time"
 
 	"github.com/google/uuid"
+	"github.com/observiq/bindplane-otel-collector/exporter/chronicleexporter/internal/ccid"
 	"github.com/observiq/bindplane-otel-collector/exporter/chronicleexporter/protos/api"
 	"github.com/shirou/gopsutil/v3/process"
 	"go.opentelemetry.io/collector/component"
@@ -68,9 +69,9 @@ func newHostMetricsReporter(cfg *Config, set component.TelemetrySettings, export
 	return &hostMetricsReporter{
 		set:        set,
 		send:       send,
+		startTime:  now,
 		agentID:    agentID[:],
 		exporterID: exporterID,
-		startTime:  now,
 		customerID: customerID[:],
 		namespace:  cfg.Namespace,
 		stats: &api.AgentStatsEvent{
@@ -108,14 +109,21 @@ func (hmr *hostMetricsReporter) start() {
 	}()
 }
 
+func (hmr *hostMetricsReporter) shutdown() {
+	if hmr.cancel != nil {
+		hmr.cancel()
+		hmr.wg.Wait()
+	}
+}
+
 func (hmr *hostMetricsReporter) getAndReset() *api.BatchCreateEventsRequest {
 	hmr.mutex.Lock()
 	defer hmr.mutex.Unlock()
 
 	now := timestamppb.Now()
 	batchID := uuid.New()
 	source := &api.EventSource{
-		CollectorId: chronicleCollectorID[:],
+		CollectorId: ccid.ChronicleCollectorID[:],
 		Namespace:   hmr.namespace,
 		CustomerId:  hmr.customerID,
 	}
@@ -143,13 +151,6 @@ func (hmr *hostMetricsReporter) getAndReset() *api.BatchCreateEventsRequest {
 	return request
 }
 
-func (hmr *hostMetricsReporter) shutdown() {
-	if hmr.cancel != nil {
-		hmr.cancel()
-		hmr.wg.Wait()
-	}
-}
-
 func (hmr *hostMetricsReporter) resetStats() {
 	hmr.stats = &api.AgentStatsEvent{
 		ExporterStats: []*api.ExporterStats{

exporter/chronicleexporter/http_exporter.go

@@ -23,6 +23,7 @@ import (
 	"io"
 	"net/http"
 
+	"github.com/observiq/bindplane-otel-collector/exporter/chronicleexporter/internal/marshal"
 	"github.com/observiq/bindplane-otel-collector/exporter/chronicleexporter/protos/api"
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/consumer"
@@ -40,12 +41,26 @@ const httpScope = "https://www.googleapis.com/auth/cloud-platform"
 type httpExporter struct {
 	cfg       *Config
 	set       component.TelemetrySettings
-	marshaler *protoMarshaler
+	marshaler *marshal.HTTP
 	client    *http.Client
 }
 
 func newHTTPExporter(cfg *Config, params exporter.Settings) (*httpExporter, error) {
-	marshaler, err := newProtoMarshaler(*cfg, params.TelemetrySettings)
+	marshaler, err := marshal.NewHTTP(marshal.HTTPConfig{
+		Config: marshal.Config{
+			CustomerID:            cfg.CustomerID,
+			Namespace:             cfg.Namespace,
+			LogType:               cfg.LogType,
+			RawLogField:           cfg.RawLogField,
+			OverrideLogType:       cfg.OverrideLogType,
+			IngestionLabels:       cfg.IngestionLabels,
+			BatchRequestSizeLimit: cfg.BatchRequestSizeLimitHTTP,
+			BatchLogCountLimit:    cfg.BatchLogCountLimitHTTP,
+		},
+		Project:   cfg.Project,
+		Location:  cfg.Location,
+		Forwarder: cfg.Forwarder,
+	}, params.TelemetrySettings)
 	if err != nil {
 		return nil, fmt.Errorf("create proto marshaler: %w", err)
 	}
@@ -79,7 +94,7 @@ func (exp *httpExporter) Shutdown(context.Context) error {
 }
 
 func (exp *httpExporter) ConsumeLogs(ctx context.Context, ld plog.Logs) error {
-	payloads, err := exp.marshaler.MarshalRawLogsForHTTP(ctx, ld)
+	payloads, err := exp.marshaler.MarshalLogs(ctx, ld)
 	if err != nil {
 		return fmt.Errorf("marshal logs: %w", err)
 	}

exporter/chronicleexporter/internal/ccid/ccid.go

@@ -0,0 +1,23 @@
+// Copyright observIQ, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// Package ccid exposes a hardcoded UUID that is used to identify bindplane agents in Chronicle.
+package ccid
+
+import (
+	"github.com/google/uuid"
+)
+
+// ChronicleCollectorID is a specific collector ID for Chronicle. It's used to identify bindplane agents in Chronicle.
+var ChronicleCollectorID = uuid.MustParse("aaaa1111-aaaa-1111-aaaa-1111aaaa1111")