feat: bump go to 1.27

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)

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

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

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

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 {

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
 		}

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)

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()

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)

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})

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 := "<anonymous>"
-	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
 		}

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",
 	}
 

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

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,
 		}
 	}
 

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{}

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)
 

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
 		},
 	}
 }

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),
 	}
 }