Merge branch 'main' into refactor/xdg-go

CHANGELOG-developer.next.asciidoc

@@ -99,6 +99,7 @@ The list below covers the major changes between 7.0.0-rc2 and main only.
 - Fix panic when more than 32767 pipeline clients are active. {issue}38197[38197] {pull}38556[38556]
 - Skip flakey metrics test on windows in filebeat httpjson input. {issue}39676[39676] {pull}39678[39678]
 - Fix flakey test on Windows 2022 in packetbeat/route. {issue}39698[39698] {pull}39822[39822]
+- Fix bug in minimum length for request trace logging. {pull}39834[39834]
 
 ==== Added
 

CHANGELOG.next.asciidoc

@@ -279,6 +279,7 @@ https://github.com/elastic/beats/compare/v8.8.1\...main[Check the HEAD diff]
 - Update CEL mito extensions to v1.12.2. {pull}39755[39755]
 - Add support for base64-encoded HMAC headers to HTTP Endpoint. {pull}39655[39655]
 - Add user group membership support to Okta entity analytics provider. {issue}39814[39814] {pull}39815[39815]
+- Add request trace support for Okta and EntraID entity analytics providers. {pull}39821[39821]
 
 *Auditbeat*
 

filebeat/input/filestream/copytruncate_prospector.go

@@ -18,14 +18,13 @@
 package filestream
 
 import (
+	"errors"
 	"os"
 	"regexp"
 	"sort"
 	"strconv"
 	"time"
 
-	"github.com/urso/sderr"
-
 	loginp "github.com/elastic/beats/v7/filebeat/input/filestream/internal/input-logfile"
 	input "github.com/elastic/beats/v7/filebeat/input/v2"
 	"github.com/elastic/beats/v7/libbeat/common/file"
@@ -230,7 +229,7 @@ func (p *copyTruncateFileProspector) Run(ctx input.Context, s loginp.StateMetada
 
 	errs := tg.Wait()
 	if len(errs) > 0 {
-		log.Error("%s", sderr.WrapAll(errs, "running prospector failed"))
+		log.Errorf("running prospector failed: %v", errors.Join(errs...))
 	}
 }
 

filebeat/input/filestream/internal/input-logfile/manager.go

@@ -25,8 +25,6 @@ import (
 	"sync"
 	"time"
 
-	"github.com/urso/sderr"
-
 	"github.com/elastic/go-concert/unison"
 
 	v2 "github.com/elastic/beats/v7/filebeat/input/v2"
@@ -141,7 +139,7 @@ func (cim *InputManager) Init(group unison.Group) error {
 	if err != nil {
 		store.Release()
 		cim.shutdown()
-		return sderr.Wrap(err, "Can not start registry cleanup process")
+		return fmt.Errorf("Can not start registry cleanup process: %w", err)
 	}
 
 	return nil

filebeat/input/filestream/prospector.go

@@ -18,11 +18,10 @@
 package filestream
 
 import (
+	"errors"
 	"fmt"
 	"time"
 
-	"github.com/urso/sderr"
-
 	loginp "github.com/elastic/beats/v7/filebeat/input/filestream/internal/input-logfile"
 	input "github.com/elastic/beats/v7/filebeat/input/v2"
 	"github.com/elastic/beats/v7/libbeat/beat"
@@ -160,7 +159,7 @@ func (p *fileProspector) Run(ctx input.Context, s loginp.StateMetadataUpdater, h
 
 	errs := tg.Wait()
 	if len(errs) > 0 {
-		log.Error("%s", sderr.WrapAll(errs, "running prospector failed"))
+		log.Errorf("running prospector failed: %v", errors.Join(errs...))
 	}
 }
 

filebeat/input/journald/input.go

@@ -20,10 +20,10 @@
 package journald
 
 import (
+	"fmt"
 	"time"
 
 	"github.com/coreos/go-systemd/v22/sdjournal"
-	"github.com/urso/sderr"
 
 	"github.com/elastic/beats/v7/filebeat/input/journald/pkg/journalfield"
 	"github.com/elastic/beats/v7/filebeat/input/journald/pkg/journalread"
@@ -181,7 +181,7 @@ func (inp *journald) open(log *logp.Logger, canceler input.Canceler, src cursor.
 		withTransports(inp.Transports),
 		withSyslogIdentifiers(inp.Identifiers))
 	if err != nil {
-		return nil, sderr.Wrap(err, "failed to create reader for %{path} journal", src.Name())
+		return nil, fmt.Errorf("failed to create reader for %s journal: %w", src.Name(), err)
 	}
 
 	return reader, nil

filebeat/input/journald/pkg/journalread/reader.go

@@ -27,7 +27,6 @@ import (
 	"time"
 
 	"github.com/coreos/go-systemd/v22/sdjournal"
-	"github.com/urso/sderr"
 
 	"github.com/elastic/beats/v7/libbeat/common/backoff"
 	"github.com/elastic/beats/v7/libbeat/common/cleanup"
@@ -96,27 +95,27 @@ func openJournal(path string) (*sdjournal.Journal, error) {
 	if path == localSystemJournalID || path == "" {
 		j, err := sdjournal.NewJournal()
 		if err != nil {
-			err = sderr.Wrap(err, "failed to open local journal")
+			err = fmt.Errorf("failed to open local journal: %w", err)
 		}
 		return j, err
 	}
 
 	stat, err := os.Stat(path)
 	if err != nil {
-		return nil, sderr.Wrap(err, "failed to read meta data for %{path}", path)
+		return nil, fmt.Errorf("failed to read meta data for %s: %w", path, err)
 	}
 
 	if stat.IsDir() {
 		j, err := sdjournal.NewJournalFromDir(path)
 		if err != nil {
-			err = sderr.Wrap(err, "failed to open journal directory %{path}", path)
+			err = fmt.Errorf("failed to open journal directory %s: %w", path, err)
 		}
 		return j, err
 	}
 
 	j, err := sdjournal.NewJournalFromFiles(path)
 	if err != nil {
-		err = sderr.Wrap(err, "failed to open journal file %{path}", path)
+		err = fmt.Errorf("failed to open journal file %s: %w", path, err)
 	}
 	return j, err
 }

filebeat/input/v2/input-cursor/input.go

@@ -19,12 +19,11 @@ package cursor
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"runtime/debug"
 	"time"
 
-	"github.com/urso/sderr"
-
 	"github.com/elastic/go-concert/ctxtool"
 	"github.com/elastic/go-concert/unison"
 
@@ -81,7 +80,7 @@ func (inp *managedInput) Test(ctx input.TestContext) error {
 
 	errs := grp.Wait()
 	if len(errs) > 0 {
-		return sderr.WrapAll(errs, "input tests failed")
+		return fmt.Errorf("input tests failed: %w", errors.Join(errs...))
 	}
 	return nil
 }
@@ -127,7 +126,7 @@ func (inp *managedInput) Run(
 	}
 
 	if errs := grp.Wait(); len(errs) > 0 {
-		return sderr.WrapAll(errs, "input %{id} failed", ctx.ID)
+		return fmt.Errorf("input %s failed: %w", ctx.ID, errors.Join(errs...))
 	}
 	return nil
 }

filebeat/input/v2/input-cursor/manager.go

@@ -20,11 +20,10 @@ package cursor
 import (
 	"context"
 	"errors"
+	"fmt"
 	"sync"
 	"time"
 
-	"github.com/urso/sderr"
-
 	"github.com/elastic/go-concert/unison"
 
 	v2 "github.com/elastic/beats/v7/filebeat/input/v2"
@@ -131,7 +130,7 @@ func (cim *InputManager) Init(group unison.Group) error {
 	if err != nil {
 		store.Release()
 		cim.shutdown()
-		return sderr.Wrap(err, "Can not start registry cleanup process")
+		return fmt.Errorf("Can not start registry cleanup process: %w", err)
 	}
 
 	return nil

go.mod

@@ -144,7 +144,7 @@ require (
 	github.com/stretchr/testify v1.9.0
 	github.com/tsg/go-daemon v0.0.0-20200207173439-e704b93fd89b
 	github.com/ugorji/go/codec v1.1.8
-	github.com/urso/sderr v0.0.0-20210525210834-52b04e8f5c71
+	github.com/urso/sderr v0.0.0-20210525210834-52b04e8f5c71 // indirect
 	github.com/vmware/govmomi v0.0.0-20170802214208-2cad15190b41
 	go.elastic.co/ecszap v1.0.2
 	go.elastic.co/go-licence-detector v0.6.0
@@ -232,7 +232,6 @@ require (
 	go.elastic.co/apm/module/apmhttp/v2 v2.6.0
 	go.elastic.co/apm/v2 v2.6.0
 	go.mongodb.org/mongo-driver v1.5.1
-	golang.org/x/exp v0.0.0-20231127185646-65229373498e
 	golang.org/x/tools/go/vcs v0.1.0-deprecated
 	google.golang.org/genproto/googleapis/api v0.0.0-20230913181813-007df8e322eb
 	gopkg.in/natefinch/lumberjack.v2 v2.0.0
@@ -386,6 +385,7 @@ require (
 	go.opentelemetry.io/otel v1.24.0 // indirect
 	go.opentelemetry.io/otel/metric v1.24.0 // indirect
 	go.opentelemetry.io/otel/trace v1.24.0 // indirect
+	golang.org/x/exp v0.0.0-20231127185646-65229373498e // indirect
 	golang.org/x/term v0.19.0 // indirect
 	golang.org/x/xerrors v0.0.0-20220907171357-04be3eba64a2 // indirect
 	google.golang.org/appengine v1.6.7 // indirect

libbeat/outputs/elasticsearch/client.go

@@ -26,7 +26,6 @@ import (
 	"time"
 
 	"go.elastic.co/apm/v2"
-	"gotest.tools/gotestsum/log"
 
 	"github.com/elastic/beats/v7/libbeat/beat"
 	"github.com/elastic/beats/v7/libbeat/beat/events"
@@ -433,7 +432,7 @@ func (client *Client) bulkCollectPublishFails(bulkResult bulkResult) ([]publishe
 
 		if client.applyItemStatus(events[i], itemStatus, itemMessage, &stats) {
 			eventsToRetry = append(eventsToRetry, events[i])
-			log.Debugf("Bulk item insert failed (i=%v, status=%v): %s", i, itemStatus, itemMessage)
+			client.log.Debugf("Bulk item insert failed (i=%v, status=%v): %s", i, itemStatus, itemMessage)
 		}
 	}
 

x-pack/filebeat/docs/inputs/input-entity-analytics.asciidoc

@@ -509,6 +509,18 @@ This is a list of optional query parameters. The default is `["accountEnabled",
 "displayName", "operatingSystem", "operatingSystemVersion", "physicalIds", "extensionAttributes",
 "alternativeSecurityIds"]`.
 
+[float]
+==== `tracer.filename`
+
+It is possible to log HTTP requests and responses to the EntraID API to a local file-system for debugging configurations.
+This option is enabled by setting the `tracer.filename` value. Additional options are available to
+tune log rotation behavior.
+
+To differentiate the trace files generated from different input instances, a placeholder `*` can be added to the filename and will be replaced with the input instance id.
+For Example, `http-request-trace-*.ndjson`.
+
+Enabling this option compromises security and should only be used for debugging.
+
 [id="provider-okta"]
 ==== Okta User Identities (`okta`)
 
@@ -797,6 +809,18 @@ The interval in which incremental updates should occur. The interval must be
 shorter than the full synchronization interval (`sync_interval`). Expressed as a
 duration string (e.g., 1m, 3h, 24h). Defaults to `15m` (15 minutes).
 
+[float]
+==== `tracer.filename`
+
+It is possible to log HTTP requests and responses to the Okta API to a local file-system for debugging configurations.
+This option is enabled by setting the `tracer.filename` value. Additional options are available to
+tune log rotation behavior.
+
+To differentiate the trace files generated from different input instances, a placeholder `*` can be added to the filename and will be replaced with the input instance id.
+For Example, `http-request-trace-*.ndjson`.
+
+Enabling this option compromises security and should only be used for debugging.
+
 [float]
 ==== Metrics
 

x-pack/filebeat/input/cel/input.go

@@ -754,7 +754,7 @@ func newClient(ctx context.Context, cfg config, log *logp.Logger, reg *monitorin
 		)
 		traceLogger := zap.New(core)
 
-		const margin = 1e3 // 1OkB ought to be enough room for all the remainder of the trace details.
+		const margin = 10e3 // 1OkB ought to be enough room for all the remainder of the trace details.
 		maxSize := cfg.Resource.Tracer.MaxSize * 1e6
 		trace = httplog.NewLoggingRoundTripper(c.Transport, traceLogger, max(0, maxSize-margin), log)
 		c.Transport = trace

x-pack/filebeat/input/entityanalytics/provider/azuread/azure.go

@@ -48,6 +48,8 @@ type azure struct {
 	logger  *logp.Logger
 	auth    authenticator.Authenticator
 	fetcher fetcher.Fetcher
+
+	ctx v2.Context
 }
 
 // Name returns the name of this provider.
@@ -71,6 +73,7 @@ func (p *azure) Test(testCtx v2.TestContext) error {
 // Run will start data collection on this provider.
 func (p *azure) Run(inputCtx v2.Context, store *kvstore.Store, client beat.Client) error {
 	p.logger = inputCtx.Logger.With("tenant_id", p.conf.TenantID, "provider", Name)
+	p.ctx = inputCtx
 	p.auth.SetLogger(p.logger)
 	p.fetcher.SetLogger(p.logger)
 	p.metrics = newMetrics(inputCtx.ID, nil)
@@ -575,7 +578,7 @@ func (p *azure) configure(cfg *config.C) (kvstore.Input, error) {
 	if p.auth, err = oauth2.New(cfg, p.Manager.Logger); err != nil {
 		return nil, fmt.Errorf("unable to create authenticator: %w", err)
 	}
-	if p.fetcher, err = graph.New(cfg, p.Manager.Logger, p.auth); err != nil {
+	if p.fetcher, err = graph.New(ctxtool.FromCanceller(p.ctx.Cancelation), p.ctx.ID, cfg, p.Manager.Logger, p.auth); err != nil {
 		return nil, fmt.Errorf("unable to create fetcher: %w", err)
 	}
 

x-pack/filebeat/input/entityanalytics/provider/azuread/fetcher/graph/.gitignore

@@ -0,0 +1 @@
+*.ndjson