diff --git a/Makefile b/Makefile index 8ef5213cdfb..78f49b7eb7a 100644 --- a/Makefile +++ b/Makefile @@ -13,7 +13,11 @@ GOTESTFLAGS?= # Prevent unintended modifications of go.[mod|sum] GOMODFLAG?=-mod=readonly -DEFAULT_TAGS=grpcnotrace,pebblegozstd,nooteloutput +# http2legacy has been added to mitigate the changes on http2 +# package on go 1.27 https://pkg.go.dev/golang.org/x/net/http2 +# An issue is created to implement long term fix on +# gmux: https://github.com/elastic/gmux/issues/103 +DEFAULT_TAGS=grpcnotrace,pebblegozstd,nooteloutput,http2legacy PYTHON_ENV?=. PYTHON_VENV_DIR:=$(PYTHON_ENV)/build/ve/$(shell go env GOOS) diff --git a/cmd/intake-receiver/go.mod b/cmd/intake-receiver/go.mod index b0a8a542863..88d4011f107 100644 --- a/cmd/intake-receiver/go.mod +++ b/cmd/intake-receiver/go.mod @@ -1,6 +1,6 @@ module github.com/elastic/apm-server/cmd/intake-receiver -go 1.26.10 +go 1.27.3 require ( github.com/stretchr/testify v1.6.1 diff --git a/go.mod b/go.mod index cf434a93a0a..00a1b9fbf35 100644 --- a/go.mod +++ b/go.mod @@ -1,6 +1,6 @@ module github.com/elastic/apm-server -go 1.26.10 +go 1.27.3 require ( github.com/KimMachineGun/automemlimit v1.0.0 diff --git a/integrationservertest/go.mod b/integrationservertest/go.mod index 7463494871e..6f1f6d94fcb 100644 --- a/integrationservertest/go.mod +++ b/integrationservertest/go.mod @@ -1,6 +1,6 @@ module github.com/elastic/apm-server/integrationservertest -go 1.26.10 +go 1.27.3 require ( github.com/dnaeon/go-vcr v1.2.0 diff --git a/internal/beater/api/intake/handler.go b/internal/beater/api/intake/handler.go index 2e5d82ec377..1edbb5a6152 100644 --- a/internal/beater/api/intake/handler.go +++ b/internal/beater/api/intake/handler.go @@ -148,8 +148,7 @@ func writeStreamResult(c *request.Context, streamResult elasticapm.Result, strea func processStreamError(err error) (request.ResultID, jsonError) { errID := request.IDResponseErrorsInternal - var invalidInput *elasticapm.InvalidInputError - if errors.As(err, &invalidInput) { + if invalidInput, ok := errors.AsType[*elasticapm.InvalidInputError](err); ok { if invalidInput.TooLarge { errID = request.IDResponseErrorsRequestTooLarge } else { diff --git a/internal/beater/middleware/middleware.go b/internal/beater/middleware/middleware.go index 03d844960dd..5b68676a8e9 100644 --- a/internal/beater/middleware/middleware.go +++ b/internal/beater/middleware/middleware.go @@ -18,6 +18,8 @@ package middleware import ( + "slices" + "github.com/elastic/apm-server/internal/beater/request" ) @@ -27,9 +29,9 @@ type Middleware func(request.Handler) (request.Handler, error) // Wrap wraps a request.Handler into given middleware functions, // maintaining order from the last to the first middleware func Wrap(h request.Handler, m ...Middleware) (request.Handler, error) { - for i := len(m) - 1; i >= 0; i-- { + for _, v := range slices.Backward(m) { var e error - h, e = m[i](h) + h, e = v(h) if e != nil { return nil, e } diff --git a/internal/beater/waitready.go b/internal/beater/waitready.go index def79161f5c..6e87385d12e 100644 --- a/internal/beater/waitready.go +++ b/internal/beater/waitready.go @@ -57,8 +57,7 @@ func waitReady( } } if err := check(ctx); err != nil { - var e *actionableError - if errors.As(err, &e) { + if e, ok := errors.AsType[*actionableError](err); ok { logger.Errorf("precondition '%s' failed: %s", e.Name, e.Error()) } else { logger.Errorf("precondition failed: %s", err) diff --git a/internal/publish/acker.go b/internal/publish/acker.go index e5308a11d40..02fbd2ff143 100644 --- a/internal/publish/acker.go +++ b/internal/publish/acker.go @@ -30,7 +30,7 @@ import ( // that blocks until all clients are closed, and all published events at the // time the clients are closed are acknowledged. type WaitPublishedAcker struct { - active int64 // atomic + active atomic.Int64 // atomic mu sync.Mutex empty *sync.Cond @@ -75,11 +75,11 @@ func (w *WaitPublishedAcker) ClientClosed() { } func (w *WaitPublishedAcker) incref(n int64) { - atomic.AddInt64(&w.active, 1) + w.active.Add(1) } func (w *WaitPublishedAcker) decref(n int64) { - if atomic.AddInt64(&w.active, int64(-n)) == 0 { + if w.active.Add(int64(-n)) == 0 { w.empty.Broadcast() } } @@ -96,7 +96,7 @@ func (w *WaitPublishedAcker) Wait(ctx context.Context) error { w.mu.Lock() defer w.mu.Unlock() - for atomic.LoadInt64(&w.active) != 0 && ctx.Err() == nil { + for w.active.Load() != 0 && ctx.Err() == nil { w.empty.Wait() } return ctx.Err() diff --git a/internal/sourcemap/elasticsearch.go b/internal/sourcemap/elasticsearch.go index aa4ab085e6c..246c4ef489e 100644 --- a/internal/sourcemap/elasticsearch.go +++ b/internal/sourcemap/elasticsearch.go @@ -74,8 +74,7 @@ func NewElasticsearchFetcher(c *elasticsearch.Client, index string, maxSourceMap func (s *esFetcher) Fetch(ctx context.Context, name, version, path string) (*sourcemap.Consumer, error) { resp, err := s.runSearchQuery(ctx, name, version, path) if err != nil { - var networkErr net.Error - if errors.As(err, &networkErr) { + if _, ok := errors.AsType[net.Error](err); ok { return nil, fmt.Errorf("failed to reach elasticsearch: %w: %v ", errFetcherUnvailable, err) } return nil, fmt.Errorf("failure querying ES: %w", err) diff --git a/internal/sourcemap/metadata_fetcher_test.go b/internal/sourcemap/metadata_fetcher_test.go index d288a28ce7b..f77bb1e3b74 100644 --- a/internal/sourcemap/metadata_fetcher_test.go +++ b/internal/sourcemap/metadata_fetcher_test.go @@ -37,11 +37,9 @@ import ( func TestMetadataFetcher(t *testing.T) { defaultID := metadata{ - identifier: identifier{ - name: "app", - version: "1.0", - path: "/bundle/path", - }, + name: "app", + version: "1.0", + path: "/bundle/path", contentHash: "foo", } @@ -177,11 +175,9 @@ type metadata struct { func TestInvalidation(t *testing.T) { defaultID := metadata{ - identifier: identifier{ - name: "app", - version: "1.0", - path: "/bundle/path", - }, + name: "app", + version: "1.0", + path: "/bundle/path", contentHash: "foo", } @@ -221,11 +217,9 @@ func TestInvalidation(t *testing.T) { set: map[identifier]string{{name: "example", version: "1.0", path: "/"}: "bar"}, searchReponse: func(w http.ResponseWriter, r *http.Request) { bar := metadata{ - identifier: identifier{ - name: "example", - version: "1.0", - path: "/", - }, + name: "example", + version: "1.0", + path: "/", contentHash: "bar", } m := sourcemapSearchResponseBody([]metadata{defaultID, bar}) diff --git a/internal/sourcemap/processor.go b/internal/sourcemap/processor.go index 5ccb45369d3..b3de309e4e3 100644 --- a/internal/sourcemap/processor.go +++ b/internal/sourcemap/processor.go @@ -21,6 +21,7 @@ import ( "context" "net/url" "path" + "slices" "time" "github.com/elastic/elastic-agent-libs/logp" @@ -99,8 +100,8 @@ func (p BatchProcessor) processException(ctx context.Context, service *modelpb.S // - sourcemap.updated is set to true func (p BatchProcessor) processStacktraceFrames(ctx context.Context, service *modelpb.Service, frames ...*modelpb.StacktraceFrame) { prevFunction := "" - for i := len(frames) - 1; i >= 0; i-- { - frame := frames[i] + for _, frame := range slices.Backward(frames) { + if mapped, function := p.processStacktraceFrame(ctx, service, frame, prevFunction); mapped { prevFunction = function } diff --git a/internal/sourcemap/sourcemap_fetcher_test.go b/internal/sourcemap/sourcemap_fetcher_test.go index 6e964705c4b..1435bf331bc 100644 --- a/internal/sourcemap/sourcemap_fetcher_test.go +++ b/internal/sourcemap/sourcemap_fetcher_test.go @@ -30,11 +30,9 @@ import ( func TestSourcemapFetcher(t *testing.T) { defaultID := metadata{ - identifier: identifier{ - name: "app", - version: "1.0", - path: "/bundle/path", - }, + name: "app", + version: "1.0", + path: "/bundle/path", contentHash: "foo", } diff --git a/systemtest/go.mod b/systemtest/go.mod index f7fc16b6f67..eb5a6c92b5e 100644 --- a/systemtest/go.mod +++ b/systemtest/go.mod @@ -1,6 +1,6 @@ module github.com/elastic/apm-server/systemtest -go 1.26.10 +go 1.27.3 require ( github.com/docker/docker v28.5.2+incompatible diff --git a/x-pack/apm-server/main.go b/x-pack/apm-server/main.go index c1a0eba76d6..1b4642965a7 100644 --- a/x-pack/apm-server/main.go +++ b/x-pack/apm-server/main.go @@ -122,13 +122,11 @@ func newTailSamplingProcessor(args beater.ServerParams) (*sampling.Processor, er policies := make([]sampling.Policy, len(tailSamplingConfig.Policies)) for i, in := range tailSamplingConfig.Policies { policies[i] = sampling.Policy{ - PolicyCriteria: sampling.PolicyCriteria{ - ServiceName: in.Service.Name, - ServiceEnvironment: in.Service.Environment, - TraceName: in.Trace.Name, - TraceOutcome: in.Trace.Outcome, - }, - SampleRate: in.SampleRate, + ServiceName: in.Service.Name, + ServiceEnvironment: in.Service.Environment, + TraceName: in.Trace.Name, + TraceOutcome: in.Trace.Outcome, + SampleRate: in.SampleRate, } } diff --git a/x-pack/apm-server/sampling/config_test.go b/x-pack/apm-server/sampling/config_test.go index 2705a333410..3391001fd23 100644 --- a/x-pack/apm-server/sampling/config_test.go +++ b/x-pack/apm-server/sampling/config_test.go @@ -41,7 +41,7 @@ func TestNewProcessorConfigInvalid(t *testing.T) { assertInvalidConfigError("invalid local sampling config: Policies unspecified") config.Policies = []sampling.Policy{{ - PolicyCriteria: sampling.PolicyCriteria{ServiceName: "foo"}, + ServiceName: "foo", }} assertInvalidConfigError("invalid local sampling config: Policies does not contain a default (empty criteria) policy") config.Policies[0].PolicyCriteria = sampling.PolicyCriteria{} diff --git a/x-pack/apm-server/sampling/groups_test.go b/x-pack/apm-server/sampling/groups_test.go index a1cb983e7b1..cfecef65b39 100644 --- a/x-pack/apm-server/sampling/groups_test.go +++ b/x-pack/apm-server/sampling/groups_test.go @@ -38,13 +38,11 @@ func TestTraceGroupsPolicies(t *testing.T) { } makePolicy := func(sampleRate float64, serviceName, serviceEnvironment, traceOutcome, traceName string) Policy { return Policy{ - SampleRate: sampleRate, - PolicyCriteria: PolicyCriteria{ - ServiceName: serviceName, - ServiceEnvironment: serviceEnvironment, - TraceName: traceName, - TraceOutcome: traceOutcome, - }, + SampleRate: sampleRate, + ServiceName: serviceName, + ServiceEnvironment: serviceEnvironment, + TraceName: traceName, + TraceOutcome: traceOutcome, } } @@ -206,9 +204,9 @@ func TestTraceGroupsRemoval(t *testing.T) { ingestRateCoefficient = 1.0 ) policies := []Policy{ - {PolicyCriteria: PolicyCriteria{ServiceName: "defined"}, SampleRate: 0.5}, + {ServiceName: "defined", SampleRate: 0.5}, {SampleRate: 0.5}, - {PolicyCriteria: PolicyCriteria{ServiceName: "defined_later"}, SampleRate: 0.5}, + {ServiceName: "defined_later", SampleRate: 0.5}, } groups := newTraceGroups(noop.Meter{}, policies, maxDynamicServices, ingestRateCoefficient) diff --git a/x-pack/apm-server/sampling/processor_test.go b/x-pack/apm-server/sampling/processor_test.go index fc20ecac015..135577f3a89 100644 --- a/x-pack/apm-server/sampling/processor_test.go +++ b/x-pack/apm-server/sampling/processor_test.go @@ -351,11 +351,11 @@ func TestProcessLocalTailSamplingUnsampled(t *testing.T) { func TestProcessLocalTailSamplingPolicyOrder(t *testing.T) { config := newTempdirConfig(t).Config config.Policies = []sampling.Policy{{ - PolicyCriteria: sampling.PolicyCriteria{TraceName: "trace_name"}, - SampleRate: 0.5, + TraceName: "trace_name", + SampleRate: 0.5, }, { - PolicyCriteria: sampling.PolicyCriteria{ServiceName: "service_name"}, - SampleRate: 0.1, + ServiceName: "service_name", + SampleRate: 0.1, }, { PolicyCriteria: sampling.PolicyCriteria{}, SampleRate: 0, @@ -1081,34 +1081,28 @@ func newTempdirConfigLogger(tb testing.TB, logger *logp.Logger) testConfig { tb.Cleanup(func() { db.Close() }) return testConfig{ - tempDir: tempdir, - metricReader: reader, - Config: sampling.Config{ - BatchProcessor: modelpb.ProcessBatchFunc(func(context.Context, *modelpb.Batch) error { return nil }), - MeterProvider: mp, - LocalSamplingConfig: sampling.LocalSamplingConfig{ - FlushInterval: time.Second, - MaxDynamicServices: 1000, - IngestRateDecayFactor: 0.9, - Policies: []sampling.Policy{ - {SampleRate: 0.1}, - }, - }, - RemoteSamplingConfig: sampling.RemoteSamplingConfig{ - Elasticsearch: pubsubtest.Client(nil, nil), - SampledTracesDataStream: sampling.DataStreamConfig{ - Type: "traces", - Dataset: "sampled", - Namespace: "testing", - }, - UUID: "local-apm-server", - }, - StorageConfig: sampling.StorageConfig{ - DB: db, - Storage: newUnlimitedReadWriter(db), - TTL: 30 * time.Minute, - ReadBatchMemoryLimit: 10 << 20, // 10MB - }, + tempDir: tempdir, + metricReader: reader, + BatchProcessor: modelpb.ProcessBatchFunc(func(context.Context, *modelpb.Batch) error { return nil }), + MeterProvider: mp, + FlushInterval: time.Second, + MaxDynamicServices: 1000, + IngestRateDecayFactor: 0.9, + Policies: []sampling.Policy{ + {SampleRate: 0.1}, + }, + Elasticsearch: pubsubtest.Client(nil, nil), + SampledTracesDataStream: sampling.DataStreamConfig{ + Type: "traces", + Dataset: "sampled", + Namespace: "testing", + }, + UUID: "local-apm-server", + StorageConfig: sampling.StorageConfig{ + DB: db, + Storage: newUnlimitedReadWriter(db), + TTL: 30 * time.Minute, + ReadBatchMemoryLimit: 10 << 20, // 10MB }, } } diff --git a/x-pack/apm-server/sampling/reservoir.go b/x-pack/apm-server/sampling/reservoir.go index ff5ce2bad53..f0fb9f5a0c2 100644 --- a/x-pack/apm-server/sampling/reservoir.go +++ b/x-pack/apm-server/sampling/reservoir.go @@ -23,11 +23,9 @@ type weightedRandomSample struct { // with the given random number generator and reservoir size. func newWeightedRandomSample(rng *rand.Rand, reservoirSize int) *weightedRandomSample { return &weightedRandomSample{ - rng: rng, - itemheap: itemheap{ - keys: make([]float64, 0, reservoirSize), - values: make([]string, 0, reservoirSize), - }, + rng: rng, + keys: make([]float64, 0, reservoirSize), + values: make([]string, 0, reservoirSize), } }