Service specific source maps

_meta/beat.yml

@@ -244,9 +244,12 @@ apm-server:
       # Sourcemapping is enabled by default.
       #enabled: true
 
-      # Source maps are always fetched from Elasticsearch, by default using the output.elasticsearch configuration.
+      # Source maps may be fetched from Elasticsearch by using the
+      # output.elasticsearch configuration.
       # A different instance must be configured when using any other output.
       # This setting only affects sourcemap reads - the output determines where sourcemaps are written.
+      # Note: Configuring elasticsearch is not allowed if apm-server is being
+      # managed by Fleet.
       #elasticsearch:
         # Array of hosts to connect to.
         # Scheme and port can be left out and will be set to the default (`http` and `9200`).

apm-server.docker.yml

@@ -244,9 +244,12 @@ apm-server:
       # Sourcemapping is enabled by default.
       #enabled: true
 
-      # Source maps are always fetched from Elasticsearch, by default using the output.elasticsearch configuration.
+      # Source maps may be fetched from Elasticsearch by using the
+      # output.elasticsearch configuration.
       # A different instance must be configured when using any other output.
       # This setting only affects sourcemap reads - the output determines where sourcemaps are written.
+      # Note: Configuring elasticsearch is not allowed if apm-server is being
+      # managed by Fleet.
       #elasticsearch:
         # Array of hosts to connect to.
         # Scheme and port can be left out and will be set to the default (`http` and `9200`).

apm-server.yml

@@ -244,9 +244,12 @@ apm-server:
       # Sourcemapping is enabled by default.
       #enabled: true
 
-      # Source maps are always fetched from Elasticsearch, by default using the output.elasticsearch configuration.
+      # Source maps may be fetched from Elasticsearch by using the
+      # output.elasticsearch configuration.
       # A different instance must be configured when using any other output.
       # This setting only affects sourcemap reads - the output determines where sourcemaps are written.
+      # Note: Configuring elasticsearch is not allowed if apm-server is being
+      # managed by Fleet.
       #elasticsearch:
         # Array of hosts to connect to.
         # Scheme and port can be left out and will be set to the default (`http` and `9200`).

beater/beater.go

@@ -20,6 +20,7 @@ package beater
 import (
 	"context"
 	"net"
+	"net/http"
 	"regexp"
 	"runtime"
 	"strings"
@@ -30,6 +31,7 @@ import (
 
 	"github.com/pkg/errors"
 	"go.elastic.co/apm"
+	"go.elastic.co/apm/module/apmhttp"
 	"golang.org/x/sync/errgroup"
 
 	"github.com/elastic/beats/v7/libbeat/beat"
@@ -606,7 +608,7 @@ func newTransformConfig(beatInfo beat.Info, cfg *config.Config) (*transform.Conf
 		},
 	}
 
-	if cfg.RumConfig.Enabled && cfg.RumConfig.SourceMapping.Enabled && cfg.RumConfig.SourceMapping.ESConfig != nil {
+	if cfg.RumConfig.Enabled && cfg.RumConfig.SourceMapping.Enabled {
 		store, err := newSourcemapStore(beatInfo, cfg.RumConfig.SourceMapping)
 		if err != nil {
 			return nil, err
@@ -618,12 +620,31 @@ func newTransformConfig(beatInfo beat.Info, cfg *config.Config) (*transform.Conf
 }
 
 func newSourcemapStore(beatInfo beat.Info, cfg config.SourceMapping) (*sourcemap.Store, error) {
-	esClient, err := elasticsearch.NewClient(cfg.ESConfig)
+	logger := logp.NewLogger(logs.Sourcemap)
+	if len(cfg.SourceMapConfigs) == 0 {
+		esClient, err := elasticsearch.NewClient(cfg.ESConfig)
+		if err != nil {
+			return nil, err
+		}
+		index := strings.ReplaceAll(cfg.IndexPattern, "%{[observer.version]}", beatInfo.Version)
+		b := sourcemap.NewESStore(esClient, index, logger)
+
+		return sourcemap.NewStore(b, logger, cfg.Cache.Expiration)
+	}
+
+	// make a copy of the default http client, configure a Transport to be
+	// the same as the esClient, instrument it, and set it on the http
+	// client.
+	c := *http.DefaultClient
+	tr, err := elasticsearch.NewHTTPTransport(cfg.ESConfig)
 	if err != nil {
 		return nil, err
 	}
-	index := strings.ReplaceAll(cfg.IndexPattern, "%{[observer.version]}", beatInfo.Version)
-	return sourcemap.NewStore(esClient, index, cfg.Cache.Expiration)
+
+	c.Transport = apmhttp.WrapRoundTripper(tr)
+	b := sourcemap.NewFleetStore(cfg.ESConfig.APIKey, cfg.SourceMapConfigs, &c)
+
+	return sourcemap.NewStore(b, logger, cfg.Cache.Expiration)
 }
 
 // WrapRunServerWithProcessors wraps runServer such that it wraps args.Reporter

beater/config/agentconfig.go

@@ -71,16 +71,20 @@ func (s *AgentConfig) setup() error {
 type Service struct {
 	Name        string `config:"name"`
 	Environment string `config:"environment"`
+	Version     string `config:"version"`
 }
 
 // String implements the Stringer interface.
 func (s *Service) String() string {
-	var name, env string
+	var name, env, version string
 	if s.Name != "" {
 		name = "service.name=" + s.Name
 	}
 	if s.Environment != "" {
 		env = "service.environment=" + s.Environment
 	}
-	return strings.Join([]string{name, env}, " ")
+	if s.Version != "" {
+		version = "service.version=" + s.Version
+	}
+	return strings.Join([]string{name, env, version}, " ")
 }

beater/config/config_test.go

@@ -194,7 +194,8 @@ func TestUnpackConfig(t *testing.T) {
 							MaxRetries: 3,
 							Backoff:    elasticsearch.DefaultBackoffConfig,
 						},
-						esConfigured: true,
+						SourceMapConfigs: []SourceMapConfig{},
+						esConfigured:     true,
 					},
 					LibraryPattern:      "^custom",
 					ExcludeFromGrouping: "^grouping",
@@ -363,8 +364,9 @@ func TestUnpackConfig(t *testing.T) {
 						Cache: Cache{
 							Expiration: 7 * time.Second,
 						},
-						IndexPattern: "apm-*-sourcemap*",
-						ESConfig:     elasticsearch.DefaultConfig(),
+						IndexPattern:     "apm-*-sourcemap*",
+						ESConfig:         elasticsearch.DefaultConfig(),
+						SourceMapConfigs: []SourceMapConfig{},
 					},
 					LibraryPattern:      "rum",
 					ExcludeFromGrouping: "^/webpack",

beater/config/rum.go

@@ -60,11 +60,12 @@ type EventRate struct {
 
 // SourceMapping holds sourecemap config information
 type SourceMapping struct {
-	Cache        Cache                 `config:"cache"`
-	Enabled      bool                  `config:"enabled"`
-	IndexPattern string                `config:"index_pattern"`
-	ESConfig     *elasticsearch.Config `config:"elasticsearch"`
-	esConfigured bool
+	Cache            Cache                 `config:"cache"`
+	Enabled          bool                  `config:"enabled"`
+	IndexPattern     string                `config:"index_pattern"`
+	ESConfig         *elasticsearch.Config `config:"elasticsearch"`
+	SourceMapConfigs []SourceMapConfig     `config:"source_maps"`
+	esConfigured     bool
 }
 
 func (c *RumConfig) setup(log *logp.Logger, dataStreamsEnabled bool, outputESCfg *common.Config) error {
@@ -79,6 +80,15 @@ func (c *RumConfig) setup(log *logp.Logger, dataStreamsEnabled bool, outputESCfg
 		return errors.Wrapf(err, "Invalid regex for `exclude_from_grouping`: ")
 	}
 
+	if c.SourceMapping.esConfigured && len(c.SourceMapping.SourceMapConfigs) > 0 {
+		return errors.New("configuring both source_mapping.elasticsearch and sourcemapping.source_maps not allowed")
+	}
+
+	// No need to unpack the ESConfig if we've set SourceMapConfigs
+	if len(c.SourceMapping.SourceMapConfigs) > 0 {
+		return nil
+	}
+
 	var apiKey string
 	if c.SourceMapping.esConfigured {
 		if dataStreamsEnabled {
@@ -115,10 +125,11 @@ func (s *SourceMapping) Unpack(inp *common.Config) error {
 
 func defaultSourcemapping() SourceMapping {
 	return SourceMapping{
-		Enabled:      true,
-		Cache:        Cache{Expiration: defaultSourcemapCacheExpiration},
-		IndexPattern: defaultSourcemapIndexPattern,
-		ESConfig:     elasticsearch.DefaultConfig(),
+		Enabled:          true,
+		Cache:            Cache{Expiration: defaultSourcemapCacheExpiration},
+		IndexPattern:     defaultSourcemapIndexPattern,
+		ESConfig:         elasticsearch.DefaultConfig(),
+		SourceMapConfigs: []SourceMapConfig{},
 	}
 }
 

beater/config/sourcemapping.go

@@ -0,0 +1,32 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package config
+
+type SourceMapConfig struct {
+	Service   Service   `config:"service"`
+	Bundle    Bundle    `config:"bundle"`
+	SourceMap SourceMap `config:"sourcemap"`
+}
+
+type Bundle struct {
+	Filepath string `config:"filepath"`
+}
+
+type SourceMap struct {
+	URL string `config:"url"`
+}

elasticsearch/config.go

@@ -88,7 +88,7 @@ func connectionConfig(config *Config) (http.RoundTripper, []string, http.Header,
 	if err != nil {
 		return nil, nil, nil, err
 	}
-	transp, err := httpTransport(config)
+	transp, err := NewHTTPTransport(config)
 	if err != nil {
 		return nil, nil, nil, err
 	}
@@ -134,7 +134,8 @@ func addresses(cfg *Config) ([]string, error) {
 	return addresses, nil
 }
 
-func httpTransport(cfg *Config) (*http.Transport, error) {
+// NewHttpTransport configures an *http.Transport according to a *Config
+func NewHTTPTransport(cfg *Config) (*http.Transport, error) {
 	proxy, err := httpProxyURL(cfg)
 	if err != nil {
 		return nil, err

model/error_test.go

@@ -30,12 +30,14 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
+	logs "github.com/elastic/apm-server/log"
 	"github.com/elastic/apm-server/sourcemap"
 	"github.com/elastic/apm-server/sourcemap/test"
 	"github.com/elastic/apm-server/tests"
 	"github.com/elastic/apm-server/transform"
 
 	"github.com/elastic/beats/v7/libbeat/common"
+	"github.com/elastic/beats/v7/libbeat/logp"
 )
 
 func baseException() *Exception {
@@ -805,7 +807,8 @@ func TestSourcemapping(t *testing.T) {
 	assert.Equal(t, 1, *event.Exception.Stacktrace[0].Lineno)
 
 	// transform with sourcemap store
-	store, err := sourcemap.NewStore(test.ESClientWithValidSourcemap(t), "apm-*sourcemap*", time.Minute)
+	b := sourcemap.NewESStore(test.ESClientWithValidSourcemap(t), "apm-*sourcemap*", logp.NewLogger(logs.Sourcemap))
+	store, err := sourcemap.NewStore(b, logp.NewLogger(logs.Sourcemap), time.Minute)
 	require.NoError(t, err)
 	transformedWithSourcemap := event.fields(context.Background(), &transform.Config{
 		RUM: transform.RUMConfig{SourcemapStore: store},

model/sourcemap_test.go

@@ -86,7 +86,8 @@ func TestInvalidateCache(t *testing.T) {
 		// create sourcemap store
 		client, err := estest.NewElasticsearchClient(estest.NewTransport(t, http.StatusOK, nil))
 		require.NoError(t, err)
-		store, err := sourcemap.NewStore(client, "foo", time.Minute)
+		b := sourcemap.NewESStore(client, "foo", logp.NewLogger(logs.Sourcemap))
+		store, err := sourcemap.NewStore(b, logp.NewLogger(logs.Sourcemap), time.Minute)
 		require.NoError(t, err)
 
 		// transform with sourcemap store

model/stacktrace_frame_test.go

@@ -28,8 +28,10 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/elastic/apm-server/elasticsearch"
+	logs "github.com/elastic/apm-server/log"
 
 	"github.com/elastic/beats/v7/libbeat/common"
+	"github.com/elastic/beats/v7/libbeat/logp"
 
 	"github.com/elastic/apm-server/sourcemap"
 	"github.com/elastic/apm-server/sourcemap/test"
@@ -422,7 +424,9 @@ func TestLibraryFrame(t *testing.T) {
 }
 
 func testSourcemapStore(t *testing.T, client elasticsearch.Client) *sourcemap.Store {
-	store, err := sourcemap.NewStore(client, "apm-*sourcemap*", time.Minute)
+	logger := logp.NewLogger(logs.Sourcemap)
+	b := sourcemap.NewESStore(client, "apm-*sourcemap*", logger)
+	store, err := sourcemap.NewStore(b, logger, time.Minute)
 	require.NoError(t, err)
 	return store
 }

sourcemap/es_store.go

@@ -44,7 +44,8 @@ var (
 	errSourcemapWrongFormat = errors.New("Sourcemapping ES Result not in expected format")
 )
 
-type esStore struct {
+// ESStore handles making sourcemap requests to an ElasticSearch backend.
+type ESStore struct {
 	client elasticsearch.Client
 	index  string
 	logger *logp.Logger
@@ -65,7 +66,13 @@ type esSourcemapResponse struct {
 	} `json:"hits"`
 }
 
-func (s *esStore) fetch(ctx context.Context, name, version, path string) (string, error) {
+// NewESStore returns an instance of ESStore for interacting with sourcemaps
+// stored in ElasticSearch.
+func NewESStore(c elasticsearch.Client, index string, logger *logp.Logger) *ESStore {
+	return &ESStore{c, index, logger}
+}
+
+func (s *ESStore) fetch(ctx context.Context, name, version, path string) (string, error) {
 	statusCode, body, err := s.runSearchQuery(ctx, name, version, path)
 	if err != nil {
 		return "", errors.Wrap(err, errMsgESFailure)
@@ -87,7 +94,7 @@ func (s *esStore) fetch(ctx context.Context, name, version, path string) (string
 	return parse(body, name, version, path, s.logger)
 }
 
-func (s *esStore) runSearchQuery(ctx context.Context, name, version, path string) (int, io.ReadCloser, error) {
+func (s *ESStore) runSearchQuery(ctx context.Context, name, version, path string) (int, io.ReadCloser, error) {
 	// build and encode the query
 	var buf bytes.Buffer
 	if err := json.NewEncoder(&buf).Encode(query(name, version, path)); err != nil {

sourcemap/es_store_test.go

@@ -36,7 +36,7 @@ import (
 	"github.com/elastic/apm-server/sourcemap/test"
 )
 
-func Test_esFetcher_fetchError(t *testing.T) {
+func Test_ESFetcher_fetchError(t *testing.T) {
 	for name, tc := range map[string]struct {
 		statusCode int
 		esBody     map[string]interface{}
@@ -77,7 +77,7 @@ func Test_esFetcher_fetchError(t *testing.T) {
 	}
 }
 
-func Test_esFetcher_fetch(t *testing.T) {
+func Test_ESFetcher_fetch(t *testing.T) {
 	for name, tc := range map[string]struct {
 		client   elasticsearch.Client
 		filePath string
@@ -101,6 +101,6 @@ func Test_esFetcher_fetch(t *testing.T) {
 	}
 }
 
-func testESStore(client elasticsearch.Client) *esStore {
-	return &esStore{client: client, index: "apm-sourcemap", logger: logp.NewLogger(logs.Sourcemap)}
+func testESStore(client elasticsearch.Client) *ESStore {
+	return NewESStore(client, "apm-sourcemap", logp.NewLogger(logs.Sourcemap))
 }