[draft, confgenerator] Configure exporter to support Log Buffering strategies in Otel Logging.

apps/otlp.go

@@ -145,7 +145,7 @@ func (r ReceiverOTLP) Pipelines(ctx context.Context) ([]otel.ReceiverPipeline, e
 		ExporterTypes: map[string]otel.ExporterType{
 			"metrics": receiverPipelineType,
 			"traces":  otel.OTel,
-			"logs":    otel.Logging,
+			"logs":    otel.LoggingPersistentQueue,
 		},
 		Receiver: otel.Component{
 			Type: "otlp",

confgenerator/confgenerator.go

@@ -57,15 +57,34 @@ func googleCloudExporter(userAgent string, instrumentationLabels bool, serviceRe
 	}
 }
 
-func googleCloudLoggingExporter() otel.Component {
-	return otel.Component{
-		Type: "googlecloud",
-		Config: map[string]interface{}{
-			// Set to mirror the 60s max limit of default retry window in Google Cloud Logging apiv2 go client :
-			// https://github.com/googleapis/google-cloud-go/blob/logging/v1.4.2/logging/apiv2/logging_client.go#L78-L90
-			"timeout": "60s",
+func googleCloudLoggingExporter(persistent_queue bool) otel.Component {
+	config := map[string]interface{}{
+		// Set to mirror the 60s max limit of default retry window in Google Cloud Logging apiv2 go client :
+		// https://github.com/googleapis/google-cloud-go/blob/logging/v1.4.2/logging/apiv2/logging_client.go#L78-L90
+		"timeout": "60s",
+		"sending_queue": map[string]interface{}{
+			"enabled":           true,
+			"block_on_overflow": true,
+			"wait_for_result":   true,
+			"batch": map[string]interface{}{
+				"flush_timeout": "200ms",
+				"min_size":      1,
+				"max_size":      1000,
+				"sizer":         "items",
+			},
 		},
 	}
+
+	if persistent_queue {
+		if sq, ok := config["sending_queue"].(map[string]interface{}); ok {
+			sq["storage"] = dbStorageExtensionID()
+		}
+	}
+
+	return otel.Component{
+		Type:   "googlecloud",
+		Config: config,
+	}
 }
 
 func ConvertPrometheusExporterToOtlpExporter(pipeline otel.ReceiverPipeline, ctx context.Context) otel.ReceiverPipeline {
@@ -152,6 +171,11 @@ func fileStorageExtensionID() string {
 	return "file_storage"
 }
 
+// dbStorageExtensionID returns the file_storage extension used by all receivers and exporters.
+func dbStorageExtensionID() string {
+	return "db_storage"
+}
+
 // fileStorageExtensionConfig returns a configured file_storage extension to be used by all receivers and exporters.
 func fileStorageExtensionConfig(stateDir string) map[string]interface{} {
 	return map[string]interface{}{
@@ -160,6 +184,14 @@ func fileStorageExtensionConfig(stateDir string) map[string]interface{} {
 	}
 }
 
+// dbStorageExtensionConfig returns a configured file_storage extension to be used by all receivers and exporters.
+func dbStorageExtensionConfig(stateDir string) map[string]interface{} {
+	return map[string]interface{}{
+		"driver":     "sqlite",
+		"datasource": fmt.Sprintf("file://%s?_pragma=busy_timeout(10000)&_pragma=journal_mode(WAL)&_pragma=synchronous(NORMAL)", path.Join(stateDir, "db_storage.db")),
+	}
+}
+
 func (uc *UnifiedConfig) getEnabledExtensions(ctx context.Context, stateDir string) map[string]interface{} {
 	extensions := map[string]interface{}{}
 	expOtlpExporter := experimentsFromContext(ctx)["otlp_exporter"]
@@ -168,6 +200,7 @@ func (uc *UnifiedConfig) getEnabledExtensions(ctx context.Context, stateDir stri
 	}
 	if uc.Logging.Service.OTelLogging {
 		extensions["file_storage"] = fileStorageExtensionConfig(stateDir)
+		extensions["db_storage"] = dbStorageExtensionConfig(stateDir)
 	}
 	return extensions
 }
@@ -222,14 +255,11 @@ func (uc *UnifiedConfig) GenerateOtelConfig(ctx context.Context, outDir, stateDi
 					},
 				},
 			},
-			otel.Logging: {
-				Exporter: googleCloudLoggingExporter(),
-				ProcessorsByType: map[string][]otel.Component{
-					// Batching logs improves log export performance.
-					"logs": {
-						otel.BatchProcessor(1000, 1000, "200ms"),
-					},
-				},
+			otel.LoggingPersistentQueue: {
+				Exporter: googleCloudLoggingExporter(true),
+			},
+			otel.LoggingNonPersistentQueue: {
+				Exporter: googleCloudLoggingExporter(false),
 			},
 		},
 	}.Generate(ctx)

confgenerator/logging_receivers.go

@@ -244,7 +244,7 @@ func (r LoggingReceiverFilesMixin) Pipelines(ctx context.Context) ([]otel.Receiv
 			"logs": nil,
 		},
 		ExporterTypes: map[string]otel.ExporterType{
-			"logs": otel.Logging,
+			"logs": otel.LoggingNonPersistentQueue,
 		},
 	}}, nil
 }
@@ -369,7 +369,7 @@ func (r LoggingReceiverSyslog) Pipelines(ctx context.Context) ([]otel.ReceiverPi
 		},
 
 		ExporterTypes: map[string]otel.ExporterType{
-			"logs": otel.Logging,
+			"logs": otel.LoggingPersistentQueue,
 		},
 	}}, nil
 }
@@ -638,7 +638,7 @@ func (r LoggingReceiverWindowsEventLog) Pipelines(ctx context.Context) ([]otel.R
 				"logs": p,
 			},
 			ExporterTypes: map[string]otel.ExporterType{
-				"logs": otel.Logging,
+				"logs": otel.LoggingNonPersistentQueue,
 			},
 		})
 	}
@@ -1028,7 +1028,7 @@ func (r LoggingReceiverSystemd) Pipelines(ctx context.Context) ([]otel.ReceiverP
 		},
 
 		ExporterTypes: map[string]otel.ExporterType{
-			"logs": otel.Logging,
+			"logs": otel.LoggingNonPersistentQueue,
 		},
 	}}, nil
 }

confgenerator/otel/modular.go

@@ -18,6 +18,7 @@ package otel
 import (
 	"context"
 	"fmt"
+	"slices"
 
 	"github.com/GoogleCloudPlatform/ops-agent/internal/platform"
 	yaml "github.com/goccy/go-yaml"
@@ -38,7 +39,8 @@ const (
 	System
 	GMP
 	OTLP
-	Logging
+	LoggingPersistentQueue
+	LoggingNonPersistentQueue
 )
 const (
 	Override ResourceDetectionMode = iota
@@ -52,8 +54,10 @@ func (t ExporterType) Name() string {
 		return ""
 	} else if t == OTel {
 		return "otel"
-	} else if t == Logging {
-		return "logging"
+	} else if t == LoggingPersistentQueue {
+		return "logging_persistent_queue"
+	} else if t == LoggingNonPersistentQueue {
+		return "logging_non_persistent_queue"
 	} else if t == OTLP {
 		return "otlp"
 	} else {
@@ -209,6 +213,7 @@ func (c ModularConfig) Generate(ctx context.Context) (string, error) {
 			extensions[extensionName] = c.Extensions[extensionName]
 			extensionsList = append(extensionsList, extensionName)
 		}
+		slices.Sort(extensionsList)
 		service["extensions"] = extensionsList
 		configMap["extensions"] = extensions
 	}

confgenerator/testdata/goldens/combined-receiver_otlp/golden/linux-gpu/otel.yaml

@@ -7,7 +7,17 @@ exporters:
       service_resource_labels: false
       skip_create_descriptor: true
     user_agent: Google-Cloud-Ops-Agent-Metrics/latest (BuildDistro=build_distro;Platform=linux;ShortName=linux_platform;ShortVersion=linux_platform_version)
-  googlecloud/logging:
+  googlecloud/logging_persistent_queue:
+    sending_queue:
+      batch:
+        flush_timeout: 200ms
+        max_size: 1000
+        min_size: 1
+        sizer: items
+      block_on_overflow: true
+      enabled: true
+      storage: db_storage
+      wait_for_result: true
     timeout: 60s
   googlecloud/otel:
     metric:
@@ -25,10 +35,6 @@ processors:
   agentmetrics/hostmetrics_0:
     blank_label_metrics:
     - system.cpu.utilization
-  batch/googlecloud/logging_logs_0:
-    send_batch_max_size: 1000
-    send_batch_size: 1000
-    timeout: 200ms
   cumulativetodelta/loggingmetrics_4:
     include:
       match_type: strict
@@ -738,11 +744,10 @@ service:
   pipelines:
     logs/logs_otlp_otlp:
       exporters:
-      - googlecloud/logging
+      - googlecloud/logging_persistent_queue
       processors:
       - transform/otlp_0
       - resourcedetection/_global_1
-      - batch/googlecloud/logging_logs_0
       receivers:
       - otlp/otlp
     metrics/default__pipeline_hostmetrics:

confgenerator/testdata/goldens/combined-receiver_otlp/golden/linux/otel.yaml

@@ -7,7 +7,17 @@ exporters:
       service_resource_labels: false
       skip_create_descriptor: true
     user_agent: Google-Cloud-Ops-Agent-Metrics/latest (BuildDistro=build_distro;Platform=linux;ShortName=linux_platform;ShortVersion=linux_platform_version)
-  googlecloud/logging:
+  googlecloud/logging_persistent_queue:
+    sending_queue:
+      batch:
+        flush_timeout: 200ms
+        max_size: 1000
+        min_size: 1
+        sizer: items
+      block_on_overflow: true
+      enabled: true
+      storage: db_storage
+      wait_for_result: true
     timeout: 60s
   googlecloud/otel:
     metric:
@@ -25,10 +35,6 @@ processors:
   agentmetrics/hostmetrics_0:
     blank_label_metrics:
     - system.cpu.utilization
-  batch/googlecloud/logging_logs_0:
-    send_batch_max_size: 1000
-    send_batch_size: 1000
-    timeout: 200ms
   cumulativetodelta/loggingmetrics_4:
     include:
       match_type: strict
@@ -707,11 +713,10 @@ service:
   pipelines:
     logs/logs_otlp_otlp:
       exporters:
-      - googlecloud/logging
+      - googlecloud/logging_persistent_queue
       processors:
       - transform/otlp_0
       - resourcedetection/_global_1
-      - batch/googlecloud/logging_logs_0
       receivers:
       - otlp/otlp
     metrics/default__pipeline_hostmetrics:

confgenerator/testdata/goldens/combined-receiver_otlp/golden/windows-2012/otel.yaml

@@ -7,7 +7,17 @@ exporters:
       service_resource_labels: false
       skip_create_descriptor: true
     user_agent: Google-Cloud-Ops-Agent-Metrics/latest (BuildDistro=build_distro;Platform=windows;ShortName=win_platform;ShortVersion=win_platform_version)
-  googlecloud/logging:
+  googlecloud/logging_persistent_queue:
+    sending_queue:
+      batch:
+        flush_timeout: 200ms
+        max_size: 1000
+        min_size: 1
+        sizer: items
+      block_on_overflow: true
+      enabled: true
+      storage: db_storage
+      wait_for_result: true
     timeout: 60s
   googlecloud/otel:
     metric:
@@ -25,10 +35,6 @@ processors:
   agentmetrics/hostmetrics_0:
     blank_label_metrics:
     - system.cpu.utilization
-  batch/googlecloud/logging_logs_0:
-    send_batch_max_size: 1000
-    send_batch_size: 1000
-    timeout: 200ms
   casttosum/iis_1:
     metrics:
     - agent.googleapis.com/iis/network/transferred_bytes_count
@@ -820,11 +826,10 @@ service:
   pipelines:
     logs/logs_otlp_otlp:
       exporters:
-      - googlecloud/logging
+      - googlecloud/logging_persistent_queue
       processors:
       - transform/otlp_0
       - resourcedetection/_global_1
-      - batch/googlecloud/logging_logs_0
       receivers:
       - otlp/otlp
     metrics/default__pipeline_hostmetrics:

confgenerator/testdata/goldens/combined-receiver_otlp/golden/windows/otel.yaml

@@ -7,7 +7,17 @@ exporters:
       service_resource_labels: false
       skip_create_descriptor: true
     user_agent: Google-Cloud-Ops-Agent-Metrics/latest (BuildDistro=build_distro;Platform=windows;ShortName=win_platform;ShortVersion=win_platform_version)
-  googlecloud/logging:
+  googlecloud/logging_persistent_queue:
+    sending_queue:
+      batch:
+        flush_timeout: 200ms
+        max_size: 1000
+        min_size: 1
+        sizer: items
+      block_on_overflow: true
+      enabled: true
+      storage: db_storage
+      wait_for_result: true
     timeout: 60s
   googlecloud/otel:
     metric:
@@ -25,10 +35,6 @@ processors:
   agentmetrics/hostmetrics_0:
     blank_label_metrics:
     - system.cpu.utilization
-  batch/googlecloud/logging_logs_0:
-    send_batch_max_size: 1000
-    send_batch_size: 1000
-    timeout: 200ms
   casttosum/iis_1:
     metrics:
     - agent.googleapis.com/iis/network/transferred_bytes_count
@@ -820,11 +826,10 @@ service:
   pipelines:
     logs/logs_otlp_otlp:
       exporters:
-      - googlecloud/logging
+      - googlecloud/logging_persistent_queue
       processors:
       - transform/otlp_0
       - resourcedetection/_global_1
-      - batch/googlecloud/logging_logs_0
       receivers:
       - otlp/otlp
     metrics/default__pipeline_hostmetrics: