[exporter/datasetexporter]: Add configuration options server_host and retry_shutdown_timeout

.chloggen/dataset-exporter-make-shutdown-timeout-configurable.yaml

@@ -0,0 +1,20 @@
+# Use this changelog template to create an entry for release notes.
+# If your change doesn't affect end users, such as a test fix or a tooling change,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: 'enhancement'
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: datasetexporter
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: "Make duration of shutdown procedure configurable to minimise data losses."
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [24415]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:

.chloggen/dataset-exporter-populate-serverhost-field.yaml

@@ -0,0 +1,20 @@
+# Use this changelog template to create an entry for release notes.
+# If your change doesn't affect end users, such as a test fix or a tooling change,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: 'enhancement'
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: datasetexporter
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: "Make sure serverHost field is correctly and always populated on the DataSet events. For more information and available configuration options, please refer to the plugin readme file."
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [20660, 24415]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:

cmd/configschema/go.mod

@@ -554,7 +554,7 @@ require (
 	github.com/relvacode/iso8601 v1.3.0 // indirect
 	github.com/rs/cors v1.9.0 // indirect
 	github.com/scaleway/scaleway-sdk-go v1.0.0-beta.15 // indirect
-	github.com/scalyr/dataset-go v0.0.9 // indirect
+	github.com/scalyr/dataset-go v0.12.1 // indirect
 	github.com/seccomp/libseccomp-golang v0.9.2-0.20220502022130-f33da4d89646 // indirect
 	github.com/secure-systems-lab/go-securesystemslib v0.5.0 // indirect
 	github.com/segmentio/asm v1.2.0 // indirect

cmd/otelcontribcol/go.mod

@@ -571,7 +571,7 @@ require (
 	github.com/rs/cors v1.9.0 // indirect
 	github.com/samber/lo v1.37.0 // indirect
 	github.com/scaleway/scaleway-sdk-go v1.0.0-beta.15 // indirect
-	github.com/scalyr/dataset-go v0.0.9 // indirect
+	github.com/scalyr/dataset-go v0.12.1 // indirect
 	github.com/seccomp/libseccomp-golang v0.9.2-0.20220502022130-f33da4d89646 // indirect
 	github.com/secure-systems-lab/go-securesystemslib v0.5.0 // indirect
 	github.com/segmentio/asm v1.2.0 // indirect

exporter/datasetexporter/README.md

@@ -25,6 +25,24 @@ See the [Getting Started](https://app.scalyr.com/help/getting-started) guide.
 
 If you do not want to specify `api_key` in the file, you can use the [builtin functionality](https://opentelemetry.io/docs/collector/configuration/#configuration-environment-variables) and use `api_key: ${env:DATASET_API_KEY}`.
 
+### Server Host Settings
+
+Specifying the server host is crucial for ensuring the correct functionality of DataSet. 
+DataSet expects the server host value to be provided in the `serverHost` attribute. 
+If the server host value is stored in a different attribute, you can use the [resourceprocessor](https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/main/processor/resourceprocessor/README.md) or [attributesprocessor](https://github.com/open-telemetry/opentelemetry-collector-contrib/tree/main/processor/attributesprocessor) to copy it into the `serverHost` attribute.
+
+You can also utilize the `server_host` settings (described below) to populate the serverHost attribute with different values.
+
+The process of populating the serverHost attribute works as follows:
+
+* If the `serverHost` attribute is specified and not empty in the log or trace, then it is used.
+* If the `serverHost` attribute is specified and not empty in the [resource](https://opentelemetry.io/docs/specs/otel/resource/sdk/), then it is used.
+* If the `host.name` attribute is specified and not empty in the [resource](https://opentelemetry.io/docs/specs/otel/resource/sdk/), then it is used.
+* If the `server_host.server_host` setting is specified and not empty, then it is used.
+* If `server_host.use_host_name` setting is set to `true`, the `hostname` of the node is used.
+
+Make sure to provide the appropriate server host value in the `serverHost` attribute to ensure the proper functionality of DataSet and accurate handling of events.
+
 ### Optional Settings
 
 - `buffer`:
@@ -33,17 +51,31 @@ If you do not want to specify `api_key` in the file, you can use the [builtin fu
   - `retry_initial_interval` (default = 5s): Time to wait after the first failure before retrying.
   - `retry_max_interval` (default = 30s): Is the upper bound on backoff.
   - `retry_max_elapsed_time` (default = 300s): Is the maximum amount of time spent trying to send a buffer.
+  - `retry_shutdown_timeout` (default = 30s): The maximum time for which it will try to send data to the DataSet during shutdown. This value should be shorter than container's grace period.
 - `logs`:
-  - `export_resource_info_on_event` (default = false): Include resource info to DataSet Event while exporting Logs. This is especially useful when reducing DataSet billable log volume.
-  - `export_scope_info_on_event` (default = false): Include LogRecord scope information (if available) on the DataSet event.
-  - `decompose_complex_message_field` (default = true): Set this to false to disable decomposing complex body / message field types (e.g. a map) into separate fields.
+    - `export_scope_info_on_event` (default = false): Include LogRecord scope information (if available) on the DataSet event.
+    - `decompose_complex_message_field` (default = true): Set this to false to disable decomposing complex body / message field types (e.g. a map) into separate fields.
+- `server_host`:
+  - `server_host` (default = ''): Specifies the server host to be used for the events.
+  - `use_hostname` (default = true): Determines whether the `hostname` of the node should be used as the server host for the events. When set to `true`, the node's `hostname` is automatically used.
 - `retry_on_failure`: See [retry_on_failure](https://github.com/open-telemetry/opentelemetry-collector/blob/main/exporter/exporterhelper/README.md)
 - `sending_queue`: See [sending_queue](https://github.com/open-telemetry/opentelemetry-collector/blob/main/exporter/exporterhelper/README.md)
 - `timeout`: See [timeout](https://github.com/open-telemetry/opentelemetry-collector/blob/main/exporter/exporterhelper/README.md)
 
 ### Example
 
 ```yaml
+processors:
+  attributes:
+    - key: serverHost
+      action: insert
+      from_attribute: container_id
+  resource:
+    attributes:
+      - key: serverHost
+        from_attribute: node_id
+        action: insert      
+
 exporters:
   dataset/logs:
     # DataSet API URL, https://app.eu.scalyr.com for DataSet EU instance
@@ -55,7 +87,15 @@ exporters:
       max_lifetime: 5s
       # Group data based on these attributes
       group_by:
-        - attributes.container_id
+        - container_id
+      # try to send data to the DataSet for at most 30s during shutdown
+      retry_shutdown_timeout: 30s
+    server_host:
+      # If the serverHost attribute is not specified or empty,
+      # use the value from the env variable SERVER_HOST
+      server_host: ${env:SERVER_HOST}
+      # If server_host is not set, use the hostname value
+      use_hostname: true
 
   dataset/traces:
     # DataSet API URL, https://app.eu.scalyr.com for DataSet EU instance
@@ -71,7 +111,7 @@ service:
   pipelines:
     logs:
       receivers: [otlp]
-      processors: [batch]
+      processors: [batch, attributes]
       # add dataset among your exporters
       exporters: [dataset/logs]
     traces:
@@ -80,3 +120,25 @@ service:
       # add dataset among your exporters
       exporters: [dataset/traces]
 ```
+
+## Examples
+
+### Handling `serverHost` Attribute
+
+Based on the given configuration and scenarios, here's the expected behavior:
+
+1. Resource: `{'node_id:' 'node-pay-01', 'host.name': 'host-pay-01'}`, Log: `{'container_id': 'cont-pay-01'}`, Env: `SERVER_HOST='server-pay-01'`, Hostname: `ip-172-31-27-19`
+   * Since the attribute `container_id` is set, `attributesprocessor` will copy this value to the `serverHost`.
+   * Used `serverHost` will be `cont-pay-01`.
+2. Resource: `{'node_id': 'node-pay-01', 'host.name': 'host-pay-01'}`, Log: `{'attribute.foo': 'Bar'}`, Env: `SERVER_HOST='server-pay-01'`, Hostname: `ip-172-31-27-19`
+   * Since the resource attribute `node_id` is set, `resourceprocessor` will copy this value to the `serverHost`. 
+   * Used `serverHost` will be `node-pay-01`.
+3. Resource: `{'host.name': 'host-pay-01'}`, Log: `{'attribute.foo': 'Bar'}`, Env: `SERVER_HOST='server-pay-01'`, Hostname: `ip-172-31-27-19`
+   * Since the resource attribute `host.name` is set, it will be used.
+   * Used `serverHost` will be `host-pay-01`.
+4. Resource: `{}`, Log: `{'attribute.foo': 'Bar'}`, Env: `SERVER_HOST='server-pay-01'`, Hostname: `ip-172-31-27-19`
+   * Since the attribute `container_id` is not set, the value from the environmental variable `SERVER_HOST`  will be copied to the `serverHost`.
+   * Used `serverHost` will be `server-pay-01`.
+5. Resource: `{}`, Log: `{'attribute.foo': 'Bar'}`, Env: `SERVER_HOST=''`, Hostname: `ip-172-31-27-19`
+   * Since the attribute `container_id` is not set and the environmental variable `SERVER_HOST` is empty, the `hostname` of the node (`ip-172-31-27-19`) will be used as the fallback value for `serverHost`.
+   * Used `serverHost` will be `ip-172-31-27-19`.

exporter/datasetexporter/config.go

@@ -11,6 +11,7 @@ import (
 	"github.com/scalyr/dataset-go/pkg/buffer"
 	"github.com/scalyr/dataset-go/pkg/buffer_config"
 	datasetConfig "github.com/scalyr/dataset-go/pkg/config"
+	"github.com/scalyr/dataset-go/pkg/server_host_config"
 	"go.opentelemetry.io/collector/config/configopaque"
 	"go.opentelemetry.io/collector/confmap"
 	"go.opentelemetry.io/collector/exporter/exporterhelper"
@@ -58,13 +59,15 @@ const bufferMaxLifetime = 5 * time.Second
 const bufferRetryInitialInterval = 5 * time.Second
 const bufferRetryMaxInterval = 30 * time.Second
 const bufferRetryMaxElapsedTime = 300 * time.Second
+const bufferRetryShutdownTimeout = 30 * time.Second
 
 type BufferSettings struct {
 	MaxLifetime          time.Duration `mapstructure:"max_lifetime"`
 	GroupBy              []string      `mapstructure:"group_by"`
 	RetryInitialInterval time.Duration `mapstructure:"retry_initial_interval"`
 	RetryMaxInterval     time.Duration `mapstructure:"retry_max_interval"`
 	RetryMaxElapsedTime  time.Duration `mapstructure:"retry_max_elapsed_time"`
+	RetryShutdownTimeout time.Duration `mapstructure:"retry_shutdown_timeout"`
 }
 
 // newDefaultBufferSettings returns the default settings for BufferSettings.
@@ -75,6 +78,20 @@ func newDefaultBufferSettings() BufferSettings {
 		RetryInitialInterval: bufferRetryInitialInterval,
 		RetryMaxInterval:     bufferRetryMaxInterval,
 		RetryMaxElapsedTime:  bufferRetryMaxElapsedTime,
+		RetryShutdownTimeout: bufferRetryShutdownTimeout,
+	}
+}
+
+type ServerHostSettings struct {
+	UseHostName bool   `mapstructure:"use_hostname"`
+	ServerHost  string `mapstructure:"server_host"`
+}
+
+// newDefaultBufferSettings returns the default settings for BufferSettings.
+func newDefaultServerHostSettings() ServerHostSettings {
+	return ServerHostSettings{
+		UseHostName: true,
+		ServerHost:  "",
 	}
 }
 
@@ -84,6 +101,7 @@ type Config struct {
 	BufferSettings                 `mapstructure:"buffer"`
 	TracesSettings                 `mapstructure:"traces"`
 	LogsSettings                   `mapstructure:"logs"`
+	ServerHostSettings             `mapstructure:"server_host"`
 	exporterhelper.RetrySettings   `mapstructure:"retry_on_failure"`
 	exporterhelper.QueueSettings   `mapstructure:"sending_queue"`
 	exporterhelper.TimeoutSettings `mapstructure:"timeout"`
@@ -116,11 +134,12 @@ func (c *Config) String() string {
 	s := ""
 	s += fmt.Sprintf("%s: %s; ", "DatasetURL", c.DatasetURL)
 	s += fmt.Sprintf("%s: %+v; ", "BufferSettings", c.BufferSettings)
+	s += fmt.Sprintf("%s: %+v; ", "LogsSettings", c.LogsSettings)
 	s += fmt.Sprintf("%s: %+v; ", "TracesSettings", c.TracesSettings)
+	s += fmt.Sprintf("%s: %+v; ", "ServerHostSettings", c.ServerHostSettings)
 	s += fmt.Sprintf("%s: %+v; ", "RetrySettings", c.RetrySettings)
 	s += fmt.Sprintf("%s: %+v; ", "QueueSettings", c.QueueSettings)
-	s += fmt.Sprintf("%s: %+v; ", "TimeoutSettings", c.TimeoutSettings)
-	s += fmt.Sprintf("%s: %+v", "LogsSettings", c.LogsSettings)
+	s += fmt.Sprintf("%s: %+v", "TimeoutSettings", c.TimeoutSettings)
 
 	return s
 }
@@ -144,16 +163,23 @@ func (c *Config) convert() (*ExporterConfig, error) {
 					RetryMaxElapsedTime:      c.BufferSettings.RetryMaxElapsedTime,
 					RetryMultiplier:          backoff.DefaultMultiplier,
 					RetryRandomizationFactor: backoff.DefaultRandomizationFactor,
+					RetryShutdownTimeout:     c.BufferSettings.RetryShutdownTimeout,
+				},
+				ServerHostSettings: server_host_config.DataSetServerHostSettings{
+					UseHostName: c.ServerHostSettings.UseHostName,
+					ServerHost:  c.ServerHostSettings.ServerHost,
 				},
 			},
-			tracesSettings: c.TracesSettings,
-			logsSettings:   c.LogsSettings,
+			tracesSettings:     c.TracesSettings,
+			logsSettings:       c.LogsSettings,
+			serverHostSettings: c.ServerHostSettings,
 		},
 		nil
 }
 
 type ExporterConfig struct {
-	datasetConfig  *datasetConfig.DataSetConfig
-	tracesSettings TracesSettings
-	logsSettings   LogsSettings
+	datasetConfig      *datasetConfig.DataSetConfig
+	tracesSettings     TracesSettings
+	logsSettings       LogsSettings
+	serverHostSettings ServerHostSettings
 }

exporter/datasetexporter/config_test.go

@@ -106,14 +106,18 @@ func TestConfigString(t *testing.T) {
 			MaxLifetime: 123,
 			GroupBy:     []string{"field1", "field2"},
 		},
-		TracesSettings:  TracesSettings{},
+		TracesSettings: TracesSettings{},
+		ServerHostSettings: ServerHostSettings{
+			ServerHost:  "foo-bar",
+			UseHostName: false,
+		},
 		RetrySettings:   exporterhelper.NewDefaultRetrySettings(),
 		QueueSettings:   exporterhelper.NewDefaultQueueSettings(),
 		TimeoutSettings: exporterhelper.NewDefaultTimeoutSettings(),
 	}
 
 	assert.Equal(t,
-		"DatasetURL: https://example.com; BufferSettings: {MaxLifetime:123ns GroupBy:[field1 field2] RetryInitialInterval:0s RetryMaxInterval:0s RetryMaxElapsedTime:0s}; TracesSettings: {}; RetrySettings: {Enabled:true InitialInterval:5s RandomizationFactor:0.5 Multiplier:1.5 MaxInterval:30s MaxElapsedTime:5m0s}; QueueSettings: {Enabled:true NumConsumers:10 QueueSize:1000 StorageID:<nil>}; TimeoutSettings: {Timeout:5s}; LogsSettings: {ExportResourceInfo:false ExportScopeInfo:false DecomposeComplexMessageField:false}",
+		"DatasetURL: https://example.com; BufferSettings: {MaxLifetime:123ns GroupBy:[field1 field2] RetryInitialInterval:0s RetryMaxInterval:0s RetryMaxElapsedTime:0s RetryShutdownTimeout:0s}; LogsSettings: {ExportResourceInfo:false ExportScopeInfo:false DecomposeComplexMessageField:false}; TracesSettings: {}; ServerHostSettings: {UseHostName:false ServerHost:foo-bar}; RetrySettings: {Enabled:true InitialInterval:5s RandomizationFactor:0.5 Multiplier:1.5 MaxInterval:30s MaxElapsedTime:5m0s}; QueueSettings: {Enabled:true NumConsumers:10 QueueSize:1000 StorageID:<nil>}; TimeoutSettings: {Timeout:5s}",
 		config.String(),
 	)
 }

exporter/datasetexporter/datasetexporter.go

@@ -15,6 +15,7 @@ import (
 	"github.com/scalyr/dataset-go/pkg/api/add_events"
 	"github.com/scalyr/dataset-go/pkg/client"
 	"go.opentelemetry.io/collector/exporter"
+	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.uber.org/zap"
 	"golang.org/x/time/rate"
 )
@@ -25,6 +26,7 @@ type DatasetExporter struct {
 	logger      *zap.Logger
 	session     string
 	exporterCfg *ExporterConfig
+	serverHost  string
 }
 
 func newDatasetExporter(entity string, config *Config, set exporter.CreateSettings) (*DatasetExporter, error) {
@@ -63,6 +65,7 @@ func newDatasetExporter(entity string, config *Config, set exporter.CreateSettin
 		session:     uuid.New().String(),
 		logger:      logger,
 		exporterCfg: exporterCfg,
+		serverHost:  client.ServerHost(),
 	}, nil
 }
 
@@ -117,3 +120,31 @@ func updateWithPrefixedValues(target map[string]interface{}, prefix string, sepa
 
 	}
 }
+
+func inferServerHost(
+	resource pcommon.Resource,
+	attrs map[string]interface{},
+	serverHost string,
+) string {
+	// first use value from the attribute serverHost
+	valA, okA := attrs[add_events.AttrServerHost]
+	if okA {
+		host := fmt.Sprintf("%v", valA)
+		if len(host) > 0 {
+			return host
+		}
+	}
+
+	// then use value from resource attributes - serverHost and host.name
+	for _, resKey := range []string{add_events.AttrServerHost, "host.name"} {
+		valR, okR := resource.Attributes().Get(resKey)
+		if okR {
+			host := valR.AsString()
+			if len(host) > 0 {
+				return host
+			}
+		}
+	}
+
+	return serverHost
+}