[exporter/sumologic]: add sticky session (#33011)

**Description:**

Adds support for sticky session in order to better support AWS LB. This
code is moved from Sumo Logic repository

**Link to tracking Issue:**

#32315

**Testing:**

Tested manually

**Documentation:**

N/A

---------

Signed-off-by: Dominik Rosiek <[email protected]>

.chloggen/drosiek-sumo-sticky-session.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# 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: sumologicexporter
+
+# A brief description of the change. Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: add sticky session support
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [32315]
+
+# (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:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user]

exporter/sumologicexporter/README.md

@@ -147,6 +147,10 @@ exporters:
  # maximum connection timeout is 55s, default = 5s
  timeout: <timeout>
 
+ # defines if sticky session support is enable.
+ # default=false
+ sticky_session_enabled: {true, false}
+
  # for below described queueing and retry related configuration please refer to:
  # https://github.com/open-telemetry/opentelemetry-collector/blob/main/exporter/exporterhelper/README.md#configuration
 

exporter/sumologicexporter/config.go

@@ -64,6 +64,10 @@ type Config struct {
  SourceHost string `mapstructure:"source_host"`
  // Name of the client
  Client string `mapstructure:"client"`
+
+ // StickySessionEnabled defines if sticky session support is enable.
+ // By default this is false.
+ StickySessionEnabled bool `mapstructure:"sticky_session_enabled"`
 }
 
 // createDefaultClientConfig returns default http client settings
@@ -132,8 +136,10 @@ const (
  DefaultClient string = "otelcol"
  // DefaultLogKey defines default LogKey value
  DefaultLogKey string = "log"
- // DefaultGraphiteTemplate defines default template for Graphite
- DefaultGraphiteTemplate string = "%{_metric_}"
+ // DefaultDropRoutingAttribute defines default DropRoutingAttribute
+ DefaultDropRoutingAttribute string = ""
+ // DefaultStickySessionEnabled defines default StickySessionEnabled value
+ DefaultStickySessionEnabled bool = false
 )
 
 func (cfg *Config) Validate() error {

exporter/sumologicexporter/exporter.go

@@ -54,6 +54,9 @@ type sumologicexporter struct {
  foundSumologicExtension bool
  sumologicExtension *sumologicextension.SumologicExtension
 
+ stickySessionCookieLock sync.RWMutex
+ stickySessionCookie string
+
  id component.ID
 }
 
@@ -289,6 +292,8 @@ func (se *sumologicexporter) pushLogsData(ctx context.Context, ld plog.Logs) err
  metricsURL,
  logsURL,
  tracesURL,
+ se.StickySessionCookie,
+ se.SetStickySessionCookie,
  se.id,
  )
 
@@ -366,6 +371,8 @@ func (se *sumologicexporter) pushMetricsData(ctx context.Context, md pmetric.Met
  metricsURL,
  logsURL,
  tracesURL,
+ se.StickySessionCookie,
+ se.SetStickySessionCookie,
  se.id,
  )
 
@@ -406,6 +413,27 @@ func (se *sumologicexporter) handleUnauthorizedErrors(ctx context.Context, errs
  }
 }
 
+func (se *sumologicexporter) StickySessionCookie() string {
+ if se.foundSumologicExtension {
+ return se.sumologicExtension.StickySessionCookie()
+ }
+
+ se.stickySessionCookieLock.RLock()
+ defer se.stickySessionCookieLock.RUnlock()
+ return se.stickySessionCookie
+}
+
+func (se *sumologicexporter) SetStickySessionCookie(stickySessionCookie string) {
+ if se.foundSumologicExtension {
+ se.sumologicExtension.SetStickySessionCookie(stickySessionCookie)
+ return
+ }
+
+ se.stickySessionCookieLock.Lock()
+ se.stickySessionCookie = stickySessionCookie
+ se.stickySessionCookieLock.Unlock()
+}
+
 // get the destination url for a given signal type
 // this mostly adds signal-specific suffixes if the format is otlp
 func getSignalURL(oCfg *Config, endpointURL string, signal component.DataType) (string, error) {

exporter/sumologicexporter/factory.go

@@ -40,9 +40,10 @@ func createDefaultConfig() component.Config {
  SourceHost: DefaultSourceHost,
  Client: DefaultClient,
 
- ClientConfig: createDefaultClientConfig(),
- BackOffConfig: configretry.NewDefaultBackOffConfig(),
- QueueSettings: qs,
+ ClientConfig: createDefaultClientConfig(),
+ BackOffConfig: configretry.NewDefaultBackOffConfig(),
+ QueueSettings: qs,
+ StickySessionEnabled: DefaultStickySessionEnabled,
  }
 }
 

exporter/sumologicexporter/sender.go

@@ -110,16 +110,18 @@ func (b *bodyBuilder) toCountingReader() *countingReader {
 }
 
 type sender struct {
- logger *zap.Logger
- config *Config
- client *http.Client
- filter filter
- sources sourceFormats
- prometheusFormatter prometheusFormatter
- dataURLMetrics string
- dataURLLogs string
- dataURLTraces string
- id component.ID
+ logger *zap.Logger
+ config *Config
+ client *http.Client
+ filter filter
+ sources sourceFormats
+ prometheusFormatter prometheusFormatter
+ dataURLMetrics string
+ dataURLLogs string
+ dataURLTraces string
+ stickySessionCookieFunc func() string
+ setStickySessionCookieFunc func(string)
+ id component.ID
 }
 
 const (
@@ -140,6 +142,7 @@ const (
  contentTypeLogs string = "application/x-www-form-urlencoded"
  contentTypePrometheus string = "application/vnd.sumologic.prometheus"
  contentTypeOTLP string = "application/x-protobuf"
+ stickySessionKey string = "AWSALB"
 )
 
 func newSender(
@@ -152,19 +155,23 @@ func newSender(
  metricsURL string,
  logsURL string,
  tracesURL string,
+ stickySessionCookieFunc func() string,
+ setStickySessionCookieFunc func(string),
  id component.ID,
 ) *sender {
  return &sender{
- logger: logger,
- config: cfg,
- client: cl,
- filter: f,
- sources: s,
- prometheusFormatter: pf,
- dataURLMetrics: metricsURL,
- dataURLLogs: logsURL,
- dataURLTraces: tracesURL,
- id: id,
+ logger: logger,
+ config: cfg,
+ client: cl,
+ filter: f,
+ sources: s,
+ prometheusFormatter: pf,
+ dataURLMetrics: metricsURL,
+ dataURLLogs: logsURL,
+ dataURLTraces: tracesURL,
+ stickySessionCookieFunc: stickySessionCookieFunc,
+ setStickySessionCookieFunc: setStickySessionCookieFunc,
+ id: id,
  }
 }
 
@@ -181,6 +188,10 @@ func (s *sender) send(ctx context.Context, pipeline PipelineType, reader *counti
  return err
  }
 
+ if s.config.StickySessionEnabled {
+ s.addStickySessionCookie(req)
+ }
+
  s.logger.Debug("Sending data",
  zap.String("pipeline", string(pipeline)),
  zap.Any("headers", req.Header),
@@ -200,6 +211,10 @@ func (s *sender) send(ctx context.Context, pipeline PipelineType, reader *counti
 }
 
 func (s *sender) handleReceiverResponse(resp *http.Response) error {
+ if s.config.StickySessionEnabled {
+ s.updateStickySessionCookie(resp)
+ }
+
  // API responds with a 200 or 204 with ConentLength set to 0 when all data
  // has been successfully ingested.
  if resp.ContentLength == 0 && (resp.StatusCode == 200 || resp.StatusCode == 204) {
@@ -684,3 +699,28 @@ func (s *sender) recordMetrics(duration time.Duration, count int64, req *http.Re
  s.logger.Debug("error for recording metric for sent request", zap.Error(err))
  }
 }
+
+func (s *sender) addStickySessionCookie(req *http.Request) {
+ currectCookieValue := s.stickySessionCookieFunc()
+ if currectCookieValue != "" {
+ cookie := &http.Cookie{
+ Name: stickySessionKey,
+ Value: currectCookieValue,
+ }
+ req.AddCookie(cookie)
+ }
+}
+
+func (s *sender) updateStickySessionCookie(resp *http.Response) {
+ cookies := resp.Cookies()
+ if len(cookies) > 0 {
+ for _, cookie := range cookies {
+ if cookie.Name == stickySessionKey {
+ if cookie.Value != s.stickySessionCookieFunc() {
+ s.setStickySessionCookieFunc(cookie.Value)
+ }
+ return
+ }
+ }
+ }
+}

exporter/sumologicexporter/sender_test.go

@@ -114,6 +114,8 @@ func prepareSenderTest(t *testing.T, compression configcompression.Type, cb []fu
  testServer.URL,
  testServer.URL,
  testServer.URL,
+ func() string { return "" },
+ func(string) {},
  component.ID{},
  ),
  }